添加新机器人#

本指南介绍如何将新的物理机器人或仿真环境接入 RPent 的 LLM-in-the-loop runner。完整的参考实现见 robots/libero/。

接入原则#

  • 复用 RPent 公共抽象。 Env、VLA、运行时和 Memory 优先复用已有组件， 例如 BaseEnvClient、BaseEnvFacade、BaseVLAClient、 BaseVLAFacade 和 MemoryManager。

  • 优先复用 RLinf 的 Env 或 VLA。 如果 RLinf 已有对应实现，RPent 尽量只增加 必要的适配层。

  • 尽量和现有机器人接入方式保持一致。 RobotSpec、Prompt、Toolkit 和运行时尽量参考已有实现，不为单个机器人引入新的公共机制。

  • 无法复用时说明原因。 如果 RPent 或 RLinf 已有相关 Env / VLA 但无法复用， 请在 PR 描述中说明或提交 issue，帮助改进现有抽象。

接入步骤概览#

RPent 的整体进程划分、服务职责和通信方式见 系统说明。 本页不再重复设计原理，只说明接入新机器人需要实现的扩展点。建议按以下顺序完成：

  1. 在 入口 中注册 RobotSpec 和 toolkit 工厂。

  2. 实现 env_client 和 env_server。如需接入 VLA 服务和 model client，参见 添加一个 VLA（或其他基于模型的原语）。

  3. 定义 prompt。

  4. 实现 toolkit 和工具函数。

  5. 注册环境参数并生成 RunConfig。

  6. 实现 runtime 钩子：同一个钩子既能为普通 CLI 初始化完整 runtime，也能为 Dashboard 初始化指定的 component 子集。

  7. 补充 环境、各组件和完整调用链的测试。

入口#

新增名为 myrobot 的机器人时，目录结构如下：

robots/myrobot/
    __init__.py            # 包入口，仅重导出两个工厂
    robot_spec.py          # RobotSpec、工厂、Dashboard 描述和 runtime 钩子
    env_client.py          # MyEnvClient —— agent 侧 RPC client (§1)
    prompt_bundle.py       # system()/user() prompt 工厂              (§2)
    toolkit.py             # Runtime、MyRobotToolkit 和观测处理 (§3)
    tools.py               # 原生工具声明与 handler (§3)
    env_server.py          # 环境侧 facade + RPC 服务                 (§1)
    vla_server.py          # （可选）VLA 模型服务

__init__.py 是机器人包入口，应保持精简，仅重导出 robot_spec.py 中实现的 工厂。rpent/robots/base.py 中的注册表会按需导入 robots.<name>，并调用 这两个函数：

# robots/myrobot/__init__.py
from robots.myrobot.robot_spec import get_robot_spec, get_toolkit

# robots/myrobot/robot_spec.py
from rpent.dashboard.events import DashboardEventSink
from rpent.memory import MemoryManager
from rpent.robots.robot_spec import RobotSpec, RunConfig
from rpent.robots.prompt_bundle import PromptBundle
from rpent.utils.config import get_memory_dir
from robots.myrobot.prompt_bundle import system_prompt, user_prompt

MYROBOT_DASHBOARD_SPEC = {...}

def get_robot_spec() -> RobotSpec:
    return RobotSpec(
        name="myrobot",
        prompts=PromptBundle(system=system_prompt, user=user_prompt),
        add_cli_args=_add_cli_args,
        parse_config=_parse_config,
        init_runtime=_init_runtime,
        dashboard=MYROBOT_DASHBOARD_SPEC,
    )

def get_toolkit(
    *,
    runtime_kwargs,
    dashboard_events: DashboardEventSink,
    config: RunConfig,
):
    from robots.myrobot.toolkit import MyRobotToolkit
    return MyRobotToolkit(
        runtime_kwargs=runtime_kwargs,
        output_dir=config.output_dir,
        dashboard_events=dashboard_events,
        memory=MemoryManager(
            root=config.prompt_vars.get("memory_dir") or get_memory_dir("myrobot"),
        ),
    )

def _add_cli_args(parser, use_dashboard) -> None:
    """向共享 parser 注册机器人参数。见第 4 节。"""
    ...

def _parse_config(args) -> RunConfig:
    """校验最终的 args，返回 RunConfig。见第 4 节。"""
    ...

def _init_runtime(
    args,
    output_dir,
    dashboard_events: DashboardEventSink,
    components: set[str] | None,
):
    """初始化全部 runtime components，或只初始化指定子集。

    返回 (daemons, runtime_kwargs)。见第 5 节。
    """
    ...

