多智能体编排框架 OpenRig：基于 Redis 持久化与状态机的 Agent 协作实践

1. 项目概述

1.1 需求背景与核心痛点

做 AI Agent 相关的项目，从单 Agent 到多 Agent，中间隔着一道很深的沟。单 Agent 跑通容易，让多个 Agent 协作干活才是真正上难度的地方。我在实际开发里遇到的最尖锐的问题不是单个 Agent 的模型调用效果不好，而是多个 Agent 之间完全没有协同感——调用链混乱、状态丢失、任务跑到一半系统一重启就全部归零，甚至两个 Agent 抢同一个资源导致数据错乱。这个项目名字叫 OpenRig，本质上就是一套我自己设计并反复打磨的多智能体编排框架，核心思路是把一个个互相独立的 Agent 当成“离散的个体”，通过编排层把它们组织起来，形成一套可持久化、可恢复、可治理的协作系统。

为什么“持久化”这个关键词这么重要？因为绝大多数跑在多 Agent 上的业务场景都不是一次性的“问一个问题拿一个结果”，而是长时间、跨步骤、有状态的任务流。举个例子，一个智能客服工单系统里，意图识别 Agent 先判断用户想干什么，然后任务分派 Agent 再决定交给哪个子 Agent 执行，子 Agent 又可能要调用外部 API，最后还有一个质检 Agent 把所有记录拉出来复核。这一整个链条可能要跑几分钟甚至更久，任何一个环节断了、挂了、内存被清了，整个流程就当场报废。很多自研的多 Agent 系统，Agent 之间的状态全部放在内存里，进程一崩就全没了，这在生产环境里是不能接受的。

OpenRig 的目标用户很清晰：已经跑通了单 Agent 应用，现在想把多个 Agent 塞进一条业务链路里的开发者或技术团队。它适合你把内部的几个 AI 服务、子流程、外部工具通过编排层组合起来，同时希望整个系统有可靠的持久化机制、任务重试能力和运行时的可观测性。如果你正在做 AI 中台、智能体平台或者企业内部自动化工具链，这个项目的思路能帮你少走很多弯路。

1.2 方案选型与整体架构

系统选型上我最后定下的技术栈是 FastAPI + LangGraph 做流程骨架，Redis 做持久化和任务队列，Agent 的注册与发现机制自己写的，大概 2000 行不到的编排核心。这套组合最讨巧的地方在于每个 Agent 依然是独立的，它可以是你写的函数、一个 LangChain 的链、一个独立部署的微服务，甚至是一段带 prompt 的 HTTP 调用封装。OpenRig 不做“约束每个 Agent 内部怎么实现”这件事，它只解决“Agent 之间怎么说话、怎么传数据、怎么保证不丢任务”这件事。

具体到编排层，OpenRig 用的是编排器模式（Orchestrator）+ 事件总线（Event Bus）的混合结构。纯编排器模式的问题是中心节点的调度压力大、链条一旦拉长就变得僵化；纯事件驱动（编舞模式）的问题则是流程不透明，出了问题不好排查。把两者混合之后，每个环节由编排器决定“下一步该谁上场”，但 Agent 之间会通过事件总线异步发送信号和数据，这样既保留了集中控制的清晰度，又给系统增加了一定的灵活性。

整个系统划分为三层：调度编排层负责把用户请求解析为任务流，调度到对应的 Agent 上；执行层是各个 Agent 的实际逻辑，可以是线程池里的本地函数，也可以是远程调用的 RPC/HTTP 服务；持久化状态层则是整个系统的核心命脉，记录了每一个编排任务的状态、Agent 的执行结果、中间数据、异常信息和重试次数，所有数据落在 Redis 里，定期落盘，支持任务从断点恢复。

2. 核心细节解析与实操要点

2.1 编排模式对比与 OpenRig 的取舍逻辑

