多智能体编排实战：用持久化状态管理突破Agent协作上限

1. 先把结论放在前面：单 Agent 的能力上限不在模型，而在状态管理

写这篇文章的时候，我刚刚把一个跑了将近一整天的多智能体编排任务恢复到断点，继续往下执行。系统没有异常，也没有丢失任何中间结论。这个名叫 OpenRig 的编排层，是我过去几个月反复推倒重写之后确定下来的方案，核心目标就一句话：把若干个互不认识的 AI Agent，组织成一个有记忆、有分工、能容错的持久化协作系统。

先交代清楚我为什么要碰这件事。最近大半年我一直在做 AI Agent 相关的业务落地，从扣子平台上的智能体应用，到基于 FastAPI + LangChain + LangGraph 的自研链路，都试过。一个非常明显的感受是：单 Agent 在真实业务里的天花板，往往不是模型能力，而是上下文和状态管理。任务一长，对话历史一多，模型就顾此失彼；中间一旦出现网络抖动、服务重启、某个工具调用超时，整个流程就得从头再来。这种体验放在“让 AI 真的下地干活”的场景里，基本是不可接受的。

OpenRig 不是从零训练或者微调模型的框架，它做的是编排。所谓编排，就是给一堆功能单一的 Agent 定义明确的角色、通信方式和任务交接规则，让它们像一个小团队那样协作。比起一个什么都干的“全栈 Agent”，一组各司其职、通过持久化状态协作的小 Agent，在可控性、可维护性、可观测性上都要好得多，这也是我在这个项目里最核心的取舍。

这篇文章会把 OpenRig 的设计思路、核心抽象、关键实现、完整实操过程以及踩坑记录都写出来，给正在做 Agent 搭建、或者被多 Agent 并发和状态丢失问题困扰的同行一个参考。无论你用的是 LangGraph、Spring AI 还是自研链路，第 2 章到第 5 章里的大部分经验和结论应该都能直接用上。

2. 整体设计与核心抽象：把“协作”变成可持久化的数据结构

2.1 四个核心抽象：Agent、Task、Message、Session

我在设计 OpenRig 的时候，给自己定了一条硬规矩：系统里所有概念，必须能落成数据库里的表。任何只存在于内存、进程内部、或者某个 Agent 的上下文窗口里的东西，都不允许成为协作的一部分。这条规矩听起来简单，但执行起来非常难，因为它直接决定了后面所有的技术选型。

围绕这条规矩，OpenRig 定义了四个核心抽象。

Agent是能力的边界，一个 Agent 只做一件具体的事。比如“搜索情报的 Agent”“写初稿的 Agent”“做格式校验的 Agent”。Agent 本身不关心业务流程，它只接收输入任务，执行，然后返回结果。Agent 的注册信息存在表里，包括名称、能力描述、支持的输入输出 schema、超时时间、重试策略。

Task是工作单元，表示一个需要被某个 Agent 执行的具体工作。Task 有状态、有依赖、有负责人。一个业务流程会被拆成一个有向无环图（DAG），每个节点是一个 Task，每个 Task 的数据进出都会被记录。为什么强调 DAG 而不是允许任意循环？因为只有无环的流程才方便持久化、恢复和追踪，循环逻辑可以靠编排层的“重新触发”来实现，不需要在依赖关系上引入环。

Message是 Agent 之间的通信载体。与直接函数调用不同，Agent 之间不直接调对方的方法，而是通过 Message 交换数据。每条 Message 都带 sender、receiver、message_type、payload、context_id 这些字段。这个设计模仿了消息总线，目的是让通信本身可记录、可回放、可审计。多 Agent 系统最容易出问题的就是“谁在什么时候给谁传了什么”，Message 全量落库之后，这个问题就变成了一个简单的 SQL 查询。

Session是贯穿始终的协作上下文。一个 Session 对应一次完整的业务目标，从触发开始，到最终结果交付结束。Session 这条记录是系统的主线，它的状态决定了现在有多少个 Task 处于可执行状态、哪些 Agent 已经被召回、哪些上下文已经被写入。Session 的设计直接决定了系统能不能“断点续跑”，所以在 2.3 节我会单独展开。

