interlatent 0.4.9

Interlatent SDK — hosted API client for robot post-training.

Interlatent Python SDK

Official Python SDK for the Interlatent platform — interpretability and analysis for autonomous policy execution.

The SDK handles three things:

1. Collect — hook PyTorch model layers and record per-step activations into a local SQLite database
2. Upload — push raw activation data and frames to S3 via presigned URLs
3. Analyze — trigger server-side processing (SAE training, autolabeling, failure classification, reports) and retrieve results

Install

pip install interlatent

Optional extras for environment integrations:

pip install 'interlatent[isaaclab]'   # Isaac Lab (requires isaaclab installed separately)
pip install 'interlatent[mjlab]'      # MuJoCo Lab (requires mjlab installed separately)
pip install 'interlatent[lerobot]'    # LeRobot (adds huggingface_hub for checkpoint naming)

Requirements: Python >= 3.10, PyTorch >= 2.0

Quickstart

from interlatent import Interlatent

client = Interlatent(api_key="ilat_...")

# 1. Hook model and collect activations
watcher = client.watch(
    model,
    env,
    model_id="my-policy",
    layer="auto",
    failure_descriptions={
        "hard_crash": "reward < -80 and done",
        "tumble": "angle > 0.6 or angle < -0.6",
    },
    success_descriptions={
        "soft_landing": "reward > 100 and done",
    },
    capture_frames=True,
)

# 2. Run your environment loop
obs, _ = env.reset()
for step in range(3000):
    action, _ = model.predict(obs, deterministic=True)
    obs, reward, done, truncated, info = env.step(action)
    client.tick(obs=obs, reward=reward, done=done, truncated=truncated, info=info)
    if done or truncated:
        obs, _ = env.reset()

# 3. Upload raw data
client.upload()

# 4. Trigger server-side analysis and wait for results
result = client.checkpoint(sae_k=32)
status = client.runs.wait(result["run_id"], timeout=600)
results = client.runs.results(result["run_id"])

client.close()

Client Constructor

client = Interlatent(
    api_key="ilat_...",       # API key (or set INTERLATENT_API_KEY env var)
    base_url=None,            # Override API base URL (default: localhost:5280, or INTERLATENT_BASE_URL env var)
    bypass_token=None,        # Vercel bypass token (or INTERLATENT_BYPASS_TOKEN env var)
    timeout=30.0,             # HTTP request timeout in seconds
    db_path=None,             # Custom path for local SQLite database
    model_id=None,            # Stable model identifier (required for upload/processing)
)

Collection

watch() — passive collection (you drive the loop)

Hook a model and start capturing activations. You control the environment loop and call tick() after each step.

watcher = client.watch(
    model,                          # PyTorch model or SB3 model
    env,                            # Gymnasium environment (optional, used for env name detection)
    layer="auto",                   # Layer to hook ("auto" detects hookable layers)
    model_id="my-policy",           # Required — stable identifier for this model
    env_name=None,                  # Override environment name (auto-detected from env if omitted)
    failure_descriptions=None,      # Dict of failure rule name -> boolean expression
    success_descriptions=None,      # Dict of success rule name -> boolean expression
    metrics=None,                   # Custom metrics (auto-detected from env if omitted)
    context_fn=None,                # Callable returning extra per-step context dict
    max_channels=None,              # Max activation channels to record
    total_steps=None,               # Expected total steps (for progress display)
    capture_frames=False,           # Capture rendered frames
    frame_every=1,                  # Capture a frame every N steps
    frame_quality=85,               # JPEG quality for saved frames
    visual_criteria=None,           # VLM scoring criteria for chunk-based scoring
    frame_dir=None,                 # Custom directory for frame storage
)

Then drive your loop:

obs, _ = env.reset()
for step in range(steps):
    action, _ = model.predict(obs, deterministic=True)
    obs, reward, done, truncated, info = env.step(action)
    client.tick(
        obs=obs,
        reward=float(reward),
        done=done,
        truncated=truncated,
        info=info,
        frame=env.render(),        # optional — pass frames directly via tick()
    )
    if done or truncated:
        obs, _ = env.reset()

collect() — automatic collection (SDK drives the loop)

Runs the full environment loop for you. The SDK calls model.predict() and env.step() internally.

result = client.collect(
    model,
    env,
    steps=5000,
    layer="auto",
    failure_descriptions={"crash": "reward < -50 and done"},
    success_descriptions={"landed": "reward > 100 and done"},
    capture_frames=True,
    deterministic=True,
)
# result = {"episode_id": "...", "run_id": "...", "steps": 5000, "env_name": "...", ...}

Multicamera frame capture

Register camera names to capture from multiple viewpoints:

client.register_cameras(["front", "side", "overhead"])

# Then pass a dict of camera images per tick:
client.tick(
    obs=obs, reward=reward, done=done, truncated=truncated,
    frame={"front": front_img, "side": side_img, "overhead": overhead_img},
)