在动手写代码之前，先把编排模式这件事想清楚非常关键。目前社区里多 Agent 的组织方式大概有三条路线：第一条是Chain 模式，就是像管道一样，一个 Agent 的输出传给下一个 Agent，简单直接，但扩展性真的很差，稍微改个流程就要改代码，而且链路里任何一个环节出了问题，后面的 Agent 全部白跑；第二条是Router 模式，核心是有一个“路由大脑”，根据任务类型分发给不同 Agent，这种模式在意图清晰、分工明确的场景下效率很高，但路由节点的判定逻辑本身就成了新的瓶颈和故障点；第三条是编排器模式，由一个中心 Orchestrator 统一调度所有 Agent，它能串联、能分流、能并行、能回滚，功能最全，但对状态管理和异常处理的设计要求也最高。

OpenRig 之所以选择混合结构，完全是从实际踩坑中总结出来的。最初的版本用的是纯 Chain 模式，代码虽然好写，但上线后很快发现问题：三个 Agent 串联的时候，中间那个 Agent 需要根据上游结果动态决定走分支一还是分支二，Chain 模式的静态编排根本表达不了这种“运行时动态路由”的需求。后来换成纯编排器模式，分支倒是能走了，但编排器代码疯狂膨胀，而且所有 Agent 都在等中心节点的指令，并发能力直线下降，一度变成整个系统的瓶颈。

最终落地的方案是把编排器设计成**“瘦控制器”**：编排器不直接执行业务逻辑，它只维护一个状态机和调度表，决定每个阶段需要触发哪些 Agent、它们之间的依赖关系是什么，真正的业务数据通过事件总线在 Agent 之间流转。这样做有三个直接的好处：第一，编排器本身逻辑薄，出 bug 的概率低得多；第二，Agent 之间是解耦的，新增一个 Agent 不需要改编排器代码，只需要注册进去；第三，因为所有事件都过总线，天然给系统留出了持久化和审计的空间。

2.2 持久化设计：Redis 机制的深度应用

既然强调持久化，Redis 的具体使用姿势就得掰开讲清楚。很多人在项目里把 Redis 只当缓存用，set 一下 key、get 一下 value，这是非常大的浪费。Redis 其实有两种持久化机制：RDB（快照）和 AOF（追加文件）。RDB 是在指定时间间隔内把内存数据生成快照写入磁盘，恢复速度快，但快照之间的数据会丢；AOF 则是把每个写操作追加到日志文件里，最多丢一秒的数据（取决于同步策略），缺点是对磁盘的写压力大。

在 OpenRig 里，我用的是 Redis 默认开启 RDB、同时按需开启 AOF，并且 AOF 的刷盘策略设置的是everysec，这意味着极端情况下最多丢失 1 秒内的状态变更。对于多 Agent 协作系统来说，这个丢失窗口是完全可接受的，因为真正核心的任务状态在应用层还有冗余备份。Redis 的持久化机制本身是运维层面的兜底，我们不能把全部可靠性押在它身上，因为无论 RDB 还是 AOF，都只能保证进程重启后数据还在，但如果你部署的容器被销毁了、Redis 实例整个没了，那数据还是会丢。所以我在应用层还加了一层独立的持久化策略，这也是 OpenRig 设计里比较重要的一点。

应用层的持久化策略才是 OpenRig 真正花心思的地方。我设计了一个叫“双写”的机制：每个 Agent 任务在执行前和执行后，状态都会先写入 Redis，同时通过异步队列把关键节点写入本地文件系统或备份数据库。这样即使 Redis 挂了，从文件里还能恢复最近一批任务。而且因为 Redis 本身速度很快，双写的性能损耗其实不高，压测的时候基本稳定在百微秒级。用一句话总结我的心得：Redis 的持久化能力是多 Agent 系统的最低保障，而不是全部保障，真正的数据可靠性要由应用层设计来兜底。

2.3 并发控制与任务排队

多 Agent 系统还有个大坑是并发控制。“AI Agent 怎么扛并发”这个问题我之前也找过很多资料，大部分答案都停留在“用异步、用线程池”这个层面，但实际跑起来你会发现真正的瓶颈不是你的代码，而是下游系统的承受能力和共享状态的一致性。

举个例子，如果同时有 50 个用户发起任务，每个任务要拆分成 3 个 Agent 子任务，瞬间会打出 150 个内部请求，如果这些子任务里有一部分要调用同一个上游大模型接口或同一个内部数据库，非常容易把下游打垮。OpenRig 的做法是维护一个全局信号量池：编排器在分发任务之前，先要去信号量池获取一个“令牌”，拿到令牌的 Agent 才能执行，执行完归还令牌。这样即使有再多的任务并发进来，真正同时跑的 Agent 数量是被限制住的。