这四个抽象之间的关系，可以理解成：Session 是一个项目的工单，Task 是工单拆出来的工序，Agent 是完成工序的工人，Message 是工序之间传递的半成品和交接单。每个环节都有记录，每个环节都能回溯，这样协作才不会是黑盒。

2.2 编排模型：星型、链式还是图式

多 Agent 编排不是只有一种模型。我在早期原型里试过三种拓扑，最后明确了一个结论：没有最好的模型，只有按场景选择后封装成统一 DSL 的模型。

链式编排是最直观的。Agent A 的输出直接作为 Agent B 的输入，一个接一个执行。适合流程固定的场景，比如“翻译 -> 校对 -> 排版”。缺点是中间只要有一个节点失败，后面就断掉；而且一旦某个 Agent 需要并行处理多个子任务，链式模型就表达不了。

星型编排是典型的主管-下属模式（supervisor pattern）。一个主管 Agent 负责任务理解、拆解、分派和结果整合，下面的执行 Agent 各自独立。这种模型适合意图开放、子任务不固定的场景，比如“用户说了一个模糊需求，系统先让规划 Agent 拆解，再分给多个执行 Agent”。OpenRig 对星型模型支持得最好，因为它和 Task 依赖表天然兼容：主管产生 Task，执行 Agent 消费 Task，结果回收后由主管继续决策。

图式编排把整个流程建模成 DAG，节点之间可以并行、汇聚、条件分支。这是表达能力最强的方案，也是 LangGraph 主推的方案，但同时也是对持久化要求最苛刻的方案。图式模型里，任何一个节点的执行状态、上下文版本、分支选择都必须持久化，否则重启恢复就是一句空话。

OpenRig 的最终做法是：底层统一用 DAG 表达，上层提供链式和星型的 DSL 封装。用户如果只想写一个顺序流程，直接写链式声明；如果需求复杂，再下沉到 DAG 层手动定义节点和依赖。这个分层让新手写得快，也让复杂场景有表达空间。需要说明的是，这是我在项目实践中为了兼顾易用性和通用性做的取舍，并不是说 DAG 是最优的，只是它作为底层模型最不容易被业务场景卡住。

2.3 持久化优先：为什么 Session 必须成为系统的脊柱

早期版本里，我把 Session 当做一个普通的运行时对象，用内存字典保存，每隔一段时间手动快照一次。结果在一次进程重启后，四个 Agent 协作到一半的场景全部丢失，人工排查花了两个多小时。那次教训之后，我把持久化从“附加功能”改成了“第一优先级”。

具体来说，Session 必须解决三个问题。

第一，上下文不断增长怎么办。Agent 协作过程中会产生大量中间文本，不可能每次全量塞进模型上下文。OpenRig 的做法是分层记忆：原始 Message 全量保存在存储里，但每个 Session 维护一个“压缩后的上下文摘要”，只有摘要和最近 N 条关键 Message 会被送进后续 Agent 的提示词。这个思路和我用 LangChain 做长对话记忆时的做法一脉相承，只是从单 Agent 的 memory 扩展成了多 Agent 的共享记忆。

第二，恢复点怎么定义。不是每条 Message 都值得作为恢复点，OpenRig 只在两类时机写检查点：一个 Task 状态变为 success/failed 的时刻，以及一个 Agent 完成全部工作并返回结果给编排器的时刻。这两个时机都是业务上的稳定节点，记录的上下文也是完整且自洽的，恢复起来不会出现“执行到一半引用了尚未存在的数据”这种问题。

第三，失败之后怎么续跑。恢复逻辑很简单：加载 Session 最近的检查点，找到所有处于 pending 和 in_progress 状态的 Task，把 in_progress 的 Task 重置为 pending（因为它的执行结果未必可靠），然后重新调度。这里有一个隐含的重要约定：Agent 必须被设计成幂等的，也就是说同一个 Task 被重复执行两次，结果不会产生副作用叠加。这一点我会在第 3 章展开讲。

持久化优先的设计让 OpenRig 在真实环境里表现得非常稳。数据库里有全量消息、有任务状态、有检查点，重启、扩容、甚至换一台机器跑调度器，都不会影响协作进度。可以说，Session 就是整个系统的脊柱，其他所有模块都是围绕它长的肉。

3. 关键模块实现：调度、通信、状态恢复，一个都不能少

