Skip to content

用现有机制组合业务流程 ​

本页面向需要审批、批处理或 Agent 工具循环的开发者。先确定业务需要保存什么、等待什么,再选择 Graph、节点、切面和 Durable 的组合。无需为每一种业务名称增加一种引擎节点。

分支、汇聚与批处理 ​

分支用 Set / Skip 明确表示有值或正常缺失。默认汇聚等各路到达后,只要至少一路有值就进入业务体;用 TryGet 分别处理缺项。要求所有输入都有值时,显式返回 WaitAll 的结果。跳过可以正常完成,真实失败仍保留错误。

固定数量的并行步骤可用 FanOut 或普通节点。数量由本轮输入决定时,可以把切片交给一个处理节点,在业务体中遍历,或让宿主按批次创建独立 Execution。只有确实需要把每一项都显示为静态节点时,才在运行前根据批次装配一份图。已封闭的共享 Graph 不在运行中追加节点,slot 也不重复打开。

独立 Execution 的额度分别计算。Sub 默认继承父执行的额度池;需要整棵静态子树共享同一份额度时,子图不要另行声明 WithMaxRunning。子图显式额度会覆盖继承,包括用 0 退出限制。多个独立根执行不会自动获得同一棵树的额度或检查点作用域。

审批放在切面的推进位置 ​

例如构建完成后等待批准,再部署。审批切面在 next(rc) 前检查批准记录,批准后才推进业务体。这个等待位置尚未占用节点业务额度,所以额度为 1 时,其他独立节点仍可工作。

短等待应监听 rc.Context().Done(),让宿主能够取消。可以用 pulse.NamedAspect("approval", gate) 记录切面的真实进出,但这不会自动创建业务审批任务;审批状态和审批人等仍由业务系统管理。

长等待需要释放进程时,由宿主组合 Durable:

  1. 先可靠保存带稳定 RunID 和步骤身份的业务等待记录。
  2. 取消本次 Execution,等待运行返回并释放资源。
  3. 收到批准后,重建同版本定义和输入,以原 RunID 调用 Resume。
  4. 切面再次检查批准记录;已确认的上游成果由 Durable 恢复,未完成步骤重新进入。

业务任务可以仍处于 waiting,本次 Execution 则是 canceled。不要返回普通的“等待中”错误代替这个协议:普通错误会消耗失败预算。正常 skip 也不表示等待,它表示本轮没有这项结果。

等待登记、取消和唤醒之间的中断窗口、重复事件及恢复扫描由宿主协议处理。完整说明和真实进程恢复用例见 Durable 组合指南。内置唤醒服务和跨机器租约仍是独立的扩展方向。

在节点里循环调用 Graph 工具 ​

一个 ReAct 形状节点可以在业务体中保存当前问题、工具结果和停止条件。工具目录把名字映射到调用函数;函数给可复用工具图提供本轮 Inputs,创建新的 Execution,等待结束后用 ReadOutput 读取声明出口。

每次工具调用都是新的执行,不修改工具图定义。循环使用 rc.Context(),处理每次调用的错误,并设置业务停止条件或最大轮数。引擎不要求模型服务,也不限制决策来自模型还是普通 Go 代码。

这种组合能完成循环和工具选择。单个节点内的循环变量不会自动逐轮持久化;如果要求进程退出后已经完成的工具调用不重做,需要稳定的轮次/调用身份及持久记录,例如由宿主为每次调用建立独立 Durable 运行。独立工具根执行也不会自动挂入调用节点的静态 Sub 树。

连续消息与渐进输出 ​

流水线使用 ProduceStream / TeeStream / ConsumeStream。受管等待让出业务额度,Ack 释放在途容量;宿主实时读取使用在 Start 前登记的 SubscribeStream。有限消息处理用 HandleMessage,渐进来源按显式位置使用 SendAt,不要把整节点 Retry 当成已发送数据的自动倒带。

完整程序见流式处理。外部数据库、HTTP 或文件操作的防重仍由业务幂等或事务协议负责。

可运行的组合证据 ​

仓库的 组合用例 验证两件事:审批切面等待时不占唯一业务额度;一个循环节点连续两次调用同一工具图并得到独立执行的结果。

sh
go test -race -count=1 -run '^(TestApprovalAspectLeavesQuotaAvailableBeforeNext|TestLoopNodeCallsReusableGraphTool)$' ./tests

这组确定性替身证明组合机制;模型质量、外部审批服务和逐轮持久恢复各自需要对应的业务验收。跨进程保证应以 Durable 的真实退出/重启用例为依据。

开发版 API 文档 · Go 1.26+ · MIT