ROMA：递归开放元智能体

轻松构建分层高性能多智能体系统！（测试版）

def solve(task):
    if is_atomic(task):                 # 第一步：原子化器
        return execute(task)            # 第二步：执行器
    else:
        subtasks = plan(task)           # 第二步：规划器
        results = []
        for subtask in subtasks:
            results.append(solve(subtask))  # 递归调用
        return aggregate(results)       # 第三步：聚合器

# 入口：
answer = solve(initial_request)

flowchart TB
    A[您的请求] --> B{原子化器}
    B -->|需要规划| C[规划器]
    B -->|原子任务| D[执行器]

    %% 规划器产生子任务
    C --> E[子任务]
    E --> G[聚合器]

    %% 递归
    E -.-> B  

    %% 执行与聚合
    D --> F[最终结果]
    G --> F

    style A fill:#e1f5fe
    style F fill:#c8e6c9
    style B fill:#fff3e0
    style C fill:#ffe0b2
    style D fill:#d1c4e9
    style G fill:#c5cae9

# 使用 uv 安装（速度提升 10-100 倍）
uv pip install roma-dspy

# 或者使用 pip
pip install roma-dspy

# 设置您的 OpenRouter API 密钥（默认使用 Claude Sonnet 4.5 + Gemini 2.5 Flash）
export OPENROUTER_API_KEY="sk-or-v1-..."

# 立即开始解决问题
python -c "from roma_dspy.core.engine.solve import solve; print(solve('2+2 是多少？'))"

# 一键部署（构建 Docker，启动服务）
just setup

# 或者使用特定配置
just setup crypto_agent

# 验证服务是否运行
curl http://localhost:8000/health

# 通过 API 解决任务
just solve "法国的首都是哪里？"

• 🚀 REST API：http://localhost:8000/docs
• 🗄️ PostgreSQL：自动持久化
• 📦 MinIO：S3 兼容存储（http://localhost:9001）
• 📊 MLflow：http://localhost:5000（使用 docker-up-full）

详细信息请参阅 快速入门指南 和 部署指南。

🧠 概念概述

ROMA 的模块层将经典的 DSPy 模式封装为专门构建的组件，这些组件反映了复杂任务执行的生命周期：

1. 原子化器 决定请求是否可以直接处理，或者需要进一步分解。
2. 规划器 将非原子目标分解为有序的子任务图。
3. 执行器 解决各个子任务，必要时通过函数或工具调用来完成。
4. 聚合器 将子任务的输出综合成一个连贯的答案。
5. 验证器（可选）在交付最终结果之前，检查汇总输出是否符合原始目标。

每个模块都采用相同的使用方式：只需传入语言模型（LM）或提供商字符串进行实例化，选择一种预测策略，然后使用任务特定的参数调用 .forward()（或异步的 .aforward()）方法。

所有模块最终都会委托给 roma_dspy.core.signatures 中定义的 DSPy 签名。这样即使内部实现不断演进，接口也能保持稳定。

📦 安装与设置

选项 1：最小化安装（最快——推荐用于评估）

适用场景： 评估 ROMA、开发、测试、快速原型制作

30 秒内即可完成安装：

# 使用 uv（推荐——速度提升 10-100 倍）
uv pip install roma-dspy

# 或者使用 pip
pip install roma-dspy

设置您的 API 密钥：

export OPENROUTER_API_KEY="sk-or-v1-..."  # 推荐
# 或者
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."

立即开始使用：

from roma_dspy.core.engine.solve import solve

# 解决任何任务
result = solve("法国的首都是什么？")
print(result)

包含内容：

• ✅ 所有核心模块（原子化器、规划器、执行器、聚合器、验证器）
• ✅ 所有 DSPy 预测策略
• ✅ 基于文件的存储（无需数据库）
• ✅ 核心工具包（计算器、文件操作）
• ✅ 兼容任何 LLM 提供商

不包含的内容（如需请单独安装）：

• PostgreSQL 持久化 → uv pip install roma-dspy[persistence]
• MLflow 可观测性 → uv pip install roma-dspy[observability]
• E2B 代码执行 → uv pip install roma-dspy[e2b]
• REST API 服务器 → uv pip install roma-dspy[api]
• S3 存储 → uv pip install roma-dspy[s3]
• 所有功能 → uv pip install roma-dspy[all]

