企业官网建设流程全解析

1. 从三五个 Agent 到协作网络：OpenRig 要解决的组织问题

先聊个我踩过的场景。前阵子我给一个内部知识库系统做智能化改造，先后拆出了六个 Agent：一个负责文档解析，一个负责语义检索，一个负责摘要生成，一个负责问答，还有一个负责把结果写成结构化报告。单独跑每一个都挺好，单个 Agent 的准确率和响应速度都能接受。可一旦要让它们协作完成一条完整任务链，问题就全出来了：Agent A 处理完的结果没人接、Agent B 需要上下文却拿不到、某一步崩了之后整个流程要从头再来，更别提进程一重启所有中间状态全部归零。

这就是典型的“离散 Agent”困境——每个智能体都是一座孤岛，能干活，但彼此之间没有组织关系。OpenRig 做的事情，就是把这一堆各自为战的 Agent 编成一个持久化的协作系统，让它们的通信、任务流转、状态保存、异常恢复都有章可循。这篇博文我会从编排原理、系统设计、Rust 落地实践、踩坑记录几个维度展开，把我在实际搭建这套框架过程中的经验和教训都倒出来。

先明确一个基本判断：OpenRig 不是要你去重写 Agent 的内部推理逻辑，它是在 Agent 之上加一个“组织层”。换句话说，你的 Agent 还干它原来的活，OpenRig 负责解决它们之间怎么找到对方、怎么传话、怎么交接任务、怎么忘记过去又怎么记住未来。这个定位非常重要，决定了你在引入它的时候不需要推倒重来，可以渐进式地把现有离散 Agent 一个一个接入到协作网络里。

我见过不少团队在讨论多智能体编排时，第一反应是“要不要用 LangGraph”“要不要上 AutoGen”这类现成框架。但实际项目里往往有一个很现实的问题：你的 Agent 可能是不同时期、不同语言、不同通信协议写出来的，强绑定某个特定框架反而会把自己锁死。OpenRig 的思路更偏向基础设施层——它只规定 Agent 之间怎么通信、状态怎么存、流程怎么编排，不规定你的 Agent 内部怎么写。这也是我选择它来做底层支撑的核心原因。

1.1 离散 Agent 的三个通病：状态丢失、上下文割裂、协作靠人肉

先说状态丢失。单个 Agent 运行时，它的对话历史、中间变量、处理进度都存在进程内存里，进程一挂什么都没了。如果你只是做一个简单的聊天机器人，这不算大事；但在多 Agent 协作场景里，一个任务往往要经过“解析 → 检索 → 推理 → 生成 → 校验”多个阶段，每个阶段都有中间产物。这些产物一旦丢失，整个任务链就要回退甚至重启，代价非常高。

再一个问题是上下文割裂。我用一个生活化类比来解释：就好比一个项目团队，每个人手里都有一份自己的笔记本，但笔记本之间不共享，A 记了什么 B 完全不知道。A 告诉 B“那个客户的需求我已经分析完了，你直接出方案吧”，但 B 手里没有 A 的分析结果，只能自己重新分析一遍，或者追问一堆细节。Agent 之间的协作如果不共享上下文，就会出现同样的尴尬——检索 Agent 明明已经找到了答案，生成 Agent 却拿不到检索结果，只能再调一次接口，既浪费 token 又降低响应速度。

第三个通病是协作靠人肉。没有编排层的情况下，多 Agent 之间的调用关系散落在业务代码里，一个业务流程串起了四个 Agent，你就得在业务代码里写四段调用逻辑、四段超时处理和四段异常分支。项目一迭代，新增一个 Agent，就得回头改业务流程代码。这种硬编码的协作方式，本质上是在把分布式系统的复杂度转嫁给应用层开发者——你本来只想做一个 AI 功能，结果先被 Agent 之间的通信搞到头大。

1.2 OpenRig 的定位：编排层而不是重写 Agent

OpenRig 解决上面三个问题的思路，可以概括成一句话：把 Agent 之间的交互从“点对点硬编码”变成“基于注册中心和消息总线的松耦合协作”。

具体来说，OpenRig 引入了三个核心组件：

这套设计的核心逻辑是：Agent 不再需要知道“谁在帮我干活”，它只需要把自己的需求发布到消息总线上，由编排引擎根据 Agent Registry 里的能力标签，找到合适的 Agent 来处理。这样一来，协作关系从代码里的硬编码变成了运行时的动态路由，新增一个 Agent 只需要注册一次，不用改任何业务流程代码。

关于这些组件的具体通信协议、状态存储格式和编排策略，我会在后面两节详细拆解。这里想强调一个容易被忽视的点：编排层最重要的价值不是“让 Agent 能通信”，而是“让 Agent 之间的通信变得可观测、可恢复、可审计”。你可以在 OpenRig 里看到每一条消息的流转路径、每一个任务的执行状态、每一次状态恢复的日志记录——这种确定性是离散 Agent 直接互调完全给不了的。