令牌池的分配可以按优先级来，比如把令牌分为“普通池”和“紧急池”，紧急任务可以插队拿令牌，普通任务排队等。这个设计的精妙之处在于它不动 Agent 内部任何逻辑，只是在编排层做流量整形，对下游系统非常友好。我实测过一个业务场景：不限制并发的时候，50 个用户进来会把内部接口的响应时间从 800 毫秒拉到 5 秒；加上令牌池限流之后，并发控制在 10 个 Agent 同时执行，响应时间稳定在 1.2 秒左右，用户体验反而大幅提升。

3. 实操过程与核心环节实现

3.1 Agent 注册机制的实施步骤

Agent 注册是 OpenRig 的入口，也是最容易被人忽视的部分。很多多 Agent 框架把 Agent 定义和编排逻辑写死在代码里，导致每加一个 Agent 就要发一次版，这在我这里是不允许的。OpenRig 把 Agent 抽象成一个声明式的配置：每个 Agent 注册时提供四样东西——Agent 的唯一标识、执行入口（可以是本地函数名、HTTP 地址或 Docker 容器命令）、输入输出数据格式、以及它关心的消息类型（即它会订阅哪些事件）。注册完成后，编排器就能通过一个注册表来查找和调度 Agent，需要新增 Agent 时只需要往注册表里插入一条记录。

注册表我放在 Redis 里，用 Hash 结构存储，key 是 agent:{agent_id}，field 是 config。这样做的另一个好处是，注册信息可以随时动态更新，不像写在配置文件里那样需要重启进程。举个例子，如果生产环境里某个 Agent 的 API 地址变了，可以直接通过一个管理接口去更新 Redis 里的注册记录，编排器下次调度时自动就会拿到新地址，完全不需要中断服务。

以下是 Agent 注册的核心代码片段，实际项目中我会把这段逻辑封装成一个装饰器：

# agent_registry.py import json import redis redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True) AGENT_REGISTRY_KEY = "openrig:agent_registry" def register_agent(agent_id: str, entrypoint: dict, input_schema: dict, output_schema: dict, topics: list): """注册一个 Agent 到全局注册表""" agent_config = { "id": agent_id, "entrypoint": entrypoint, # {"type": "function", "name": "my_agent_func"} "input_schema": input_schema, # JSON Schema，用于校验输入 "output_schema": output_schema, "topics": topics, "status": "active", "created_at": time.time() } redis_client.hset(AGENT_REGISTRY_KEY, agent_id, json.dumps(agent_config)) return agent_config def get_agent(agent_id: str): raw = redis_client.hget(AGENT_REGISTRY_KEY, agent_id) if not raw: return None return json.loads(raw) def list_agents_by_topic(topic: str): """根据订阅的消息类型找到所有相关 Agent""" all_agents = redis_client.hgetall(AGENT_REGISTRY_KEY) result = [] for agent_id, raw in all_agents.items(): config = json.loads(raw) if topic in config.get("topics", []): result.append({"id": agent_id, "config": config}) return result

这段代码看着简单，但它把“编排器找 Agent”这件事变成了“查表”，完全解耦了调度逻辑与 Agent 实现。你甚至可以把某个 Agent 的动态信息，比如当前负载、最近心跳时间、错误次数，全部塞进这个配置里去。运维和监控人员只需要查 Redis 里的注册表，就能对系统内所有 Agent 的运行状态一目了然。

3.2 持久化状态机的实现细节

多 Agent 系统里最难设计的一个模块，我觉得就是状态机。因为每个任务从创建到结束，可能要经历十几个状态。而且这些状态不是线性走的，有的任务会失败重试，有的任务会等待外部回调，有的任务会中途被人为取消。为了让系统在任何一个节点崩溃后都能恢复，我设计了一套基于 Redis 的持久化状态机结构。

