أنماط البنية لتطبيقات LLM الإنتاجية

مقدمة

بناء تطبيق LLM يعمل في عرض تجريبي أمر بسيط نسبيًا. بناء تطبيق موثوق وفعّال من حيث التكلفة وقابل للمراقبة في الإنتاج تحدٍ مختلف تمامًا. تتطلب تطبيقات LLM الإنتاجية أنماط بنية تعالج قضايا الزمن والتكلفة والموثوقية والمراقبة التي لا تنشأ في النماذج الأولية.

يغطي هذا المقال أنماط البنية الأساسية لأنظمة LLM الإنتاجية، مع إرشادات تنفيذ عملية لكل منها.

التخزين المؤقت

التخزين المؤقت هو أكثر الطرق فعالية لتقليل الزمن والتكلفة في تطبيقات LLM. العديد من أحمال العمل الواقعية لها تكرار كبير — يطرح المستخدمون أسئلة مشابهة، ويُعالَج نفس السياق بشكل متكرر.

التخزين المؤقت الدلالي

التخزين المؤقت التقليدي بالمطابقة التامة يفوّت الاستعلامات المتكافئة دلاليًا. يستخدم التخزين المؤقت الدلالي التضمينات لتحديد الاستعلامات السابقة المشابهة:

import hashlib
from datetime import timedelta
from redis import Redis
import numpy as np

redis = Redis()

def get_cached_response(query: str, query_embedding: list, threshold: float = 0.95):
    # التحقق من المطابقة التامة أولًا
    cache_key = f"llm:exact:{hashlib.md5(query.encode()).hexdigest()}"
    cached = redis.get(cache_key)
    if cached:
        return cached.decode()
    
    # البحث الدلالي في الذاكرة المؤقتة
    # خزّن التضمينات في فهرس متجهي (مثل Redis مع RediSearch)
    similar = search_similar_queries(query_embedding, threshold)
    if similar:
        return similar[0]["response"]
    
    return None

def cache_response(query: str, query_embedding: list, response: str, ttl: int = 3600):
    cache_key = f"llm:exact:{hashlib.md5(query.encode()).hexdigest()}"
    redis.setex(cache_key, ttl, response)
    store_embedding(query, query_embedding, response)

إبطال الذاكرة المؤقتة

إبطال الذاكرة المؤقتة حرج للدقة:

• مبني على الوقت: انتهاء صلاحية الإدخالات بعد فترة محددة
• مبني على المحتوى: إبطال عند تغير المستندات المصدر
• مبني على الإصدار: وسم إدخالات الذاكرة بإصدارات النماذج

تحديد المعدل

واجهات LLM API مكلفة ولها حدود معدل. تنفيذ تحديد المعدل يحمي ميزانيتك ويمنع الإساءة.

تحديد المعدل بالرموز

from datetime import datetime, timedelta
from collections import defaultdict

class TokenRateLimiter:
    def __init__(self, max_tokens_per_minute: int = 100000):
        self.limits = defaultdict(list)
        self.max_tokens = max_tokens_per_minute
    
    def check_and_consume(self, user_id: str, estimated_tokens: int) -> bool:
        now = datetime.now()
        window_start = now - timedelta(minutes=1)
        
        # تنظيف الإدخالات القديمة
        self.limits[user_id] = [
            (ts, tokens) for ts, tokens in self.limits[user_id]
            if ts > window_start
        ]
        
        current_usage = sum(tokens for _, tokens in self.limits[user_id])
        
        if current_usage + estimated_tokens > self.max_tokens:
            return False
        
        self.limits[user_id].append((now, estimated_tokens))
        return True

تحديد المعدل المتدرّج

نفّذ حدودًا مختلفة لمستويات المستخدمين المختلفة:

• المستوى المجاني: 10 طلبات/يوم، 1000 رمز/يوم
• المستوى الاحترافي: 100 طلب/ساعة، 50000 رمز/ساعة
• المؤسسات: حدود مخصصة مع قدرة اندفاع

البدائل وتوجيه النماذج

لا يوجد LLM واحد مثالي لكل مهمة. يجب أن توجّه الأنظمة الإنتاجية الطلبات إلى النموذج الأنسب وتتراجع بسلاسة عند الفشل.

توجيه متعدد النماذج

from enum import Enum
from dataclasses import dataclass

class ModelTier(Enum):
    FAST = "gpt-4o-mini"
    BALANCED = "gpt-4o"
    POWERFUL = "gpt-4o"
    FALLBACK = "gpt-4o-mini"

class ModelRouter:
    def __init__(self):
        self.fallback_chain = [
            ModelTier.BALANCED,
            ModelTier.FAST,
            ModelTier.FALLBACK
        ]
    
    def route(self, query: str, context_length: int, complexity: str) -> ModelTier:
        # التوجيه بناءً على تعقيد المهمة
        if complexity == "simple" or context_length < 500:
            return ModelTier.FAST
        elif complexity == "complex":
            return ModelTier.POWERFUL
        return ModelTier.BALANCED
    
    def execute_with_fallback(self, prompt: str, model: ModelTier) -> str:
        start_index = self.fallback_chain.index(model)
        for tier in self.fallback_chain[start_index:]:
            try:
                return self.call_model(tier, prompt)
            except Exception as e:
                log_error(tier, e)
                continue
        raise RuntimeError("فشلت جميع النماذج")

