今年我在团队里主要折腾一件事把零散的AI Agent落成能扛事儿的系统。项目代号OpenRig目标很简单——做一套多智能体编排实践把那些互相不认识的离散AI Agent织成一张持久化协作系统的网。这篇文章不是产品发布会是我整理出来的设计思路、落地流程和排障记录。做AI应用落地的人尤其是被多Agent协作、任务恢复、中间状态丢失坑过的人应该能从里面拿走一些东西。先说结论单个Agent的能力再强解决不了“集成”的问题。OpenRig真正解决的是通信、状态、记忆这三件事。通信解决Agent之间怎么喊话状态解决任务跑到哪一步记忆解决Agent换实例之后怎么想起之前聊到哪。下面我会从设计动机、核心模块、完整实操和问题排查四个方面把整个实践过程摊开讲。1. 为什么需要把一堆Agent“织”起来1.1 单Agent在真实业务里撑不过三个来回我们最开始也是老老实实跑单Agent。流程不复杂过一个客服工单先查订单再算运费最后生成退款单。demo阶段一切正常模型调用一次LLM工具调用几次API看起来挺聪明。但一放到真实业务里问题马上暴露。第一是上下文撑不住。对话超过十轮早先用户填的地址、商品ID、优惠券信息就开始模糊。模型不是不记得是它的“注意力”被后面的内容稀释了。你让它在一次会话里又检索又计算又填表越往后越容易丢关键字段。第二是会话太脆弱。一次普通网络抖动LLM接口超时整个任务直接失败。客户端重试模型重新开始推理但前面的工具调用结果已经丢了。用户得从头再描述一遍问题这在内部流程自动化里几乎是不可接受的。第三是没法拆人。很多任务天然需要不同角色协作一个Agent做数据清洗一个Agent写报告一个Agent做合规检查。如果全塞在一个Prompt里Prompt会膨胀到几千行改一个角色逻辑就要重新调整个系统。所以我们后来把每个能力拆成独立Agent。拆完之后更痛苦调研Agent做完了写作Agent不知道写作Agent写完了审核Agent拿不到结果。各自跑得欢整体是一盘散沙。这也是“离散AI Agent”这个词的由来——每个节点是好的连不成线。1.2 编排的隐藏底座不只是“喊人干活”我一开始理解的编排就是让A调用BB调用C。于是写了各种内部HTTP接口A处理完直接POST给B。这在两个Agent的时候挺顺三个Agent也凑合五个Agent以上就开始失控。失控的原因在于直接调用只解决了“数据传递”没解决“契约管理”。B接口挂了怎么办B处理到一半超时怎么办B返回的结果A没收到重发是幂等的吗这些全是问题。我后来想明白一个类比一个十几人的项目组不可能所有人都靠一对一喊话协作。必须有工单系统、共享文档、周报、会议纪要。任务派给谁、做到什么程度、卡在哪个环节这些信息不在任何一个人的脑子里而是沉淀在系统里。这就是持久化协作系统的基本要求。OpenRig把“项目组”那套运作机制搬到了Agent世界里。每个Agent不是直接用函数调用去喊别人而是往公共总线上投递事件从公共总线上订阅自己关心的事件。谁处理、处理到什么状态、为什么失败全部有记录。这样系统不再是靠一组脆弱的实时连接维持而是一个可以回放、恢复、审计的事件流。1.3 选型为什么OpenRig用事件驱动而不是静态流程图在动手前我们也评估了LangGraph这类图编排方案。坦白说静态图对固定流程是合适的节点确定边确定每个节点做什么一目了然。但真实的多Agent协作很快就会遇到麻烦。比如审核不通过需要回退回退之后可能还要人工介入人工改完可能再触发新的分支。这些动态行为如果全画到图里图会变成一团乱麻。更麻烦的是图编排通常把执行状态保存在内存里进程一重启整个图的状态就丢了。OpenRig换了个思路不把流程定义成一张写死的图而是用“事件驱动 状态机”来描述协作。每个Agent只关心自己订阅的事件收到事件后做自己的事再把结果作为新事件发出去。任务当前处于什么阶段由独立的状态机管理。事件是事实状态机是规则Agent是执行者。三者拆开之后协作系统的可维护性一下子高了很多。我在后面会详细讲这三个模块。需要先说明的是这个选型并不是说图编排一无是处。流程极简、场景固定的系统静态图开发效率确实高。但如果你想做一个能长期演进、能容忍故障、能回溯审计的多Agent系统事件驱动几乎是必经之路。2. OpenRig的三个核心模块总线、状态、记忆2.1 消息总线让Agent之间不直接喊话OpenRig的消息总线默认使用NATS JetStream。为什么选它而不是Kafka或者Redis Stream因为Kafka有点重Redis Stream的持久化和消费组机制在丢消息场景下不够干净。NATS JetStream轻量、原生支持持久化、消费组和ack机制对Agent之间的异步消息已经足够。在总线上传递的消息格式是统一的。一个标准事件大概长这个样子{ event_id: evt_01HZ3K8X9XYZ, task_id: task_20250601_001, producer: researcher, type: task.research.completed, payload: { topic: 智能客服, report: ... }, timestamp: 2025-06-01T10:30:00Z, correlation_id: req_00A1B2 }event_id是全局唯一的事件标识。task_id用来标记这是哪一次业务任务。producer表示是哪个Agent产出的。type是事件类型也是Agent订阅时的匹配依据。payload放实际业务数据。correlation_id是链路追踪ID我要求所有Agent在消费和产出时都必须把它原样透传。不要小看correlation_id。多Agent协作最怕的就是问题来了不知道怎么串链路。有了这个字段你可以在日志系统里按correlation_id查询整个任务从创建到结束的所有事件。我们团队内部有个不成文的规定任何一个Agent日志里没有correlation_id这条日志就当作没打。2.2 任务状态机用乐观锁防住并发碰车只有消息总线还不够因为Agent收到消息之后任务还是那个任务它处于什么阶段必须有统一凭证。OpenRig里每个任务对应一张状态记录表结构大概是这样字段说明task_id业务任务IDcurrent_status当前状态version乐观锁版本号owner_agent当前正在处理该任务的Agentpayload任务的公共数据JSONBupdated_at最后更新时间状态流转被定义成一个状态机pending - assigned - running - succeeded失败则进入failed允许重试的话进入retryingretrying之后回到assigned。状态机的好处是非法转移直接拒绝。比如一个已经succeeded的任务不允许被另一个Agent改成running。这里有一个特别值得讲的设计乐观锁。每个Agent要更新任务状态时必须带上自己读到的version。SQL是这样的UPDATE tasks SET current_status succeeded, version version 1, updated_at now() WHERE task_id ? AND version ?;如果更新的行数是0说明别人已经改过了这个Agent就不能再覆盖需要重新读取最新状态再决定下一步。为什么不直接用数据库行锁因为Agent处理一次任务可能耗时几十秒甚至几分钟你不可能让一个事务把行锁Hold那么久。悲观锁会浪费连接和数据库资源乐观锁加版本号冲突了重试就好。2.3 事件溯源把“记忆”从Agent脑子里搬出来很多Agent框架会把对话历史、中间结果存在LLM调用的上下文里。这在单轮demo里没问题但一旦Agent实例崩溃、容器重启、Session过期记忆就没了。OpenRig的做法是事件溯源所有Agent产生的中间结果都作为事件追加写入event_store表。这个表是一个append-only的日志。设计上只有插入操作没有更新和删除。每次Agent完成一个阶段就往里面追加一条事件。这样做有三个直接好处。第一个是可恢复。新启动的Agent实例想了解任务上下文不需要去翻旧Agent的内存直接查该task_id的事件列表按顺序回放一遍当前状态自然就重建出来了。第二个是可审计。谁在什么时间、基于什么输入、产出了什么内容全部有记录。我们接审核类业务流程时这个能力几乎是被业务方强制要求的。第三个是可调试。系统出了奇怪问题可以直接把指定任务的事件日志拉出来一步一步看状态是怎么转移的。比起看代码推断效率高太多了。但事件溯源也有代价。任务跑得越久事件越多回放就越慢。OpenRig的解法是快照。系统每隔一段时间或者每积累N条事件就把任务的最新状态生成一个checkpoint存下来。恢复的时候只需要加载最新快照再回放快照之后的新事件就行。这个机制和数据库的WAL加checkpoint是同一个道理。3. 实操搭一个“调研-写作-审核”持久化协作链路3.1 三个Agent的分工与协作协议如果你手边没有具体场景我建议先照着我这个例子跑一遍改改就能用到自己的业务里。这个链路包含三个Agent调研Agent、写作Agent、审核Agent。需求是这样的用户给定一个主题调研Agent先搜集整理背景资料并生成报告写作Agent看到报告之后基于报告写一篇初稿审核Agent检查初稿是否符合合规要求通过就发布不通过就要写上修改意见退回给写作Agent重新写。这个例子覆盖了多Agent协作的三种典型交互顺序传递、异步并行和回退重试。而且整个过程都是事件驱动不存在任何一个Agent在自己的代码里写死“调用下一个Agent”。3.2 环境准备数据库、消息队列、OpenRig运行时OpenRig核心是用Rust写的提供Python SDK和命令行工具。为了演示方便下面统一用Python SDK。起步前先把基础设施跑起来我用Docker Compose。version: 3 services: postgres: image: postgres:16 environment: POSTGRES_DB: openrig POSTGRES_USER: openrig POSTGRES_PASSWORD: openrig ports: - 5432:5432 nats: image: nats:2.10 command: [-js, -m, 8222] ports: - 4222:4222 - 8222:8222启动服务之后安装Python SDK。pip install openrig-sdk然后初始化OpenRig运行时。openrig init --db postgres://openrig:openriglocalhost/openrig --nats nats://localhost:4222init命令会创建事件表、任务表以及必要的索引。这一步做完系统骨架就立起来了。3.3 配置Agent订阅什么、产出什么、失败怎么办OpenRig用一个YAML文件描述所有Agent的协作协议。协议的粒度是事件类型每个Agent声明自己订阅什么事件、产出什么事件。这样后续新增Agent不需要改旧Agent的代码改配置就行。agents: researcher: subscribe: [task.research.request] publish: [result.research] max_retries: 3 timeout_sec: 600 prompt_template: 你是行业调研专家围绕{topic}输出结构化报告 writer: subscribe: [result.research] publish: [draft.article] max_retries: 2 timeout_sec: 900 prompt_template: 基于调研报告撰写文章初稿语言要专业 reviewer: subscribe: [draft.article] publish: [review.result] max_retries: 0 timeout_sec: 300 prompt_template: 请审核初稿输出通过或修改意见 rules: - name: reject-to-writer when: review.result.status rejected action: rewrite target_agent: writer这里再说一下rules的作用。审核Agent产出review.result事件如果是个拒绝OpenRig会根据规则把这个事件重新路由给writer而不是让writer自己写代码轮询。也就是说回退逻辑被收编到系统里了Agent本身保持无状态。配置写好之后执行加载命令openrig apply -f agents.yamlapply命令会把订阅关系和规则下发到运行时里它会负责监听消息、投递给对应Agent。3.4 写一个Agent的Python运行主循环Agent这边不需要理解状态机整套细节只需要写一个回调函数。以写作Agent为例from openrig import OpenRigClient client OpenRigClient() client.subscribe(result.research) async def handle_research(msg): task_id msg[task_id] topic msg[payload][topic] report msg[payload][report] # 恢复这个任务的历史上下文防止实例重启后丢失 context await client.load_context(task_id) history context.events prompt f 基于以下调研报告写一篇初稿。 主题{topic} 报告{report} 历史事件数{len(history)} draft await llm_generate(prompt) # 完成当前阶段产出新事件 await client.complete( task_idtask_id, event_typedraft.article, payload{draft: draft}, correlation_idmsg[correlation_id], ) client.run()注意几个关键点。load_context不是简单地从数据库捞一段文本而是把该任务之前的事件列表传进来让Agent能感知到上下文已经发生过什么。complete是标准动作它内部会做三件事追加事件、更新任务状态、把事件发到消息总线。如果Agent处理失败OpenRig会按配置里的max_retries自动重试重试次数用完就标记为failed。所以Agent的代码可以写得很薄。它不需要关心消息可能重复投递不需要关心状态机怎么转移那些是运行时和状态机去协调的事情。3.5 发布任务与观察状态流转现在把整个链路跑起来。先创建一个任务。curl -X POST http://localhost:8000/v1/tasks \ -H Content-Type: application/json \ -d { type: task.research.request, payload: {topic: 智能客服系统} }返回结果里会有task_id。然后开另一个终端观察任务。openrig task show task_20250601_001正常情况下会看到任务状态一步步变化。刚开始是pending调研Agent消费后变成assigned再变running最后succeeded。事件列表会记录每个阶段调研完成、写作完成、审核通过。如果审核拒绝事件列表里能看到review.result的status是rejected然后状态机会自动把事件重新投递给writer任务状态进入retrying。整个过程不需要人工介入。我这里强烈建议你把这个例子原封不动跑下来然后故意让审核Agent的Prompt改成“永远拒绝”。你会看到系统自动回退三次最后任务进入failed。这个“失败路径”比“成功路径”更能让你理解持久化协作系统为什么需要状态机和事件溯源。4. 多智能体持久化排障实录4.1 消息重复消费幂等键救场多Agent系统里消息至少一次投递是常态。NATS JetStream的ack如果超时或者消费端处理完后网络抖动消息就会重投。于是同一个事件可能被同一个Agent消费两次。我们遇到的实际案例写作Agent处理调研报告时因为数据库事务提交慢触发了超时重投。结果两个进程同时消费了同一份报告生成了两篇初稿还都写进了事件表。排查的时候才发现问题不是NATS的问题是我们自己的事件表没有幂等约束。后面在event_store表上加了一个唯一约束。ALTER TABLE events ADD CONSTRAINT uq_task_event UNIQUE (task_id, event_id);插入事件时使用ON CONFLICT DO NOTHING消费时如果插入影响行数为0说明已经处理过直接ack并跳过。这样就算消息重复投递事件也只会落库一次。这个坑的教训是不要指望消息队列帮你做到恰好一次那个成本极高。分布式系统里更现实的方案是“至少一次”加上“业务幂等”。幂等键最好提前设计到了线上再补数据洗起来会很痛苦。4.2 状态冲突和任务死锁状态机不止画图第二个高频问题是状态冲突。多个Agent并行跑的时候两个Agent可能同时认领同一个任务。比如审核Agent和定时补偿Agent同时发现某个任务超时都尝试把任务状态改成running。没有乐观锁的时候后提交的会覆盖先提交的owner_agent就乱了。加上version乐观锁之后冲突发生时更新行数为0Agent就知道自己“抢任务失败”了。正确的处理是等一下随机时间重新读取任务看是否还需要处理。不要立刻重试否则两个Agent会一直碰撞。还有一个比冲突更隐蔽的问题是任务级死锁。A任务在等B任务完成B任务又在等A任务完成双方都不进行下一步。纯消息总线系统里这种死锁不会自动暴露。OpenRig给每个任务加了一个心跳字段。任何Agent处理任务时都会定期刷新heartbeat。运行时有一条后台扫描任务如果发现某个任务心跳长时间没更新就把它标记为stalled。stalled状态会触发告警业务上我们配置了统一补偿回调把任务重新投递给空闲Agent。设计上有一条铁律禁止Agent在同步等待另一个Agent完成。所有跨Agent等待必须是异步事件驱动。你只要允许Agent写“我先睡一会儿等对方处理完再继续”死锁排查会变得无比痛苦。4.3 并发与性能怎么扛住一大波任务很多人问过我“AI Agent怎么扛并发”这里的瓶颈通常不在模型本身而在编排层的数据库和消息队列。我们的event_store一开始是全表无脑写入任务量上来之后单表查询越来越慢回放任务上下文时要扫描大量无关记录。后来做了三件事。第一是按时间分区。event_store按月分区老数据自动归档到冷分区热分区保持较小体积。查询任务事件时走task_id加created_at的联合索引回放速度快了很多。第二是批量写。Agent的complete操作默认是单条事件落库在高并发时数据库连接很容易被打满。OpenRig加了本地缓冲攒够一批事件后统一批量insert。这个改动把数据库写入TPS提升了一个数量级。第三是定期快照。前面说过事件溯源会积累大量事件如果一个任务跑了几百个步骤每次恢复都要回放几百条事件太慢。快照机制让恢复操作变成“加载快照 回放增量”绝大多数任务的恢复时间可以控制在几十毫秒。做性能优化的时候要记住一点先测量再优化。我们一开始凭感觉觉得是数据库慢结果看了NATS的监控才发现是消费组不平衡导致部分Agent堆积。后来把消费组配置改了问题就没了。持久化协作系统里消息队列、数据库、Agent计算三个环节都可能成为瓶颈别只盯着一个地方。最后分享两点真实体会做完OpenRig这套实践我最大的体会有两点。第一Agent的上下文不能当持久化存储用状态和记忆要沉淀在系统里。模型会话是柔软易散的系统状态必须是坚硬可查的两者混在一起迟早出事。第二多Agent协作的能力上限不取决于某个Agent有多聪明而取决于通信、状态、记忆这些基础设施做得有多扎实。聪明是模型的可靠是系统的。最后再分享一个小技巧调试这类系统时事件日志一定要打全。一条事件里带上task_id和correlation_id还不够还要在日志里打清楚“你是谁、你消费了哪个事件、你产出了哪个事件”。多Agent系统跟单机程序不一样问题常常不在报错的地方而在某条消息悄悄被丢弃的缝隙里。把这些细节做到位系统才能从“demo很酷”走向“生产能用”。