LeRobot 文件
除錯您的處理器管線
並獲得增強的文件體驗
開始使用
除錯您的處理器管線
處理器管線 (Processor pipelines) 可能會很複雜，特別是在串接多個轉換步驟時。與簡單的函式呼叫不同，管線缺乏直觀的可觀測性；您無法輕易查看每個步驟之間發生了什麼，或是問題出在哪裡。本指南提供了專為解決這些挑戰而設計的除錯工具與技術，協助您理解資料流經管線的過程。
我們將探討三種互補的除錯方法：用於執行時期監控的勾子 (hooks)、用於詳細檢查的逐步除錯 (step-through debugging)，以及用於捕捉結構不符的特徵驗證 (feature validation)。每一種方法都有其不同的用途，結合起來便能全面掌握管線的行為。
理解勾子 (Hooks)
勾子是會在管線執行期間特定時間點被呼叫的函式。它們提供了一種無需更動管線程式碼即可檢查、監控或修改資料的方法。您可以將它們視為管線的「事件監聽器」。
什麼是勾子 (Hook)？
勾子是一種回呼函式 (callback function)，會在管線執行過程中的特定時刻自動被觸發。這個概念源於事件驅動程式設計；試想您可以「掛載」到管線的執行流程中，藉此觀察正在發生的事情或對其做出反應。
您可以將勾子想像成在管線中插入檢查點。每當管線到達這些檢查點時，它會短暫暫停並呼叫您的勾子函式，讓您有機會檢查目前的狀態、記錄資訊並驗證資料。
勾子就是一個接收兩個參數的函式：
step_idx: int- 目前處理步驟的索引（0、1、2 等）transition: EnvTransition- 管線在該點的資料轉變狀態
勾子的優點在於其非侵入性：您可以加入監控、驗證或除錯邏輯，而無需修改管線程式碼的任何一行。管線保持簡潔並專注於其核心邏輯，而勾子則處理如日誌記錄、監控與除錯等橫切關注點 (cross-cutting concerns)。
Before 與 After 勾子
管線支援兩種勾子：
- Before 勾子 (
register_before_step_hook) - 在每個步驟執行前呼叫 - After 勾子 (
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)勾子的內部運作方式
理解內部機制有助於更有效地使用勾子。管線維護了兩個獨立的列表：一個用於 Before 步驟勾子，另一個用於 After 步驟勾子。當您註冊一個勾子時，它僅僅是被加入到對應的列表中。
執行期間，管線遵循嚴格的順序：對於每個處理步驟，它首先依註冊順序呼叫所有的 Before 勾子，執行實際的步驟轉換，最後依註冊順序呼叫所有的 After 勾子。這在每個步驟前後建立了一種可預測的、三明治般的結構。
關鍵在於勾子不會改變核心管線邏輯——它們純粹是附加的。管線的 _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() 方法是一個產生器 (generator)，它會在每個處理步驟後產生 (yield) 轉變狀態，允許您檢查中間結果。您可以將其視為在資料流經管線時為其建立了一系列快照——每個快照都向您顯示了在套用另一個轉換後資料的具體模樣。
逐步除錯的運作方式
step_through() 方法從根本上改變了管線的執行方式。它不再是依序執行所有步驟並僅傳回最終結果，而是將管線轉換為一個會產出中間結果的迭代器。
其內部運作如下：該方法首先將您的輸入資料轉換為管線的內部轉變格式，然後產生 (yield) 此初始狀態。接著，它對該結果應用第一個處理步驟並再次產生結果，隨後將第二個步驟應用於該結果並再次產生，依此類推。每一次的 yield 都會為您提供該點處完整的轉變快照。
這種產生器模式非常強大，因為它是惰性的 (lazy)——管線只有在您要求時才會計算下一步。這意味著您可以在任何一點停止、徹底檢查當前狀態，並決定是否要繼續。您不必為了除錯某個有問題的步驟而被迫執行整個管線。
與其執行整個管線並只看到最終結果，step_through() 會在每個步驟後暫停並為您提供中間轉變狀態。
# This creates a generator that yields intermediate states
for i, intermediate_result in enumerate(processor.step_through(input_data)):
print(f"=== After step {i} ===")
# 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 迴圈中加入斷點以進行互動式除錯。
# Step through the pipeline with debugging
for i, intermediate in enumerate(processor.step_through(data)):
print(f"Step {i}: {processor.steps[i].__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]以獲取行動轉換。 - 檢查轉變過程的任何部分，以了解每個步驟的功能。
逐步除錯非常適合理解資料轉換，但那些資料的結構又該如何確認？雖然勾子和逐步除錯有助於除錯執行時期的行為，但您也需要確保管線產生的資料格式符合下游元件的預期。這就是特徵契約驗證 (feature contract validation) 的作用。
驗證特徵契約
特徵契約定義了您的管線預期作為輸入及預期作為輸出的資料結構。驗證這些契約有助於儘早捕捉不匹配的問題。
理解特徵契約
每個處理器步驟都有一個 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.IMAGE, 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}")驗證預期的特徵
檢查您的管線是否產生了您預期的特徵。
# 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) - 用於在不修改管線程式碼的情況下進行執行時期監控與驗證。
- 逐步除錯 (Step-through) - 用於檢查中間狀態並理解轉換過程。
- 特徵驗證 (Feature validation) - 用於確保符合資料結構契約。
何時使用各個方法：
- 當您需要了解管線的功能或發生非預期情況時，請從逐步除錯開始。
- 在開發與生產環境中加入勾子進行持續監控，以自動捕捉問題。
- 在部署前使用特徵驗證，確保您的管線能與下游元件順利運作。
這三種工具相輔相成，為您提供複雜管線所缺乏的全面可觀測性。透過勾子監視問題、逐步除錯協助理解行為，以及特徵驗證確保相容性，您將能自信且高效地除錯任何管線。 在 GitHub 上更新