实时流式通信
Real-time Streaming Communication
SSE 流式架构实现 Agent 响应实时推送与前端消费
子问题
1.SSE 事件格式化
2.生产者/消费者解耦
3.流式 chunk 段落聚合
4.多事件类型路由
5.生产者/消费者生命周期不对称导致执行中断
6.流式 chunk 段落聚合与 stable item_id upsert
7.POST-based SSE 替代 EventSource 的工程实现
8.流式 JSON 不完整时的容错解析
各项目的解法2 solutions
Signals
横向对比
| 维度 | ValueCell | seedance-2 |
|---|---|---|
| 传输协议 | SSE over POST (fetch+ReadableStream)，非 EventSource | — |
| 生产消费解耦 | asyncio.Queue + background task，消费者断开不中断生产者 | — |
| 事件类型体系 | 4 枚举 16 事件类型，Pydantic BaseResponse 统一信封 | — |
| 段落聚合 | ResponseBuffer 双策略(immediate/buffered)，stable item_id upsert | — |
| 中断恢复 | ExecutionContext 保存 planning stage，TTL 过期 + 用户校验 | — |
| 前端容错 | best-effort-json-parser 容忍不完整 JSON，AbortController 超时 | — |
最佳实践
1.asyncio.Queue 解耦生产消费
2.ResponseBuffer 段落聚合
3.背景任务独立于 SSE 连接生命周期
4.asyncio.Queue + active flag 实现消费者断开不中断生产者
5.immediate 事件触发 flush 保证 buffered 内容持久化顺序
6.fetch+ReadableStream 替代 EventSource 支持 POST 和自定义 headers
7.handlersRef 模式避免 React hook 中的 stale closure