Hugging Face's logo

LeRobot 文件

環境處理器

Hugging Face's logo
加入 Hugging Face 社群

並獲得增強的文件體驗

開始使用

環境處理器 (Environment Processors)

環境處理器是 LeRobot 資料處理架構中的關鍵層，負責處理特定於環境的轉換，並與策略相關的處理分開。這種職責分離實現了更簡潔的程式碼、更好的模組化，以及針對不同環境和策略進行實驗的便利性。

為什麼需要環境處理器？

當使用不同的機器人環境（LIBERO、MetaWorld、Aloha 等）時，每個環境通常都有獨特的資料格式、座標系和規範，需要在策略處理之前進行標準化。如果沒有環境處理器，這些轉換將會：

  1. 硬編碼在環境程式碼中 - 導致難以對不同的狀態表示進行實驗
  2. 在不同策略間重複 - 每個策略都需要處理特定環境的怪癖
  3. 與策略邏輯混合 - 違反了職責分離原則，並使除錯變得更困難

環境處理器透過在原始環境觀測值和策略輸入之間提供一個專用的處理層，解決了上述問題。

處理管線

以下是評估期間資料在完整處理管線中的流動方式：

# In lerobot_eval.py rollout() function:

# 1. Raw environment observation (numpy arrays, various formats)
raw_observation = env.step(action)

# 2. Convert numpy to torch, normalize images [0,1]
observation = preprocess_observation(raw_observation)

# 3. Add task metadata (for multi-task environments)
observation = add_envs_task(env, observation)

# 4. ENVIRONMENT-SPECIFIC preprocessing (NEW!)
#    - Flatten robot states
#    - Rotate images to match dataset conventions
#    - Handle environment-specific coordinate systems
observation = env_preprocessor(observation)

# 5. POLICY-SPECIFIC preprocessing
#    - Normalize with dataset statistics
#    - Add batch dimensions
#    - Move to GPU
#    - Tokenize language instructions
observation = preprocessor(observation)

# 6. Policy inference
action = policy.select_action(observation)

# 7. POLICY-SPECIFIC postprocessing
#    - Unnormalize actions
#    - Remove batch dimensions
action = postprocessor(action)

# 8. ENVIRONMENT-SPECIFIC postprocessing (NEW!)
#    - Convert action formats if needed
#    - Apply environment-specific constraints
action_transition = {"action": action}
action_transition = env_postprocessor(action_transition)
action = action_transition["action"]

# 9. Execute in environment
env.step(action)

優勢

1. 職責分離

環境處理器處理特定於環境資料格式的轉換，而策略處理器則處理特定於模型需求的轉換。

# ❌ Before: Mixed concerns
class LiberoVLAPolicy:
    def preprocess(self, obs):
        # Environment-specific: Flatten robot state (shouldn't be in policy!)
        state = self._flatten_robot_state(obs["robot_state"])
        # Policy-specific: Normalize with dataset stats
        state = self.normalizer(state)
        return state

# ✅ After: Clear separation
# Environment processor: Handles LIBERO's nested robot state
env_preprocessor = LiberoProcessorStep()  # Flattens robot_state

# Policy processor: Handles model requirements
policy_preprocessor = NormalizerProcessorStep(stats=dataset_stats)

2. 靈活性與可重用性

同一策略可以與不同的環境處理器搭配使用，同樣地，同一環境處理器也可以與不同的策略搭配使用。

# Use SmolVLA policy with LIBERO environment
libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(libero_cfg)
smolvla_preprocessor, smolvla_postprocessor = make_pre_post_processors(smolvla_cfg)

# Or use ACT policy with the same LIBERO environment
libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(libero_cfg)
act_preprocessor, act_postprocessor = make_pre_post_processors(act_cfg)

3. 更容易實驗

想要針對 LIBERO 嘗試不同的狀態表示嗎？只需建立一個新的處理器即可。

# Original: 8D state (pos + quat→axisangle + gripper)
@ProcessorStepRegistry.register("libero_processor")
class LiberoProcessorStep(ObservationProcessorStep):
    def _process_observation(self, obs):
        eef_pos = robot_state["eef"]["pos"]          # 3D
        eef_axisangle = quat2axisangle(quat)         # 3D
        gripper = robot_state["gripper"]["qpos"]     # 2D
        state = torch.cat([eef_pos, eef_axisangle, gripper], dim=-1)  # 8D
        return state

# Experiment: Add velocity for better control
@ProcessorStepRegistry.register("libero_velocity_processor")
class LiberoVelocityProcessorStep(ObservationProcessorStep):
    def _process_observation(self, obs):
        # Include velocities for 14D state
        eef_pos = robot_state["eef"]["pos"]          # 3D
        eef_axisangle = quat2axisangle(quat)         # 3D
        eef_vel = robot_state["eef"]["vel"]          # 3D  (NEW)
        gripper_pos = robot_state["gripper"]["qpos"] # 2D
        gripper_vel = robot_state["gripper"]["qvel"] # 3D  (NEW)
        state = torch.cat([eef_pos, eef_axisangle, eef_vel,
                          gripper_pos, gripper_vel], dim=-1)  # 14D
        return state

