企业官网建设流程全解析

做 Multi-Agent 应用做到一定规模，你一定会撞上同一堵墙：agent 一多，消息满天飞，任务卡死，谁该等谁、谁该先动，全都说不清。这时候你才会真正意识到，多 agent 之间稀缺的资源不是大模型的 token，而是一套能让它们有序协作的通信协议与编排机制。这篇文章就来拆一个我最近在落地的编排中枢设计，核心三件套：状态机管生命周期，DAG 管任务依赖，事件总线管消息解耦。读完你至少能回答三个问题：为什么单靠 prompt 约束 agent 协作靠不住？状态机、DAG、事件总线各自到底解决什么问题、边界在哪？把它们拼成一个实际系统时，又会在哪些细节上翻车？

1. 没有编排中枢的 Multi-Agent，本质是一群失控的协作者

1.1 多 agent 不是多线程：每个 agent 都有“自己的意志”

很多人第一次写多 agent 系统，会下意识把它当多线程来用：开几个线程并行跑、加锁、同步，以为这样就能协作。但 agent 和线程有一点根本区别——线程没有自我意志，agent 有。

一个 agent 拿到任务之后，它会自己做判断：先看哪份资料、调用哪个工具、什么时候觉得“我干完了”。如果你不约束它，它可能在你预期之外的地方停下来，也可能反复做同一件事而不自知。更麻烦的是，多个 agent 同时按自己的逻辑工作，彼此之间的交互靠互相发消息，消息又没有一个统一的格式和路由规则，系统最后就会变成一个谁也管不住的消息风暴。

我见过一个真实案例：一个客服工单处理应用里，分类 agent 和分析 agent 同时收到工单，各自跑完之后都往“下一步”发指令，结果回复生成 agent 被触发了三次，生成了三份相互矛盾的回复草稿。这不是模型能力问题，是编排问题——系统没有一种机制告诉每个 agent“你现在处于什么阶段、该由谁在什么条件下触发下一步”。

1.2 乱象的三个典型症状，你大概率都见过

我把多 agent 系统失控的典型症状归纳成三类，大家可以对号入座：

• 状态不收敛：同一个任务，有的 agent 认为它还在处理中，有的 agent 认为它已经失败，没有统一的生命周期管理，最终谁也说不清这个任务到底算什么状态。
• 依赖不明：任务之间没有“先决关系”的表达，所有 agent 都并行启动，靠运气决定谁先产出谁后消费，结果常常是后续 agent 拿不到前置数据，空转等待。
• 通信无秩序：agent 之间直接点对点调用，没有统一的消息通道。一旦某个 agent 挂了，调用方立刻就阻塞，错误沿着调用链层层传递，最后整个任务进入不知所云的状态。

这三个症状的共同根源，是缺少一个“中枢”。注意，我这里说的中枢不是一个中央处理器替 agent 思考，而是一个负责通信协议、状态管理、拓扑编排的基础设施层。它不告诉 agent 怎么理解问题，但告诉 agent 什么时候该跑、什么时候该等、什么时候算完。

1.3 编排中枢的三根支柱，正好对应三类问题

我自己在项目里把编排中枢拆成三根支柱，每一根精准解决一类问题：

这三样不是可选的“锦上添花”，而是多 agent 系统走到一定复杂度之后的基本生存工具。下面我分别展开讲每一根支柱的细节，以及它们为什么必须是长成这个样子的。

2. 状态机：把 agent 的一生管起来，从诞生到销毁都有章法

2.1 经典状态集：为什么就这几个状态够用

单 agent 的状态机设计，我建议先从最小状态集开始，不要一上来就搞十几个状态，否则你自己都会晕。

我目前最常用的一套状态集是这样的：

• idle：已创建，等待被调度
• running：正在执行，持有执行权
• waiting：执行被挂起，通常在等某个事件、某个人审阅或者外部的异步确认
• blocked：依赖条件不满足，卡住了
• error：执行出错，需要补偿或者人工介入
• done：正常完成，输出已落库

为什么要这么设计？因为它刚好覆盖了“一个 agent 从被创建到释放资源”的全部生命周期阶段，而且每个阶段都有明确的“我接下来能干嘛”的约束。比如 running 状态下，它不能同时做两件事；waiting 状态下，它不能自行往下推进，必须等一个外部事件把它重新顶到 running。

