Hugging Face's logo

LeRobot documentation

处理器

Hugging Face's logo
Join the Hugging Face community

and get access to the augmented documentation experience

to get started

处理器

处理器是机器人、数据集和策略之间的数据转换层。流水线是一系列ProcessorStep的链式组合；每个步骤都声明了它如何转换数据以及特征契约。

请参阅 机器人处理器简介 了解相关概念，实现你自己的处理器 以编写一个步骤，并在流水线出现异常时参考 调试你的处理器流水线。

ProcessorStep

类 lerobot.processor.ProcessorStep

< >

( )

数据处理流水线中单个步骤的抽象基类。

每个步骤必须实现 __call__ 方法，以在数据转换上执行其变换，并实现 transform_features 方法，以描述它如何更改数据特征的形状或类型。

子类可以通过实现 state_dict 和 load_state_dict 来可选地支持有状态。

get_config

< >

( )

返回用于序列化的步骤配置。

load_state_dict

< >

( state: dict[str, torch.Tensor] )

参数

  • state — 状态张量字典。

从状态字典中加载该步骤的状态。

重置

< >

( )

如果存在，重置处理器步骤的内部状态。

save_artifacts

< >

( save_directory: Path )

保存非张量资产，并将构造函数参数映射到相对路径。

state_dict

< >

( )

返回步骤的状态（例如，学习到的参数、运行均值）。

transform_features

< >

( features: dict[PipelineFeatureType, dict[str, PolicyFeature]] )

参数

  • features — 一个字典，用于描述观测、动作等输入特征。

定义此步骤如何修改流水线功能的描述。

此方法用于跟踪数据在流经管道过程中形状、数据类型或模态的变化，而无需处理实际数据。

DataProcessorPipeline

类 lerobot.processor.DataProcessorPipeline

< >

( steps: Sequence[ProcessorStep] = <factory>name: str = 'DataProcessorPipeline'to_transition: Callable[[TInput], EnvTransition] = <factory>to_output: Callable[[EnvTransition], TOutput] = <factory>before_step_hooks: list[Callable[[int, EnvTransition], None]] = <factory>after_step_hooks: list[Callable[[int, EnvTransition], None]] = <factory> )

参数

  • steps — 组成流水线的 ProcessorStep 对象序列。
  • name — 管道的描述性名称。
  • to_transition — 一个将原始输入数据转换为标准化 EnvTransition 格式的函数。
  • to_output — 用于将最终的 EnvTransition 转换为所需输出格式的函数。
  • before_step_hooks — 一个函数列表，将在执行每个步骤之前被调用。
  • after_step_hooks — 一个函数列表，将在每个步骤执行完毕后被调用。

用于处理数据的顺序流水线，并与 Hugging Face Hub 集成。

此类将多个 ProcessorStep 实例串联起来，以形成完整的数据处理工作流。它是通用的，支持自定义的输入和输出类型，这些类型由 to_transition 和 to_output 转换器处理。

from_config

< >

( config: dict[str, Any]state_dict: dict[str, dict[str, torch.Tensor]] | None = Noneoverrides: dict[str, Any] | None = Noneto_transition: Callable[[TInput], EnvTransition] | None = Noneto_output: Callable[[EnvTransition], TOutput] | None = None )

参数

  • config — 一个配置字典，其结构与保存的处理器 JSON 相同。
  • state_dict — 可选的按无后缀状态键分组的内存中流水线状态。
  • overrides — 可选的构造函数覆盖项，以注册表名称或类名称作为键。
  • to_transition — 可选的转换器，用于将输入数据转换为 EnvTransition。
  • to_output — 可选的从 EnvTransition 到输出数据的转换器。

从内存中的配置和可选的状态张量构建流水线。

from_pretrained

< >

( pretrained_model_name_or_path: str | Pathconfig_filename: strforce_download: bool = Falseresume_download: bool | None = Noneproxies: dict[str, str] | None = Nonetoken: str | bool | None = Nonecache_dir: str | Path | None = Nonelocal_files_only: bool = Falserevision: str | None = Noneoverrides: dict[str, Any] | None = Noneto_transition: Callable[[TInput], EnvTransition] | None = Noneto_output: Callable[[EnvTransition], TOutput] | None = None**kwargs )

