ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

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

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 可能还没跑起来。返回体里只有四个东西,其中两个是你后续所有操作的钥匙:

字段干什么用
submissionId这次提交的唯一身份。Agent 跑完时,结算事件会带着它回来
offset断点标记(resume token)。从它之后读事件流,看到的就是 Agent 接下来产生的一切,不重放历史
streamUrl这条对话的事件流地址
uid对话实例标识,可用于条件发送

所以整条链路的心智模型是:写走 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统一承载):

模式特点什么时候选
'long-poll'(默认)长轮询,兼容性好大多数场景的稳妥选择
'sse'SSE 长连接,连接常驻逐字显示 token 这类对延迟敏感的场景

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

配合两个状态字段,UI 就活了:

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

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

流上传输的最小单元是 ConversationStreamChunk,认识即可:

块类型含义
message-delta文本/推理的增量片段,打字机效果的来源
tool-input/tool-output/tool-output-error工具调用的入参、结果、报错
message-started/message-completed一条消息的开始与结束
data-partAgent 主动写入的命名结构化数据
conversation-reset对话被重置,携带全新快照
submission-settled某次提交的终态结算

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

可靠性这块,SDK 替你把脏活全干了(流协议参考 有完整的线级说明):

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

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

新手最容易踩的四个坑

坑后果正确做法
把offset当数字做算术续读位置错乱、请求被拒把它当不透明字符串,原样存、原样传
拿到事件就渲染,不去重重连重放后同一内容"打出两遍"交给observe()维护状态;若自己消费原始流,按position去重
刷新页面直接重新 observe,不补快照开场白缺失,界面"凭空断片"先history()补一次快照,再让observe()接管实时部分
等send()的返回里出现回复永远等不到send()只做受理;结果走read()/wait()/ 事件流

我想要 X,就做 Y

你想要做法
聊天界面逐字显示observe()+subscribe(),按part.state区分生成中/完成
后台批处理只要结果read()一行拿回复文本
页面刷新后恢复现场先history()补快照,再observe()接管
确认某次提交成败wait()挂submissionId,resolve 成功 / reject 失败
审计监控原始事件wait()的onEvent回调,按position去重

延伸阅读

  • 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

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进