LeRobot v0.6.0 把机器人开发推向“想象、评估、改进”闭环

mkdir -p lerobot-loop-demo
cd lerobot-loop-demo
python -m venv .venv
. .venv/bin/activate
python - <<'PY'
from __future__ import annotations

import json
import random
from dataclasses import dataclass, asdict
from pathlib import Path

RUN_DIR = Path("runs")
RUN_DIR.mkdir(exist_ok=True)

@dataclass
class Task:
    task_id: str
    object_x: float
    object_y: float
    lighting: str
    distractors: int

@dataclass
class EvalResult:
    task_id: str
    success: bool
    steps: int
    reason: str


def imagine_tasks(n: int = 12) -> list[Task]:
    """Generate task variants. Replace this with simulator/domain randomization later."""
    lighting_options = ["normal", "dim", "backlit"]
    return [
        Task(
            task_id=f"task-{i:03d}",
            object_x=round(random.uniform(-0.25, 0.25), 3),
            object_y=round(random.uniform(0.10, 0.45), 3),
            lighting=random.choice(lighting_options),
            distractors=random.randint(0, 3),
        )
        for i in range(n)
    ]


def evaluate_policy(tasks: list[Task]) -> list[EvalResult]:
    """Toy evaluator. Replace scoring with LeRobot policy rollout metrics."""
    results: list[EvalResult] = []
    for task in tasks:
        difficulty = abs(task.object_x) + task.distractors * 0.08
        if task.lighting != "normal":
            difficulty += 0.10
        success = random.random() > difficulty
        results.append(
            EvalResult(
                task_id=task.task_id,
                success=success,
                steps=random.randint(18, 80),
                reason="ok" if success else "missed_grasp_or_poor_visibility",
            )
        )
    return results


def improve_dataset(tasks: list[Task], results: list[EvalResult]) -> list[dict]:
    """Return failed cases that deserve more data, simulation, or manual review."""
    by_id = {task.task_id: task for task in tasks}
    failures = []
    for result in results:
        if not result.success:
            task = by_id[result.task_id]
            failures.append({
                "task": asdict(task),
                "failure_reason": result.reason,
                "next_action": "collect_more_episodes_or_add_domain_randomization",
            })
    return failures


def main() -> None:
    random.seed(7)
    tasks = imagine_tasks()
    results = evaluate_policy(tasks)
    failures = improve_dataset(tasks, results)

    success_rate = sum(r.success for r in results) / len(results)
    report = {
        "success_rate": round(success_rate, 3),
        "total_tasks": len(tasks),
        "failed_tasks": len(failures),
        "failures": failures,
    }

    out = RUN_DIR / "report.json"
    out.write_text(json.dumps(report, indent=2), encoding="utf-8")
    print(json.dumps(report, indent=2))
    print(f"\nWrote {out}")


if __name__ == "__main__":
    main()
PY

# 假设你使用 Python 环境管理 LeRobot 实验；版本号按你的项目约束调整
python -m venv .venv
. .venv/bin/activate
pip install "lerobot==0.6.0"

mkdir -p experiments/pick-place/{configs,reports,artifacts}
cat > experiments/pick-place/configs/run.json <<'JSON'
{
  "project": "pick-place",
  "loop": "imagine-evaluate-improve",
  "policy_version": "replace-with-your-policy-id",
  "dataset_version": "replace-with-your-dataset-id",
  "metrics": ["success_rate", "steps", "interventions", "failure_reason"]
}
JSON