Streaming

1. 为什么需要流式输出

在具体的测量中，我们会发现 LLM 生成占到了总耗时的 70-80%，这个比例会随着模型、网络和上下文长度变化，不能当作固定指标。一段中等长度的完整回复可能需要等待 500-2000ms，如果 AI 伴侣要生成约 200 字的安慰内容，等待时间还可能超过 3 秒

LLM 并不是先生成完整答案，再一次性返回。模型会逐步生成 token，但接口传回来的一个 chunk 可能包含一个或多个 token。首个文本片段可能在 200ms 内就已经可用；如果服务端仍然等待全部内容生成完毕，用户就会额外等待 1-2 秒。实际首字节时间仍取决于模型、上下文长度、网络和服务端实现。

流式输出会在文本片段可用后立即传给客户端，让用户从等待完整回复，变成直接看到文字逐步出现。

从首字节时间来看，两种模式通常存在明显差异。下面的数字只是测量示例，实际结果需要结合模型、网络和部署环境验证：

模式	首字节时间	用户感知
非流式（等完整回复）	1500-3000ms	它在想什么？卡了吗？
流式（收到首个片段后开始推送）	200-500ms	它在说话了

对于实时对话产品，流式输出不是额外的视觉效果，而是控制用户感知延迟的基础能力。

人的阅读速度大约是每秒 5-8 个汉字，而模型的输出速度会因模型和语言而变化。对于中文对话，生成速度有时会高于阅读速度；在这种情况下，持续追加文字可以明显减轻等待感，也更接近自然的打字过程。

2. SSE 协议

服务端向前端持续推送数据时，常见方案包括 HTTP Streaming、SSE 和 WebSocket：

维度	HTTP Streaming	SSE（Server-Sent Events）	WebSocket
方向	单向（服务端 → 客户端）	单向（服务端 → 客户端）	双向
协议	基于 HTTP 的响应流（HTTP/1.1 常用 chunked）	基于 HTTP 的 text/event-stream	独立协议（ws://）
自动重连	无	EventSource 内置，fetch 需自行实现	需手动实现
事件类型	无	支持 event 字段分类	自行约定
边缘环境兼容性	好	好	可用，但连接协调通常更复杂

AI 伴侣当前采用的是明确的请求与响应模式：用户发送一条消息，AI 返回一条回复。WebSocket 提供的双向通道在这个场景中没有得到充分利用，却需要额外处理连接管理、心跳和断线重连。如果后续要在 Cloudflare Workers 上协调连接、维护房间状态或广播消息，通常还需要结合 Durable Objects，架构复杂度和运行成本也会随之增加。若产品以后需要服务端主动推送或多人实时协作，WebSocket 仍然是合理的选择。

SSE 是建立在 HTTP 响应流之上的事件格式：HTTP/1.1 下通常使用 chunked transfer，HTTP/2 和 HTTP/3 则由协议帧负责传输。SSE 在数据流之上定义了 event、data 和空行分隔规则，服务端可以明确标记事件类型，前端也能按照同一套规则解析数据。

下面是一段完整的 SSE 响应报文：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache

event: thinking
data: {"node": "memory_retrieval"}

event: token
data: {"content": "我"}

event: token
data: {"content": "记得"}

event: token
data: {"content": "你上次"}

event: done
data: {"emotion": "happy", "memoriesUsed": 3, "outputCharacters": 24}

理解 SSE 时，可以先记住四条基本规则。

第一，响应头需要使用 Content-Type: text/event-stream。浏览器或客户端据此把响应识别为持续推送的事件流，不会等待整个响应结束，而是在收到完整事件后立即交给调用方处理。Connection: keep-alive 是否需要设置，则取决于 HTTP 版本和中间层，不应作为 SSE 的必需条件。

第二，每条事件由若干 字段: 值 行组成。常用字段包括 event、data 和 id，分别表示事件类型、数据负载和用于断线重连的事件 ID。字段名与冒号之间没有空格，冒号和值之间保留一个空格。

第三，事件之间使用空行，也就是 \n\n 分隔。解析器遇到双换行时，就可以确认一条完整事件已经结束。

第四，data 字段可以跨越多行。多个 data: 行会使用 \n 拼接成一个字符串，不过在本文的 LLM 流式场景中，每条事件只使用一个 data 行。

