بناء روبوت دردشة ذكاء اصطناعي بمفتاح 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 دولار للبدء في البناء.