1. 从单兵作战到团队协同：OpenRig 到底想解决什么问题

如果你最近半年一直在折腾 AI Agent，大概率经历过这样一个阶段：一开始用单个 Agent 跑通一个任务，感觉惊为天人；接着想让它干点复杂的活，比如同时处理代码审查、文档生成、数据抓取、消息推送，结果发现单个 Agent 的上下文窗口不够用、工具调用互相打架、任务一多就乱套。这时候你自然会想到——能不能搞多个 Agent，各管一摊，互相配合？

想法很美好，现实很骨感。我试过用最朴素的方式做多 Agent 协作：开几个终端窗口，每个窗口跑一个 Agent 进程，然后用文件或者消息队列传递中间结果。跑通 Demo 没问题，但一旦任务链条拉长、Agent 数量增加，问题就全冒出来了。进程挂了怎么恢复？Agent 之间的状态怎么同步？某个 Agent 卡住了怎么排查？任务执行到一半系统重启，之前的状态全丢了怎么办？

OpenRig 这个项目，就是冲着这些痛点来的。它的核心思路可以用一句话概括：把离散的 AI Agent 编织成一个持久化的协作系统。注意这里有两个关键词——“持久化”和“协作系统”。持久化意味着 Agent 的状态、任务队列、执行历史都不会因为进程重启而丢失；协作系统意味着多个 Agent 之间有明确的角色分工、通信机制和调度策略，而不是简单地把几个进程扔到一起。

这个项目适合谁看？如果你已经写过至少一个能跑通的 AI Agent，对工具调用、提示词工程、任务编排有基本概念，现在想往多 Agent 协作方向深入，那 OpenRig 的实践思路会很有参考价值。如果你还在纠结怎么让单个 Agent 稳定输出 JSON，建议先把基础打牢再来看这个。

我接下来会从整体设计思路、核心细节、实操过程、常见问题四个维度，把 OpenRig 这套多智能体编排方案拆开来讲。里面会涉及 tmux 的会话管理、MCP 协议的工具集成、持久化状态存储的设计取舍，以及我在实际搭建过程中踩过的坑。

2. 整体设计与思路拆解：为什么是“编织”而不是“堆叠”

2.1 多 Agent 协作的三种常见架构及其局限

在聊 OpenRig 的具体设计之前，有必要先梳理一下目前多 Agent 协作的主流架构。我把它们大致分为三类，每一类都有各自的适用场景和明显短板。

第一类是中心化调度架构。一个 Orchestrator Agent 负责接收任务、拆解步骤、分发给 Worker Agent，最后汇总结果。这种架构逻辑清晰，适合任务流程相对固定的场景。但问题在于 Orchestrator 容易成为瓶颈——它要维护所有 Worker 的状态，一旦某个 Worker 执行时间过长或者返回异常，整个链条就卡住了。而且 Orchestrator 本身的上下文也会随着任务复杂度增加而爆炸。

第二类是去中心化对等架构。多个 Agent 之间直接通信，没有明确的中心节点。这种架构灵活性高，但调试难度极大。Agent A 给 Agent B 发消息，B 处理完再给 C 发，C 发现需要 A 的额外信息又回头找 A——这种循环依赖在真实场景中非常常见，但排查起来简直是一场噩梦。

第三类是流水线架构。Agent 按照预定义的顺序依次执行，前一个的输出是后一个的输入。这种架构最简单，但灵活性最差，只适合步骤完全固定的批处理任务。

OpenRig 的设计思路和这三类都不太一样。它更像是把多个 Agent 当作独立的“服务节点”，通过一个持久化的消息层把它们连接起来。每个 Agent 有自己的生命周期、状态存储和工具集，Agent 之间的通信通过消息队列或者共享状态存储来完成。这种设计的好处是：单个 Agent 挂掉不会导致整个系统崩溃，重启后可以从持久化状态中恢复；新增 Agent 只需要注册到消息层，不需要修改现有 Agent 的逻辑。

