Metadata-Version: 2.4
Name: interlatent
Version: 0.4.12
Summary: Interlatent SDK — hosted API client for robot post-training.
Author: Interlatent Contributors
License: MIT
Project-URL: Homepage, https://github.com/seanpixel/interlatent
Project-URL: Issues, https://github.com/seanpixel/interlatent/issues
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.31.0
Requires-Dist: torch>=2.0
Requires-Dist: numpy>=1.21
Requires-Dist: grpcio>=1.60
Requires-Dist: protobuf>=4.25
Requires-Dist: sonora>=0.2.2
Provides-Extra: mjlab
Provides-Extra: isaaclab
Provides-Extra: lerobot
Requires-Dist: huggingface_hub>=0.20; extra == "lerobot"
Requires-Dist: lerobot>=0.4.0; extra == "lerobot"
Requires-Dist: pyarrow>=14; extra == "lerobot"
Requires-Dist: Pillow>=9; extra == "lerobot"

# Interlatent Python SDK

Official Python SDK for the [Interlatent](https://interlatent.com) 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

```bash
pip install interlatent
```

Optional extras for environment integrations:

```bash
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

```python
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

```python
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.

```python
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:

```python
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.

```python
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:

```python
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.

```python
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.

```python
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:

```python
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

```python
# 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:

```python
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:

```python
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:

```python
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

```python
# 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.

```python
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.

```bash
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:

```bash
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:

```bash
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:

```python
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:

```python
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`:

```bash
# 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:

```python
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:

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