企业 AI 平台从 0 到 1（四）：做一个 AI 对话后端，从模型输出到用户屏幕这一段，我踩过的坑

前面三篇讲了怎么听懂问题、怎么捞证据、怎么合成答案。这一篇换个视角：假设你也要做一个类似的企业内部 AI 对话后端，从 “模型开始吐 token” 到 “用户屏幕上看到答案” 这一段，你会遇到哪些问题？每个问题通常有哪几种解法？我们项目在单台服务器、团队不大、并发不高的约束下是怎么选的？代码只是我们落地的样子，不是标准答案。

阅读前提：这是一篇经验复盘，不是教程。每节按 “问题 → 解法选项 → 我们的选择与代价 → 代码片段” 组织。第三篇讲了检索和有据合成（温度 0.0、证据门控、引用对齐），本篇不重复。

---

1. 推送协议怎么选：SSE、WebSocket、长轮询，各自什么时候用

问题：模型是一个 token 一个 token 吐的，你要把这个过程实时传给浏览器。

常见解法有三种：

方案	适合什么	代价
长轮询	老系统兼容、简单	延迟高、空请求多、服务端要挂起
WebSocket	双向交互：用户随时打断、插入指令、做协同	要自己写心跳、重连、粘包、鉴权；反代配置复杂；服务端要维护连接状态
SSE	服务端单向推送：聊天、日志流、进度条	只能服务端→客户端

我们的选择：SSE。 理由不是它 “更好”，而是我们当前场景就是单向的 —— 用户发一条消息，服务端流式回，用户输入走普通 POST。浏览器原生 EventSource 支持断线自动重连，走普通 HTTP 能复用现有鉴权中间件。如果以后要做 “生成过程中用户随时打断、改主意” 这种双向交互，再上 WebSocket，架构上把它隔离在传输层，业务不用动。

一个容易忘的部署细节：SSE 过 Nginx 默认会被缓冲，逐字效果变成 “一顿一顿”，必须显式关：

StreamingResponse(event_gen(), media_type="text/event-stream",
    headers={"Cache-Control": "no-cache",
             "Connection": "keep-alive",
             "X-Accel-Buffering": "no"})   # 告诉 Nginx 别缓冲

前端拿到的事件顺序，我们固定成这样，前后端都省心：

meta → status(正在分析) → step/thinking(阶段) → source_preview(来源卡片)
→ delta×N(逐字正文) → done(答案+来源+用量)

source_preview（命中的文档片段）特意放在 delta 之前 —— 用户看答案前先知道 “AI 引用了什么”，这是反幻觉体验的关键，不是随便排的。

---

2. 模型调用中途报错：为什么不抛异常

问题：流式请求过程中可能出各种错 —— 网关超时、模型名不对、鉴权失败、网络抖动。传统做法是抛异常让上层处理。

这里有个反直觉的点：在流式场景里，“抛异常” 比 “报错文本” 更糟糕。因为连接已经建立、已经吐了几个字了，这时候抛异常 = 连接中断 = 用户看到白屏或者 “连接已断开”。

我们的做法：错误也变成一段文本，流照常结束。 上层（SSE 层）只需要统一处理 “文本 chunk” 和 “结束事件”，错误路径和正常路径合流：

async def iter_stream_chunks(self, prompt, *, provider="", model="", messages=None, ...):
    parsed_provider, parsed_model = parse_model_ref(model)
    api_base, api_key, request_model = await self._resolve_runtime_config(...)

    try:
        client = self._get_client(api_base, api_key)
    except Exception as e:
        # 错误变成文本推出去，连接正常结束，而不是 500 中断
        yield LlmStreamChunk(text=f"LLM 客户端初始化失败 ({provider}/{model}): {e}")
        return

    stream = await self._chat_create_with_model_fallback(...)
    finish_reason = None                                   # 初始化，防空流
    async for chunk in stream:
        usage = getattr(chunk, "usage", None)
        if usage is not None:
            stream_prompt_tokens = int(getattr(usage, "prompt_tokens", 0) or 0)
            stream_completion_tokens = int(getattr(usage, "completion_tokens", 0) or 0)
        choices = getattr(chunk, "choices", None) or []   # 最后一个 usage chunk 的 choices 是空列表
        if not choices:
            continue
        choice0 = choices[0]
        piece = (getattr(choice0.delta, "content", None) or "")
        if piece:
            yield LlmStreamChunk(text=piece)
        reason = getattr(choice0, "finish_reason", None)
        if reason:
            finish_reason = reason
    if finish_reason:
        yield LlmStreamChunk(finish_reason=finish_reason)