2.2 持久化为什么是多 Agent 系统的生命线

很多人做多 Agent 系统时，第一反应是追求“智能”——怎么让 Agent 更聪明地拆解任务、更准确地调用工具。但实际跑起来你会发现，稳定性比智能更重要。一个不够聪明但稳定运行的 Agent 系统，价值远大于一个偶尔惊艳但经常崩溃的系统。

持久化解决的就是稳定性问题。具体来说，它要保证三件事：

• 状态可恢复：Agent 执行到一半进程挂了，重启后能接着上次的进度继续跑，而不是从头再来。
• 消息不丢失：Agent A 发给 Agent B 的消息，即使 B 当时不在线，也要能在 B 上线后送达。
• 执行可追溯：每个 Agent 做了什么、什么时候做的、结果是什么，都要有完整的日志记录，方便排查问题。

OpenRig 在持久化层面的选择很务实。它没有一上来就搞复杂的分布式数据库，而是根据任务规模提供了分层方案：小规模场景用 SQLite 或者本地文件系统就够了，中等规模可以用 Redis 做消息队列和状态缓存，大规模才需要考虑 PostgreSQL 或者专业的消息中间件。这种渐进式的设计思路很值得借鉴——不要为了“架构先进”而过度设计，先让系统跑起来，再根据实际瓶颈去优化。

2.3 tmux 在 Agent 编排中的角色定位

tmux 这个工具出现在多 Agent 编排的语境里，可能有些人会觉得意外。毕竟 tmux 通常被认为是终端复用工具，和 AI Agent 有什么关系？

但如果你实际搭建过多 Agent 系统，就会发现 tmux 解决了一个非常实际的问题：进程管理和会话隔离。每个 Agent 本质上是一个长期运行的进程，它需要独立的终端会话、独立的输入输出通道、独立的日志文件。用 tmux 来管理这些 Agent 进程，好处非常明显：

• 每个 Agent 跑在独立的 tmux window 或 pane 里，互不干扰。
• 可以通过 tmux 的会话管理功能，批量启动、停止、重启 Agent。
• Agent 的输出可以直接重定向到 tmux 的日志，方便实时查看和事后回溯。
• 即使 SSH 连接断开，tmux 会话中的 Agent 依然在后台运行。

我自己的做法是：给每个 Agent 分配一个 tmux window，window 名称就是 Agent 的角色名，比如coder、reviewer、fetcher。然后用一个管理脚本统一控制这些 window 的启动和停止。这样调试的时候非常直观——直接tmux attach就能看到所有 Agent 的实时输出。

2.4 MCP 协议如何统一工具调用接口

MCP（Model Context Protocol）是最近半年热度飙升的一个概念。简单来说，它定义了一套标准化的协议，让 AI Agent 能够以统一的方式调用外部工具和数据源。在没有 MCP 之前，每个 Agent 框架都有自己的工具定义方式——LangChain 有 Tool 类，Semantic Kernel 有 Plugin，OpenAI 有 Function Calling。你想把一个工具从 A 框架迁移到 B 框架，基本要重写一遍。

MCP 的价值在于解耦。工具提供方只需要按照 MCP 协议暴露接口，Agent 框架只需要按照 MCP 协议调用接口，双方不需要知道对方的具体实现。这就像 USB 接口统一了外设连接标准一样——你不需要为每个外设准备不同的接口，只要都支持 USB 就行。

在 OpenRig 的实践中，MCP 主要用来解决两个问题：一是工具的动态注册和发现，Agent 启动时自动拉取可用的 MCP 工具列表；二是工具调用的标准化，不管底层是 HTTP API、本地脚本还是数据库查询，对 Agent 来说都是统一的 MCP 调用。

注意：MCP 目前还在快速演进中，不同实现之间的兼容性参差不齐。在实际项目中，建议先锁定一个稳定的 MCP 版本，不要盲目追新。

3. 核心细节解析与实操要点：从零搭建一个持久化 Agent 协作系统