方案 2：使用 Docker 的完整安装（生产环境）

适合场景： 生产部署、团队协作、全面可观测性

先决条件：

• Docker 和 Docker Compose
• Python 3.12+（用于本地开发）
• Just 命令运行工具（可选，推荐使用）

一键式设置：

# 交互式设置（会提示配置 E2B、S3 等）
just setup

# 或者指定特定配置文件
just setup crypto_agent

手动启动 Docker：

just docker-up       # 基础版（PostgreSQL + MinIO + API）
just docker-up-full  # 包含 MLflow 可观测性的完整版

环境变量（由 just setup 自动配置）：

# LLM 提供商（必填）
OPENROUTER_API_KEY=...     # 推荐
# 或者
OPENAI_API_KEY=...
ANTHROPIC_API_KEY=...

# 可选：高级功能
E2B_API_KEY=...           # 代码执行
COINGECKO_API_KEY=...     # 加密货币工具包

Docker 的其他功能：

• 📊 PostgreSQL（执行历史、检查点）
• 📈 MLflow（实验跟踪、指标）
• 🌐 REST API（FastAPI，附带文档）
• 📦 MinIO（兼容 S3 的存储）
• 🎨 TUI 可视化界面

方案 3：开发环境安装

适用于贡献或扩展 ROMA：

# 克隆仓库
git clone https://github.com/sentient-agi/roma.git
cd roma

# 安装开发工具（包括 pytest、ruff、mypy）
uv pip install -e ".[dev]"

# 运行测试
just test

# 格式化代码
just format

# 类型检查
just typecheck

对比：哪种安装方式适合您？

⚡ 快速入门：端到端工作流

以下示例模拟了一个典型的编排流程。它使用了三个不同的提供商，展示了各个模块如何轻松地与不同模型和策略协同工作。

import dspy
from roma_dspy import Aggregator, Atomizer, Executor, Planner, Verifier, SubTask
from roma_dspy.types import TaskType

# 执行器可能调用的可选工具
def get_weather(city: str) -> str:
    """返回该城市的固定天气预报"""
    return f"{city} 的天气是晴朗的。"

# 面向 ReAct 的执行器，使用 Fireworks 模型
executor_lm = dspy.LM(
    "fireworks_ai/accounts/fireworks/models/kimi-k2-instruct-0905",
    temperature=0.7,
    cache=True,
)
executor = Executor(
    lm=executor_lm,
    prediction_strategy="react",
    tools=[get_weather],
    context_defaults={"track_usage": True},
)

# 原子化器决定何时分支进入规划阶段
atomizer = Atomizer(
    lm=dspy.LM("openrouter/google/gemini-2.5-flash", temperature=0.6, cache=False),
    prediction_strategy="cot",
    context_defaults={"track_usage": True},
)

# 规划器为非原子目标生成可执行的子任务
planner = Planner(
    lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.85, cache=True),
    prediction_strategy="cot",
    context_defaults={"track_usage": True},
)

aggregator = Aggregator(
    lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.65),
    prediction_strategy="cot",
)

verifier = Verifier(
    lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.0),
)

def run_pipeline(goal: str) -> str:
    atomized = atomizer.forward(goal)
    if atomized.is_atomic 或 atomized.node_type.is_execute:
        execution = executor.forward(goal)
        candidate = execution.output
    else:
        plan = planner.forward(goal)
        results = []
        for idx, subtask in enumerate(plan.subtasks, start=1):
            execution = executor.forward(subtask.goal)
            results.append(
                SubTask(
                    goal=subtask.goal,
                    task_type=subtask.task_type,
                    dependencies=subtask.dependencies,
                )
            )
        aggregated = aggregator.forward(goal, results)
        candidate = aggregated.synthesized_result

    verdict = verifier.forward(goal, candidate)
    if verdict.verdict:
        return candidate
    return f"验证器标记了输出：{verdict.feedback 或 '未返回反馈'}"