3.1 消息协议与路由：Agent 之间到底怎么说话

Agent 之间的通信协议，我吃了不少苦头才定下来。最开始的版本里，Agent 之间直接传递 Python 对象，爽是爽，但一旦 Agent 跑在独立进程或者另一台服务器上，这套设计就完全崩了。后来我把协议收敛成 JSON 格式的消息，字段固定，payload 部分兼容任意合法 JSON。

一个典型的 Message 长这样：

{ "message_id": "msg_01HZX...", "session_id": "ses_01HZW...", "sender": "search_agent", "receiver": "writer_agent", "message_type": "task_result", "correlation_id": "task_01HZY...", "timestamp": "2025-01-18T10:23:11Z", "payload": { "status": "success", "data": {"topic": "OpenRig", "materials": ["..."]} } }

几个字段值得单独解释。message_type用来区分是任务结果、请求补充信息、还是错误回报；correlation_id把 Message 和触发它的 Task 关联起来，这一步非常关键，因为没有它，你没法在消息日志里还原“这条结果到底是响应哪个任务的”。payload只装业务数据，不装系统控制字段，避免业务和框架耦在一起。

路由逻辑在实现上其实很轻。每个 Session 内部维护一张“接收者表”，记录当前处于等待状态的 Task 对应的 Agent。当一条 Message 到达时，调度器根据correlation_id找到对应的 Task，再把 payload 写入 Task 的结果字段，最后推进依赖状态。这里完全不需要一个全局的消息中心，因为所有的路由信息都能从持久化的 Task 表里查出来。用 SQL 做路由，是我这个项目里最满意的一个决定，它让消息系统变得极其简单可靠。

3.2 调度引擎与并发控制：多 Agent 怎么扛得住并发

“AI Agent 怎么扛并发”是很多人问我的问题。OpenRig 的做法并不神秘，核心就是三点：一是 Session 级隔离，二是 Task 级并行，三是全局限流。

Session 级隔离的意思是，不同 Session 之间完全不共享可变状态，每个 Session 有自己的上下文、自己的任务列表、自己的执行线程。隔离保证了一个业务方的高负载不会污染另一个业务方的上下文。Task 级并行是整个系统吞吐量的来源：一个 Session 内部如果有多个互不依赖的 Task，调度器会同时执行它们。比如“情报收集”阶段要查三个来源，那就起三个并行 Task，分别交给三个 Agent 实例执行。

实现上我用了一组工作线程池，线程数可以通过配置调整。调度器每次从所有未完成 Session 里，挑出当前可执行的 Task，放进一个优先级队列，然后由固定数量的 worker 消费。这里最重要的细节是：同一个 Session 内可并行的 Task 可以同时执行，但同一个 Session 的状态提交必须串行。我用了每个 Session 一把独立锁，Task 完成后，worker 需要先抢到 Session 锁再更新状态和写检查点。这个设计避免了“两个并行任务同时修改上下文导致覆盖”的经典问题。

全局限流则是为了照顾模型 API 的速率限制。OpenRig 里维护了一个简单的令牌桶，每个 Agent 调模型之前都要申请令牌。令牌桶的速率按模型供应商和模型名分开配置，比如 GPT-4 类慢速模型限流严一点，本地小模型可以放宽。没有这个限流，高并发下最先挂掉的一定不是你的服务，而是上游的模型 API。

3.3 状态快照、检查点与恢复：重启不再是灾难

先看快照的数据结构。每次检查点保存三样东西：Session 的元信息、所有 Task 的状态快照、以及从上一个检查点以来新增的 Message。这三样合起来，足够在任意时刻重建整个协作现场。

恢复的代码逻辑很直接，核心流程如下：

def resume_session(session_id: str) -> Session: session = load_session(session_id) checkpoint = load_latest_checkpoint(session_id) tasks = load_tasks(session_id) # 把执行中的任务重置为待执行，等重新调度 for task in tasks: if task.status == TaskStatus.IN_PROGRESS: task.status = TaskStatus.PENDING task.assignee_id = None save_task(task) # 重建消息索引和路由表 rebuild_message_index(session_id) scheduler.wake_up(session_id) return session

