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

يشرح هذا البرنامج التعليمي كيفية بناء خدمة روبوت دردشة صغيرة ولكنها جاهزة للإنتاج باستخدام FastAPI، وبث SSE، وذاكرة المحادثة، وتبديل النماذج. الهدف ليس تقديم عرض تجريبي بسيط، بل الوصول إلى خلفية برمجية (backend) يمكنك وضعها فعلياً خلف واجهة منتج وتطويرها بأمان.

إذا قمت بالفعل بتوجيه أحد الـ SDK المتوافقة مع OpenAI إلى LemonData، فإن هذا المقال يكمل من تلك النقطة. إذا لم تقم بتغيير الـ base URL بعد، فاقرأ دليل الهجرة أولاً. إذا كان اهتمامك الأساسي هو تنظيم الطلبات (request shaping) والتراجع (backoff) تحت الضغط، فاقرأ هذا الدليل مع دليل تحديد معدل طلبات AI API.

ما الذي سنقوم ببنائه

تتكون الخدمة النهائية من ستة أجزاء متحركة:

1. نقطة نهاية (endpoint) متزامنة /chat لاختبارات التأكد من التشغيل (smoke tests).
2. نقطة نهاية للبث /chat/stream لواجهة المستخدم الحقيقية.
3. حالة المحادثة مرتبطة بـ conversation_id.
4. قائمة مسموحة (allowlist) للنماذج حتى لا تتمكن الواجهة الأمامية من طلب معرفات عشوائية.
5. معالجة الأخطاء التي لا تنهار عند أول خطأ 429.
6. مسار واضح للانتقال من نموذج أولي في الذاكرة إلى Redis أو PostgreSQL.

هذا كافٍ لتشغيل بوت دعم، أو مساعد داخلي، أو النسخة الأولى من أداة دردشة مدمجة.

تثبيت الحزمة الأدنى

pip install fastapi uvicorn openai pydantic redis

يمكنك حذف redis في المرحلة الأولى، ولكن من المفيد ربط الاستيراد (import) الآن ليكون مسار الترقية واضحاً.

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

أسرع طريقة للضياع في بناء روبوتات الدردشة هي البدء بـ websockets، واستخدام الأدوات (tool use)، وتنسيق الوكلاء (agent orchestration) قبل أن يكون مسار الطلب الأساسي مستقراً. ابدأ بنقطة نهاية واحدة صغيرة تثبت أن مفتاحك، والـ base URL، وتوجيه النماذج تعمل بشكل صحيح.

from fastapi import FastAPI
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: أضف البث (Streaming) لأن المستخدمين يشعرون بالزمن قبل قياسه

معظم منتجات روبوتات الدردشة تبدو بطيئة ليس لأن النموذج بطيء، بل لأن واجهة المستخدم تظل فارغة حتى يصل الرد الكامل. يعد SSE كافياً للعديد من منتجات الدردشة وله عبء تشغيلي أقل من websockets.

from fastapi.responses import StreamingResponse

@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")

على الواجهة الأمامية، لا يزال أبسط عميل من جانب المتصفح جيداً بما يكفي:

async function sendMessage(payload) {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value, { stream: true });
    console.log(chunk);
  }
}

إذا كان منتجك يستخدم بالفعل عميل متصفح و standard HTTP، فإن SSE يحافظ على بساطة المعمارية.

الخطوة 3: نقل حالة المحادثة خارج جسم الطلب (Request Body)

عادةً ما يحتفظ العرض التجريبي الأول لروبوت الدردشة بالنص الكامل في المتصفح ويرسله في كل مرة. هذا يعمل مع النماذج الأولية، ولكنه يصبح فوضوياً بمجرد حاجتك إلى إعادة المحاولة، أو الجلسات القابلة للاستئناف، أو الأدوات من جانب الخادم.

مخزن في الذاكرة (in-memory store) جيد للبداية:

from collections import defaultdict
import uuid

conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    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

مسار الترقية إلى Redis هو في الغالب أعمال برمجية للتخزين:

import json
import redis

redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

def load_history(conv_id: str) -> list:
    raw = redis_client.get(f"chat:{conv_id}")
    return json.loads(raw) if raw else []

def save_history(conv_id: str, history: list) -> None:
    redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))

استخدم Redis إذا كانت المحادثات تحتاج إلى TTL، أو قابلية الاستئناف، أو النشر على عدة مثيلات (multi-instance). استخدم PostgreSQL إذا كان نص المحادثة نفسه يمثل بيانات للمنتج.

الخطوة 4: تعامل مع الأخطاء كسلوك للمنتج، وليس مجرد استثناءات

إذا كان روبوت الدردشة الخاص بك موجهاً للعملاء، فإن مسار الفشل يهم بقدر مسار النجاح. لا يهتم المستخدم ما إذا كان الفشل ناتجاً عن تحديد معدل الطلبات (rate limiting)، أو الرصيد، أو انقطاع النموذج. ما يهمه هو ما إذا كانت واجهة المستخدم تتجمد.

