OpenPipe Agent Reinforcement Training (ART)

ART automatically logs metrics every time model.log(...) is called. These are stored in history.jsonl in the run directory and can be sent to W&B if enabled.

Automatically Logged Metric Types

Derived Metrics

ART derives several metrics if the underlying inputs are provided:

• Cumulative metrics: e.g., time/cum/trainer_s, data/cum/num_unique_scenarios, costs/cum/all.
• Cost rollups: costs/train, costs/eval, costs/all.
• Throughput: throughput/avg_trainer_tok_per_s, throughput/avg_actor_tok_per_s (requires data/step_actor_tokens and time/step_actor_s).

To train an agent in an OpenEnv environment, define an asynchronous rollout function with the following pattern:

1. Reset: Call await asyncio.to_thread(env_client.reset) to initialize the environment state.
2. Initialize Trajectory: Create an art.Trajectory object containing your initial system prompt.
3. Generate Action: Use await model.openai_client().chat.completions.create(...) to get a model response.
4. Execute Step: Send the action to the environment using await asyncio.to_thread(env_client.step, action) to receive an observation and reward.
5. Record Results: Append the model's choice to traj.messages_and_choices and update traj.reward with the environment's reward.
6. Finalize: Return traj.finish().

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()

RULER (Relative Universal LLM-Elicited Rewards) is a general-purpose reward function that uses an LLM-as-judge to rank multiple agent trajectories. It requires no labeled data or hand-crafted reward functions. It works by comparing trajectories against each other and scoring them from 0 to 1 based on goal achievement, which is ideal for GRPO training as it provides the necessary relative rankings.

import art
from art.rewards import ruler_score_group

# Create a TrajectoryGroup from your trajectories
group = art.TrajectoryGroup([...])  # List of art.Trajectory objects

# Use RULER to score them
judged_group = await ruler_score_group(
    group,
    "openai/o3",
    debug=True  # Shows the judge's reasoning
)

# Access the scores
if judged_group:
    for traj in judged_group.trajectories:
        print(f"Reward: {traj.reward}")
        print(f"RULER explanation: {traj.logs[-1]}")

Use the art.TrainableModel class to initialize the client for generating tokens and training. You must provide a name (for observability platforms like W&B), a project (to group metrics for a specific task), and a base_model (the starting model).

After initialization, you must register a backend (such as ServerlessBackend or LocalBackend) using await model.register(backend) to enable inference and training.

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",
)

# Register a backend (e.g., ServerlessBackend or LocalBackend)
await model.register(backend)

The complete MCP•RL training pipeline involves:

1. Server Discovery: Querying the MCP server for tools.
2. Scenario Generation: Creating training tasks using generate_scenarios.
3. Trajectory Gathering: Running the model on scenarios to create TrajectoryGroup objects.
4. RULER Scoring: Evaluating those trajectories with ruler_score_group.
5. Reinforcement Learning: Training the model using the scored groups via backend.train.

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
result = await backend.train(model, scored_groups, learning_rate=1e-5)
await model.log(scored_groups, metrics=result.metrics, step=result.step, split="train")