多Agent协作下的轻量Workflow设计：从if-else到显式流程编排

过去大半年，我一直在做一件不算性感但特别磨人的事：把一套靠人肉连线、层层 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 果断抛弃的五种重量级能力

我第一版方案参考了不少可视化编排平台的思路，差点就往里塞了一堆能力，后来逐条砍掉。砍的过程就是轻量方案定位的过程。

我个人的判断标准很简单：如果某个能力不是为了解决"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_factory=dict) trace_id: str = field(default_factory=lambda: uuid.uuid4().hex) current_node: str = "" history: List[dict] = field(default_factory=list) 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_factory=dict) 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_router=True, 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_ascii=False))

这样每跑一次流程，日志里都有完整的节点执行轨迹。配合 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，都特别好用。这套东西不复杂，但用着用着就离不开了。