1.3 适用场景判断：什么情况该上编排框架，什么情况不需要

不是所有多 Agent 项目都需要 OpenRig 这种编排框架。在决定引入之前，值得先问问自己三个问题：

第一，你的 Agent 之间是否需要共享状态？如果每个 Agent 完全无状态、输入输出都是独立请求，那编排层的作用就很有限，一个简单的 HTTP 调用就能解决问题。第二，你的任务链路是否有复杂的交接和分支？如果只有一个 Agent 干活，或者即使有多个 Agent 但它们的调用顺序是固定不变的直线，那也未必需要引入编排框架，代码里写死顺序即可。第三，你的系统是否需要从故障中恢复？如果你能接受某个 Agent 挂了之后整个任务重新开始，那确实可以省掉状态持久化这套复杂度。

我的建议是：当你的 Agent 数量超过三个、它们之间存在数据依赖、且你对任务的可靠性有要求时，再考虑引入 OpenRig 这类编排层。用量化的标准衡量，就是：如果你的业务流程里出现了“Agent 的输出要作为另一个 Agent 的输入”这样的情况，而且这个链条超过两跳，那编排框架的收益就已经大于它的引入成本了。

2. 编织的关键机制：注册中心、消息总线与状态补全

这一节拆解 OpenRig 的三大核心机制。我会先讲设计动机，再讲具体的实现选择，最后用一段 Rust 代码做示例，方便你直接理解在代码层面它是怎么工作的。

2.1 Agent Registry：谁在场、能干什么、怎么找到它

Agent Registry 是一个服务注册中心，不是什么新鲜的概念——微服务架构里早就有 Eureka、Consul 这样的基础设施。但在多智能体场景里，Agent Registry 需要多做一些事情：它不仅要知道“有哪些 Agent 在线”，还要知道“每个 Agent 具备什么能力”“每个 Agent 当前的负载状况”“每个 Agent 支持什么样的消息格式”。

能力标签是 Registry 里最重要的元数据。比如一个检索 Agent，它的能力标签可能是semantic_search、vector_db_query；一个摘要 Agent，它的能力标签可能是text_summarization。当某个 Agent 在消息总线上发布一个需求“我需要做语义检索”，编排引擎会根据能力标签做匹配，找到合适的检索 Agent 并把消息路由过去。

具体实现上，Agent 启动时需要向 Registry 执行注册，上报自己的地址、能力标签、输入输出格式 schema。Registry 会维护心跳机制，如果某个 Agent 超过一定时间没有上报心跳，就把它标记为离线并从路由候选集中移除。这个设计和微服务的服务发现几乎一样，但它多了一个关键点：能力匹配不是简单的字符串相等，而是基于 schema 兼容性判断。例如，某个 Agent 声称自己接受text/plain格式输入，那它就能处理发布者用text/markdown发出的请求吗？这取决于编排引擎里配的兼容规则。

我在初始版本里做得很简单，就是用字符串精确匹配。结果很快发现一个问题：团队里不同 Agent 的作者对同一个能力起了不同的名字，有的叫search_document，有的叫doc_search，Registry 根本匹配不上。后来改成了带同义词映射的标签体系，才基本解决。这个细节提醒我，编排层不仅要解决技术问题，还要解决组织问题——能力标签的规范化需要从一开始就定好规则。

2.2 消息总线：Topic 路由和请求-响应的取舍

消息总线是 Agent 之间通信的通道。OpenRig 实现了两种通信模式，适用场景完全不同：

模式一：请求-响应（Request-Reply）。适用于一个有明确调用者和被调用者的场景，比如编排引擎让摘要 Agent 对某段文本做摘要，调用方发出请求后等待响应。这个模式下消息是点对点的，需要包含消息 ID、请求者地址、超时时间等元数据。实现起来相对简单，语义也直观。

模式二：发布-订阅（Publish-Subscribe）。适用于一个 Agent 产生了一个事件、多个 Agent 都可能关心的场景。比如“文档解析完成”这个事件，检索 Agent 关心它，因为需要把解析结果写入向量库；报告生成 Agent 也关心它，因为要拿解析结果去做后续处理。发布者不需要知道有哪些订阅者，只需要往 Topic 上发布消息即可。

两种模式在 OpenRig 里是通过消息头里的路由策略字段区分的，如果routing字段是direct，消息总线路由到指定 Agent；如果是topic，则根据 Topic 匹配所有订阅者。这种设计在工程上非常实用，因为一条任务链里，有些环节是明确的调用关系（谁需要谁处理），有些环节是事件驱动的订阅关系（谁关心谁接收），用同一套消息总线统一处理，比混用两套通信系统要清爽得多。

