بناء روبوت محادثة (Chatbot) يعمل بالذكاء الاصطناعي باستخدام مفتاح API واحد: من الصفر إلى مرحلة الإنتاج في 30 دقيقة

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

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

النقاط الرئيسية

• يحتاج روبوت المحادثة الجاهز للإنتاج إلى ستة أجزاء: نقطة نهاية متزامنة (sync endpoint)، ونقطة نهاية للبث (streaming endpoint)، وحالة محادثة من جانب الخادم، وقائمة سماح للنماذج (allowlist)، ومعالجة حقيقية للأخطاء، ومسار واضح لترقية التخزين.
• أثبت صحة مفتاحك، و الـ base URL، والتوجيه باستخدام نقطة نهاية محادثة صغيرة واحدة قبل إضافة البث أو الذاكرة أو استدعاءات الأدوات.
• تغطي تقنية SSE للبث معظم منتجات المحادثة وتتحمل عبئاً تشغيلياً أقل من الـ websockets.
• اعرض النماذج من خلال قائمة سماح في الخلفية، وليس كحقل نصي حر، حتى لا يتمكن الواجهة الأمامية من طلب معرفات نماذج عشوائية أو قديمة.
• تتغير توفر النماذج وتشكيلاتها بشكل متكرر. تحقق من دليل النماذج الخاص بـ TokenLab (تمت الملاحظة بتاريخ 2026-07-07) قبل تثبيت قائمة السماح الخاصة بك في بيئة الإنتاج.

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

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

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

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

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

pip install fastapi uvicorn openai pydantic redis

يمكنك تخطي redis في المرحلة الأولى، لكن إعداد الاستيراد الآن يجعل الترقية لاحقاً أمراً بسيطاً بدلاً من إعادة هيكلة الكود.

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

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

from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-tokenlab-xxx",
    base_url="https://api.tokenlab.sh/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "deepseek-v4-flash"
    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")

في الواجهة الأمامية، لا يزال قارئ fetch العادي كافياً:

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);
  }
}

إذا كان منتجك يعمل بالفعل على متصفح عبر HTTP قياسي، فإن SSE تجعل البنية أبسط من طبقة websocket.

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

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

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

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، أو قابلية الاستئناف، أو النشر على مثيلات متعددة. استخدم PostgreSQL عندما يكون النص نفسه بيانات منتج تحتاج إلى الاستعلام عنها أو تدقيقها أو إعداد تقارير عنها لاحقاً.

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

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

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: rate limited, please retry shortly]\n\n"
        except APIConnectionError:
            yield "data: [error: connection issue, please retry]\n\n"
        except APIError:
            yield "data: [error: something went wrong on our end]\n\n"
        finally:
            if full_response:
                conversations[conv_id].append(
                    {"role": "assistant", "content": "".join(full_response)}
                )
        yield "data: [DONE]\n\n"

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

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

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

ALLOWED_MODELS = {
    "default": "deepseek-v4-flash",
    "flagship": "gpt-5.5",
    "balanced": "claude-sonnet-5",
}

class ChatRequest(BaseModel):
    message: str
    tier: str = "default"
    conversation_id: str | None = None

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
    # بقية منطق البث يستخدم `model` بدلاً من سلسلة نصية خام يوفرها العميل

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

الخطوة 6: تعامل مع بقية متطلبات الإنتاج

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

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

• أضف معرفات الطلبات (request IDs) حتى تتمكن من ربط فشل الواجهة الأمامية بسجلات الخلفية.
• حدد التزامن لكل مستخدم وحجم الطلب.
• قلص السجلات الطويلة قبل أن تفجر ميزانية الـ tokens الخاصة بك.
• سجل النموذج، وزمن الاستجابة، وحجم الإدخال، وسبب الانتهاء.
• افصل رسائل الخطأ المرئية للمستخدم عن تفاصيل الخطأ الداخلية.
• اختبر نموذجاً بديلاً واحداً حتى تعرف أن التراجع (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

الهدف ليس محاسبة دقيقة للـ tokens، بل إيقاف تضخم السياق الواضح قبل أن يؤثر على فاتورتك أو زمن الاستجابة.

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

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

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

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

اختبار الدخان

uvicorn main:app --reload --port 8000

curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "model": "deepseek-v4-flash"}'

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

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

أنشئ مفتاح API على TokenLab، ووجه الـ OpenAI SDK الخاص بك إلى https://api.tokenlab.sh/v1، ويمكنك شحن الإصدار الأول من روبوت المحادثة الخاص بك دون إدارة حسابات منفصلة عبر المزودين.

النموذج	الفئة النموذجية	ملاحظات
DeepSeek V4 Flash	سريع / افتراضي	خيار افتراضي جيد للدورات ذات الحجم الكبير وزمن الاستجابة المنخفض
GPT-5.5	رائد	استخدمه للدورات التي تحتاج إلى استنتاج أقوى
Claude Sonnet 5	متوازن	خيار قوي للبرمجة والردود بأسلوب المراجعة
Gemini 3.5 Flash	اقتصادي / بديل سريع	بديل منخفض التكلفة وسريع للتوجيه عالي الحجم

تتغير أسعار الـ tokens الدقيقة بشكل متكرر عبر المزودين ولا يتم إعادة إنتاجها هنا كأرقام ثابتة. تحقق من الأسعار الحالية على دليل النماذج (تمت الملاحظة بتاريخ 2026-07-07) قبل وضع الميزانية. من الناحية العملية، توجيه معظم المحادثات إلى فئة سريعة/افتراضية مثل DeepSeek V4 Flash أو Gemini 3.5 Flash وحجز GPT-5.5 أو Claude Sonnet 5 للدورات التي تحتاج إليها يبقي معظم التطبيقات ضمن فاتورة شهرية منخفضة، ولكن تأكد من الأسعار الفعلية لكل مليون token لحسابك قبل الالتزام بميزانية.

الأسئلة الشائعة

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

كيف أعرف النموذج الافتراضي الذي يجب استخدامه؟ ابدأ بنموذج سريع ومنخفض التكلفة مثل DeepSeek V4 Flash أو Gemini 3.5 Flash للفئة الافتراضية، وأضف فئة متوازنة أو استنتاجية على Claude Sonnet 5 أو GPT-5.5 خلف قائمة السماح الموضحة في الخطوة 5. تحقق من دليل النماذج (تمت الملاحظة بتاريخ 2026-07-07) للحصول على الخيارات الحالية، حيث يتم شحن نماذج جديدة ويتم إيقاف النماذج القديمة وفق جدول زمني خارج سيطرتك.

ما الذي يتعطل أولاً عندما ينتقل روبوت المحادثة من العرض التوضيحي إلى حركة المرور الحقيقية؟ غالباً ما يكون مسار الخطأ، وليس مسار النجاح. عمليات إعادة المحاولة غير المحدودة، وفقدان حدود التزامن لكل مستخدم، وسجل المحادثة غير المحدود هي الأسباب الثلاثة الأكثر شيوعاً لانهيار نظام خلفي لروبوت محادثة تحت حمل حقيقي. الخطوتان 4 و 6 أعلاه تعالجان الثلاثة بشكل مباشر.

---

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