流式传输架构

流式传输架构

学习目标

读完本章后，你将能够：

• 区分 token 级流式、块级流式和预览流式的适用场景
• 设计块级分块算法（边界保护、合并策略）
• 评估不同流式模式对用户体验和系统资源的影响

前置知识

• HTTP/SSE 基础
• 异步流处理（ReadableStream、AsyncIterator）

---

核心概念

1. 三层流式模型

生产级 Agent 系统通常同时存在三种流式抽象：

层级	粒度	适用场景	延迟	复杂度
Token 级	单个 token	模型 API 原始输出	最低	低
块级	语义块（段落/句子）	消息渠道的最终回复	中	中
预览	增量文本更新	实时打字效果	中	中

关键洞察：大多数消息渠道（WhatsApp、Telegram、Slack）不支持真正的 token 级流式，而是通过 发送+编辑 消息模拟实时效果。

2. Token 级流式（模型层）

模型 API（OpenAI、Anthropic 等）通过 SSE（Server-Sent Events）返回 token 增量：

data: {"choices":[{"delta":{"content":"Hello"}}]}
data: {"choices":[{"delta":{"content":" world"}}]}
data: [DONE]

Agent 系统消费 token 流，将其聚合为更高层级的抽象（块或预览）。

3. 块级流式（Block Streaming）

块级流式将模型输出分割为语义完整的块，每完成一个块就发送给用户。

分块算法要素：

1. 低界（Low Bound）：缓冲区达到最小字符数前不发射（避免碎片化）
2. 高界（High Bound）：优先在最大字符数前分割，强制时在最大字符数处分割
3. 断点偏好（Break Preference）：paragraph → newline → sentence → whitespace → hard break
4. 代码围栏保护：不在 Markdown 代码块内分割；强制分割时关闭并重新打开围栏

边界语义：

• text_end：每当 chunker 发射块时立即发送
  • 优点：响应更快
  • 缺点：可能产生较多消息气泡
• message_end：等待 assistant 消息完成后再刷新缓冲
  • 优点：更紧凑，减少消息数量
  • 缺点：延迟更高

合并策略（Coalescing）：

• 基于空闲间隙（idle gap）合并连续块
• 设置最小字符数防止发送微小片段
• 连接符由 break preference 决定（\n\n / \n / 空格）

人性化 pacing：

• 块之间添加随机化暂停（如 800–2500ms）
• 仅应用于块回复，不应用于最终回复或工具摘要

4. 预览流式（Preview Streaming）

预览流式通过发送临时消息并不断编辑来模拟实时打字效果：

实现模式：

模式	行为	适用渠道
off	禁用预览	所有
partial	单条预览消息，不断替换为最新文本	Telegram、Discord
block	预览消息按块追加更新	Telegram、Discord
progress	进度/状态预览，完成后显示最终答案	Slack

技术实现：

• Telegram：sendMessage + editMessageText
• Discord：发送草稿消息 + edit
• Slack：原生 streaming API 调用

预览 vs 块流的冲突：

• 当 block streaming 启用时，应跳过 preview streaming 以避免双重流式
• 某些渠道支持 preview streaming 但不支持 block streaming，需要独立配置

5. 双层流式的协作

问题与规避

消息气泡过多

• text_end + 短文本导致频繁发送单条消息
• 对策：启用 coalescing 或切换到 message_end 模式

代码块被截断

• 分块算法在代码围栏中间分割，破坏 Markdown 渲染
• 对策：实现代码围栏保护逻辑，强制分割时关闭+重开围栏

渠道 API 限制

• 频繁编辑可能触发 Telegram/Discord 的速率限制
• 对策：设置合理的 coalescing idle gap，批量更新而非逐 token 编辑

设计取舍

Token 级 vs 块级流式

• Token 级：最低延迟，适合打字机效果的 UI
• 块级：语义完整，适合消息渠道的最终交付
• 多数生产系统同时消费 token 流（内部）并发射块流（外部）

即时发射 vs 延迟刷新

• text_end：用户体验更接近实时对话
• message_end：减少消息数量，更适合长文本输出
• 可配置 break 策略让用户/管理员根据场景选择

参考来源

• OpenAI Streaming
• Server-Sent Events

---

补充：企业级 AI 应用的 SSE 事件类型体系

本补充来自 JeecgBoot v3.9.2 的源码探索，展示了生产环境中 SSE 事件类型的设计模式。

多事件类型体系

在复杂 AI 应用中，单一 text 事件不足以描述完整的交互过程。一个完善的 SSE 事件系统需要定义多种事件类型：

事件类型	用途	数据载荷
初始化事件	携带 requestId、conversationId、topicId	会话标识
内容事件	流式输出 AI 响应内容	partial text
思考事件	推理模型的 reasoning_content 实时推送	thinking token
思考结束事件	标记推理模型思考阶段结束	空
工具执行前事件	工具调用前的名称和参数	tool name + args
工具执行完成事件	工具调用后的结果	tool name + result
流程完成事件	流程引擎最终输出	final output
流程异常事件	执行失败的错误信息	error message
消息结束事件	标记单次消息推送完成	空

推理模型 Thinking 推送适配

对于 DeepSeek 等推理模型，reasoning_content 字段与正常回答内容需要分离推送：

1. onPartialThinking 回调捕获 thinking token，推送 EVENT_THINKING
2. thinking 结束推送 EVENT_THINKING_END
3. 正常回答继续通过 onPartialResponse 推送 EVENT_MESSAGE

关键陷阱：历史消息持久化时若未保存 reasoning_content，重建 AiMessage 时 thinking 为 null，导致 DeepSeek 后续多轮请求校验失败（“reasoning_content must be passed back”）。解决方案是在发送前为历史 AI 消息注入占位 thinking。

SSE 连接生命周期管理

• 使用 AiragLocalCache 缓存 SseEmitter，支持断线重连时继续接收（通过 receiveByRequestId）
• 超时计时器：无新消息 120 秒自动释放等待线程
• closeSSE 封装：确保无论成功/异常/完成，都发送 EVENT_MESSAGE_END 并正确关闭 emitter