### Run ART Server Locally with Python

Source: https://art.openpipe.ai/getting-started/installation-setup

Example Python code demonstrating how to initialize and register a `TrainableModel` with a `LocalBackend`. This code assumes the `backend` extra has been installed and is intended for local execution on a machine with a GPU.

```python
from art import TrainableModel, gather_trajectory_groups
from art.local.backend import LocalBackend

backend = LocalBackend()

model = TrainableModel(
    name="agent-001",
    project="my-agentic-task",
    base_model="OpenPipe/Qwen3-14B-Instruct",
)

await model.register(backend)

... the rest of your code ...
```

--------------------------------

### Install ART Client

Source: https://art.openpipe.ai/getting-started/installation-setup

Installs the core ART client package using pip. This is the base installation required for all ART functionalities.

```bash
pip install openpipe-art
```

--------------------------------

### Use ART Managed Autoscaling Backend with Python

Source: https://art.openpipe.ai/getting-started/installation-setup

Example Python code for initializing and registering a `TrainableModel` with a `ServerlessBackend`. This setup directs inference and training to the autoscaling W&B Training cluster, eliminating the need for local GPU management.

```python
from art import TrainableModel, gather_trajectory_groups
from art.serverless.backend import ServerlessBackend

backend = ServerlessBackend()

model = TrainableModel(
    name="agent-001",
    project="my-agentic-task",
    base_model="OpenPipe/Qwen3-14B-Instruct",
)

await model.register(backend)

... the rest of your code ...
```

--------------------------------

### Install ART Backend Dependencies

Source: https://art.openpipe.ai/getting-started/installation-setup

Installs the ART client along with backend dependencies required for local training and inference on machines with a GPU. This enables local execution of ART's advanced features.

```bash
pip install openpipe-art[backend]
```

--------------------------------

### Install ART for Remote GPUs with SkyPilot

Source: https://art.openpipe.ai/getting-started/installation-setup

Installs the ART client with the optional SkyPilot dependency, enabling connections to remote servers and automatic GPU provisioning for ART tasks.

```bash
pip install openpipe-art[skypilot]
```

--------------------------------

### Run ART with Remote Dedicated GPUs via SkyPilot

Source: https://art.openpipe.ai/getting-started/installation-setup

Example Python code for initializing a `SkyPilotBackend` and registering a `TrainableModel`. This configuration connects the local ART client to a remote cluster managed by SkyPilot, allowing for GPU-intensive operations on provisioned hardware.

```python
from art import TrainableModel, gather_trajectory_groups
from art.skypilot.backend import SkyPilotBackend

backend = await SkyPilotBackend.initialize_cluster(
    cluster_name="my-cluster",
    gpu="H100"
)

model = TrainableModel(
    name="agent-001",
    project="my-agentic-task",
    base_model="OpenPipe/Qwen3-14B-Instruct",
)

await model.register(backend)

... the rest of your code ...
```

--------------------------------

### Complete Email Agent Example Setup

Source: https://art.openpipe.ai/integrations/langgraph-integration

This Python code sets up a complete, runnable example for training a LangGraph email search agent. It includes imports for asyncio, dataclasses, typing, art, weave, langchain_core, langgraph, litellm, pydantic, and tenacity. It also initializes the model and backend, and defines various Pydantic data models for scenarios and responses.

```python
import asyncio
import uuid
from dataclasses import asdict
from textwrap import dedent
from typing import List

import art
import weave
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from litellm import acompletion
from pydantic import BaseModel, Field
from tenacity import retry, stop_after_attempt

from art.langgraph import init_chat_model, wrap_rollout
from art.utils import iterate_dataset

# Initialize model and backend
model = art.Model(name="Qwen/Qwen2.5-7B-Instruct")
backend = art.backends.SkyPilotBackend()

# Data models
class EmailResult(BaseModel):
    message_id: str
    subject: str
    from_address: str
    date: str
    snippet: str

class FinalAnswer(BaseModel):
    answer: str
    source_ids: List[str]

class Scenario(BaseModel):
    id: str
    question: str
    answer: str
    inbox_address: str
    query_date: str

class EmailScenario(BaseModel):
    step: int
    scenario: Scenario

class ProjectTrajectory(art.Trajectory):
    final_answer: FinalAnswer | None = None

class CorrectnessJudgeResponse(BaseModel):
    reasoning: str = Field(description="Explanation of the reasoning process.")
    accept: bool = Field(description="Whether the AI answer should be accepted.")

```

--------------------------------

### Install openpipe-art Package

Source: https://art.openpipe.ai/index

This command installs the openpipe-art package, which is required to use the ART client on your machine. It enables you to integrate ART into your existing Python projects for training agents.

```bash
pip install openpipe-art

```

--------------------------------

### Install ART with LangGraph extras

Source: https://art.openpipe.ai/integrations/langgraph-integration

