跳到主要内容

数据截至 (上游 commit 0de51d76e1dc)

Chat UI — 流式生成管线

本章讲主线:用户点“发送”之后,服务端到底发生了什么,token 怎么一边流给浏览器一边写回数据库。读完你能讲清 Chat UI 最核心的那条路径。

1. 它要解决的小问题​

聊天产品有三个相互冲突的诉求,要在一条流里同时满足:

  • 要边生成边显示(打字机效果),不能等模型说完才一次性返回。
  • 要能随时停,而且停下来之后历史里存的是“用户最后看到的样子”。
  • 生成期间不能断线——有些推理模型首 token 要等很久,中间得有心跳。

Chat UI 的答案是:把一切都建模成一串 MessageUpdate 事件,用 async generator 流式产出,再用一个 ReadableStream 把它们逐行序列化给浏览器。

2. MessageUpdate:统一事件协议​

整条管线里流动的不是裸字符串,而是带类型的事件对象。这是理解一切的钥匙。

事件类型(src/lib/types/MessageUpdate.ts:21 MessageUpdateType):

类型含义
Status状态:Started / Finished / Error / KeepAlive(心跳)
Stream一个内容 token(正文)
Reasoning推理过程(子类型:Stream 流式思考 / Status 思考状态)
Tool工具调用(子类型:Call/Result/Error/ETA/Progress)
FinalAnswer最终答案(带 interrupted 标志)
Title对话标题(后台异步生成的)
RouterMetadata路由/provider 元信息(“via together”之类)
File生成的文件(如工具产出的图片)

为什么这样设计: 生成、推理、工具、标题本来是异构的过程,统一成一个可辨识联合类型后,(a)服务端写回逻辑只要一个 switch(+server.ts:435 的 update() 函数),(b)浏览器解析也只要一个 switch,(c)这些事件还能原样存进 message.updates[] 供回放。

3. 三条流合并:编排器​

textGeneration() 是顶层编排器。它的精妙在于同时跑三件事并合并成一条流:

textGeneration(ctx)
   │
   ├── titleGen      —— 后台生成对话标题(仅"New Chat"时)
   ├── textGen       —— 主生成(工具流 或 普通生成)
   └── keepAliveGen  —— 每 100ms 发一个 KeepAlive 心跳,直到主生成完成
   │
   └─▶ mergeAsyncGenerators([...])  —— 谁先有值就先 yield 谁

真实代码很短(textGeneration/index.ts:35):

export async function* textGeneration(ctx: TextGenerationContext) {
	const done = new AbortController();
	const titleGen = generateTitleForConversation(ctx.conv, ctx.locals);
	const textGen = textGenerationWithoutTitle(ctx, done);
	const keepAliveGen = keepAlive(done.signal);
	// keep alive until textGen is done
	yield* mergeAsyncGenerators([titleGen, textGen, keepAliveGen]);
}

心跳的实现也直白(index.ts:25 keepAlive):一个 while (!done.aborted) 循环,每 100ms yield 一个 KeepAlive 状态。主生成结束时 done.abort()(index.ts:122),心跳随之停。心跳事件不会被持久化——写回时被显式过滤掉(+server.ts:374)。心跳事件不会被持久化——写回时被显式过滤掉(+server.ts:374)。

主生成的分叉(index.ts:71): 先 runMcpFlow(...)。它的返回值决定后续:

runMcpFlow() 返回
   ├── "completed"      —— 工具流已完整生成答案,结束
   ├── "aborted"        —— 用户中断,结束(不回退)
   ├── "exhausted"      —— 工具轮数耗尽,已发一段收尾答案,结束(不回退)
   └── "not_applicable" —— 没工具/模型不支持(且尚未输出任何内容)→ 回退到 generate()

注意容错的细节:如果工具流抛异常,只要不是中断错误、且这轮还没给用户显示过任何输出(mcpProducedOutput 为 false),就回退到普通 generate()(index.ts:104-121);已经显示过就重新抛出,避免丢弃用户已看到的工具过程——也就是说“工具坏了”不会让用户拿不到回复,但也不会把已展示的内容作废重来。

4. generate():消费 token 流​

