策略后端

后端封装一台 VLA 策略服务器，每次推理返回一个动作 chunk。EVA 内置六个，用 policy.type 选择。

概览

四个后端连接外部 VLA 策略服务器，由你自行启动；EVA 只提供客户端那一半。两个无需服务器：mock 产生合成运动，replay 回放已录制的轨迹，二者用于流水线与 3D 视图的冒烟测试。

policy.type连接方式是否需要服务器？用途
openpiWebSocket是OpenPI 兼容的策略服务器。
openpi_rtcWebSocket是OpenPI 服务器，带实时分块（见下文）。
starvlaWebSocket是starVLA 策略服务器。
gr00tZeroMQ是Isaac GR00T 策略服务器。
mock无否用于离线测试的合成运动。
replay无否回放一个已录制的数据集 episode。

协议由后端决定，无需配置。

前置条件

使用需要服务器的后端时：OpenPI / starVLA / GR00T 服务器已在网络可达的机器上运行、检查点已加载，你已知其 IP 与端口。从 configs/01_deploy/<机器人>/ 下的部署预设入手；共享默认值在 configs/00_base/defaults.py。mock 与 replay 无需上述准备。

policy 配置块

每个后端都读取相同的四个顶层键。各后端的细调项放在 backend_options。

键默认值含义
type"openpi"选用哪个后端。
host"127.0.0.1"策略服务器 IP。
port9000策略服务器端口。
backend_options{}各后端的额外设置；可用键取决于 type。

最小块：

configs/01_deploy/<机器人>/your_preset.py
policy = dict(
    type='openpi',             # 选择后端：openpi / openpi_rtc / starvla / gr00t / mock / replay
    host='<策略服务器IP>',      # ← 填入你的策略服务器地址
    port=<端口>,             # ← 填入服务器监听的端口
)
host / port 可能已设好。许多机器人预设把 host/port 放在共享的 _base.py 中，由预设深合并覆盖。自加之前先查 _base.py。

openpi 与 openpi_rtc

两者都连接 OpenPI 兼容服务器。openpi_rtc 用于在更高控制频率下获得更平滑的运动。

openpi

无状态：每次观测发出、动作 chunk 返回。仅需 type、host、port。

openpi_rtc——实时分块

把上一个 chunk 作为提示喂回策略，使前后 chunk 衔接对齐。唯一旋钮是 latency_k：把反馈向前移动多少步以补偿往返延迟。值越大，吸收延迟越多，跟随策略越松。起点 4。

python
policy = dict(
    type='openpi_rtc',
    host='<策略服务器IP>',             # ← 填入你的策略服务器地址
    port=<端口>,                    # ← 填入服务器监听的端口
    backend_options=dict(latency_k=4),  # 先填 4，动作有延迟就调大。
)
RTC 后端 vs RTC 策略。openpi_rtc 是后端（与服务器的传输协议）。它与单独的 rtc 推理策略（chunk 的时间调度，见推理策略）配合，两者各有自己的 latency_k；R1-Lite / ARX 预设两者都设。

starvla

连接 starVLA 服务器。starVLA 接受单张相机图像；camera_key 选择发送哪个（默认第一个）。unnorm_key 指定服务器用以反归一化输出的数据集——其值由训练好的模型决定，向服务器负责人确认。两键均可选。

python
policy = dict(
    type='starvla',
    host='<策略服务器IP>',  # ← 填入你的策略服务器地址
    port=<端口>,         # ← 填入服务器监听的端口
    backend_options=dict(
        camera_key='<相机名>',    # ← 发送哪个 EVA 相机（如 cam_high）
        unnorm_key='<数据集键>',  # ← 来自你的模型，向服务器负责人确认
    ),
)

gr00t

连接 Isaac GR00T 服务器。GR00T 给每份数据贴一个模态键标识字段（相机画面、关节位置、任务文字）。名字因模型而异，必须把 EVA 的相机与状态名映射到服务器期望的键，取值向服务器负责人索取。

backend_options 键默认值含义
video_keys{}把每个 EVA 相机名映射到一个 GR00T video.* 键。
state_key"state.qpos"关节位置向量的键。
language_key"annotation.human.task_description"任务指令文字的键。
action_keys["action.qpos"]拼接成动作 chunk 的键。
timeout_ms15000应答超时（毫秒）。
api_tokenNone可选访问令牌。

