Python API

Public Imports

The package root exposes the common entry models:

from robosmith import TaskSpec, ForgeConfig, SmithConfig

ForgeConfig and SmithConfig are aliases. TaskSpec is the central task schema passed into the training pipeline.

Run Pipeline

from robosmith import TaskSpec, ForgeConfig
from robosmith.agent.graphs.run import run_pipeline, resume_pipeline

spec = TaskSpec(task_description="A quadruped walks forward")
config = ForgeConfig(max_iterations=2)

state = run_pipeline(spec, config)
print(state["run_id"])

resumed = resume_pipeline(state["run_id"])

run_pipeline(task_spec, config=None, on_step=None) Runs the full LangGraph training workflow and returns `PipelineState`.

resume_pipeline(run_id, runs_dir=None, on_step=None) Loads `checkpoint.json`, skips completed nodes, and continues the graph.

on_step is an optional callback with (node_name, log_line).

Config Models

from robosmith.config import (
    Algorithm,
    Decision,
    EnvironmentType,
    ForgeConfig,
    LLMConfig,
    RewardSearchConfig,
    RobotType,
    RunState,
    SafetyConstraint,
    StageRecord,
    StageStatus,
    SuccessCriterion,
    TaskSpec,
)

Important fields:

Model	Purpose
TaskSpec	Structured task description used by every pipeline stage.
SuccessCriterion	Metric, operator, and threshold for success.
SafetyConstraint	Human-readable and optional metric-based safety constraint.
LLMConfig	Provider, model, fast model, temperature, retries.
RewardSearchConfig	Candidate count, evolution iterations, eval budget.
ForgeConfig	Top-level pipeline config, paths, stage skipping, scout source.
RunState	Serializable run summary used by delivery and run inspection.

Environment API

from robosmith.envs.registry import EnvRegistry, EnvEntry
from robosmith.envs.wrapper import make_env
from robosmith.envs.adapter_registry import EnvAdapterRegistry
from robosmith.envs.adapters import EnvAdapter, EnvConfig
from robosmith.envs.reward_wrapper import ForgeRewardWrapper

EnvRegistry(registry_path=None) Loads environment entries from YAML.

EnvRegistry.get(entry_id) Returns one `EnvEntry` or `None`.

EnvRegistry.search(...) Filters by robot type, environment type, framework, model, tags, or action type.

make_env(entry, **kwargs) Creates a live environment through the matching adapter.

ForgeRewardWrapper(env, reward_fn, reward_clip=100.0) Replaces environment rewards with a custom reward function.

Environment Synthesis

from robosmith.stages.env_synthesis.synthesis import EnvMatch, match_task_to_env

match = match_task_to_env(spec, EnvRegistry(), framework="gymnasium")

EnvMatch contains entry, score, and match_reason.

Trainer API

from robosmith.trainers.base import (
    LearningParadigm,
    Policy,
    Trainer,
    TrainingConfig,
    TrainingResult,
)
from robosmith.trainers.registry import TrainerRegistry
from robosmith.trainers.selector import PolicyApproach, select_policy_approach

TrainerRegistry().list_available() Returns installed training backends.

TrainerRegistry().list_all() Returns all known backends with availability, required packages, paradigm, and algorithms.

TrainerRegistry().get_trainer(algorithm, backend=None, paradigm=None) Returns a ready trainer or raises when dependencies are missing.

select_policy_approach(...) Chooses paradigm, algorithm, backend, reason, confidence, and alternatives.

Inspection API

from robosmith.inspect.dispatch import (
    inspect_dataset,
    inspect_env,
    inspect_policy,
    inspect_robot,
)
from robosmith.inspect.compat import check_compatibility
from robosmith.inspect.models import (
    CameraSpec,
    CompatReport,
    DatasetInspectResult,
    EnvInspectResult,
    PolicyInspectResult,
    RobotInspectResult,
)

dataset = inspect_dataset("lerobot/aloha_mobile_cabinet")
env = inspect_env("Ant-v5")
policy = inspect_policy("lerobot/smolvla_base")
robot = inspect_robot("path/to/robot.urdf")
report = check_compatibility(policy.model_id, dataset.repo_id)

check_compatibility(a, b, c=None) returns CompatReport with compatible, errors, warnings, and info.

Diagnostics API

from robosmith.diagnostics.trajectory_analyzer import (
    analyze_trajectory,
    compare_trajectories,
)
from robosmith.diagnostics.diag_models import (
    ActionStats,
    FailureCluster,
    TrajectoryCompareResult,
    TrajectoryDiagResult,
)

result = analyze_trajectory("path/to/rollout.hdf5")
comparison = compare_trajectories("rollout_a.hdf5", "rollout_b.hdf5")

Generator API

from robosmith.generators.gen_wrapper import generate_wrapper

code = generate_wrapper(
    policy_id="lerobot/smolvla_base",
    target_id="lerobot/aloha_mobile_cabinet",
    output_path="adapter.py",
    use_llm=False,
)

Agent Models

from robosmith.agent.models.base import BaseAgent
from robosmith.agent.models.reward import RewardAgent, RewardCandidate
from robosmith.agent.models.decision import DecisionAgent

Agents use LiteLLM and are wrapped so structured JSON calls can retry after parse failures. Most users interact with these through the pipeline, not directly.