每个编排任务在系统里有一个全局唯一的task_id，所有状态都挂在task:{task_id}这个 key 上。数据结构用 Redis 的 Hash，字段包括status（当前状态）、payload（任务数据）、created_at、updated_at、retry_count、history（历史状态列表）。状态机的流转逻辑放在编排器核心模块里，每个状态变更都是“读当前状态 → 校验合法性 → 写新状态”三步，写入时用 Lua 脚本保证原子性，避免并发情况下两个 Agent 同时把状态写了。

-- state_transition.lua -- 原子地更新任务状态，只有预期的当前状态匹配时才执行更新 if redis.call('hget', KEYS[1], 'status') ~= ARGV[1] then return -1 end redis.call('hset', KEYS[1], 'status', ARGV[2]) redis.call('hset', KEYS[1], 'updated_at', ARGV[3]) redis.call('rpush', KEYS[1] .. ':history', ARGV[1] .. '->' .. ARGV[2]) return 0

用 Lua 脚本做状态流转的好处很简单：Redis 的 Lua 脚本是原子的，天然解决了“并发下状态更新互相覆盖”的问题。很多做多 Agent 系统的同学在这块容易踩坑，以为直接读 Redis 再写 Redis 就行了，但中间隔着网络 IO，两个进程可能读到同一个旧状态，最后后写的人会覆盖先写的人。用 Lua 之后，状态机的每一个步骤都变成了一个原子操作，这是整个系统可靠性的重要基石。

状态机的枚举我建议尽量细化，不要只定义“新建、跑完、失败”三个状态。OpenRig 里有九个状态：PENDING（已创建未调度）、SCHEDULED（已分配 Agent 等待执行）、RUNNING（正在执行）、WAITING（等待外部回调）、SUCCEEDED（成功完成）、FAILED（失败）、RETRYING（重试中）、COMPENSATING（补偿中）、CANCELLED（已取消）。每个状态之间允许的转移关系在代码里有一个专门的矩阵表，任何非法流转都会直接被拒绝并记录日志。状态建得细，后续做监控、做回溯、做补偿都会方便得多。

3.3 多 Agent 协作链路：一个真实场景的完整拆解

为了让上面的理论落地，我把一个真实场景完整拆解出来：做一个自动舆情监控任务，用户提交一个关键词，系统需要去采集平台抓数据、用大模型分析情感倾向、做摘要、最后生成报表发送到指定邮箱。这个任务如果用单个 Agent 一次性完成，prompt 要写得极复杂，而且要区分不同执行阶段很容易乱。用 OpenRig 拆分后，任务流长这样：

第一个环节是collector_agent（采集 Agent），它的职责是接收{keyword: "OpenRig"}的输入，去调用采集平台的 API 拿到一批原始帖子数据；第二个环节是sentiment_agent（情感分析 Agent），它订阅data_collected事件，拿到采集 Agent 的输出结果，调用大模型接口对每条帖子做情感打分；第三个环节是summary_agent（摘要 Agent），它订阅sentiment_analyzed事件，把所有打分结果聚合成一段综合摘要；第四个环节是reporter_agent（报表 Agent），它把摘要渲染成 HTML 或 PDF 发送邮件。每个环节的输出都会持久化到 Redis，同时通过事件总线通知下一个环节。

在 OpenRig 里，这个链路的描述数据是一个 JSON 配置，存到 Redis 中，而不是写死在一堆代码里：

{ "workflow_id": "wf_news_monitor", "name": "舆情监控与报告生成", "trigger": {"event": "user_task_created"}, "steps": [ {"id": "collector", "agent": "collector_agent", "inputs": ["task.payload"]}, {"id": "sentiment", "agent": "sentiment_agent", "inputs": ["collector.output"], "subscribes": ["data_collected"]}, {"id": "summary", "agent": "summary_agent", "inputs": ["sentiment.output"], "subscribes": ["sentiment_analyzed"]}, {"id": "reporter", "agent": "reporter_agent", "inputs": ["summary.output"], "subscribes": ["summary_ready"]} ] }

这个配置本身存储在 Redis 里意味着什么呢？意味着你可以线上动态修改一个工作流的步骤顺序、增删环节，编排器下次调度时直接读取新配置就生效了，不需要重新部署系统。这就是“把离散 Agent 编织成系统”的含义：Agent 是独立的零件，工作流配置是把它们穿起来的线，而 Redis 持久化是让这根线不会因为系统重启而断掉。这边我强烈建议大家把工作流配置和核心逻辑彻底分离，这不仅让系统灵活，排查问题也会轻松很多。

