بناء روبوت محادثة باستخدام واجهة برمجة تطبيقات النموذج الكبير: واجهة خلفية FastAPI مع واجهة أمامية للتدفق

ارسم مخططًا صغيرًا أولاً: ثلاثة أدوار، كلٌ يدير جزءًا.

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

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

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

الواجهة الخلفية: FastAPI مع SQLite

قم بتثبيت الاعتماديات وتشغيل الخادم:

复制pip install fastapi uvicorn openai
export API_KEY=你的密钥
uvicorn server:app --reload --port 8000

ثم اكتب server.py. يحتوي الكود على أربع نقاط تستحق التوقف عندها:

复制# server.py
import os
import sqlite3
import uuid

from fastapi import FastAPI
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(base_url="https://api.apidamoxing.com/v1", api_key=os.environ["API_KEY"])

SYSTEM = {"role": "system", "content": "你是『小墨』，一个说话简洁、爱用比喻的聊天伙伴。回答控制在 200 字内。"}
KEEP = 20  # 每次只带最近 20 条消息

db = sqlite3.connect("chat.db", check_same_thread=False)
db.execute("create table if not exists msg(id integer primary key autoincrement, sid text, role text, content text)")

def load(sid, n=KEEP):
    rows = db.execute("select role, content from msg where sid=? order by id desc limit ?", (sid, n)).fetchall()
    return [{"role": r, "content": c} for r, c in reversed(rows)]

def save(sid, role, content):
    db.execute("insert into msg(sid, role, content) values (?,?,?)", (sid, role, content))
    db.commit()

class ChatIn(BaseModel):
    session_id: str
    message: str

@app.get("/")
def index():
    return FileResponse("index.html")

@app.post("/session")
def new_session():
    return {"session_id": uuid.uuid4().hex}

@app.post("/chat")
async def chat(body: ChatIn):
    save(body.session_id, "user", body.message[:4000])
    messages = [SYSTEM] + load(body.session_id)

    async def gen():
        parts = []
        try:
            stream = await client.chat.completions.create(
                model="uncensored", messages=messages,
                stream=True, max_tokens=800, temperature=0.8,
            )
            async for chunk in stream:
                if not chunk.choices:        # 末尾的 usage 块没有 choices
                    continue
                delta = chunk.choices[0].delta.content
                if delta:
                    parts.append(delta)
                    yield delta
        except Exception as e:
            yield f"\n[请求失败：{type(e).__name__}]"
        finally:
            if parts:
                save(body.session_id, "assistant", "".join(parts))

    return StreamingResponse(gen(), media_type="text/plain; charset=utf-8")

1. load() يستخرج أحدث 20 رسالة فقط لمنع تجاوز نافذة السياق البالغة 100,000 رمز.
2. gen() هو مولد غير متزامن (async generator)، حيث يُعيد كل قطعة صغيرة فور استلامها، مما يسمح للواجهة الأمامية بعرض النص حرفًا حرفًا.
3. يحتوي نهاية التدفق على كتلة بيانات تحتوي فقط على الاستخدام، ولا تحتوي على اختيارات، لذا يجب التحقق من أنها فارغة وتجاوزها.
4. يتم حفظ الرد الكامل في كتلة finally، مما يضمن عدم فقدان الجزء الذي تم إنشاؤه حتى في حالة حدوث خطأ.

إذا أردت عدم التعامل مع قواعد البيانات في البداية، يمكنك استبدال دالتي load وsave بقراءة وكتابة القاموس دون تغيير باقي الكود. وعلى العكس، عند زيادة عدد الزوار، يمكنك استبدال SQLite بأي قاعدة بيانات أخرى مألوفة لديك، مع الحفاظ على شكل هاتين الدالتين. ملاحظة: المثال يستخدم sqlite3 المتزامن، وهو سريع بما يكفي للنماذج الأولية؛ يُنظر في استخدام برامج التشغيل غير المتزامنة في سيناريوهات الحمل العالي.

الواجهة الأمامية: قراءة التدفق باستخدام fetch

لا يحتاج المتصفح إلى أي مكتبات. احصل على قارئ باستخدام resp.body.getReader()، ثم قم بحلقة على read(). احصل على كتلة بيانات، فك تشفيرها باستخدام TextDecoder، وأضفها إلى الصفحة. احفظ الكود التالي في index.html في نفس مجلد server.py:

复制<!doctype html>
<meta charset="utf-8">
<title>小墨</title>
<div id="gate">
  <label><input type="checkbox" id="adult"> 我已年满 18 周岁</label>
  <button id="enter">进入</button>
</div>
<div id="app" hidden>
  <div id="log" style="white-space:pre-wrap;min-height:300px"></div>
  <input id="box" placeholder="说点什么"> <button id="send">发送</button>
</div>
<script>
let sid = localStorage.getItem("sid");
const log = document.getElementById("log");

document.getElementById("enter").onclick = async () => {
  if (!document.getElementById("adult").checked) return;
  if (!sid) {
    const r = await fetch("/session", { method: "POST" });
    sid = (await r.json()).session_id;
    localStorage.setItem("sid", sid);
  }
  document.getElementById("gate").hidden = true;
  document.getElementById("app").hidden = false;
};