dashboard 是可选项；环境不支持 Dashboard 控制时保持为 None。支持时， 在机器人包中定义该 spec：其中 task 描述命令、校验字段、展示模板和输出目录 slug；机器人专用的 Session 设置继续使用普通命令行参数； runtime_components 描述服务行；primitives 按顺序列出 Dashboard 展示并允许直接执行的 Toolkit 动作。相机标签从每步记录的 PNG 工件中自动发现。 任务候选项应直接保存在 spec 中，避免导入机器人包时依赖仿真器包。 完整结构参考 robots/libero/robot_spec.py。

_resolve_robot(name) 通过 importlib.import_module(f"robots.{name}") 动态加载机器人包。因此，只需将机器人包放在 robots/ 下，无需维护中央注册列表。

下文依次说明这些模块需要实现的内容。_add_cli_args 和 _parse_config 见第 4 节，runtime 钩子见第 5 节。Dashboard spec 只由 Dashboard runner 使用。

1. env_client.py + env_server.py#

这两个文件连接 agent 进程与 env_server。client 在 agent 进程内将方法调用 转换成 RPC 请求，env_server 负责处理这些请求。

1.1 Env client（agent 侧）#

继承 rpent.robots.components.env_client_base.BaseEnvClient。它已经负责启动时校验 env.get_env_meta、执行首次 reset、缓存 last_obs，并实现公共的 reset、step、chunk_step、render_camera、 get_camera_meta 和 get_task_language RPC。子类只需增加环境专用方法； 扩展方法需要独立超时时，再扩展超时表。RPC 名称需要保持稳定，因为服务端 facade 会显式注册每个名称。

from rpent.robots.components.env_client_base import BaseEnvClient

class MyEnvClient(BaseEnvClient):
    _TIMEOUT_S = {
        **BaseEnvClient._TIMEOUT_S,
        "env.custom_method": 30.0,
    }

    def custom_method(self, arg):
        return self._client.call(
            "env.custom_method",
            args=(arg,),
            timeout_s=self._TIMEOUT_S["env.custom_method"],
        )

env = MyEnvClient(rpc_client, expected_meta=expected_meta)

1.2 Env server（环境侧）#

在 env_server 中定义与 client API 对应的 facade 类，例如 MyEnvFacade。该类继承 rpent.robots.components.env_facade_base.BaseEnvFacade；基类已提供公共 RPC 路由和 读写分派锁。子类实现公共环境方法，并通过 _register_rpc 增加环境专用路由。 方法接收与 client 一致的位置参数和关键字参数，返回传输层支持的 Python / NumPy 值（不要返回 torch；agent 进程不导入 torch）。

from rpent.robots.components.env_facade_base import BaseEnvFacade

class MyEnvFacade(BaseEnvFacade):
    def __init__(self, env, meta):
        self._env = env
        self._meta = meta
        super().__init__()

    def _register_rpc(self):
        super()._register_rpc()
        # 自定义方法需额外注册
        self._rpc["env.custom_method"] = self.custom_method

    # BaseEnvFacade 要求的抽象方法必须实现
    def get_env_meta(self): ...
    def reset(self): ...
    def step(self, action): ...
    def chunk_step(self, actions, **kwargs): ...
    def get_camera_meta(self, camera_name, **kwargs): ...
    def render_camera(self, camera_name, **kwargs): ...
    def get_task_language(self): ...

    def custom_method(self, arg): ...

facade = MyEnvFacade(env, meta)
facade.serve(transport="http", host=host, port=port)

BaseEnvFacade 通过 _register_rpc 注册公共路由，并使用读写锁串行化会改变 状态的调用。只有确认某个扩展路由可以安全地与其他读操作并发时，才把它加入 _readonly_methods。继承的 RpcFacade.serve 负责绑定传输方式（HTTP 或 socket）、提供 healthz 和 shutdown、检测父进程退出并执行资源清理。

2. prompt_bundle.py#

定义 system_prompt() 和 user_prompt() 两个 prompt 工厂，并在机器人的 robot_spec.py 中构造 PromptBundle(system=system_prompt, user=user_prompt) （见上面的“入口”）。 每个工厂返回一个有序的 dict[str, PromptNode]，其中包含带标题的分节； PromptBundle.render 负责组装和填充。一套 prompt 供 API loop、Claude Code 和 Codex 等 planner 共用。正文使用工具的裸名（如 move_to），并说明 Claude Code 和 Codex SDK 会将其显示为 mcp__rpent__<name>；无需分别维护 CLI 与 API 版本。