参数

  • pretrained_model_name_or_path — Hugging Face Hub 上仓库的标识符、本地目录的路径或单个配置文件的路径。
  • config_filename — 流水线 JSON 配置文件的名称。当存在多个配置（例如 preprocessor 与 postprocessor）时，始终需要指定此参数以避免歧义。
  • force_download — 是否强制（重新）下载文件。
  • resume_download — 为保持向后兼容而保留，但会被忽略。下载会自动恢复。
  • proxies — 为保持向后兼容而保留，但会被忽略。请改用 HTTP_PROXY 和 HTTPS_PROXY。
  • token — 用于私有 Hub 仓库的 HTTP 承载者身份验证的令牌。
  • cache_dir — 用于存储下载文件的特定缓存文件夹的路径。
  • local_files_only — 如果为 True，则避免从 Hub 下载文件。
  • revision — 要使用的特定模型版本（例如分支名称、标签名称或提交 ID）。
  • overrides — 用于覆盖特定步骤配置的字典。键应与步骤的类名或注册表名称匹配。
  • to_transition — 用于将输入数据转换为 EnvTransition 的自定义函数。
  • to_output — 用于将最终的 EnvTransition 转换为输出格式的自定义函数。
  • **kwargs — 其他参数（未使用）。

引发

FileNotFoundError 或 ValueError 或 ImportError 或 KeyError 或 ProcessorMigrationError

  • FileNotFoundError — 如果找不到配置文件。
  • ValueError — 如果配置存在歧义或实例化失败。
  • ImportError — 如果某个步骤的类无法导入。
  • KeyError — 如果覆盖键与流水线中的任何步骤都不匹配。
  • ProcessorMigrationError — 如果模型需要迁移到处理器格式。

从本地目录、单个文件或 Hugging Face Hub 仓库加载 pipeline。

此方法实现了一个简化的加载流程，并具备智能迁移检测功能：

简化加载策略：

  1. 配置加载 (_load_config):

    • 目录：从目录中加载指定的 config_filename
    • 单个文件：直接加载文件（忽略 config_filename）
    • Hub 仓库：从 Hub 下载指定的 config_filename

  2. 配置验证 (_validate_loaded_config):

    • 格式验证：确保 config 是有效的处理器格式
    • 迁移检测：引导用户迁移旧版 LeRobot 模型
    • 清除错误：提供可操作的错误信息

  3. 步骤构建 (_build_steps_with_overrides):

    • 类解析：注册表查找或动态导入
    • 覆盖合并：用户参数会覆盖已保存的配置
    • 状态加载：为有状态步骤加载 .safetensors 文件

  4. 覆盖验证 (_validate_overrides_used):

    • 确保所有用户覆盖项均已应用（捕获拼写错误）
    • 提供包含可用键的有用错误消息

迁移检测：

  • 智能检测：分析 JSON 文件以检测旧版 LeRobot 模型
  • 精准定位：避免在其他 HuggingFace 模型上出现误报
  • 清晰的指导：提供要运行的确切迁移命令
  • 错误模式：始终引发 ProcessorMigrationError，以便用户明确执行操作

加载示例：

# Directory loading
pipeline = DataProcessorPipeline.from_pretrained("/models/my_model", config_filename="processor.json")

# Single file loading
pipeline = DataProcessorPipeline.from_pretrained(
    "/models/my_model/processor.json", config_filename="processor.json"
)

# Hub loading
pipeline = DataProcessorPipeline.from_pretrained("user/repo", config_filename="processor.json")

# Multiple configs (preprocessor/postprocessor)
preprocessor = DataProcessorPipeline.from_pretrained(
    "model", config_filename="policy_preprocessor.json"
)
postprocessor = DataProcessorPipeline.from_pretrained(
    "model", config_filename="policy_postprocessor.json"
)

覆盖系统：

  • 关键匹配：使用注册表名称或类名称作为覆盖键
  • 配置合并：用户覆盖项优先于已保存的配置
  • 验证：确保所有覆盖键与实际步骤匹配（捕获拼写错误）
  • 示例：overrides={“NormalizeStep”: {“device”: “cuda”}}

get_config

< >

( )

返回可 JSON 序列化的流水线配置。

load_state_dict

< >

( state_dict: dict[str, dict[str, torch.Tensor]] )