关于 Topic 命名规范，我建议采用层级结构，比如task.parse.completed、task.retrieve.results、task.summarize.failed。层级化的好处是可以用通配符做订阅——Agent 可以只订阅task.parse.*，就能收到所有解析相关的事件，不用关心具体是哪一步产生的。

消息总线还有一个必须考虑的点：消息可靠性和顺序性。Rust 生态里最常用的消息中间件是 Apache Kafka 和 NATS JetStream，它们都支持持久化消息和消费者组的负载均衡。如果要处理的消息量不大，也可以用内存队列加 Postgres 兜底——把消息先持久化到数据库，再由后台任务分发，简单可靠，适合小团队快速上线。

2.3 状态持久化的三层设计：会话快照、事件日志、外部存储

前面提到，状态丢失是离散 Agent 协作最致命的通病之一。OpenRig 的 State Store 用了三层设计来解决这个问题。

第一层：会话快照（Session Snapshot）。每个 Agent 在处理任务的过程中，会把关键状态定期保存为快照。快照包括：当前的上下文窗口、已经完成的工作产物、待处理的队列、以及一些自定义的 Agent 状态字段。快照机制解决的是“Agent 崩溃后恢复到哪一步”的问题——只要最近一次快照成功，崩溃恢复后就能从快照点继续执行，而不是从头再来。

第二层：事件日志（Event Log）。快照是周期性保存的，两次快照之间的操作记录会丢失。为了弥补这个空缺，OpenRig 把 Agent 过程中的每个重要操作都记录为事件，追加写入日志。恢复时先加载最近快照，再回放快照之后的事件日志，就能恢复到崩溃前的精确状态。这种做法借鉴了事件溯源（Event Sourcing）的核心思想：状态是事件流的投影，把事件流保存下来，任何时刻的状态都可以重建。

第三层：外部业务存储。有些状态不属于 Agent 本身，而属于业务系统——比如某个任务关联的订单信息、用户信息、产品资料。这些数据不应该塞进 Agent 的会话快照里，而是应该存放在业务数据库，Agent 只在需要时通过外部接口访问。把这三层分开，是为了遵守单一职责原则：快照负责 Agent 内部状态、事件日志负责可追溯性、业务存储负责业务数据，各司其职，互不干扰。

这三层设计的取舍在于：快照和事件日志都要消耗存储空间，尤其是事件日志，随着运行时间增长会越来越大。我的做法是定期做“日志压缩”——合并相同状态字段的冗余事件，只保留最终状态变化，类似 Kafka 的 log compaction。还有一个值得注意的点：快照和事件日志必须做原子性处理，不能让两者处于不一致状态，否则恢复出来的 Agent 状态就是错乱的。我在实际项目里采用了两阶段提交的方案：先写事件日志并标记为“已持久化”，再写快照并标记为“已包含到事件序号 N”，恢复时取快照里记录的最后一个事件序号作为回放起点，保证一致性。

3. 编排引擎的工作方式：从 DAG 到自适应协商

注册中心解决了 Agent 之间如何找到对方，消息总线解决了如何通信，状态存储解决了如何记住过去。接下来是编排引擎——它决定 Agent 之间的任务流转逻辑。这是整个系统里最灵活、也最容易设计过度的地方。

3.1 静态 DAG 编排与动态协商的适用边界

市面上主流的多 Agent 编排框架通常都支持两种任务编排方式，一种是静态 DAG（有向无环图），一种是动态协商。两者各有利弊，我的实践结论是：能用静态 DAG 就不用动态协商，动态协商只留给真正需要灵活决策的场景。

静态 DAG 的思路是：在任务开始前，把整个协作流程定义成一张图，每个节点是一个 Agent，每条边是一次消息传递，边上的条件表达式决定是否走到下一个节点。这个方案的优点是确定性极强——你可以预知任务流转的每一步，出问题时能精确定位到哪个环节，测试也容易写。缺点是不够灵活，如果业务流程本身就复杂多变，静态 DAG 会让流程图变得异常庞大且难以维护。

动态协商的思路是：不预定义流程，而是让 Agent 在运行时根据当前状态决定下一步找谁、做什么。典型实现是“黑板系统”——多个 Agent 共享一块工作区域，各自声明“我擅长处理什么”，看到适合自己的任务就接手处理，处理完把结果写回黑板。这个方案非常灵活，适合开放性任务，但代价是行为不可预测——同一个任务这次走这条路、下次可能走另一条路，调试和复现都是噩梦。

我在 OpenRig 里采取的是折中策略：默认用静态 DAG 定义主干流程，在 DAG 的某些节点上允许挂载“动态协商分支”。比如在一个文档问答任务里，主干流程是“解析 → 检索 → 生成”，但如果检索 Agent 返回的结果质量评分低于阈值，就会触发协商分支——由几个候选的辅助 Agent 竞争处理这个低质量结果，谁的处理方案评分高就用谁的结果。用这种方式，既保留了 DAG 的稳定性，又给了系统在局部环节上的灵活性。

