Flue SDK 事件流指南：从 send 到实时逐字渲染，Agent 执行进度全链路掌握

Flue SDK 事件流指南：从 send 到实时逐字渲染，Agent 执行进度全链路掌握

【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue

用 Flue 部署的 Agent 是怎么把执行进度推给你的？答案是@flue/sdk客户端的事件流：send()受理后，回复、工具调用、推理过程都以事件块的形式持续流回。本文带你走完整条链路——submissionId、offset断点、observe()实时订阅、长轮询与 SSE 双模式、自动重连去重，外加一份新手踩坑清单，看完就能搭出一个实时盯着 Agent 干活的界面。

发完消息，回复去哪了

大概率你第一次写 Flue 客户端代码会遇到这个场景：

const result = await client.send({ kind: 'user', body: '帮我总结一下' }); // 打印 result.reply —— undefined？没有这个字段。

消息发出去了，但send()的返回值里没有回复。卡住你的人，多半是拿它当"一问一答"的同步接口在调。

这不是 bug，而是 Flue 刻意的202 受理模型：服务端把消息持久化落库之后立刻返回 202，此刻 Agent 可能还没跑起来。返回体里只有四个东西，其中两个是你后续所有操作的钥匙：

所以整条链路的心智模型是：写走 HTTP 一发即走，读全部走同一条事件流。

全景：消息从哪进，事件从哪出

把全局立起来只需要记住三个位置：

• 入口：POST一条消息 → 202 受理（submissionId+offset）
• 出口：同一条对话的事件流，持续吐出消息、工具调用、结算等事件块
• 续传点：每次读响应都会带下一个offset，存下它，断了从这里接着读

SDK 的 FlueClient 就是围绕"写一次、读一条流"封装的，下面按你会遇到的四个问题来讲，而不是按方法名罗列。

四个关键动作

发起任务：send()，拿到断点就走

send()的职责就一件：投递消息，拿到受理回执。不关心过程的场景（比如后台批量下发）用它就够了，之后想读进度，凭手里的offset随时能接上流。

const admission = await client.send({ kind: 'user', body: '帮我总结一下' }); // admission.submissionId、admission.offset —— 存好，后面全靠它

盯住过程：observe()，订阅可渲染的状态

这是实时界面的核心。observe.ts 里的观察器会先拉一遍历史快照补齐上下文，再切到实时尾巴（live tailing）持续跟进，并且替你维护好一份可以直接渲染的FlueConversationState。你只需要订阅它，把状态画出来：

const observation = client.observe(); observation.subscribe(() => { const { conversation } = observation.getSnapshot(); for (const message of conversation?.messages ?? []) { if (message.display !== 'visible') continue; for (const part of message.parts) { if (part.type === 'text') renderText(part.text, part.state); } } });

拿到结果：read() 要回复，wait() 只要成败

两种"拿结果"的需求，两个方法：

// 只要回复文本：等这次提交结算，直接给答案 const reply = await client.read(admission);

// 只要成败信号：结算成功 resolve，失败/中止抛 FlueExecutionError try { await client.wait(admission); } catch (error) { console.error('submission failed:', error); }

wait()还能挂onEvent回调，逐块观察原始事件（做监控、审计时常用），配合submissionId精确对上"哪一次提交"。

恢复现场：history()，刷新后补快照

页面刷新、进程重启、打开一个别人正在跑的对话——先history()拿一次当前完整快照（FlueConversationSnapshot），再observe()接管后续实时部分，现场就接上了。

const snapshot = await client.history(); // snapshot.messages 是完整对话，snapshot.offset 可以直接作为续读断点

把实时感做对：live 模式、相位和 part.state

observe()的实时尾巴有两种传输方式（stream.ts 中的FlueEventStream统一承载）：

两者推的是同一批事件块、同样的投递保证，差别只在连接形态。

配合两个状态字段，UI 就活了：

• phase：观察器的连接生命周期，按loading → connecting → live → closed走（出错进error，短暂重连时会从live回落connecting）。进度条、加载转圈直接订阅它驱动即可。
• part.state：每条文本/推理片段的生成状态——'streaming'表示正在流式生成，'done'表示已完整。逐字打字机效果就是靠它区分"半截"和"完整"。

底层不用记：事件块长什么样，可靠性谁来兜底

流上传输的最小单元是 ConversationStreamChunk，认识即可：

你基本不用亲手处理这些块——observe()已把它们归约成完整状态；只有wait()的onEvent回调会接触原始块。

可靠性这块，SDK 替你把脏活全干了（流协议参考 有完整的线级说明）：

1. offset是不透明断点：每次响应带回下一个值，原样保存、原样回传。别解析它，别对它做加减。
2. 至少一次投递：连接中途断开时，传输层从断点重放在途的那批事件——宁可重复，不可丢失。
3. 按position自动去重：每个块都带单调递增的position，observe()会丢弃重放的重复块，界面上不会出现"同一段话打出两遍"。
4. 自动恢复：内置僵尸流看门狗，流长时间无动静就自动取消重连；遇到 401/403 按退避策略重试而不是立刻失败。

结论一句话：网络抖动、刷新、重启，你不用写一行补偿逻辑，业务代码只负责"收到状态就渲染"。

新手最容易踩的四个坑

我想要 X，就做 Y

延伸阅读

• SDK 客户端与五个方法定义：client.ts
• 事件流迭代器（重连、断点重放）：stream.ts
• 观察器状态机（去重、看门狗、重试策略）：observe.ts
• chunk 协议与归约逻辑：conversation-stream.ts
• 结算等待与wait()实现：settle.ts
• 官方事件文档：sdk/events.md
• 线级流协议参考：streaming-protocol.md
• FlueClient 完整 API 参考：flue-client.md

想实时盯住 Agent 的每一刻，核心就一句话：send()把任务交出去，剩下的交给事件流——断点续读、去重、重连全是它的事，你只管把渲染好的状态画出来。

【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue