如果你最近半年在做 Agent 应用大概率遇到过这个场景用户提一句“帮我整理上个月的报销单、生成汇总、再发给财务”Agent 前两步做得挺好到第三步突然开始自由发挥——要么多调了一个无关工具要么把中间结果覆盖掉最后输出一份完全没法用的东西。面向 Agent 的轻量 Workflow 设计就是为这类问题准备的一套实用解法。它不引入重型流程引擎也不把流程硬编码到业务代码里而是用一套轻量的状态机加节点编排机制把 Agent 的工具调用、思考决策、人工确认都收拢到清晰可控的执行框架内。这篇内容适合正在做 Agent 应用、被多步骤任务稳定性折磨的开发者参考后面也会写到我实际项目里踩过的坑和一些测量数据。1. 为什么 Agent 需要 Workflow一次失败的多步骤任务复盘1.1 复杂流程的失控点在哪先说一个我实际经历过的案例。当时在做一个客服工单处理 Agent需求是收到用户邮件后先识别问题类型再查询订单信息判断是否要退款最后回复用户并创建工单。听起来不算复杂。但上线第一周就频繁出问题。我复盘下来发现失控点不在单个步骤而在步骤之间的衔接。单步拆开看模型调用搜索工具、读取订单数据、判断用户意图准确率都在九成以上一旦串起来第二步的查询结果会在第三步被新的对话历史挤掉第四步的模型看到的是混入了旧数据的上下文于是开始脑补不存在的订单状态。这就是 Agent 应用和传统后端应用最大的区别传统流程里的数据是显式的参数传递都是代码写死的Agent 流程里每一步的输入输出靠自然语言和上下文隐式传递模型又有随机性一步错位后面全乱。轻量 Workflow 在这里的价值不是替代模型的能力而是给模型划定边界。每个步骤的输入从哪来、输出写到哪、下一步必须走哪个节点由 Workflow 控制。模型只在节点内部做自己擅长的事理解、生成、决策。1.2 三种流程组织方式的取舍硬编码、重引擎与轻量 Workflow我在做这套设计之前先对比过三种方式。第一种是代码硬编码。直接用 if-else 把流程写死比如“如果查询成功调用生成函数如果失败回到上一步”。这种方式最简单但坏处很明显流程逻辑和业务代码完全耦合改一个分支要改代码发版Agent 场景里分支数量多硬编码很快会变成一坨无法维护的条件嵌套。我试过维护一个十几个分支的硬编码流程每次加需求都提心吊胆。第二种是重型 BPM 或工作流引擎。这类产品功能非常完整可视化编排、版本管理、审批流、定时调度都有。但它们的定位是“跨系统、长周期、强管控”的业务流程一个流程从设计到跑起来需要大量配置建数据库表、部署服务、配置权限太重了。Agent 场景是“一次请求触发一个微型流程”生命周期几秒到几分钟节点数量几个到几十个用重引擎就像开卡车去菜市场买菜光是热车就要半天。第三种就是我最终选择的轻量 Workflow。核心只有三样东西节点定义、状态流转、上下文传递。用一个 JSON 文件描述流程用状态机控制走向节点之间通过共享上下文交换数据。接入成本按小时算不需要额外部署服务。三者的取舍可以概括成下面这张表维度代码硬编码重型流程引擎轻量 Workflow接入成本低高低流程调整灵活度差要改代码中配置繁琐好改配置即可动态分支支持差中好运行资源占用低高低状态可恢复性差好好与 Agent 场景匹配度勉强不匹配推荐1.3 轻量 Workflow 重点解决的三件事总结下来轻量 Workflow 要解决的核心问题有三个这三个也是我在实际项目里判断设计是否到位的标准。流程可观测。每一步执行了什么节点、停在哪个状态、输入输出是什么都要能查。Agent 是黑盒流程必须是透明的。可观测性解决不了问题但没有可观测性出了问题只能靠猜。状态可恢复。Agent 执行过程中一旦超时、崩溃、模型返回异常格式流程要能从断点恢复而不是整条流程从头再来。多步骤任务跑到第八步挂了让用户重新描述一遍需求体验极差。决策可配置。哪些步骤必须固定执行哪些步骤让 Agent 决策哪些步骤需要人工确认应该通过配置表达而不是靠代码堆。这样改流程不用发版产品和运营也能参与调整。2. 轻量 Workflow 的架构骨架节点、状态机与共享上下文2.1 节点抽象四个基本类型我习惯把节点分成四种基本类型覆盖掉绝大多数 Agent 场景。任务节点执行具体动作。比如调用搜索工具、查询数据库、调用模型生成一段文本。这类节点一般由代码实现输入从上下文取输出写回上下文。决策节点决定下一步走哪个分支。这里要区分两种情况能用规则判断的用规则比如“工具返回结果是否为空”“金额是否超过阈值”只有开放性问题才交给 Agent 决策比如“用户这句话属于投诉还是咨询”。人工节点暂停流程等人确认或补充信息。比如退款操作前要求人工确认生成报告后等人修改意见。并行节点同时执行多个任务节点。比如同时查询多个数据源、同时调用多个工具。并行能提速但会引入数据竞争问题后面专门讲。每个节点只做一件事输入输出尽量小而明确。这跟写函数的直觉一样函数粒度越细越容易复用节点粒度越细越容易编排和排查。2.2 状态流转用有限状态机框住 Agent 的每一步状态机是整个 Workflow 的地基。每条流程实例的状态可以定义成(node_id, status)的组合node_id 表示执行到哪个节点status 表示这个节点当前处于什么状态。我常用的状态枚举就六个pending、running、completed、failed、timeout、paused。pending 表示等待执行running 表示执行中completed 表示成功完成failed 表示业务失败timeout 表示超时paused 表示等待人工处理或外部事件。节点执行完后的流转逻辑由引擎统一处理规则很简单如果返回下一节点 ID就进入该节点的 pending如果返回 failed 或 timeout则进入重试逻辑或人工兜底如果流程到达 end则整条流程完成。这里有个关键设计Agent 的自由度被限定在节点内部。节点里你可以让模型自由发挥思维链、任意调用工具但节点之间的路径必须由流程定义控制。这个边界是整个系统稳定性的来源——模型负责战术Workflow 负责战略。2.3 共享上下文结构化的“工作台”而非聊天记录多轮 Agent 最常见的错误做法是把所有历史都塞给模型。对话一长token 超限、信息混杂、模型被无关上下文干扰。Workflow 场景需要的是结构化的共享上下文。我一般把上下文存储分成三个区。持久区存用户原始输入、业务对象 ID、最终结果这类关键数据临时区存中间计算结果、工具返回的原始数据节点跑完可以清理公开区是模型可以直接读取的数据比如当前任务说明、当前可见的工具结果。每个区都要设置容量上限。比如持久区最多保留 50 个字段临时区最多保留 20 条记录超过上限自动丢弃最旧数据或者触发一次摘要总结。这样模型每次看到的上下文是经过筛选的工作台而不是一条越来越长的聊天记录。上下文写入的时候我强烈建议带 source 标记记录是哪个节点写入的。排查问题时能直接回答“这个字段是谁改的、什么时候改的”这个信息在数据竞争场景下几乎是救命稻草。2.4 不选 DAG 编排引擎的理由有人会问为什么不直接用现成的 DAG 编排引擎比如 Airflow 或者 Prefect。我并不是说这些工具不好而是它们和 Agent 场景的匹配度不高。DAG 引擎的核心定位是“定时批量任务”。任务数量多、依赖关系在 DAG 里是静态的、调度周期是小时或者天级。Agent 场景恰好相反一次用户请求触发一个微型流程生命周期几秒到几分钟任务数量少、依赖关系可能根据 Agent 决策动态变化而且往往需要实时响应。用 DAG 引擎你要维护调度服务、队列、执行器加上一套 SDK 和运维配套。而轻量 Workflow 可以完全跑在应用进程内不依赖额外服务。所谓轻量不只是代码量少更重要的是运维成本低、依赖少、启动快。依赖也少意味着出问题时的排查面小。Agent 链路本身已经涉及模型调用、工具调用、外部 API环节够多了没必要再引入一套重量级调度基础设施。3. 四个关键机制设计超时重试、条件分支、人工审批与并发边界3.1 超时与重试给节点划红线模型调用和工具调用都可能卡住。模型服务超时、第三方 API 无响应、函数执行陷入死循环这些不是“可能发生”的问题而是“一定会发生”的问题。每个节点必须有超时上限。我实际项目里的参数给一个参考值普通 LLM 调用超时 30 到 60 秒工具调用超时 10 到 20 秒文件读取、数据库查询这类内部操作超时 5 秒左右。模型生成是 Agent 链路里最不可控的环节超时给足工具类外部接口也不可预测但一般比模型快。超时后进入重试。重试策略推荐指数退避加最大重试次数。假设基础间隔 1 秒第 n 次重试的等待时间是 1 乘以 2 的 n-1 次方秒也就是第一次失败等 1 秒、第二次等 2 秒、第三次等 4 秒三次重试总等待 7 秒。超过最大重试次数节点进入 failed 状态流程走兜底分支。重试次数不是越多越好。工具调用重试 2 到 3 次比较合理因为第三方接口服务抖动一般很快恢复模型调用重试 1 次就够因为同一个请求大概率得到差不多的结果重试太多只是在烧钱。超时和重试参数应该在节点配置里声明不要散落在代码里。3.2 条件分支与循环上限规则优先模型兜底分支是最容易做歪的地方。我在几个项目里反复折腾后的结论是能用规则判断的分支绝对不要交给模型。比如“查询结果是否为空”“金额是否超过 5000”“状态是否为已完成”这类分支用代码判断即可一条 if 语句的事快速、稳定、零成本。模型判断有概率性一次误判就能让整条流程走向错误分支而且复现成本高。只有“用户意图属于哪一类”“当前回复是否满足要求”这类开放性问题才适合交给 Agent 决策节点。即使交给模型我也建议给决策节点加后置校验规则。比如让模型输出一个 JSON 结构指定格式为{choice: A, confidence: 0.95}然后再用规则校验置信度阈值低于阈值就走默认分支。模型可以参与决策但最终要有一个规则兜底。循环同样要设置上限。Agent 场景最常见的循环是“自我修正循环”让模型重新生成方案、不满意再生成、再不满意还要生成。没有上限时模型可能无限自我否定token 消耗和延迟都会失控。我在所有循环节点上加最大轮数参数一般 2 到 3 轮超过轮数强制输出当前版本并记录一条“修正次数超限”的日志。3.3 人工审批节点人机协同的正确姿势不是所有流程都能全自动跑完。涉及资金操作、删除数据、对外发送敏感信息时我们需要一个“人进来确认一下”的环节。人工节点是 Agent 流程里最容易被实现得不专业的地方。我这里的实现思路分四步。第一步节点进入 paused 状态同时把待办信息推送到消息队列第二步外部系统比如一个审批前端消费队列展示待办第三步用户在页面点同意或拒绝调用接口回调第四步引擎拿到回调结果写入上下文继续流转到下一节点。有几个细节值得注意。一是超时处理人工节点最好不要设硬超时人没有及时处理是常态不是异常。超过一定时间可以发提醒但状态应该保持 paused而不是标记失败。二是幂等用户可能重复点击审批按钮回调要能识别同一个审批事件重复处理会把流程推两次。三是审批结果要写进上下文持久区后续节点如果需要对账可以直接读取审批记录和时间。3.4 并发执行的范围与数据隔离并行节点能显著提升效率尤其在数据采集阶段。比如同时调用两个搜索工具、同时读取多个订单文档可以省一半时间。但并行带来的数据竞争问题我在项目里踩过一次很深的坑。问题出在并行分支写回结果。多个分支同时执行完后每个分支都把结果写进上下文的同一个字段比如search_results。后完成的分支覆盖先完成的导致最终上下文里只剩最后一个分支的数据其他分支的结果全部丢失。而且由于时序不稳定每次丢的还不一样问题表现成偶发性的数据错误排查起来极其痛苦。解决办法是强制要求并行分支使用独立的 key 分区。比如分支 A 写results.a分支 B 写results.b互不干扰。合并后续节点时再统一从分区里读取。这个约束可以直接做进引擎并行子节点写上下文时自动用分支 ID 作为 key 前缀开发者无法绕过。并行并发度也要设上限。LLM 调用并行过多会造成资源紧张也容易触发限流。我一般把并行度控制在 3 到 5具体看模型服务的配额和业务容忍的延迟。4. 徒手实现轻量 Workflow 运行时配置协议、核心代码与断点恢复4.1 用 JSON 编写一份流程定义轻量 Workflow 最关键的一点是流程定义和业务代码分离。我把流程定义写成 JSON业务代码只负责实现节点。下面是一个报销汇总流程的配置示例。流程包括启动、查询报销数据、生成汇总、人工审批、发送财务、结束。{ workflow_id: expense_report_v1, nodes: { start: { type: start, next: collect_data }, collect_data: { type: tool, tool: query_expense, timeout_seconds: 15, max_retry: 3, next: gen_summary }, gen_summary: { type: agent, prompt: 请根据报销数据生成汇总报告, timeout_seconds: 60, max_retry: 1, next: human_approve }, human_approve: { type: human, notify_channel: approval_queue, remind_hours: 24, next: send_finance }, send_finance: { type: tool, tool: send_email, timeout_seconds: 20, max_retry: 2, next: end } } }每个节点配置里包含类型、执行参数、超时重试和流转目标。流转目标也可以是一个条件表达式比如next_if_empty: ask_user、next_if_not_empty: summary这样分支逻辑就留在配置层而不是埋在代码里。4.2 节点抽象与上下文接口实现定义配置协议之后需要一个简单的运行时。我用 Python 实现核心是两个类WorkflowContext 和 BaseNode。import time class WorkflowContext: def __init__(self, workflow_id: str): self.workflow_id workflow_id self.storage {} def read(self, key: str): return self.storage.get(key) def write(self, key: str, value, source_node: str): self.storage[key] { value: value, node: source_node, ts: time.time() }每次写入都保存来源节点和时间戳这在排查数据覆盖问题时非常有用。生产环境里这个 storage 可以直接换成 Redis 或者数据库天然支持跨进程共享和状态持久化。节点基类的接口只需要一个 execute 方法接收上下文返回下一节点 ID 或特殊状态。class BaseNode: def __init__(self, node_id: str, config: dict): self.node_id node_id self.config config async def execute(self, ctx: WorkflowContext) - str: raise NotImplementedError接收入口执行具体的业务逻辑返回下一节点 ID。这样设计的好处是节点的实现完全独立可以单测、可以复用。比如一个“查询报销单”的节点可以在多个不同流程里被引用只要配置里的 tool 参数不一样。4.3 事件驱动的主循环有了节点和上下文下一步是执行引擎的主循环。稍微核心的一段代码class WorkflowEngine: def __init__(self, definition: dict): self.definition definition self.current_node start self.status pending async def run(self, initial_input: dict): ctx WorkflowContext(self.definition[workflow_id]) ctx.write(input, initial_input, system) while True: node_def self.definition[nodes].get(self.current_node) if not node_def: raise ValueError(f未知节点: {self.current_node}) node create_node(self.current_node, node_def) self.status running try: next_node await self._execute_with_policy(node, ctx) except TimeoutError: next_node self._handle_timeout(node_def) except Exception as exc: next_node self._handle_failure(node_def, exc) if next_node in (end, failed): self.status next_node break self.current_node next_node return self.statusengine 只需要在一个循环里取当前节点、执行、拿下一节点、继续。执行策略里封装超时和重试失败则走兜底分支。整个实现不超过两百行完全没有额外依赖这就是轻量的意义。如果不满足于线性流程还可以让 next 字段支持条件表达式。比如配置成next: {if_empty: ask_user, default: summary}引擎执行完节点后根据节点返回值匹配表达式。这样分支能力不需要改引擎代码直接在配置层表达。4.4 状态持久化崩溃后从断点继续Agent 跑到第八步崩溃了用户要求继续最朴素的做法是整条重跑。但重跑意味着模型调用消费翻倍、用户体验恶化而且部分节点执行了两次。我更推荐做状态持久化和断点恢复。实现方式不复杂定义一个状态表包含 workflow_id、current_node、ctx_snapshot、status、updated_at。每次节点执行完成后把当前节点和上下文快照落一次库。这就是 checkpoint。恢复时读库把 ctx_snapshot 反序列化回 WorkflowContext然后从 current_node 继续执行。唯一要注意的是节点幂等性恢复后重放的节点如果之前已经执行成功应该跳过。比如发邮件节点如果断点前已经发了邮件恢复后再执行一次就会重复发送。解决办法是给节点执行结果也落库执行前先检查该节点是否已有 completed 标记。节点必须可重入这是断点恢复的底线。这套机制实现成本很低但在多步骤 Agent 场景里收益极大。用户看到的效果是“点击继续从崩溃处接着跑”整个产品稳定性感知提升一个档次。5. 上线落地踩坑数据竞争、循环失控与幂等恢复5.1 并行分支写同一个 key数据竞争的教训前面提过并行分支的数据竞争这里把排查过程详细说一下。当时某数据采集流程开了两个并行分支分别查询不同供应商的库存然后统一汇总。上线后经常出现汇总数据只有一个供应商的结果而且不同批次丢的还不一样。最开始怀疑是模型问题把日志翻出来看才发现两个分支都正常返回了但写回上下文时都写到了inventory_data这个 key。分支 A 写完后分支 B 又覆盖了一次。由于节点是并发执行谁后写完全由调度时序决定于是变成偶发bug。修复方式就是在引擎层强制分区并行分支的写 key。分支 A 自动映射到inventory_data.branch_a分支 B 映射到inventory_data.branch_b。后续汇总节点读两个分区再合并。这个约束应该配置在引擎层不要让业务开发自己记得用不同 key因为人一定会忘。5.2 模型自我修正的循环失控熔断参数怎么定还有一个常见问题是循环失控。某次做文案优化流程设计了“生成版本、自我评审、不满意就重新生成”的循环。上线后某条流程的 token 消耗异常高查日志发现模型连续重写了十几个版本每次都觉得不够好但每次修改幅度又很小。这是典型的缺少循环上限导致的失控。我们给循环节点加了 max_loops 参数设为 3超过轮数直接选当前版本输出同时写入超限日志。加完之后这类流程的 token 消耗降了 70%用户体验反而更好——因为改到第三版时模型通常会收敛再改下去边际收益很低。循环上限这个参数我的建议是宁可设小也不要设大。Agent 决策的质量不会因为多改几轮就显著提升反而累积的错误概率会上升。2 到 3 轮是实测收益和成本的平衡点。5.3 断点恢复的幂等陷阱节点必须可重入断点恢复还有一个隐藏陷阱。某个流程里有一个“通知用户进度”的节点恢复时重放该节点用户就收到了两条重复通知。排查后发现恢复流程没有检查节点的历史执行结果。解决办法是引入节点执行记录表。每个节点启动时先登记执行完成后更新状态为 completed。恢复执行前引擎检查当前节点是否已有 completed 记录有就跳过直接走下一节点。这是幂等恢复的关键。所有有外部副作用的节点发消息、发邮件、写数据库、调用第三方写接口都必须支持幂等。反面例子是只读节点比如查询和生成重放没有副作用可以不做幂等检查但做完检查统一处理更省心。5.4 常见问题速查表整理一个速查表这部分内容都来自我实际排查过的现场。问题现象排查思路解决方案上下文数据覆盖后续节点读到错误数据查看上下文写入日志的 node 字段并行分支 key 分区隔离重试风暴多个节点同时反复重试统计各节点超时时间与重试次数超时时间错开设置全局并发上限循环不退出token 消耗异常增长查看循环节点执行日志设置 max_loops超限强制输出恢复重复执行用户收到重复通知或邮件检查断点恢复日志节点执行结果落库检查幂等标记人工审批不生效流程一直停在等待状态检查消息队列消费情况和回调接口日志增加回调轮询与超时提醒机制模型决策置信度低分支选择经常不符合预期查看决策节点的置信度分布增加置信度阈值校验低于阈值走默认分支写到最后说点我自己的体会。面向 Agent 的轻量 Workflow 设计最难的地方不是写引擎而是克制——克制住把流程做得更复杂的冲动。我最初设计过一版支持几十种节点类型和可视化编排的引擎写了大半发现真正稳定跑在生产环境的反而是后来重构的这版节点类型就四种、状态转换就六种、上下文分三个区配置用 JSON 表达。Agent 的难度已经在模型那里了流程层要做的就是把不可控的边界画清楚。如果你也在做 Agent 应用我建议从一张纸开始把用户请求拆成 5 到 10 个步骤标出哪些固定执行、哪些让 Agent 决策、哪些需要人工确认然后再迁移到代码和配置。先跑通一个最小闭环比一开始就设计一个万能框架有用得多。我后来在可观测性上又补了很多东西把每个节点的输入输出都记录成结构化日志排错效率提升非常明显。到最后你会发现轻量 Workflow 不是一个框架而是一种设计习惯。