Section 21.2: Behavior cloning; the distribution-shift problem

A Careful Control Loop

This section develops the technical contract for behavior cloning; the distribution-shift problem into a usable mental model. First we define the object of study, then we connect it to the agent loop, then we test it with a compact implementation.

The key question in Behavior cloning; the distribution-shift problem is practical: what must the agent know, what can it observe, what action is available, and what evidence shows that the action worked under the stated conditions?

Theory

For Behavior cloning; the distribution-shift problem, the practical design rule is to make the interface inspectable before optimization begins: inputs, outputs, units, latency, bounds, and failure labels should all be visible in the saved artifact.

Worked Example

For Behavior cloning; the distribution-shift problem, keep one concrete rollout in view. A sensor reading becomes an estimate, the estimate constrains an action, the action changes the world, and the next observation confirms or contradicts the assumption. The section's idea is useful only if it improves that loop.

from pathlib import Path

dataset_root = Path("robot_demos")
for episode in sorted(dataset_root.glob("episode_*")):
    print("inspect", episode.name)
print("next step: convert demonstrations to the LeRobotDataset format")

Expected output: the printed trace for Behavior cloning; the distribution-shift problem should expose the method configuration, the measured evidence field, and the failure label. If one of those fields is missing or unchanged under the perturbation, the example is not yet an evaluation artifact.

# Compute a behavior-cloning loss and expose the distribution-shift issue.
# The second line shows how a small action error moves the next observation.
import numpy as np

expert_action = np.array([0.10, 0.00])
policy_action = np.array([0.07, 0.04])
bc_loss = np.mean((policy_action - expert_action) ** 2)
next_observation_error_cm = np.linalg.norm(policy_action - expert_action) * 100
print(f"BC loss: {bc_loss:.4f}")
print(f"next observation offset: {next_observation_error_cm:.1f} cm")

Practical Recipe

1. Write the observation, action, and success metric before choosing a model.
2. Build a baseline that is simple enough to debug by inspection.
3. Add the library implementation only after the baseline behavior is understood.
4. Record failures as structured cases: perception error, state error, planning error, control error, or evaluation error.
5. Run at least one perturbation test before trusting the result.

Behavior cloning; the distribution-shift problem becomes useful when it is tied to a closed-loop contract. In this Part V section on Behavior cloning; the distribution-shift problem, the contract names the observation stream, the state estimate, the action representation, the timing budget, and the evaluation artifact. Without that contract, a model can look capable in a notebook while failing the first time a sensor drops a frame or a controller saturates.

For Behavior cloning; the distribution-shift problem, separate the conceptual claim, the systems claim, and the evidence claim. A plausible mechanism, a clean interface, and a closed-loop result are different claims; the section should keep their evidence separate.

For Behavior cloning; the distribution-shift problem, start with a small baseline that logs inputs, outputs, units, timestamps, and termination conditions before moving to Gymnasium or PettingZoo. The library run should keep the same artifact schema, so the comparison remains a same-task evaluation.

1. Write a one-paragraph task contract with observation, action, success, and failure fields.
2. Start with the smallest simulator, dataset, or wrapper that exposes the task contract faithfully.
3. Run one deterministic smoke test and one perturbation test before scaling.
4. Save a single result artifact containing configuration, seed, metrics, videos or traces, and failure labels.
5. Compare methods only when one script evaluates them on the same task panel.

When Behavior cloning; the distribution-shift problem fails, avoid labeling the whole method as weak. First assign the failure to perception, state estimation, planning, control, timing, data coverage, or evaluation. Then rerun one controlled perturbation that isolates the suspected cause. This pattern turns a disappointing rollout into a reusable diagnostic asset.

pip install pandas pydantic

# Define the fields every demonstration episode must expose.
# Include timing and failure-label fields before evaluation.
from pydantic import BaseModel

class EpisodeCard(BaseModel):
    robot: str
    observation: str
    action: str
    demonstrator: str
    split: str
    license: str
    control_hz: int = 20
    failure_label: str = "none"
    def as_row(self) -> dict[str, object]:
        return self.model_dump()

episode_card = EpisodeCard(
    robot="mobile_manipulator",
    observation="front_rgbd plus proprioception",
    action="delta_end_effector_pose",
    demonstrator="teleop",
    split="train",
    license="CC-BY-4.0",
    control_hz=20,
    failure_label="none"
)
print(episode_card.as_row())

# Create one normal episode and one stress episode for comparison.
# The example includes the stress condition explicitly so the audit can run end to end.
episodes = [
    EpisodeCard(
        robot="dual-arm",
        observation="front and wrist cameras",
        action="joint deltas",
        demonstrator="teleop",
        split="train",
        license="CC-BY-4.0",
        control_hz=20,
        failure_label="none",
    ),
    EpisodeCard(
        robot="dual-arm",
        observation="front and wrist cameras",
        action="joint deltas",
        demonstrator="teleop",
        split="stress",
        license="CC-BY-4.0",
        control_hz=10,
        failure_label="delayed_grasp_recovery",
    ),
]

if isinstance(episodes, list):
    print({"rows": len(episodes), "first": episodes[0] if episodes else None})
elif isinstance(episodes, dict):
    print({"fields": sorted(episodes), "audit_ready": all(value not in (None, "") for value in episodes.values())})
else:
    print({"value": episodes})

# Save one audit table for the baseline and library route.
# Add metric columns after the rollout script runs so the artifact is evaluable.
import pandas as pd

rows = [episode.model_dump() for episode in episodes]
pd.DataFrame(rows).to_csv("part_v_episode_audit.csv", index=False)
print("saved", len(rows), "episodes")

# Validate the maintained-tool route without changing the audit schema.
library_route = {"tool": "LeRobot", "artifact": "part_v_episode_audit.csv"}
required_fields = {"tool", "artifact"}
missing = sorted(required_fields - set(library_route))
assert not missing
print({"loader_ready": True, "tool": library_route["tool"], "artifact": library_route["artifact"]})

# Complete solution for the Part V audit lab.
from pydantic import BaseModel
import pandas as pd

class EpisodeCard(BaseModel):
    robot: str
    observation: str
    action: str
    demonstrator: str
    split: str
    license: str
    timing_hz: int
    failure_label: str

    def as_row(self) -> dict[str, object]:
        return self.model_dump()

episodes = [
    EpisodeCard(
        robot="dual-arm",
        observation="front and wrist cameras",
        action="joint deltas",
        demonstrator="teleop",
        split="train",
        license="CC-BY-4.0",
        timing_hz=30,
        failure_label="none",
    ),
    EpisodeCard(
        robot="dual-arm",
        observation="front and wrist cameras",
        action="joint deltas",
        demonstrator="teleop",
        split="stress",
        license="CC-BY-4.0",
        timing_hz=30,
        failure_label="object-slip",
    ),
]
rows = [episode.as_row() for episode in episodes]
pd.DataFrame(rows).to_csv("part_v_episode_audit.csv", index=False)
print("saved", len(rows), "episodes to part_v_episode_audit.csv")