بناء روبوت دردشة ذكي باستخدام مفتاح API واحد: من الصفر إلى الإنتاج في 30 دقيقة

في هذا الدرس، سنبني خلفية روبوت دردشة ذكي جاهزة للإنتاج مع استجابات متدفقة، وتاريخ المحادثة، وتبديل النماذج، ومعالجة الأخطاء بشكل صحيح. سنستخدم Python وFastAPI وOpenAI SDK موجهًا إلى مجمع API حتى تتمكن من استخدام أي نموذج.

المتطلبات الأساسية

pip install fastapi uvicorn openai

الخطوة 1: نقطة نهاية دردشة أساسية

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-lemon-xxx",
    base_url="https://api.lemondata.cc/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "gpt-4.1-mini"
    conversation_id: str | None = None

@app.post("/chat")
async def chat(req: ChatRequest):
    response = client.chat.completions.create(
        model=req.model,
        messages=[{"role": "user", "content": req.message}]
    )
    return {"reply": response.choices[0].message.content}

هذا يعمل ولكنه لا يدعم البث، ولا يحتفظ بالتاريخ، ولا يتعامل مع الأخطاء. دعنا نصحح ذلك.

الخطوة 2: إضافة البث

البث يرسل الرموز أثناء توليدها بدلاً من الانتظار حتى اكتمال الرد. يرى المستخدمون الرد يتشكل في الوقت الحقيقي.

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    def generate():
        stream = client.chat.completions.create(
            model=req.model,
            messages=[{"role": "user", "content": req.message}],
            stream=True
        )
        for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                yield f"data: {delta.content}\n\n"
        yield "data: [DONE]\n\n"

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

الخطوة 3: تاريخ المحادثة

تخزين تاريخ المحادثة في الذاكرة (يمكن استبداله بـ Redis أو قاعدة بيانات في الإنتاج).

from collections import defaultdict
import uuid

conversations: dict[str, list] = defaultdict(list)

SYSTEM_PROMPT = "أنت مساعد مفيد. كن مختصرًا ومباشرًا."

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    conv_id = req.conversation_id or str(uuid.uuid4())

    # بناء تاريخ الرسائل
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend(conversations[conv_id])
    messages.append({"role": "user", "content": req.message})

    # تخزين رسالة المستخدم
    conversations[conv_id].append(
        {"role": "user", "content": req.message}
    )

    def generate():
        full_response = []
        stream = client.chat.completions.create(
            model=req.model,
            messages=messages,
            stream=True
        )
        for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                full_response.append(delta.content)
                yield f"data: {delta.content}\n\n"

        # تخزين رد المساعد
        conversations[conv_id].append(
            {"role": "assistant", "content": "".join(full_response)}
        )
        yield f"data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"X-Conversation-ID": conv_id}
    )

الخطوة 4: معالجة الأخطاء

يمكن أن تفشل مكالمات API الخاصة بالذكاء الاصطناعي لعدة أسباب: حدود المعدل، رصيد غير كافٍ، النموذج غير متاح. تعامل مع كل حالة:

from openai import (
    APIError,
    RateLimitError,
    APIConnectionError
)

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)

    def generate():
        try:
            full_response = []
            stream = client.chat.completions.create(
                model=req.model,
                messages=messages,
                stream=True
            )
            for chunk in stream:
                delta = chunk.choices[0].delta
                if delta.content:
                    full_response.append(delta.content)
                    yield f"data: {delta.content}\n\n"

            conversations[conv_id].append(
                {"role": "assistant", "content": "".join(full_response)}
            )

        except RateLimitError as e:
            yield f"data: [ERROR] تم تجاوز الحد. يرجى الانتظار قليلاً.\n\n"
        except APIConnectionError:
            yield f"data: [ERROR] فشل الاتصال. جارٍ إعادة المحاولة...\n\n"
        except APIError as e:
            yield f"data: [ERROR] {e.message}\n\n"

        yield "data: [DONE]\n\n"

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

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    # الاحتفاظ بآخر 10 جولات لإدارة طول السياق
    history = conversations[conv_id][-20:]
    messages.extend(history)
    messages.append({"role": "user", "content": user_msg})
    conversations[conv_id].append({"role": "user", "content": user_msg})
    return messages

الخطوة 5: تبديل النماذج

دع المستخدمين يبدلون النماذج أثناء المحادثة. نماذج مختلفة لاحتياجات مختلفة:

AVAILABLE_MODELS = {
    "fast": "gpt-4.1-mini",
    "smart": "claude-sonnet-4-6",
    "reasoning": "o3",
    "budget": "deepseek-chat",
    "creative": "claude-sonnet-4-6",
}

@app.get("/models")
async def list_models():
    return {"models": AVAILABLE_MODELS}

يمكن للواجهة الأمامية عرض هذه كخيارات. وبما أن جميع النماذج تستخدم نفس تنسيق OpenAI المتوافق عبر المجمع، فإن التبديل هو فقط تغيير معلمة model.

الخطوة 6: إدارة نافذة السياق

المحادثات الطويلة تتجاوز حدود سياق النموذج. نفذ نافذة انزلاقية:

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    """الاحتفاظ بموجه النظام + الرسائل الأخيرة ضمن ميزانية الرموز."""
    # تقدير تقريبي: 1 رمز ≈ 4 أحرف
    system = messages[0]  # دائمًا احتفظ بموجه النظام
    history = messages[1:]

    total_chars = len(system["content"])
    trimmed = []

    for msg in reversed(history):
        msg_chars = len(msg["content"])
        if total_chars + msg_chars > max_tokens * 4:
            break
        trimmed.insert(0, msg)
        total_chars += msg_chars

    return [system] + trimmed

التطبيق الكامل

# شغّل باستخدام: uvicorn main:app --reload --port 8000
# اختبار: curl -N -X POST http://localhost:8000/chat/stream \
#   -H "Content-Type: application/json" \
#   -d '{"message": "Hello!", "model": "gpt-4.1-mini"}'

الكود الكامل أقل من 100 سطر. من هنا يمكنك إضافة:

• المصادقة (مفاتيح API أو JWT)
• التخزين الدائم (PostgreSQL أو Redis للمحادثات)
• تحديد معدل الاستخدام لكل مستخدم
• تتبع الاستخدام والفوترة
• دعم WebSocket للبث ثنائي الاتجاه
• الواجهة الأمامية (React أو Vue أو JavaScript عادي مع EventSource)

تقدير التكلفة

لروبوت دردشة يتعامل مع 1000 محادثة يوميًا (بمتوسط 5 جولات لكل منها):

النموذج	التكلفة اليومية	التكلفة الشهرية
GPT-4.1-mini	~2.40$	~72$
GPT-4.1	~12.00$	~360$
Claude Sonnet 4.6	~18.00$	~540$
DeepSeek V3	~1.68$	~50$

استخدام GPT-4.1-mini لمعظم المحادثات والترقية إلى Claude Sonnet 4.6 فقط عند طلب المستخدمين يحافظ على التكاليف تحت 100 دولار شهريًا لمعظم التطبيقات.

---

احصل على مفتاح API الخاص بك: lemondata.cc يوفر أكثر من 300 نموذج عبر نقطة نهاية واحدة. رصيد مجاني بقيمة 1 دولار للبدء في البناء.