[!info] related notes

• 知识地图: LLM Streaming MOC
• 理论基础: LLM 输出四层模型, Delta Stream, Append Event Stream
• 架构设计: LLM Streaming 分层架构
• 上层应用: 会话与消息流设计
• SSE 管道: 多服务 SSE 管道
• 前端消费: 前端 SSE 消费
• Tool Call: Function Calling 流式累积
• AI 架构: AI Gateway 模型路由器

LLM Streaming 协议设计（项目实践版）

范围

本文基于 LLM Streaming MOC 中的协议理论，结合 BodySense 项目的实际场景，给出推荐的协议设计方案。

理论知识（三种基础协议、五层架构、三种高级模式）已拆分为独立原子笔记，本文聚焦项目实践。

BodySense 场景分析

BodySense 的流式输出需求：

需求	说明
文本回答	AI 助手的回复，需要打字机效果
Function Calling	调用症状提取、体态分析等工具
结构化症状提取	从对话中提取结构化症状信息
安全警告	Red Flag 检测结果
引用来源	RAG 检索的参考文献
阶段变化	咨询流程的状态推进

推荐方案：Delta Stream + 语义事件扩展

外层用 Append 管理流程，内层用 Delta 管理文本生成。

协议设计

{ "type": "start" }

{ "type": "text.delta", "content": "你好" }
{ "type": "text.delta", "content": "，我可以帮你" }

{ "type": "extracted_info", "data": { "body_part": "肩部", "symptom": "疼痛" } }

{ "type": "tool_call", "name": "symptom_extract" }

{ "type": "end" }

更完善的协议版本

采用 Structured Streaming + Multi-channel Stream：

{ "type": "message.started", "channel": "text", "data": { "message_id": "m1", "role": "assistant" } }

{ "type": "message.delta", "channel": "text", "data": { "message_id": "m1", "content": "我先帮你整理一下症状。" } }

{ "type": "tool_call.started", "channel": "tool", "data": { "tool_call_id": "t1", "name": "symptom_extract" } }

{ "type": "extraction.updated", "channel": "structured", "data": { "symptoms": ["头痛", "发热"] } }

{ "type": "red_flag.detected", "channel": "status", "data": { "has_red_flags": true, "flags": ["高热"] } }

{ "type": "message.delta", "channel": "text", "data": { "message_id": "m1", "content": "根据你的描述，建议尽快就医。" } }

{ "type": "message.completed", "channel": "text", "data": { "message_id": "m1" } }

事件类型映射

事件类型	channel	前端处理
message.delta	text	追加到当前消息（打字机效果）
tool_call.started	tool	显示工具调用状态
tool_call.completed	tool	显示工具结果
extraction.updated	structured	更新人体可视化面板
red_flag.detected	status	显示安全警告
citation.added	text	显示引用来源
phase.changed	status	更新诊断面板
message.completed	text	流结束，更新状态

后端实现结构

LLM stream (OpenAI / Claude / Qwen)
   ↓
Chunk Aggregator (token → 语义 chunk)
   ↓
Semantic Router (text / tool / structured / status)
   ↓
Event Builder (构建统一格式事件)
   ↓
SSE Emitter (发送到前端)

Chunk Aggregator 策略

• 文本：按句号、感叹号、换行聚合
• Tool call：按 function call boundary 切分
• 结构化数据：工具执行完成后一次性发送

前端实现结构

// 统一事件 reducer
function streamReducer(state: StreamState, event: StreamEvent): StreamState {
  switch (event.type) {
    case "message.delta":
      return { ...state, buffer: state.buffer + event.data.content }
    case "tool_call.started":
      return { ...state, toolCalls: [...state.toolCalls, event.data] }
    case "extraction.updated":
      return { ...state, extractedInfo: event.data }
    case "red_flag.detected":
      return { ...state, redFlags: event.data }
    case "message.completed":
      return { ...state, messages: [...state.messages, { content: state.buffer }], buffer: "" }
    default:
      return state
  }
}

常见错误：char-level split

如果前端出现逐字拆分过细（每个字符一个事件），问题不在 SSE 本身，而是 Chunk Aggregator 层没有正确聚合——后端直接把每个 token 透传给了前端。

详见 Chunk Aggregator。

升级路径

当前：Delta Stream + 手动事件类型
  ↓
升级 1：采用 Structured Streaming 统一事件格式
  ↓
升级 2：增加 Multi-channel 区分推理/回答
  ↓
升级 3：需要调试/审计时引入 Event Sourcing

一句话总结

BodySense 推荐 Delta Stream + 语义事件扩展，兼顾打字机效果和 tool_call / 结构化数据传输。随着系统复杂度增长，可逐步升级到 Structured Streaming + Multi-channel 的工业级协议。