很多初学者把状态机理解成“给运行中/已结束起个名字”，这是不对的。状态机的价值在于让非法跃迁变得不可表达——一个 done 的 agent 不应该再收到运行指令，一个 waiting 的 agent 不应该自己觉得“我等够了”就去继续干。如果这些判断都靠 if-else 写在业务代码里，早晚会有个开发者漏掉一个分支，然后系统在一个奇怪的状态里运行一整晚。

2.2 状态转移表：谁触发了跃迁，谁就要负责响应

状态机的核心产物不是状态列表，而是状态转移表。我在设计的时候会先画一张表，明确每个状态在什么事件下跃迁到什么状态，谁负责触发这个事件：

这张表看起来简单，但它强制你做了一件特别重要的事：每个状态跃迁都有明确的责任方。比如“input_received”这个事件，会让一个 waiting 状态的 agent 重新回到 running，这个事件不是 agent 自己瞎编的，而是事件总线上一个真实消息触发的结果。这就在状态机、DAG、事件总线三者之间埋下了第一根连接线。

我会把这个转移表实现成一个配置化的状态机引擎，而不是每写一个 agent 就重写一套 switch-case。引擎只需要知道三件事：当前状态、收到的外部事件、转移表，就能推导出下一状态并调用对应的动作。这样即使未来增加新的 agent 类型，状态机引擎本身也不需要大改。

2.3 状态机的部署形态：外部统一管理比 agent 自治更稳

关于状态机到底放在 agent 内部还是放在编排中枢统一管理，我踩过坑后坚定选择后者。

第一版实现里，我让每个 agent 自己维护状态，美其名曰“自治”。结果很快就发现，agent 的 LLM 调用一旦超时或者 token 被截断，它自己的状态就卡在一个半死不活的位置，外部根本不知道。而且多个 agent 各自有一套状态命名，A 里的“finished”和 B 里的“complete”其实是同一个意思，调试的时候光是翻译状态名就让人头大。

统一管理的做法是：状态存放在编排中枢的状态存储里，agent 只上报事件，由中枢根据转移表计算并持久化新状态。例如 agent 跑完一个子任务，不是写“我完成了”，而是发一个AgentCompleted(task_id, agent_id, output)事件给中枢，中枢查转移表，把该 agent 的状态从 running 改成 done，然后广播AgentStateChanged事件给所有关心这个状态的模块。这相当于所有 agent 都共用一个“状态账本”，谁都不会谎报军情。

3. DAG 编排：把“谁先谁后”画成图，而不是写死在代码里

3.1 为什么用 DAG，而不是工作流脚本或者自由对话

多 agent 协作里有一个核心问题：产物之间是有依赖的。举一个很普通的例子——写一份市场分析报告，通常得先由“搜索 agent”收集素材，再由“分析 agent”加工成观点，最后由“写作 agent”落笔成文。这里的依赖关系是明确的，前一步不完成，后一步启动就是浪费算力和时间。

表达这种依赖关系，业界已经验证过的最稳定方案就是有向无环图（DAG）。节点是任务或 agent 实例，边表示“依赖关系”。为什么不能是有环图？因为有环意味着循环依赖，A 等 B，B 又等 A，如果不加干预，这个任务永远无法结束，等于程序死锁。用 DAG 就从根本上堵死了循环依赖——图如果有环，拓扑排序做不出来，直接报错。

有人可能会问：为什么不用 if-else 脚本？如果任务只有四个步骤，脚本确实够用。但多 agent 任务的分支、合并、并行组合一旦多起来，脚本会迅速爆炸成一团没人敢改的面条代码。DAG 的优势在于它是数据，不是代码。你可以任意调整节点、增加并行分支、查看整张执行计划图，甚至可以序列化成 JSON 发到前端可视化。这种灵活性，脚本很难给到。

3.2 编排执行周期：构建 DAG、拓扑排序、按依赖触发

DAG 编排模块的工作流程，我习惯分成三步：

1. 构建 DAG：根据任务类型，从模板库加载一张拓扑图，或者由上层业务动态生成。比如“写报告”模板的拓扑可能是：collect -> analyze -> write -> review，其中 analyze 还可以拆成parallel: [insight1, insight2]。
2. 拓扑排序：每次运行前，对图做一次拓扑排序，得到执行顺序。这一步顺便能发现环——如果排不出来，立刻抛异常，不让任务带病上路。
3. 按依赖触发：调度器遍历拓扑序列，当一个节点的所有上游节点都标记为 done，就把该节点关联的 agent 状态从 idle 置为 running，并发出启动指令。