3.2 会话状态机如何驱动 Agent 交接

多个 Agent 协作过程中，最麻烦的其实不是干活，而是“交接”。所谓交接，就是 Agent A 完成自己的部分后，把任务连同上下文一起完整地递给 Agent B，B 基于 A 的结果继续处理。交接如果做不好，最常见的问题就是上下文信息丢失和重复劳动。

OpenRig 用会话状态机来管理交接过程。每个任务对应一个会话 Session，会话的状态包括pending、in_progress、waiting_dependency、completed、failed、compensating等。当一个 Agent 完成任务中的一步，会话状态会从in_progress变为waiting_dependency，等待依赖的下一步 Agent 就绪，然后状态再变为in_progress并转移给下一个 Agent。

每次交接都会打包一份“交接单”，里面包含三个东西：上一环节的输出产物、当前任务的完整上下文摘要、以及下一步需要注意的约束条件。为什么需要上下文摘要而不是完整上下文？因为完整上下文可能非常大，如果每个交接都带着全部对话历史和中间结果传递，通信开销和存储开销都会失控。摘要则可以用 LLM 生成，也可以由 Agent 自行产生，关键是抓取对下一步最核心的信息。

这个设计的妙处在于：它把“协作上下文”从一个隐性的、靠 Agent 记忆维持的东西，变成了显性的、可序列化可传递的结构化数据。即使中间某个 Agent 需要重新调度，新接手的 Agent 也可以依靠交接单快速恢复上下文，不用从零开始理解任务。

3.3 失败处理：超时、重试、降级、死信

多 Agent 协作系统里，失败是常态而不是异常。我见过太多人在设计时只考虑“理想路径”，结果线上有人工介入救火的场景多到怀疑人生。OpenRig 里我把失败处理分成了四个层级：

第一层，超时处理。每个任务节点都配置超时时间，超过时间未收到响应，编排引擎先把任务标记为超时状态，再按配置决定重试还是降级。超时时间的配置有一个矛盾：设短了容易误杀慢任务，设长了拖慢后续流程。我的经验是一个粗略基线：内部 Agent 之间的调用设 30 秒，涉及外部 API 调用设 60 秒，涉及多轮 LLM 推理的环节设 120 秒，然后根据实测结果逐步调整。

第二层，重试策略。重试要区分错误类型：可重试错误（网络抖动、临时资源不足）和不可重试错误（参数错误、数据格式不合法）。可重试错误用指数退避加抖动的方式重试，最多 3 次；不可重试错误直接标记失败，进入补偿流程。特别提醒一点：重试必须保证消息的幂等性，同一个任务不能因为重试被处理两次，否则可能出现重复扣费、重复写库这类事故。

第三层，降级方案。当某个 Agent 持续失败时，编排引擎可以启用降级——换一个能力相近的替代 Agent，或者直接跳过这个环节，或者用规则引擎加启发式方法给出一个“够用但不完美”的结果。降级方案一定要在设计阶段就定好，而不是等故障发生时再想，因为故障发生时你已经没有足够时间去设计方案了。

第四层，死信队列。所有重试和降级都无法处理的任务，统一进入死信队列，等待人工介入。死信队列里的任务必须保留完整的上下文信息和失败日志，这样人工处理时才能知道这个任务经历了什么、为什么失败。我经常说：一个设计良好的死信队列，等于给系统装了一个“安全阀”——最坏情况下，任务不会凭空消失，而是停留在一个可以追溯和干预的地方。

4. Rust 落地实践：搭一个可持久化的最小协作系统

前面讲了理论框架，这一节给一套最小可运行的系统。我直接在 OpenRig 的基础上选 Rust 技术栈做实现，包含 Agent 注册、消息路由和状态持久化三个核心模块。

4.1 技术选型栈说明

为什么选 Rust？两个原因。第一，Agent 协作系统本质是一个并发密集型任务系统，Rust 的 async/await 和 actor 模型生态非常适合表达“大量 Agent 同时在线、互相通信”这类场景。第二，Rust 在内存安全上的保证，能让我在处理大并发时不担心数据竞争和悬垂引用的问题——这在状态持久化场景里尤其重要，因为涉及多处共享状态的读写。

具体选型：

这套组合的特点是：每个组件都足够简单可靠，没有引入复杂的分布式协调组件（比如 etcd 或 ZooKeeper），适合中小规模的 Agent 协作场景。如果 Agent 数量巨大，可以再引入更多的横向扩展组件，但起步阶段不要过度设计。

4.2 最小代码实现：Agent 定义、消息路由、状态持久化

先定义 Agent 的核心 trait。每个 Agent 只需要实现handle_message方法和snapshot_state方法，前者处理消息，后者保存状态快照。

