企业官网建设流程全解析

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

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

给 Agent 发了条消息，却石沉大海？先别怀疑网络——这是 Flue 事件流的"202 受理"设计，而非故障。本文带你用 3 步接上@flue/sdk，实时消费 Agent 的消息生成、工具调用与推理过程。

发完消息石沉大海？先别怀疑网络

你写下client.send(...)，await 完拿到一个 202 响应，然后……安静了。没有回复，没有中间过程，也没有报错。新手通常先怀疑三件事：自己代码有 bug、网络不通、模型卡死。其实都不是。

根因一句话：Flue 采用异步受理 + 事件流回读模型。服务器把消息"持久化受理"就立刻返回，Agent 可能还要跑上几分钟，回复和整个过程不会推给你，而是写进这场对话的事件流里等你去读。所以问题从来不是"怎么让send()把回复带回来"，而是"怎么订阅这条事件流"。

📺 现场直播 + 随时回放：一个比喻讲透它

接下来只用一个比喻：Agent 的执行是一场现场直播。

• send()是向导播台下单。导播台收单（HTTP 202）就立刻开给你一张回单，不会等节目播完再回话。
• 回单上有两个关键字段：submissionId是这单节目的节目单号——节目结束时，结算结果会带着它回来；offset是直播的回放点——一个时间戳，从它开始读，节目之后的所有内容都归你。
• 事件流就是直播信号本身。你可以随时"进直播间"，也可以倒回任意offset重放。断网重连、页面刷新恢复、多进程盯同一场对话，全都建立在这套回放机制上。

还有个细节能帮你少踩坑：offset是不透明断点，原样保存、原样回传即可，别解析它、别对它做算术——就像影院座位号，重要的是拿它换票，而不是研究它为什么是那个号。

跟着走一遍：从 send() 到逐字看到回复

整条链路是个三幕剧，每一幕决定你用五个方法（send/wait/read/observe/history）里的哪一个。

第一幕，下单。send({ message })投递消息，立刻返回submissionId、offset和流地址streamUrl。你不关心过程的话，后面可以只用它。

第二幕，进直播间。这一步最见取舍：

• 要做聊天界面或进度面板、逐字看→ 用observe()。它先拉历史快照补齐上下文，再切到实时尾巴持续跟进：默认'long-poll'长轮询，延迟敏感就传'sse'走 SSE 长连接，两种模式推的是同一批事件。它替你维护好一份可渲染的FlueConversationState，你subscribe()即可。
• 只想知道做完没、成没成→ 用wait(submission)。它盯住submissionId直到结算，只回"成功 / 失败"，不取回复，开销最小；想顺带看中间事件就挂一个onEvent回调。
• 想一行代码问一句答一句→ 用read(submission)。它替你组合wait()+history()，直接返回回复文本，后台批处理的首选。
• 页面刷新后恢复现场→ 先history()一次性读当前对话快照，再让observe()接管继续直播。

整个流程你真正要写的代码只有这一段（observe()是主力，其余方法按需加）：

import { createFlueClient } from '@flue/sdk'; const client = createFlueClient({ url: 'https://app.example.com/api/support/thread-42' }); const observation = client.observe(); observation.subscribe(() => { const { conversation } = observation.getSnapshot(); for (const message of conversation?.messages ?? []) { for (const part of message.parts) { if (part.type === 'text') renderText(part.text, part.state); } } });

留意两个能直接驱动 UI 的细节：part.state非'streaming'（正在生成）即'done'（已完成），打字机效果全靠它；快照里的phase字段走过loading → connecting → live → closed（出错为error），进度条直接订阅它即可。方法的完整形状见 client.ts 与 官方 FlueClient 文档。

流里跑的是什么：扫一眼就行 🧵

流的最小单元是ConversationStreamChunk（定义在 conversation-stream.ts），不用背，认识几个熟脸即可：

• message-delta— 文本/推理的增量片段，逐字打出的来源
• tool-input/tool-output/tool-output-error— 工具调用的入参、结果与报错
• message-started/message-completed— 一条消息的开始与结束
• data-part— Agent 主动附加的结构化数据
• conversation-reset— 整场对话被重置，携带全新快照
• submission-settled— 某次提交结算，wait()等的就是它

好消息是你几乎不用亲手处理它们：observe()已经把 chunk "归约"成完整状态。只有做底层调试、或在wait()的onEvent里看原始事件时，才会直接接触 chunk。

网络抽风时，断网重连为什么不丢消息

这是最容易被忽视、也最让人安心的部分，状态机在 observe.ts 里：

1. 至少一次投递。连接断了，传输层会从offset断点重放在途的那批事件——宁可重复，不可丢失。
2. 按position自动去重。每个 chunk 带单调递增的position，observe()丢掉已应用过的块，重复事件不会让界面"打出两遍"；重连时它会重新取一份干净快照，而不是在脏状态上继续补。
3. 僵尸流看门狗。流长时间没有心跳，SDK 自动取消死连接并重建，不会在半开连接上挂到天荒地老。
4. 鉴权退避。遇到 401/403 按退避策略重试，连续失败达到上限才放弃，不会一遇错就死。

一句话：网络抖动、页面刷新、进程重启，交给offset+ 自动重连兜底，你的业务代码只写"收到状态就渲染"。

照着抄就完事

• 如果你做聊天界面、要逐字显示 →observe()+subscribe()，渲染part.state
• 如果你跑后台批处理、只要结果 →read()一行拿回复
• 如果你要知道某次提交成败 →wait()，用submissionId关联
• 如果你要刷新后恢复现场 → 先history()补快照，再observe()接管
• 如果你做监控审计、要原始事件 →wait()的onEvent，按position去重
• 如果你要逐字低延迟推送 →observe({ live: 'sse' })，其余不变

一句话带走这个模型

send()下单、observe()看直播、wait()/read()等结算、offset兜底断网——"202 受理 + 事件流回读"就是 Flue 事件流的全部。

延伸阅读：SDK 客户端入口 client.ts、事件流迭代器 stream.ts、SDK 事件文档、流协议线级参考。

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