第 3 步的主动触发机制是关键。DAG 里每个节点完成之后，调度器会做一次“依赖满足检查”，看看有没有新的节点从“等上游”变成“可以跑”。这个过程就是前面说的prerequisite_met事件来源，也是 DAG 和状态机的握手环节。

用伪代码描述核心调度循环的话，大概是这样的：

def on_node_done(node): for nxt in dag.get_downstream(node): upstreams = dag.get_upstream(nxt) if all(state_store.is_done(u) for u in upstreams): state_store.transition(nxt, "idle", "running") launch_agent(nxt)

这个循环极其稳定，因为它只有两个动作：检查依赖、触发启动。不会出现“某个节点被启动了两次”或者“没等齐就开工”这类低级问题。

3.3 失败与重试：DAG 里的局部回滚策略

DAG 编排最难的地方不是正常路径，而是失败路径。一个节点失败之后，你应该把它下游的所有节点都取消掉吗？还是让下游先跑别的分支？

我的经验是：默认取消失败节点的所有直接下游，把它们的状态重置为 blocked，由人工或业务规则决定是否重跑。原因很朴素：下游消费的是上游的产物，上游失败了，下游拿到残缺数据，做出的结果大概率是错的，不如干脆停下来。这个“停止传播”的行为我塞进了调度的dispatch_failure逻辑里：

def handle_failure(node): state_store.transition(node, "running", "error") for down in dag.traverse_downstream(node): state_store.transition(down, "waiting", "blocked") notify_event_bus("TaskBranchFailed", node)

这样一次失败不会拖垮整张图，只是把受影响的分支冻结，等某个事件（比如人工重试指令）到来时，再把对应节点集体放行。如果你希望系统更激进一点，可以配置自动重试：第一次失败直接重跑当前节点，重跑机会耗尽后才触发分支冻结。重试逻辑放在调度器里而非 agent 里，因为 agent 自己往往不具备“我应该重试”的全局判断力。

4. 事件总线：让 agent 之间不用互相点名，也能协同干活

4.1 点对点通信的教训：调用链纠缠，是系统崩坏的开始

我在通讯协议这个问题上走过弯路。最早我让 agent 之间通过 HTTP 互相调用：A 调 B 的接口拿结果，B 再调 C 的接口。这相当于让每个 agent 都记住一大串“谁在哪儿、谁什么时候上线”。一旦某个 agent 挂了或者换了地址，所有依赖它的 agent 全要跟着改配置。

这种点对点的通信方式还有一个更隐蔽的坏处：调用链太长之后，一次任务的失败根本无法定位。到底是 B 调 C 超时了，还是 A 解析 B 的返回时出了错？你只能顺着调用链一路打日志，一个节点一个节点地扒。这绝不是多 agent 系统该有的体验。

事件总线扭转这个局面的方式很简单：所有 agent 都不需要知道消息从哪来、到哪去，它们只往总线上发事件、订阅自己关心的事件。通信协议从“你调我我调你”变成“我发布你订阅”，耦合度瞬间降下来。这也让每一段消息传递都可以被记录、重放、审计。

4.2 事件类型学：event、command、ack 的语义边界

事件总线上并不是所有消息都一个用法。我在实践中严格区分三种消息语义：

• event（事件）：表示“已经发生了一件事实”，例如FileUploaded、AgentCompleted。事件是过去时，接收者收到后只能响应事实，不能拒绝“它已经发生”这一点。
• command（指令）：表示“请你去执行某件事”，例如GenerateReply。command 是将来时，接收者可以选择执行或者抛错。
• ack（确认）：表示“我收到了且打算处理”，用来做消息投递确认，配合至少一次投递语义使用。

区分这三者最重要的意义，在于设计接收方的容错策略。事件可以重复处理（配合幂等键去重）；command 需要一个明确的执行者，不能被多个 agent 同时抢着执行；ack 则是事件总线不能丢消息的最后一道保险。如果你把这些语义混在一起，比如让多个 agent 订阅同一个 command 流，消息一进来，所有 agent 都会觉得自己有义务执行，这就是重复执行的温床。

4.3 消息契约：schema、版本、幂等、死信，一样都不能省

事件总线看似只是“发消息、收消息”，实际要让它在生产环境可靠运行，必须提前把契约定明白。我这里列出四个最容易忽略的点：