# robots/myrobot/prompt_bundle.py
from robots.myrobot.prompts import system as system_parts
from robots.myrobot.prompts import user as user_parts
from rpent.prompt.utils import PromptNode

def system_prompt() -> PromptNode:
    return {
        "INTRO": system_parts.PREAMBLE,
        "GOAL": system_parts.GOAL,
        "RULES": system_parts.RULES,
        "WORKFLOW": system_parts.WORKFLOW,
        "ENVIRONMENT": system_parts.ENVIRONMENT,
        "OUTPUT": system_parts.OUTPUT,
    }

def user_prompt() -> PromptNode:
    return {
        "TASK": user_parts.TASK,
        "BEGIN": user_parts.BEGIN,
    }

将 prompt 内容保存在机器人包内，例如 robots/myrobot/prompts/system.py 和 user.py。分节内容可以是普通字符串，也可以使用 BulletList 或 Numbered。占位符 {{suite}} / {{task}} / {{seed}} / {{output_dir}} / {{recipe_tag}} 在渲染时填充。

3. toolkit.py#

这个模块负责组织 LLM 可以调用的工具，以及这些工具需要的客户端和环境状态。 现有的 LIBERO、RoboCasa 和 RoboTwin 实现将工具函数放在 tools.py， 其余部分放在 toolkit.py，新增机器人时可以沿用这一划分。

通常需要实现以下四部分：

运行时对象（例如 MyRobotRuntime）由 toolkit 持有，保存 EnvClient、 VLA 客户端和本次会话的状态。工具函数通过 ctx.robot 访问它，从而共用同一个 环境、缓存观测和任务进度。

工具定义和处理函数 放在 tools.py 中，用 @tool 声明。函数签名描述 工具参数，Google 风格 docstring 提供工具说明，函数本身负责执行动作并返回 ToolResult。将这些声明收集到 MYROBOT_TOOLS 元组中，就得到了该机器人 提供的工具集合。具体写法见 添加动作原语。

每步状态保存 由 dump_state(runtime, env_state, log) 完成。它通过 env_state.record_step(...) 创建步骤记录，并用 env_state.save(...) 保存观测。在 record_step 块内省略 step 时，文件属于当前步骤；指定 step=<int> 可以保存到其他步骤，step=None 则用于整次运行的文件。 每次保存成功后，文件名会自动加入该 StepRecord 的 artifacts 集合。 再由 build_observation(state, record) 将记录整理成文本数据和图片，供动作 响应与 view_env_state 共用。

Toolkit 类 继承 Toolkit[MyRobotRuntime]，将前面几部分连接起来：

  • 在 __init__ 中构造运行时对象和 EnvState，并将它们连同 memory、 output_dir、tools=MYROBOT_TOOLS 和 dashboard_events 传给基类。 运行时对象使用 robot 参数传入；memory 的访问权限在 MemoryManager 上配置，评测时默认为只读。

  • 初始化环境并保存第 0 步状态。如果支持 Dashboard，同时发布 StepRecordEvent，让页面显示初始观测。

  • 实现 _capture_observation(*, command, result, elapsed_s)，调用上述 状态保存与观测整理函数，返回观测数据和 PNG 图片。Toolkit 会在动作执行后 自动调用它；原始执行结果可通过 result.to_dict() 保存到步骤日志中。

  • 实现 solved()，根据环境状态判断任务是否成功。工具调用、取消和录像收尾 由基类处理；若需要额外的关闭逻辑，在重写 close() 时保留对基类的调用。

runtime_kwargs 由 robot_spec.py:get_toolkit 转发给 toolkit，用于构造 运行时对象。其中通常包含 {"env": MyEnvClient(...), "model": VLAClient(...)} 及其他辅助客户端。

建议遵循的约定#

  • output_dir 是 runner 为单次运行创建的工作目录。环境观测由 EnvState 管理；调用方使用逻辑文件名，不自行拼接存储路径。 transcript 等运行记录与环境文件共享该目录。

  • 工具定义来自带 @tool 的函数，planner 负责将它们转换为各自 SDK 所需的格式。 机器人工具只需实现一次，就能供不同 planner 调用。

  • 环境侧返回传输层支持的 Python / NumPy 数据，不包含 torch 对象。

  • 动作工具执行后，由 toolkit 保存新的状态快照，让下一次 view_env_state 能读到动作后的环境。工具需要录制过程画面时，通过 ctx.record_frame 提交帧。

  • 新增触觉、力等观测模态时，仍通过 dump_state 保存，再由 build_observation 提供给 planner。