print(run_pipeline("计划一个巴塞罗那周末之旅，并附上打包清单。"))

亮点：

• 不同模块可以运行在不同的 LM 和温度下。
• 工具可以在构建时提供，也可以在每次调用时动态传入。
• context_defaults 确保每个 .forward() 调用都会进入带有模块 LM 的正确 dspy.context()。

⚙️ 配置与存储

ROMA-DSPy 使用 OmegaConf 进行分层配置，并结合 Pydantic 进行验证；同时提供 执行范围内的存储，以实现完全的任务隔离。

快速配置示例

from roma_dspy.config import load_config

# 加载特定配置文件并覆盖默认值
config = load_config(
    profile="crypto_agent",
    overrides=["agents.executor.llm.temperature=0.3"]
)

可用配置文件： general、crypto_agent（可通过 just list-profiles 查看列表）

更多信息： 配置指南 提供关于配置文件、代理配置、LLM 设置、工具包配置以及任务感知代理映射的完整说明。

存储

存储是自动且按执行范围管理的——每个任务都会获得一个独立的目录。较大的工具包响应（超过 100KB）会自动保存为 Parquet 文件。

from roma_dspy.core.engine.solve import solve

# 存储路径自动生成：{base_path}/executions/{execution_id}/
result = solve("分析区块链交易")

特点： 执行隔离、兼容 S3、自动 Parquet 存储、Docker 管理

更多信息： 部署指南 提供生产环境存储配置，包括 S3 集成。

🧰 工具包

ROMA-DSPy 包含 9 个内置工具包，用于扩展智能体的能力：

核心工具包: FileToolkit、CalculatorToolkit、E2BToolkit（代码执行） 加密货币工具包: CoinGeckoToolkit、BinanceToolkit、DefiLlamaToolkit、ArkhamToolkit 搜索工具包: SerperToolkit（网页搜索） 通用工具包: MCPToolkit（连接任意 MCP 服务器）

快速配置

agents:
  executor:
    toolkits:
      - class_name: "FileToolkit"
        enabled: true
      - class_name: "E2BToolkit"
        enabled: true
        toolkit_config:
          timeout: 600

参阅: 工具包参考文档，其中包含所有工具的完整说明、配置选项、MCP 集成以及自定义工具包开发指南。

🌐 REST API 和 CLI

ROMA-DSPy 同时提供了 REST API 和命令行界面，适用于生产环境使用。

REST API

基于 FastAPI 的服务器，附带交互式文档：

# 使用 Docker 自动启动
just docker-up

# API 文档：http://localhost:8000/docs
# 健康检查：http://localhost:8000/health

端点: 执行管理、检查点、可视化、指标

CLI

# 本地任务执行
roma-dspy solve "您的任务" --profile general

# 服务器管理
roma-dspy server start
roma-dspy server health

# 执行管理
roma-dspy exec create "任务"
roma-dspy exec status <id> --watch

# 交互式 TUI 可视化（建议配合 MLflow 使用以获得最佳效果）
just viz <execution_id>

# 完整帮助信息
roma-dspy --help

参阅: /docs 端点中的 API 文档，获取完整的 OpenAPI 规范和交互式测试功能。

🏗️ 核心构建模块：BaseModule

所有模块均继承自位于 roma_dspy/core/modules/base_module.py 的 BaseModule。它实现了以下标准化：

• 通过 DSPy 预测策略进行签名绑定；
• 大模型实例化与上下文管理；
• 工具规范化与合并；
• 支持同步/异步调用，并具备安全的关键字过滤机制。

上下文与大模型管理

实例化模块时，您可以提供一个已有的 dspy.LM，也可以让模块根据提供商字符串（model）及可选关键字参数（model_config）自动构建一个。

from roma_dspy import Executor

executor = Executor(
    model="openrouter/openai/gpt-4o-mini",
    model_config={"temperature": 0.5, "cache": True},
)

在内部，BaseModule 确保每次调用 .forward() 时，都会将预测器的调用包裹在以下上下文中：

with dspy.context(lm=self._lm, **context_defaults):
    ...

您可以通过 get_model_config() 检查实际的大模型配置，确认提供商、缓存设置或经过清理的关键字参数。