恢复完成后，调度器会像正常流程一样，把所有 PENDING 且依赖已满足的 Task 继续派发给对应 Agent。这里有个容易踩的坑：如果 Agent 本身不是幂等的，重置 IN_PROGRESS 任务会导致同一步被执行两次。比如一个“扣款”Agent 被重复调用，用户就会被扣两遍钱。所以 OpenRig 官方推荐所有 Agent 的副作用操作都走后置补偿或者幂等键。对多数内容生成、检索类 Agent 来说，天然就是幂等的，这一点不太用担心。

另一个值得注意的小细节是检查点的存储频率。写检查点本身有 I/O 成本，频率太高会影响吞吐。我的经验值是：按 Task 粒度写入，不要按 Message 粒度写入。一个 Task 可能会产生 10 条 Message，但只有它完成时才有必要落一次检查点。这样既保证了恢复粒度足够细，又不至于把数据库写入变成瓶颈。

3.4 可观测性与问题定位：多 Agent 系统的救星

多 Agent 系统最大的噩梦不是写不出来，而是出了问题不知道去哪里查。两个 Agent 来回传了几轮信息，最后结果不对，你连是哪一步开始错的都找不到。OpenRig 的可观测性设计是围绕“链路追踪”展开的。

每条 Message、每个 Task、每次 Agent 调用，都会带上同一个trace_id。这个 trace_id 会在 Session 创建时生成，贯穿全程。当业务方来投诉时，我只需要拿到 trace_id，就能把完整链路从数据库里拉出来：先查 Session，再查这个 Session 下所有 Task 的执行顺序，再查每个 Task 消耗的 Message，最后定位到具体是哪一步产生了错误输出。

在实现上，我写了一个简单的查询接口，按时间线输出每个阶段的输入摘要、输出摘要、耗时和 Token 消耗。摘要不会存全文，只存前几百字符和关键统计字段，避免日志表膨胀得太快。早期我遇到过“排查问题要翻几万条原始记录”的情况，加了摘要层之后，90% 的问题一眼就能定位。

还有一个小工具值得做：Agent 输出对比。当同一个 Task 被重试多次时，系统会把每次重试的输入参数和输出结果并排展示。这个功能在排查“为什么 Agent 这次回答和上次不一样”时特别好用，因为多数时候不是随机性问题，而是重试时上下文里混进了新的消息，导致 Agent 决策变了。

4. 实操：用 OpenRig 搭一个“情报收集 + 内容生成”协作流程

4.1 初始化工程与环境

理论讲再多，不如动手跑一遍。下面用一个完整的例子演示 OpenRig 的使用流程：任务是把一个技术主题做成一篇结构化的调研简报。整个过程分三步：先让“情报收集 Agent”搜集素材，然后让“内容规划 Agent”整理大纲，最后让“写作 Agent”生成全文。三个 Agent 之间通过持久化 Session 协作，任何一步失败都可以从断点恢复。

先初始化一个 Python 工程，安装依赖：

mkdir openrig_demo && cd openrig_demo python -m venv venv && source venv/bin/activate pip install openrig fastapi uvicorn "pydantic>=2" sqlalchemy

OpenRig 默认使用 SQLite 起步，零配置，跑通之后再换 PostgreSQL 也不费劲。配置好环境变量后，启动一个最小的 FastAPI 服务作为宿主进程，OpenRig 会把编排调度和 API 服务都挂在这个进程里。

4.2 定义两个角色 Agent

Agent 的定义非常简单，核心就是一个函数加上注册信息。以情报收集 Agent 为例：

from openrig import Agent, register_agent def search_materials(topic: str, max_results: int = 5) -> dict: # 这里是实际检索逻辑，可以是搜索 API、内部知识库或本地文档 results = fake_search(topic, max_results=max_results) return { "topic": topic, "materials": [{"title": r.title, "snippet": r.snippet} for r in results], "count": len(results) } search_agent = Agent( name="search_agent", description="负责检索主题相关的资料、文章和案例，返回结构化素材列表", execute=search_materials, input_schema={"topic": "string", "max_results": "integer"}, output_schema={"topic": "string", "materials": "array", "count": "integer"}, timeout_seconds=60, retry_policy={"max_retries": 3, "backoff": 2.0} ) register_agent(search_agent)

