สร้างแชทบอทด้วย API โมเดลใหญ่: แบ็กเอนด์ FastAPI พร้อมสตรีมมิง

วาดแผนภาพเล็กๆ ก่อน: แต่ละฝ่ายรับผิดชอบงานของตัวเอง

คิดแชทบอทเหมือนร้านส่งอาหาร: เบราว์เซอร์คือลูกค้า แบ็กเอนด์ของคุณคือพนักงานต้อนรับ และ API โมเดลคือครัว ลูกค้าไม่เข้าครัวโดยตรง ทุกคำสั่งซื้อผ่านพนักงานต้อนรับ นี่คือเหตุผลที่คีย์ API ไม่ควรวางที่หน้าเว็บ

การไหลของหนึ่งการสนทนาคือ: หน้าเว็บส่งข้อความหนึ่งบรรทัด แบ็กเอนด์เก็บไว้ ดึงประวัติล่าสุดมา รวมกับ system ส่งไปยัง API รับข้อมูลแล้วส่งต่อให้หน้าเว็บทันที และเก็บข้อความตอบกลับทั้งหมดไว้เมื่อเสร็จสิ้น

ทำไมให้แบ็กเอนด์ส่งต่อแทนที่จะให้หน้าเว็บเชื่อมต่อโดยตรง? นอกจากความปลอดภัยของคีย์ API แล้ว ยังมีเหตุผลสองประการ: หนึ่งคือคุณต้องเพิ่มประวัติ กำหนดขีดจำกัดอัตรา และกรองข้อมูลระหว่างการส่งต่อ ซึ่งทำได้เฉพาะที่แบ็กเอนด์ สองคือเมื่อเปลี่ยนโมเดลหรือบริการในอนาคต คุณ只需แก้ไขแบ็กเอนด์เพียงจุดเดียว โดยหน้าเว็บไม่ต้องเปลี่ยนแปลงเลย

แบ็กเอนด์: FastAPI พร้อม SQLite

ติดตั้ง dependencies และเริ่มทำงานก่อน:

复制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 ซึ่งส่งค่า yield ออกมาทีละส่วนเมื่อได้รับข้อมูล ทำให้ฝั่งไคลเอนต์สามารถแสดงผลตัวอักษรได้ทีละตัว
3. ที่ท้ายสตรีมจะมี data block ที่บรรจุเฉพาะข้อมูลการใช้งาน ซึ่งไม่มี choices ดังนั้นจึงต้องตรวจสอบค่าว่างแล้วข้ามไป
4. ใน finally จะบันทึกข้อความตอบกลับทั้งหมด แม้เกิดข้อผิดพลาดระหว่างทาง ข้อความที่สร้างไว้แล้วจะไม่สูญหาย

หากคุณต้องการหลีกเลี่ยงการเข้าถึงฐานข้อมูล คุณสามารถเปลี่ยนการ load และ save ให้เป็นการอ่านเขียน dict แทนได้ โดยไม่ต้องแก้ไขโค้ดส่วนอื่น หากปริมาณการใช้งานเพิ่มขึ้น คุณสามารถเปลี่ยน SQLite เป็นฐานข้อมูลใดก็ได้ที่คุณคุ้นเคย โดยรักษาโครงสร้างของฟังก์ชันทั้งสองไว้เหมือนเดิม นอกจากนี้ ตัวอย่างในโค้ดใช้ sqlite3 แบบ synchronous ซึ่งเขียนข้อมูลได้เร็วและเพียงพอสำหรับโปรโตไทป์ แต่ในสถานการณ์ที่มี parallel requests สูง ควรพิจารณาใช้ driver แบบ async

หน้าเว็บ: อ่านสตรีมมิงด้วย fetch

เบราว์เซอร์ไม่ต้องใช้ library ใดๆ resp.body.getReader() ได้ reader มา แล้ว loop read() รับข้อมูลแบบไบต์ทีละส่วน แล้วใช้ TextDecoder decode แล้ว append ลงหน้าเว็บ บันทึกโค้ดด้านล่างเป็น 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 ต้องปิดการบัฟเฟอร์การตอบสนองสำหรับเส้นทางนี้

ประวัติการสนทนา: ทำไมต้องเก็บในแบ็กเอนด์ของคุณ

API การสนทนาเป็นแบบ stateless เหมือนพนักงานต้อนรับที่ดูเฉพาะกระดาษที่คุณยื่นให้: มีอะไรในกระดาษ เขาก็รู้แค่นั้น ดังนั้นการจำบริบทคือหน้าที่ของคุณ โดยมีวิธีการสามระดับ:

1. แบบง่ายที่สุด:เก็บในหน่วยความจำโดยใช้พจนานุกรม เมื่อรีสตาร์ทข้อมูลจะหายไป เหมาะสำหรับขั้นตอนการดีบัก
2. แบบทั่วไป:จัดเก็บแบบ SQLite ตามตัวอย่าง โดยบันทึกข้อความแต่ละรายการเป็นแถว และค้นหาข้อความล่าสุด N รายการตาม session_id
3. ขั้นสูง:เมื่อประวัติยาวเกินไป ให้ใช้โมเดลสรุปเนื้อหาเก่าให้เป็นบทสรุป แล้ววางไว้หลัง system message โดยยังคงเก็บข้อความล่าสุดเป็นข้อความต้นฉบับไว้