Installs the openpipe-art package with the necessary extras for LangGraph integration and the training backend. This command ensures all required dependencies for using ART with LangGraph are present.

```bash
uv pip install -U openpipe-art[backend,langgraph]>=0.4.9
```

--------------------------------

### Initialize ART TrainableModel in Python

Source: https://art.openpipe.ai/fundamentals/art-client

Initializes the ART TrainableModel for generating tokens and training. It requires a model name, project name, and the base model to train from. This setup is crucial for connecting with observability platforms and ensuring consistent metric grouping.

```python
import art

model = art.TrainableModel(
    # the name of your model as it will appear in W&B
    # and other observability platforms
    name="agent-001",
    # keep your project name constant between all the models you train
    # for a given task to consistently group metrics
    project="my-agentic-task",
    # the model that you want to train from
    base_model="OpenPipe/Qwen3-14B-Instruct",
)
```

--------------------------------

### Run Summarizer Model Training Script

Source: https://art.openpipe.ai/tutorials/summarizer

Executes the Python training script for the summarizer model. This command assumes that dependencies have been installed and the environment is set up. The training process involves spinning up a cluster, registering the model, downloading checkpoints, training, and uploading final checkpoints.

```bash
uv run python src/summarizer/train.py
```

--------------------------------

### Define Training Scenarios in Python

Source: https://art.openpipe.ai/fundamentals/art-client

Scenarios define specific examples for the agent to learn from. Each scenario can have unique fields that represent real-world variations. This code defines a `Scenario` class and a list of sample scenarios.

```python
class Scenario:
    # add whatever fields differ from one real-world scenario to another
    field_1: str
    field_2: float

scenarios = [
    Scenario(
        field_1= "hello",
        field_2= 0
    ),
    Scenario(
        field_1= "world!",
        field_2= 1
    )
]
```

--------------------------------

### Initialize SkyPilotBackend Cluster for ART

Source: https://art.openpipe.ai/fundamentals/art-backend

Initializes a SkyPilotBackend cluster for remote training. This involves provisioning a remote machine with a GPU, installing openpipe-art, setting up a LocalBackend with a training server, and forwarding requests. It requires specifying cluster details like name, ART version, environment variables, and GPU type.

```python
from art.skypilot import SkyPilotBackend

backend = await SkyPilotBackend.initialize_cluster(
    # name of the cluster in SkyPilot's registry
    cluster_name="my-cluster",
    # version of openpipe-art that should be installed on the remote cluster
    # default to version installed on the client
    art_version="0.3.12",
    # path to environment variables (e.g. WANDB_API_KEY) to set on the remote cluster
    env_path=".env",
    # the GPU the cluster is equipped with
    gpu="H100"
    # alternatively, more complicated requirements can be specified in
    # the `resources` argument
)
```

--------------------------------

### Train Agent with OpenEnv Echo Environment using Python

Source: https://art.openpipe.ai/integrations/openenv-integration

This code example demonstrates training an AI agent using OpenEnv's echo environment with ART. It sets up a ServerlessBackend, defines a trainable model, creates a pool of environment clients, and runs a training loop. The `rollout` function defines how the agent interacts with the environment. Dependencies include `asyncio`, `datetime`, `art`, `art.serverless.backend`, `dotenv`, `env.echo_env`, and `weave`.

```python
import asyncio
from datetime import datetime

import art
from art.serverless.backend import ServerlessBackend
from dotenv import load_dotenv
from envs.echo_env import EchoAction, EchoEnv
import weave

PROMPT = "Use at most 100 tokens; maximize the total character length of the output."
NUM_STEPS = 50
ROLLOUTS_PER_GROUP = 4


# The rollout function defines how your agent interacts with the environment
async def rollout(model: art.TrainableModel, env_client: EchoEnv) -> art.Trajectory:
    # Reset the environment to get initial state
    await asyncio.to_thread(env_client.reset)

    # Create a trajectory to store interactions and rewards
    traj = art.Trajectory(
        messages_and_choices=[{"role": "system", "content": PROMPT}],
        reward=0.0
    )

    # Use the model to generate an action
    choice = (
        await model.openai_client().chat.completions.create(
            model=model.inference_model_name,
            messages=traj.messages(),
            max_completion_tokens=100,
            timeout=30,
        )
    ).choices[0]
    reply = (choice.message.content or "").strip()

    # Send the action to the environment and get observation/reward
    result = await asyncio.to_thread(
        env_client.step,
        EchoAction(message=reply)
    )

    # Record the model's output and reward
    traj.messages_and_choices.append(choice)
    traj.reward = result.reward

    return traj.finish()


async def main() -> None:
    load_dotenv()
    weave.init("openenv-demo")

    # Set up the training backend
    backend = ServerlessBackend()

    # Define the model to train
    model = art.TrainableModel(
        name=f"openenv-echo-{datetime.now().strftime('%Y-%m-%d-%H%M%S')}",
        project="openenv-demo",
        base_model="OpenPipe/Qwen3-14B-Instruct",
    )
    await model.register(backend)

    # Create a pool of environment clients for efficient training
    env_pool = [
        EchoEnv.from_docker_image("quixote13/echo-env:latest")
        for _ in range(ROLLOUTS_PER_GROUP)
    ]

    # Training loop
    for step in range(await model.get_step(), NUM_STEPS):
        print(f"Gathering groups for step {step}")

        # Run multiple rollouts in parallel
        groups = await art.gather_trajectory_groups([
            art.TrajectoryGroup(
                rollout(model, env_client)
                for env_client in env_pool
            )
        ])

        # Train the model on collected trajectories
        await model.train(groups)


if __name__ == "__main__":
    asyncio.run(main())

```