event 字段可以把思考、生成、完成和错误四种状态明确区分开。前端根据事件类型切换对应的 UI，不需要再从连续字节流中推测当前阶段。相比之下，裸 HTTP Streaming 只提供连续数据，事件分类和边界规则都需要应用自行定义。需要注意的是，浏览器的 EventSource 只支持 GET；本例使用 POST 携带消息，因此前端会用 fetch 自己解析 SSE。

3. 三层流式管线

完整的流式响应可以分为生成、传输和渲染三个层次。

生成层由 LangGraph 和 LLM 组成，负责逐步产生文本片段。它向外提供一个异步迭代器，每次 yield 一个文本片段或状态事件；片段边界不一定与模型 token 一一对应。

传输层运行在 Hono 和 Workers 中，负责把生成层输出转换为 SSE 格式，并通过 HTTP 持续发送给客户端。超时、异常捕获和连接中断检测也在这一层处理。

渲染层位于 React 前端，负责消费 SSE 流并更新界面，包括增量文本拼接、Markdown 渲染和状态切换动画。

三层之间的数据流是单向的：

LangGraph 节点执行
    ↓ yield { type: 'thinking', node: 'memory_retrieval', status: 'start' }
    ↓ yield { type: 'thinking', node: 'memory_retrieval', status: 'end' }
    ↓ yield { type: 'thinking', node: 'llm_generate', status: 'start' }
    ↓ yield { type: 'token', content: '我' }
    ↓ yield { type: 'token', content: '记得' }
    ↓ ...
    ↓ yield { type: 'done', metadata: { emotion, memoriesUsed } }
Hono SSE 中间层
    ↓ event: thinking\ndata: {...}
    ↓ event: token\ndata: {...}
    ↓ event: done\ndata: {...}
React 前端
    → 显示 "正在思考..."
    → 逐字追加文字
    → 完成，显示情绪标签

除了模型生成的文本片段，管线节点的执行状态也会进入事件流。前端不必一直展示静态 loading，而是可以根据事件提示用户当前正在检索记忆、组织语言或开始回复。

接下来从服务端开始实现这条流式管线。

4. Hono 流式接口

实现接口之前，我们先比较普通 HTTP 响应与流式响应的执行方式。

普通响应会先在服务端准备好全部数据，再一次性写入 Response body。浏览器拿到完整响应后才开始处理。

normal.ts

import { Hono } from 'hono'

const app = new Hono()

// 普通响应：等 LLM 生成完毕，一次性返回
app.post('/chat', async (c) => {
  const { message } = await c.req.json()
  const reply = await callLLM(message)  // 等待 1-3 秒
  return c.json({ reply })               // 全部完成后才返回
})

在这种方式下，用户发送消息后需要等待 1-3 秒，随后一次性看到完整回复。

流式响应会先返回 Response Header，再持续、分块地向 body 写入数据。浏览器每收到一块内容就可以立即处理，不必等待全部生成完成。

streaming.ts

import { Hono } from 'hono'
import { streamSSE } from 'hono/streaming'

const app = new Hono()

// 流式响应：边生成边返回
app.post('/chat/stream', async (c) => {
  const { message } = await c.req.json()
  // 立即返回响应头，body 持续写入
  return streamSSE(c, async (stream) => {
    const llmStream = callLLM(message, { stream: true })
    for await (const chunk of llmStream) {
      await stream.writeSSE({           // 每个文本片段立即推送
        event: 'token',
        data: JSON.stringify({ content: chunk.text })
      })
    }
  })
})

用户发送消息后，首个片段可能在数百毫秒内出现，后续内容再随着模型生成持续追加。这个时间需要通过实际监控确认，不能直接写死为某个固定值。

两者的关键区别在于 Response 的创建时机。普通响应会等数据准备完成后再创建 Response，流式响应则先创建一个可以持续写入的 Response，再逐步向其中发送数据。streamSSE 内部通过可写流把事件写入响应，服务端可以在生成新片段时继续发送 SSE 事件。

Hono 提供 streamSSE 函数来创建 SSE 响应。下面先看一个最小示例：

minimal.ts

import { Hono } from 'hono'
import { streamSSE } from 'hono/streaming'

const app = new Hono()

