OpenRig多智能体编排框架：从Agent孤岛到可恢复协作系统

先聊几句题外话。最近这一年，我身边做 AI 应用的朋友几乎都在同一个问题上卡壳：单个 Agent 跑 demo 很惊艳，一旦放进真实业务里，立刻变成“一个聪明但没组织的实习生”——你给它一个任务，它能干，但你要它和别人配合，它就彻底乱套。你需要的其实不是更多更强的 Agent，而是一套能把它们“织”起来、还能在故障后恢复现场的系统。OpenRig 正是我在这类实践中沉淀下来的一个多智能体编排框架，核心就两件事：把离散的 AI Agent 组织成协作网络，以及让这种协作具备持久化能力。这篇博文我会把设计思路、核心机制、落地细节和生产环境里踩过的坑一次性讲透。

先说清楚它解决什么问题。假设你的业务里已经有三个 Agent：一个负责抓取信息，一个负责分析，一个负责生成报告。最粗暴的接法是代码里硬编码“A 调完调 B，B 调完调 C”，每个 Agent 之间用 HTTP 互相呼唤。这种模式撑到十几个 Agent、几十条任务链路时就会失控：你不知道现在任务跑到第几步，某个 Agent 挂了之后整个链条无法恢复，并发一上来接口互相等待直接雪崩。OpenRig 的答案是：Agent 不直接互相调用，而是把任务描述、消息路由、状态存储、失败重试全部收归到一个编排层，由编排层统一调度。这样每个 Agent 继续保持“离散”的独立性，但整体变成一个有状态、可恢复的协作系统。

我见过不少团队一上来就追求复杂的 Agent 协议、自研通信框架，结果光基础设施就写了三个月。OpenRig 给我的启发是：编排的重点不在“灵巧”，而在“可控”。所以这篇内容，也适合那些正在从单 Agent demo 走向多 Agent 生产落地的开发者，尤其适合后端背景、对系统稳定性有要求的人。

1. 为什么单机 Agent 撑不起真实业务：OpenRig 要解决的三个核心问题

1.1 离散 Agent 的“孤岛效应”

单个 AI Agent 本质上是一个“输入-处理-输出”的独立单元。选型时大家习惯把 Agent 做得越“纯粹”越好：这个 Agent 只负责阅读理解，那个 Agent 只负责写代码。听起来很模块化，但真实业务不是模块化流程，而是一张互相依赖的网。一个需求分析 Agent 的结果要喂给方案设计 Agent，方案又分成前端、后端、测试三个分支，三个分支完成后合并成交付物。如果每个 Agent 都只关注自己的输入输出，没人知道全局数据流长什么样。

孤岛效应的直接后果是上下文断裂。A 产出的信息放在自己的 memory 里，B 完全看不到，于是 B 只能让用户重新提供一次数据，或者 A 在自己的回复里把数据带上——这又导致 token 浪费和上下文污染。OpenRig 的做法是引入一个共享的消息层，消息被编上 task_id、agent_id、sequence_id，任何一个 Agent 都可以订阅它关心的消息，而不是靠人工在代码里把 A 的输出拼到 B 的输入上。

1.2 临时调用和持久协作的差距

很多团队做 Agent 集成的第一步是“用代码把几个 API 串起来”。这种临时调用（transient invocation）有几个致命问题：没有状态、没有重试、没有审计。任务进行到第 3 步时服务重启了，整个链路从零开始；某个 Agent 调第三方 API 超时，你甚至不知道它到底有没有执行成功；事后想复盘“为什么这次回答质量差”，找不到任何中间过程记录。

我把这比作开公司：临时调用相当于大家用私人微信沟通工作，出问题后聊天记录删了就没证据；持久化协作系统相当于公司信息管理系统，每一次任务派发、状态变更、结果提交都在系统里留痕。OpenRig 的核心不只是一个调度器，它是一套以事件日志为底座的协作系统——一切协作行为都是事件，事件落库，系统状态由此重建。

这也是“持久化协作系统”这个关键词里“持久化”的真正含义：不是把几个结果存进数据库就算完，而是整个协作过程的状态都具备可重放性。系统崩溃了，你从最近一个快照恢复，然后重放未完成的事件，所有 Agent 就知道自己该继续干什么，而不是茫然地重新开始。

1.3 面向真实场景：OpenRig 适合谁用

根据我的实践经验，OpenRig 最适合这些场景：

• 业务流程固定但步骤较多的场景，比如“资料收集 → 数据清洗 → 分析建模 → 生成报告”，每步有明确输入输出。
• 需要多个 Agent 协作产出同一交付物的场景，比如“前端 Agent + 后端 Agent + 测试 Agent 共建一个功能”。
• 对可审计性有要求的场景，比如金融数据分析、企业内部知识库问答，事后需要知道某条结论是哪几个 Agent、用了哪些数据得出来的。