use async_trait::async_trait; use serde::{Deserialize, Serialize}; use std::collections::HashMap; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AgentMessage { pub id: String, pub sender: String, pub receiver: String, pub msg_type: String, pub payload: serde_json::Value, pub routing: RoutingMode, pub timestamp: i64, } #[derive(Debug, Clone, Serialize, Deserialize)] pub enum RoutingMode { Direct, Topic(String), } #[async_trait] pub trait Agent: Send + Sync { fn name(&self) -> &str; fn capabilities(&self) -> Vec<String>; async fn handle_message(&self, msg: AgentMessage) -> Result<AgentMessage, AgentError>; async fn snapshot_state(&self) -> Result<serde_json::Value, AgentError> { Ok(serde_json::json!({})) } async fn restore_state(&self, _snapshot: serde_json::Value) -> Result<(), AgentError> { Ok(()) } }

这个 trait 只有三个核心方法：name告诉 Registry 我是谁，capabilities告诉 Registry 我能干什么，handle_message是 Agent 的核心处理逻辑，接收一条消息、返回一条响应。快照和恢复作为默认实现提供，Agent 按需覆盖。

再实现一个简单的 Agent 注册中心。这里用内存 HashMap 存储所有注册信息，生产环境可以换成 redis 或 etcd，但最小实现里 HashMap 足够演示：

#[derive(Debug, Clone, Default)] pub struct AgentRegistry { agents: Arc<RwLock<HashMap<String, RegistryEntry>>>, } #[derive(Debug, Clone)] pub struct RegistryEntry { pub name: String, pub capabilities: Vec<String>, pub addr: String, pub last_heartbeat: i64, pub status: AgentStatus, } impl AgentRegistry { pub fn new() -> Self { Self { agents: Arc::new(RwLock::new(HashMap::new())) } } pub async fn register(&self, entry: RegistryEntry) { let mut agents = self.agents.write().await; agents.insert(entry.name.clone(), entry); } pub async fn find_by_capability(&self, capability: &str) -> Option<RegistryEntry> { let agents = self.agents.read().await; agents .values() .find(|e| e.status == AgentStatus::Online && e.capabilities.contains(&capability.to_string())) .cloned() } pub async fn heartbeat(&self, name: &str) { let mut agents = self.agents.write().await; if let Some(entry) = agents.get_mut(name) { entry.last_heartbeat = chrono::Utc::now().timestamp(); entry.status = AgentStatus::Online; } } }

接下来是消息总线的简化实现。OpenRig 的消息总线基于 NATS JetStream，核心逻辑是订阅和发布：

pub struct MessageBus { nc: nats::asynk::Connection, js: nats::asynk::jetstream::Context, } impl MessageBus { pub async fn connect(url: &str) -> Result<Self, Box<dyn std::error::Error>> { let nc = nats::asynk::connect(url).await?; let js = nats::asynk::jetstream::new(nc.clone()); Ok(Self { nc, js }) } pub async fn publish(&self, subject: &str, msg: &AgentMessage) -> Result<(), Box<dyn std::error::Error>> { let bytes = serde_json::to_vec(msg)?; self.js.publish(subject, bytes.into()).await?; Ok(()) } pub async fn request_reply( &self, subject: &str, msg: &AgentMessage, timeout: Duration, ) -> Result<AgentMessage, Box<dyn std::error::Error>> { let bytes = serde_json::to_vec(msg)?; let reply = self.nc.request(subject, bytes.into()).await?; let reply_msg: AgentMessage = serde_json::from_slice(&reply.data)?; Ok(reply_msg) } }

这个实现很直观：publish是发布-订阅模式，request_reply是请求-响应模式。底层依赖 NATS 的消息路由和持久化能力。

最后是状态持久化的核心实现，用 PostgreSQL 存储会话快照和事件日志：

pub struct StateStore { pool: sqlx::PgPool, } impl StateStore { pub async fn new(database_url: &str) -> Result<Self, sqlx::Error> { let pool = sqlx::PgPool::connect(database_url).await?; Ok(Self { pool }) } pub async fn save_snapshot( &self, session_id: &str, agent_name: &str, snapshot: serde_json::Value, ) -> Result<(), sqlx::Error> { sqlx::query( "INSERT INTO agent_snapshots (session_id, agent_name, snapshot, created_at) VALUES ($1, $2, $3, NOW()) ON CONFLICT (session_id, agent_name) DO UPDATE SET snapshot = $3, created_at = NOW()", ) .bind(session_id) .bind(agent_name) .bind(snapshot) .execute(&self.pool) .await?; Ok(()) } pub async fn append_event( &self, session_id: &str, event: serde_json::Value, ) -> Result<(), sqlx::Error> { sqlx::query( "INSERT INTO session_events (session_id, event, created_at) VALUES ($1, $2, NOW())", ) .bind(session_id) .bind(event) .execute(&self.pool) .await?; Ok(()) } }