工具的参数、返回值和执行约定见 核心接口。

4. _add_cli_args + _parse_config (runner 钩子)#

机器人特有的 CLI 参数通过两个钩子接入 rpent/cli/main.py 的解析流程，并参与 最终的 argparse 解析：

``_add_cli_args(parser, use_dashboard) -> None``。 将机器人参数注册到 main.py 已创建的共享 parser。use_dashboard 决定原本必填的参数是否保持可选。 每个 Dashboard TaskRun 会在 parse_config 调用前，由机器人 Dashboard spec 定义的任务命令提供其声明的字段。main.py 会在 parser.parse_args() 之前调用 该钩子，因此 argparse 的 usage 和错误信息也会包含机器人参数。

``_parse_config(args) -> RunConfig``。 普通 CLI 模式下，该钩子在 parser.parse_args() 后调用；Dashboard 模式下，每个 TaskRun 会先把请求字段 写入任务参数，再调用该钩子。该钩子校验这些字段并返回 RunConfig：

  • recipe_tag —— 单次运行的机器人标签，用于 transcript 文件名和 recipe 路径 （LIBERO 使用 f"{suite.replace('libero_', '')}_t{task}_s{seed}"）。

  • output_dir —— 单次运行的临时目录路径。main.py 随后调用 init_output_dir 创建目录并配置日志。

  • prompt_vars —— 传给 PromptBundle.render 的字典，通常包含运行标识和 prompt 引用的其他变量。

  • task_desc —— 机器人特定的任务标识字典，会原样写入 transcript JSON 记录 （LIBERO 使用 {"suite": ..., "task": ..., "seed": ...}）。

def _add_cli_args(parser, use_dashboard) -> None:
    required = not use_dashboard
    parser.add_argument("--suite", default=None, required=required)
    parser.add_argument("--task", type=int, default=None, required=required)
    # ... 其他机器人参数 ...

def _parse_config(args) -> RunConfig:
    if not args.suite: raise ValueError("--suite is required")
    # ... 生成 recipe_tag、output_dir 和 prompt_vars ...
    return RunConfig(
        recipe_tag=recipe_tag,
        output_dir=output_dir,
        prompt_vars=prompt_vars,
        task_desc={"suite": args.suite, "task": args.task, "seed": args.seed},
    )

5. Runtime 初始化钩子#

init_runtime 返回 (owned_daemons, runtime_kwargs)：

  • owned_daemons: list[ProcessDaemon] 只包含当前进程实际启动的子进程， 当前 runner 会在清理阶段停止它们。连接外部 endpoint 时，不能把外部服务加入 该列表。

  • runtime_kwargs: dict 包含本次运行所需的客户端，通常是 {"env": MyEnvClient(...), "model": VLAClient(...)} 及其他辅助客户端。 Toolkit 使用这些参数构造运行时对象，供工具函数共用。

第四个参数 components 指定要初始化的服务名称。None 表示全部服务，普通 CLI 会传入这个值。Dashboard 根据 dashboard.runtime_components 得到两个子集， 每个 component 都必须显式声明 scope: "shared" 或 scope: "unique"。Dashboard 先初始化一次 shared components，再为每个新的环境实例初始化 unique components。两次都调用同一个钩子，最后合并返回的 runtime_kwargs。在 LIBERO 中，这两个子集分别是 {"vla", "sam3"} 和 {"env"}。

实现应在启动任何服务前拒绝未知 component 名称。如果多个选中的本地服务初始化 较慢，应先全部启动，再依次等待 ready，让初始化过程可以重叠。参考实现见 robots/libero/robot_spec.py 中的有序 component registry。

endpoint（--env-endpoint、--vla-endpoint，以及 LIBERO 的 --sam3-endpoint）解析和环境专用服务命令，应放在拥有对应服务的钩子中。 这些 spawner 应通过 rpent.robots.runtime.try_spawn_server 和 try_wait_server 组合，使各环境的状态事件、就绪失败和 owned daemon 清理保持 一致；runner 不处理这些环境细节。参考模式见 robots/libero/robot_spec.py 和 robots/robocasa/robot_spec.py。