工具的使用

工具可以以列表、元组或映射的形式提供，这些工具应为 DSPy 的 ReAct/CodeAct 策略所接受的可调用对象。

executor = Executor(tools=[get_weather])
executor.forward("安曼的天气如何？", tools=[another_function])

BaseModule 会自动根据对象身份去重工具，并将构造函数默认值与每次调用的覆盖值合并。

预测策略

ROMA 通过 PredictionStrategy 枚举（位于 roma_dspy/types/prediction_strategy.py）暴露了 DSPy 的策略。您可以直接使用枚举，也可以使用不区分大小写的字符串别名：

from roma_dspy.types import PredictionStrategy

planner = Planner(prediction_strategy=PredictionStrategy.CHAIN_OF_THOUGHT)
executor = Executor(prediction_strategy="react")

可用的策略包括 Predict、ChainOfThought、ReAct、CodeAct、BestOfN、Refine、Parallel、majority 等。对于需要工具的策略（如 ReAct 或 CodeAct），模块会自动应用您传递的工具。

异步执行

每个模块都提供 aforward() 方法。当底层 DSPy 预测器支持异步操作（acall/aforward）时，ROMA 会以异步方式调度；否则，它会优雅地回退到同步实现，同时保持可等待性。

result = await executor.aforward("下载最新的销售报告")

📚 模块参考

⚛️ 分解器

位置: roma_dspy/core/modules/atomizer.py

用途: 判断目标是否为原子级，还是需要规划。

构造函数:

Atomizer(
    prediction_strategy: Union[PredictionStrategy, str] = "ChainOfThought",
    *,
    lm: Optional[dspy.LM] = None,
    model: Optional[str] = None,
    model_config: Optional[Mapping[str, Any]] = None,
    tools: Optional[Sequence|Mapping] = None,
    **strategy_kwargs,
)

输入 (AtomizerSignature):

• goal: str

输出 (AtomizerResponse):

• is_atomic: bool — 任务是否可以直接执行。
• node_type: NodeType — 用于下游路由的提示，取值为 PLAN 或 EXECUTE。

使用示例:

atomized = atomizer.forward("策划一份包含餐厅预订的东京五日行程")
if atomized.is_atomic:
    ...  # 直接发送给执行器
else:
    ...  # 转交给规划器

分解器与具体策略无关，但通常会使用 ChainOfThought 或 Predict 策略。您可以通过 call_params 传递提示信息（例如 max_tokens）：

atomizer.forward(
    "总结这份 PDF 文件",
    call_params={"max_tokens": 200},
)

📋 规划器

位置: roma_dspy/core/modules/planner.py

用途: 将目标分解为有序的子任务，并可选生成依赖关系图。

构造函数: 与分解器相同。

输入 (PlannerSignature):

• goal: str

输出 (PlannerResult):

• subtasks: List[SubTask] — 每个子任务包含目标、任务类型和依赖关系。
• dependencies_graph: Optional[Dict[str, List[str]]] — 当大模型返回时提供的显式邻接矩阵。

使用示例:

plan = planner.forward("在六周内举办一场 B2B 网络研讨会")
for subtask in plan.subtasks:
    print(subtask.goal, subtask.task_type)

SubTask.task_type 是一个遵循 ROMA MECE 框架的枚举类型（检索、写作、思考、代码解释、图像生成）。

⚙️ 执行器

位置: roma_dspy/core/modules/executor.py

用途: 解决原子级目标，可选地通过 DSPy 的 ReAct、CodeAct 策略调用工具或函数。

构造函数: 与前面一致；常用的策略是 ReAct、CodeAct 或 ChainOfThought。

输入 (ExecutorSignature):

• goal: str

输出 (ExecutorResult):

• output: str | Any
• sources: Optional[List[str]] — 来源或引用。

使用示例:

execution = executor.forward(
    "整理一份为期三天的滑雪旅行行李清单",
    config={"temperature": 0.4},  # 每次调用时对大模型的覆盖
)
print(execution.output)

若仅希望在特定调用中启用工具：

execution = executor.forward(
    "巴黎的天气如何？",
    tools=[get_weather],
)