如果你只是做一个“问一句答一句”的单轮聊天机器人，那用不上 OpenRig，LangChain 直接调模型就够了。但只要你发现 Agent 之间开始互相传数据、前后结果有依赖、系统崩溃后想恢复进度，它就是值得考虑的方案。

2. 把 Agent 织成网的四个核心机制

2.1 通信层：Agent 之间不搞点对点，全部走消息总线

我在设计 OpenRig 时给自己定了一条铁律：Agent 之间不允许直接调用。两个 Agent 要协作，只能通过消息总线（Message Bus）交换消息。这样做有三个直接好处：第一，Agent 的耦合度降到最低，替换一个 Agent 实现不影响上下游；第二，所有消息都经过总线，自然可以统一做鉴权、限流、审计；第三，消息的投递状态可以被总线跟踪，谁发出去没收到、谁收到了没回复，一目了然。

消息结构我建议至少包含这些字段：

这整套机制说起来简单，但实际踩坑不少，最要命的是“消息丢失”问题。我自己的方案是引入“发送确认”，收到方处理完毕后必须回一条 ack 消息，发送方在一定时间内收不到 ack 就重新发送，配合消息表里的状态字段做去重，避免重复消费。

2.2 编排模型：用任务图描述“谁先做、谁能并行、谁失败了怎么办”

有了通信层之后，下一步是解决“流程怎么走”的问题。OpenRig 里用的是有向无环图（DAG）编排模型，这算是目前多智能体编排领域的主流做法。每个节点代表一个 Agent 或一个执行单元，节点之间的边代表依赖关系。这样你一眼就能看出：

• 哪些节点可以并行跑（入度为 0 的节点）
• 哪些节点必须等前置节点完成（有依赖的节点）
• 整个流程的瓶颈在哪（最长链路）

DAG 的好处是天然防死锁。只要你的依赖图满足无环性质，调度器就一定找得到执行顺序。实际落地时我用 YAML 描述编排定义，比纯代码可读性好很多，业务人员也能看懂。

2.3 持久化状态：事件溯源加快照，重启不是重来

讲编排不可能绕过持久化。我的观点是：事件溯源（Event Sourcing）是持久化协作系统的最佳实践。解释一下，传统做法是保存“当前状态”，比如任务表里记录“进行中”“已完成”；事件溯源则是把每次状态变化当成一条不可变的事件记录下来，状态只是事件的投影。

举个例子，任务从“排队中”变成“执行中”，不是去 UPDATE 一条记录，而是 INSERT 一条“任务已启动”的事件。下次要恢复状态，把事件日志从零开始重放一遍，或者基于最近的快照加上后续事件重放，你得到的一定是确定性的结果。

我常用的组合是“快照 + 事件日志”双写：每处理完 100 个事件生成一个快照，恢复时先载入快照，再从快照位置往后重放增量事件。这样既保证确定性，又不需要从第一天开始把所有历史事件重放一遍，性能也能接受。

2.4 可观测性：一次协作窗口的完整追踪链

多 Agent 系统比传统微服务更难排查问题，因为每个 Agent 都可能调用 LLM，LLM 的输出又存在不确定性。你问“为什么这次任务结果这么差”，传统日志只能告诉你每个 Agent 的入参出参，但你不知道整个链路的耗时分布和关键决策点。

OpenRig 的做法是引入 trace_id，从顶层任务创建时生成一个 ID，贯穿所有子任务和 Agent 调用。每个 Agent 在处理消息时记录处理耗时、token 消耗、输入摘要。这样在观测面板上就能看到“任务在 agent_c 上耗费了 80% 时间”或者“agent_b 的输入被截断导致后续分析失败”这类关键结论。我一直觉得，没有 trace 的多 Agent 系统就是盲人摸象，所以可观测性必须第一天就纳入架构，而不是等出了线上事故再补。

3. 从零搭建 OpenRig 协作系统的关键实操细节

3.1 基础架构：选型与部署形态

先明确技术栈选择。Rust 在并发安全和性能上有天然优势，核心调度器我确实用 Rust 写过一版，性能表现很好，但对于大多数团队来说，Python 生态对接 LLM 更方便。因此我建议的架构是：核心编排、消息总线、持久化模块用 Python 实现（FastAPI 提供接口），对性能要求极高的 Agent 用 Rust 单独写作独立服务。这样既控制开发成本，又能在必要时做热点替换。

部署形态上，我强烈建议把“编排器”和“Agent 执行器”分开部署。编排器是无状态服务，可以水平扩展；Agent 执行器是有状态服务，连接模型 API 和外部工具。两者之间用消息总线解耦，Agent 执行器处理完消息后回写事件。具体架构可以这样映射：