两个踩过的坑：

• 流式必须带 stream_options={"include_usage": True}，否则 OpenAI 兼容网关默认不回 usage，token 统计全是 0；

• 开启 usage 后最后一个 chunk 是纯用量、choices 为空列表，直接 chunk.choices[0] 会 IndexError，必须判空。

经验：在用户已经 “看到你在工作” 的场景里，错误的体验优先级是 “不断流、能看懂” > “状态码正确”。后端工程师习惯抛异常，在这里是个思维转换。

---

3. 事件队列：为什么要有界、为什么终态不能丢

问题：生产端（检索、模型、工具）吐事件的速度，和 SSE 消费端（网络）把事件推给浏览器的速度，不匹配。中间隔一个队列。

三种做法：

1. 无界队列：内存会被慢消费拖死；

2. 满了直接丢：可能把 done 事件丢了，前端永远转圈；

3. 分级对待：中间帧可丢，终态必达。

我们选第三种。 队列容量 256，中间状态帧（thinking、delta）满了就丢，done/error 这种终态事件必须挤进去 —— 队列满了就挤掉最旧的中间帧腾位置：

class AgentStreamBus:
    def __init__(self):
        self._queue = asyncio.Queue(maxsize=256)
        self._closed = False

    def _force_enqueue(self, event):
        while True:
            try:
                self._queue.put_nowait(event)
                return
            except asyncio.QueueFull:
                with contextlib.suppress(asyncio.QueueEmpty):
                    self._queue.get_nowait()

    def push_nowait(self, event):
        if self._closed:
            return
        if str(event.get("event") or "") in {"done", "error"}:
            self._force_enqueue(event)      # 终态强制入队
            return
        try:
            self._queue.put_nowait(event)
        except asyncio.QueueFull:
            pass                             # 中间帧：丢了就丢了

经验：这种 “中间态尽力而为、终态必须达成” 的分级思想，在流式系统里是性价比最高的可靠性投资。别上来就做持久化队列、 Exactly-Once，你没那个规模。

---

4. 长答案被截断：三种做法，我们选了分段续写

问题：模型输出有 token 上限，长回答经常说到一半被 finish_reason="length" 掐断。这在知识库问答里尤其常见 —— 用户问 “总结一下这篇文档的架构”，答案天然就长。

业界常见做法：

1. 强制重生成：截断了就让模型从头再来一遍。浪费 token，而且下次还可能在同一个地方截断；

2. 分段独立生成：让模型自己列大纲，按段落分别生成。可控但破坏了行文连贯性；

3. 检测 + 续写：停了之后先判断是不是 “没说完”，没说完就把已生成内容喂回去，让它接着写。

我们选第三种。 关键是 “判断没说完” 不能只看 finish_reason——stop 是模型自己说 “我写完了”，但它经常写了半截代码块就 stop。程序要再查一遍文本形态：

def answer_looks_incomplete(text: str) -> bool:
    t = (text or "").rstrip()
    if not t:
        return False
    if _INCOMPLETE_MD_IMAGE_RE.search(t):     # 以不完整的 ![](... 结尾
        return True
    if t.count("```") % 2 == 1:               # 代码围栏开了没关
        return True
    tail = t[-120:]
    if re.search(r"\[[^\]]*$", tail):          # 链接/引用括号没闭合
        return True
    return False

续写主流程，段数封顶（默认 6）—— 既防死循环烧钱，也控制 messages 长度不无限膨胀：

