面向Agent的轻量Workflow引擎设计与实践

最近在搭一套多Agent协同调度模块时被一件事反复折磨Agent的任务链路越来越长中途要调工具、要等人审、要并行跑子任务代码里if-else堆到没法维护。我意识到缺的不是模型能力而是一个能把这些步骤“编排”起来的东西。网上搜了一轮市面上成熟的Workflow引擎基本都偏重——部署一套要起好几个服务配置模型复杂光是理解它的抽象概念就得花一周。对Agent场景来说这套东西太重了杀鸡用牛刀还给鸡添负担。最后我决定自己设计一个面向Agent的轻量Workflow设计目标很明确核心逻辑能在一千行内讲清楚部署时只要一个进程Agent只需按约定上报状态剩下的编排、重试、恢复全部交给Workflow层。这篇文章就记录我整个的设计思路、取舍过程和落地实测。如果你是正在做Agent应用、被任务编排搞得头疼的开发者或者想理解轻量工作流内部到底怎么回事这篇应该能给你一个可以直接抄作业的参考。1. 起因传统流程引擎在Agent场景下的别扭1.1 我的实际痛点是什么我们最初用的是任务列表手动推进的方式。每个Agent执行完一步往数据库写一条记录下一步的触发靠定时任务扫描。听起来简单但Agent场景有个特点任务不是固定流水线而是高度动态的。比如一个“生成市场分析报告”的任务可能要先判断数据源是否可用可用就并行拉取三个渠道的数据不可用就要进入人工确认分支。人工确认完成后还要重新评估分支条件。这种动态拓扑用固定状态字段根本描述不了。我代码里不得不维护一张“当前任务类型”的表每种类型配一套推进逻辑加一个新任务类型就要动核心代码。更痛苦的是失败处理。Agent调外部API超时了怎么办重试几次重试间隔多久如果Agent进程崩溃重启当时跑到哪个步骤了这些逻辑散落在各个Agent实现里每个Agent写的还不一样。有的Agent重试三次有的重试一次就放弃排查线上问题时根本不知道某个任务卡住是因为什么。1.2 为什么不用现成的Workflow引擎我评估过两个方向。一个是重型工作流引擎它能做复杂状态机、分布式事务、可视化编排但部署成本太高。在我这个场景里Agent本身就够复杂了再引入一套需要独立维护的编排系统运维负担直接翻倍。另一个是云厂商提供的Serverless编排服务功能倒是全但任务是跑在我们自建的Agent容器里跨网络调用的延迟和费用都吃不消。我也认真想过另一个思路完全不用Workflow让Agent自己负责所有步骤。这个方案确实最轻但它把Agent搞成了一个上帝类既要懂业务又要懂调度。而且一旦任务涉及多个Agent协作比如A做完B做B的结果要回传给A没有中间层根本没法管理。我得有一个能记录“当前谁在做什么、做到哪一步”的公共视图。1.3 我定义的“轻量”边界在哪基于这些痛点我给自己定了几条硬约束整体代码量控制在两千行以内核心执行器不超过八百行不依赖独立服务嵌入现有Agent进程即可用数据库一张表存Workflow实例状态不引入额外的存储系统支持顺序执行、并行执行、条件分支、失败重试、人工审批五种基本能力任何操作都要能在五分钟内向新人讲清楚轻量的核心不是功能少而是剥离一切“在这个场景下不常用”的能力。我不需要可视化拖拽编排不需要高可用集群不需要分布式事务。我需要的是任务能跑起来、失败了能恢复、出问题能定位。这三件事做到就已经覆盖了Agent编排90%的诉求。2. 核心模型设计一张表加一个循环就够了2.1 任务节点的数据结构设计这个Workflow时我把自己代入一个Agent的视角去思考Agent接到的指令应该是什么它只需要一个节点描述对象。最终我定义的数据结构是这样的dataclass class TaskNode: node_id: str # 节点唯一ID比如 step_1 agent_type: str # 要调度哪个Agent比如 data_fetcher payload: dict # 传给Agent的具体参数 depends_on: list[str] # 依赖哪些节点空表示无依赖 next_on_success: str # 成功后进入哪个节点 next_on_failure: str # 失败后进入哪个节点 retry_count: int 3 # 最大重试次数 retry_interval: int 30 # 重试间隔单位秒 timeout: int 300 # 单次执行超时时间depends_on这一个字段解决了并行问题。我把依赖关系显式声明出来执行器只需要检查“我的依赖都完成了吗”就能判断自己是否可以启动。这样避免了在代码里写死调度顺序拓扑结构完全由数据驱动。next_on_success和next_on_failure则解决了分支问题。传统Workflow引擎用专门的分支节点类型我觉得没必要。普通节点的成功或失败本身就是天然的布尔分支条件。只要在节点定义里写明成功往哪走、失败往哪走分支能力自然就有了。2.2 状态机的流转规则有了节点定义我还需要一个表示整体执行进度的状态机。我用了最直白的六状态模型状态含义触发条件PENDING待执行节点创建等待依赖完成RUNNING执行中依赖已完成Agent领取任务SUCCEEDED已成功Agent返回成功结果FAILED已失败Agent返回失败结果RETRYING等待重试失败但未超过重试次数IN_REVIEW待人工审批进入审批节点等待人审结果这六种状态的流转规则我写成了一个纯函数方便单元测试和排查def build_mermaid(instance_id): print(mermaid_code)不对方法想岔了实际我是写成状态转移表来校验合法转移。比如 RUNNING 只能变成 SUCCEEDED、FAILED 或 RETRYING绝对不允许直接跳到 PENDING。这些规则保证状态机不会出现“死循环”或“跳变”的诡异情况。状态的归属粒度是“节点”而非“整个工作流”。整个Workflow实例也有一个总状态但它是根据节点状态实时计算出来的只要存在RUNNING节点就是RUNNING只要有一个FAILED节点就是FAILED全部SUCCEEDED就是完成。这样省去了一堆状态同步的逻辑。2.3 执行循环心跳式调度我设计执行器的核心是一个死循环加一个“可运行节点查询”def scheduler_loop(): while True: runnable db.query( SELECT * FROM nodes WHERE status PENDING AND EXISTS (SELECT 1 FROM nodes AS deps WHERE deps.node_id ANY(nodes.depends_on) AND deps.status SUCCEEDED) ) for node in runnable: assign_to_agent(node) time.sleep(SCHEDULE_INTERVAL)这个查询表达的核心逻辑是找所有依赖都已完成且自身还没启动的节点。之所以用“轮询”而不是“事件驱动”是为了简化恢复逻辑。如果某个Agent领取任务后崩溃了节点状态会一直停在RUNNING直到超时。轮询模式下超时检测是顺手的事而事件驱动的模式里丢失的事件会留下永久的挂起节点。执行器采用单线程循环。并行能力不靠执行器的多线程而是靠“一次调度多个节点分配给多个Agent”。每个Agent本身就是独立进程天然并行调度器只管分发和回收不需要自己处理并发竞争。3. 编排能力的低成本实现顺序、并行、分支、重试与人审3.1 顺序执行与条件分支的约定顺序执行其实是最简单的。只要在定义工作流时让每个节点的depends_on指向前一个节点执行器自然按顺序跑。我习惯把整条链的next_on_success也串联起来这样即使依赖断了成功路径还是清晰的。条件分支我分成了两种实现方式取决于分支条件的位置。一种是在调度层判断前置节点成功后调度器根据返回值决定把哪个后续节点置为可运行。比如“数据源可用性检查”节点返回{available: true}调度器就把“拉取数据”节点置为可运行返回{available: false}就把“人工确认”节点置为可运行。这个逻辑我放在一个叫decide_next的钩子函数里def decide_next(node_result, workflow_definition): if node_result.get(available) is True: return fetch_data_node return human_review_node另一种是在Agent内部判断Agent自己根据业务逻辑返回不同的成功标记比如{success: true, action: retry}或{success: true, action: giveup}。调度器根据action字段决定走next_on_success还是直接跳到终态。这种方式适合“业务判断比较复杂、但路径固定”的场景把决策留给Agent自己做。3.2 并行分支与聚合并行是这套设计里最讨巧的部分。前面说了depends_on字段天然描述并行关系。如果三个节点都只依赖同一个父节点那么父节点成功后三个节点会同时被调度器发现并分配给三个Agent。但并行之后往往要聚合——等三个任务都完成才能进入下一步。这时我引入了一个虚拟聚合节点。它的Agent类型是aggregatordepends_on填写那三个并行子节点{ node_id: aggregate_1, agent_type: aggregator, payload: {mode: join, sources: [child_1, child_2, child_3]}, depends_on: [child_1, child_2, child_3] }聚合节点本身不调用外部Agent而是由调度器内置的聚合逻辑处理。它会等待所有依赖节点返回SUCCEEDED收集它们的输出结果组装成一个聚合结果然后传递给下游节点。这里有个坑我必须提醒千万不要在并行子节点里写数据库或做有副作用的操作因为并行出错重试时副作用可能叠加。我踩过这个坑后面专开一节讲。3.3 重试、超时与人工审批的实现位重试和超时理解起来不复杂但实现位置的选择会影响整个系统的行为。我把重试逻辑交给调度器统一处理。节点执行失败时调度器不会立刻把节点置为FAILED而是检查重试次数。如果还有剩余次数就把状态置为RETRYING并记录下次重试的绝对时间戳调度循环扫描到当前时间超过时间戳时再次把节点分配给Agent。超时检测则利用了状态时间戳。每个节点记录started_at调度循环里每个周期检查一次当前时间减去started_at如果超过节点的timeout直接判断为执行失败。这个逻辑很朴素但足够可靠——Agent崩溃、容器被杀、网络分区导致消息丢失最终都会表现为“超时”超时机制把它们统一收编了。人工审批的实现则是一个特殊的Agent类型。节点进入RUNNING时调度器不是去调外部模型而是给审批人推一条待办通知然后进入等待。审批人在管理后台点了“通过”或“拒绝”接口会直接修改节点状态。为了让流程不卡死我给审批节点也加了超时——超时后默认拒绝并触发失败分支。实际上线后发现没有超时上限的话一个流程可能被人为晾一周下游全部堵塞所以超时这个设计保留了。4. 持久化与断点恢复如何做到崩溃后接着跑4.1 状态持久化只落两个时机为了做到进程崩溃后能恢复持久化是躲不开的。但追高频的持久化会拖垮性能。我定的策略是只有两个时机强制写数据库。第一个时机是节点状态发生转移时。比如从PENDING变成RUNNING从RUNNING变成SUCCEEDED这些状态变更必须同步落库确保“当前进度”不会丢。第二个时机是Agent领取任务时。调度器把任务派发出去后会立即把节点状态置为RUNNING并写入当前执行的Agent实例ID。这样做的好处是崩溃后恢复时我知道这个节点的任务被谁领走了可以去查它的最终状态。Agent执行过程中的中间日志不落库。比如Agent每一步日志、临时计算结果这些丢了我认因为它们是可重算的。只有“节点状态”这种不可丢失的进展信息才需要持久化。这个取舍遵循一个原则状态是硬数据过程是软数据硬数据必须保软数据丢了重跑就行。4.2 恢复流程与幂等设计进程重启后的恢复逻辑比我预想的要简单。核心就三步def recover_after_restart(): # 1. 找到所有 RUNNING 且超时的节点 hanging db.query( SELECT * FROM nodes WHERE status RUNNING AND started_at now() - interval 5 minutes ) for node in hanging: # 2. 标记为执行失败走重试逻辑 mark_failed(node, reasonagent crash, no heartbeat) # 3. 其余状态不变调度循环自动继续所有处于PENDING、RETRYING状态的节点恢复后天然会被调度循环重新发现。RUNNING节点如果还在正常执行会正常返回如果Agent已经死了等超时后被置为失败并触发重试。这里最精妙的一点是恢复逻辑不需要遍历所有节点只需要处理“悬挂”的少数节点。但“重新跑”是有风险的所以我给所有会触发外部副作用的节点设计了幂等策略。核心方法是给每个节点分配一个全局唯一的execution_id这个ID会被写进外部API请求的幂等键里。比如Agent调支付接口、写文件、发消息时带上这个ID外部系统看到同样的ID就认为是重试直接返回上一次的结果避免重复扣款和重复发消息。5. 实测某跨平台系统接入这套Workflow的真实表现5.1 从领任务到结果回传的完整链路我拿一个实际业务来验证整个设计某跨平台系统的“多平台内容同步发布”服务这个任务链路包含数据收集、内容审核、多平台发布、结果汇总四个大步骤。数据收集阶段并行调用了三个平台的接口每个平台单独一个Agent。三个平台数据都回来后进入审核Agent判断内容是否符合各平台规范。审核通过后进入发布阶段发布这个动作我拆成了三个并行节点分别发布到三个平台。全部发布成功后汇总节点收集三个平台的返回结果并写入最终报表。整个流程共8个节点。我把这套流程定义成JSON放在数据库里5分钟完成配置。第一次跑的时候三个并行的数据收集Agent同时启动确实做到了并行。其中一个平台的接口当时因限流返回502数据收集节点走了两次重试后成功整个过程没有人工介入。完整链路跑完从开始到汇总报告生成花了约40秒比之前手动拼接流程快了将近3倍。5.2 并发、容错和性能实测数据我关心三个指标调度器本身的吞吐、节点状态变更的延迟、以及容错场景下的恢复时间。调度器单循环间隔我设的是1秒一次轮询最多拉取50个可运行节点。实测中50个节点全部分配出去从“符合调度条件”到“所有Agent领取完毕”耗时约1.2秒瓶颈在数据库查询不在调度器本身。对Agent场景来说这个吞吐完全够了。容错场景我用了一个粗暴但有效的测试方法在聚合节点执行前直接kill掉调度器进程等了10秒再重启。重启后恢复逻辑扫描到有3个已完成的节点和1个悬挂的RUNNING节点。悬挂节点被标记为失败并触发重试Agent重新执行后成功。整个恢复过程耗时约3秒没有人工干预。后来我在代码注释里写了一句评价最满意的就是这里崩溃恢复在软件层面几乎没成本。性能方面状态变更落库一次的平均耗时约8毫秒。对单节点多次重试、多分支并行这类常见情况状态变更的总次数通常不超过20次数据库压力完全可以忽略。我没有做一万并发这种压力测试因为这不是轻量方案的目标场景硬去够那种规模只会把架构重新逼回复杂。6. 踩过的坑和必须提前给后来者说的话6.1 三个印象最深的坑第一个坑是并行节点的幂等处理。最初设计时我完全没考虑重试带来的副作用叠加。某次数据收集节点在重试时向内部消息队列重复推送了两个完全相同的数据包下游统计直接翻倍。排查时我看到节点重试日志才反应过来问题出在重新执行任务时没有使用幂等键。后来给所有写入操作的外呼都固定带上execution_id前后对比才正常。第二个坑是恢复逻辑只处理了RUNNING节点没处理“PENDING但依赖已完成”的节点。事情是这样的调度器把节点A分发给Agent后Agent执行完也返回了但调度器在写库前崩溃了。恢复后节点A还是RUNNING状态它不是悬挂状态但永远不会被重新调度。那段时间我有几个任务卡在“执行中”一周没动。解决办法是恢复时把所有RUNNING且没有心跳超时的节点也强制重跑一遍代价是偶发重复执行但在幂等键保护下可以接受。第三个坑是审批节点的超时策略。最初我把审批节点的超时设成无限大结果一个需要人工确认的任务被审批人晾了两个星期下游整条链路瘫痪。后来加了一个总量超时——整个工作流的执行期限超过直接失败。审批确实可能耗时但系统必须有兜底。我现在建议给每个审批节点单独设置一个合理的超时上限而不是用全局默认值。6.2 如果让我重来哪些设计我会保留重新回看这个方案有三件事我做得比较对。第一是坚持用“数据驱动而非代码驱动”的编排。所有流程定义都是JSON新增一个流程不需要改代码、不需要发布版本运营人员在后台填一张表就能上线。这让迭代速度提高非常多。第二是状态的“级联简化”。我一开始也想设计复杂的子任务关系、任务组后来一刀切每个节点只有六种状态节点之间只有依赖关系整个工作流的总状态从节点状态推导。简单模型带来的直接好处是排查问题方便看到节点状态就知道卡在哪不需要一层层展开去看父子关系。第三是“写完执行器再写恢复器”。很多设计文档喜欢把恢复机制当成附加模块我这次是先把正常执行路径的代码写完紧接着就把恢复逻辑写成同等级核心组件并且每周做一次kill进程恢复测试。恢复能力不是上线前临时加的保险而是与主流程并列的基础设施。如果你也在给Agent场景设计Workflow我的建议是先把需要的能力压缩到五个以内用最直接的数据结构表达清楚然后让执行器足够“笨”笨到把所有聪明都留给数据定义。这种方向下做出来的东西维护成本会低到超出你的直觉预期。