أنماط تصميم API لـ LLM

كيفية تصميم واجهات API لـ LLM تؤثر بشكل كبير على تجربة المستخدم والتكلفة وقابلية التوسع. يغطي هذا الدرس الأنماط الأساسية التي يجب أن يعرفها كل مهندس ذكاء اصطناعي.

المتزامن مقابل غير المتزامن

النمط المتزامن

# بسيط لكن يحجب العميل
@app.post("/chat")
async def chat(request: ChatRequest):
    response = await llm.complete(request.message)  # يحجب لـ 1-5 ثوانٍ
    return {"response": response}

استخدمه عندما:

• وقت الاستجابة أقل من 3 ثوانٍ
• تفاعلات بسيطة ذات دورة واحدة
• تطبيقات منخفضة الحركة

النمط غير المتزامن

# غير حاجب مع استطلاع المهام
@app.post("/chat")
async def chat(request: ChatRequest):
    job_id = await queue.enqueue(process_chat, request)
    return {"job_id": job_id, "status": "processing"}

@app.get("/chat/{job_id}")
async def get_result(job_id: str):
    result = await queue.get_result(job_id)
    if result is None:
        return {"status": "processing"}
    return {"status": "complete", "response": result}

استخدمه عندما:

• وقت الاستجابة أكثر من 5 ثوانٍ
• سير عمل معقد متعدد الخطوات
• تحتاج قدرة إعادة المحاولة/الاسترداد

نمط البث

النمط الأهم لتطبيقات الذكاء الاصطناعي الحديثة.

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for chunk in llm.stream(request.message):
            yield f"data: {json.dumps({'chunk': chunk})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

التعامل من جانب العميل:

// مثال EventSource في JavaScript
const eventSource = new EventSource('/chat/stream');
eventSource.onmessage = (event) => {
    if (event.data === '[DONE]') {
        eventSource.close();
        return;
    }
    const data = JSON.parse(event.data);
    appendToUI(data.chunk);
};

لماذا البث مهم:

• الوقت حتى أول رمز: 200 مللي ثانية مقابل 3000 مللي ثانية للاستجابة الكاملة
• المستخدمون يرون التقدم فوراً
• أداء محسوس أفضل

تجميع الطلبات

جمع طلبات متعددة لتقليل العبء.

class BatchProcessor:
    def __init__(self, max_batch_size=10, max_wait_ms=100):
        self.batch = []
        self.max_size = max_batch_size
        self.max_wait = max_wait_ms

    async def process(self, request):
        self.batch.append(request)

        if len(self.batch) >= self.max_size:
            return await self._flush()

        await asyncio.sleep(self.max_wait / 1000)
        if self.batch:
            return await self._flush()

    async def _flush(self):
        batch = self.batch
        self.batch = []
        # معالجة جميع الطلبات في استدعاء LLM واحد
        combined = self._combine_requests(batch)
        response = await llm.complete(combined)
        return self._split_response(response, len(batch))

المقايضة:

النهج	زمن الاستجابة	الإنتاجية	التكلفة
فردي	منخفض	منخفضة	عالية
مجمّع	متوسط	عالية	منخفضة

إصدار API

دائماً أضف إصداراً لواجهات AI API—النماذج والمطالبات تتطور.

# الإصدار في URL
@app.post("/v1/chat")
async def chat_v1(request: ChatRequest):
    return await process_with_gpt35(request)

@app.post("/v2/chat")
async def chat_v2(request: ChatRequest):
    return await process_with_gpt4(request)

# أو الإصدار في الرأس
@app.post("/chat")
async def chat(
    request: ChatRequest,
    api_version: str = Header(default="2024-01")
):
    if api_version == "2024-01":
        return await process_v1(request)
    return await process_v2(request)

تصميم استجابة الخطأ

أخطاء الذكاء الاصطناعي تحتاج تواصلاً واضحاً:

class AIErrorResponse(BaseModel):
    error_code: str
    message: str
    retry_after: Optional[int] = None
    fallback_available: bool = False

# أمثلة أخطاء
ERRORS = {
    "RATE_LIMITED": AIErrorResponse(
        error_code="RATE_LIMITED",
        message="طلبات كثيرة جداً. يرجى إعادة المحاولة.",
        retry_after=30
    ),
    "CONTEXT_TOO_LONG": AIErrorResponse(
        error_code="CONTEXT_TOO_LONG",
        message="المدخل يتجاوز الحد الأقصى للرموز.",
        fallback_available=True  # يمكن المحاولة بسياق أقصر
    ),
    "CONTENT_FILTERED": AIErrorResponse(
        error_code="CONTENT_FILTERED",
        message="تم تصفية الطلب للسلامة.",
        fallback_available=False
    )
}

بعد ذلك، سنستكشف أنظمة إدارة المطالبات للإنتاج. :::