3.1 环境准备与基础依赖安装

在开始搭建之前，先把基础环境准备好。我假设你用的是 Linux 或者 macOS，Windows 用户建议用 WSL2。以下是核心依赖清单：

安装命令以 Ubuntu 为例：

sudo apt update sudo apt install -y tmux redis-server sqlite3 python3.11 python3.11-venv

Python 虚拟环境创建：

python3.11 -m venv openrig-env source openrig-env/bin/activate pip install redis aiosqlite mcp httpx

这里我特意没有列具体的 Agent 框架依赖，因为 OpenRig 的设计是框架无关的。你可以用 LangChain、AutoGen、CrewAI，甚至自己手写一个简单的 Agent 循环。关键是 Agent 要能通过 MCP 调用工具，并且能把状态写入持久化存储。

3.2 Agent 角色划分与职责定义

多 Agent 系统的第一步不是写代码，而是定义角色。角色划分得好，后续的通信和调度会顺畅很多；角色划分得不好，Agent 之间会互相推诿或者重复劳动。

我的经验是，按照“输入-处理-输出”的链条来划分角色，每个角色只负责一个明确的职责。以一个代码审查场景为例：

• Fetcher Agent：负责从代码仓库拉取待审查的代码变更，输出结构化的 diff 数据。
• Analyzer Agent：负责分析代码变更，识别潜在问题，输出问题列表。
• Reviewer Agent：负责根据问题列表生成审查意见，输出格式化的审查报告。
• Notifier Agent：负责把审查报告推送到指定的消息渠道。

每个 Agent 的输入和输出都定义为明确的数据结构，通过消息队列传递。这样做的好处是，任何一个 Agent 都可以被替换或升级，只要保持输入输出格式不变，其他 Agent 不需要做任何修改。

实操心得：角色划分时，尽量让每个 Agent 的职责单一。我一开始把“分析”和“审查”合并成一个 Agent，结果发现提示词很难写——既要它客观分析问题，又要它给出主观审查意见，模型经常混淆两种输出格式。拆开之后，两个 Agent 的提示词都变得非常清晰。

3.3 持久化状态存储的表结构设计

持久化存储的设计直接决定了系统的可恢复能力。我用 SQLite 来举例，因为它足够轻量，适合中小规模场景。核心表结构如下：