async def iter_llm_stream_with_continuation(llm, *, messages=None, max_segments=6, ...):
    seg_cap = int(getattr(settings, "agent_synthesis_max_continuation_segments", 6) or 6)
    max_segments = min(max(1, max_segments), seg_cap)
    segment_text = ""

    for seg_idx in range(max_segments):
        async for chunk in llm.iter_stream_chunks(messages=messages, **run_kwargs):
            if chunk.text:
                segment_text += chunk.text
                yield chunk.text                          # 分段内容直接往下推，用户无感
            if chunk.finish_reason:
                finish_reason = chunk.finish_reason

        if not should_continue_synthesis(segment_text, finish_reason,
                                          segment_index=seg_idx, max_segments=max_segments):
            break
        # 已生成内容作为 assistant 消息回喂，让模型接着写而不是重写
        messages.append({"role": "assistant", "content": segment_text})
        messages.append({"role": "user", "content": "请基于上文继续输出剩余内容，不要重复已有部分。"})

判定函数：

def should_continue_synthesis(segment_text, finish_reason, *, segment_index, max_segments):
    if segment_index >= max_segments - 1:     # 到段数上限了，别续了
        return False
    if not (segment_text or "").strip():
        return False
    if finish_reason == "length":            # 被长度上限掐的，必续
        return True
    if finish_reason in (None, "stop") and answer_looks_incomplete(segment_text):
        return True                          # 自己说写完了但文本是残的，续
    return False

经验：续写时证据上下文已经在 messages 里，不用重新检索 —— 这是 RAG 链路的便利。但要警惕 messages 越攒越长，段数封顶是必须的，别让 “防截断” 变成 “烧钱循环”。

---

5. token 用量：为什么要三层落库，而不是记一笔

问题：一次对话不是一次模型调用。Query 改写一次、合成一次、长答案续写好几次、工具循环又好几次。token 成本是真金白银，你得知道哪次对话、哪个 Agent、哪个模型花了多少。

常见做法：

• 只在消息表记个总数：事后没法归因；

• 每次调用记一笔明细：能查，但前端展示要聚合；

• 上下文累加 + 消息表总数 + 明细 + trace：用户看得见、后台能查、出问题能串起来。

我们选第三种。 用量用 ContextVar 在请求级别累加，异步落库：

_llm_usage_ctx: ContextVar = ContextVar("llm_usage_ctx", default=None)

def add_pending_llm_usage_call(*, prompt_tokens, completion_tokens, model=""):
    """单次 LLM 调用的 usage 累加到当前请求上下文"""
    ...

单次调用 usage → 累加到请求上下文
→ 对话消息表写入（用户能看到"本次回答消耗 X tokens"）
→ llm_usage_logs 明细表逐条写（哪条消息、哪个模型、多少 token）
→ 同时汇总进 trace（排查"这次对话为什么这么贵"时一条链路看全）

一个兜底问题：部分网关流式不回 usage，本地粗估。注意这只是 “不让用量为空” 的兜底，不是精确值；真实 tokenizer 下中文通常接近 1 字 1 token，有 usage 时一律以网关返回为准：

def estimate_tokens_from_text(text: str) -> int:
    cjk = sum(1 for ch in text if 0x4E00 <= ord(ch) <= 0x9FFF)
    other = len(text) - cjk
    return max(1, int(cjk / 1.5 + other / 4.0))

经验：用量系统不要等 “要算账了” 才做。第一天就把明细落上，成本归因是后期优化模型选型、Agent 配置的依据。

---

6. 让模型调工具：为什么必须有界

问题：不带知识库时，模型要自己决定查数据库、跑代码、搜网页。这就是 ReAct：想一步、调一个工具、看结果、再想下一步。

风险很明显：模型可能陷入 “反复调工具” 的死循环，或者调出名单外的工具名。

我们的做法：

1. 轮数有界：默认 4 轮，硬上限 6。少于 3 轮复杂任务做不完，超过 6 轮基本是在原地打转烧钱；

2. 到上限强制收尾：追加一条 “别再调工具了，直接给答案”；

3. 工具选择结果过白名单：模型说要调什么不算数，必须在候选名单里。