快照的保存策略是 upsert 语义：同一个 session 和同一个 agent 的快照会被新快照覆盖，避免重复存储。事件日志则是 append-only 追加，每条事件都保留了发生的时间戳，方便后续按时间顺序回放。

4.3 跑起来：验证多 Agent 协作流转

代码写完后，搭一个三个 Agent 的最小场景来验证协作流转：一个ParseAgent负责把原始文本解析成结构化文档，一个RetrieveAgent负责根据查询在文档库里做语义检索，一个SummarizeAgent负责对检索结果生成摘要。

我先手动注册三个 Agent，然后在业务代码里用编排引擎发起一个任务：

#[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { // 初始化组件 let registry = AgentRegistry::new(); let bus = MessageBus::connect("nats://localhost:4222").await?; let store = StateStore::new("postgres://user:pass@localhost/agentdb").await?; // 注册 Agent registry.register(RegistryEntry { name: "parse_agent".into(), capabilities: vec!["document_parse".into()], addr: "nats://localhost:4222".into(), last_heartbeat: chrono::Utc::now().timestamp(), status: AgentStatus::Online, }).await; registry.register(RegistryEntry { name: "retrieve_agent".into(), capabilities: vec!["semantic_search".into()], addr: "nats://localhost:4222".into(), last_heartbeat: chrono::Utc::now().timestamp(), status: AgentStatus::Online, }).await; registry.register(RegistryEntry { name: "summarize_agent".into(), capabilities: vec!["text_summarization".into()], addr: "nats://localhost:4222".into(), last_heartbeat: chrono::Utc::now().timestamp(), status: AgentStatus::Online, }).await; // 发起任务：解析文档 let parse_msg = AgentMessage { id: uuid::Uuid::new_v4().to_string(), sender: "orchestrator".into(), receiver: "parse_agent".into(), msg_type: "parse_document".into(), routing: RoutingMode::Direct, payload: serde_json::json!({ "content": "..." }), timestamp: chrono::Utc::now().timestamp(), }; let parse_result = bus.request_reply("agent.parse", &parse_msg, Duration::from_secs(30)).await?; // 保存快照 store.save_snapshot("session-001", "parse_agent", parse_result.payload.clone()).await?; // 第二步：语义检索 let retrieve_msg = AgentMessage { id: uuid::Uuid::new_v4().to_string(), sender: "orchestrator".into(), receiver: "retrieve_agent".into(), msg_type: "semantic_search".into(), routing: RoutingMode::Direct, payload: serde_json::json!({ "query": "如何配置 OpenRig 状态存储", "documents": parse_result.payload }), timestamp: chrono::Utc::now().timestamp(), }; let retrieve_result = bus.request_reply("agent.retrieve", &retrieve_msg, Duration::from_secs(60)).await?; // 第三步：生成摘要 let summarize_msg = AgentMessage { id: uuid::Uuid::new_v4().to_string(), sender: "orchestrator".into(), receiver: "summarize_agent".into(), msg_type: "summarize".into(), routing: RoutingMode::Direct, payload: retrieve_result.payload, timestamp: chrono::Utc::now().timestamp(), }; let summarize_result = bus.request_reply("agent.summarize", &summarize_msg, Duration::from_secs(120)).await?; println!("最终摘要: {}", summarize_result.payload); Ok(()) }

这个示例里，三个阶段用的是串行的request_reply调用，在业务代码里可以清晰地看到每一步的输入输出。任务执行过程中，每个 Agent 的关键状态都通过store.save_snapshot持久化了，模拟进程崩溃后，只要从快照里恢复 ParseAgent 的输出，就能跳过解析阶段直接进入检索阶段。

当然这只是一个最小演示。真实项目中，编排引擎不会像这样在业务代码里一串到底，而是会基于 DAG 定义自动执行任务流转。下面的进阶设计部分我会说明如何把这段串行逻辑改造成 DAG 驱动的自动编排。

5. 踩坑实录：我在编排过程中遇到的四个典型问题

理论和示例归理论与示例，真正把 OpenRig 用到生产环境，才知道坑有多深。这一节记录我在实际部署和调优过程中遇到的四个典型问题，每一个都是真实场景，每一个都花了不止一天才排查清楚。

5.1 序列化灾难：Agent 输出总是丢字段

最初版本的 Agent 之间用自定义二进制格式传递消息，结果调试的时候经常发现：A Agent 发送的消息，B Agent 反序列化之后出现字段缺失，而且报错信息非常隐晦——不是说你少传了字段，而是说“找不到对应方法”，让你一度以为是对端 Agent 类型定义不一致。

后来排查发现根因出在版本兼容性上：A Agent 更新了消息定义，加了一个priority字段，但 B Agent 使用的旧版反序列化库直接忽略了未知字段；等 B Agent 也更新之后，C Agent 又停下了。多条链路上的 Agent 版本不一致，导致字段在传递过程中时有时无，丢得神不知鬼不觉。

