[!info] related notes

• 所属 MOC: LLM Streaming MOC
• 组合: Multi-channel Stream, Event Sourcing AI
• 综合: Streaming 高级协议模式
• 基础协议: Delta Stream, Append Event Stream
• 分流层: Semantic Router

Structured Streaming

一句话定义

Structured Streaming 是 LLM Streaming 的协议封装模式——为每个流事件定义稳定、可扩展、可版本化的结构，让前端 reducer、后端日志、回放系统、监控系统都能统一消费。

核心机制

解决什么问题

没有统一结构时，流事件可能是各种随意的 JSON：

{ "text": "你好" }
{ "tool": "search", "args": {} }
{ "info": {} }
{ "done": true }

前端需要对每种事件写不同的解析逻辑，扩展新事件类型时容易出错。

示例事件封装

这是用于说明字段职责的外壳，不是可直接验证网络输入的完整 schema；具体 type 应绑定对应 data 类型，并定义版本和未知事件策略。

interface StreamEvent {
  id?: string           // 事件唯一 ID（可选）
  type: string          // 事件类型（必填）
  channel?: string      // 输出通道（可选）
  data: object          // 事件数据（必填）
  timestamp?: number    // 时间戳（可选）
}

统一后的事件示例

{
  "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": "message.completed",
  "channel": "text",
  "data": { "message_id": "m1" }
}

事件类型命名规范

推荐使用 entity.action 格式：

type	语义
message.started	消息开始生成
message.delta	文本增量
message.completed	消息生成完成
tool_call.started	工具调用开始
tool_call.completed	工具调用完成
extraction.updated	结构化信息更新
status.changed	流程状态变化
error.occurred	错误发生

最小场景

单消息顺序流的 reducer 片段（StreamState 与按 type 区分的 StreamEvent 联合类型由应用另定义）。这里只示意投影，不处理并发、重放去重、工具完成和 Run 终态；外壳接口中的 data: object 不足以让这段代码直接通过类型检查：

function streamReducer(state: StreamState, event: StreamEvent): StreamState {
  switch (event.type) {
    case "message.started":
      return { ...state, currentMessage: { id: event.data.message_id, content: "" } }
    case "message.delta":
      if (!state.currentMessage || state.currentMessage.id !== event.data.message_id) return state
      return { ...state, currentMessage: { ...state.currentMessage, content: state.currentMessage.content + event.data.content } }
    case "tool_call.started":
      return { ...state, toolCalls: [...state.toolCalls, event.data] }
    case "message.completed":
      if (!state.currentMessage || state.currentMessage.id !== event.data.message_id) return state
      return { ...state, currentMessage: null, messages: [...state.messages, state.currentMessage] }
    default:
      return state
  }
}

优势

• 事件类型清晰：type 字段一目了然
• 前端 reducer 分发简单：switch/case 即可
• 可扩展，但新增必需事件或修改字段可能破坏旧消费者，需明确兼容策略
• 方便日志和回放：所有事件结构统一，存储和查询简单
• 可增加 version 表达版本；兼容性仍需 schema 演进规则和实际迁移，字段本身不会自动实现兼容

与 Delta / Append 的关系

Structured Streaming 不是和 Delta / Append 竞争的协议，而是它们的封装层：

• Delta 事件放进结构：{ "type": "message.delta", "data": { "content": "你好" } }
• Append 事件放进结构：{ "type": "tool_call.started", "data": { "name": "search" } }

也就是说：

Delta / Append 是”怎么更新”；Structured streaming 是”事件长什么样”。

边界与易混淆点

• Structured streaming 不是 Apache Spark 的 Structured Streaming：这里指的是 LLM 流式事件的结构化封装，不是大数据处理框架。
• 不要过度设计：简单场景不需要这么复杂的结构，可以从 { type, data } 起步，但仍须定义合法 payload 和完成/失败语义。
• type 命名要稳定：一旦前端依赖了某个 type 值，修改成本很高。命名要谨慎。
• data 字段的 schema 也需要约定：type 只是入口，data 内部结构也需要文档化。