قواطع الدائرة

منع الفشل المتسلسل عند تعطل مزود LLM:

import time
from circuitbreaker import CircuitBreaker

class LLMCircuitBreaker(CircuitBreaker):
    FAILURE_THRESHOLD = 5
    RECOVERY_TIMEOUT = 60

    @LLMCircuitBreaker
    def call_llm(self, prompt: str) -> str:
        return self.llm.invoke(prompt)

تدفق الاستجابات

يحسّن التدفق الأداء المُدرك بإظهار الرموز أثناء توليدها. هذا أساسي لواجهات الدردشة.

تدفق Server-Sent Events

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/chat")
async def chat(request: dict):
    async def generate():
        async for chunk in llm.astream(request["message"]):
            data = json.dumps({"token": chunk.content})
            yield f"data: {data}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

الاستهلاك من جانب العميل

const eventSource = new EventSource('/chat');
let fullResponse = '';

eventSource.onmessage = (event) => {
    if (event.data === '[DONE]') {
        eventSource.close();
        return;
    }
    const data = JSON.parse(event.data);
    fullResponse += data.token;
    updateUI(fullResponse);
};

المراقبة

تتطلب المراقبة في تطبيقات LLM تتبع المدخلات والمخرجات والزمن والتكاليف ومقاييس الجودة.

هيكل التسجيل

import logging
import json
from datetime import datetime

class LLMLogger:
    def __init__(self):
        self.logger = logging.getLogger("llm_app")
    
    def log_request(self, request_id: str, model: str, prompt: str, 
                    response: str, latency_ms: int, tokens_in: int, 
                    tokens_out: int, cost: float):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "request_id": request_id,
            "model": model,
            "prompt_length": len(prompt),
            "response_length": len(response),
            "latency_ms": latency_ms,
            "tokens_in": tokens_in,
            "tokens_out": tokens_out,
            "cost_usd": cost
        }
        self.logger.info(json.dumps(log_entry))

المقاييس الرئيسية للتتبع

• الزمن: الوقت حتى أول رمز وزمن التوليد الكلي
• استخدام الرموز: رموز الإدخال والإخراج لكل طلب
• التكلفة: تكلفة الدولار لكل طلب والمجموع
• معدل الأخطاء: نسبة الطلبات الفاشلة
• معدل إصابة الذاكرة المؤقتة: نسبة الطلبات المخدومة من الذاكرة
• درجات الجودة: ملاحظات المستخدم أو مقاييس جودة آلية

استخدام LangSmith أو Langfuse

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe()
def generate_response(query: str) -> str:
    # هذه الدالة تُتتبع تلقائيًا
    docs = retrieve(query)
    context = "\n".join([d.page_content for d in docs])
    response = llm.invoke(f"السياق: {context}\n\nالسؤال: {query}")
    return response

تحسين التكلفة

ضغط المطالبات

قلّل رموز الإدخال بضغط المطالبات:

def compress_context(documents: list, max_tokens: int = 4000) -> str:
    """ضغط المستندات لتلائم ميزانية الرموز."""
    total_text = ""
    for doc in documents:
        # لخّص كل مستند إلى الجمل الرئيسية
        summary = summarize_document(doc)
        if count_tokens(total_text + summary) > max_tokens:
            break
        total_text += summary + "\n"
    return total_text

المعالجة الدفعية

لأحمال العمل غير الفورية، دفعة الطلبات لتقليل عبء كل طلب:

async def batch_process(queries: list[str]) -> list[str]:
    """معالجة استعلامات متعددة في دفعة واحدة."""
    messages = [[{"role": "user", "content": q}] for q in queries]
    responses = await llm.abatch(messages)
    return [r.content for r in responses]

خاتمة

تتطلب تطبيقات LLM الإنتاجية أنماط بنية تعالج التحديات الفريدة للعمل مع نماذج اللغة الكبيرة. التخزين المؤقت يقلل التكلفة والزمن. تحديد المعدل يحمي الميزانيات. البدائل وتوجيه النماذج يحسّن الموثوقية. التدفق يعزز تجربة المستخدم. المراقبة توفر رؤية لسلوك النظام.

ابدأ بتنفيذ التخزين المؤقت والمراقبة — فهما يقدّمان أكبر قيمة فورية. ثم أضف تحديد المعدل والبدائل وتوجيه النماذج مع توسع تطبيقك. الرؤية الرئيسية هي أن بنية LLM الإنتاجية تدور حول بناء أنظمة تتدهور بسلاسة، وتكلّف بشكل متوقع، ويمكن تصحيحها عند حدوث مشكلات.