这个问题给我三个教训。第一，Agent 之间的通信协议必须有明确的版本号，消息头里带schema_version字段，接收方校验版本号，不匹配就直接报错而不是静默忽略。第二，建议直接用 JSON 或 MessagePack 这类自带 schema 容忍的格式，至少缺少字段时能给出清晰的报错信息。第三，建立一套消息契约测试，每个 Agent 在 CI 里可以发起一条完整的协作链路，确保最新代码和其他所有 Agent 的兼容性。自从加了版本协议和契约测试，序列化类问题基本绝迹。

5.2 死锁与饥饿：两个 Agent 互相等待

多 Agent 协作系统里，死锁是非常容易出现的并发问题。我遇到过的最典型场景是：两个 Agent 在处理任务时需要互相调用对方的结果。Agent A 需要 Agent B 生成一个中间分析，Agent B 需要 Agent A 提供一份数据预处理结果，两边等不到对方的结果就一直阻塞，直到双双超时失败。

为什么会出现这种循环依赖？因为我在设计 DAG 时没有做依赖图的环检测，任务编排引擎直接按照注册的调用关系执行了。排查过程其实不难：我把所有 Agent 的等待关系画成一张有向图，一眼就看到 A ← B 和 B ← A 两条边形成了一个环。

修复方案有两个角度。一是从设计上避免循环依赖：在 DAG 定义阶段就做拓扑排序检查，存在环就拒绝启动任务。二是从运行上增加超时和取消机制：给每个 Agent 调用设置强制超时，超时后主动释放资源并上报编排引擎，由编排引擎做回退处理。现在的 OpenRig 里，我在 DAG 编辑器里就嵌入了环检测逻辑，同时给所有底层调用加了超时兜底——双保险。

另外还有一个容易忽略的饥饿问题：如果某个 Agent 因为优先级低、一直被其他任务占用，导致它的任务迟迟得不到处理，本质上是一种“活锁”。解法是在编排引擎里增加公平调度的机制——按等待时间动态提升任务优先级，确保长时间等待的任务最终能被处理。

5.3 状态漂移：快照恢复后 Agent 记忆错乱

状态恢复机制上线后，出现了一个诡异的问题：Agent 从快照恢复后，有时候会出现“记忆错乱”——它明明应该记得任务做到哪一步，却表现出对上下文陌生或者执行了重复的操作。

排查了很久，最后定位到问题根源在快照和事件日志的同步性上。我之前的设计是先保存事件日志，再周期性保存快照。理论上恢复时会从快照里记录的“最后事件序号”开始回放。但问题是，save_snapshot和append_event这两个操作不是原子的，中间可能发生进程崩溃——快照保存成功了，但“快照对应的事件序号”没有正确写入，导致恢复时误以为快照已经包含全部事件，跳过了一部分日志的回放。

解决方式是我在前面提到过的两阶段提交：先把事件的持久化作完，再更新快照的元数据，用数据库事务把这两个操作包在一起。这样快照总是对应一个明确的事件偏移量，恢复时就能精确地知道该从哪里开始回放。经过这个修复，状态漂移问题就不再出现了。

这个教训非常典型，本质上是分布式系统中的“原子性”问题。你永远不要把两个有关联的状态变更当成两个独立操作来做，一定要通过事务或两阶段提交保证它们要么全部成功、要么全部失败。

5.4 拓扑风暴：不可靠网络下的消息放大

另一个让我印象深刻的问题是“拓扑风暴”。起因是有一次网络抖动，某个 Agent 发送的任务请求超时了，编排引擎按规则做了重试。但那个 Agent 的请求本身是慢处理类型的——它已经收到了请求，真的在处理，只是处理得比较慢。重试请求到达后，这个 Agent 又开了一个新的处理实例。两个实例同时处理同一个任务，都产生了结果，导致下游收到了重复数据。

这还不是最严重的。因为下游 Agent 处理重复数据时触发了异常，它又对这个任务发起了重试请求，形成了级联放大的效果。最后的结果是：一个小小的网络抖动，引发了几十倍的消息风暴，把消息总线和数据库都打满了。

这个问题对应的核心设计原则就是幂等性。消息总线上传入的每一条消息都必须携带一个全局唯一的message_id，接收方在处理前先检查这个 ID 是否已经处理过，如果处理过就直接返回上次的结果，不再重复执行。我在消息总线层实现了幂等机制，用一个 Redis 或者数据库表记录所有已处理的消息 ID，这块的开销比较小，但带来的稳定性收益是巨大的。

如果你在设计阶段就考虑消息的幂等性，这个坑是可以完全避免的。但大多数系统都是在出事故后才想起来补这个机制——我也是踩了一次线上故障才有深刻体会。