--------------------------------

### Initialize LangGraph Chat Model with ART

Source: https://art.openpipe.ai/integrations/langgraph-integration

Initializes a chat model using ART's integration for LangGraph, allowing for reinforcement learning-based training. Includes optional Weave tracking setup for monitoring agent interactions.

```python
import uuid
import weave
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from art.langgraph import init_chat_model
import art

# Initialize Weave tracking (optional)
if os.getenv("WANDB_API_KEY", ""):
    weave.init(model.project, settings={"print_call_link": False})
```

--------------------------------

### Example Trajectory with Compacted History (Python)

Source: https://art.openpipe.ai/features/additional-histories

An example of a Trajectory object representing a conversation that has undergone compaction. It includes the current active messages and references a previous, compacted conversation segment within `additional_histories`.

```python
trajectory = Trajectory(
    # Current active conversation
    messages_and_choices=[
        {"role": "system", "content": "You are a helpful assistant. ngoại"},
        {"role": "user", "content": "Explain quantum entanglement"},
        {"role": "assistant", "content": "Quantum entanglement is..."},
        # ... many more messages ...
    ],
    additional_histories=[
        # Previous conversation segment before compaction
        History(
            messages_and_choices=[
                {"role": "system", "content": "You are a helpful assistant. ngoại"},
                {"role": "user", "content": "Compacted conversation history: the user asked about quantum entanglement, and the assistant explained..."},
                {"role": "user", "content": "Tell me more about the history of quantum entanglement"},
                {"role": "assistant", "content": "Quantum entanglement was first..."},
            ]
        )
    ]
)
```

--------------------------------

### Initialize LocalBackend Instance

Source: https://art.openpipe.ai/fundamentals/art-backend

Demonstrates how to create an instance of the LocalBackend class. Key parameters include `in_process` to control automatic shutdown and `path` to specify the directory for logs and model weights. This is useful for running local vLLM and Unsloth/torchtune instances on a GPU.

```python
from art.local import LocalBackend

backend = LocalBackend(
    # set to True if you want your backend to shut down automatically
    # when your client process ends
    in_process=False,
    # local path where the backend will store trajectory logs and model weights
    path='./.art',
)
```

--------------------------------

### Initialize TrainableModel with SFT LoRA Adapter in Python

Source: https://art.openpipe.ai/getting-started/faq

Demonstrates how to initialize the `TrainableModel` from the ART framework using an existing Hugging Face-style LoRA adapter. This allows reinforcement learning to begin from a pre-trained state, improving efficiency. The `base_model` parameter should point to the directory containing the adapter.

```python
import art

model = art.TrainableModel(
    name="agent-001",
    project="my-agentic-task",
    base_model="/path/to/my_sft_lora_adapter",  # HF-style adapter dir
)
```

--------------------------------

### Initialize ART TrainableModel from SFT LoRA in Python

Source: https://art.openpipe.ai/fundamentals/art-client

Initializes the ART TrainableModel using an existing SFT LoRA adapter. This allows warm-starting RL training from task-aligned weights, reducing training steps and GPU costs, and stabilizing early training. The base_model parameter should point to the adapter directory.

```python
import art

model = art.TrainableModel(
    name="agent-001",
    project="my-agentic-task",
    # Point to the local SFT LoRA adapter directory
    # (e.g., contains adapter_config.json and adapter_model.bin/safetensors)
    base_model="/path/to/my_sft_lora_adapter",
)
```

--------------------------------

### Initialize ServerlessBackend for ART

Source: https://art.openpipe.ai/fundamentals/art-backend

Initializes the ServerlessBackend for remote training on autoscaling GPUs. Requires a W&B API key, which can be provided as an argument or an environment variable. This backend automatically saves LoRA checkpoints as W&B Artifacts and deploys them for inference.

```python
from art.serverless.backend import ServerlessBackend

backend = ServerlessBackend(
  api_key="my-api-key",
  # or set WANDB_API_KEY in the environment
)
```

