过去大半年我一直在做一件不算性感但特别磨人的事把一套靠人肉连线、层层 if/else 拼出来的自动化流水线重构成真正由多个 Agent 协作驱动的系统。单跑任何一个 Agent 都很顺畅但只要把三个以上的 Agent 串在一起工作流编排就成了比 prompt 调优更让人崩溃的问题。市面上的工作流引擎不是不好而是普遍偏向组织级流程管理配置项多、学习曲线陡、跑起来还得养一套运维。所以后来我干脆自己写了一层极简的编排层这就是我今天想聊的面向 Agent 的轻量 Workflow 设计。如果你手头已经有一堆能跑的 Agent 脚本不想引一个重型框架只想要一层清晰的流程骨架那这篇文章就是写给你读的。1. 为什么 Agent 一多任务编排就成了逃不掉的坎1.1 单 Agent 的假洒脱与多 Agent 的真混乱我很长时间里只有一个 Agent。给它一个任务描述、几份参考资料、一套函数接口它自己决定怎么拆、怎么调。单个 Agent 的场景里根本不需要什么 workflow因为模型自带的思维链本身就充当了临时编排者用户说帮我写一篇文章Agent 内部就自动规划成查资料 → 列大纲 → 逐段写 → 校对。整个流程是隐式的没有外部可见的状态机。问题不在这问题在于所有任务都混在一个 Agent 里的时候prompt 会越来越长工具函数会越来越多最后模型开始频繁地选错工具忘记前面的结论。于是我把职责拆成了四个小 Agent一个负责意图判断一个负责资料检索一个负责信息抽取和结构化一个负责生成最终报告。每个 Agent 单独测效果立竿见影。但拆完当天下午我就意识到麻烦来了谁来告诉它们执行的顺序谁决定什么时候需要回到上一步如果第一个 Agent 判断错了意图后面三个全白跑如果资料检索返回的内容不够是不是应该回退到意图判断重新拆解这些都是业务流程问题不是一个模型能自己拍板解决的。模型只能基于上下文给出下一步建议但流程的骨架得有人显式画出来。这正是轻量 Workflow 要解决的起点把Agent 内部怎么思考和Agent 之间怎么衔接两件事分开。前者交给模型后者交给流程。1.2 从隐式 if-else 到显式路径图编排需求的必然性拆完 Agent 之后我最开始的接法非常粗暴主进程里按顺序调用遇到判断就用 if-else。如果意图是改写走改写路径如果抽取结果为空重新调一次抽取如果报告生成失败退回改写路径。两三个分支时还能接受等分支逐步多到十几条这个主进程就变成了一团没人敢改的意大利面条。每次加需求我都要把所有逻辑从头读一遍才敢在中间插一个条件判断。真正让我意识到必须换思路的是一次线上故障某个输入触发了三个分支叠加结果同一个报告被两个 Agent 各生成了一遍。由于没有统一的流程视图我花了大半天才从日志里把执行路径拼出来。那段时间我突然想明白一件事我的这些 if-else 其实早就构成了一张隐性的有向依赖图只是它散落在每个函数、每段状态判断里既不可见也不可测。既然如此不如把它显式建模。这就是轻量 Workflow 的本质——不创造新的复杂度只是把散落在代码里的流程逻辑收拢成一张可读、可查、可限制步数的图。否则等 Agent 数量从四个涨到几十个复杂度会在你看不到的地方指数膨胀。2. 轻量 Workflow 的边界与取舍不做什么比做什么更重要做轻量 Workflow 时最容易犯的错不是设计得太简陋而是做加法停不下来。我第一版方案也参考了很多开源编排框架的思路后来发现大部分重量级能力对我来说都是负担。轻量的核心不是代码少而是能力边界清晰你知道它擅长什么也知道它不擅长什么遇到不擅长的情况直接换方案而不是硬塞。2.1 必须保留的三类核心抽象轻量 Workflow 我只保留了三个抽象。第一个是节点Node。节点是流程的最小执行单元通常对应一个 Agent 调用或者一次工具调用。节点只管一件事接收上下文、执行动作、返回结果。我踩过的所有糟糕设计几乎都是因为节点承担了太多职责既能调 Agent又能改流程、又能做分支判断。最稳定的做法是让节点只做执行把下一步怎么走交给路由。第二个是路由Router。路由负责决策下一步它可以是硬编码规则也可以是让模型输出一个路由 key。很多 Agent 框架喜欢把路由写死在节点内部导致流程改动时必须动节点代码。我把路由独立出来之后绝大多数流程调整都变成了改配置、改 next_map不用碰业务代码省了非常多事。第三个是共享状态Context。所有节点共用一个可变的上下文对象。这个设计的优缺点都很明显优点是不用为每个节点定义复杂的参数签名信息要传就放进 Context缺点是容易变成一个大杂烩后面我会专门讲怎么给 Context 做减法。这三类抽象加上执行器的循环控制就是全部。没有持久化、没有子流程嵌套、没有可视化编辑先别急着补功能你会发现大部分场景用不上。2.2 果断抛弃的五种重量级能力我第一版方案参考了不少可视化编排平台的思路差点就往里塞了一堆能力后来逐条砍掉。砍的过程就是轻量方案定位的过程。能力取舍理由可视化拖拽界面去掉轻量 Workflow 的图是给开发者看的手写数据结构比拖拽快得多可视化往往意味着序列化格式复杂持久化与事件溯源去掉需要时自己在节点里加不要在执行器层面扛执行器保持无状态最简单子流程嵌套去掉嵌套会显著提高心智负担轻量方案用一个大流程就能覆盖绝大多数场景跨服务分布式调度去掉第一版根本不需要单体进程内跑就够分布式需求出现时已经该换重型方案了补偿事务/复杂重试去掉节点内部自己管重试执行器只管记录失败并走 fallback各司其职我个人的判断标准很简单如果某个能力不是为了解决Agent 之间的路径选择与状态传递就先不做。Workflow 不是来当万能框架的它是来给 Agent 协作画路线的。打个比方施工队进场装修包工头手里有一张施工计划图知道每个工种什么时候进场、什么时候撤场就够用了你非要他上一套仓库管理系统去管板材库存那属于本末倒置。3. 最小可用实现一个能跑生产环境的轻量编排骨架这一章给出可以照抄的骨架。我选择用 Python 的 dataclass 来表达因为它足够简单、可读性强。如果你习惯其他语言思路完全一样。3.1 数据结构用最简单的方式画一张有向图先定义节点和上下文两个基础类型。节点保存名称、处理函数、以及一份下一跳映射表上下文就是所有节点共享的袋子外加执行历史。from dataclasses import dataclass, field from typing import Any, Callable, Dict, Optional, List import time, uuid dataclass class WorkflowContext: data: Dict[str, Any] field(default_factorydict) trace_id: str field(default_factorylambda: uuid.uuid4().hex) current_node: str history: List[dict] field(default_factorylist) steps: int 0 def record(self, node_name: str, status: str, detail: Any None): self.history.append({ node: node_name, status: status, ts: time.time(), detail: detail, }) dataclass class Node: name: str handler: Callable[[WorkflowContext], Any] is_router: bool False next_map: Dict[str, str] field(default_factorydict) fallback: Optional[str] None为什么这样设计第一节点函数签名统一只接收 WorkflowContext这样任何节点都可以单独测试给一个模拟 Context 就能跑。第二next_map 就是图的边key 是路由返回的决策值value 是下一个节点名。路由节点返回什么 key执行器就跳到哪里。3.2 执行器逻辑调度、路由与循环闸class LightWorkflow: def __init__(self, name: str, max_steps: int 20): self.name name self.nodes: Dict[str, Node] {} self.max_steps max_steps def add(self, node: Node) - LightWorkflow: self.nodes[node.name] node return self def run(self, ctx: WorkflowContext, start: str start) - WorkflowContext: current start while current: node self.nodes[current] ctx.current_node current ctx.steps 1 if ctx.steps self.max_steps: ctx.record(current, error, {reason: max_steps_exceeded}) break try: result node.handler(ctx) ctx.record(current, ok, result) if node.is_router: key result if isinstance(result, str) else (result or {}).get(next) current node.next_map.get(key, node.fallback) if current is None: ctx.record(node.name, error, {reason: no_route, key: key}) break else: current node.next_map.get(__default__, ) except Exception as e: ctx.record(current, error, {reason: str(e)}) current node.fallback if current is None: break return ctx执行器就这么多。核心逻辑是普通节点跑完默认走__default__定义的边路由节点跑完用结果当 key 去查 next_map。每走一步都记录执行历史步数超过 max_steps 就强制终止这是防止 Agent 自循环的最后一道闸门。轻量指的就是这几十行逻辑能解决全部路径控制问题剩下的业务都留在节点 handler 里。下面给一个示例流程三个节点意图判断路由、执行任务、异常兜底。wf LightWorkflow(demo) def judge(ctx): # 假设这里调用 LLM根据用户输入返回 search 或 rewrite return {next: search} def search(ctx): ctx.data[result] search done return {next: done} def rewrite(ctx): ctx.data[result] rewrite done return {next: done} wf.add(Node(start, judge, is_routerTrue, next_map{search: search, rewrite: rewrite, unknown: search})) wf.add(Node(search, search, next_map{__default__: done})) wf.add(Node(rewrite, rewrite, next_map{__default__: done})) wf.add(Node(done, lambda ctx: None))这段代码一眼就能看出执行路径长什么样。流程要改改 next_map 就行要新增分支加一个节点加一条边就行。它的最大价值在于图是显式的而不是藏在 20 层 if-else 里。3.3 状态管理共享上下文必须学会做减法共享 Context 最自然的使用方式是把每一步的原始结果都存进去。问题是 LLM 对上下文长度有硬限制你不可能把几万字的中间结果都塞给下一个 Agent。我实践下来的规则是节点产出后先归纳成摘要再放 Context原始大对象放磁盘或缓存Context 里只放引用。def context_trim(ctx: WorkflowContext, keep_keys: List[str]): for k in list(ctx.data.keys()): if k not in keep_keys: del ctx.data[k] return ctx不要小看这一步。我线上出过一次事故某个 Agent 的上下文里塞了十几篇文章全文token 数直接爆掉最后整条流程静默失败。从那以后我每个节点结束都会主动做减法宁可后面节点按需去缓存读原文也不让 Context 无限膨胀。还有一点路由节点的结果要尽量短判断结果就一个 key不要让它把大段推理过程也塞进共享状态推理过程写日志就够了。3.4 执行轨迹日志低成本可观测性执行器跑完拿到的是结构化 history。不要只把它留在内存里最简单的方式是每条记录转成一行 JSON 写进日志。def log_trace(ctx: WorkflowContext): for rec in ctx.history: print(json.dumps({ trace_id: ctx.trace_id, node: rec[node], status: rec[status], ts: rec[ts], detail: rec[detail], }, ensure_asciiFalse))这样每跑一次流程日志里都有完整的节点执行轨迹。配合 trace_id 搜索等于给每个流程实例发了一张全景图这套东西不到 20 行代码但上线排查时能救命。4. 实测三个月里我踩过的四个深坑骨架搭好之后真正的磨难才开始。我把三个月里印象最深的问题按排查过程拆开讲每个都是先现象、后定位、再修复你可以把排查链路直接拿去参考。4.1 节点重复执行LLM 超时重试引发的重复副作用现象很简单某次线上运行用户收到了两条内容完全相同的结果通知可流程日志里明明只有一次成功的生成。一开始我怀疑是用户侧重复点击查了入口层没有重复请求。然后把注意力放到 Agent 的 HTTP 调用上下游服务的 SDK 默认开启了超时重试第一次调用其实已经成功但因为网络抖动响应超时SDK 自动重发了一次于是两个请求都完成了任务通知也就发了两遍。定位思路是先看 trace。同一个 trace_id 下面该 Agent 节点只记录了一次ok但下游服务日志里出现了两个时间间隔不到一千毫秒的相同请求。这说明重试发生在节点内部执行器完全感知不到。这类问题的通用解法是幂等节点内部维护一个已执行集合执行前先检查这个 Key 是否已经做过了。Key 用 trace_id 加操作类型加内容哈希生成。修复后我重放故障流量通知只发出一次。从那以后所有会产生副作用的节点都必须自带幂等保护不能指望执行器。4.2 上下文膨胀把整个库里文章都塞进了 state这个坑在上一节简单提过。过程是这样的资料检索节点为了图省事把所有命中文章的正文直接塞进 Context后面做摘要和生成报告的节点必须读取这些正文。最开始文章少没发现后来检索结果上百条、每条几千字某个模型接口开始频繁报上下文长度超限。诡异的是报错不是每次都出现因为超限取决于当次实际命中条数。做排查时我第一反应是看 token 消耗曲线发现某个节点调用前后 token 数差了将近三倍才意识到 Context 早已失控。修复方案就是上面说的摘要进状态、原文进缓存。摘要是给后续 Agent 做全局判断用的原文是给最终生成节点按需读取的两件事必须分开。我后来还加了一道保险节点 handler 入口处统计 ctx.data 的总大小超过阈值直接报错。宁可显式失败也不让它在某个 AI 接口上静默爆炸。4.3 Agent 自循环没有步数上限就等着烧 token第三个坑是最典型的 Agent 自循环。某个 Agent 负责生成回复并自带质量检查逻辑如果它觉得结果不够好就返回 redo流程跳回生成节点重新来一遍。问题来了模型有时候会陷入反复自我否定的循环同一段内容生成七八遍之后依然觉得不满意于是不断 redo。如果没有上限这个过程会一直烧 token 烧到会话超时。排查时我在执行器里临时加了日志发现历史记录中节点的访问序列是 A - B - A - B 无限往复而且每次 B 返回的结果 key 都是 redo。这里的本质问题是流程没有把同一节点重复执行次数作为终止条件。修复方法是给每个节点加一个 visit 计数器某个节点连续进入超过 3 次路由就强制切到 fallback让模型换一种处理方式或者直接转人工。另外跳回上游节点时要顺手清掉触发循环的那些 Context 字段否则下一次判断还会得到同样的结论。max_steps 能兜底但不能解决问题明确循环必须被打断的规则才是关键。4.4 故障定位困难没有 trace 的编排等于黑盒最后一个坑不是功能问题而是排查体验问题。项目早期流程一旦出错我经常要同时开三四个窗口业务日志、模型调用日志、数据库状态然后手动拼出是哪一步出的错。有一次线上反馈某个流程总是卡住我却只能靠猜。后来我把执行器的历史记录完整 dump 成结构化 JSON每一行包含 node、status、耗时、关键入参、出口 key再把 trace_id 贯穿到所有下游调用的日志里。改造之后排查一个坏流程通常只要一分钟看一眼执行历史就知道卡在哪个节点、路由决策 key 是什么、异常在哪一行。我还顺手写了一个小脚本把执行历史渲染成可读的路径图树状文本就够用对复盘和给同事解释都很有帮助。轻量 Workflow 的执行器很小这恰恰意味着你可以把执行历史当成一等公民来记录这是重型引擎反而不容易做到的事。5. 什么时候该果断放弃轻量方案轻量方案不是银弹。它最适合的是单体进程内、步骤有限、不需要长时间挂起、不需要多人协作开发和审计的 Agent 流程。如果你的场景突破了这几个假设请果断放弃别硬撑。5.1 三个必须切换重型方案的危险信号第一个信号是流程需要跨服务、跨团队协作。轻量方案的执行器是单体内存循环天然不考虑进程崩溃恢复。一旦流程跑到一半服务重启状态就丢了。要么你自己去实现持久化和恢复要么直接换有持久化能力的编排平台。这个门槛其实很低当你的流程里出现这一步必须等三天后的回调轻量方案基本就不适用了。第二个信号是流程涉及正式的人工审批环节。需要完整审批链和审计日志的场景通常还要求每一步操作都可以追溯、可以解释。轻量 Workflow 的历史记录足够你 Debug但不够你应付合规审计。这类需求别再自己写直接上重型方案。第三个信号更加隐蔽流程数量多、版本管理严格、需要多人并行修改。轻量工作流用代码描述流程必然要求所有人都理解同一套代码结构。当团队有五个人以上同时改流程代码合并冲突和流程图漂移会不断消耗你。这时候需要的不是更聪明的执行器而是规范化的流程描述和版本管理机制。判断标准可以用一句话概括如果你的流程可以用一张 A4 纸画清楚轻量方案就够用当它需要一张 A0 大图还画不完就该升级了。5.2 从轻量到重量的平滑过渡路径我个人的建议是不要一上来就上重型引擎。先用轻量方案把流程跑熟把节点、路由、数据依赖关系理清楚这个过程会把很多隐藏的业务逻辑暴露出来。等你确定这套流程的价值和稳定性之后再带着已经验证过的流程定义去选型重型平台迁移就是纯粹的机械工作风险要小得多。反过来我也见过有人直接上了重型平台做一个小流程结果大量时间耗在平台配置、权限、环境问题上业务反而没进展。轻量方案最大的价值不只是省时间而是帮你更快看清问题本身——流程到底长什么样、哪里容易断、哪里需要人介入。这些东西搞明白了选什么平台心里都有数。最后再分享一个小技巧我一直为轻量 Workflow 保留一个简陋但实用的回放工具。每次运行结束后把执行历史丢给它它就能生成一条带耗时标记的路径路线图。排查问题、跟业务对齐预期、临时做 demo都特别好用。这套东西不复杂但用着用着就离不开了。