使用大型模型 API 建構聊天機器人：FastAPI 後端搭配串流前端

先畫一張小圖：三個角色各管一攤

把聊天機器人想成一家外送店：瀏覽器是顧客，你的後端是櫃檯，大型模型介面是後廚。顧客從不直接進後廚，所有訂單都由櫃檯轉交，這就是金鑰為什麼不能放在前端的原因。

一次對話的流轉是：網頁傳送一句話，後端儲存下來，取出最近歷史，拼上 system，請求介面，邊收邊轉發給網頁，收完後把完整回覆也儲存起來。

為什麼選擇讓後端轉發，而不是前端直連？除了金鑰安全，還有兩個現實原因：一是你需要在轉發時加入歷史、做速率限制和過濾，這些都只能在後端做；二是將來換模型或換服務時，只需要改後端一處，前端完全不用動。

後端：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 token 的上下文視窗。
2. gen() 為非同步生成器，每收到一小段內容便 yield 出去，前端因此能逐字顯示。
3. 串流末尾會有一個只含用量的資料區塊，它沒有 choices，所以要判空跳過。
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，每則訊息佔一行，依 session_id 查詢最近 N 則。
3. 進階：當歷史過長時，將較早的內容交由模型摘要成一段摘要，置於 system 之後，近期訊息仍保留原文。

儲存於自有後端的另一個好處是你完全掌控保留時長。建議提供「清空對話」按鈕，真正刪除對應 session 的記錄，並在隱私權聲明中清楚說明你儲存了什麼。

摘要法的觸發條件可以很簡單：當估算的歷史 token 超過一萬，就把最早的一半訊息交給模型摘要成不超過 300 字的一段，寫回儲存並刪除原始訊息。這樣既保留了脈絡，又能讓每次請求的體量穩定在可控範圍內。要注意的是，摘要本身也是模型產生的，可能遺漏細節，重要的設定（例如使用者的暱稱、禁忌話題）更適合放在 system 裡固定下來，而不是依賴摘要保留。

面向成年人的產品：入口與邊界

這項服務僅面向 18 歲及以上的成年使用者，你的應用程式也應如此。範例前端放了一個最簡單的確認入口，更嚴格的產品可以加上更完整的年齡驗證流程。幾項實務建議：

• 入口頁明確提示年齡限制，未確認不顯示聊天介面。
• 不要把產品推廣給學生或未成年人群體，也不要放入面向他們的場景。
• 涉及未成年人的性內容無論是否虛構都會被介面攔截並返回 403，你的後端要能識別這個狀態並給使用者一句友善的提示，而不是顯示原始錯誤。

從產品設計上，也可以讓使用者在設定裡自行設定一個暱稱和語氣偏好，這些寫進 system 即可，既讓對話更個人化，也不必讓模型去猜。

上線前的加固與擴充

• 限制輸入：範例已將單則訊息截斷至 4000 字元，可依需求調整。
• 限制頻率：每支金鑰每分鐘 300 次請求，依使用者或 IP 於後端先進行限流。
• 錯誤顯示：後端捕獲異常，向前端返回簡短提示，不要把堆疊追蹤（stack trace）暴露出去。
• 用量監控：想統計 token，可在非串流請求中讀取 usage，或在串流的最後一個區塊中讀取，細節見token 與計費。
• 換框架：前端換成 Vue 或 React 時，讀取串流的邏輯一模一樣。後端換成 Express 也只是把 StreamingResponse 換成相應的串流寫法。

需要更完整的參數說明時，看介面參數詳解；更多問題見常見問題。

最後做一次上線前的自查：金鑰是否僅在伺服器環境變數中；單則輸入和歷史長度是否有上限；錯誤是否有友善提示；清空對話是否真的刪除記錄；年齡確認是否在入口處。五項都滿足，這個原型就可以提供給真實使用者試用。