document.getElementById("send").onclick = async () => {
  const box = document.getElementById("box");
  const text = box.value.trim();
  if (!text) return;
  box.value = "";
  log.textContent += "\n你：" + text + "\n小墨：";
  const resp = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ session_id: sid, message: text }),
  });
  const reader = resp.body.getReader();
  const decoder = new TextDecoder("utf-8");
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    log.textContent += decoder.decode(value, { stream: true });
  }
  log.textContent += "\n";
};
</script>

تفاصيل مهمتان: أضف { stream: true } أثناء فك التشفير، لأن بايتات الحرف الواحد قد تُقسم بين كتلتين، وعدم إضافته سيؤدي إلى ظهور رموز عشوائية. ثانيًا، يجب تعطيل زر الإرسال أثناء تنفيذ الطلب لمنع النقر المتكرر من قبل المستخدم؛ لم يُضف المثال ذلك اختصارًا، يرجى إضافته عند النشر.

بعد التشغيل، افتح المتصفح على المنفذ 8000، أكد العمر، وأرسل رسالة. إذا ظهرت النصوص دفعة واحدة بدلاً من حرفًا حرفًا، فغالبًا ما يكون هناك وسيط يقوم بتخزين الاستجابة؛ جرب الاتصال المباشر بالمنفذ المحلي لاستبعاد ذلك. عند النشر خلف Nginx، يجب تعطيل تخزين الاستجابة لهذا المسار.

سجل المحادثات: لماذا يجب حفظه في واجهتك الخلفية

واجهة المحادثة "عديمة الحالة"، مثل موظف الاستقبال الذي ينظر فقط إلى الورقة التي تقدمها له. لذا، مسؤولية "تذكر السياق" تقع عليك، ويمكن تحقيق ذلك بثلاث مستويات:

1. الأبسط:يُخزَّن في الذاكرة فقط باستخدام قاموس. يُفقد عند إعادة التشغيل، وهو مناسب للاختبار.
2. الشائع:يُخزَّن في SQLite كما هو موضح في المثال، مع تخزين كل رسالة في سطر، والبحث عن آخر N رسالة حسب session_id.
3. المتقدم:عندما يصبح تاريخ المحادثات طويلاً جداً، اطلب من النموذج تلخيص الأجزاء الأقدم في ملخص واحد، وضعه بعد رسالة system، مع الاحتفاظ برسائل الفترة الأخيرة كنص كامل.

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

يمكن أن يكون شرط تفعيل طريقة الملخص بسيطًا: عندما يتجاوز عدد رموز السياق المقدّر عشرة آلاف، قم بتلخيص نصف الرسائل الأقدم للنموذج في فقرة لا تتجاوز 300 حرف، ثم اكتبها في المستودع واحذف الرسائل الأصلية. بهذه الطريقة تحافظ على السياق مع الحفاظ على حجم كل طلب ضمن نطاق قابل للتحكم. يجب الانتباه إلى أن الملخص نفسه يولّده النموذج وقد يفتقد بعض التفاصيل؛ لذا فإن الإعدادات المهمة (مثل اسم المستخدم أو المواضيع المحظورة) تناسب وضعها في نظام الرسائل الثابت بدلاً من الاعتماد على الملخص للحفاظ عليها.

المنتجات الموجهة للبالغين: نقاط الدخول والحدود

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

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

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

التعزيز والتوسعة قبل الإطلاق

• تقييد الإدخال:يقتصر المثال الحالي الرسالة الواحدة على 4000 حرف، ويمكنك تعديل ذلك حسب الحاجة.
• تقييد التكرار:300 طلب في الدقيقة لكل مفتاح API، مع تطبيق تقييد التكرار في الخادم الخلفي بناءً على المستخدم أو عنوان IP.
• عرض الأخطاء:التقط الخادم الخلفي الاستثناءات وأرسل تلميحات موجزة إلى الواجهة الأمامية، ولا تكشف عن تفاصيل تتبع الأخطاء (stack traces).
• مراقبة الاستخدام:لحساب عدد الرموز، يمكنك قراءة حقل usage في الطلبات غير المتدفقة، أو في آخر كتلة من التدفق. لمزيد من التفاصيل، راجعرموز (token) والفوترة.
• تغيير إطار العمل:عند استبدال الواجهة الأمامية بـ Vue أو React، يظل منطق قراءة التدفق كما هو. وعند استبدال الخادم الخلفي بـ Express، يكفي استبدال StreamingResponse بكتابة التدفق المناسبة.

لمزيد من التفاصيل حول المعاملات، راجعتفاصيل معاملات الواجهة البرمجية؛ وللمزيد من الأسئلة، راجعالأسئلة الشائعة.

قم بإجراء مراجعة ذاتية نهائية قبل الإطلاق: هل يتم تخزين مفتاح API فقط في متغيرات البيئة في الخادم؟ هل توجد حدود لطول الإدخال الفردي وتاريخ المحادثات؟ هل توجد رسائل خطأ ودودة؟ هل يؤدي مسح المحادثة إلى حذف السجلات فعلياً؟ هل يوجد تأكيد للعمر في نقطة الدخول؟ إذا استوفيت هذه الشروط الخمسة، يمكن اختبار النموذج الأولي مع المستخدمين الحقيقيين.