LeRobot documentation
调试您的处理器流水线
调试您的处理器流水线
处理器流水线可能非常复杂，尤其是在串联多个转换步骤时。 与简单的函数调用不同，流水线缺乏天然的可视化能力，你很难直观地看到每个步骤之间发生了什么，或者问题出在哪里。 本指南提供了专门针对这些挑战设计的调试工具和技术，帮助你理解数据在流水线中的流动过程。
我们将探讨三种互补的调试方法：钩子用于运行时监控，单步调试用于详细检查，以及特征验证用于捕获结构不匹配。每种方法各有侧重，共同为您提供对管道行为的全面可见性。
理解 Hooks
Hooks 是在流水线执行的特定时间点被调用的函数。 它们提供了一种在不修改流水线代码的情况下检查、监控或修改数据的方式。 可以将它们视为流水线的“事件监听器”。
什么是 Hook？
Hook 是一种回调函数，会在管道执行过程中的特定时刻被自动调用。 这一概念源自事件驱动编程，想象一下你可以“挂钩”到管道的执行流程中，以观察或响应正在发生的事情。
可以将钩子（hooks）想象为在流水线中插入的检查点。每当流水线到达其中一个检查点时，它会短暂暂停以调用你的钩子函数，从而让你有机会检查当前状态、记录信息并验证数据。
钩子只是一个接受两个参数的函数：
step_idx: int- 当前处理步骤的索引（0、1、2 等）transition: EnvTransition- 流水线中该位置的数据转换
钩子（hooks）的美妙之处在于其非侵入性：你可以在不改动流水线代码任何一行的情况下，添加监控、验证或调试逻辑。流水线保持简洁并专注于其核心逻辑，而钩子则负责处理日志记录、监控和调试等横切关注点。
Before 与 After 钩子
该管道支持两种类型的钩子：
- 前置钩子（
register_before_step_hook）- 在每个步骤执行前调用 - 钩子之后 (
register_after_step_hook) - 在每个步骤完成后调用
def before_hook(step_idx: int, transition: EnvTransition):
"""Called before step processes the transition."""
print(f"About to execute step {step_idx}")
# Useful for: logging, validation, setup
def after_hook(step_idx: int, transition: EnvTransition):
"""Called after step has processed the transition."""
print(f"Completed step {step_idx}")
# Useful for: monitoring results, cleanup, debugging
processor.register_before_step_hook(before_hook)
processor.register_after_step_hook(after_hook)实现 NaN 检测钩子
下面是一个检测 NaN 值的钩子的实用示例：
def check_nans(step_idx: int, transition: EnvTransition):
"""Check for NaN values in observations."""
obs = transition.get(TransitionKey.OBSERVATION)
if obs:
for key, value in obs.items():
if isinstance(value, torch.Tensor) and torch.isnan(value).any():
print(f"NaN detected in {key} at step {step_idx}")
# Register the hook to run after each step
processor.register_after_step_hook(check_nans)
# Process your data - the hook will be called automatically
output = processor(input_data)
# Remove the hook when done debugging
processor.unregister_after_step_hook(check_nans)Hooks 的内部工作原理
理解内部机制有助于你更有效地使用钩子。流水线维护着两个独立的列表：一个用于 before-step 钩子，另一个用于 after-step 钩子。当你注册一个钩子时，它只是被追加到相应的列表中。
在执行过程中，流水线遵循严格的顺序：对于每个处理步骤，它首先按注册顺序调用所有 before-hooks，然后执行实际的步骤转换，最后按注册顺序调用所有 after-hooks。这为每个步骤创建了一种可预测的、类似三明治的结构。
关键洞察在于，钩子不会改变核心流水线逻辑——它们纯粹是附加的。流水线的 _forward 方法协调了钩子与处理步骤之间的配合，确保你的调试或监控代码恰好在正确的时机运行，而不会干扰主数据流。
以下是流水线执行钩子的简化视图：
class DataProcessorPipeline:
def __init__(self):
self.steps = [...]
self.before_step_hooks = [] # List of before hooks
self.after_step_hooks = [] # List of after hooks
def _forward(self, transition):
"""Internal method that processes the transition through all steps."""
for step_idx, processor_step in enumerate(self.steps):
# 1. Call all BEFORE hooks
for hook in self.before_step_hooks:
hook(step_idx, transition)
# 2. Execute the actual processing step
transition = processor_step(transition)
# 3. Call all AFTER hooks
for hook in self.after_step_hooks:
hook(step_idx, transition)
return transition
def register_before_step_hook(self, hook_fn):
self.before_step_hooks.append(hook_fn)
def register_after_step_hook(self, hook_fn):
self.after_step_hooks.append(hook_fn)执行流程
执行流程如下：
Input → Before Hook → Step 0 → After Hook → Before Hook → Step 1 → After Hook → ... → Output例如，使用 3 个步骤和两种钩子类型：
def timing_before(step_idx, transition):
print(f"⏱️ Starting step {step_idx}")
def validation_after(step_idx, transition):
print(f"✅ Completed step {step_idx}")
processor.register_before_step_hook(timing_before)
processor.register_after_step_hook(validation_after)
# This will output:
# ⏱️ Starting step 0
# ✅ Completed step 0
# ⏱️ Starting step 1
# ✅ Completed step 1
# ⏱️ Starting step 2
# ✅ Completed step 2多个钩子
您可以注册多个同类型的钩子 - 它们将按照注册的顺序执行：
def log_shapes(step_idx: int, transition: EnvTransition):
obs = transition.get(TransitionKey.OBSERVATION)
if obs:
print(f"Step {step_idx} observation shapes:")
for key, value in obs.items():
if isinstance(value, torch.Tensor):
print(f" {key}: {value.shape}")
processor.register_after_step_hook(check_nans) # Executes first
processor.register_after_step_hook(log_shapes) # Executes second
# Both hooks will be called after each step in registration order
output = processor(input_data)虽然钩子非常适合监控特定问题（例如 NaN 检测）或在正常流水线执行期间收集指标，但有时你需要更深入地探究。当你想要确切了解每个步骤中发生了什么，或者调试复杂的转换逻辑时，单步调试（step-through debugging）能提供你所需要的详细检查。
单步调试
逐步调试就像为你的管道提供慢动作回放。你不必看着数据从输入到输出在一瞬间模糊地变换，而是可以暂停并检查每个单独步骤之后发生了什么。
这种方法在尝试理解复杂流水线、调试意外行为或验证每个转换是否按预期工作时尤为有价值。与钩子（hooks）不同，后者非常适合自动化监控，而逐步调试（step-through debugging）则让你能够手动、交互式地控制检查过程。
step_through() 方法是一个生成器，它在每个处理步骤后产出转换状态，从而允许您检查中间结果。可以将其视为在数据流经管道时创建的一系列快照——每个快照都精确展示了数据在应用了另一次转换后的样子。
逐步执行的工作原理
step_through() 方法从根本上改变了流水线的执行方式。它不再按顺序运行所有步骤并仅返回最终结果，而是将流水线转换为一个迭代器，逐步产出中间结果。
内部流程如下：该方法首先将输入数据转换为流水线内部的过渡格式，然后生成（yield）此初始状态。接着，它应用第一个处理步骤并生成结果。随后，它对该结果应用第二个步骤并再次生成，依此类推。每次 yield 都会提供该时刻过渡状态的完整快照。
这种生成器模式之所以强大，是因为它是惰性的——只有当你请求时，流水线才会计算下一步。这意味着你可以在任意时刻停止，彻底检查当前状态，并决定是否继续。你不必为了调试某个有问题的步骤而被迫运行整个流水线。
与其运行整个流水线并只查看最终结果，step_through() 会在每一步之后暂停，并为您提供中间转换：
# This creates a generator that yields intermediate states.
# The first yield is the input itself; yield i (i >= 1) is the output of step i - 1.
for i, intermediate_result in enumerate(processor.step_through(input_data)):
print("=== Input ===" if i == 0 else f"=== After step {i - 1} ===")
# Inspect the observation at this stage
obs = intermediate_result.get(TransitionKey.OBSERVATION)
if obs:
for key, value in obs.items():
if isinstance(value, torch.Tensor):
print(f"{key}: shape={value.shape}, dtype={value.dtype}")使用断点进行交互式调试
您可以在逐步执行循环中添加断点，以进行交互式调试：
# Step through the pipeline with debugging
for i, intermediate in enumerate(processor.step_through(data)):
print("Input" if i == 0 else f"After step {i - 1}: {processor.steps[i - 1].__class__.__name__}")
# Set a breakpoint to inspect the current state
breakpoint() # Debugger will pause here
# You can now inspect 'intermediate' in the debugger:
# - Check tensor shapes and values
# - Verify expected transformations
# - Look for unexpected changes在调试器会话期间，你可以：
- 检查
intermediate[TransitionKey.OBSERVATION]以查看观测数据 - 检查
intermediate[TransitionKey.ACTION]以查看动作转换 - 检查转换的任意部分以了解每个步骤的作用
逐步调试非常适合理解数据的转换过程，但数据的结构呢？虽然钩子和逐步调试有助于排查运行时行为，但你还需要确保流水线生成的数据符合下游组件所期望的格式。这正是特征契约验证发挥作用的地方。
验证特征合约
功能合约定义了您的管道所期望的输入数据结构以及所产生的输出数据结构。 验证这些合约有助于尽早发现不匹配的问题。
理解特征合约
每个处理器步骤都有一个 transform_features() 方法，用于描述它如何更改数据结构：
# Get the expected output features from your pipeline
initial_features = {
PipelineFeatureType.OBSERVATION: {
"observation.state": PolicyFeature(type=FeatureType.STATE, shape=(7,)),
"observation.image": PolicyFeature(type=FeatureType.VISUAL, shape=(3, 224, 224))
},
PipelineFeatureType.ACTION: {
"action": PolicyFeature(type=FeatureType.ACTION, shape=(4,))
}
}
# Check what your pipeline will output
output_features = processor.transform_features(initial_features)
print("Input features:")
for feature_type, features in initial_features.items():
print(f" {feature_type}:")
for key, feature in features.items():
print(f" {key}: {feature.type.value}, shape={feature.shape}")
print("\nOutput features:")
for feature_type, features in output_features.items():
print(f" {feature_type}:")
for key, feature in features.items():
print(f" {key}: {feature.type.value}, shape={feature.shape}")验证预期功能
检查您的 pipeline 是否生成了预期的特征：
# Define what features you expect the pipeline to produce
expected_keys = ["observation.state", "observation.image", "action"]
print("Validating feature contract...")
for expected_key in expected_keys:
found = False
for feature_type, features in output_features.items():
if expected_key in features:
feature = features[expected_key]
print(f"✅ {expected_key}: {feature.type.value}, shape={feature.shape}")
found = True
break
if not found:
print(f"❌ Missing expected feature: {expected_key}")此验证有助于确保您的管道能够与期望特定数据结构的下游组件正确配合工作。
摘要
现在，您已经了解了这三种调试方法，可以系统地解决任何流水线问题：
- 钩子 - 用于在不修改流水线代码的情况下进行运行时监控和验证
- 逐步执行 - 用于检查中间状态并理解转换过程
- 功能验证 - 用于确保满足数据结构契约
何时使用每种方法：
- 当您需要了解流水线的工作方式或遇到意外情况时，请从逐步调试开始
- 添加 hooks 以在开发和生产环境中进行持续监控，从而自动捕获问题
- 在部署前使用 feature validation，以确保您的管道能够与下游组件协同工作
这三个工具协同工作，为您提供复杂流水线天然缺乏的完整可观测性。借助钩子监控问题、逐步执行帮助理解行为以及特征验证确保兼容性，您将能够自信且高效地调试任何流水线。
Update on GitHub