多Agent协作系统落地：基于OpenRig的编排与状态恢复实践

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 的定位编排层而不是重写 AgentOpenRig 解决上面三个问题的思路可以概括成一句话把 Agent 之间的交互从“点对点硬编码”变成“基于注册中心和消息总线的松耦合协作”。具体来说OpenRig 引入了三个核心组件组件作用解决的通病Agent Registry维护所有在线 Agent 的地址、能力标签、状态信息协作靠人肉找不到对方Message Bus负责 Agent 之间的消息路由支持请求-响应和发布-订阅两种模式上下文割裂通信混乱State Store持久化每个 Agent 的会话状态、任务进度、事件日志状态丢失无法恢复这套设计的核心逻辑是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_searchRegistry 根本匹配不上。后来改成了带同义词映射的标签体系才基本解决。这个细节提醒我编排层不仅要解决技术问题还要解决组织问题——能力标签的规范化需要从一开始就定好规则。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 BB 基于 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 在内存安全上的保证能让我在处理大并发时不担心数据竞争和悬垂引用的问题——这在状态持久化场景里尤其重要因为涉及多处共享状态的读写。具体选型模块选型说明异步运行时tokio主流选择生态完善文档多消息通信NATS JetStream支持持久化和流式消费单机部署简单数据库PostgreSQL sqlx存会话快照和事件日志sqlx 能在编译期检查 SQL序列化serde JSON初始版本用 JSON灵活直观后续可以换 MessagePackAgent SDKasync-trait用 trait 抽象 Agent 行为方便接入不同类型 Agent这套组合的特点是每个组件都足够简单可靠没有引入复杂的分布式协调组件比如 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) - VecString; async fn handle_message(self, msg: AgentMessage) - ResultAgentMessage, AgentError; async fn snapshot_state(self) - Resultserde_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: ArcRwLockHashMapString, RegistryEntry, } #[derive(Debug, Clone)] pub struct RegistryEntry { pub name: String, pub capabilities: VecString, 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) - OptionRegistryEntry { 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) - ResultSelf, Boxdyn 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(), Boxdyn 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, ) - ResultAgentMessage, Boxdyn 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) - ResultSelf, 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(), Boxdyn std::error::Error { // 初始化组件 let registry AgentRegistry::new(); let bus MessageBus::connect(nats://localhost:4222).await?; let store StateStore::new(postgres://user:passlocalhost/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 需要打通不妨按这套思路去设计——先别急着写流程代码先想清楚注册、通信、状态、编排和失败处理这五件事大概率能避开不少我在实践中踩过的坑。