这里面最容易被忽略的字段是description。OpenRig 里的规划 Agent 靠这个描述来决定任务分给谁，描述写得越具体，分派准确率越高。我见过很多人写“负责搜索”就完事了，结果规划 Agent 把写作任务也分给它。描述至少要包含职责边界、擅长和不擅长的内容。

写作 Agent 的写法类似，输入是主题和素材列表，输出是 Markdown 文本。为了让 Agent 幂等，我要求它在开头带上session_id作为标识，同一个 Session 重复生成时直接复用已有草稿，不会产生重复内容。

4.3 创建 Session 并编排任务依赖

接下来是核心的编排逻辑。OpenRig 提供一套声明式的编排接口，用户可以定义一个工作流，指定 Agent 之间的依赖关系：

from openrig import Workflow, SessionManager, task workflow = Workflow("material_report") @task(assignee="search_agent", depends_on=[]) def search_step(ctx): # ctx 里自动带上了 Session 上下文，包括用户输入 topic return search_agent.execute(ctx.inputs["topic"]) @task(assignee="writer_agent", depends_on=["search_step"]) def write_step(ctx): # 这里能直接拿到 search_step 的结果，因为持久化上下文自动注入了 materials = ctx.result("search_step") return writer_agent.execute(ctx.inputs["topic"], materials) workflow.add_task(search_step) workflow.add_task(write_step) manager = SessionManager() session = manager.create_session(workflow_id="material_report", inputs={"topic": "OpenRig 多智能体编排"}) manager.start_session(session.id)

这段代码最值得解释的是ctx.result("search_step")。OpenRig 允许后续 Task 按前置 Task 名称拉取结果，实现上是直接从持久化存储读的，而不是从内存变量读。这意味着即使写作 Agent 被调度到另一台机器上，它依然能拿到前置结果。跨进程、跨机器的数据依赖，就靠这一步打通了。

4.4 运行验证与效果检查

跑完上面的流程，可以检查一下 Session 的最终状态：

openrig-cli session show ses_01HZW...

输出会列出 Session 下每个 Task 的执行状态、耗时、Token 消耗、输入输出摘要。正常情况应该看到两个 Task 都是success，写作结果已经写入了 Session 的最终产物字段。为了确认协作过程确实落库，我习惯直接查一下消息表，看看 search_agent 发送给 writer_agent 的那条 Message 是否包含完整的素材列表。这一步能验证消息路由是否真的把数据传到了正确的地方。

在本地跑通后，这个流程可以用容器镜像部署成常驻服务，FastAPI 会暴露一个简单的 HTTP 接口，外部业务系统通过 POST 请求创建 Session、查询进度、获取结果。OpenRig 本身不绑定特定前端，扣子、微信机器人后端、企业内部工作台，都可以通过这个接口接进来。

4.5 断点恢复与并发压测

演示完正常流程，我再手动模拟一次故障。跑完搜索步骤之后，直接把服务进程 kill 掉，然后重新启动，调用恢复接口：

manager.resume_session(session_id="ses_01HZW...")

观察日志会发现，search_step 的状态从 success 被保留，而尚未执行的 write_step 被重新调度并运行。整个恢复过程在秒级完成，没有丢失搜索结果，也没有重复执行搜索 Agent。这就是“持久化协作系统”和普通脚本最本质的区别：脚本挂了就是挂了，协作系统挂了还能接着干。

压测方面，我模拟了 50 个并发 Session，每个 Session 内部又拆出 3 个并行子任务。在 SQLite 环境下，吞吐量大概到每分钟 300 个 Task 就开始有写入竞争；切到 PostgreSQL 之后，同样配置下每分钟能跑到 1800 个 Task 以上。这个数据说明瓶颈主要在后端存储而不是编排层，生产环境建议直接上 PostgreSQL，连接池调到 20 起步。这部分压测结果是我在自己项目里的实测数据，不同机器上会有差异，但结论方向是一致的：存储先行，别先优化 Agent 调用，那多半不是瓶颈。

5. 常见问题与排查速查

5.1 高频问题实录与解决方案

多 Agent 编排系统在真实环境里的问题，往往集中在这几类：Agent 输出格式失控、任务卡死、上下文串台、恢复后行为不一致。下面是我实战里遇到的高频问题，整理成速查表。