--------------------------------

### Run Model Benchmark Script

Source: https://art.openpipe.ai/tutorials/summarizer

Executes the benchmark script to compare AI model performance against specified models. Requires OPENROUTER_API_KEY and AWS credentials for S3 uploads. The script iterates through validation documents, records accuracy percentages, and uploads results.

```bash
uv run python benchmarks/benchmark_models.py
```

--------------------------------

### Basic MCP Server Training Pipeline with ART

Source: https://art.openpipe.ai/features/mcp-rl

This comprehensive Python script demonstrates a basic training pipeline for an MCP server using ART. It initializes a trainable model, generates scenarios, gathers trajectories, scores them with RULER, and trains the model.

```python
import art
from art.mcp import generate_scenarios
from art.rewards import ruler_score_group
from art import gather_trajectory_groups

# Initialize the model
model = art.TrainableModel(
    model="OpenPipe/Qwen3-14B-Instruct",
    openrouter_api_key="your_openrouter_key"
)

# Generate training scenarios automatically
scenario_collection = await generate_scenarios(
    tools=tools_list,
    resources=resources_list,
    num_scenarios=100,
    show_preview=False,
    generator_model="gpt-4o-mini",
    generator_api_key="your_openrouter_key",
)

# Gather trajectory groups
groups = await gather_trajectory_groups(
    (
        art.TrajectoryGroup(
            rollout(model, scenario, False)
            for _ in range(4)  # rollouts per group
        )
        for scenario in scenario_collection
    ),
    pbar_desc="train gather step",
)

# Score groups using RULER
scored_groups = [
    await ruler_score_group(
        group,
        judge_model="gpt-4o-mini",
        debug=True,
        swallow_exceptions=True
    )
    for group in groups
]

# Train the model
await model.train(
    scored_groups,
    config=art.TrainConfig(learning_rate=1e-5),
)
```

--------------------------------

### Combine RULER with Original Rewards (Preserving)

Source: https://art.openpipe.ai/fundamentals/ruler

This example shows how to combine RULER scores with pre-existing 'independent_reward' metrics calculated within a rollout function. RULER automatically preserves these original rewards, allowing them to be added to the final trajectory reward after judging.

```python
# Your trajectories already have rewards from rollout
judged_group = await ruler_score_group(group, "openai/o3", debug=True)

# Combine RULER scores with original rewards
for traj in judged_group.trajectories:
    traj.reward += traj.metrics["independent_reward"]
```

--------------------------------

### Implement Checkpoint Deletion within Training Loop (Python)

Source: https://art.openpipe.ai/features/checkpoint-deletion

Provides an example of integrating checkpoint deletion into a training loop to manage storage space efficiently. After each training step, `delete_checkpoints` is called, configured to use the `train/reward` metric. This ensures that only the most recent and best-performing checkpoint according to `train/reward` is kept, significantly reducing storage usage from ~6GB to ~240MB.

```python
import art
from art.serverless.backend import ServerlessBackend

from .rollout import rollout
from .scenarios load_train_scenarios

TRAINING_STEPS = 50

model = art.TrainableModel(
    name="agent-001",
    project="checkpoint-deletion-demo",
    base_model="OpenPipe/Qwen3-14B-Instruct",
)
backend = ServerlessBackend()
await model.register(backend)


train_scenarios = load_train_scenarios()

# training loop
for _step in range(await model.get_step(), TRAINING_STEPS):
    train_groups = await art.gather_trajectory_groups(
        (
            art.TrajectoryGroup(rollout(model, scenario, step) for _ in range(8))
            for scenario in train_scenarios
        ),
        pbar_desc=f"gather(train:{step})",
    )
    # trains model and automatically persists each LoRA as a W&B Artifact
    # ~120MB per step
    await model.train(
        train_groups,
        config=art.TrainConfig(learning_rate=5e-5),
    )
    # clear all but the most recent and best-performing checkpoint on the train/reward metric
    await model.delete_checkpoints(best_checkpoint_metric="train/reward")

# ~240MB of storage used by checkpoints
```

--------------------------------

### Pass Extra LiteLLM Parameters to RULER (Python)

Source: https://art.openpipe.ai/fundamentals/ruler

This Python code demonstrates how to pass additional parameters to LiteLLM when using the RULER scoring function. It shows how to adjust parameters like `temperature` and `max_tokens` to fine-tune the LLM judge's behavior. It also includes an example of specifying a custom `api_base` for local model deployments, enabling more granular control over the judging process.

```python
# Adjust temperature and max tokens
await ruler_score_group(
    group,
    "openai/o3",
    extra_litellm_params={"temperature": 0.7, "max_tokens": 1000}
)

# Use custom API base for local models
await ruler_score_group(
    group,
    "openai/gpt-4",
    extra_litellm_params={"api_base": "http://localhost:8000"}
)

```