ข้อดีอีกอย่างของการเก็บใน backend ของคุณคือคุณควบคุมอายุการเก็บข้อมูลได้ แนะนำให้มีปุ่ม "ล้างการสนทนา" เพื่อลบ session นั้นจริงๆ และระบุในนโยบายความเป็นส่วนตัวว่าเก็บข้อมูลอะไรไว้

เงื่อนไขการเรียกใช้วิธีสรุปอาจเรียบง่ายได้: เมื่อจำนวนโทเคนในประวัติเกิน 10,000 ให้ส่งข้อความครึ่งแรกให้โมเดลสรุปเป็นข้อความไม่เกิน 300 คำ แล้วบันทึกกลับลงที่เก็บพร้อมลบข้อความต้นฉบับ วิธีนี้ช่วยรักษาความต่อเนื่องของบทสนทนา และทำให้ปริมาณข้อมูลในแต่ละคำขออยู่ในขอบเขตที่ควบคุมได้ ควรทราบว่าบทสรุปเกิดจากโมเดลจึงอาจมีรายละเอียดที่ตกหล่น การตั้งค่าสำคัญเช่น ชื่อเล่นของผู้ใช้ หรือหัวข้อที่ห้ามพูด ควรกำหนดไว้ใน system message แทนที่จะพึ่งพาบทสรุป

ผลิตภัณฑ์สำหรับผู้ใหญ่: จุดเข้าและขอบเขต

บริการนี้สำหรับผู้ใหญ่อายุ 18 ปีขึ้นไป แอปของคุณก็ควรทำเช่นเดียวกัน ตัวอย่าง frontend มี entrance ยืนยันอายุแบบง่ายที่สุด สำหรับผลิตภัณฑ์ที่ต้องการความเข้มงวดสูงอาจเพิ่มขั้นตอนการยืนยันอายุที่สมบูรณ์ขึ้น ข้อแนะนำในการปฏิบัติ:

• หน้าจุดเข้าใช้งานระบุข้อจำกัดอายุอย่างชัดเจน ไม่แสดงหน้าแชทจนกว่าจะยืนยันแล้ว
• อย่าโปรโมตผลิตภัณฑ์ไปยังกลุ่มนักเรียนหรือเด็ก และอย่าใส่ไว้ในบริบทที่มุ่งเน้นกลุ่มเหล่านี้
• เนื้อหาทางเพศที่เกี่ยวข้องกับเด็กไม่ว่าจะเป็นจริงหรือไม่จะถูก interface intercept และส่ง 403 กลับมา backend ของคุณต้องตรวจจับสถานะนี้และแสดงข้อความแจ้งเตือนผู้ใช้ที่สุภาพ ไม่ใช่แสดง error ดิบ

จากการออกแบบผลิตภัณฑ์ ผู้ใช้สามารถกำหนดชื่อเล่นและรูปแบบการพูดในหน้าตั้งค่าได้ การตั้งค่าเหล่านี้เขียนลงใน system message ได้ทันที ทำให้บทสนทนาเป็นส่วนตัวมากขึ้น โดยไม่ต้องให้โมเดลเดา

การเสริมความมั่นคงและขยายขีดความสามารถก่อนเปิดใช้งาน

• จำกัดข้อมูลนำเข้า:ตัวอย่างตัดข้อความแต่ละรายการไว้ที่ 4,000 ตัวอักษร ปรับตามความต้องการ
• จำกัดความถี่:จำกัด key แต่ละคีย์ไว้ที่ 300 คำขอต่อนาที โดยทำระบบจำกัดอัตราที่แบ็กเอนด์ก่อน โดยแบ่งตามผู้ใช้หรือ IP
• การแสดงข้อผิดพลาด:แบ็กเอนด์ดักจับข้อยกเว้นและส่งข้อความสรุปสั้นๆ กลับไปยังหน้าเว็บ อย่าเปิดเผย stack trace
• การตรวจสอบการใช้งาน: หากต้องการนับโทเคน ให้อ่านค่า usage จากคำขอแบบไม่สตรีม หรืออ่านจากบล็อกสุดท้ายของสตรีม ดูรายละเอียดในโทเคนและการคิดเงิน
• เปลี่ยนเฟรมเวิร์ก:เมื่อเปลี่ยนหน้าเว็บเป็น Vue หรือ React การอ่านสตรีมทำงานเหมือนกันทุกประการ การเปลี่ยนแบ็กเอนด์เป็น Express เพียงเปลี่ยน StreamingResponse เป็นรูปแบบการเขียนสตรีมที่สอดคล้องกัน

เมื่อต้องการรายละเอียดพารามิเตอร์ที่สมบูรณ์ยิ่งขึ้น ดูรายละเอียดพารามิเตอร์ API; คำถามเพิ่มเติมดูที่คำถามที่พบบ่อย

ตรวจสอบรายการก่อนเปิดใช้งานอีกครั้ง: คีย์ API ถูกเก็บไว้ในตัวแปรสภาพแวดล้อมของเซิร์ฟเวอร์เท่านั้นหรือไม่; มีการจำกัดความยาวของข้อมูลนำเข้าและประวัติหรือไม่; ข้อผิดพลาดมีข้อความแจ้งเตือนที่เป็นมิตรหรือไม่; ปุ่มล้างการสนทนาลบบันทึกจริงๆ หรือไม่; การยืนยันอายุอยู่ที่จุดเข้าใช้งานหรือไม่ หากครบทั้ง 5 ข้อนี้ โปรโตไทป์นี้พร้อมให้ผู้ใช้จริงทดลองใช้แล้ว