墨台 MUOTAI

Appearance

流式事件、轨迹流与外部渠道 ​

AI 输出不是一条字符串，而是一组带 session、run、sequence 和事件类型的可见事件。普通聊天 SSE 负责实时 UI；JSONL 轨迹 SSE 负责可恢复审计和进度；外部 IM 负责把最终结果投递到原渠道。

1. 响应通道层次 ​

text
Agent node
  -> StreamHandler
  -> AgentResponseChannel
  -> AgentResponseChannelRegistry
  -> SSE / trajectory stream / external IM adapter

StreamContext 在线程内绑定当前 StreamHandler，图节点通过它发送 token、节点状态、RAG 引用、工具结构化块和终态。dispatcher 不应直接拼接模型文本；它只负责找到 channel、等待 channel ready、处理超时和释放 channel。

2. 普通聊天流式事件 ​

接口：

text
POST /v1/chatbot/{agentId}/sessions/{sessionId}/chat?sse=sync&stream=true
POST /v1/chatbot/{agentId}/sessions/{sessionId}/chat?sse=async&stream=true
GET  /v1/chatbot/{agentId}/sessions/{sessionId}/sse/connect

sse=sync 在提交请求时建立 emitter；sse=async 先入队返回 messageId，客户端随后连接 connect。连接建立前的错误、完成和取消会进入 PendingSseEventStore，连接成功后按 session drain；迟到事件按 TTL 和每 session 最大数量清理。

Dispatcher 默认最多等待 channel 60 秒，轮询间隔至少 50ms；超过等待时间写入错误事件，不应让虚拟线程永久阻塞。

3. 事件类型 ​

SseStreamHandler 当前会产生以下可见类别：

事件主要字段/用途
token文本增量；由约 200ms buffer 批量发送
nodegraph node started/completed、action、intent
ragDataset、查询、命中数、置信度和预览
referenceWeb Search/RAG 来源引用
thinking_start/content/end受规则过滤的思考内容
structured_blockloadConfig、toolCall、skillCall、rag、taskEnd 等卡片字段
agent_staterunning、paused、resumed、completed 等状态
error可展示错误和 messageId/sessionId
complete最终完成、响应、run 信息
artifact/文件事件产物发布和文件引用

token、thinking 和 structured block 不能全部拼成回答文本；客户端应按事件类别分别渲染。rawResultMode 下子 Agent 可能已经直接流出结果，TaskEndNode 会静默总结或不再重复发送。

4. 刷新与终态顺序 ​

token buffer 使用定时器和同步 flush：节点结束、错误、取消和 complete 前都会强制刷新。完成顺序必须保持：

text
flush token
 -> flush assistant response snapshot
 -> send visible complete/error/cancel event
 -> update run status
 -> close emitter/channel

这样可以避免客户端先收到 COMPLETED，却因最后一批文本尚未写入而看到空回答。运行状态更新不能替代可见事件。

5. 轨迹流式事件与断线恢复 ​

持久化轨迹接口：

text
GET /v1/agents/{agentId}/sessions/{sessionId}/trajectory/events
GET /v1/agents/{agentId}/sessions/{sessionId}/trajectory/stream

events 支持 afterSequence 分页；stream 先重放游标之后的 JSONL，再订阅新提交事件，支持 afterSequence 或 Last-Event-ID。大型 payload 只返回 blob reference，详情接口再读取完整内容。

普通聊天 SSE 没有把所有 token 当作可无限重放的事件流，而是依靠 pending event、assistant response snapshot 和 execution snapshot 补偿。因此重连策略应区分两类通道：聊天窗口优先读取快照并去重，审计/执行监控使用轨迹 sequence。

6. 外部即时通信 ​

微信、企业微信、钉钉、QQ 等渠道先把 webhook 解析为 AgentMessage，再使用 submitLocal 或渠道适配器进入同一 session 执行器。接入适配器必须保存：

  • 外部消息 ID、会话/群 ID、回复目标和租户。
  • 原始输入与模型增强输入的映射。
  • 幂等键和已回复状态。
  • 失败重试次数、退避时间和死信记录。

外部 IM 的回复目标通常只在接入节点内存中有效；如果跨节点使用共享队列，必须共享路由、粘性策略或把回复地址持久化。不要让 Agent 节点直接持有渠道 SDK 的全局状态。

7. 客户端状态机 ​

客户端必须用 sessionId + runId/runVersion 判断事件归属，不能因为同一 session 收到迟到 token 就覆盖新一轮回答。刷新页面时先查询 run-status/execution snapshot，再建立 SSE。