可选的运行结果 finalizer#

RobotSpec.finalize_run 是面向所有机器人、与具体 benchmark 无关的通用运行结束 钩子。RoboCasa 是当前使用方，用它记录单 cell 评测结果，供后续结果统计和聚合。 任何需要发布机器可读评测产物的机器人都可以注册该钩子。其默认值为 None，不会 改变 runner 行为。配置该钩子后，普通终端 runner 会在关闭 toolkit 前读取 toolkit.solved()，完成运行时清理后再把结构化的 RunFinalizationContext 传给 钩子。产物 schema 与文件名由钩子负责，RPent 只定义生命周期边界。

JSON 产物应使用 write_json_atomic，避免中断写入留下不完整结果：

from rpent.evaluation import RunFinalizationContext, write_json_atomic

def _finalize_run(context: RunFinalizationContext):
    return write_json_atomic(
        context.output_dir / "result.json",
        {
            "robot": context.robot_name,
            "task": dict(context.task_desc),
            "success": context.environment_success,
        },
    )

通过 RobotSpec(..., finalize_run=_finalize_run) 注册回调。该钩子目前只用于普通 终端运行，Dashboard 不会调用。benchmark manifest、机器人专用 runtime 字段和 聚合逻辑应继续放在机器人目录中，而不是共享 CLI。

6. 需要补充的测试#

新增机器人时，应先分别测试它使用的每个 runtime 组件，再跑一次完整调用链。 例如，使用 Env、Pi0.5 和 SAM3 的机器人，需要分别提供环境测试、Pi0.5 推理测试、 SAM3 分割测试，以及完整调用链测试。每个组件都应实际调用一次并检查结果， 仅能导入模块或通过服务健康检查还不够。

测试放在哪里#

以下 myrobot 替换为新机器人的包名：

  • tests/unit_tests/robots/myrobot/：离线单元测试，覆盖配置解析、client 参数处理、工具分派，以及 runtime 启动和清理逻辑。用 fake 替代仿真器和模型， 保证可在 CPU 上运行。

  • tests/e2e_tests/myrobot/test_components.py：为每个真实组件分别编写测试， 例如 test_environment_component、test_pi05_component 和 test_sam3_component；只需覆盖该机器人实际使用的组件。

  • tests/e2e_tests/myrobot/test_policy_chain.py：编写一个串起 planner、 toolkit、模型和环境的完整调用链测试。

  • Fixture 放在同目录的 conftest.py，可复用的场景初始化和调用放在 scenario.py。生命周期与断言辅助函数复用 tests/e2e_tests/common.py， 目录组织可参考 tests/e2e_tests/libero/。

每个组件测什么#

  • Env： 启动真实环境，用固定 task/seed 执行 reset 并获取观测，检查所需 相机图像和状态字段的 shape、dtype。至少执行一个合法动作，再检查下一帧观测、 终止信息和成功判定。

  • VLA 或其他动作模型： 加载真实 checkpoint，按该机器人的输入格式传入 观测和指令，执行一次推理。检查返回动作非空、数值有限，且维度符合环境要求。

  • 感知或其他服务： 用已知输入实际调用每个服务，并验证输出。例如让 SAM3 分割图像中的已知物体，检查 mask 尺寸与图像一致且包含前景。

通过受支持的 runtime 接口启动各组件，并检查测试结束后自己启动的 daemon 全部退出。复用已有组件时可以复用其测试，但要说明已有覆盖位置，并补测新增的 输入输出适配。

完整调用链测什么#

组件测试通过后，复用 tests/e2e_tests/common.py 中的 run_scripted_policy_chain，以固定 task/seed 和有限动作数运行公开 CLI。 其中的本地 OfflinePlannerServer 会请求一个真实动作原语，再调用 finish， 无需外部 LLM API。环境和模型服务保持真实，不要 monkeypatch CLI 或 runtime 内部实现。

检查至少执行了一个环境动作、transcript 记录了 finish、states.json 包含无错误的动作记录、生成了预期观测工件，以及自己启动的 daemon 全部退出。 这种有界接入测试不要求任务成功。

离线测试使用 pytest tests/unit_tests/robots/myrobot -v 运行。安装机器人 extra 并准备好所需 GPU、checkpoint 和资产后，使用 pytest tests/e2e_tests/myrobot -v 运行真实组件和调用链测试。干净环境下的 GPU 套件运行方式见 tests/README.md 和 tests/e2e_tests/run_gpu_suite.sh。