[!info] related notes

• 所属 MOC: AI Agent Application MOC, LLM Streaming MOC
• 相关: 事件契约, SSE, SSE Gateway

AI 应用流式协议

一句话定义

AI 应用流式协议是定义全链路（Python → Go → React）流式事件传输的规范。它不只是”用 SSE 推数据”，而是要定义事件格式、连接生命周期、错误恢复和背压控制。

核心原理

全链路架构

Python AI Service ──SSE──→ Go 后端 ──SSE──→ React 前端
  (事件产生)        (事件转发)      (事件消费)

协议选型

链路	推荐协议	原因
Python → Go	SSE 或 gRPC streaming	简单、可靠
Go → React	SSE	浏览器原生支持

事件生命周期

连接建立
    │
    ├─ heartbeat (每 15s)
    ├─ text_delta (LLM 输出)
    ├─ tool_call_start (工具调用开始)
    ├─ tool_call_result (工具调用结果)
    ├─ progress (进度更新)
    ├─ interrupt (需要审批)
    ├─ error (错误)
    └─ done (完成)
    │
连接关闭

连接管理

# Python: 产生事件流
async def event_stream(context) -> AsyncIterator[str]:
    yield f"event: heartbeat\ndata: {json.dumps({})}\n\n"

    async for event in agent_run(context):
        yield event.to_sse()

    yield f"event: done\ndata: {json.dumps({'status': 'completed'})}\n\n"

// Go: 转发事件流
func proxySSE(w http.ResponseWriter, pythonStream <-chan Event) {
    w.Header().Set("Content-Type", "text/event-stream")
    flusher := w.(http.Flusher)

    for event := range pythonStream {
        fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.Type, event.JSON())
        flusher.Flush()
    }
}

// React: 消费事件流
const es = new EventSource('/api/chat/stream');
es.addEventListener('text_delta', handleTextDelta);
es.addEventListener('tool_call_start', handleToolCallStart);
es.addEventListener('done', handleDone);
es.addEventListener('error', handleError);

常见坑

1. 不做心跳: 连接被代理断开
2. 不做重连: 网络抖动后连接丢失
3. 事件格式不统一: Python 和 Go 用不同的格式
4. 不做背压: 产生太快，消费太慢

参考资料

• LLM Streaming MOC
• Event Contract
• SSE