Failure taxonomy

Failure and success rules are boolean expressions evaluated per step. Available variables include reward, done, truncated, episode_reward, and any auto-detected or custom metrics.

client.watch(
    model, env,
    failure_descriptions={
        "hard_crash":   "reward < -80 and done",
        "drift_crash":  "y_velocity < -0.8 and y_position < 0.2 and done",
        "out_of_frame": "x_position < -0.9 or x_position > 0.9",
        "tumble":       "angle > 0.6 or angle < -0.6",
    },
    success_descriptions={
        "soft_landing": "reward > 100 and done",
    },
)

If an OPENAI_API_KEY is available, you can also pass natural-language failure descriptions — the SDK will translate them to compiled expressions via LLM. Without the key, plain-text descriptions are stored for VLM-based classification at processing time.

Upload and Processing

upload() — push raw data to the server

Registers episodes, uploads the local SQLite database and frames to S3 via presigned URLs, and confirms the upload.

client.upload(
    tags={"experiment": "v2"},    # optional metadata
    workers=8,                     # parallel upload threads
    reward_config=None,            # optional reward config dict
)

checkpoint() — upload + trigger server-side analysis

Calls upload() internally, then triggers the full analysis pipeline on the server:

result = client.checkpoint(
    sae_k=32,                      # SAE dictionary size
    vlm_chunk_size=None,           # frames per VLM chunk
    vlm_frames_between=None,       # frame sampling interval
)
# result = {"run_id": "...", "job_id": "...", "status": "pending", ...}

The server-side pipeline runs: SAE training, latent statistics, autolabeling (if OPENAI_API_KEY is set on server), episode export, failure classification, and checkpoint report generation.

Poll for results

# Block until processing completes
status = client.runs.wait(result["run_id"], timeout=600, poll=5.0)

# Or poll manually
while True:
    data = client.runs.status(run_id)
    if data["processing"]["status"] in ("completed", "failed"):
        break

# Retrieve results
results = client.runs.results(run_id)
report = results["report"]

Stable Baselines3 Integration

Use the SB3 callback to automatically collect activations during training:

client = Interlatent(api_key="...", model_id="my-sb3-agent")

client.watch(model, env, layer="auto", capture_frames=True)

callback = client.sb3_callback(checkpoint_every=10_000)
model.learn(100_000, callback=callback)

client.upload()
client.close()

Isaac Lab Integration

The IsaacSimCollectionEnv wraps an Isaac Lab environment for activation collection. It supports both ManagerBasedRLEnv and DirectRLEnv.

Mode 1 — standalone collection

The wrapper drives its own rollout loop:

from interlatent import Interlatent
from interlatent.isaaclab.collection_env import IsaacSimCollectionEnv

client = Interlatent(api_key="...", model_id="spot-velocity")
env = gym.make("Isaac-Velocity-Flat-Spot-v0", cfg=env_cfg)

col_env = IsaacSimCollectionEnv(
    env,
    interlatent_client=client,
    env_name="Isaac-Velocity-Flat-Spot-v0",
    failure_rules={"fell_over": "base_height < 0.3"},
    success_rules={"goal_reached": "episode_reward > 50"},
)

runner = OnPolicyRunner(col_env, asdict(agent_cfg), device=device)
runner.load(checkpoint_path, load_cfg={"actor": True}, strict=True)
col_env.attach(runner.alg.actor)

result = col_env.collect(steps=2000)
# result = {"episode_id": "...", "steps": 2000, "env_name": "..."}

Mode 2 — passive collection (hooks into an existing training/eval loop)

The wrapper intercepts env.step() calls and automatically feeds data to the Interlatent client:

col_env = IsaacSimCollectionEnv(env, interlatent_client=client, env_name="Spot-v0")
runner = OnPolicyRunner(col_env, asdict(agent_cfg), device=device)

with col_env.collecting(runner.alg.actor) as episode_id:
    runner.learn(num_learning_iterations=500)

print("episode:", episode_id)

Both modes automatically:

• Auto-detect observation and action labels from Isaac Lab managers
• Capture rendered frames
• Upload data on completion
• Pass reward configuration to the server

Reward inspection

# Snapshot current reward manager state
rewards = col_env.inspect_rewards(env_idx=0)

# JSON-serializable reward config
config = col_env.reward_config_json(env_idx=0)

MuJoCo Lab Integration

The CollectionEnv wrapper in interlatent.mjlab.collection_env has the same API as IsaacSimCollectionEnv, adapted for mjlab environments. Usage is identical — see the Isaac Lab section above.

from interlatent.mjlab.collection_env import CollectionEnv

col_env = CollectionEnv(
    env,
    interlatent_client=client,
    env_name="my-mjlab-env",
    actor_obs_key="actor",  # default is "actor" for mjlab (vs "policy" for isaaclab)
)

LeRobot Integration

Two CLI entry points are provided for instrumenting LeRobot policy servers with Interlatent activation capture.