iteration = 0
max_iterations = min(max(1, int(cfg.max_iterations or 4)), 6)
while iteration < max_iterations:
    iteration += 1
    async for item in self._stream_chat_completion(
            client=client, messages=messages,
            tool_schemas=tool_schemas or None,
            tool_choice="auto" if tool_schemas else None):
        if isinstance(item, CompletionRoundResult):
            round_result = item
            continue
        yield item

    tool_calls = round_result.tool_calls
    if not tool_calls:
        break                              # 模型没调工具 → 这就是最终答案

    messages.append({
        "role": "assistant", "content": round_result.content or "",
        "tool_calls": [{"id": tc.id, "type": "function",
                        "function": {"name": tc.function.name,
                                     "arguments": tc.function.arguments or "{}"}}
                       for tc in tool_calls],
    })
    async for evt in self._execute_tool_calls_parallel(tool_calls=tool_calls, ...):
        yield evt

工具选择器还有个细节：一个助手开 6-7 个工具时，每次把所有工具的 JSON schema 全塞给模型，既费 token 又容易选错。做法是先让模型从候选里挑本轮要用的子集，再进主循环。但返回值必须过白名单：

def _normalize_selection(data, candidates):
    allowed = {str(item.get("name")) for item in candidates if item.get("kind") == "tool"}
    tools = [s for s in (data.get("selected_tools") or []) if s in allowed]
    ...

经验：让模型做选择（“用 AI 做路由”）是省事的，但程序必须兜底—— 模型输出永远是不可信输入，白名单、轮数上限、强制收尾，这三个是 ReAct 系统的保险丝。

---

7. 让模型跑代码：安全分级思路

问题：让模型生成 Python 做计算（数据分析、公式求解）是很自然的需求，但直接 exec 是找死。

业界的安全分级：

级别	手段	防什么	适合
脚本沙箱	AST 扫描、禁危险函数、子进程、超时	误操作、低风险代码	内部可信用户的日常计算
进程隔离	独立用户、cgroup 限内存 CPU、禁网	中等风险	多租户、不可信代码
容器 / 微虚拟机	每个代码块一个容器、只读文件系统	真恶意	公网开放、第三方代码

我们选第一级。 AST 静态扫描 + 子进程 + 8 秒超时 + 输出超过 16000 字符截断：

_FORBIDDEN_NAMES = {"eval", "exec", "__import__", "compile", "open", "input"}
_FORBIDDEN_DUNDERS = {"__globals__", "__builtins__", "__subclasses__"}

def _static_scan(code, denylist, allowlist):
    try:
        tree = ast.parse(code, mode="exec")
    except SyntaxError as e:
        return f"代码存在语法错误：{e.msg} (line {e.lineno})"

    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            root = node.names[0].name.split(".")[0]
            if root in denylist:
                return f"禁止导入模块：{node.names[0].name}"
            if allowlist and root not in allowlist:
                return f"未在白名单内的模块：{node.names[0].name}"
        elif isinstance(node, ast.Call):
            name = ""
            if isinstance(node.func, ast.Name):
                name = node.func.id
            elif isinstance(node.func, ast.Attribute):
                name = node.func.attr
            if name in _FORBIDDEN_NAMES:
                return f"禁止调用 {name}"
        elif isinstance(node, ast.Attribute) and node.attr in _FORBIDDEN_DUNDERS:
            return f"禁止访问 {node.attr}"
    return None

经验：别上来就上容器。我们是单台服务器、内部可信用户，脚本沙箱够用，成本和复杂度都可控。但要把 “防到哪一层” 写清楚 ——过度宣称安全比承认有限更危险。真要跑外部不可信代码，升容器，别在脚本沙箱上堆补丁。

---

8. 模型路由与别名回退：一种 “防呆” 设计

问题：模型不是硬编码的。我们用 provider:model 引用，配置存数据库。但模型升级后，配置里写的请求名可能变成别名（服务商改了模型 ID），直接请求会报 “模型不存在”。

做法：自动用原生名重试一次，但只对 “模型名类错误” 回退—— 参数错、鉴权错不重试，重试也没用：

async def _chat_create_with_model_fallback(self, client, *, model, request_model, ...):
    kwargs = {"model": request_model, "messages": messages, "stream": stream, ...}
    if stream:
        kwargs["stream_options"] = {"include_usage": True}
    try:
        return await client.chat.completions.create(**kwargs)
    except Exception as exc:
        if request_model != model and self._should_fallback_model_alias(exc):
            kwargs["model"] = model
            return await client.chat.completions.create(**kwargs)
        raise