3.4 持久化任务队列：基于 Redis Streams 的可靠消息传递

Team 的 Agent 之间通信用 Redis Streams 来做，这是一个很多人在项目中没用到的特性。Redis Streams 比简单的 Pub/Sub 好在哪里？关键在于Pub/Sub 是即发即弃的，消息发出去了，如果此刻没有任何消费者在监听，消息就丢了；而 Streams 把消息持久化存储在 Redis 里，消费组里的消费者可以按需拉取，处理完再确认。对于多 Agent 协作系统来说，这个“处理完再确认”的能力可以说是刚需。

我在 OpenRig 里定义了一个全局 Stream 叫openrig:events，所有 Agent 之间的事件都写入这个 Stream。每个消费者在处理完事件后，会通过XACK命令确认，这样如果某个消费者在确认前崩溃了，消息还留在 Stream 里，其他消费者（或重启后的同一个人）可以继续消费。这个机制从本质上解决了“任务跑到一半崩溃导致丢失”这个老大难问题。

# event_bus.py import redis, json, time redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True) STREAM_KEY = "openrig:events" GROUP_NAME = "openrig_workers" def publish_event(event_type: str, payload: dict): """往事件总线上发一条消息""" message = { "event_type": event_type, "payload": json.dumps(payload, ensure_ascii=False), "ts": time.time() } redis_client.xadd(STREAM_KEY, message) return message def create_consumer_group(): """创建消费者组，如果已存在则跳过""" try: redis_client.xgroup_create(STREAM_KEY, GROUP_NAME, id="0") except redis.exceptions.ResponseError: pass def consume_events(consumer_name: str, batch_size: int = 10, block_ms: int = 5000): """消费者循环拉取事件""" while True: results = redis_client.xreadgroup( GROUP_NAME, consumer_name, {STREAM_KEY: ">"}, count=batch_size, block=block_ms ) if not results: continue # 没有消息就继续阻塞等待 for stream_name, messages in results: for message_id, fields in messages: event_type = fields.get("event_type") payload = json.loads(fields.get("payload", "{}")) # 将业务处理回调注入进来 try: dispatch_event(event_type, payload) redis_client.xack(STREAM_KEY, GROUP_NAME, message_id) except Exception as exc: # 异常时记录日志，然后转移至死信队列 redis_client.xadd("openrig:dead_letter", {"message_id": message_id, "error": str(exc)})

用 Streams 还有一个额外的好处是回放能力。比如某天排查问题时，你想知道过去 3 个小时系统里到底流转了哪些事件，可以直接从 Stream 里按时间范围捞数据，不用去翻各种日志。多 Agent 系统一旦跑起来，数据流转路径五花八门，想单靠日志定位问题太痛苦了，事件流本身就是一份极佳的审计数据。

4. 常见问题与排查技巧实录

4.1 典型故障案例与解决过程

我把 OpenRig 上线后遇到过的几个最典型的问题整理出来，每个都是真实踩过的坑，也是大家在自己实践时大概率会撞上的。

第一个是“任务卡在 RUNNING 状态死活不走”。现象是任务列表里有几个任务永远停在 RUNNING，既不成功也不失败。排查半天发现是执行 Agent 的进程在处理某个数据时直接崩溃了，但崩溃前还没把状态改成 FAILED，而 Redis Streams 里的消息又已经消费过了。要解决这类问题，必须依靠“超时判断”：编排器启动一个定时检查器，每隔一段时间扫一次状态机，发现某个任务停留在 RUNNING 超过预设阈值（例如 10 分钟），就把它强制置为 FAILED 并触发重试或补偿流程。这个看门狗机制非常管用，没有它，系统会凭白多出大量僵尸任务。

第二个是“事件消息丢失”，现象不是整个任务失败，而是某个下游 Agent 迟迟收不到上游的数据。最后查出来居然是没有启用消费者组的XACK确认机制，消费者把消息读出来了但没确认，Redis 那边也不知道你到底处理完没有。后来把所有消费者的处理逻辑统一改成“先处理业务、再 ACK 消息、出现异常就塞死信队列”，这个问题就彻底消失了。