默认值覆盖标准 *.qpos 命名；通常只需填 video_keys：

python
policy = dict(
    type='gr00t',
    host='<策略服务器IP>',  # ← 填入你的策略服务器地址
    port=<端口>,         # ← 填入服务器监听的端口
    backend_options=dict(
        video_keys=dict(
            cam_high='<GR00T视频键>',        # ← 把你的相机映射到服务器的键
            cam_left_wrist='<GR00T视频键>',  # ← 每个相机一条
        ),
        # state_key / language_key / action_keys 除非服务器不同，否则保留默认。
    ),
)

mock 与 replay（无需服务器）

两者均不需要策略服务器。

mock

返回与观测无关的平滑合成运动，用于流水线和 3D 视图的冒烟测试，而非真实控制。可选 chunk_size（默认 50）。

replay

replay 需要 dataset 传输：必须与 transport = dict(type='dataset', ...) 搭配，否则 EVA 拒绝启动。dataset 路径与 episode 选择见传输层。

python
transport = dict(
    type='dataset',
    dataset_dir='<录制数据集路径>',  # ← LeRobot v2.1 录制所在的文件夹
    episode_id=0,                  # 回放哪一段录制（0 = 第一段）
)
policy = dict(
    type='replay',
    backend_options=dict(chunk_size=50),  # 每个 chunk 回放的步数，默认即可。
)

运行并验证

按修改后的配置启动控制台：

bash
eva --config configs/01_deploy/<机器人>/your_preset.py --web-port 8080

打开 DEBUG 标签页：

如何添加新的策略后端

策略后端是 PolicyClient（src/policy_client/base.py）的子类，注册进 POLICY_REGISTRY。src/policy_client/ 下的包 __init__ 会自动 import 同级所有模块，因此在该目录下放一个文件即可被选用。

  1. 继承 PolicyClient，实现 infer、reset 与 metadata 属性。
  2. 用 @POLICY_REGISTRY.register_client("<your_name>") 装饰类，并提供 from_config(cls, config, ctx) 类方法 —— 注册表会用解析后的 policy 配置和 PolicyBuildContext（携带 action_dim、chunk_size 等）调用它。
  3. 保存为 src/policy_client/<your_name>.py。src/policy_client/__init__.py 的包扫描会在启动时 import 它，注册即作为副作用完成，无需额外接线。
  4. 在任意配置中选用：policy = dict(type='<your_name>', host=..., port=..., backend_options=dict(...))。
python
# src/policy_client/mybackend.py  ← 新文件；包会自动 import
from __future__ import annotations

import numpy as np

from core.config import ConfigDict
from core.registry import POLICY_REGISTRY
from policy_client.base import PolicyBuildContext, PolicyClient


@POLICY_REGISTRY.register_client("mybackend")   # ← policy.type 中使用的字符串
class MyBackend(PolicyClient):
    def __init__(self, action_dim: int, chunk_size: int = 50) -> None:
        self._action_dim = action_dim
        self._chunk_size = chunk_size
        # ← 在此打开与模型服务的连接

    @classmethod
    def from_config(cls, config: ConfigDict, ctx: PolicyBuildContext) -> MyBackend:
        opts = config.backend_options
        return cls(action_dim=ctx.action_dim, chunk_size=opts.get("chunk_size", 50))

    def infer(self, observation: dict, prev_action: np.ndarray | None = None,
              rtc_params: dict | None = None) -> dict:
        # ← 把 observation 发送给服务，收到形如 [T, action_dim] 的 chunk
        actions = np.zeros((self._chunk_size, self._action_dim), dtype=np.float32)
        return {"actions": actions}

    def reset(self) -> None:
        # ← 清理 episode 级状态
        ...

    @property
    def metadata(self) -> dict:
        return {"server_name": "mybackend", "chunk_size": self._chunk_size}

契约定义在 src/policy_client/base.py：infer 返回 {"actions": ndarray[T, action_dim] float32}；reset 清空 episode 状态；metadata 上报 server_name / chunk_size / action_mode。无服务器最小示例参考 src/policy_client/base.py 中的 RandomPolicyClient；WebSocket 真实后端见 src/policy_client/openpi.py。