-- Agent 注册表 CREATE TABLE agents ( id TEXT PRIMARY KEY, role TEXT NOT NULL, status TEXT DEFAULT 'idle', last_heartbeat TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 任务队列表 CREATE TABLE tasks ( id TEXT PRIMARY KEY, parent_task_id TEXT, assigned_agent TEXT, status TEXT DEFAULT 'pending', payload TEXT, result TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (assigned_agent) REFERENCES agents(id) ); -- 消息表 CREATE TABLE messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, sender TEXT NOT NULL, receiver TEXT NOT NULL, content TEXT NOT NULL, status TEXT DEFAULT 'unread', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 执行日志表 CREATE TABLE execution_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, task_id TEXT, action TEXT NOT NULL, detail TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

这几张表的设计有几个关键点：

• tasks表用parent_task_id来支持任务拆解，一个父任务可以拆成多个子任务，子任务的结果汇总后更新父任务。
• messages表用status字段来标记消息是否已读，Agent 启动时先拉取未读消息，处理完再标记为已读。
• execution_logs表记录每个 Agent 的每一步操作，排查问题时非常有用。

注意：SQLite 的并发写入能力有限，如果 Agent 数量超过 10 个，建议换成 PostgreSQL 或者用 Redis 做消息队列。我实测下来，5 个 Agent 以内 SQLite 完全够用，超过 10 个就会出现明显的写入锁等待。

3.4 tmux 会话管理脚本的编写要点

用 tmux 管理 Agent 进程，核心是写一个可靠的启动脚本。以下是我在实际项目中用的脚本框架：

#!/bin/bash SESSION_NAME="openrig" AGENTS=("fetcher" "analyzer" "reviewer" "notifier") # 如果会话已存在，先杀掉 tmux has-session -t $SESSION_NAME 2>/dev/null if [ $? -eq 0 ]; then tmux kill-session -t $SESSION_NAME fi # 创建新会话 tmux new-session -d -s $SESSION_NAME -n "manager" # 为每个 Agent 创建独立的 window for agent in "${AGENTS[@]}"; do tmux new-window -t $SESSION_NAME -n "$agent" tmux send-keys -t "$SESSION_NAME:$agent" \ "source openrig-env/bin/activate && python -m openrig.agent --role $agent" C-m done # 切回 manager window tmux select-window -t "$SESSION_NAME:manager" echo "OpenRig session started with ${#AGENTS[@]} agents."

这个脚本的关键点：

• 每次启动前先清理旧会话，避免残留进程干扰。
• 每个 Agent 跑在独立的 window 里，日志互不干扰。
• 用send-keys发送启动命令，而不是直接在 tmux 外面启动进程，这样 Agent 的输出会留在 tmux 的缓冲区里，方便回溯。
• 最后切回 manager window，方便统一管理。

实操心得：建议在 manager window 里跑一个监控脚本，定期检查各个 Agent 的心跳。如果某个 Agent 超过一定时间没有更新心跳，就自动重启对应的 window。这个机制能大幅提升系统的自愈能力。

3.5 MCP 工具注册与调用的实现细节

MCP 工具调用的核心流程分为三步：工具注册、工具发现、工具调用。

工具注册是工具提供方要做的事。假设你有一个查询天气的工具，按照 MCP 协议暴露出来：

from mcp.server import Server from mcp.types import Tool, TextContent server = Server("weather-tool") @server.list_tools() async def list_tools(): return [ Tool( name="get_weather", description="查询指定城市的天气", inputSchema={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"} }, "required": ["city"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): if name == "get_weather": city = arguments["city"] # 实际查询逻辑 result = f"{city}今天晴，气温 22-28 度" return [TextContent(type="text", text=result)]

工具发现是 Agent 启动时要做的事。Agent 连接到 MCP Server，拉取可用的工具列表，然后把这些工具转换成自己框架能识别的格式。以 LangChain 为例：

from mcp import ClientSession from langchain.tools import StructuredTool async def load_mcp_tools(session: ClientSession): tools_response = await session.list_tools() langchain_tools = [] for tool in tools_response.tools: langchain_tools.append( StructuredTool.from_function( func=lambda **kwargs: session.call_tool(tool.name, kwargs), name=tool.name, description=tool.description, args_schema=tool.inputSchema ) ) return langchain_tools

工具调用就是 Agent 在推理过程中决定调用某个工具，然后通过 MCP 协议发起调用，拿到结果后继续推理。

这套流程的好处是，工具的实现和 Agent 的实现完全解耦。你可以随时新增工具、修改工具逻辑，只要 MCP 接口不变，Agent 端不需要做任何改动。

4. 实操过程与核心环节实现：一个完整的代码审查协作系统

4.1 任务定义与拆解逻辑

假设我们要搭建一个自动代码审查系统，输入是一个 Git 仓库的 PR 链接，输出是一份结构化的审查报告。任务拆解如下：

1. Fetcher Agent 拉取 PR 的 diff 内容。
2. Analyzer Agent 分析 diff，识别出变更的文件、函数、潜在风险点。
3. Reviewer Agent 根据分析结果生成审查意见。
4. Notifier Agent 把审查意见推送到指定渠道。

每个步骤的输出都写入tasks表，下一个 Agent 从表中读取上一个步骤的输出作为输入。这种设计的好处是，任何一个步骤失败，都可以单独重试，不需要从头开始。

4.2 Fetcher Agent 的实现与持久化写入

Fetcher Agent 的核心逻辑是调用 Git 命令拉取 diff，然后把结果写入任务表。

import subprocess import json import aiosqlite async def fetch_pr_diff(pr_url: str, db_path: str): # 拉取 diff result = subprocess.run( ["git", "fetch", "origin", "pull/123/head"], capture_output=True, text=True ) diff = subprocess.run( ["git", "diff", "main...FETCH_HEAD"], capture_output=True, text=True ).stdout # 写入任务表 async with aiosqlite.connect(db_path) as db: await db.execute( """INSERT INTO tasks (id, assigned_agent, status, payload, result) VALUES (?, ?, ?, ?, ?)""", ("task-001", "analyzer", "pending", json.dumps({"pr_url": pr_url}), json.dumps({"diff": diff})) ) await db.commit() return diff

这里的关键点是：Fetcher Agent 完成拉取后，不是直接把 diff 传给 Analyzer，而是写入数据库，然后更新任务状态。Analyzer 启动时从数据库读取待处理任务。这样做的好处是，即使 Analyzer 当时不在线，任务也不会丢失。

4.3 Analyzer Agent 的推理与工具调用

Analyzer Agent 需要调用大模型来分析 diff 内容。它的输入是 diff 文本，输出是结构化的分析结果。

async def analyze_diff(diff: str, llm_client): prompt = f"""分析以下代码变更，识别出： 1. 变更涉及的文件和函数 2. 潜在的风险点（如空指针、资源泄漏、并发问题） 3. 建议的审查重点 代码变更： {diff} 请以 JSON 格式输出，包含 files、risks、focus_points 三个字段。""" response = await llm_client.chat(prompt) analysis = json.loads(response) # 写入任务表 async with aiosqlite.connect(db_path) as db: await db.execute( """UPDATE tasks SET status = 'completed', result = ? WHERE id = 'task-001'""", (json.dumps(analysis),) ) await db.execute( """INSERT INTO tasks (id, assigned_agent, status, payload) VALUES (?, ?, ?, ?)""", ("task-002", "reviewer", "pending", json.dumps(analysis)) ) await db.commit() return analysis

注意：大模型的输出格式稳定性是个大问题。我试过直接用json.loads解析模型输出，结果经常因为模型多输出了一段解释文字而报错。后来改成先用正则提取 JSON 块，再做解析，稳定性提升了很多。

4.4 Reviewer Agent 的意见生成与格式化

Reviewer Agent 拿到 Analyzer 的分析结果后，生成人类可读的审查意见。

async def generate_review(analysis: dict, llm_client): prompt = f"""根据以下代码分析结果，生成一份审查报告。 要求： - 按文件分组 - 每个问题标注严重程度（高/中/低） - 给出具体的修改建议 分析结果： {json.dumps(analysis, ensure_ascii=False, indent=2)} 请以 Markdown 格式输出。""" review = await llm_client.chat(prompt) async with aiosqlite.connect(db_path) as db: await db.execute( """UPDATE tasks SET status = 'completed', result = ? WHERE id = 'task-002'""", (review,) ) await db.execute( """INSERT INTO tasks (id, assigned_agent, status, payload) VALUES (?, ?, ?, ?)""", ("task-003", "notifier", "pending", review) ) await db.commit() return review

4.5 Notifier Agent 的消息推送与状态回写

Notifier Agent 负责把审查报告推送到指定渠道，比如企业微信、钉钉或者邮件。

import httpx async def push_notification(review: str, webhook_url: str): async with httpx.AsyncClient() as client: response = await client.post( webhook_url, json={"msgtype": "markdown", "markdown": {"content": review}} ) response.raise_for_status() async with aiosqlite.connect(db_path) as db: await db.execute( """UPDATE tasks SET status = 'completed' WHERE id = 'task-003'""" ) await db.commit()

4.6 完整流程的启动与监控

把所有 Agent 启动起来，整个流程就自动运转了。启动脚本如下：

#!/bin/bash SESSION="openrig-review" tmux new-session -d -s $SESSION -n "manager" tmux new-window -t $SESSION -n "fetcher" tmux new-window -t $SESSION -n "analyzer" tmux new-window -t $SESSION -n "reviewer" tmux new-window -t $SESSION -n "notifier" tmux send-keys -t "$SESSION:fetcher" \ "python -m openrig.agents.fetcher" C-m tmux send-keys -t "$SESSION:analyzer" \ "python -m openrig.agents.analyzer" C-m tmux send-keys -t "$SESSION:reviewer" \ "python -m openrig.agents.reviewer" C-m tmux send-keys -t "$SESSION:notifier" \ "python -m openrig.agents.notifier" C-m echo "Review system started."

监控方面，我在 manager window 里跑一个简单的轮询脚本，每 30 秒检查一次各 Agent 的心跳和任务队列长度。如果某个 Agent 超过 2 分钟没有心跳，就自动重启对应的 window。

import asyncio import aiosqlite from datetime import datetime, timedelta async def monitor(db_path: str): while True: async with aiosqlite.connect(db_path) as db: async with db.execute( "SELECT id, role, last_heartbeat FROM agents" ) as cursor: async for row in cursor: agent_id, role, heartbeat = row if heartbeat: last = datetime.fromisoformat(heartbeat) if datetime.now() - last > timedelta(minutes=2): print(f"Agent {role} ({agent_id}) is stale, restarting...") # 触发重启逻辑 await asyncio.sleep(30)

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

5.1 Agent 启动后无法连接到消息队列

这是最常见的问题之一。表现是 Agent 进程启动了，但一直卡在“等待消息”状态，任务队列里的任务没有被消费。

排查思路：

1. 检查 Redis 或者 SQLite 的连接配置是否正确。特别是 Redis 的 host 和 port，很容易因为环境变量没设置对而连到默认的 localhost。
2. 检查 Agent 是否成功注册到了agents表。如果注册失败，任务分配逻辑就找不到这个 Agent。
3. 检查消息队列的 key 或者表名是否一致。我遇到过因为两个 Agent 用了不同的队列名称，导致消息发到了“空队列”里。

避坑技巧：在 Agent 启动时加一行日志，打印它监听的队列名称和连接地址。这样排查时一眼就能看出配置是否一致。

5.2 任务执行到一半卡住，没有报错也没有进展

这种情况通常是因为 Agent 在等待一个永远不会到达的消息，或者大模型调用超时但没有设置超时时间。

排查步骤：

我的做法是给每个 Agent 的关键操作都加上超时控制。比如模型调用设置 60 秒超时，数据库写入设置 10 秒超时。超时后记录日志并进入重试或者失败状态，而不是无限等待。

5.3 多个 Agent 同时写入数据库导致锁冲突

SQLite 的写入锁是数据库级别的，同一时间只能有一个写入操作。如果多个 Agent 同时尝试写入，就会出现database is locked错误。

解决方案有三个层次：

• 最简单：在写入时增加重试逻辑，捕获database is locked异常后等待随机时间再重试。
• 中等规模：把写入操作集中到一个专门的 Writer Agent，其他 Agent 通过消息队列把写入请求发给它。
• 大规模：换成 PostgreSQL，利用行级锁和更好的并发控制。

我实测下来，5 个 Agent 以内用重试逻辑就够了，超过 10 个建议直接上 PostgreSQL。

5.4 MCP 工具调用返回格式不一致

不同 MCP Server 返回的数据格式可能不一样。有的返回纯文本，有的返回 JSON 字符串，有的返回结构化对象。Agent 在解析时如果假设了固定格式，很容易出错。

我的处理方式是在 Agent 和 MCP Server 之间加一层适配器，统一把返回值转换成 Agent 期望的格式。适配器的逻辑很简单：

def normalize_mcp_result(result): if isinstance(result, str): try: return json.loads(result) except json.JSONDecodeError: return {"text": result} elif isinstance(result, list): return {"items": result} elif isinstance(result, dict): return result else: return {"raw": str(result)}

5.5 Agent 重启后状态丢失或重复执行

这个问题通常是因为状态更新和任务执行不是原子操作。比如 Agent 执行完任务后，先更新了任务状态，然后进程挂了，重启后从数据库读取状态发现任务已完成，但实际上结果没有写入。

解决方案是引入“执行中”状态和幂等性设计：

• 任务状态增加executing中间态。Agent 开始执行时先把状态改为executing，执行完成后再改为completed。
• 重启后检查executing状态的任务，根据执行日志判断是重新执行还是继续等待。
• 关键操作设计成幂等的，比如写入结果时用INSERT OR REPLACE而不是INSERT。

实操心得：我一开始没有设计executing状态，结果 Agent 重启后经常出现任务重复执行的情况。加上这个中间态之后，配合执行日志，基本可以做到精确恢复。

6. 性能调优与扩展方向

6.1 Agent 并发数的合理设置

Agent 数量不是越多越好。每个 Agent 都要消耗内存和 API 调用配额，而且 Agent 之间的通信开销会随着数量增加而上升。

我的经验值是：对于 IO 密集型的 Agent（比如调用模型 API、查询数据库），并发数可以设置为 CPU 核心数的 2-4 倍；对于 CPU 密集型的 Agent（比如本地代码分析），并发数不要超过 CPU 核心数。

具体到 OpenRig 的场景，我一般会这样配置：

• Fetcher Agent：2-3 个，因为 Git 操作相对较快。
• Analyzer Agent：3-5 个，因为模型调用是主要瓶颈。
• Reviewer Agent：2-3 个，模型调用为主。
• Notifier Agent：1-2 个，推送操作很快。

6.2 消息队列的选型对比

我个人的建议是：先用 SQLite 把系统跑通，遇到性能瓶颈再换 Redis。不要一上来就搞复杂的消息中间件，那样调试成本太高。

6.3 从单机到分布式的演进路径

OpenRig 目前的设计是单机多进程，所有 Agent 跑在同一台机器上。如果任务量继续增长，可以考虑往分布式方向演进：

• 第一步：把数据库换成 PostgreSQL，支持多机访问。
• 第二步：把消息队列换成 Redis 或者 RabbitMQ，支持跨机通信。
• 第三步：引入服务发现机制，Agent 可以动态注册和发现彼此。
• 第四步：增加任务调度层，根据各机器的负载情况分配任务。

但我要提醒一句：分布式系统的复杂度是指数级上升的。在单机能够支撑的情况下，不要轻易往分布式走。我见过太多项目在单机阶段就问题百出，却急着上分布式，结果调试成本翻了好几倍。

6.4 监控与告警的必备指标

一个稳定的多 Agent 系统，至少需要监控以下指标：

• Agent 心跳：每个 Agent 是否在正常运行。
• 任务队列长度：待处理任务是否堆积。
• 任务平均执行时间：是否有任务执行异常缓慢。
• 模型调用成功率：API 调用是否稳定。
• 数据库写入延迟：持久化层是否成为瓶颈。

这些指标可以先用简单的日志和定时脚本采集，等系统稳定后再考虑接入专业的监控平台。

6.5 后续可以扩展的方向

OpenRig 这套框架搭好之后，可以往几个方向扩展：

• 增加 Agent 类型：比如专门做测试的 Tester Agent、专门做部署的 Deployer Agent。
• 引入优先级队列：紧急任务优先处理，普通任务排队等待。
• 支持人工介入：某些关键决策点暂停，等待人工确认后再继续。
• 增加学习机制：记录每次任务执行的结果，逐步优化任务拆解和 Agent 调度策略。

我在实际使用中体会最深的一点是：多 Agent 系统的价值不在于“智能”，而在于“可靠”。一个能够稳定运行、故障可恢复、状态可追溯的系统，比一个偶尔惊艳但经常崩溃的系统有用得多。OpenRig 这套思路的核心，就是把工程化的可靠性原则应用到 AI Agent 的编排中，让多个 Agent 真正像一个团队一样协作，而不是各自为战。