第三个是“多个消费者重复处理同一个事件”，这是把消息读出来之后并发处理的经典副作用。当时我用的是多线程消费，一个 worker 线程读完消息还没 ACK 又被另一个线程读了一遍，导致下游 Agent 收到了重复指令。解决方式是在应用层维护一个去重表：每条事件生成一个唯一 ID，处理前先查去重表，处理完写进去，下次再收到同样的 ID 直接跳过。去重表也放在 Redis 里，天然共享，多线程、多进程都能查到。

4.2 压测数据与性能调优参考

聊性能调优之前得先说清楚一点：多 Agent 系统的性能瓶颈通常不在编排器本身，而在模型调用和外部 API 的延迟上。OpenRig 在本地压测的时候，编排器本身的调度延迟可以控制在 500 微秒到 2 毫秒之间，但这没有任何参考意义，真正决定用户体验的是上下游 IO。不过编排层的效率会影响系统的最高并发阈值，所以还是值得调一调。

我压测用的配置是 8 核 CPU、16GB 内存的虚拟机，Redis 放在同一内网，模拟 100 个用户同时提交舆情监控任务，每个任务包含 4 个 Agent 串联。压测结果大致如下表：

优化空间主要在两块：第一是Redis 连接池，一次性创建太多连接会让 Redis 报错，后续把连接池上限从默认的 10 提到 50，才把高并发场景下的连接饥饿问题解决掉；第二是事件批处理，把多个事件的 ACK 合并成一次批量提交，能显著降低网络往返次数。如果发现 Redis 的 CPU 使用率飙到 70% 以上，建议先检查事件消息里的 payload 是否过大，比如把大段文本塞进了事件里导致频繁序列化和反序列化。我可以负责任地说，把不必要的数据从事件里拿出来，会是你在多 Agent 系统中做的性价比最高的优化之一。

4.3 状态恢复与补偿策略的实战经验

所谓“持久化协作系统”，最终要验证的就是一件事：进程全挂了，能不能恢复。我做过一次演练：手动 kill 掉编排器进程，重启后观察系统行为。第一次测试的结果非常难看，任务是恢复了几条，但很多后来成功执行完的 Agent 结果没有被正确写入，因为 Agent 执行完后回调的状态更新也丢了。后来我在设计里加了一条“写回重试”规则：Agent 执行完成后，状态更新最多重试 3 次，每次间隔 1 秒，仍然写失败的话就放入本地落盘队列，等编排器进程重启后再补写。经过这么调整之后，恢复成功率基本能达到接近 100%。

另外就是补偿策略，多 Agent 系统里一定会出现“前面几步成功、后面几步失败”的局面。这时候不能让数据留下中间幂等的不一致状态。以舆情监控任务为例，如果摘要 Agent 失败了，但采集 Agent 已经抓了很多平台数据，直接判定整条任务失败，那采集的数据就白抓了，下次重跑要从头再来一遍，浪费大量 API 配额。更好的做法是引入补偿 Agent：当主链路任一步骤失败时，会触发一个补偿流程，把已经采集到的数据和中间结果缓存起来，下次重试时直接从缓存恢复，不需要重跑整个链路。这就是为什么我在状态机里单列了COMPENSATING这个状态。在系统设计的时候给自己留出这种“优雅的失败路径”，会省下来大量的时间和接口成本。

5. 工具选型解析与扩展方向

5.1 为什么是 LangGraph + FastAPI + Redis

很多朋友看到技术栈第一反应是“都用 LangGraph 了，还用得着自己写编排吗？”这里得澄清一个问题：LangGraph 本身解决的是单个 Agent 内部的状态图流转，比如一个带条件分支的复杂 prompt 链，重点是可控性和可视化；而 OpenRig 解决的是跨 Agent 的分布式协作与持久化，重心在 Agent 间通信、任务恢复、并发控制这些系统工程问题上。两者其实是互补的关系，LangGraph 在 OpenRig 里作为一个 Agent 的执行引擎出现，而不是替代 OpenRig 的编排层。

