OpenRig：构建高可靠多智能体协作系统的编排实践

1. 为什么需要OpenRig从单体Agent到多智能体协作的工程困境做AI Agent开发时间长了你会发现一个扎心的事实单体Agent的能力边界是清晰存在的。单靠一个Agent加上一堆工具能解决的需求天花板很低。你自己搭过一个完整的Agent就会明白检索增强、工具调用、上下文管理、多轮对话这些单点技术其实都有成熟方案真正让人头疼的是把多个Agent组织起来分工协作——这才是工程上最容易被低估的深坑。我最初也尝试过最朴素的思路用代码写死一个调用链Agent A处理完把结果喂给Agent BB再调用Agent C。听起来很直接但真实场景一跑就崩。原因有三层第一Agent的返回结果天然带随机性模型温度调到0也不可能保证每次输出完全一致写死调用链等于把不确定性引入关键路径第二Agent之间互相传递的数据结构会随着业务迭代不断膨胀你根本没法用固定接口约束第三也是最致命的——一旦某一个Agent调用超时或返回异常整个链路的故障排查难度是几何级增长的。OpenRig这个项目的出发点就是解决这类问题。它的核心思路听起来其实不复杂把每个Agent当成一个有独立能力的节点通过一个统一编排层来管理它们的注册、调度、状态持久化和消息传递让原本一个个孤立的Agent变成长时间运行、可恢复、可观测的协作系统。比起传统工作流引擎那种预先画好DAG靠流程引擎推动的模式OpenRig更像是在Agent之间织了一张网——所以名字里用了Rig这个词它本来有装配、钻井架的意思引申过来就是把松散的部件装配成一个稳固结构。1.1 多智能体系统真正的痛点在哪里在做OpenRig之前我认真复盘过团队之前那几个多Agent项目为什么最终都没能落地。你会发现它们失败的点高度重合基本集中在四类问题上一是Agent之间的通信协议完全是野路子。有人用自然语言让Agent互相传话看起来灵活实际上不可控Agent A理解的汇总报告和Agent B输出的汇总报告根本不是一回事有人用JSON硬编码传递又回到了上面说的接口僵化问题。二是状态管理接近空白。Agent执行到一半任务状态全存在内存里服务一重启全部丢失。我们当时有个业务场景是Agent需要连续处理五个子任务再汇总每次跑到第三个子任务时进程崩溃就得从头开始跑。一天崩两次业务方直接骂娘。三是并发控制靠运气。多个Agent同时访问同一个共享资源没有锁、没有队列、没有幂等机制数据错乱成为常态。四是整个系统没有观测性。每个Agent内部发生了什么事只能靠日志里偶然打印出来的一行半句话去猜。这些问题单个拎出来都不算大但叠加在一起就会让系统变得极其脆弱。OpenRig在设计时就是冲着这四个点去的确立一套通用的消息协议、把关键状态持久化下来、在编排层做并发控制与调度、最后把所有节点的运行轨迹接到统一的日志与追踪体系里。1.2 编排层选型的核心权衡在真正动手写第一行代码之前最需要想清楚的问题其实是编排层到底应该多厚市面上有不少多智能体编排框架大致可以分成两类。一类是重编排编排层管着Agent的每一步行动Agent本身只是被动的执行器每一步干什么都由中央大脑说了算另一类是轻编排编排层只负责把任务拆解并分配给合适的AgentAgent内部怎么做有相当大的自由度编排层只关心结果。两种路线各有代价。重编排的好处是可控性极强坏处是编排层的单点压力非常大而且Agent一旦遇到预期外的分支编排层就得有足够的兜底逻辑来响应这会把编排逻辑写得异常复杂轻编排的好处是Agent的自主性比较强系统整体更灵活坏处是任务质量很大程度上依赖Agent自身的规划能力。我个人在实际工程化过程中的体会是如果你想要的是一个生产级别的多智能体系统任何一个纯路线都会让你痛不欲生更好的方向是编排层定边界Agent内在灵活。OpenRig采取的就是中间路线编排层负责任务的注册、依赖分析、状态持久化以及消息路由Agent内部流程由Agent自己决定。这种做法最大的收益在于明确了编排层和Agent层各自的职责边界出了问题你起码能迅速断定是编排层的调度问题还是Agent本身的执行问题。2. OpenRig整体架构把离散Agent编织为一体2.1 全局协调器与Agent节点的关系OpenRig顶层结构里最关键的两个概念是全局协调器Coordinator和Agent节点。协调器本身不执行具体业务逻辑它的工作只有四件事接收任务请求、拆解任务并生成任务清单、把清单里的每个子任务路由给注册过的Agent、跟踪每个子任务的执行状态。这里有一个设计细节我觉得特别值得分享协调器对Agent的调用必须走异步消息而不是同步HTTP调用。同步调用会导致两个问题。第一个是超时传染Agent A执行60秒超时调用链上所有等待方都会被拖住第二个是状态无法持久化调用是即发即收的如果协调器在这个过程中崩溃正在执行的任务状态就丢了。改成异步消息之后协调器把任务信息写入消息队列就立刻返回Agent从队列里取任务执行完再通过回调接口上报结果。整个执行链路是断点续传式的任何一个环节挂了消息还在队列里重启之后可以继续消费。Agent节点侧则遵循着非常简单的接口约定每个Agent对外暴露一个统一的执行入口输入是一个任务消息对象输出是一个结果消息对象。任务消息对象里包含任务ID、任务类型、输入数据和幂等键结果消息对象包含任务ID、状态码、输出数据和错误信息。我们可以看一下Agent节点的最低限度接口长得像什么# agent_base.py from pydantic import BaseModel from typing import Any, Optional class TaskMessage(BaseModel): task_id: str task_type: str input_data: dict[str, Any] idempotency_key: str class ResultMessage(BaseModel): task_id: str status: str # succeeded / failed / retryable_failed output_data: Optional[dict[str, Any]] None error_info: Optional[str] None class BaseAgent: async def execute(self, task: TaskMessage) - ResultMessage: raise NotImplementedError只要每个Agent插件化地把这个接口实现掉它就可以被OpenRig纳入管理。定义这个接口的时候我去掉了所有跟业务相关的字段只保留编排层关心的通用信息因为编排系统最忌讳的就是跟业务耦合太深。2.2 任务状态机的设计与持久化模型Coordinator收到一个用户请求之后会把它拆解成一张任务清单然后逐个调度。为了让整个执行过程可以随时被恢复所有任务状态都用状态机来表达。这里的状态机我设计了五个状态PENDING等待执行、RUNNING正在执行、RETRY_WAIT重试等待中、COMPLETED已完成、FAILED不可恢复的失败。状态转换的路径有两个地方会产生。一个在Agent侧Agent执行成功写COMPLETED遇到可重试错误比如上游API临时超时返回 RETRY_WAIT另一个在协调器侧从RETRY_WAIT重新入队之后转回PENDING接着分发给Agent继续跑。用了这个状态机之后任务恢复的逻辑就变得干净了任何时刻系统崩溃重启之后扫描一遍状态表把RUNNING或者RETRY_WAIT的任务找出来重新入队即可。这里再分享一下我对持久化的建模理解。原本容易犯的错误是为每一个Agent建立单独的状态数据表后来发现边界根本没有办法划清——因为Agent之间的数据是有依赖关系的Agent B用的很可能是Agent A的中间结果。OpenRig最终采取的是两种持久化对象一种叫任务状态记录TaskStateRecord记录任务本身的流转状态另一种叫共享工作内存WorkingMemory本质上是一棵带版本号的键值树用来存放Agent之间需要交换的中间产物版本号用来做并发控制。这样设计之后每个Agent执行时不是去直接读另一个Agent的输出而是从工作内存里按key取数据配合版本号判断数据是否过期。2.3 共享工作内存的具体实现工作内存的实现我用了Redis的Hash结构这是我觉得性价比相当高的做法。键名格式是wm:{memory_key}value是一段JSON包含data和version两个字段。# memory_ops.py import json, aioredis redis await aioredis.from_url(redis://redis:6379/2) async def write_memory(key: str, data: dict, expected_version: int): current await redis.hget(wm: key, payload) if current: current_version json.loads(current)[version] if current_version ! expected_version: raise ValueError(fversion conflict for key {key}) new_version expected_version 1 payload json.dumps({data: data, version: new_version}) await redis.hset(wm: key, payload, payload) async def read_memory(key: str): raw await redis.hget(wm: key, payload) if not raw: return None return json.loads(raw)写入时带上期望版本号本质上就是乐观锁。并发环境下两个Agent同时尝试更新同一个key时只有版本号匹配的那个能写成功另一个会收到异常自行决定重读重试。实测下来这个版本号机制能够挡掉绝大多数数据竞争问题。有几次我在跟同行交流时听到有人问工作内存直接放Redis里那Agent老去读写Redis不是拉低了模型生成的速度吗这里有一个认知误区Agent的大部分耗时都耗在模型推理上Redis操作是微秒级的对整体时延的影响几乎可以忽略。真正需要花心思优化的是减少Agent读写工作内存的次数也就是尽量减少无效的跨Agent数据交换。3. 持久化层的工程落地从Redis RDB/AOF到任务恢复3.1 Redis持久化机制详解与选型理由编排层的协调器和工作内存都依赖Redis那Redis本身挂了怎么办Redis默认是把数据放内存里的如果进程退出或者机器断电内存中的数据会全部丢失。所以持久化是绕不开的一环这里就有必要把Redis持久化背景说透。Redis的持久化机制有两种主要方案。第一种叫RDB快照按照配置的时间间隔把当前内存里的全量数据生成一份二进制快照写入磁盘。RDB的优点是恢复速度快、文件紧凑适合备份缺点是它会丢失最后一次快照之后写入的数据。举例来说配置是save 900 1表示15分钟内至少1次写入才触发快照如果在第899秒写入了一条数据然后机器断电这条数据就丢了。第二种叫AOF日志每条写命令都追加到日志文件恢复时重放日志即可。AOF有三种刷盘策略always、everysec和no。always最安全但性能最差每条写命令都要落盘everysec折中最多丢1秒的数据no则完全依赖操作系统自己刷盘性能最好但顺序无法保证。OpenRig这边我实际采用的是双开策略也就是RDB和AOF同时开启并且把AOF的刷盘策略配成everysec。用everysec是因为Agent编排场景对性能挺敏感的大量任务状态要高频写入用always会让写吞吐直接掉一个档次但相比RDB的分钟级快照everysec最多丢一秒数据这个损失在可接受范围内。恢复时Redis会优先用AOF文件恢复因为AOF中的数据一般来说更新。关于持久化配置你还需要关注一个细节Redis有一个后台子进程重写AOF的机制重写期间会fork子进程。如果你的机器内存本身就紧张fork需要复制页表可能会导致瞬间的内存翻倍。所以跑编排系统的Redis节点内存规格要留足余量。3.2 状态快照的恢复流程设计有了持久化的底子OpenRig的任务恢复流程就是顺理成章的事情。我把恢复分成两个层面协调器恢复和Agent恢复。协调器恢复的触发条件是协调器进程异常退出。重启的时候它会做三步第一步从Redis的任务状态表里拉出所有RUNNING和RETRY_WAIT状态的任务记录第二步将这些任务的状态统一重置为PENDING并写入一条recovered:true的标记到任务消息里第三步把这些任务重新推入待处理队列等待空闲Agent消费。Agent恢复相对更复杂一些。因为Agent执行到一半失败时它的内部上下文比如对话历史、中间推理步骤是需要被重新构建的。我的做法是让Agent自己上报checkpoint。每执行一个关键子步骤Agent就把当前上下文序列化成JSON存到以agent_ctx:{agent_id}:{task_id}为键的Redis结构里。下次任务重新分发到同样类型的一个Agent时它先从Redis里读上下文如果有就接着跑没有就从头开始。用大白话说相当于写文档时每写一段就按一次CtrlS崩溃了也不用担心从零开始。这种设计下你会发现一个有意思的现象Agent有状态还是无状态变得不明确了。从业务角度看Agent确实是有状态的它有checkpoint但从调度角度看Agent又是无状态的任何一台机器上的同类型Agent实例都能接着执行。这给水平扩展省了很多事。3.3 持久化对象清理与生命周期管理持久化不是存得越多越好。如果你用一段时间不去管Redis里会堆满完成状态的任务记录和Agent上下文内存被吃光只是时间问题。OpenRig里我加了一个生命周期管理组件规则比较直接任务状态记录完成任务后保留24小时用于追踪和排查之后打包归档到冷存储。工作内存里的中间结果在父任务COMPLETED后立刻标记为可清理延迟一小时物理删除。Agent上下文checkpoint在任务结束后保留最长7天超过7天自动清理。所有清理操作都通过定时任务执行跑完记录一份清理报告。忘了说一个比较反直觉的坑清理任务本身也可能和正在运行的任务竞争Redis资源。所以清理程序一律跑在低峰期比如每天凌晨两点并发度限制在2。我在早期版本里踩过这个坑清理程序白天跑结果把正在执行任务链路上的工作内存给删了导致下游Agent到处找不到数据。后来加了两个保险清理前检查任务状态确认是COMPLETED以及延迟一小时物理删除。4. 编排层的并发与扛量FastAPI LangGraph路径下的实战配置4.1 从同步调用改为全链路异步聊完持久化再来解决标题里那个很现实的问题多个Agent同时跑怎么扛住并发我的选择是基于FastAPI LangGraph把整条链路做成异步。FastAPI本身的Async支持很成熟LangGraph原生也支持异步图执行两者拼接起来非常顺。给你搭个大概结构。编排服务对Agent的调用改造成两层第一层是Commander负责接收外部请求并拆解子任务把子任务全部写入一个Redis Stream第二层是Worker从Stream里消费任务通过LangGraph定义的Agent图执行。FastAPI的对外接口保持纯异步# api.py from fastapi import FastAPI, BackgroundTasks app FastAPI() app.post(/dispatch) async def dispatch_task(payload: dict): task_id uuid.uuid4().hex await commander.register_task(task_id, payload) await commander.enqueue_subtasks(task_id) return {task_id: task_id, status: accepted}这里的关键点是接口只保证任务被接受而不保证任务被执行完真正执行放在后台的Worker里异步做。如果接口同步等待任务执行完一个长任务20秒接口就得占用20秒的连接很快整个服务就会被为数不多的长任务拖垮。4.2 Worker数量、并发度与限流策略Worker并发度到底开多大经验值先给你起步2个Worker每个Worker跑2个并发的任务循环。这个参数不是越大越好因为Agent在执行过程中会大量调用外部API比如模型推理API这些API自身有rate limit你把Worker开到50表面上吞吐上去了实际上大量请求会在第三方接口排队超时反而拖慢整体效率。我实际调整时是这么操作的先压测找到外部依赖的QPS上限然后反推Worker数量。举个例子模型推理API限制是每分钟60次单个Agent一次任务平均要调用3次模型推理那每分钟最多能完成20个任务。如果你把Worker并发度开到每分钟30个任务的量就会有1/3的任务因限流失败重试。正确的做法是给每个Agent加一个信号量# worker.py import asyncio from langgraph.graph import StateGraph semaphore asyncio.Semaphore(4) async def consume(subtask): async with semaphore: graph build_agent_graph() result await graph.ainvoke(subtask.to_state()) await reporter.report_result(subtask.task_id, result)顺手还加了指数退避重试。第一次重试等5秒第二次10秒第三次20秒最多重试五次。这比无脑交织重试的效果好得多至少避免了Agent在限流恢复后集体涌入造成二次雪崩。4.3 Redis Stream的消费组与故障转移任务队列这里我用的是Redis Stream而不是简单的List。Stream的消费者组模型带来一个很重要的能力多个Worker消费同一个任务流时一条消息只会被一个Worker取走天然避开分布式重复消费的问题。更实用的地方在于消费者组里有pending entries列表如果一个Worker处理到一半崩溃了消息会一直停在pending列表里另一个Worker可以用XCLAIM命令把它抢过来继续处理。实操里有一个经验消费者组的消息确认机制一定要用对。Worker从Stream里取出消息后先处理业务逻辑成功完事之后才执行XACK确认删除消息。绝对不能先ACK再处理——那样一旦处理后半段崩了消息就彻底丢了。我们团队在早期就是先ACK再执行上线不到一周就出了数据丢失的事故后来改成先处理后确认就把问题解决了。4.4 分布式锁与幂等性保障多Agent并发跑起来之后防止同一个任务被执行两次就是必须解决的硬问题了。比如网络抖动导致Worker没收到结果协调器重试时可能重新触发Agent执行。OpenRig里每个任务消息都要求携带幂等键Agent执行前先通过SET NX命令在Redis里抢占这个键# idempotency.py ok await redis.set(fidem:{idempotency_key}, task_id, nxTrue, ex3600) if not ok: existing await redis.get(fidem:{idempotency_key}) if existing task_id: # 同一任务重试直接读取上次执行结果 result await redis.hgetall(fresult:{task_id}) return result raise ValueError(idempotency key already used by another task)这套逻辑做完至少可以保证同一个任务在大多数并发场景下不会重复跑。过程中还遇到过一个隐蔽问题两个不同任务如果用了相同的业务唯一ID拼接出的幂等键会被误判成重复任务。所以幂等键的生成规则必须包含任务类型前缀避免跨任务串号。4.5 水平扩展与多节点部署做到前面这些之后单个服务节点的并发瓶颈依然存在。Redis里的消息和状态可以支持任意节点共享访问所以OpenRig的编排层天生支持多节点水平扩展——你只需要在服务前面挂一个负载均衡器把请求分发到多个无状态编排服务节点上这些节点共享同一个Redis集群。但这里要提醒一句不是无限加节点就能无限扛量。瓶颈通常会转移最常见的是转移到Redis本身其次是外部API的限流。我做了两轮压测后的数据单节点编排服务大概能稳定扛住每秒钟25个任务的调度量指拆解和入队阶段不含Agent执行水平扩展到3个节点可以跑到每秒70个左右再往上加节点提升就非常有限了因为热点集中在共享的Redis和统一的外部依赖API上。如果业务量还要继续涨更值得做的投资是拆分Redis集群把任务队列、工作内存、状态记录分到独立的Redis实例上。5. 常见问题与排查技巧实录从线上事故中总结的避坑经验5.1 Agent状态丢失任务重跑却找不到checkpoint现象一个Agent执行长任务时崩溃协调器重试后发现任务从头开始跑之前上报的checkpoint全都没了。排查过程登录Redis看agent_ctx键是否存在结果发现键确实存在但却是一个空字符串。仔细查代码后发现是Agent在跑完第一步时就把空上下文覆盖写进去了后续步骤没有更新。根因是checkpoint的写入时机写错了不是因为执行到关键步骤才写而是每个步骤结束都会写而第一步写入的恰好是一个空上下文。解决把Agent执行流程里的checkpoint写入逻辑改成有实际进展才写判断条件是上下文里是否新增了有效的中间结果。后来我还在Agent里加了每一步的checkpoint序号恢复时如果发现序号低于崩溃前上报的最大序号就拒绝旧checkpoint强制要求Agent从最近的有效位置重跑。这个事故给我留下的教训是做持久化设计时必须先定义清楚什么算值得持久化的中间状态而不是机械地每隔几步存一次。5.2 消息重复消费Redis Stream的pending机制引起的重复执行现象生产环境偶尔出现同一个任务被执行两次导致下游数据重复写入。排查过程查看日志发现两个Worker几乎同时收到同一条任务消息。原因是其中一个Worker消费消息后还没来得及ACK进程就因为内存压力被系统杀掉了消息留在pending列表里另一个Worker在等待一段时间后通过XCLAIM抢走了这条消息执行了任务。两个执行操作最终都写到了同一个业务表。解决这是典型的at-least-once语义问题。靠消息队列本身做不到exactly-once所以把幂等保障下沉到业务执行层。两个改进点一是写入业务数据前通过数据库唯一索引拦一遍二是业务写入动作改为update if existsinsert if not exists。这两个兜底做了之后重复执行最多产生一次无效计算不会再产生脏数据。这里也想说一句很多在讲消息队列的文档里都会提到at-least-once、exactly-once但你真在Agent编排场景里遇到的时候才会理解这坨概念的现实分量。5.3 Redis持久化配置导致的RDB阻塞问题现象大促流量进来后任务处理延迟突然飙高有一批任务集体超时。排查过程先看Redis的慢日志发现大量RDB save操作耗时超过2秒。查配置发现RDB触发条件被调得很激进数据量大时fork子进程加上写磁盘耗时阻塞了Redis主线程。编排系统写入任务状态的频率又很高结果就是Redis时不时卡上两秒。解决调整持久化策略RDB触发条件放宽由每天一次的定时备份兜底AOF继续保持everysec策略。调整后Redis主线程的阻塞消失了。这个事故说明一个道理持久化配置不是安全但慢的二元选择它需要针对具体的写入模型做参数调优盲目的安全策略会反过来拖死你的线上业务。5.4 Agent返回结果不稳定同样输入输出不同现象同一个任务分发给两个同类型Agent产出的结果字段结构不一致下游Agent解析时报错。排查过程检查Agent提示词发现返回格式是靠提示词自觉保证的没有在代码层强制约束。模型一抽风多给了一个字段或者少了一个数组下游就炸了。解决把所有Agent输出强制接入一个输出校验器用Pydantic模型做运行时校验。模型输出的JSON先解析不符合schema就直接让Agent重新生成一次最多重试两次。实话说这个方案也不能保证100%成功但它把脏数据的比例从肉眼可见的5%降到了接近0.1%以内。安全兜底还有一层校验失败且重试仍然失败时返回RETRY_WAIT状态让协调器重新调度一个全新的Agent实例来跑。在LangGraph里实现这个校验器的姿势是把输出Parser挂到图的末端节点# langgraph_output_validator.py from pydantic import BaseModel, ValidationError from langgraph.graph import END class AgentOutput(BaseModel): summary: str decisions: list[str] confidence: float def validate_output(state): raw state[generated] try: parsed AgentOutput.model_validate_json(raw) return {parsed_output: parsed} except ValidationError: return {parsed_output: None, needs_repair: True}5.5 时间戳乱序跨节点协调器的时间不同步现象任务状态记录里出现结束时间早于开始时间的怪异记录重试逻辑判断异常。排查过程两台编排服务节点所在机器的系统时间有将近3秒的偏差。任务状态记录的时间戳由各个节点自己生成导致全局视角的时间线是乱序的。解决时间戳统一由Redis生成用Redis INCR一个时间序列字段或者直接用Redis的TIME命令取服务器时间。跨节点之间的时钟差异在这种场景下就不再影响排序了。后来我还养成了一个习惯任何分布式系统的状态排序字段一律不信任业务节点本地时钟。5.6 排查工具与启动参数速查既然聊到排查顺手整理一份我在OpenRig现场最常用的排查清单按场景分好类可以直接参考场景排查命令关键指标任务卡住不执行XINFO GROUPS查看Stream消费组积压pending数量持续增长说明Worker不足Worker无响应htop 进程状态看线程栈线程卡在IO等待说明外部依赖慢了Redis内存占用过高redis-cli --bigkeys定位大key大key多为工作内存残留手动清理任务执行超时慢日志 调用链追踪Agent执行时间分布是否集中在某类节点重复执行查询幂等键命中数命中数大于预期说明消息有重复消费Agent输出格式错误输出校验器错误日志错误集中在哪个Agent表明需要调提示词上面这套是我从真实故障中不断补全的速查表。刚开始做OpenRig的时候排查问题全靠人肉翻日志效率极低后来逐步补上观测性和这些排查命令平均故障定位时间从小时级降到分钟级。6. 一些后续可以继续深挖的方向OpenRig做到现在这个程度已经能满足中小规模多智能体场景的持久化编排需求。我自己的使用感受是真正让这套系统稳定跑起来的关键不是某个框架或者某个组件而是把状态可恢复、消息不丢失、并发可控制这三件事认真做到位。技术选型上用Redis做状态和队列的存储中枢用FastAPI和LangGraph搭服务与Agent执行层用Stream和乐观锁做并发控制整条路线很朴实但实测下来非常稳。后续我觉得有三个方向值得继续试第一个是把工作内存升级成一个支持向量检索的长期记忆库让Agent之间不仅能传结构化数据还能传语义化记忆第二个是给编排层增加动态扩缩容的能力根据队列积压情况自动调整Worker数量而不是像我现在的版本这样靠人工配置第三个是沉淀一套Agent协作的可视化追踪界面把每个任务的流转路径完整画出来。这些都是能明显提升多智能体系统生产可用度的方向我自己会继续在OpenRig里迭代下去。如果你们团队也在做类似的多智能体编排系统欢迎从这套思路里找灵感也欢迎给我这边提需求和分析案例大家一起把这条路走宽。