经验：这种 “防呆” 设计看着小，但生产环境里模型 ID 变更、网关别名、配置打错字这种事每天都在发生。关键是有选择地重试—— 不是所有错误都值得重试，盲目重试会放大故障。

---

9. 配置自由度：能力做成配置，边界在哪

问题：平台能力越来越多 ——Agent 要能配模型、温度、工具集；新技能要能加；提示词要能改。是写死在代码里，还是做成配置？

我们的原则：自由度向上开放给运营，护栏向下握在代码里。

• Agent 是配置的：模型名、温度、top_p、max_tokens、系统提示词、启用哪些工具，界面上配。同一个执行引擎，配出不同的助手；

• 工具开关：勾选即用，mcp_ 前缀的外部 MCP 工具勾上就能接；

• 技能自动激活：每个技能带 slug/name/description/ 是否可自动激活，对话时按问题匹配，命中的技能指令注入合成提示词；

• 提示词配置化：有据作答规则这类文案存系统配置表，后台改了生效，不用发版。

技能选择算法的一个细节 —— 相对阈值，避免一次激活一堆：

def _select_profile_skill_slugs(catalog, question, explicit_skill_slugs):
    requested = {slug for slug in explicit_skill_slugs}
    invoked = {m.group(1) for m in _EXPLICIT_SKILL_INVOCATION.finditer(question)}
    for entry in entries:
        if entry.slug in requested or entry.slug in invoked or _is_direct_skill_mention(question, entry):
            selected.add(entry.slug)
            continue
        if not getattr(entry, "auto_activatable", False):
            continue
        score = _skill_metadata_relevance_score(question, entry)
        if score > 0:
            auto_scores.append((score, entry.slug))
    if auto_scores:
        max_score = max(s for s, _ in auto_scores)
        floor = max(2, max_score // 2)
        selected |= {slug for s, slug in auto_scores if s >= floor}
    return selected

但配置不是失控：温度夹到 [0,1]、工具和技能选择结果过白名单、Agent 绑了知识库就强制走 RAG 路由。自由度向上开，护栏向下握—— 这句话是我们做配置化时反复对自己说的。

---

10. 踩坑清单（可以当上线检查表用）

坑	解法
流式请求忘了 include_usage	token 统计全 0，必须显式带
直接 chunk.choices[0]	最后一个 usage chunk 的 choices 是空列表，必须判空；finish_reason 先初始化为 None
队列满了直接丢事件	done/error 走强制入队，中间帧才丢
只信 finish_reason=stop	叠加文本残缺检测（奇数 ```、未闭合括号等）
续写让模型重写	已生成内容作为 assistant 消息回喂，让它接着写
续写不设段数上限	封顶 6 段，防死循环烧钱、messages 膨胀
模型输出名单外的工具名	选择结果过白名单过滤
ReAct 不设轮数上限	默认 4 轮、硬上限 6，到上限强制收尾
直接 exec 模型代码	AST 扫描 + 子进程 + 8s 超时 + 16000 字符输出截断
异常在流式中途抛出	错误转成文本 chunk，流正常结束
中文 token 按字符估	粗估兜底，有 usage 一律以网关返回为准
别名模型报错直接 500	只对模型名类错误回退原生名重试一次
Nginx 缓冲 SSE	响应头加 X-Accel-Buffering: no
没绑 KB 却走了 RAG	路由判断：绑 KB 才走检索链路，否则走通用助手

---

11. 小结

这一篇的核心不是 “我们怎么做的”，而是做这类系统时的几个判断：

1. 流式场景里，不断流比状态码正确更重要—— 错误也变成文本；

2. 可靠性投资要分级—— 中间态尽力而为，终态必须达成，别上来就上 Exactly-Once；

3. 模型做选择，程序做兜底—— 白名单、轮数上限、强制收尾，一个都不能少；

4. 安全要分级—— 脚本沙箱、进程隔离、容器，按你的用户信任度选，别过度宣称；

5. 自由度要配护栏—— 配置化让运营能扩展能力，但参数 clamp、白名单、路由约束必须握在代码里。

下一期讲 Agent 与工作流编排：多步任务怎么串、可视化工作流怎么跑、技能怎么沉淀复用。