FastAPI 在这里的角色是提供异步能力和一个轻量的 API 网关。多 Agent 系统天然是 IO 密集型的，FastAPI 的 async 特性让请求处理线程不会因为等待 Agent 执行而空转。Redis 则承担了四重职责：Agent 注册表、状态存储、事件总线、持久化介质。一个中间件能同时干四件事，这很大程度上降低了系统的运维复杂度，至少不用同时维护消息队列、状态数据库和缓存三套基础设施。

这套选型的核心逻辑就一句话：能用简单的基础设施解决的，绝不上重型框架。多 Agent 编排的复杂度本身已经够高了，如果每家组件都有各自的学习成本和部署成本，团队协作与排障难度会成倍上升。Redis 作为唯一的中间件，家人上手快、监控好做、持久化机制足够成熟，是我权衡了 Zookeeper、etcd、RabbitMQ 几套方案后的最终选择。

5.2 项目后续的扩展方向和 AI Agent 产品化思考

OpenRig 目前还只是我用来解决企业内部自动化问题的实践项目，但它的设计已经为几种产品化方向留好了接口。第一个方向是接入多模型路由，现在的 Agent 内部直接调用大模型 API，后续可以抽象出一层模型网关，根据任务类型自动选择不同模型，比如长文本摘要走性价比更高的模型，复杂推理走更强力的模型。第二个方向是支持用户自定义工作流，目前工作流配置已经全部 JSON 化，且存在 Redis 里，那么完全可以做一个可视化拖拽界面，让业务人员自己组装 Agent 流程。第三个方向是监控告警体系的完善，目前状态机和事件总线已经天然产生了一套高质量监控数据，后续接上 Prometheus 和 Grafana，能非常直观地看到每个 Agent 的调用量、延迟、失败率。

在 AI Agent 产品化层面，我自己最大的心得是：不要掉进“Agent 万能化”的陷阱。很多做 AI 中台的公司喜欢把所有业务逻辑都交给 Agent 去理解、去执行，结果模型幻觉一出现，整个业务流程就跟着出错。OpenRig 的哲学恰恰相反——能用规则和状态机明确表达的控制流，就用代码表达；把不确定性留给模型真正擅长的地方，比如意图识别、内容生成、理解判断。这个边界划分得越清晰，你的多 Agent 系统就越稳定，越接近生产可用。

6. 收尾：真实环境中沉淀下来的几条心得

这几条经验是 OpenRig 从第一版跑到现在，逐步沉淀下来的，花了大几十个小时的排障时间，应该能帮你避开很多弯路。

第一，先把状态机定义完整，再写任何 Agent 逻辑。很多做多 Agent 项目的朋友上来就让各个 Agent 先跑起来，之后才发现 Agent 之间的衔接混乱、状态无法对齐。正确的顺序一定是先设计状态机和消息类型，把所有流转关系画清楚，让每个 Agent 知道自己什么时候被调用、把结果交给谁、失败时找谁。状态机是这个系统最重要的契约。

第二，事件总线上不要传大对象。我踩过最蠢的一个坑，是有人把 5MB 的 PDF 文件 base64 后塞进事件里，导致 Redis 内存用量直接爆炸。事件消息应该只传“必要的数据索引和轻量的上下文”，真正的大文件、大对象放进对象存储或数据库里，事件里只带一个引用 ID。如果数据量达到一定级别，建议直接用 HTTP 或 RPC 做点对点传输，绕过事件总线。

第三，监控和日志从一开始就要做，不要等项目跑起来再补。多 Agent 系统的故障排查难度随 Agent 数量指数级上升，你根本无从判断问题出在哪个环节，除非每一步都有日志、有事件记录、有状态变化的历史。OpenRig 里每个任务的history字段就是为此而生的，它是全链路调用的“黑匣子”，航向正确与否，只有记录下来才能复盘。

最后想说的是，多智能体编排这件事目前还没有一套标准答案，OpenRig 也只是我和团队在探索过程中的一套可行方案。如果你有更巧妙的设计思路，或者在某些特定业务场景下跑出了更优的架构，非常欢迎拿这套思路去做你自己的版本，折腾起来你会有自己的收获。