• 编排器（Orchestrator）：解析 DAG，派发任务，跟踪状态，至少两个实例做负载均衡
• 消息总线（Broker）：用 Redis Stream 或 Kafka，按 task_id 做分区，保证同一任务的消息有序
• 持久化存储：PostgreSQL 存事件日志和快照，Redis 存内存态（可丢失，重建用事件日志）
• Agent 执行器：按业务拆成独立服务，每个执行器消费对应主题的消息

3.2 编排配置示例与核心代码实现

这是整个项目最值得参考的部分。我放一个最简但完整的编排定义，基于 YAML：

workflow: content_generation_flow description: 从资料收集到报告生成的完整流程 nodes: - id: fetcher agent: news_fetcher input: { topic: "多智能体编排实践" } - id: parser agent: data_parser input: { source: "fetcher.output.text" } depends_on: [fetcher] - id: analyzer agent: analysis_agent input: { data: "parser.output.structured" } depends_on: [parser] - id: writer agent: report_writer input: { analysis: "analyzer.output.conclusion" } depends_on: [analyzer] - id: reviewer agent: quality_reviewer input: { report: "writer.output.draft" } depends_on: [writer] edges: - { from: fetcher, to: parser, on_success: proceed, on_failure: retry_max_3 } - { from: parser, to: analyzer, on_success: proceed, on_failure: fail_workflow } - { from: analyzer, to: writer, on_success: proceed } - { from: writer, to: reviewer, on_success: proceed }

调度器核心代码里，最关键的逻辑是“从事件日志恢复 DAG 执行状态”。我用 Python 写过一个简版，核心思路如下：

# scheduler.py（简化版） from dataclasses import dataclass, asdict from typing import Dict, List, Optional @dataclass class WorkflowState: workflow_id: str task_id: str node_states: Dict[str, str] # node_id -> pending/running/success/failed current_ready: List[str] class EventSourcedScheduler: def __init__(self, event_store, snapshot_store): self.event_store = event_store self.snapshot_store = snapshot_store def restore(self, workflow_id: str) -> WorkflowState: # 1. 加载最近快照 snapshot = self.snapshot_store.load(workflow_id) if snapshot: state = WorkflowState(**snapshot) else: state = self._init_state(workflow_id) # 2. 从快照位置向后重放事件 events = self.event_store.load_from(workflow_id, snapshot.seq if snapshot else 0) for event in events: self._apply(state, event) # 3. 重放后重新计算 ready 队列 state.current_ready = self._calc_ready(state) return state def _apply(self, state: WorkflowState, event: dict): node_id = event["node_id"] if event["type"] == "NODE_SUCCEEDED": state.node_states[node_id] = "success" elif event["type"] == "NODE_FAILED": state.node_states[node_id] = "failed" elif event["type"] == "NODE_STARTED": state.node_states[node_id] = "running" def _calc_ready(self, state: WorkflowState) -> List[str]: # 根据 DAG 依赖关系，找出所有前置节点都成功的 pending 节点 ready = [] for node_id, depends in self.dag_edges.items(): if state.node_states[node_id] == "pending": if all(state.node_states[a] == "success" for a in depends): ready.append(node_id) return ready

这段代码看似简单，但它是整个系统的“定海神针”。我遇到过几次宕机事故，都是靠跑这个恢复逻辑，几秒钟内就从最近快照继续跑了，业务无感知。事件日志里只有“已发生的事实”，调度器只是把事实投射回状态，这种设计让系统状态永远可以被解释，不会被脏写搞乱。

3.3 持久化层的设计与恢复逻辑

持久化层是另一个大头。事件日志表我会设计成：

