Flue Workflows 实战指南用 CLI、Node.js 脚本、Agent SDK 与 Durable Workflows 驱动 Agent【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue本篇技术指南以 Flue 官方文档 guide/workflows.md 为核心脉络系统讲解 Flue 中工作流Workflow的四种驱动方式终端单次运行flue run、Node.js 进程内的 Flue JS API、面向已部署 Agent 的 HTTP Agent SDK以及具备跨崩溃续跑能力的 Durable Workflows。读者读完将掌握每种模式的适用场景、完整可运行的代码示例以及它们背后的持久化与恢复原理——从一次 CI 脚本调用到生产级多步编排都能直接落地。什么是 Flue 中的 Workflow在 Flue 里任何运行了一个 Agent 的脚本或程序都被称为工作流。需要强调的一点是Workflow 本身并不是 Flue 的一项独立特性而是一种被官方认可、便于指代的使用模式——它覆盖了在常规的部署式聊天机器人体验之外用 Flue 搭建脚本化自动化的各种方式。这一定义决定了工作流的适用范围非常宽你可以用终端命令跑一次 Agent 完成工单分流用定时任务在凌晨生成日报用 HTTP 客户端与线上 Agent 对话也可以用 Cloudflare Workflows、Inngest 等托管编排平台驱动多步业务流程。关于如何先部署一个 Agent参见部署指南。选择方案的决策表具体选哪种方式做可编程自动化通常取决于代码在哪里运行以及运行结束后你需要拿回什么方案适用场景flue run需要从终端初始化并提示一个本地 Agent最适合 CI 工作流。Flue JS API需要从 Node.js 初始化并控制本地 Agent最适合本地脚本、cron 定时任务等。Flue Agent SDK需要通过 HTTP 初始化并控制托管已部署Agent最适合与线上生产 Agent 对话或监听其输出。Durable Workflows需要在托管运行时中初始化并控制托管 Agent且要求持久性保证最适合多步编排和必须扛住中断的产品。这些方式并非互斥一个 durable workflow 内部用的还是与独立 Node.js 脚本相同的start()和init()API一个 CI 任务本质上也是把你在终端敲的那条flue run命令包进了 shell step。flue run在终端里初始化并提示本地 Agent最小的工作流就是一次flue run调用。它在本地进程中加载 Agent 模块、提交一条消息、把最终回复打印到 stdout然后在运行落定settle时退出——退出码直接报告成功或失败flue run src/agents/triage.ts --message Triage issue 17307. --id issue-17307关键设计是输出通道的分离除了最终回复以外的一切流式文本、工具活动、状态行都写到 stderr因此 stdout 始终可被管道消费传入--json后stdout 不再输出纯文本回复而是输出一个结构化结果信封。有了这两点就能在 shell 脚本里把多个 Agent 串起来summary$(flue run src/agents/reporter.ts -m Summarize yesterdays deploys. --json | jq -r .message) flue run src/agents/notifier.ts -m Post this summary to #eng: $summary对话在多次调用之间是持久化的存储在项目配置的数据库中所以复用同一个--id可以跨多次运行延续同一段对话。在 CI 中承载周期性工作流同一命令在 CI 的任意 shell step 里都能运行这让 CI 成为托管周期性工作流最省事的地方。Provider 凭据来自 job 的环境变量--new配合确定性的--id能让对话创建做到 exactly-once——被重试的 job 不会重复创建对话jobs: triage: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - run: npm ci - run: npx flue run src/agents/triage.ts --message Triage issue #${{ github.event.issue.number }}. --id issue-${{ github.event.issue.number }} --new --json triage.json env: ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}triage.json里的信封携带 outcome、回复文本和 conversation id供后续步骤使用。GitHub Actions 与 GitLab CI 的平台细节分别见GitHub Actions 生态页与GitLab CI 生态页。完整参数与输出约定flue run的完整参数列表来自 CLI 参考参数说明path要运行的 Agent 模块文件路径相对当前工作目录解析。-m, --message text提交给 Agent 的用户消息必填。--name agent模块定义了多个 Agent 时选择运行哪一个模块导出多个 Agent 时必填没有静默默认值。--id id要创建或继续的对话 id默认生成一个新的 ULID 并打印到 stderr。--data json实例创建数据JSONAgent 内用useInitialData()读取仅在本次运行创建对话时生效继续已有对话时被忽略不能与--uid同用。--uid uid仅继续指定 uid 的对话实例不能与--new或--data同用。--new仅创建当对话 id 已存在时本次运行被拒绝。--json向 stdout 输出 JSON 结果信封而非回复文本。--env path在运行前加载一个备选.env格式文件替代默认.env。--json模式下的 stdout 始终是恰好一个信封按outcome区分{ id: support-4821, agent: hello, submissionId: …, outcome: completed, message: The final assistant reply., uid: inst_… }outcome: completed携带message最终回复outcome: failed与outcome: aborted携带error对象{ message, type?, details?, dev? }底层错误为 Flue 错误时才有类型化字段而非回复运行开始前的设置/准入失败模块解析、配置、创建数据校验打印{ outcome: error, error: { … } }。信封是对退出码的补充而非替代0表示 completed1表示 failed 与设置类错误130表示 aborted。这些输出约定与参数校验如--uid与--new、--uid与--data互斥都能在 CLI 实现源码 packages/cli/src/commands/run.ts 与信封组装逻辑run.ts中找到对应实现。Flue JS API在 Node.js 脚本中控制本地 Agent当工作流需要的不仅是flue run——循环、错误处理、复杂数据结构——就写一个 Node.js 脚本。start()会在你自己的进程内启动 Flue 运行时没有 server、没有app.ts见独立脚本init()返回一个 Agent 对话的句柄。句柄的dispatch()提交一条消息并以持久化收据durable receipt解析read()等待落定后的回复import { init } from flue/runtime; import { sqlite, start } from flue/runtime/node; import { Reporter } from ../src/agents/reporter.ts; await using flue await start({ agents: [Reporter], db: sqlite(./nightly.db), }); const reporter init(Reporter, { id: nightly-2026-07-17 }); const receipt await reporter.dispatch(Produce the nightly report.); const reply await reporter.read(receipt); console.log(reply.text);失败或被中止的运行会让read()以AgentRunError拒绝所以脚本只需要普通的try/catch就完成了全部错误处理。db选项决定对话是否比脚本活得更久省略它则使用进程退出即消失的内存态配置一个数据库适配器后之后的运行可以继续同一段对话。关于句柄本身Agent API 参考给出了一个值得记住的实现细节init()创建的句柄是一个地址address而非资源——init()本身不创建任何东西、不做任何 I/O实例在首次接触时才创建运行时的解析发生在句柄被使用之时。因此在模块作用域调用init()是安全的同时init(agent, { id, uid })的选项只接受id/uid把id/uid放进消息 payload 会直接抛错。init()句柄的其余 API 面参见其参考页。Flue Agent SDK通过 HTTP 与已部署 Agent 对话Flue Agent SDKflue/sdk通过 HTTP 连接一个已部署的 Agent发送消息并流式接收响应。一个客户端包裹一个对话 URLimport { createFlueClient } from flue/sdk; const conversation createFlueClient({ url: https://example.com/agents/release-auditor/release-${version}, token: process.env.FLUE_TOKEN, }); const admission await conversation.send({ message: { kind: user, body: Audit the ${version} rollout. }, }); const reply await conversation.read(admission); console.log(reply.text);SDK 在 HTTP 之上复刻了本地句柄的模型send()在准入admission时即以提交的标识符解析服务端返回 202此时 Agent 尚未运行read()等待该提交落定并以回复解析失败或被中止时抛出FlueExecutionError。read()也接受一个裸的 submission id——这样只持久化了准入信息的进程之后可以重新挂接与本地句柄read()的恢复故事一致。选择依据是 Agent 在哪里运行脚本自己运行 Agent 就用start()Agent 在某个部署里运行就用 SDK。客户端的方法面与发送选项createFlueClient返回的FlueClient接口定义于 packages/sdk/src/client.ts包含url客户端地址的完整对话 URL、send()202 准入、read()等待落定并读取回复、wait()仅等待落定不取回复、abort()中止对话中在途与排队的工作、history()读取物化快照、observe()跨历史补齐与实时更新的持续视图与attachmentUrl()解析附件字节的 URL。send()的选项packages/sdk/src/public/send.ts除message外还包括initialData实例创建数据仅在本次发送创建对话时生效配合uid: null可在已存在时报错uid发送条件——省略为无条件继续或创建实例传字符串表示仅继续该 uid 的化身不存在或 uid 不匹配返回 404 且不投递传null表示仅创建实例已存在返回 409agent_instance_exists其 uid 在error.meta.uididempotencyKey调用方命名的投递键重试携带相同键会收敛到原提交而非准入重复项重放时deduplicated: true键复用但 payload 不同则返回 409submission_conflict。构造选项url、fetch、headers、token的完整说明见 createFlueClient 参考其中fetch覆盖使得 SDK 可以跑在任何 fetch 形态的传输上例如 Cloudflare 的 service binding 或测试用的 canned Response包为 ESM-only、可在任意支持fetch的环境运行浏览器、Node.js、edge runtime。整体概览参见 SDK 总览。Durable Workflows让脚本本身跨崩溃续跑缺口Flue 保证单次发送不保证发送之间的脚本Flue 已经为每一次 send 保证了持久化的 outcome消息一旦被准入该 submission 就会穿越崩溃、重启与重新部署最终落定——这份契约由持久化Durability指南完整覆盖。但 Flue不保证发送之外的那段脚本一个在两次 dispatch 之间死掉的工作流会从头重新执行。对快速脚本这没问题但对一旦启动就必须完成的多步编排就不够用了。这正是durable workflow要填补的空缺一个托管脚本其每一步都检查点checkpoint自己的结果从而能够重试失败的步骤、恢复跨越数天的运行、在重启后不丢进度。Cloudflare Workflows、Inngest、Temporal 都是构建在这一原语之上的产品Flue 与它们不需要任何特殊集成——在你选择的平台上写 durable workflow然后把 Flue 当作普通服务来调用即可。通用模式给收据单独开一个 step在下面的示例中dispatch 运行在自己的 workflow step 里因此收据——submission 的持久化取票凭证——在它一产生的瞬间就被检查点记录第二个 step 再读取落定后的回复。一个已完成 dispatch 的 step 永远不会重发一个崩溃的 read step 会用同一张收据重新挂接而不是再次提示 Agent。这个拆分同时把每步的时间上限也拆开了一个 20 分钟的 step 超时端到端变成最多 40 分钟。如果操作带有单一截止时间就在 dispatch step 的结果里检查点该截止时间让 read step 强制执行剩余部分。示例Cloudflare Workflows在 Cloudflare 上把 Workflow 写在与你 Flue 应用同一个 Worker 里在 step 中直接调用init()句柄import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from cloudflare:workers; import { init } from flue/runtime; import { Reviewer } from ./agents/reviewer.ts; import { collectFindings, fileReport } from ./shared/nightly.ts; type Params { date: string }; export class NightlyReview extends WorkflowEntrypoint { async run(event: WorkflowEventParams, step: WorkflowStep) { const findings await step.do(collect findings, () collectFindings(event.payload.date)); const agent init(Reviewer, { id: nightly-${event.payload.date} }); const receipt await step.do(dispatch review, () agent.dispatch(Review these findings:\n${findings}), ); const review await step.do(read review, async () { const reply await agent.read(receipt); return { text: reply.text, data: reply.data }; }); await step.do(file report, () fileReport(review)); } }该类从src/cloudflare.ts导出其 workflow binding 在wrangler.jsonc中声明wrangler.jsonc只声明应用自有资源Flue 生成的FLUE_*binding 不要手写详见 Cloudflare target 指南。其余平台细节同样见该指南。示例Inngest与 Temporal 的同一模式在 Inngest 中发送放在step.run内部。函数运行在 Flue 应用进程内时调用init()作为独立服务运行时改用 Agent SDKimport { init } from flue/runtime; import { inngest } from ./client.ts; import { Reviewer } from ../agents/reviewer.ts; import { fileReport } from ../shared/nightly.ts; export const nightlyReview inngest.createFunction( { id: nightly-review }, { event: reports/nightly.requested }, async ({ event, step }) { const agent init(Reviewer, { id: nightly-${event.data.date} }); const receipt await step.run(dispatch review, () agent.dispatch(Review the nightly findings.), ); const review await step.run(read review, async () { const reply await agent.read(receipt); return { text: reply.text, data: reply.data }; }); await step.run(file report, () fileReport(review)); }, );同一模式适用于任何其他 durable workflow 引擎在 Temporal 里dispatch 与 read 各自放进一个 activity 即可。崩溃后重新挂接Re-attachingread()不持有任何内存态落定结果与回复都是持久化的对话记录因此任何进程在任意时刻都能读取某个 submission而一个在 workflow 宕机期间已落定的 submission 会立即解析。这正是收据要单独占一个 step 的原因——一旦引擎检查点记录了它read step 的每次重试都重新挂接到同一个 submission而不会再次提示 Agent。唯一剩下的崩溃窗口在 dispatch step 内部发送已被准入但 step 在检查点收据之前死掉了。引擎会重跑该 step、再次发送——此时实例的发送条件决定了重发的含义无条件发送无uid重发会投递一条重复消息它会在一个 turn 边界加入实时响应两个 submission 都以同一份合并后的回复落定所以重试拿到的新收据读到的是同一个答案。仅创建发送uid: null重复发送在准入时即被拒绝抛出AgentInstanceExistsError——消息永远不会两次到达 Agent而这个拒绝就是 workflow 失败本次运行或回退的信号。与底层持久化契约的关系上述模式成立的前提是 Flue 的 accepted-work 契约每次 send 都是一条被持久化记录的submission在任何模型工作开始之前payload 就已落盘submission 按准入顺序构成持久队列每次处理以attempt为单位中断消耗一次 attempt恢复则申领新 attempt直到耗尽重试预算默认 10 次 attempt、每个 submission 一小时墙钟超时可用durabilitystatic 按 Agent 覆盖。整体纪律是at-least-once 执行 exactly-once 记录已持久提交的工作已记录的回复、工具结果、状态写入永不重跑中断前未提交的工作在下一次 attempt 重跑。恢复决策矩阵、重试预算与超时、durable tools 与step.do、delegated tasks 等完整契约见持久化指南。值得注意的是Flue 不会检查点任意 TypeScript 执行并从最后一行恢复——检查点边界就是 Agent 本身Agent 之外的 endpoint、脚本或 cron job 正是要靠你平台的 workflow 引擎Cloudflare Workflows、Inngest 或朴素重跑来提供。延伸阅读持久化Durability——每一次 send 背后的 accepted-work 契约、attempt 与重试预算。Schedules定时任务——时间触发的 dispatch应用内与外部皆可。flue run参考——完整 CLI 参数与输出信封约定。SDK 总览——面向已部署应用的对话客户端。部署指南——承载这些工作流的应用如何托管上线。【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考