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 受理submissionIdoffset出口同一条对话的事件流持续吐出消息、工具调用、结算等事件块续传点每次读响应都会带下一个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.stateobserve()的实时尾巴有两种传输方式stream.ts 中的FlueEventStream统一承载模式特点什么时候选long-poll默认长轮询兼容性好大多数场景的稳妥选择sseSSE 长连接连接常驻逐字显示 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 替你把脏活全干了流协议参考 有完整的线级说明offset是不透明断点每次响应带回下一个值原样保存、原样回传。别解析它别对它做加减。至少一次投递连接中途断开时传输层从断点重放在途的那批事件——宁可重复不可丢失。按position自动去重每个块都带单调递增的positionobserve()会丢弃重放的重复块界面上不会出现同一段话打出两遍。自动恢复内置僵尸流看门狗流长时间无动静就自动取消重连遇到 401/403 按退避策略重试而不是立刻失败。结论一句话网络抖动、刷新、重启你不用写一行补偿逻辑业务代码只负责收到状态就渲染。新手最容易踩的四个坑坑后果正确做法把offset当数字做算术续读位置错乱、请求被拒把它当不透明字符串原样存、原样传拿到事件就渲染不去重重连重放后同一内容打出两遍交给observe()维护状态若自己消费原始流按position去重刷新页面直接重新 observe不补快照开场白缺失界面凭空断片先history()补一次快照再让observe()接管实时部分等send()的返回里出现回复永远等不到send()只做受理结果走read()/wait()/ 事件流我想要 X就做 Y你想要做法聊天界面逐字显示observe()subscribe()按part.state区分生成中/完成后台批处理只要结果read()一行拿回复文本页面刷新后恢复现场先history()补快照再observe()接管确认某次提交成败wait()挂submissionIdresolve 成功 / reject 失败审计监控原始事件wait()的onEvent回调按position去重延伸阅读SDK 客户端与五个方法定义client.ts事件流迭代器重连、断点重放stream.ts观察器状态机去重、看门狗、重试策略observe.tschunk 协议与归约逻辑conversation-stream.ts结算等待与wait()实现settle.ts官方事件文档sdk/events.md线级流协议参考streaming-protocol.mdFlueClient 完整 API 参考flue-client.md想实时盯住 Agent 的每一刻核心就一句话send()把任务交出去剩下的交给事件流——断点续读、去重、重连全是它的事你只管把渲染好的状态画出来。【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考