--------------------------------

### Backend Initialization and Model Registration

Source: https://art.openpipe.ai/fundamentals/art-backend

Illustrates a pattern for initializing different backend types (ServerlessBackend, SkyPilotBackend, LocalBackend) based on a configuration variable `BACKEND_TYPE`. After initializing the chosen backend, it shows how to register a trainable model with it for subsequent training operations.

```python
BACKEND_TYPE = "serverless"

if BACKEND_TYPE == "serverless":
    from art.serverless.backend import ServerlessBackend
    backend = await ServerlessBackend()
else if BACKEND_TYPE=="remote":
    from art.skypilot import SkyPilotBackend
    backend = await SkyPilotBackend.initialize_cluster(
        cluster_name="my-cluster",
        gpu="H100"
    )
else:
    from art.local import LocalBackend
    backend = LocalBackend()

model = art.TrainableModel(...)

await model.register(backend)

# ...training code...
```

--------------------------------

### Generate and Rank Joke Trajectories with RULER (Python)

Source: https://art.openpipe.ai/fundamentals/ruler

This Python code demonstrates a complete example of using the RULER library to generate and rank joke quality trajectories. It sets up initial messages, creates three distinct trajectories (good, mediocre, off-topic) with different joke responses, groups them, and then uses `ruler_score_group` to score them. Finally, it prints the ranked results based on the RULER scores. Dependencies include `asyncio`, `art`, and `openai`.

```python
import asyncio
import art
from art.rewards import ruler_score_group
from openai.types.chat.chat_completion import Choice
from openai.types.chat import ChatCompletionMessage

async def main():
    # Initial messages shared by all trajectories
    initial_messages = [
        {"role": "system", "content": "You are a comedy writer. Generate funny jokes based on the given topic."},
        {"role": "user", "content": "Tell me a funny joke about computers"}
    ]

    # Create three trajectories with different quality responses
    good_trajectory = art.Trajectory(
        messages_and_choices=[
            *initial_messages,
            Choice(finish_reason="stop", index=0, message=ChatCompletionMessage(
                role="assistant",
                content="Why don't computers ever get invited to parties?\n\nBecause they always crash! 🥁\n\nBut seriously, have you tried turning them off and on again?"
            ))
        ],
        reward=0.0
    )

    mediocre_trajectory = art.Trajectory(
        messages_and_choices=[
            *initial_messages,
            Choice(finish_reason="stop", index=0, message=ChatCompletionMessage(
                role="assistant",
                content="What do you call a computer that doesn't work?\n\nBroken."
            ))
        ],
        reward=0.0
    )

    off_topic_trajectory = art.Trajectory(
        messages_and_choices=[
            *initial_messages,
            Choice(finish_reason="stop", index=0, message=ChatCompletionMessage(
                role="assistant",
                content="I don't really know jokes about computers, but here's a fact: The sky is blue because of Rayleigh scattering."
            ))
        ],
        reward=0.0
    )

    # Create a TrajectoryGroup and use RULER to score
    group = art.TrajectoryGroup([good_trajectory, mediocre_trajectory, off_topic_trajectory])
    judged_group = await ruler_score_group(group, "openai/o3", debug=True)

    # Display rankings
    if judged_group:
        sorted_trajectories = sorted(judged_group.trajectories, key=lambda t: t.reward, reverse=True)
        for rank, traj in enumerate(sorted_trajectories, 1):
            messages = traj.messages()
            print(f"Rank {rank}: Score {traj.reward:.3f}")
            print(f"  Response: {messages[-1]['content'][:50]}...")

asyncio.run(main())

```

--------------------------------

### Query MCP Server for Available Tools

Source: https://art.openpipe.ai/features/mcp-rl

This Python snippet queries an MCP server to retrieve a list of available tools. It's the first step in understanding the capabilities of the MCP server for agent training.

```python
tools_list = await mcp_client.list_tools()
```

--------------------------------

### Register Backends with ART Model in Python

Source: https://art.openpipe.ai/fundamentals/art-client

Registers different backend types (ServerlessBackend, SkyPilotBackend, LocalBackend) with the initialized ART model. This step connects the model to the chosen compute environment for training and inference. Ensure the necessary backend classes are imported.

```python
# managed training
backend = ServerlessBackend()

# remote training
backend = SkyPilotBackend.initialize_cluster(
    cluster_name="art", gpu="H100"
)

# local training
backend = LocalBackend()

await model.register(backend)
```

--------------------------------

### Fork Checkpoint Locally - Python

Source: https://art.openpipe.ai/features/checkpoint-forking

Demonstrates how to fork a training run from an existing local checkpoint using the OpenPipe Python SDK. This is useful for restarting training from a known good state or experimenting with hyperparameters. It requires specifying the new model's name, project, base model, and the source model for forking, along with an optional step limit.

