Flue SDK 事件流完整指南：3 步实时消费 Agent 执行进度

Flue SDK 事件流完整指南3 步实时消费 Agent 执行进度【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flueFlue 是一个面向生产环境的沙箱 Agent 框架其官方客户端flue/sdk把 Agent 的执行进度——消息生成、工具调用、推理过程——变成可订阅、可续传的事件流。本文带你弄懂 Flue SDK 事件流背后的202 受理模型并给出 3 个步骤搭出一个能盯着 Agent 干活的实时界面。send() 为什么拿不到回复202 受理模型先别急着写代码先把一个最容易踩坑的设计讲清楚Flue 的send()是受理不是请求-应答。你发送一条消息后服务器只要把消息持久化受理就立即返回 HTTP 202不同步等待 Agent 跑完。所以响应里装不下回复装的是一张收据共 4 个字段字段含义submissionId本次提交的唯一标识回复结算时带着它回来offset断点续传的位置标记从这里读起正好看到之后的事件streamUrl这条对话事件流的读取 URLuid会话实例标识后续消息可凭它精确续接同一实例换句话说回复和全部中间过程都要去这条对话的事件流里读回。SDK 里的FlueClient就是围绕这一条事件流构建的对外暴露 5 种读法。同一条事件流的 5 种读法5 个方法封装的是同一条对话事件流只是抽象层次不同。对照下表选一个即可方法作用适用场景send()投递消息立即返回受理收据发起任务read()阻塞到提交完成直接返回回复文本简单的问一句答一句wait()只等终态不返回回复内容只关心做完没、成没成observe()持续观察对话自动维护可渲染状态聊天界面、进度面板history()一次性读取当前对话快照页面刷新后恢复现场两个值得注意的点read()与wait()的差别在于前者多取一次回复它内部就是wait()加一次history()读取。observe()是主力方法先用历史快照补齐上下文再切到实时尾巴持续跟进。它的live选项支持两种传输——long-poll默认兼容性好和sse长连接token 级延迟更低两者推送的是同一批 chunk只是传输方式不同。完整的参数与返回类型参考官方文档apps/docs/src/content/docs/sdk/flue-client.md。3 步跑通最小实时进度示例第 1 步创建客户端。一个客户端对应一条对话URL 是Agent 挂载路径 你自己起的对话 ID开新对话就是换个 ID 再建一个客户端。import { createFlueClient } from flue/sdk; const client createFlueClient({ url: https://app.example.com/api/support/thread-42, });第 2 步建立观察。observe()同步返回、不发网络请求首次subscribe()才真正开始工作const observation client.observe(); observation.subscribe(() { const { conversation, phase } observation.getSnapshot(); for (const message of conversation?.messages ?? []) { for (const part of message.parts) { if (part.type text) render(part.text, part.state); } } });part.state非streaming生成中即done已完成这就是实时进度的直接来源phase会经历loading → connecting → live失败进error、关闭进closed拿它驱动进度条即可。第 3 步发送消息。进度就会在已订阅的视图里逐字打出来await client.send({ message: { kind: user, body: 排查一下这个工单 } });如果你写的是脚本而不是前端界面最小往返只需两行先send()拿到受理收据再client.read(admission)阻塞到完成并返回回复文本。事件流里到底有什么chunk 类型一览事件流传输的最小单元是ConversationStreamChunk每个 chunk 都带一个单调递增的position。常见的几类Chunk 类型携带内容message-delta文本/推理的增量片段逐字效果的来源message-started/message-completed一条消息的开始与结束tool-input/tool-output/tool-output-error工具调用的入参、结果、报错data-partAgent 主动附加的结构化数据submission-settled提交终止completed / failed / abortedconversation-reset整场对话被重置携带全新快照stream-checkpoint传输连续性标记非内容SDK 内部消化好消息是你几乎不需要自己处理 chunk。observe()已经把它们归约成可直接渲染的FlueConversationState——消息、parts、settlements 全是现成的。只有两种场合会直接接触原始 chunkwait()的onEvent回调做进度打印以及底层调试。另外注意 chunk 类型并非稳定应用 API官方建议应用代码始终消费归约后的状态。断网重连为什么不丢消息这是事件流最容易被忽视、也最让新手安心的设计机制分四层offset 是不透明断点每次响应都带回下一个 offset原样保存、原样回传即可不要解析或做算术从-1读起会重放整段历史。至少一次投递连接在一批数据中途断开时传输层会从这批之前的 offset 重连并重放在途的这批事件——宁可重复不可丢失。自动去重observe()记录已应用 chunk 的position把小于等于它的重放 chunk 直接丢弃所以界面不会打出两遍。自动恢复流连续 90 秒毫无动静即判定为死连接断流重连重连按 1 秒起步、翻倍递增、封顶 30 秒的退避进行HTTP 400 直接失败属协议错误重试无益401/403 连续重试到 3 次仍失败才进入error相位此时可调用refresh()手动重检。重连策略上还有个有意思的选择observe()断线后不做增量续传而是重新拉一份历史快照来重建状态——快照由服务端物化、成本低且天然重置去重基线。简单说网络抖动、页面刷新、进程重启交给offset 自动重连即可你的业务代码只写收到状态就渲染。这套状态机的完整实现在 packages/sdk/src/public/observe.ts。按场景选读法场景推荐做法聊天界面逐字显示observe()subscribe()渲染part.state长对话只看最近observe({ limit })配historyBefore()往前翻页只要结果read()一行拿回复只想知道提交成败wait()配合submissionId页面刷新恢复现场先history()补快照再observe()接管还有一条捷径写 React 聊天页可以直接用flue/react的useFlueAgent()它直接构建在observe()之上默认使用sse模式以降低 token 级延迟。收束你接下来该做什么Flue 的事件流把Agent 在干什么变成了可订阅、可续传、自动去重的实时数据send()发起observe()看过程wait()/read()拿结果offset兜底断网。掌握202 受理 事件流回读这一个模型Agent 实时界面的关键路径就都通了。下一步建议二选一读 SDK 事件文档 apps/docs/src/content/docs/sdk/events.md对齐物化状态与流 chunk两层词汇或直接翻线级协议参考 apps/docs/src/content/docs/reference/streaming-protocol.md看 wire 上每个字节长什么样。【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考