from openai import APIConnectionError, APIError, RateLimitError

@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():
        full_response = []
        try:
            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"
        except RateLimitError:
            yield "data: [ERROR] The model is busy. Please retry in a few seconds.\n\n"
        except APIConnectionError:
            yield "data: [ERROR] Temporary network issue. Please retry.\n\n"
        except APIError as error:
            yield f"data: [ERROR] {error.message}\n\n"
        else:
            conversations[conv_id].append(
                {"role": "assistant", "content": "".join(full_response)}
            )
        finally:
            yield "data: [DONE]\n\n"

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

إذا كنت تقدم حملاً كبيراً، فيجب عليك أيضاً تنظيم الطلبات قبل وصولها إلى المصدر. الأنماط التفصيلية موجودة في دليل تحديد معدل الطلبات، ولكن النسخة القصيرة هي: استخدم عمليات إعادة محاولة محدودة، واستخدم التذبذب (jitter)، ولا تستخدم أبداً except Exception: time.sleep(1).

الخطوة 5: تبديل النماذج يحتاج إلى قائمة مسموحة، وليس صندوق نص حر

يمكن لمفتاح API واحد الوصول إلى مئات النماذج. هذا لا يعني أن واجهة المستخدم الخاصة بك يجب أن تعرض مئات النماذج. يجب أن تنشر الخلفية البرمجية قائمة مسموحة صغيرة تناسب حالة الاستخدام الخاصة بك.

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

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

هذا يحقق ثلاثة أشياء مفيدة:

• يمنع الواجهة الأمامية من طلب معرفات نماذج غير صالحة أو مهجورة.
• يتيح لك إعادة تعيين فئة لاحقاً دون إعادة نشر كل عميل.
• يمنحك مكاناً واحداً لفرض ضوابط التكلفة.

إذا كان فريقك لا يزال يقرر أي المزودين سيعتمد عليهم، فإن مقارنة الأسعار و مقارنة OpenRouter مقابل LemonData هما صفحتان تستحقان القراءة قبل قفل القائمة المسموحة.

الخطوة 6: أضف اللمسات النهائية للإنتاج قبل وصول حركة المرور

تصبح الخلفية البرمجية لروبوت الدردشة جاهزة للإنتاج عندما يتم التعامل مع التفاصيل المحيطة، وليس عندما تكون مكالمة الدردشة الأساسية ذكية.

قائمة المراجعة قصيرة:

• أضف معرفات الطلبات (request IDs) حتى تتمكن من ربط إخفاقات الواجهة الأمامية بسجلات الخلفية البرمجية.
• حدد سقفاً للتزامن لكل مستخدم وحجم الطلب.
• قم بتقليم التواريخ الطويلة قبل أن تستهلك ميزانية الـ token الخاصة بك.
• سجل النموذج، وزمن الوصول، وحجم المدخلات، وسبب الإنهاء.
• افصل رسائل الخطأ المرئية للمستخدم عن تفاصيل الخطأ الداخلية.
• اختبر نموذجاً بديلاً واحداً حتى تعرف أن التراجع (fallback) يعمل قبل حدوث أول انقطاع.

تقليم السجل يمكن أن يظل بسيطاً:

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    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

النقطة ليست المحاسبة الدقيقة للـ token، بل هي إيقاف تضخم السياق الواضح.

من العرض التجريبي إلى المنتج

بمجرد استقرار هذه الخلفية البرمجية، نادراً ما تكون الترقية التالية هي "المزيد من الذكاء الاصطناعي". عادة ما تكون بنية تحتية مملة:

• المصادقة (auth) حتى لا يتمكن مستخدم من قراءة محادثة مستخدم آخر.
• الاستمرارية (persistence) حتى تنجو الجلسات من عمليات النشر.
• تحديد معدل الطلبات (rate limiting) حتى لا يتمكن مستخدم واحد من حرق حصتك.
• الفوترة أو تتبع الاستخدام إذا كان روبوت الدردشة موجهاً للعملاء.
• التلخيص في الخلفية إذا كانت المحادثات تحتاج إلى ذاكرة طويلة المدى.

لهذا السبب تساعد البوابة الموحدة. بمجرد الانتهاء من هجرة الـ base URL، تتوقف تغييرات النماذج عن كونها إعادة كتابة للمنصة وتصبح مجرد تكوين (configuration).

اختبار التأكد من التشغيل

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"}'

إذا كان بإمكانك بث دورة واحدة، والحفاظ على محادثة واحدة، وإرجاع خطأ نظيف عند فشل قسري، فأنت تمتلك الأساس الصحيح.

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

قم بإنشاء مفتاح API في LemonData، ووجه OpenAI SDK الخاص بك إلى https://api.lemondata.cc/v1، ويمكنك شحن أول نسخة إنتاج من روبوت الدردشة الخاص بك دون إدارة حسابات مزودين منفصلة.

Model	Daily Cost	Monthly Cost
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 دولار لبدء البناء.