4. 更乾淨的環境程式碼

環境公開所有可用資料，而無需知道下游模型將使用哪些資料。

# LIBERO environment exposes full robot state
observation = {
    "pixels": {"image": img, "image2": img2},
    "robot_state": {
        "eef": {"pos": ..., "quat": ..., "vel": ..., "mat": ..., "axisangle": ...},
        "gripper": {"qpos": ..., "qvel": ...},
        "joints": {"pos": ..., "vel": ...}
    }
}

# Environment processor decides what to use
# Policy processor handles model-specific transformations

使用環境處理器

工廠函式 (Factory Function)

make_env_pre_post_processors 函式遵循與策略的 make_pre_post_processors 相同的模式。

from lerobot.envs.factory import make_env_pre_post_processors
from lerobot.envs.configs import LiberoEnv, PushtEnv

# For LIBERO: Returns LiberoProcessorStep in preprocessor
libero_cfg = LiberoEnv(task="libero_spatial", camera_name=["agentview"])
env_preprocessor, env_postprocessor = make_env_pre_post_processors(libero_cfg)

# For other environments: Returns identity processors (no-op)
pusht_cfg = PushtEnv()
env_preprocessor, env_postprocessor = make_env_pre_post_processors(pusht_cfg)

在 envs/factory.py 中的實作

def make_env_pre_post_processors(
    env_cfg: EnvConfig,
) -> tuple[
    PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
    PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
]:
    """
    Create preprocessor and postprocessor pipelines for environment observations.

    Args:
        env_cfg: The configuration of the environment.

    Returns:
        A tuple containing:
            - preprocessor: Pipeline that processes environment observations
            - postprocessor: Pipeline that processes environment outputs
    """
    # For LIBERO environments, add the LiberoProcessorStep to preprocessor
    if isinstance(env_cfg, LiberoEnv) or "libero" in env_cfg.type:
        preprocessor = PolicyProcessorPipeline(steps=[LiberoProcessorStep()])
    else:
        # For all other environments, return an identity preprocessor
        preprocessor = PolicyProcessorPipeline(steps=[])

    # Postprocessor is currently identity for all environments
    # Future: Could add environment-specific action transformations
    postprocessor = PolicyProcessorPipeline(steps=[])

    return preprocessor, postprocessor

評估中的整合

在 lerobot_eval.py 中，環境處理器會被建立一次並全程使用。

def eval_main(cfg: EvalPipelineConfig):
    # Create environment
    envs = make_env(cfg.env, n_envs=cfg.eval.batch_size)

    # Create policy
    policy = make_policy(cfg=cfg.policy, env_cfg=cfg.env)

    # Create policy processors
    preprocessor, postprocessor = make_pre_post_processors(
        policy_cfg=cfg.policy,
        pretrained_path=cfg.policy.pretrained_path,
    )

    # Create environment processors (NEW!)
    env_preprocessor, env_postprocessor = make_env_pre_post_processors(env_cfg=cfg.env)

    # Run evaluation with both processor types
    eval_policy_all(
        envs=envs,
        policy=policy,
        env_preprocessor=env_preprocessor,      # Environment-specific
        env_postprocessor=env_postprocessor,    # Environment-specific
        preprocessor=preprocessor,              # Policy-specific
        postprocessor=postprocessor,            # Policy-specific
        n_episodes=cfg.eval.n_episodes,
    )

範例：LIBERO 環境處理器

LiberoProcessorStep 展示了一個真實的環境處理器。

from lerobot.processor.pipeline import ObservationProcessorStep

@dataclass
@ProcessorStepRegistry.register(name="libero_processor")
class LiberoProcessorStep(ObservationProcessorStep):
    """
    Processes LIBERO observations into the LeRobot format.

    **State Processing:**
    - Extracts end-effector position (3D)
    - Converts quaternion to axis-angle representation (3D)
    - Extracts gripper joint positions (2D)
    - Concatenates into 8D state vector

    **Image Processing:**
    - Rotates images 180° to match HuggingFaceVLA/libero convention
    """

    def _process_observation(self, observation):
        processed_obs = observation.copy()

        # Process images: Flip 180° for camera convention
        for key in list(processed_obs.keys()):
            if key.startswith("observation.images."):
                img = processed_obs[key]
                img = torch.flip(img, dims=[2, 3])  # Flip H and W
                processed_obs[key] = img

        # Process robot_state: Flatten to 8D vector
        if "observation.robot_state" in processed_obs:
            robot_state = processed_obs.pop("observation.robot_state")

            eef_pos = robot_state["eef"]["pos"]           # (B, 3)
            eef_quat = robot_state["eef"]["quat"]         # (B, 4)
            gripper_qpos = robot_state["gripper"]["qpos"] # (B, 2)

            # Convert quaternion to axis-angle
            eef_axisangle = self._quat2axisangle(eef_quat)  # (B, 3)

            # Concatenate into single state vector
            state = torch.cat((eef_pos, eef_axisangle, gripper_qpos), dim=-1)
            state = state.float()

            processed_obs["observation.state"] = state

        return processed_obs