```python
import art
from art.local import LocalBackend

async def train():
    with LocalBackend() as backend:
        # Create a new model that will fork from an existing checkpoint
        model = art.TrainableModel(
            name="my-model-v2",
            project="my-project",
            base_model="OpenPipe/Qwen3-14B-Instruct",
        )

        # Copy the checkpoint from another model
        await backend._experimental_fork_checkpoint(
            model,
            from_model="my-model-v1",
            not_after_step=500,  # Use checkpoint at or before step 500
            verbose=True,
        )

        # Register and continue training
        await model.register(backend)
        # ... rest of training code
```

--------------------------------

### Create and Run LangGraph ReAct Agent with ART

Source: https://art.openpipe.ai/integrations/langgraph-integration

Creates and runs a LangGraph ReAct agent using ART's integrated chat model. This snippet demonstrates initializing the chat model, defining tools, configuring agent execution, and invoking the agent with system and user messages.

```python
@weave.op
async def rollout(model: art.Model, email_scenario: EmailScenario) -> ProjectTrajectory:
    # Initialize chat model with temperature
    chat_model = init_chat_model(model.name, temperature=1.0)

    # Define available tools
    tools = [search_inbox_tool, read_email_tool, return_final_answer_tool]

    # Create the LangGraph ReAct agent
    react_agent = create_react_agent(chat_model, tools)

    # Configure agent execution
    config = {
        "configurable": {"thread_id": str(uuid.uuid4())},
        "recursion_limit": MAX_TURNS,
    }

    # Run the agent with system and user messages
    await react_agent.ainvoke(
        {
            "messages": [
                SystemMessage(content=system_prompt),
                HumanMessage(content=scenario.question),
            ]
        },
        config=config,
    )
```

--------------------------------

### Python Training Loop for OpenPipe AI

Source: https://art.openpipe.ai/integrations/langgraph-integration

This Python code defines the main asynchronous function for training the OpenPipe AI model. It includes setting up training scenarios, registering the model, configuring training parameters, iterating through the dataset, gathering trajectories, applying RULER scoring, and executing the model training step. Dependencies include asyncio, the openpipe-art library, and custom Scenario and EmailScenario classes.

```python
import asyncio
import openpipe_art as art

# Assume Scenario and EmailScenario are defined elsewhere
# Assume model, backend, iterate_dataset, wrap_rollout, ruler_score_group are defined elsewhere

async def main():
    # Sample training scenarios (replace with real data)
    training_scenarios = [
        Scenario(
            id="1",
            question="Find emails about the quarterly budget",
            answer="Budget meeting scheduled for Q4 review",
            inbox_address="user@company.com",
            query_date="2024-01-20"
        ),
        Scenario(
            id="2",
            question="Look for urgent project updates",
            answer="Project deadline moved to next month",
            inbox_address="user@company.com",
            query_date="2024-01-20"
        ),
    ]

    # Register model with backend
    await model.register(backend)

    # Training configuration
    training_config = {
        "groups_per_step": 2,
        "num_epochs": 3,
        "rollouts_per_group": 4,
        "learning_rate": 1e-5,
        "max_steps": 5,
    }

    # Training iterator
    training_iterator = iterate_dataset(
        training_scenarios,
        groups_per_step=training_config["groups_per_step"],
        num_epochs=training_config["num_epochs"],
        initial_step=await model.get_step(),
    )

    # Training loop
    for batch in training_iterator:
        print(f"Training step {batch.step}, epoch {batch.epoch}")

        # Create trajectory groups
        groups = []
        for scenario in batch.items:
            groups.append(
                art.TrajectoryGroup([
                    wrap_rollout(model, rollout)(
                        model, EmailScenario(step=batch.step, scenario=scenario)
                    )
                    for _ in range(training_config["rollouts_per_group"])
                ])
            )

        # Gather trajectories
        finished_groups = await art.gather_trajectory_groups(
            groups,
            pbar_desc="gather",
            max_exceptions=training_config["rollouts_per_group"] * len(batch.items),
        )

        # Apply RULER scoring
        judged_groups = []
        for group in finished_groups:
            judged_group = await ruler_score_group(group, "openai/o4-mini")
            judged_groups.append(judged_group)

        # Train model
        await model.train(
            judged_groups,
            config=art.TrainConfig(learning_rate=training_config["learning_rate"]),
        )

        print(f"Completed training step {batch.step}")

        if batch.step >= training_config["max_steps"]:
            break

if __name__ == "__main__":
    asyncio.run(main())

```

--------------------------------

### Run Inference with ART Model in Python

Source: https://art.openpipe.ai/fundamentals/art-client

Generates inference tokens using the ART model by making requests to a vLLM server on the registered backend. This code snippet shows how to obtain the OpenAI client from the model and make a chat completion request, including system and user messages, model name, and other parameters like max_tokens and timeout.