• 消息 schema 必须固定：每条事件都用 JSON Schema 或 Protobuf 定义字段，字段增删要走版本号。我见过直接在事件里加自定义字段的，结果消费方 parser 直接崩掉，全链路告警。
• 幂等键必须自带：任何事件都带一个由task_id + step + attempt生成的idempotency_key，消费方在本地做去重，保证同样的事件即便被投递两次，业务副作用也只执行一次。
• 事件要支持重放：总线的消息必须持久化一段时间，至少保留一周。因为一旦消费方的代码有 bug，修好之后你大概率需要“从现在开始重放过去一小时的事件”来把状态补平。
• 死信队列必须存在：消费方反复处理失败的消息，不能被静默丢弃，要进入死信队列，并触发告警。否则你会遇到一种诡异情况——某个 agent 一直在处理一条坏消息，后面的任务全部堵死，但没有任何日志提示。

这些契约层面的规矩，比选哪个消息队列中间件重要得多。用 Kafka、Redis Stream 还是 NATS，只是性能差异；契约不规范，用什么都会乱。

5. 一个编排中枢的落地样板：工单分析流水线

5.1 系统拓扑与数据流

理论讲多了容易飘，我拿一个实际落地的项目来说明三件套如何协同工作。这个项目的需求是：收到用户投诉工单后，系统自动完成“分类 -> 摘要 -> 回复起草 -> 人工审核 -> 回复发送”全流程，中间涉及五个不同职能的 agent。

整个系统的数据流是这样的：

1. HTTP 网关收到新工单，进入编排中枢，构建一张 DAG：classify -> summarize -> write_reply -> (waiting) human_review -> send，其中 classify 完成后，summarize 和 write_reply 可以并行启动前半段，但 write_reply 需要先拿到 classify 的分类标签，所以它跟 summarize 并非完全并行，而是有个先决条件。
2. 调度器对 DAG 做拓扑排序，把classify节点的状态从 idle 置为 running，启动分类 agent。
3. 分类 agent 把结果以AgentCompleted事件发到事件总线。调度器收到事件后，查询状态转移表，把 classify 标记为 done，再做一次依赖检查，发现 summarize 和 write_reply 的下游条件均满足，于是把这两个节点的 agent 状态切到 running 并启动。
4. write_reply 完成后进入waiting状态，等待HumanReviewed事件。人工审核页面上点了“通过”，前端不发 API 直接调 agent，而是往事件总线发一条HumanReviewed事件，调度器收到后把 send 节点放行，发送 agent 执行最后一步。
5. 整个任务结束时，中枢把 DAG 所有节点标记为 done，并把整条执行链路的 traceId 存进数据库，后续排查问题全靠它。

5.2 核心伪代码：从 DAG 构建到调度循环

编排中枢的调度器核心伪代码，我压到最简版本，方便你照这个思路去搭：

class Orchestrator: def __init__(self, state_store, bus, dag_builder): self.state_store = state_store self.bus = bus self.dag_builder = dag_builder def submit_task(self, task): dag = self.dag_builder.build(task.type, task.payload) state_store.create_task(task.id, dag) for node in dag.roots(): state_store.transition(node, "idle", "running") self.launch(node) bus.subscribe("AgentCompleted", self.on_agent_completed) bus.subscribe("AgentFailed", self.on_agent_failed) def on_agent_completed(self, event): node = state_store.get_node(event.task_id, event.agent_name) state_store.transition(node, "running", "done") for nxt in dag.get_downstream(node): if all(state_store.is_done(u) for u in dag.get_upstream(nxt)): state_store.transition(nxt, "waiting", "running") self.launch(nxt)

这里的launch并不是直接调一个 agent 的入口函数，而是往事件总线上发一条StartAgent的 command。至于谁来执行这个 command，可以是本地进程内的 worker，也可以是远程的一台独立部署 agent 服务。这个设计保持了编排中枢对所有执行单元的彻底解耦。

5.3 可观测性设计：一条 traceId 贯穿全部节点

多 agent 系统最让人崩溃的就是排查问题。为了减少这种痛苦，我在编排中枢里做了一个硬性要求：每个任务从生命第一天就绑定一个全局 traceId，所有 agent 的输出、所有事件的日志、所有状态跃迁记录都必须带上这个 traceId。