為什麼要進行這些轉換？

  1. 影像旋轉：HuggingFaceVLA/libero 資料集中的影像與原始 LIBERO 模擬器相比旋轉了 180°。處理器負責解決這種慣例不匹配的問題，使得在該資料集上訓練的策略能夠順利運作。

  2. 狀態扁平化：原始 LIBERO 環境公開了巢狀字典，包含所有可用的狀態資訊（位置、四元數、速度、矩陣表示等）。處理器執行以下操作：

    • 選取相關組件（位置、四元數、夾爪狀態）
    • 將四元數轉換為軸角（更適合學習）
    • 扁平化為策略預期的單一 8D 向量

  3. 靈活性：環境仍然公開所有原始資料。如果您想嘗試不同的狀態表示（例如：包含速度、使用矩陣表示而非軸角），您可以建立一個新的處理器，而無需修改環境程式碼。

為新環境新增環境處理器

若要為新環境新增環境處理器：

1. 建立處理器步驟

# In src/lerobot/processor/env_processor.py

@dataclass
@ProcessorStepRegistry.register(name="myenv_processor")
class MyEnvProcessorStep(ObservationProcessorStep):
    """Process observations from MyEnv."""

    def _process_observation(self, observation):
        processed = observation.copy()

        # Your environment-specific transformations
        if "myenv.specific.state" in processed:
            state = processed.pop("myenv.specific.state")
            # Transform to standard format
            processed["observation.state"] = self._transform_state(state)

        return processed

2. 更新工廠

# In src/lerobot/envs/factory.py

def make_env_pre_post_processors(env_cfg: EnvConfig):
    if isinstance(env_cfg, LiberoEnv) or "libero" in env_cfg.type:
        preprocessor = PolicyProcessorPipeline(steps=[LiberoProcessorStep()])
    elif isinstance(env_cfg, MyEnvConfig) or "myenv" in env_cfg.type:
        preprocessor = PolicyProcessorPipeline(steps=[MyEnvProcessorStep()])
    else:
        preprocessor = PolicyProcessorPipeline(steps=[])

    postprocessor = PolicyProcessorPipeline(steps=[])
    return preprocessor, postprocessor

3. 在評估中使用

無需更動！評估指令碼會自動使用適當的處理器。

lerobot-eval \
    --policy.path=lerobot/my_policy \
    --env.type=myenv \  # Automatically uses MyEnvProcessorStep
    --eval.n_episodes=10

未來展望：環境後處理器 (Postprocessors)

目前，後處理器在所有環境中均為恆等處理（無作用）。未來的應用案例包括：

動作空間轉換

@dataclass
class MyEnvActionPostprocessor(ProcessorStep):
    """Convert policy actions to environment-specific format."""

    def __call__(self, transition: EnvTransition) -> EnvTransition:
        action = transition["action"]

        # Example: Convert from Cartesian to joint space
        if self.action_space == "joint":
            action = self.ik_solver(action)

        # Example: Apply environment-specific safety limits
        action = torch.clamp(action, self.min_action, self.max_action)

        transition["action"] = action
        return transition

座標系轉換

@dataclass
class CoordinateTransformPostprocessor(ProcessorStep):
    """Transform actions between coordinate systems."""

    def __call__(self, transition: EnvTransition) -> EnvTransition:
        action = transition["action"]

        # Example: Policy outputs in world frame, env expects base frame
        action = self.world_to_base_transform(action)

        transition["action"] = action
        return transition

最佳實踐

  1. 保持環境處理器簡潔：它們應該只處理特定於環境的資料格式問題，而不是複雜的學習相關轉換。

  2. 使用策略處理器處理模型需求：正規化、批次處理、裝置分配和分詞應歸類於策略處理器中。

  3. 公開環境中的所有資料：讓處理器決定使用哪些資料，而不是在環境中硬編碼選擇。

  4. 文件化規範：清楚地說明您的處理器所處理的任何座標系慣例、相機方位或資料格式。

  5. 獨立測試：環境處理器應該要在無需載入完整策略或環境的情況下進行測試。

總結

環境處理器提供了環境特定資料轉換與策略特定模型需求之間的清晰分離。這種架構：

  • ✅ 能夠輕鬆試驗不同的狀態表示
  • ✅ 允許策略在不同環境間無縫運作
  • ✅ 使環境程式碼專注於模擬/硬體介面
  • ✅ 使處理器管線更具可維護性和可除錯性
  • ✅ 遵循單一職責原則

核心洞見：環境定義資料格式，處理器將其標準化，策略則取用已標準化的資料。 每一層都有明確且專注的職責。

在 GitHub 上更新

←機器人與遠端操作員處理器 動作表示→
© . This site is unofficial and not affiliated with Hugging Face, Inc.