🔀 聚合器

位置: roma_dspy/core/modules/aggregator.py

目的: 将多个子任务的结果合并为最终的叙述或决策。

构造函数: 模式相同。

输入（AggregatorResult 签名）：

• original_goal: str
• subtasks_results: List[SubTask] — 通常是规划器的提案，并附加了执行结果。

输出（AggregatorResult 基类模型）：

• synthesized_result: str

使用方法：

aggregated = aggregator.forward(
    original_goal="计划数据迁移",
    subtasks_results=[
        SubTask(goal="盘点当前数据库", task_type=TaskType.RETRIEVE),
        SubTask(goal="起草迁移时间表", task_type=TaskType.WRITE),
    ],
)
print(aggregated.synthesized_result)

由于它继承自 BaseModule，如果您的聚合策略需要外部调用，仍然可以附加工具（例如知识库检索函数）。

✅ 验证器

位置: roma_dspy/core/modules/verifier.py

目的: 验证合成输出是否满足原始目标。

输入（VerifierSignature）：

• goal: str
• candidate_output: str

输出：

• verdict: bool
• feedback: Optional[str]

使用方法：

verdict = verifier.forward(
    goal="起草符合 GDPR 的隐私政策",
    candidate_output=aggregated.synthesized_result,
)
if not verdict.verdict:
    print("需要修订:", verdict.feedback)

🎯 高级模式

运行时替换模型

使用 replace_lm() 可以在同一个模块中使用不同的语言模型（适用于 A/B 测试或备用方案）。

fast_executor = executor.replace_lm(dspy.LM("openrouter/anthropic/claude-3-haiku"))

每次调用的覆盖

您可以在不重新构建模块的情况下，改变语言模型的行为或提供额外参数。

executor.forward(
    "总结会议记录",
    config={"temperature": 0.1, "max_tokens": 300},
    context={"stop": ["Observation:"]},
)

call_params（或关键字参数）会被过滤，以匹配 DSPy 预测器接受的参数，从而防止意外错误。

仅工具执行

如果您希望实现确定性的工具路由，可以设置一个虚拟语言模型（或极低温模型），并传入纯 Python 可调用对象。

from roma_dspy import Executor

executor = Executor(
    prediction_strategy="code_act",
    lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.0),
    tools={"get_weather": get_weather, "lookup_user": lookup_user},
)

ROMA 将确保构造函数和每次调用中提供的工具都可供策略使用。

🧪 测试

# 运行所有测试
just test

# 运行特定测试
pytest tests/unit/ -v
pytest tests/integration/ -v

参阅: justfile 中包含所有可用的测试命令。

💡 故障排除与提示

• ValueError: Either provide an existing lm — 在构造模块时，请提供 lm= 或 model=。
• Invalid prediction strategy — 检查拼写；字符串不区分大小写，但必须匹配已知别名。
• 缓存 — 在您的语言模型上启用 cache=True，或在 model_config 中设置，以重用之前的完成内容。
• 异步上下文 — 在混合同步和异步调用时，确保事件循环正在运行（例如使用 asyncio.run）。
• 工具重复 — 工具会按身份去重；如果您需要变体，请创建不同的函数。

📖 术语表

核心概念

• DSPy: 斯坦福大学的声明式框架，用于提示、规划和工具集成。
• 预测策略: 驱动推理的 DSPy 类或函数（CoT、ReAct 等）。
• 子任务: 描述分解后工作单元的 Pydantic 模型（goal、task_type、dependencies）。
• 节点类型: Atomizer 选择的是 PLAN 还是 EXECUTE。
• 任务类型: 子任务的 MECE 分类（RETRIEVE、WRITE、THINK、CODE_INTERPRET、IMAGE_GENERATION）。
• 上下文默认值: 每次调用时传递给 dspy.context(...) 的关键字参数。

配置与存储

• FileStorage: 执行范围内的存储管理器，为每次任务执行提供隔离的目录。
• DataStorage: 自动 Parquet 存储系统，用于大型工具包响应（基于阈值）。
• 执行 ID: 每次任务执行的唯一标识符，用于存储隔离。
• 基础路径: 所有存储操作的根目录（本地路径或 S3 存储桶）。
• 配置文件: 命名的配置预设（例如 general、crypto_agent）。
• 配置覆盖: 运行时值，会覆盖配置文件或默认设置。