```python
openai_client = model.openai_client()

messages: art.Messages = [
    {
        "role": "system",
        "content": "...",
    },
    {
        "role": "user",
        "content": "..."
    }
]
chat_completion = await openai_client.chat.completions.create(
    messages=messages,
    model=model.name,
    max_tokens=100,
    timeout=100,
    tools=[...]
)
print(chat_completion.choices[0].message.tool_calls)
```

--------------------------------

### Generate Training Scenarios for MCP Servers

Source: https://art.openpipe.ai/features/mcp-rl

This Python code automatically generates diverse training scenarios for an MCP server. It leverages ART's `generate_scenarios` function, specifying tools, number of scenarios, and the generator model.

```python
from art.mcp import generate_scenarios

scenario_collection = await generate_scenarios(
    tools=tools_list,
    num_scenarios=24,
    show_preview=True,
    generator_model="openai/gpt-4.1-mini",
    generator_api_key="your_openrouter_key",
    generator_base_url="https://openrouter.ai/api/v1",
)
```

--------------------------------

### Tokenization Process with Additional Histories (Python)

Source: https://art.openpipe.ai/features/additional-histories

Illustrates the tokenization pipeline for a Trajectory that contains additional histories. It shows how the main history and each additional history are tokenized independently before the training weight is distributed.

```python
# Assuming create_history_from_trajectory and tokenize_history are defined elsewhere

# In the tokenization pipeline
histories = [create_history_from_trajectory(trajectory)]  # Main history
histories.extend(trajectory.additional_histories)  # Add all additional histories

# Each history is tokenized independently
for history in histories:
    tokenized_result = tokenize_history(history)
    # Weight is distributed across all results
```

--------------------------------

### Implement Rollout Function in Python

Source: https://art.openpipe.ai/fundamentals/art-client

The rollout function executes the agent through a single scenario, generating a trajectory. It uses an OpenAI client to create a completion and assigns a reward based on the agent's performance. This function is crucial for evaluating agent behavior during training.

```python
# define a rollout function that puts the model through its paces for a given scenario
async def rollout(model: art.Model, scenario: Scenario) -> art.Trajectory:
    openai_client = model.openai_client()

    trajectory = art.Trajectory(
        messages_and_choices=[
            {
                "role": "system",
                "content": "..."
            },
            {
                "role": "user",
                "content": "..."
            }
        ]
    )

    # generate a completion using the client
    chat_completion = await openai_client.chat.completions.create(
        messages=trajectory.messages(), model=model.name
    )
    choice = chat_completion.choices[0]
    trajectory.messages_and_choices.append(choice)

    # determine how well the agent did during this particular run
    agent_performance_score: float = ...
    trajectory.reward = agent_performance_score

    return trajectory
```

--------------------------------

### Execute Model Training Loop in Python

Source: https://art.openpipe.ai/fundamentals/art-client

This code snippet demonstrates the training loop for the model. It iterates for a specified number of steps, gathering trajectory groups for each scenario and then sending them to the backend for model training. The loop utilizes `art.gather_trajectory_groups` and `model.train`.

```python
async def train(): # train for 50 steps
    for _ in range(await model.get_step(), 50):
        # Trajectories produced using the same training scenario are automatically grouped
        train_groups = await art.gather_trajectory_groups(
            (
                art.TrajectoryGroup(rollout(model, scenario) for _ in range(8))
                for scenario in scenarios
            ),
            pbar_desc="gather",
        )

        print("num train groups:", len(train_groups))
        # num train groups: 2

        print("length of each train group:", len(train_groups[0]))
        # length of each train group: 8

        # send the grouped trajectories to the backend and wait until training finishes
        await model.train(
            train_groups,
            config=art.TrainConfig(learning_rate=1e-5),
        )
        # once model.train finishes for the current step
        # the backend updates the LoRA weights for inference and
        # the training loop continues until 50 steps have completed
```

--------------------------------

### Reinforcement Learning Training with RULER Feedback

Source: https://art.openpipe.ai/features/mcp-rl

This Python code trains a model using reinforcement learning based on feedback from RULER. It gathers trajectory groups, scores them using RULER, and then trains the model with specified learning rate.

```python
from art import gather_trajectory_groups

# Train using RULER feedback
groups = await gather_trajectory_groups(
    trajectory_groups_generator,
    pbar_desc="train gather step",
)

scored_groups = [
    await ruler_score_group(
        group,
        judge_model="openai/o4-mini",
    )
    for group in groups
]

await model.train(
    scored_groups,
    config=art.TrainConfig(learning_rate=1e-5),
)
```

--------------------------------

### Fork Checkpoint from S3 - Python

Source: https://art.openpipe.ai/features/checkpoint-forking

Illustrates forking a training run from a checkpoint stored in an S3 bucket using the OpenPipe Python SDK. This method is beneficial when checkpoints are backed up or stored remotely. It requires specifying the source model, S3 bucket name, and an optional step limit to select the appropriate checkpoint.