这样当你发现某个工单最终没有发出回复，你只需要做一件事：在事件总线日志里搜这个 traceId，然后把时间轴拉出来。你会看到类似这样的节奏：12:00:01 AgentCompleted(classify)、12:00:03 StartAgent(write_reply)、12:00:05 HumanReviewed(submitted)……一眼就能定位到哪个环节断了，而不需要在五个 agent 的日志目录里挨个翻。这一条经验，在系统规模超过三个 agent 之后，几乎是救命级别的。

6. 踩过的坑，希望你别再踩：状态不同步、DAG 成环、事件风暴

6.1 状态不同步：分布式系统的幽灵

我的第一版编排中枢，每个 agent 的状态同时存在 agent 进程内存和中央状态库里，更新的时候偶尔图省事只改了内存。结果战区汇报时，状态库显示某个工单还在 running，内存里 agent 早已 done。这种状态不同步问题，比功能 bug 可怕得多，因为它不会立刻让系统崩掉，而是让状态逐步腐化，最后你发现所有 recover 逻辑都是错的，因为存储的“真相”早就被污染了。

改进方案只有一个：状态更新的唯一权威，必须在中央状态存储。agent 进程内存里的状态只作为本地缓存，绝不参与决策。任何一次状态跃迁，都是调度器先写数据库，再广播事件。内存状态一旦和存储不一致，以存储为准，并且立刻告警。这样虽然多了一次数据库读写的开销，但换来了全局一致性。

6.2 DAG 里悄悄长出的环

用 DAG 的本意是杜绝环，但实际开发中依赖关系往往是函数式的，改着改着就会发现：节点 A 的完成事件被用来触发 B，B 的依赖检查又反过来引用了 A 的某个中间结果，形成了一个逻辑上的环。

我测试过很多 DAG 场景，发现一个有效的防线是：在 DAG 构建阶段就把“是否有环”作为硬性校验，有环就直接拒绝发布任务，而不是等到运行时去撞南墙。具体实现可以简单粗暴一些，每次传入新的依赖边时，做一次拓扑检测，如果引入该边会让拓扑排序失败，就拒绝这条边的注册。这个校验只花十几毫秒，但能把无数小时的生产事故扼杀在摇篮里。

6.3 事件风暴：同一个事件被多个消费者抢着处理

事件总线的高吞吐也会带来麻烦。我最开始用订阅模式时，一个AgentCompleted事件会被好几个消费者同时处理，有的消费者只是想做监控统计，有的却想据此启动下一个节点。结果就是同一个事件被当成“启动下一个节点”的指令处理了两次，重复执行了整整两条链路。

如果系统对消息投递有至少一次（at-least-once）保证，那么消费方侧必须用幂等键去重。我在事件总线上给所有事件统一加了dedup_key字段，消费方在处理前先查 Redis：这个 key 见过没有，见过就直接返回，没见就写 key 再执行业务逻辑。这件事一开始做的时候有点繁琐，但确实可以有效地兜底重复投递那一类问题。

6.4 超时补偿：别让任务永远“等待中”

前面讲的状态机里，waiting 状态是最容易出问题的。一个 agent 在等一个用户确认事件，理论上这个事件一定会来，但如果用户一直不点确认呢？任务就永远挂在 waiting 状态。我遇到过几次之后，现在所有 waiting 状态都默认配上超时补偿机制：waiting 超过预设的 TTL，自动生成一个WaitTimeout事件，转移到 error 或者 blocked，同时把整条任务链路推给人工跟进队列。

超时补偿看似是一个小细节，但只要缺了它，你的编排中枢就随时可能积压一大堆处于“僵尸状态”的任务，既占用数据库空间，又污染监控面板，还可能干扰后续 DAG 的依赖判断——上游明明没有完成，下游却傻等。

最后说一点个人体会。我在搭完这套状态机、DAG、事件总线三件套之后，最大的变化不是系统变稳了，而是我的开发方式变了：现在每接一个新场景，我会先画三样东西——状态转移表、DAG 依赖图、事件清单，这三样定稿之后才开始写业务代码。多 agent 系统的复杂度大部分集中在协作关系上，而不是单个 agent 的聪明程度上。先把协作关系用状态机、DAG、事件总线这些“硬结构”固定下来，剩下的空间才敢交给大模型去自由发挥。这也是我目前能够有效控制多 agent 工程复杂度的唯一路线，希望这套思路对你也同样有用。