6. 经验沉淀：让协作系统跑得更稳的进阶建议

如果把最后一节的坑比作“开刀手术”，那接下来的建议就是“健身养生”——让系统从一开始就长得健康，降低出问题的概率。这些建议来自于我对 OpenRig 的使用和维护经验，不一定每个项目都适用，但值得作为设计阶段的参考答案。

6.1 添加可观测性：追踪、指标、日志

多 Agent 协作系统最可怕的一点是“黑盒”——消息在哪个环节丢了、任务卡在哪个 Agent 上了、哪个 Agent 的内存占用异常，如果你没有可观测性设计，排查这些问题就像在黑暗里找钥匙。

我强烈建议在系统里三个层面都加上可观测性：

消息追踪（Tracing）。给每个会话分配一个全局唯一的trace_id，消息流转到任何 Agent 时都把trace_id透传下去。配合 OpenTelemetry 或类似的分布式追踪系统，你可以在一条时间线上完整地看到任务经历的所有 Agent、每次消息的耗时、每一步的处理结果。没有这个，你只能在各个 Agent 的日志里手工比对message_id，效率极低。

运行指标（Metrics）。至少采集这几类指标：每个 Agent 的消息处理速率、平均处理耗时、队列积压量、失败率、重试次数、状态快照大小。指标配合告警规则——比如队列积压超过阈值就告警——可以让你在系统真正出问题之前提前介入。

结构化日志（Logging）。所有 Agent 的日志全部结构化输出，统一格式里包含时间戳、Agent 名、trace_id、日志级别、消息摘要。我在项目里统一用 JSON 格式输出日志，这样可以直接接入 ELK 或 Loki 做集中检索。

这三件事要在系统搭建初期就做，而不是等出了问题再补。因为如果一开始没有在代码路径里埋点，后面再补救等于在所有 Agent 的代码里翻腾一遍，成本非常高。

6.2 粒度的平衡：Agent 职责怎么切分才算合理

Agent 的职责粒度，是一个没有标准答案但极其影响系统复杂度的设计决策。粒度太粗，一个 Agent 什么活都干，出问题时难以定位；粒度太细，Agent 数量爆炸，编排成本高于收益。

我自己思考这个问题用的标准是：当两个 Agent 之间的依赖关系变成“闭环”时，就应该考虑合并。什么意思呢？如果 Agent A 的结果几乎总是被 Agent B 使用，而 B 的结果又可以反过来帮助 A 优化，那它们其实是同一个工作单元的两个阶段，合并可能更合适。反过来，如果两个 Agent 的依赖关系很少、它们各自服务的业务目标差异明显，那就保持独立。

另一个实用的判断标准是看 Agent 的复用频率。一个被多个不同任务复用的功能单元，应该独立成 Agent；一个只被单一任务使用的功能单元，保留在业务流程里可能更简单。组织好 Agent 的边界，比在代码层面优化任何一个 Agent 的实现都要重要，因为协作系统的复杂度来自于连接数，而不是节点数。

6.3 从 OpenRig 出发还能扩展什么

最后聊聊扩展方向。OpenRig 这套协作框架的价值在于它把“低层次的通信和状态管理”抽离出去了，那么你可以在更高的层次上做更多事情。

一个自然的扩展是智能调度。目前的编排引擎还基于预定义的 DAG 和规则，未来可以考虑引入基于强化学习的调度策略——根据任务的难度、Agent 的历史表现、当前负载状况，让系统自动决定任务分配给哪个 Agent、用什么样的并行度执行。这个方向叫做“自适应编排”，现在很多研究机构都在探索，但工程落地还很不成熟，当前阶段我更建议在规则和启发式方法上做文章，等基础设施稳定了再逐步智能化。

另一个扩展方向是 Agent 的自我进化。既然每个 Agent 的每次运行都被记录成事件日志，你实际上拥有了一座“行为数据库”。你可以分析哪些 Agent 经常失败、哪些组合效果优于其他组合、哪些环节最耗时，然后基于这些数据对 Agent 的编排策略做持续优化。我设想中的 OpenRig 演进方向，就是让协作系统本身成为一个可以学习和进化的组织，而不仅仅是任务的执行管道。

回到最初的问题：为什么要把离散的 AI Agent 编织成持久化的协作系统？因为 Agent 的数量增多之后，组织的价值就会超越个体的价值。OpenRig 解决的不是“单个 Agent 变聪明”的问题，而是“多个 Agent 像一个团队一样干活”的问题。这套框架的最核心收益，用一句话概括就是：它把 Agent 之间的协作变得可设计、可运维、可恢复、可优化。如果你手头正有多个离散 Agent 需要打通，不妨按这套思路去设计——先别急着写流程代码，先想清楚注册、通信、状态、编排和失败处理这五件事，大概率能避开不少我在实践中踩过的坑。