```python
await backend._experimental_fork_checkpoint(
    model,
    from_model="my-model-v1",
    from_s3_bucket="my-backup-bucket",
    not_after_step=500,
    verbose=True,
)
```

--------------------------------

### Evaluate Agent Responses with RULER

Source: https://art.openpipe.ai/features/mcp-rl

This Python snippet uses ART's RULER (Reinforcement Learning Understanding and Evaluating Responses) to evaluate agent responses without labeled data. It analyzes task accomplishment, tool usage quality, efficiency, and error handling.

```python
from art.rewards import ruler_score_group

# RULER evaluates responses without labeled data
scored_group = await ruler_score_group(
    group,
    judge_model="openai/o4-mini",
)
```

--------------------------------

### Mock Email Search and Read Functions (Python)

Source: https://art.openpipe.ai/integrations/langgraph-integration

Provides mock implementations for searching emails and reading a specific email by its message ID. These functions simulate interaction with an email service and are intended to be replaced with actual API calls. They return structured data representing email results.

```python
def search_emails(inbox: str, keywords: List[str], sent_before: str) -> List[EmailResult]:
    """Mock email search function - replace with real implementation"""
    return [
        EmailResult(
            message_id="msg_123",
            subject=f"Subject matching {keywords[0]}",
            from_address="sender@example.com",
            date="2024-01-15",
            snippet=f"Email snippet containing {keywords[0]}"
        )
    ]

def read_email(message_id: str) -> EmailResult | None:
    """Mock email read function - replace with real implementation"""
    return EmailResult(
        message_id=message_id,
        subject="Full email subject",
        from_address="sender@example.com",
        date="2024-01-15",
        snippet="Full email content here..."
    )
```

--------------------------------

### Main Rollout Function for Email Agent (Python)

Source: https://art.openpipe.ai/integrations/langgraph-integration

Orchestrates the AI agent's workflow for handling email-based scenarios. It initializes a ReAct agent, defines tools for searching and reading emails, and returns the final answer along with source message IDs. The function also includes correctness evaluation and error handling.

```python
# Main rollout function
@weave.op
async def rollout(model: art.Model, email_scenario: EmailScenario) -> ProjectTrajectory:
    scenario = email_scenario.scenario
    MAX_TURNS = 10

    traj = ProjectTrajectory(
        reward=0.0,
        messages_and_choices=[],
        metadata={
            "scenario_id": scenario.id,
            "step": email_scenario.step,
        },
    )

    system_prompt = dedent(f"""
        You are an email search agent. Use the tools to search emails and find answers.
        User's email address: {scenario.inbox_address}
        Today's date: {scenario.query_date}

        When you find the answer, use return_final_answer_tool with the answer and source message IDs.
    """)

    final_answer = None

    @tool
    def search_inbox_tool(keywords: List[str]) -> List[dict]:
        """Search inbox for emails matching keywords"""
        results = search_emails(scenario.inbox_address, keywords, scenario.query_date)
        return [asdict(result) for result in results]

    @tool
    def read_email_tool(message_id: str) -> dict | None:
        """Read a specific email by message ID"""
        email = read_email(message_id)
        return email.model_dump() if email else None

    @tool
    def return_final_answer_tool(answer: str, reference_message_ids: List[str]) -> dict:
        """Return final answer with source message IDs"""
        nonlocal final_answer
        final_answer = FinalAnswer(answer=answer, source_ids=reference_message_ids)
        return final_answer.model_dump()

    tools = [search_inbox_tool, read_email_tool, return_final_answer_tool]
    chat_model = init_chat_model(model.name, temperature=1.0)
    react_agent = create_react_agent(chat_model, tools)

    try:
        config = {
            "configurable": {"thread_id": str(uuid.uuid4())},
            "recursion_limit": MAX_TURNS,
        }

        await react_agent.ainvoke({
            "messages": [
                SystemMessage(content=system_prompt),
                HumanMessage(content=scenario.question),
            ]
        }, config=config)

        if final_answer:
            traj.final_answer = final_answer
            correctness_judge_response = await judge_correctness(scenario, final_answer.answer)
            traj.metrics["correct"] = float(correctness_judge_response.accept)

    except Exception as e:
        print(f"Error running agent: {e}")
        traj.messages_and_choices.append({"role": "assistant", "content": f"Error: {str(e)}"})

    return traj
```

--------------------------------

### Shut Down SkyPilot Cluster via CLI

Source: https://art.openpipe.ai/tutorials/summarizer

Command-line interface command to shut down a SkyPilot managed cluster. Replace `<cluster-name>` with the actual name of your cluster. This is a recommended step after completing training and benchmarks to avoid incurring unnecessary costs.

```bash
uv run sky down <cluster-name>
```