普通生成路径(generate.ts:16)做的事:

  1. 调 endpoint({...}) 拿到一个统一格式的 token 流(TextGenerationStreamOutput)。
  2. for await 逐个消费:
    • 是 routerMetadata → 转成 RouterMetadata 事件。
    • 是普通 token → 包成 Stream 事件 yield(generate.ts:244)。
    • 处于 reasoning 模式 → 包成 Reasoning/Stream 事件(第 3 章细讲)。
    • 是 generated_text(终止)→ 处理 stop token、推理裁剪,最后 yield FinalAnswer(generate.ts:160)。
  3. 每个 token 之间检查中断:查 AbortedGenerations 缓存,若有比本次更晚的停止时间就 abort() 并 break(generate.ts:248)。

5. 边流边写回:update() 闭包​

生成出来的事件去哪了?HTTP 处理器里有个 ReadableStream,它的 start(controller) 里跑着主循环(+server.ts:674):

// 极简示意,非源码
for await (const event of textGeneration(ctx)) {
  await update(event);   // ← 关键:一个事件既写回内存消息、又 enqueue 给浏览器
}

update()(+server.ts:435)对每种事件做两件事:

  • 累加到内存里的 assistant 消息:Stream → content += token;Reasoning/Stream → reasoning += token;Title → 立刻单独写库;FinalAnswer → 定稿 content 和 interrupted。
  • 序列化发给浏览器:controller.enqueue(JSON.stringify(event) + "\n")(+server.ts:587),所以响应是 application/jsonl——一行一个 JSON 事件。

两个不显眼但聪明的细节:

  • 侧信道防护:Stream 事件发出前,token 被 padEnd(16, "\0") 补成定长(+server.ts:579)。这是为了防一种“靠观察网络包长度反推内容”的 LLM 侧信道攻击(代码里直接贴了论文链接)。
  • 客户端断开:enqueue 抛错就标记 clientDetached,转去 persistConversation() 把已生成的内容存库(+server.ts:591),用户回来还能看到。

6. 消息是一棵树,不是一个数组​

你可能以为对话就是 messages: []。实际上 Chat UI 把它存成树:每条消息有 ancestors[](到根的路径)和 children[](src/lib/types/Message.ts)。

为什么?为了分支重试。当你编辑一条用户消息重发,它不是覆盖,而是给那条消息加一个兄弟(addSibling),再在兄弟下挂一条空 assistant 消息(+server.ts:283)。原分支保留,你可以来回切换。

构建 prompt 时,buildSubtree(conv, id)(tree/buildSubtree.ts:3)沿 ancestors 把从根到目标的那一条线性路径摘出来,送给模型——模型永远只看到一条干净的对话线,看不到树。

        root(user)
          │
     assistant-A
          │
   ┌──────┴───────┐
 user-edit1    user-edit2   ← 编辑产生兄弟分支
     │             │
 assistant-B   assistant-C

7. 巧妙之处​

  • 一切皆事件:把生成/推理/工具/标题/路由都收敛成 MessageUpdate 联合类型,使“流式传输”、“持久化”、“回放”三件事共用一套数据(MessageUpdate.ts:10)。
  • 三流合并:用 mergeAsyncGenerators 把心跳、标题、主生成织成一条流,标题生成不阻塞正文(index.ts:44)。
  • 工具流失败自动降级:工具坏了且尚未输出任何内容时退回普通生成,用户无感(index.ts:119)。
  • 定长 padding 防侧信道:对抗按包长度推测 token 内容的攻击(+server.ts:579)。

8. 边界与局限​

  • 这个 build 只支持 OpenAI 兼容 + 流式;completion(非 chat)路径虽在 endpointOai.ts:118 保留,但 models.ts 装配的 endpoint 永远是 chat 类型。
  • 标题生成有个 HACK:靠 conv.title !== "New Chat" 判断“是不是新会话”(title.ts:14),改了这个魔法字符串会失效。

代码地图​

主题文件符号
编排 / 心跳src/lib/server/textGeneration/index.tstextGeneration, keepAlive, textGenerationWithoutTitle
普通生成src/lib/server/textGeneration/generate.tsgenerate
写回 + 流式src/routes/conversation/[id]/+server.tsPOST, update, persistConversation
事件协议src/lib/types/MessageUpdate.tsMessageUpdateType, MessageStreamUpdate
对话树src/lib/utils/tree/buildSubtree.tsbuildSubtree, addChildren, addSibling