app.get('/hello-stream', async (c) => {
  return streamSSE(c, async (stream) => {
    // stream 对象是你和前端之间的管道
    // 通过 writeSSE 方法往管道里写入 SSE 事件

    await stream.writeSSE({
      event: 'greeting',
      data: '你好'
    })

    // 模拟延迟
    await stream.sleep(1000)

    await stream.writeSSE({
      event: 'greeting',
      data: '世界'
    })
  })
  // 回调函数执行完毕后，流自动关闭
})

export default app

streamSSE(c, callback) 接收两个参数。

第一个参数 c 是 Hono Context，也就是当前请求的上下文。streamSSE 使用它创建 Response，并设置 Content-Type: text/event-stream 等 SSE 所需的响应头。是否需要额外设置缓存或代理相关响应头，要结合部署平台的行为决定。

第二个参数 callback 是一个接收 stream 对象的异步函数。服务端通过这个对象向前端写入事件，它提供三个常用方法：

方法	作用
stream.writeSSE({ event, data, id })	写入一条 SSE 事件
stream.sleep(ms)	异步等待指定时间，不占用 CPU 执行时间
stream.close()	手动关闭流

writeSSE 接收一个对象，其中 event 是字符串形式的事件类型，data 是字符串形式的数据负载，id 是可选的事件 ID。data 必须是字符串，因此发送 JSON 时需要先调用 JSON.stringify。

callback 正常 return 或执行到末尾后，流会自动关闭，前端的 reader.read() 会返回 { done: true }。如果 callback 抛出未捕获异常，连接同样会结束，但前端无法获得具体原因。因此，服务端需要在 callback 内部捕获异常，并通过 error 事件明确通知前端。

实现完整接口前，需要先定义前后端共同遵守的事件协议。服务端根据管线状态发送事件，前端再依据事件类型更新 UI。

events.ts

// 管线阶段事件：告诉前端当前执行到了哪个节点
interface ThinkingEvent {
  type: 'thinking'
  node: string       // 'safety_check' | 'memory_retrieval' | 'llm_generate' | ...
  status: 'start' | 'end'
}

// 文本生成事件：LLM 持续输出文本片段
interface TokenEvent {
  type: 'token'
  content: string    // 一个或多个字符，片段边界不等同于 token 边界
}

// 完成事件：对话成功结束，携带元信息
interface DoneEvent {
  type: 'done'
  emotion: string
  memoriesUsed: number
  outputCharacters: number
}

// 错误事件：出了问题，附带兜底回复
interface ErrorEvent {
  type: 'error'
  code: string       // 'safety_blocked' | 'llm_timeout' | 'internal_error'
  message?: string
  fallbackReply?: string  // 兜底话术，前端直接展示即可
}

type StreamEvent = ThinkingEvent | TokenEvent | DoneEvent | ErrorEvent

这里把业务决策放在服务端，前端只负责渲染。收到 thinking 时显示加载状态，收到 token 时追加文字，收到 done 时结束本轮生成，收到 error 时展示兜底回复。安全拦截、情绪路由和降级策略都不需要在前端重复判断。

fallbackReply 用来处理不需要用户额外操作的错误。例如发生安全拦截时，前端可以直接展示“这个话题我们换个方向聊聊吧~”，而不是弹出“您的消息违规”一类生硬的错误提示。

DeepSeek、OpenAI 和 Claude 都支持流式输出，但响应格式并不完全一致。可以在服务层封装统一接口，让上层管线不依赖具体的模型提供商。

以下以 OpenAI 兼容格式为例，DeepSeek 也提供类似格式。请求设置 stream: true 后，API 会返回 SSE 响应流，并在生成文本片段时逐条推送事件；具体字段仍应以所使用模型供应商的文档为准：

llm-response.txt

data: {"choices": [{"delta": {"content": "我"}}]}

data: {"choices": [{"delta": {"content": "记得"}}]}

data: {"choices": [{"delta": {"content": "你"}}]}

data: [DONE]

每个 data: 行都包含一段 JSON，其中 delta.content 是本次新增的文本片段，data: [DONE] 表示生成结束。

这里使用 AsyncGenerator 封装解析过程。它适合处理内容持续产生、调用方持续消费的场景，上层代码可以通过 for await...of 逐个读取结果：

llm.ts

// AsyncGenerator 函数：用 async function* 声明，内部用 yield 逐个产出结果
// 调用方用 for await...of 消费
async function* callLLMStream(
  systemProm