工具包

• BaseToolkit: 提供存储集成和工具注册的所有工具包的抽象基类。
• REQUIRES_FILE_STORAGE: 元数据标志，表示工具包需要 FileStorage（例如 FileToolkit）。
• 工具包配置: 工具包特有的设置，如 API 密钥、超时时间和阈值。
• 工具选择: 包含/排除列表，用于筛选工具包中可用的工具。
• 存储阈值: 大小限制（KB），超过该限制的响应将以 Parquet 格式存储。

架构

• 执行范围隔离: 每次执行都会获得唯一的存储目录的模式。
• Parquet 集成: 自动列式存储，用于大型结构化数据。
• S3 兼容性: 可通过 Docker 卷挂载使用兼容 S3 的存储。
• 工具注册: 自动发现并注册工具包方法为可调用工具。

祝您构建愉快！如果您扩展或自定义模块，请保持签名一致，以便您的高层编排保持稳定。

附加资源：

• 快速入门指南 - 10 分钟内上手
• 配置指南 - 完整配置参考
• 工具包参考 - 所有内置和自定义工具包
• 部署指南 - 使用 Docker 进行生产部署
• E2B 设置 - 代码执行工具包设置
• 可观测性 - MLflow 跟踪与监控
• 配置系统 - 配置文件及示例

📊 基准测试

我们使用 ROMA 实现的一个简单搜索系统——ROMA-Search，在三个基准测试中进行了评估：SEAL-0、FRAMES 和 SimpleQA。以下是每个基准测试的性能图表。

SEAL-0

SealQA 是一个新的具有挑战性的基准测试，用于评估搜索增强型语言模型在事实查询方面的能力，尤其是在网络搜索产生冲突、噪声或无帮助结果的情况下。

FRAMES

SimpleQA

🧩 基础与渊源

尽管 ROMA 引入了一个实用的开源分层任务执行框架，但它直接建立在两项基础性研究贡献之上，这些贡献首次发表于 WriteHERE：

• 异构递归规划 — ROMA 的整体架构遵循先前关于 异构递归规划 的工作所提出的框架，在该框架中，复杂任务会被递归地分解为子任务图，每个子任务都被赋予一种独特的认知类型。

• 分解中的类型指定 — ROMA 的“三大通用操作”（THINK 🤔、WRITE ✍️、SEARCH 🔍）推广了 分解中的类型指定 假说，该假说指出推理、创作和检索是三种基本的认知类型。

这些贡献在 WriteHERE 仓库和论文中得到了详细描述。通过明确采纳并扩展这一基础，ROMA 提供了一个 可泛化的框架、智能体系统、多功能性和可扩展性，它基于这些洞见，并使其能够被跨领域的开发者所使用。

🙏 致谢

如果没有这些出色的开源贡献，这个框架将无法实现！

• 受 Xiong 等人发表的《超越提纲：异构递归规划》(https://arxiv.org/abs/2503.08275) 中描述的分层规划方法启发。
• Pydantic - 使用 Python 类型注解进行数据验证
• DSPy) - 用于编程 AI 智能体的框架
• E2B - 用于 AI 智能体的云端运行时

📚 引用

如果您在研究中使用了 ROMA 仓库，请引用以下内容：

@misc{alzubi2026romarecursiveopenmetaagent,
      title={ROMA: 面向长周期多智能体系统的递归式开放元智能体框架}, 
      author={Salaheddin Alzu'bi、Baran Nama、Arda Kaz、Anushri Eswaran、Weiyuan Chen、Sarvesh Khetan、Rishab Bala、Tu Vu、Sewoong Oh},
      year={2026},
      eprint={2602.01848},
      archivePrefix={arXiv},
      primaryClass={cs.AI},
      url={https://arxiv.org/abs/2602.01848}, 
}

🌟 星标历史

📄 许可证

本项目采用 Apache 2.0 许可证授权 — 详情请参阅 LICENSE 文件。