Async inference (gRPC policy server)

Drop-in replacement for lerobot's PolicyServer. Hooks into the policy after it loads and captures activations on every forward pass. Uploads all data on client disconnect or shutdown.

interlatent-rollout \
    --host=0.0.0.0 \
    --port=8080 \
    --fps=30 \
    --layer=auto \
    --api-key="ilat_..." \
    --env-name="my-robot-env" \
    --model-name="my-policy"    # optional — auto-derived from policy path

Then run lerobot's robot client as normal:

python -m lerobot.async_inference.robot_client \
    --robot.type=so100_follower \
    --server_address=HOST:8080

Sync inference (local rollout)

Replaces lerobot's teleop loop — runs policy inference locally on the robot with Interlatent instrumentation:

interlatent-sync-rollout \
    --robot.type=so100_follower \
    --robot.port=/dev/tty.usbmodem58760431541 \
    --robot.cameras="{ front: {type: opencv, index_or_path: 0, width: 640, height: 480, fps: 30}}" \
    --robot.id=black \
    --pretrained_name_or_path=user/model \
    --policy_type=smolvla \
    --task="pick up the cube" \
    --fps=30 \
    --layer=auto \
    --api-key="ilat_..."

The environment and model must already exist in the Interlatent dashboard. The model name is derived from the policy path (HuggingFace repo org/model-name becomes model-name; local file uses the basename). Override with --model-name.

VLA Layer Inspection

The vla module provides a utility for discovering hookable layers in vision-language-action models:

from interlatent.vla.hook_vla_layers import inspect_policy

attn_layers, linear_layers = inspect_policy(
    policy,
    max_attn_layers=10,
    max_linear_layers=10,
    verbose=True,
)

# Each layer has: full_name, module, kind ("attention"/"linear"), metadata
for layer in attn_layers:
    print(f"{layer.full_name}: heads={layer.num_heads}, dim={layer.embed_dim}")

This is an inspection utility — it is not integrated into watch() and does not collect activations on its own.

HTTP Resources

The client also exposes HTTP resource objects for direct API access:

client = Interlatent(api_key="ilat_...")

# Environments
envs = client.environments.list()
env = client.environments.create(slug="ant-v5", display_name="Ant-v5")

# Episodes (also aliased as client.runs)
episode = client.episodes.retrieve("episode-id")
status = client.episodes.status("episode-id")
results = client.episodes.results("episode-id")
meta = client.episodes.meta("episode-id")
chunk = client.episodes.chunk("episode-id", 0)
frame_bytes = client.episodes.frame("episode-id", "frame_0000000.jpg")

# Models
model_config = client.models.get("my-model-id")
job = client.models.process("my-model-id", sae_k=32)

# Latents
latents = client.latents.list("model-id")
latent = client.latents.retrieve("model-id", 0)

# Checkpoints / analysis reports
reports = client.checkpoints.list("model-id")
report = client.checkpoints.retrieve("model-id", "report-id")

Resource	Methods
client.index	retrieve()
client.environments	list(), create(), episodes()
client.episodes	retrieve(), create(), update(), upload_urls(), upload_complete(), status(), results(), wait(), meta(), chunk(), frame()
client.runs	Alias for client.episodes
client.models	get(), process()
client.latents	list(), retrieve(), create()
client.checkpoints	list(), retrieve(), create()
client.auth	login()

Running the Demo Script

The repository includes a full end-to-end demo at scripts/demo_processing.py:

# Install dependencies
pip install interlatent stable-baselines3 gymnasium box2d-py

# Run with the hosted API
python scripts/demo_processing.py --api-key "ilat_..."

# Customize training and collection
python scripts/demo_processing.py \
    --api-key "ilat_..." \
    --train-steps 50000 \
    --collect-steps 5000 \
    --sae-k 64

# Skip training and load a saved model
python scripts/demo_processing.py \
    --skip-train \
    --model-path models/lunarlander.zip \
    --api-key "ilat_..."

The demo trains a PPO agent on LunarLander-v3, collects activations with failure rules, uploads to the server, triggers the full analysis pipeline, polls until completion, and prints the results including the dashboard URL.

Environment Management

Create and configure environments programmatically:

client.create_environment(
    env_id="my-robot-env",
    slug="my-robot-env",
    display_name="My Robot Environment",
    robot_type="so100",
    num_cameras=2,
    camera_names=["front", "wrist"],
    task_description="Pick and place task",
    environment_type="robotics",
    vlm_enabled=True,
    failure_cases={"dropped": "object falls from gripper"},
    vlm_scoring_targets={"grasp_quality": "How well is the robot grasping the object?"},
)

Context Manager

The client supports context manager usage for automatic cleanup:

with Interlatent(api_key="...", model_id="my-policy") as client:
    client.watch(model, env, layer="auto")
    # ... collect data ...
    client.upload()
# client.close() called automatically