用大模型 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 字符，按需调整。
• 限制频率：每把 key 每分钟 300 次请求，按用户或 IP 在后端先做限流。
• 错误展示：后端捕获异常，向前端返回简短提示，不要把堆栈暴露出去。
• 用量监控：想统计 token，可在非流式请求里读 usage，或在流的最后一个块里读取，细节见token 与计费。
• 换框架：前端换成 Vue 或 React 时，读流的逻辑一模一样。后端换成 Express 也只是把 StreamingResponse 换成相应的流式写法。

需要更完整的参数说明时，看接口参数详解；更多问题见常见问题。

最后做一次上线前的自查：密钥是否只在服务端环境变量里；单条输入和历史长度是否有上限；错误是否有友好提示；清空对话是否真的删除记录；年龄确认是否在入口处。五项都满足，这个原型就可以拿给真实用户试用了。