LeRobot documentation

环境处理器

Hugging Face's logo
Join the Hugging Face community

and get access to the augmented documentation experience

to get started

环境处理器

环境处理器在策略自身的处理器之前转换环境的观测值,并在环境之前处理策略的动作。每个环境配置都会通过 get_env_processors() 返回其对应的处理器对,而 make_env_pre_post_processors 会为评估、训练期间的评估以及仿真机器人构建这些处理器。

模拟环境无需

模拟环境到达策略时已经是规范化的:图像为 observation.images.<camera>(或 observation.image),状态为一个扁平的 observation.state,每个模拟器的图像方向和状态组合都由其 src/lerobot/sim/integrations/<simulator>.py 集成统一处理,无论是评估、rollout 还是录制,本地运行还是远程服务(参见 模拟:本地或远程服务)。以 LIBERO 为例,其观测适配器会翻转两个相机并构建 8 维状态(末端执行器位置、轴角姿态、夹爪关节),因此其环境处理器为恒等映射。

将新模拟器的观测映射放在其集成中,将其控制约定放在该集成的 conventions() 中,而不是放在环境处理器中,这样模拟机器人和被提供的环境也能获取到它们(参见 添加基准测试)。

当环境处理器仍然是该工具

  • lerobot.sim 之外的环境。Isaac Lab Arena 通过其仓库自身的 make_env 运行,并使用批量的 torch 观测值;IsaaclabArenaProcessorStep 会选取并拼接其状态键,并转换其相机数据。
  • 针对特定策略的环境适配。 make_env_pre_post_processors 允许策略替换环境中的成对处理器,正如 XVLA 和 Cosmos3 在 LIBERO 中所做的那样。

Evaluation 允许检查点对其环境处理器生成的内容进行校验:即经过预处理器处理后的观测值,以及经过后处理器处理后的动作,两者均从其 transform_features 中读取。因此,一个步骤会在此处将其输出声明为输入的函数,以反映其运行时行为,例如,仅在其源键存在时才构建状态的步骤。远程推理不应用任何环境处理器,因此在那里提供的检查点必须自行接收模拟器的观测值并命令其动作。

评估期间的处理顺序为:

observation = env_preprocessor(observation)   # environment-specific
observation = preprocessor(observation)       # policy's saved preprocessing (normalization, ...)
action = postprocessor(policy.select_action(observation))
action = env_postprocessor(action)            # environment-specific

编写一个

环境处理器是一个普通的处理器步骤,由配置返回:

from dataclasses import dataclass

from lerobot.envs.configs import EnvConfig
from lerobot.processor import ObservationProcessorStep, PolicyProcessorPipeline, ProcessorStepRegistry


@dataclass
@ProcessorStepRegistry.register(name="myenv_processor")
class MyEnvProcessorStep(ObservationProcessorStep):
    def observation(self, observation):
        processed = observation.copy()
        processed["observation.state"] = processed.pop("observation.myenv_state")
        return processed

    def transform_features(self, features):
        return features  # update if names or shapes change


@EnvConfig.register_subclass("myenv")
@dataclass
class MyEnvConfig(EnvConfig):
    ...

    def get_env_processors(self):
        return PolicyProcessorPipeline(steps=[MyEnvProcessorStep()]), PolicyProcessorPipeline(steps=[])

保持环境特定的内容(按键布局、坐标系),并将归一化和模型特定的变换交由策略的处理器处理。

Update on GitHub