参数

  • state_dict — 一个字典，将无后缀的状态键映射到步骤状态字典。

引发

KeyError

  • KeyError — 如果加载时发现有缺失的预期状态或意外的额外状态。

将流水线状态张量加载到现有步骤中。

process_action

< >

( action: PolicyAction | RobotAction | EnvAction )

参数

  • action — 动作数据。

仅通过流水线处理转换中的动作部分。

process_complementary_data

< >

( complementary_data: dict[str, Any] )

参数

  • complementary_data — 补充数据字典。

仅通过管道处理转换的互补数据部分。

process_done

< >

( done: bool | torch.Tensor )

参数

  • done — done 标志。

仅通过流水线处理 transition 的 done 标志。

process_info

< >

( info: dict[str, Any] )

参数

  • info — 信息字典。

仅通过流水线处理转换的信息字典。

process_observation

< >

( observation: RobotObservation )

参数

  • observation — 观测字典。

仅通过流水线处理转移中的观察部分。

process_reward

< >

( reward: float | torch.Tensor )

参数

  • reward — 奖励值。

仅通过流水线处理转移中的奖励部分。

process_truncated

< >

( truncated: bool | torch.Tensor )

参数

  • truncated — 截断标志。

仅处理通过管道时的截断标志。

register_after_step_hook

< >

( fn: Callable[[int, EnvTransition], None] )

参数

  • fn — 一个可调用对象，接受步数索引和当前转换作为参数。

注册一个函数，以便在每一步之后调用。

register_before_step_hook

< >

( fn: Callable[[int, EnvTransition], None] )

参数

  • fn — 一个可调用对象，接受步数索引和当前转换作为参数。

注册一个函数，以便在每一步之前调用。

重置

< >

( )

重置管道中所有有状态步骤的状态。

save_pretrained

< >

( save_directory: str | Path | None = Nonerepo_id: str | None = Nonepush_to_hub: bool = Falsecard_kwargs: dict[str, Any] | None = Noneconfig_filename: str | None = None**push_to_hub_kwargs )

参数

  • save_directory — 流水线将保存到的目录。如果为 None，则保存到 HF_LEROBOT_HOME/processors/{sanitized_pipeline_name}。
  • repo_id — 您在 Hub 上的仓库 ID。仅在 push_to_hub=true 时使用。
  • push_to_hub — 是否在保存对象后将其推送到 Hugging Face Hub。
  • card_kwargs — 传递给卡片模板以自定义卡片的额外参数。
  • config_filename — JSON 配置文件的名称。如果为 None，则根据流水线的 name 属性生成名称。
  • **push_to_hub_kwargs — Additional key word arguments passed along to the push_to_hub method.

将流水线的配置和状态保存到目录中。

此方法会创建一个 JSON 配置文件，用于定义流水线的结构（名称和步骤）。对于每个有状态的步骤，它还会保存一个 .safetensors 文件，其中包含其状态字典。

state_dict

< >

( )

按状态键分组返回流水线状态张量。

step_through

< >

( data: TInput )

参数

  • data — 输入数据。

逐步处理数据，并在每个阶段输出转换结果。

这是一个生成器方法，可用于调试和检查数据在通过管道过程中的中间状态。

transform_features

< >

( initial_features: dict[PipelineFeatureType, dict[str, PolicyFeature]] )

参数

  • initial_features — 描述初始特征的字典。

按顺序应用所有步骤中的特征转换。

此方法通过每个步骤的 transform_features 方法传播特征描述字典，使流水线能够在不处理任何实际数据的情况下静态确定输出特征规范。

unregister_after_step_hook

< >

( fn: Callable[[int, EnvTransition], None] )

参数

  • fn — 之前注册的精确函数对象。

引发

ValueError

  • ValueError — 如果在列表中未找到该钩子。

注销一个‘after_step’钩子。

unregister_before_step_hook

< >

( fn: Callable[[int, EnvTransition], None] )

参数

  • fn — 之前注册的精确函数对象。

引发

ValueError

  • ValueError — 如果在列表中未找到该钩子。

注销一个‘before_step’钩子。

PolicyProcessorPipeline

lerobot.processor.DataProcessorPipeline

( *args**kwargs )

Update on GitHub

←策略 环境→