CREATE TABLE event_log ( seq BIGSERIAL PRIMARY KEY, workflow_id TEXT NOT NULL, task_id TEXT NOT NULL, node_id TEXT NOT NULL, event_type TEXT NOT NULL, payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX idx_event_workflow_seq ON event_log (workflow_id, seq); CREATE TABLE snapshot ( workflow_id TEXT PRIMARY KEY, seq BIGINT NOT NULL, state_json JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL );

这里有个实践要点：业务数据不要只放在事件的 payload 里，还需要单独建一张结果表存 Agent 的最终产出物。否则事件日志巨大之后，重放成本很高。我的策略是：事件日志负责记录“进度”和“元数据”，业务大对象（如长文本报告）直接写对象存储，事件里只保留引用 ID。

恢复逻辑的顺序也不能乱：先恢复快照，再重放事件，重放完再恢复各 Agent 内部的缓冲状态（比如消息队列中未处理的请求）。整个流程我建议封装成一个recover_workflow(workflow_id)的方法，并配套自动化测试，每次变更调度逻辑都回归一遍。

3.4 并发与高负载参数调优

“AI Agent 怎么扛并发”是最近热门话题。真实瓶颈其实很少在模型 API，更多在事件写入、消息排队和调度器处理能力上。基于我的压测数据，有几条参数值得记录：

事件批量写入：不要每产生一个事件立刻 INSERT，而是攒批后批量写入 PostgreSQL。我用psycopg2的execute_values把 50 条事件一次性写入，吞吐能从每秒 200 提升到 1200。代价是事件可见性延迟约 200ms，对于多数协作场景完全可接受。

消息分区键：Redis Stream 消费时，用 task_id 作为 consumer group 的 partition key，确保同一个任务的消息始终被同一个执行器消费。否则两个 worker 同时处理同一任务的上下游消息，会产生竞态，前面提到的重放逻辑会因此不稳定。

调度器轮询频率：不要用“每 100ms 轮询所有任务”的粗暴方案，那样在任务数上百时会白白烧掉 CPU。建议只轮询state == ready的任务，并且配合 Redis 的阻塞弹出（XREAD BLOCK），让调度器空闲时挂起而不是空转。

限流与背压：每个 Agent 执行器的消费速率要兜底限流，否则一个 Agent 回复变慢，消息队列会疯狂堆积，最终拖垮整个总线。我通常给执行器设置最大并发数（比如 10），超过则暂停消费，靠消息队列自身的背压机制缓冲。

4. 生产环境踩坑记录与排查经验

这一节的内容，大部分是我在生产环境里用真金白银换来的经验，比第二节的原理更值得认真看。

4.1 编排死锁与循环依赖：别让 Agent 互相等

第一次上线时我遇到一个诡异现象：整个工作流卡住，没有任何节点在执行，日志也正常，就是不出结果。查了半天发现是 DAG 里存在一个环——A 等 B、B 等 C、C 等 A，三个 Agent 永远在互相等待。虽然我在设计文档里写了“DAG 必须无环”，但实际编排定义是人写的，难免有疏漏。

解决办法是两层防护：第一层，提交编排定义时做拓扑排序校验，如果检测到环直接拒绝发布；第二层，调度器启动时对每个工作流做依赖深度检测，执行超过 N 层依赖还没到达叶子节点就告警。另外，给每个节点加上超时熔断，即使出现死锁，也会在超时后被强制标记为失败，避免无限期占住资源。

4.2 重放不一致：事件顺序和幂等性的连锁反应

有一次线上系统崩溃后恢复，结果发现同一个 Agent 执行了两次同样的任务，产生了重复收费。排查下来问题出在：事件日志里节点的“完成”事件和“开始”事件顺序错乱，恢复时把已完成节点重新加入 ready 队列了。

这个坑的根因是写入事件时的顺序没有严格保证。同一个节点的事件落到不同的数据库事务里，恢复时自然乱序。我的修复方案是：

• 事件的 created_at 采用数据库统一时间，业务代码不传时间戳。
• 每个节点的事件写入必须发生在同一个数据库事务中，用事务的原子性保证“开始”必然先于“完成”。
• 消费端增加事件去重表，记录 node_id + last_event_type，重复事件直接跳过。

这之后再也没有出现过重复执行的问题。这里强烈建议把“幂等性”当成第一等公民来设计，特别是你的 Agent 有外部副作用（比如发邮件、扣费、调第三方接口）时。

4.3 状态膨胀与存储瓶颈

事件日志方案很好用，但它有一个绕不开的问题：时间久了日志量会非常大。我们压测数据是每条事件约 1KB，单日 10 万条就是 100MB，跑一个月就是 3GB。这还不算慢查询带来的性能问题。

我的处理方式是三层：

• 定期快照：每 1000 条事件做一次快照，恢复时从最近快照开始，历史事件在快照完成后有机会归档到冷存储（比如 S3/OSS）。
• 事件压缩：对已经快照的事件做轻量聚合，比如“节点完成”事件可以合并成“节点结果”一条记录。
• 只读归档：超过 30 天的历史事件迁移到归档表，线上只保留最近 30 天。审计场景需要旧数据时再从归档表查询。

4.4 常见问题速查表

排查这类问题，我建议给每个生产环境的 Agent 准备一份“标准操作手册”，记录它的输入格式、输出格式、依赖的第三方资源、常见失败模式。排障时优先看 trace 而不是漫天搜日志，先定位“哪一步没完成”，再定位“为什么没完成”。

最后再说一个我自己反复踩了多次才想明白的小技巧。多智能体系统里，不要让 Agent 自己决定下一步做什么，所有“下一步”都应当由编排层决定。Agent 只负责提交结果和异常，是否重试、是否走分支、是否终止，全部由编排逻辑控制。这样虽然牺牲了一些“智能感”，但换来了系统的确定性。尤其在生产环境，确定性比“偶尔聪明一下”重要得多。这套思路在 OpenRig 里贯穿始终——把智能留给模型，把秩序留给系统。