最近在做某平台侧的能力落地时我接了一个需求让前端通过自然语言提交“帮我查一下某商品的库存不够就提醒采购有货就把报价单生成出来发我”。业务方说得很简单恨不得一个对话框全解决。但真开始做我就意识到直接把所有逻辑都交给一个 Agent 大模型自由发挥上线之后一定会出问题。输入稍微偏一偏模型就可能在某个环节答非所问或者做了一个完全不在预期内的操作。后来我把思路改成“Agent 负责判断流程负责兜底”也就是这篇要分享的面向 Agent 的轻量 Workflow 设计。这套方案的核心就两句话让模型在有限的、确定的路由集合里做选择而不是让它无限制地自由行动。它解决了三个实际问题Agent 不可控、多轮调用上下文混乱、以及业务人员需要对“系统最终会走哪条路”有确定性预期。适合正在做 AI Agent 应用开发、想把模型调用稳定地嵌入业务系统的一线开发者和架构师参考。我这边最终没有引入任何重量级的流程引擎而是自己在项目里用状态机的方式写了一个只有几百行代码的轻量编排内核。这篇文章会把设计思路、核心数据结构、具体落地的三个迭代版本以及我实测踩过的坑完整写出来。代码以 Python 为主思路不挑语言。1. 为什么要用 Workflow 圈住 Agent1.1 Agent 的规划能力很强但它不可控大语言模型原生具备一定的任务规划能力你可以让它“自己决定该调用哪些工具、按什么顺序调用”。demo 阶段跑起来很惊艳但一旦进入真实业务问题就暴露了。我见过太多类似的情况模型把“查库存”和“下采购单”两个动作的先后关系搞错了直接跳过了确认环节。面对两个选项时模型想到了第三个“创造性的”动作而系统没有执行该动作的权限流程卡住。同样一句话今天走的是 A 分支明天模型参数或 prompt 稍变它走了 B 分支业务方完全不知道发生了什么。这些问题的根子不在于模型不够聪明而在于系统把决策权全部交给了模型却没有给它一个可以行动的空间边界。规划的宽度太大出错的概率就呈指数级上升。1.2 轻量 Workflow限制行动空间保留判断能力我在设计里做的第一件事就是给 Agent 画一个“活动范围”。整个业务过程被拆成一个一个有明确语义的节点每个节点只做一件事。节点之间怎么跳转可以由代码硬编码也可以由模型决策。模型依然在关键岔路口做判断但它只能在我给定的候选分支里选不能发明新的路径。这里有个类比很合适Workflow 是铁轨Agent 是火车司机。司机可以决定前方扳道岔时走哪条路但他不能脱离轨道开到马路上去。铁路系统虽然灵活性低但安全性、可预期性和运营成本远远优于“可以任意行驶的自动驾驶汽车”。这也是“轻量”这个词的核心含义。轻的不是功能而是约束方式。它不需要像传统 BPM 引擎那样把每个业务对象、每个表单字段都建模成数据实体它只需要帮助你把“模型的下一步选择”变成流程转移事件。2. 核心设计三种节点、一份上下文、一条状态边2.1 以状态机作为骨架而不是通用流程图一开始我考虑过直接用有向无环图但在设计过程中放弃了。Agent 场景下经常会遇到“失败重试”“人工复核后回到上一步修改参数”这样的环路需求DAG 表达起来很别扭。状态机天然支持环状态、事件、转移。当前节点执行完成之后产生一个事件事件驱动系统决定走向下一个状态这个模型非常吻合 Agent 场景。完整状态机实现相对复杂但只是在流程层面自己定义“节点 条件边 转移”编写量其实很小。2.2 三种基础节点工具、决策、审批我在这套方案里定义了三类节点覆盖绝大多数 Agent 应用的落地场景。工具节点Tool Node执行确定性逻辑比如查数据库、调内部 API、发送消息。工具节点不调用大模型结果稳定可靠是流程真正的“受力点”。决策节点LLM Node调用大模型输入当前上下文和若干候选选项模型输出其中一个选项 ID流程根据该 ID 沿着对应分支走下去。这里有一个重要原则模型输出的只能是候选 ID不直接携带业务数据。业务数据必须由后续的工具节点重新查询获取避免模型“编造数据”。人工审批节点Approval Node流程暂停等待用户点同意或拒绝。超时或者拒绝时可以选择退回指定节点或者终止。人工节点是安全兜底的关键它保证了即使模型决策错误系统也不会一条路走到黑。2.3 轻量上下文不要复制传统工作流的复杂变量域传统工作流引擎里有全局变量、局部变量、表达式解析器我们不需要那一套。我的做法极简整个运行实例持有一个ctx就是一个兼容字典的对象存放所有节点运行后产生的字段。每个节点有独立的输出槽位节点运行完把自己的结果写入ctx[节点的名称]下。进入 LLM 决策节点时不是把整个ctx丢给模型而是从一个白名单里挑选字段拼进 prompt。控制 token 量的同时也减少模型被无关信息干扰的概率。这个设定最大的好处是心智负担低。新同事接手维护时打开执行日志就能看懂整个上下文演变过程不需要理解一套专门的变量绑定语法。3. 实操记录三步实现轻量编排内核3.1 第一版固定流程先跑通路径我没有一上来就接入大模型。第一步先实现一个最基础的框架配置驱动、节点自注册、条件转移。用 Pydantic 定义了配置模型加载后交给一个极简执行器。from typing import Any, Callable, Literal from pydantic import BaseModel class ToolNodeConfig(BaseModel): type: Literal[tool] id: str tool_name: str class DecisionNodeConfig(BaseModel): type: Literal[decision] id: str model: str candidates: list[str] prompt_template: str class Edge(BaseModel): source: str target: str # 空字符串表示无条件转移 condition_field: str condition_value: str class FlowConfig(BaseModel): start: str nodes: list[ToolNodeConfig | DecisionNodeConfig] edges: list[Edge] default_target: str 执行器的主循环是这段程序最核心的地方class GraphExecutor: def __init__(self, config: FlowConfig, tools: dict[str, Callable[..., Any]]): self.config config self.tools tools self.max_steps 50 # 防死循环 def run(self, ctx: dict[str, Any]) - dict[str, Any]: node_id self.config.start current_ctx ctx steps 0 while node_id in self.node_map: steps 1 if steps self.max_steps: raise RuntimeError(workflow exceeded max steps) node self.node_map[node_id] if node.type tool: result self.run_tool(node, current_ctx) elif node.type decision: result self.run_decision(node, current_ctx) else: raise ValueError(funknown node type: {node.type}) next_node_id self.transit(node_id, result, current_ctx) current_ctx[node.id] result node_id next_node_id if node_id end: break return current_ctx这一版完全不需要大模型所有节点都是工具节点。它证明了一件事任何复杂流程都可以被配置化。拿到一个新的业务需求时我会先不考虑模型把每一步该调用什么接口、判断条件是什么先全部用代码写死跑通一遍。3.2 第二版接入 LLM 决策节点流程骨架稳定之后我加入了 LLM 决策节点。它的输入是一个问题描述和候选项输出是一个 JSON里面包含选中的选项 ID 和理由。关键代码如下def route_by_llm( ctx: dict[str, Any], node: DecisionNodeConfig, llm_client, ) - str: # 从上下文中提取白名单字段避免把无关内容全部塞给模型 prompt_data {k: ctx.get(k) for k in node.ctx_fields if k in ctx} user_prompt node.prompt_template.format(**prompt_data) response llm_client.chat_completion( system你是流程路由助手只能从用户给定的候选列表中选择一个ID。, useruser_prompt, response_format{type: json_object}, temperature0, ) text response[content] # 第二版先不做复杂解析直接按 JSON 处理 choice json.loads(text)[choice] # 安全校验必须属于候选集合否则走默认分支 if choice not in node.candidates: return node.default_target return choice这里有一个很关键的设计模型只返回选项 ID不返回任何业务字段。例如模型判断“该工单属于售后”它只负责返回选项 ID “after_sale”之后流程会进入名为“查询售后知识库”的工具节点由代码而非模型来生成答案。这保证最后业务动作的执行是确定的、可以被审计的。我在实际落地时把 prompt 也设计得很死板甚至可以说带一点“命令式”的风格当前上下文 库存数量: {stock_count} 订单状态: {order_status} 这是你唯一的选择依据。从以下候选中选择最合适的一个只输出 JSON [{id: to_restock, desc: 需要补货}, {id: to_deliver, desc: 可以发货}, {id: to_review, desc: 需要人工复核}] JSON 格式{choice: to_restock, reason: 简要理由}唯一可变的部分是ctx_fields白名单。这个节点天然支持不同业务场景复用换一个流程就换一组字段和候选。3.3 第三版工具异常、重试与人工兜底跑到第三版业务方开始要求更复杂的能力失败了要重试、重试还不行要让人介入。我加了两个节点配置字段max_retries和fallback_node。工具节点执行异常时执行器捕获异常并记录错误信息到ctx重试次数加一。超过上限就跳转到兜底节点。def run_tool_with_retry(self, node: ToolNodeConfig, ctx: dict[str, Any]): attempt 0 error None while attempt node.max_retries: try: result self.tools[node.tool_name](ctx) return {ok: True, data: result} except Exception as e: error str(e) attempt 1 ctx.setdefault(errors, []).append( {node: node.id, attempt: attempt, error: error} ) # 记录失败的节点信息走兜底或者转人工 ctx[last_error] error if node.fallback_node: return {ok: False, fallback: node.fallback_node} raise RuntimeError(error)这种做法看起来非常直白但它很好地兜住了真实世界的脏乱差第三方接口超时、库存系统 ACK 失败、数据库抖动……过去这些都要靠写 JSON 嵌套的异常处理代码现在全被 Workflow 的转移逻辑吸收了。3.4 第四版人工审批与外部状态恢复人工审批本来不在我的最小设计里后来客户提了一个很现实的需求某些操作不能自动执行必须等财务确认。我把审批节点做成“外部回调驱动”的模式class ApprovalNodeConfig(BaseModel): type: Literal[approval] id: str approve_target: str reject_target: str timeout_seconds: int 86400执行到审批节点时执行器把当前实例 ID 写入持久化存储我用的是数据库然后返回一个挂起状态。外部系统通过单独的服务接口提交“同意”或“拒绝”事件事件携带实例 ID流程从断点继续执行。这里有一个用状态机天然就带来的好处挂起中的实例不会占用进程资源。再次唤醒只需要执行“转移”逻辑即可。相比传统“每步一个回调函数”的实现方式我更喜欢这种集中式的状态控制它让异常处理变得特别简单——任何节点异常都可以落到一个统一的error_target上通过人工事件再拉起。4. 实战案例用轻量 Workflow 编排一个库存处理助手我用一个真实场景串一遍流程用户在前端输入一句话“某供应商的物料库存不足了请帮我确认并发送提醒”系统需要自动完成工单分诊、库存查询、生成处理建议、人工确认、发送消息五个环节。流程定义如下表节点 ID类型用途出边条件start工具初始化上下文无route决策判断是否需要库存处理候选need_check/no_actioncheck_stock工具调用物料系统查库存无suggest决策基于库存数据生成建议候选auto_send/manual_reviewhuman_review审批人工确认是否发送同意 / 拒绝send_msg工具发送消息通知采购无end工具结束无路由节点接入大模型而check_stock和send_msg走代码逻辑。其中suggest节点比较有意思它会根据库存量和上次采购时间来决定是自动发送还是要人工复核suggest_prompt 历史采购数据 最近采购日期{last_purchase_date} 当前库存{stock_count} 当库存小于安全库存且最近采购日期超过30天时建议 manual_review 否则建议 auto_send。直接输出JSON{choice: auto_send | manual_review, reason: ... }实际执行时route节点先返回need_check流程进入check_stock工具节点ctx中新增了stock_count120等字段。随后suggest节点发现last_purchase_date距今已有 45 天于是返回manual_review。流程停在审批节点推送一条消息给业务人员业务人员点确认后才执行send_msg。整个过程里大模型只参与了两处判断其余环节全部可控耗时也完全可预期。这比让 Agent 一步到位“理解需求、自己取数、自己发消息”稳妥太多。5. 常见问题与排查技巧实录5.1 模型输出不稳定JSON 解析失败解决思路层层递进。首先是降温度temperature调成 0绝大多数模型的稳定性会明显改善。其次是在 prompt 里规定只输出 JSON不输出任何解释。第三是在代码里做容错——如果严格 JSON 解析失败尝试用正则提取第一个{...}块再解析。最后如果解析仍然失败就放弃本次决策走default_target。这个兜底策略看起来很简单但它是保证生产环境不挂的关键。我见过一些团队在json.loads处直接抛异常结果线上流程大面积中断。宁可多走一个后续节点也不要让整个流程因为一个解析错误卡死。5.2 上下文越来越大prompt 严重超长一个流程跑 20 多个节点时如果每次决策节点把全量ctx传给模型token 开销会非常惊人。解决办法不复杂每个决策节点配置ctx_fields白名单只把你真正需要模型看到的关键字段放进去。对于历史步骤用摘要替代原始数据。比如查询数据库的结果只要保留“命中行数”和“关键状态”不要保留完整明细。5.3 流程死循环我在执行器里硬编码了一个max_steps默认 50。一旦超过限制直接抛异常转人工。这是 Workflow 引擎设计的通用防御姿势任何“自动规模化”的系统都需要这一步。另外设计转移条件时建议避免让一条边同时满足两个分支的判断。这个需要在测试阶段用表格穷举所有上下文组合来验证不要指望靠代码 review 找出来。5.4 流程持久化恢复架构升级后我发现把全部上下文字段存在内存里对调试很不友好而且服务一重启所有挂起中的审批流程就全丢了。所以我把ctx序列化成 JSON 存进数据库表里每次节点运行结束更新一行记录。节点设计上没有保存任何内部状态所以恢复动作就是“读取记录—反序列化—继续执行下一步”。5.5 幂等性问题当工具节点实际已经执行成功但网络断掉Workflow 收到的是“请求失败”重试时工具可能会被执行两次。比如“发送消息”如果重试就会给用户发两条一模一样的通知。解决方式是在工具节点支持幂等键同一个ctx实例 ID 和节点 ID 组合成的业务键如果工具端检测到键已存在直接返回上一次的结果不再重复执行。加了这个机制之后重试才真正变得安全。6. 后续可以扩展的方向有一个我个人非常看好的方向就是把这套轻量 Workflow 设计成“可被 Agent 自己修改的配置”。既然 Workflow 本质是一份 JSON那完全可以写一个“流程生成 Agent”让它在收到业务需求时先输出一份流程配置草稿再由开发人员 review 之后上线运行。这样既能结合大模型理解自然语言的能力又能保证最终上线的流程是经过人工确认的。另外一个方向是引入更丰富的转移条件表达式比如支持ctx[value] 100这类数值判断。我目前用字符串相等条件就够用但未来接入复杂业务之后必然需要一套小的表达式引擎。到时候可以考虑接入一个轻量级的表达式库而不是自己写解析器。再就是多实例并发控制。同一用户同一业务并发提交时需要基于用户 ID 做锁避免两个实例同时对同一个物料做“补货”和“停用”这种冲突操作。这个在共享型业务场景下尤其重要。最后简单说说我这几个月做过这个设计之后的体会。一个常见的认知偏差是觉得引入 Workflow 会增加开发量、拖慢 Agent 响应速度。但实际情况恰恰相反Workflow 把每个动作收敛成确定性的函数调用之后调试成本大幅下降业务方对你的系统也会有实实在在的信任感。以后接到“做一个自动 XX 的 Agent”这种需求时我都会先花半天时间把客户的真实流程画成状态机再考虑让模型上场。流程画得越清楚写代码就越省心模型也越不容易“自由发挥”。 
长河君 · 2026/10/17 7:59:23 · 阅读 1.2k Agent 不可控？用轻量 Workflow 状态机让大模型决策更可控
最近在做某平台侧的能力落地时我接了一个需求让前端通过自然语言提交“帮我查一下某商品的库存不够就提醒采购有货就把报价单生成出来发我”。业务方说得很简单恨不得一个对话框全解决。但真开始做我就意识到直接把所有逻辑都交给一个 Agent 大模型自由发挥上线之后一定会出问题。输入稍微偏一偏模型就可能在某个环节答非所问或者做了一个完全不在预期内的操作。后来我把思路改成“Agent 负责判断流程负责兜底”也就是这篇要分享的面向 Agent 的轻量 Workflow 设计。这套方案的核心就两句话让模型在有限的、确定的路由集合里做选择而不是让它无限制地自由行动。它解决了三个实际问题Agent 不可控、多轮调用上下文混乱、以及业务人员需要对“系统最终会走哪条路”有确定性预期。适合正在做 AI Agent 应用开发、想把模型调用稳定地嵌入业务系统的一线开发者和架构师参考。我这边最终没有引入任何重量级的流程引擎而是自己在项目里用状态机的方式写了一个只有几百行代码的轻量编排内核。这篇文章会把设计思路、核心数据结构、具体落地的三个迭代版本以及我实测踩过的坑完整写出来。代码以 Python 为主思路不挑语言。1. 为什么要用 Workflow 圈住 Agent1.1 Agent 的规划能力很强但它不可控大语言模型原生具备一定的任务规划能力你可以让它“自己决定该调用哪些工具、按什么顺序调用”。demo 阶段跑起来很惊艳但一旦进入真实业务问题就暴露了。我见过太多类似的情况模型把“查库存”和“下采购单”两个动作的先后关系搞错了直接跳过了确认环节。面对两个选项时模型想到了第三个“创造性的”动作而系统没有执行该动作的权限流程卡住。同样一句话今天走的是 A 分支明天模型参数或 prompt 稍变它走了 B 分支业务方完全不知道发生了什么。这些问题的根子不在于模型不够聪明而在于系统把决策权全部交给了模型却没有给它一个可以行动的空间边界。规划的宽度太大出错的概率就呈指数级上升。1.2 轻量 Workflow限制行动空间保留判断能力我在设计里做的第一件事就是给 Agent 画一个“活动范围”。整个业务过程被拆成一个一个有明确语义的节点每个节点只做一件事。节点之间怎么跳转可以由代码硬编码也可以由模型决策。模型依然在关键岔路口做判断但它只能在我给定的候选分支里选不能发明新的路径。这里有个类比很合适Workflow 是铁轨Agent 是火车司机。司机可以决定前方扳道岔时走哪条路但他不能脱离轨道开到马路上去。铁路系统虽然灵活性低但安全性、可预期性和运营成本远远优于“可以任意行驶的自动驾驶汽车”。这也是“轻量”这个词的核心含义。轻的不是功能而是约束方式。它不需要像传统 BPM 引擎那样把每个业务对象、每个表单字段都建模成数据实体它只需要帮助你把“模型的下一步选择”变成流程转移事件。2. 核心设计三种节点、一份上下文、一条状态边2.1 以状态机作为骨架而不是通用流程图一开始我考虑过直接用有向无环图但在设计过程中放弃了。Agent 场景下经常会遇到“失败重试”“人工复核后回到上一步修改参数”这样的环路需求DAG 表达起来很别扭。状态机天然支持环状态、事件、转移。当前节点执行完成之后产生一个事件事件驱动系统决定走向下一个状态这个模型非常吻合 Agent 场景。完整状态机实现相对复杂但只是在流程层面自己定义“节点 条件边 转移”编写量其实很小。2.2 三种基础节点工具、决策、审批我在这套方案里定义了三类节点覆盖绝大多数 Agent 应用的落地场景。工具节点Tool Node执行确定性逻辑比如查数据库、调内部 API、发送消息。工具节点不调用大模型结果稳定可靠是流程真正的“受力点”。决策节点LLM Node调用大模型输入当前上下文和若干候选选项模型输出其中一个选项 ID流程根据该 ID 沿着对应分支走下去。这里有一个重要原则模型输出的只能是候选 ID不直接携带业务数据。业务数据必须由后续的工具节点重新查询获取避免模型“编造数据”。人工审批节点Approval Node流程暂停等待用户点同意或拒绝。超时或者拒绝时可以选择退回指定节点或者终止。人工节点是安全兜底的关键它保证了即使模型决策错误系统也不会一条路走到黑。2.3 轻量上下文不要复制传统工作流的复杂变量域传统工作流引擎里有全局变量、局部变量、表达式解析器我们不需要那一套。我的做法极简整个运行实例持有一个ctx就是一个兼容字典的对象存放所有节点运行后产生的字段。每个节点有独立的输出槽位节点运行完把自己的结果写入ctx[节点的名称]下。进入 LLM 决策节点时不是把整个ctx丢给模型而是从一个白名单里挑选字段拼进 prompt。控制 token 量的同时也减少模型被无关信息干扰的概率。这个设定最大的好处是心智负担低。新同事接手维护时打开执行日志就能看懂整个上下文演变过程不需要理解一套专门的变量绑定语法。3. 实操记录三步实现轻量编排内核3.1 第一版固定流程先跑通路径我没有一上来就接入大模型。第一步先实现一个最基础的框架配置驱动、节点自注册、条件转移。用 Pydantic 定义了配置模型加载后交给一个极简执行器。from typing import Any, Callable, Literal from pydantic import BaseModel class ToolNodeConfig(BaseModel): type: Literal[tool] id: str tool_name: str class DecisionNodeConfig(BaseModel): type: Literal[decision] id: str model: str candidates: list[str] prompt_template: str class Edge(BaseModel): source: str target: str # 空字符串表示无条件转移 condition_field: str condition_value: str class FlowConfig(BaseModel): start: str nodes: list[ToolNodeConfig | DecisionNodeConfig] edges: list[Edge] default_target: str 执行器的主循环是这段程序最核心的地方class GraphExecutor: def __init__(self, config: FlowConfig, tools: dict[str, Callable[..., Any]]): self.config config self.tools tools self.max_steps 50 # 防死循环 def run(self, ctx: dict[str, Any]) - dict[str, Any]: node_id self.config.start current_ctx ctx steps 0 while node_id in self.node_map: steps 1 if steps self.max_steps: raise RuntimeError(workflow exceeded max steps) node self.node_map[node_id] if node.type tool: result self.run_tool(node, current_ctx) elif node.type decision: result self.run_decision(node, current_ctx) else: raise ValueError(funknown node type: {node.type}) next_node_id self.transit(node_id, result, current_ctx) current_ctx[node.id] result node_id next_node_id if node_id end: break return current_ctx这一版完全不需要大模型所有节点都是工具节点。它证明了一件事任何复杂流程都可以被配置化。拿到一个新的业务需求时我会先不考虑模型把每一步该调用什么接口、判断条件是什么先全部用代码写死跑通一遍。3.2 第二版接入 LLM 决策节点流程骨架稳定之后我加入了 LLM 决策节点。它的输入是一个问题描述和候选项输出是一个 JSON里面包含选中的选项 ID 和理由。关键代码如下def route_by_llm( ctx: dict[str, Any], node: DecisionNodeConfig, llm_client, ) - str: # 从上下文中提取白名单字段避免把无关内容全部塞给模型 prompt_data {k: ctx.get(k) for k in node.ctx_fields if k in ctx} user_prompt node.prompt_template.format(**prompt_data) response llm_client.chat_completion( system你是流程路由助手只能从用户给定的候选列表中选择一个ID。, useruser_prompt, response_format{type: json_object}, temperature0, ) text response[content] # 第二版先不做复杂解析直接按 JSON 处理 choice json.loads(text)[choice] # 安全校验必须属于候选集合否则走默认分支 if choice not in node.candidates: return node.default_target return choice这里有一个很关键的设计模型只返回选项 ID不返回任何业务字段。例如模型判断“该工单属于售后”它只负责返回选项 ID “after_sale”之后流程会进入名为“查询售后知识库”的工具节点由代码而非模型来生成答案。这保证最后业务动作的执行是确定的、可以被审计的。我在实际落地时把 prompt 也设计得很死板甚至可以说带一点“命令式”的风格当前上下文 库存数量: {stock_count} 订单状态: {order_status} 这是你唯一的选择依据。从以下候选中选择最合适的一个只输出 JSON [{id: to_restock, desc: 需要补货}, {id: to_deliver, desc: 可以发货}, {id: to_review, desc: 需要人工复核}] JSON 格式{choice: to_restock, reason: 简要理由}唯一可变的部分是ctx_fields白名单。这个节点天然支持不同业务场景复用换一个流程就换一组字段和候选。3.3 第三版工具异常、重试与人工兜底跑到第三版业务方开始要求更复杂的能力失败了要重试、重试还不行要让人介入。我加了两个节点配置字段max_retries和fallback_node。工具节点执行异常时执行器捕获异常并记录错误信息到ctx重试次数加一。超过上限就跳转到兜底节点。def run_tool_with_retry(self, node: ToolNodeConfig, ctx: dict[str, Any]): attempt 0 error None while attempt node.max_retries: try: result self.tools[node.tool_name](ctx) return {ok: True, data: result} except Exception as e: error str(e) attempt 1 ctx.setdefault(errors, []).append( {node: node.id, attempt: attempt, error: error} ) # 记录失败的节点信息走兜底或者转人工 ctx[last_error] error if node.fallback_node: return {ok: False, fallback: node.fallback_node} raise RuntimeError(error)这种做法看起来非常直白但它很好地兜住了真实世界的脏乱差第三方接口超时、库存系统 ACK 失败、数据库抖动……过去这些都要靠写 JSON 嵌套的异常处理代码现在全被 Workflow 的转移逻辑吸收了。3.4 第四版人工审批与外部状态恢复人工审批本来不在我的最小设计里后来客户提了一个很现实的需求某些操作不能自动执行必须等财务确认。我把审批节点做成“外部回调驱动”的模式class ApprovalNodeConfig(BaseModel): type: Literal[approval] id: str approve_target: str reject_target: str timeout_seconds: int 86400执行到审批节点时执行器把当前实例 ID 写入持久化存储我用的是数据库然后返回一个挂起状态。外部系统通过单独的服务接口提交“同意”或“拒绝”事件事件携带实例 ID流程从断点继续执行。这里有一个用状态机天然就带来的好处挂起中的实例不会占用进程资源。再次唤醒只需要执行“转移”逻辑即可。相比传统“每步一个回调函数”的实现方式我更喜欢这种集中式的状态控制它让异常处理变得特别简单——任何节点异常都可以落到一个统一的error_target上通过人工事件再拉起。4. 实战案例用轻量 Workflow 编排一个库存处理助手我用一个真实场景串一遍流程用户在前端输入一句话“某供应商的物料库存不足了请帮我确认并发送提醒”系统需要自动完成工单分诊、库存查询、生成处理建议、人工确认、发送消息五个环节。流程定义如下表节点 ID类型用途出边条件start工具初始化上下文无route决策判断是否需要库存处理候选need_check/no_actioncheck_stock工具调用物料系统查库存无suggest决策基于库存数据生成建议候选auto_send/manual_reviewhuman_review审批人工确认是否发送同意 / 拒绝send_msg工具发送消息通知采购无end工具结束无路由节点接入大模型而check_stock和send_msg走代码逻辑。其中suggest节点比较有意思它会根据库存量和上次采购时间来决定是自动发送还是要人工复核suggest_prompt 历史采购数据 最近采购日期{last_purchase_date} 当前库存{stock_count} 当库存小于安全库存且最近采购日期超过30天时建议 manual_review 否则建议 auto_send。直接输出JSON{choice: auto_send | manual_review, reason: ... }实际执行时route节点先返回need_check流程进入check_stock工具节点ctx中新增了stock_count120等字段。随后suggest节点发现last_purchase_date距今已有 45 天于是返回manual_review。流程停在审批节点推送一条消息给业务人员业务人员点确认后才执行send_msg。整个过程里大模型只参与了两处判断其余环节全部可控耗时也完全可预期。这比让 Agent 一步到位“理解需求、自己取数、自己发消息”稳妥太多。5. 常见问题与排查技巧实录5.1 模型输出不稳定JSON 解析失败解决思路层层递进。首先是降温度temperature调成 0绝大多数模型的稳定性会明显改善。其次是在 prompt 里规定只输出 JSON不输出任何解释。第三是在代码里做容错——如果严格 JSON 解析失败尝试用正则提取第一个{...}块再解析。最后如果解析仍然失败就放弃本次决策走default_target。这个兜底策略看起来很简单但它是保证生产环境不挂的关键。我见过一些团队在json.loads处直接抛异常结果线上流程大面积中断。宁可多走一个后续节点也不要让整个流程因为一个解析错误卡死。5.2 上下文越来越大prompt 严重超长一个流程跑 20 多个节点时如果每次决策节点把全量ctx传给模型token 开销会非常惊人。解决办法不复杂每个决策节点配置ctx_fields白名单只把你真正需要模型看到的关键字段放进去。对于历史步骤用摘要替代原始数据。比如查询数据库的结果只要保留“命中行数”和“关键状态”不要保留完整明细。5.3 流程死循环我在执行器里硬编码了一个max_steps默认 50。一旦超过限制直接抛异常转人工。这是 Workflow 引擎设计的通用防御姿势任何“自动规模化”的系统都需要这一步。另外设计转移条件时建议避免让一条边同时满足两个分支的判断。这个需要在测试阶段用表格穷举所有上下文组合来验证不要指望靠代码 review 找出来。5.4 流程持久化恢复架构升级后我发现把全部上下文字段存在内存里对调试很不友好而且服务一重启所有挂起中的审批流程就全丢了。所以我把ctx序列化成 JSON 存进数据库表里每次节点运行结束更新一行记录。节点设计上没有保存任何内部状态所以恢复动作就是“读取记录—反序列化—继续执行下一步”。5.5 幂等性问题当工具节点实际已经执行成功但网络断掉Workflow 收到的是“请求失败”重试时工具可能会被执行两次。比如“发送消息”如果重试就会给用户发两条一模一样的通知。解决方式是在工具节点支持幂等键同一个ctx实例 ID 和节点 ID 组合成的业务键如果工具端检测到键已存在直接返回上一次的结果不再重复执行。加了这个机制之后重试才真正变得安全。6. 后续可以扩展的方向有一个我个人非常看好的方向就是把这套轻量 Workflow 设计成“可被 Agent 自己修改的配置”。既然 Workflow 本质是一份 JSON那完全可以写一个“流程生成 Agent”让它在收到业务需求时先输出一份流程配置草稿再由开发人员 review 之后上线运行。这样既能结合大模型理解自然语言的能力又能保证最终上线的流程是经过人工确认的。另外一个方向是引入更丰富的转移条件表达式比如支持ctx[value] 100这类数值判断。我目前用字符串相等条件就够用但未来接入复杂业务之后必然需要一套小的表达式引擎。到时候可以考虑接入一个轻量级的表达式库而不是自己写解析器。再就是多实例并发控制。同一用户同一业务并发提交时需要基于用户 ID 做锁避免两个实例同时对同一个物料做“补货”和“停用”这种冲突操作。这个在共享型业务场景下尤其重要。最后简单说说我这几个月做过这个设计之后的体会。一个常见的认知偏差是觉得引入 Workflow 会增加开发量、拖慢 Agent 响应速度。但实际情况恰恰相反Workflow 把每个动作收敛成确定性的函数调用之后调试成本大幅下降业务方对你的系统也会有实实在在的信任感。以后接到“做一个自动 XX 的 Agent”这种需求时我都会先花半天时间把客户的真实流程画成状态机再考虑让模型上场。流程画得越清楚写代码就越省心模型也越不容易“自由发挥”。