其中 Agent 输出格式失控是最常见也最隐蔽的。模型不是不会输出 JSON，而是它会很自然地“顺带”输出解释文字。我的经验是第一层用正则抽取 JSON 片段，抽不到就调一次“修复 Agent”专门整理格式，交给“修复 Agent”前把原始输出原样保存下来，方便定位是哪里开始坏的。

卡死问题也值得多说一句。OpenRig 默认给每个 Task 设置了超时时间，但单纯超时往往不够，很多任务卡住是因为依赖关系里有隐式的环。比如 A 等 B 的结果，B 又在等 A 的确认，两个 Task 都处于 waiting 状态，谁也不会先动。排查这种问题，我的工具是把所有 waiting 任务的依赖表导出来做一次拓扑排序，有环就是编排定义错了，没有环那就是超时配置太小。

5.2 性能与并发调优：从能跑到跑得快

如果你的多 Agent 系统已经从“能跑”进入“跑得快”阶段，下面的调优经验可以直接抄。

第一，把模型调用和业务逻辑分离。Agent 的函数体内，不要一边调模型一边写数据库。模型调用放在独立的执行器里，方便统计耗时和 Token；业务逻辑放在编排层，方便做事务控制。这个分离对性能排查有决定性的帮助：出问题时，你能立刻分清是模型慢还是代码慢。

第二，上下文压缩要跑在后台。长会话的摘要生成本身也要调模型，如果放在关键路径上，会拖慢整个链路的响应。我的做法是：Task 完成后，先落库、先标记状态、先返回给上层，摘要压缩放到异步队列里慢慢算。上层用户感知到的延迟，就只包括真正必要的 Agent 调用耗时。

第三，高频读取走缓存，状态写入走数据库。Session 的当前状态读取频率很高（每次调度决策都要读一次），但写入频率低。所以我把 Session 的实时状态放在 Redis 里做缓存，数据库只保存可恢复的检查点。注意这里有个一致性代价：缓存和数据库之间可能有秒级延迟，但多 Agent 协作系统天然容忍这种延迟，因为最终一致性足够满足业务需求。实测下来，这个改动把调度器的决策开销从 30ms 降到了 3ms 以内，收益非常明显。

第四，预留一个“降级通道”。业务高峰期模型 API 不稳定，我给 OpenRig 加了一个全局开关：可以把所有 Agent 的执行切换到“模拟模式”，直接返回缓存的历史结果。这样在联调环境和上游故障时，系统依然能完整跑完整个编排流程。这个模式在生产排障时救过我很多次，强烈建议保留。

6. 一些个人体会

最后分享一点个人感受，不算总结，就是些实在话。

多智能体编排这件事，最难的地方其实不是技术，而是克制。我见过很多团队一上来就搭了十几个 Agent，看起来热闹，实际上职责重叠、通信混乱，最后连自己都说不清哪个 Agent 干了什么。OpenRig 的设计让我学会了一件事：Agent 的粒度宁愿偏大一点，也要职责清晰。一个 Agent 从搜索到筛选到总结全都包了，虽然在概念上“不够优雅”，但在排障和调优上的收益，远大于概念上的损失。

另一个体会是：持久化不是“为了不丢数据”这么简单，它实际上是整个系统的调试工具。有了全量消息、任务状态和检查点，你就拥有了多 Agent 系统的黑匣子。我最近处理的一个线上问题，就是靠查 Message 历史定位到是某个 Agent 在第三轮协作时把价格信息写进了错误字段，进而导致下游决策偏差。没有持久化日志，这种问题基本只能靠猜。

如果你正在做自己的 Agent 项目，我的建议是先别急着上编排框架，先把“状态打出来、日志留下来、断点续跑跑通”这三件事做好。这三件事做好了，哪怕你用的是最简单的顺序调用，系统的健壮性也会超过大多数花哨的编排方案。等确实需要并行、需要分工、需要跨进程协作的时候，再引入 OpenRig 这样的编排层，你会发现自己对“系统需要什么”有了远比之前清晰的认识。

最后再分享一个小技巧：给每个 Agent 取名的时候，把它的 Router 路径和职责缩写加进去，比如search_agent_v2。这个小习惯在日志检索的时候能省下大量时间，排查问题靠grep的时候，你才会感谢当初命名的用心。