Evaluation

This guide covers running evaluations, computing metrics, and saving results.

TrajectoryRunner

The TrajectoryRunner orchestrates evaluation trajectories:

from llenvs.core.registry import environment_registry
from llenvs.inference.backends import OpenAIBackend
from llenvs.inference import SamplingParams, build_standard_pipeline
from llenvs.evaluation import TrajectoryRunner

env = environment_registry.get(name="leg_counting", adapter="reasoning_gym", size=100, seed=42)
backend = OpenAIBackend(model="gpt-4o")

runner = TrajectoryRunner(
    environment=env,
    backend=backend,
    sampling_params=SamplingParams(temperature=0.0, max_tokens=1024),
    prompt_pipeline=build_standard_pipeline(
        system_prompt="You are a helpful assistant.",
        use_cot=True,
        answer_format="xml_answer",
    ),
)

# Run single trajectory
result = runner.run_trajectory(task_index=0)
print(f"Success: {result.success}")
print(f"Total reward: {result.total_reward}")

# Run batch
batch_result = runner.run_batch(
    task_indices=list(range(50)),
    progress_callback=lambda c, t: print(f"\r{c}/{t}", end=""),
)
print(f"\nSuccess rate: {batch_result.success_rate:.2%}")

Tool Environments

TrajectoryRunner auto-detects tool environments via available_tools on the observation and uses generate_with_tools_batch() automatically:

from llenvs.core.registry import environment_registry
from llenvs.evaluation import TrajectoryRunner

env = environment_registry.get(name="math:GSM8K", adapter="gem", tool_types=("python",))

runner = TrajectoryRunner(
    environment=env,
    backend=backend,
    sampling_params=SamplingParams(temperature=0.0),
    system_prompt="Use Python to solve. Submit your final answer.",
)

result = runner.run_trajectory(task_index=0)

Batched Evaluation

run_batch() automatically batches inference calls using lockstep execution. All active trajectories advance one step together, and the backend's generate_chat_batch() (or generate_with_tools_batch()) is called once per step with all active conversations. Trajectories that finish early drop out of subsequent batches.

This applies to both runner types: - TrajectoryRunner: Batches generate_chat_batch() calls per step. For tool environments, uses generate_with_tools_batch() automatically. - SegmentedTrajectoryRunner: Batches segment generation via generate_segment_batch() on the continuation strategy. Callbacks (step_callback) still run per-trajectory after each step.

This gives significant speedups: - vLLM / HuggingFace: All prompts go through the GPU in one batched call per step. - API backends (OpenAI, Anthropic, OpenRouter): Concurrent async HTTP requests limited by max_concurrency.

Single-trajectory run_trajectory() is unchanged.

# This automatically batches inference across all 100 tasks
batch_result = runner.run_batch(
    task_indices=list(range(100)),
    progress_callback=lambda c, t: print(f"\r{c}/{t}", end=""),
)

Use batch_size to limit how many trajectories run in each lockstep batch (useful for GPU memory management):

# Process 32 tasks at a time instead of all 100
batch_result = runner.run_batch(
    task_indices=list(range(100)),
    batch_size=32,
)

The same progress_callback(completed, total) pattern is also available on run_batch_from_states() and run_batch_from_state_actions(), which is useful for Monte-Carlo rollout code that starts from arbitrary states instead of resetting tasks from scratch.

For cross-environment batching (interleaving trajectories from multiple environments into a single lockstep loop), see run_multi_evaluation() in the Parallelization guide.

See the Parallelization guide for architecture details, max_concurrency tuning, and performance tips.

Logging

Structured logging during evaluation runs is configured via LogConfig. Three targets are available: console (Python logging), file (JSONL), and W&B.

from llenvs.evaluation import LogConfig

# Console logging (uses Python logging at INFO/DEBUG levels)
result = run_evaluation(env, backend, log=LogConfig(targets=("console",)))

# File logging (JSONL in .logs/{env_name}/)
result = run_evaluation(env, backend, log=LogConfig(targets=("file",)))

# W&B logging
result = run_evaluation(env, backend, log=LogConfig(targets=("wandb",), wandb_project="my-evals"))

# Multiple targets
result = run_evaluation(env, backend, log=LogConfig(targets=("console", "file", "wandb")))

# W&B with existing run (e.g. from verl)
result = run_evaluation(env, backend, log=LogConfig(targets=("wandb",), wandb_run=existing_run))

The log parameter is available on TrajectoryRunner, SegmentedTrajectoryRunner, run_evaluation(), run_segmented_evaluation(), and run_multi_evaluation() (via per-entry runner configs).

File logging writes JSONL to {log_dir}/{env_name}/{timestamp}.jsonl with one JSON object per line. Each line has an "event" key (batch_start, step, trajectory_end, batch_end, error).

Console logging uses logging.getLogger("llenvs.evaluation"). To see step-level details, set the logger to DEBUG:

import logging
logging.getLogger("llenvs.evaluation").setLevel(logging.DEBUG)

progress_callback is unchanged and orthogonal to logging.

Metrics and Statistics

The evaluation module separates metrics (what we measure) from statistics (how we summarize):

• Metrics: Measurable quantities like action_reward, trajectory_reward, accuracy
• Statistics: Summary computations like mean, std_dev, quantiles, confidence intervals

Two statistics types exist for different metric categories:

• ContinuousStatistics: For continuous-valued metrics (rewards, scores, etc.)
• BinaryStatistics: For binary metrics (success/failure) with pass@k support

Computing Metrics

from llenvs.evaluation import (
    compute_accuracy,
    compute_trajectory_reward,
    compute_action_reward,
    compute_format_compliance,
)

# Binary metrics (accuracy, format_compliance) return BinaryStatistics
accuracy = compute_accuracy(batch_result.trajectory_results)
stats = accuracy.statistics
print(f"Accuracy: {stats.mean:.3f} ({stats.count}/{stats.n})")
print(f"95% CI: [{stats.ci_lower:.3f}, {stats.ci_upper:.3f}]")

# Continuous metrics (rewards) return ContinuousStatistics
trajectory_reward = compute_trajectory_reward(batch_result.trajectory_results)
stats = trajectory_reward.statistics
print(f"Mean trajectory reward: {stats.mean:.3f}")
print(f"Std dev: {stats.std_dev:.3f}")
print(f"Min: {stats.min:.3f}, Max: {stats.max:.3f}")

# Action-level metrics
action_reward = compute_action_reward(batch_result.trajectory_results)
stats = action_reward.statistics
print(f"Mean action reward: {stats.mean:.3f}")
print(f"Median: {stats.median:.3f}")
print(f"Q25-Q75: [{stats.q25:.3f}, {stats.q75:.3f}]")

format_compliance = compute_format_compliance(batch_result.trajectory_results)
stats = format_compliance.statistics
print(f"Format compliance: {stats.mean:.1%} ({stats.count}/{stats.n} actions)")

Statistics Types

ContinuousStatistics (for rewards, scores, etc.):

Field	Description
n	Sample size
mean	Arithmetic mean
std_dev	Sample standard deviation
std_error	Standard error of the mean
min	Minimum value
max	Maximum value
median	Median (50th percentile)
q25	25th percentile
q75	75th percentile
ci_lower	Lower bound of 95% CI
ci_upper	Upper bound of 95% CI

BinaryStatistics (for success/failure metrics):

Field	Description
n	Sample size
mean	Success rate (proportion of successes)
count	Number of successes
std_error	Standard error sqrt(p(1-p)/n)
ci_lower	Lower bound of Wilson score CI
ci_upper	Upper bound of Wilson score CI

BinaryStatistics also provides the pass_at_k(k) method.

Computing Statistics Directly

You can compute statistics from any sequence of values:

from llenvs.evaluation import compute_continuous_statistics, compute_binary_statistics

# Continuous data
values = [0.8, 0.85, 0.9, 0.78, 0.92]
stats = compute_continuous_statistics(values, confidence_level=0.95)
print(f"Mean: {stats.mean:.3f}")
print(f"Std dev: {stats.std_dev:.3f}")
print(f"Range: [{stats.min:.3f}, {stats.max:.3f}]")

# Binary data
successes = [1, 1, 0, 1, 0, 1, 1, 0, 1, 1]  # 7 successes out of 10
stats = compute_binary_statistics(successes)
print(f"Success rate: {stats.mean:.1%} ({stats.count}/{stats.n})")

Pass@k

Pass@k measures the probability of at least one success in k samples. It's available as a method on BinaryStatistics:

# Run multiple times per task
results = []
for _ in range(10):  # 10 samples
    result = runner.run_trajectory(task_index=0)
    results.append(result)

# Get accuracy statistics (BinaryStatistics)
accuracy = compute_accuracy(results)
stats = accuracy.statistics

# Compute Pass@k for different values of k
print(f"Pass@1: {stats.pass_at_k(1):.3f}")
print(f"Pass@5: {stats.pass_at_k(5):.3f}")
print(f"Pass@10: {stats.pass_at_k(10):.3f}")

The pass_at_k(k) method uses the exact formula: 1 - C(n-c, k) / C(n, k) where n is the total number of samples and c is the number of successes.

All Metrics at Once

from llenvs.evaluation import compute_all_metrics, ContinuousStatistics, BinaryStatistics

metrics = compute_all_metrics(batch_result)

for name, metric in metrics.metrics.items():
    stats = metric.statistics
    if isinstance(stats, ContinuousStatistics):
        print(f"{name}: {stats.mean:.4f} (std: {stats.std_dev:.4f})")
    else:  # BinaryStatistics
        print(f"{name}: {stats.mean:.4f} ({stats.count}/{stats.n})")

Aggregating Metrics

Combine metrics from multiple evaluations:

from llenvs.evaluation import (
    aggregate_continuous_metrics,
    aggregate_binary_metrics,
)

# Aggregate trajectory rewards from multiple task groups
math_rewards = compute_trajectory_reward(math_results)
logic_rewards = compute_trajectory_reward(logic_results)

combined = aggregate_continuous_metrics(
    [math_rewards, logic_rewards],
    name="all_reasoning_reward",
)
print(f"Combined: {combined.statistics.mean:.3f} (n={combined.statistics.n})")

# Aggregate accuracy across task groups
math_acc = compute_accuracy(math_results)
logic_acc = compute_accuracy(logic_results)

combined_acc = aggregate_binary_metrics([math_acc, logic_acc])
print(f"Combined accuracy: {combined_acc.statistics.mean:.1%}")
print(f"Pass@5: {combined_acc.statistics.pass_at_k(5):.3f}")

Note: Aggregated continuous metrics have median, q25, q75 set to None because these cannot be accurately reconstructed from summary statistics alone.

Saving Results

from datetime import datetime
from llenvs.evaluation.results import create_evaluation_result, print_summary

# Create formatted result
eval_result = create_evaluation_result(
    batch_result=batch_result,
    model_name="gpt-4o",
    environment_name="leg_counting",
    start_time=datetime.now(),
    config={"temperature": 0.0, "max_tokens": 1024},
    include_detailed_results=True,
)

# Print summary
print_summary(eval_result)

# Save to JSON
eval_result.save("results/eval_20240115.json")

# Save without per-trajectory details (smaller file)
eval_result.save("results/eval_summary.json", include_results=False)

Result Structure

{
    "metadata": {
        "timestamp": "2024-01-15T10:30:00",
        "model": "gpt-4o",
        "environment": "leg_counting",
        "duration_seconds": 123.4,
        "num_trajectories": 100,
        "config": {...},
    },
    "metrics": {
        # Binary metrics (BinaryStatistics)
        "accuracy": {"n": 100, "mean": 0.85, "count": 85, "std_error": 0.036, ...},
        "format_compliance": {"n": 100, "mean": 0.95, "count": 95, ...},
        # Continuous metrics (ContinuousStatistics)
        "trajectory_reward": {"n": 100, "mean": 0.85, "std_dev": 0.12, "min": 0.0, ...},
        "action_reward": {"n": 100, "mean": 0.82, "std_dev": 0.15, ...},
    },
    "summary": {
        "success_rate": 0.85,
        "mean_reward": 0.85,
        "num_trajectories": 100,
    },
    "results": [...]  # Per-trajectory details if included
}

Prompt Templates and Model Profiles

The runner supports prompt templates (wrapping questions) and model profiles (model-specific adjustments):

from llenvs.inference import TEMPLATE_REGISTRY, PROFILE_REGISTRY

runner = TrajectoryRunner(
    environment=env,
    backend=backend,
    sampling_params=SamplingParams(temperature=0.0),
    system_prompt="You are an expert mathematician.",
    prompt_template=TEMPLATE_REGISTRY["math"],          # Wraps questions
    model_profile=PROFILE_REGISTRY["deepseek_r1"],      # Model-specific
)

The message build order is: system prompt → observation → prompt template → model profile → prompt pipeline. See the Prompts guide for full details.

For multi-turn environments with structured observations, you can also control how much conversation history the model sees. See the History Control guide.

Turn Info

Multi-turn environments have a maximum step count (EnvironmentSpec.max_steps) but the model doesn't see it by default. TurnInfoConfig provides an opt-in mechanism to inject turn counters and limits into the structured messages the runner builds.

from llenvs.evaluation import TrajectoryRunner, TurnInfoConfig

# Enable with defaults
runner = TrajectoryRunner(
    environment=env,
    backend=backend,
    turn_info=True,  # shorthand for TurnInfoConfig()
)

With turn_info=True (default config), the runner: - Appends "\n\nYou have a maximum of {max_steps} turns to complete this task." to the task description - Prepends "[Turn {turn}/{max_steps}]\n" to each state observation

Turn numbers are 1-indexed. When the environment doesn't declare max_steps, the _no_max variants are used instead (empty task suffix, "[Turn {turn}]\n" state prefix).

Custom formatting:

runner = TrajectoryRunner(
    environment=env,
    backend=backend,
    turn_info=TurnInfoConfig(
        task_suffix="\n\n({max_steps} steps available)",
        state_prefix="Step {turn}: ",
        task_suffix_no_max="",
        state_prefix_no_max="Step {turn}: ",
    ),
)

Available placeholders: {max_steps}, {turn} (1-indexed), {turns_remaining}.

Turn info only applies in structured mode (when obs.task/obs.state are set). Legacy mode is unaffected. History entries are not modified — only the task description and the current state observation receive injections.

The run_evaluation() convenience function also accepts turn_info:

result = run_evaluation(
    environment=env,
    backend=backend,
    turn_info=True,
)

Convenience Function

from llenvs.evaluation import run_evaluation
from llenvs.inference import TEMPLATE_REGISTRY

result = run_evaluation(
    environment=env,
    backend=backend,
    num_tasks=50,
    sampling_params=SamplingParams(temperature=0.0),
    system_prompt="Think step by step. Use <answer>...</answer> tags.",
    prompt_template=TEMPLATE_REGISTRY["math"],  # Optional
)

print(f"Accuracy: {result.success_rate:.2%}")

Error Handling

The runner handles errors gracefully:

batch_result = runner.run_batch(task_indices)

# Check for errors
for result in batch_result.trajectory_results:
    if "error" in result.metadata:
        print(f"Task {result.metadata['task_index']} failed: {result.metadata['error']}")

Built-in Reward Functions

StepPenalty

StepPenalty emits a small negative reward on every step, incentivizing agents to solve tasks in fewer turns:

from llenvs.core import StepPenalty

# Add as extra_rewards on any environment
env = SomeEnvironment(
    ...,
    extra_rewards=(StepPenalty(penalty=0.1),),
)

The penalty parameter is a positive float (default 0.1) that gets negated in the signal. Combined with an OUTCOME reward for correctness, this creates pressure to be efficient without sacrificing accuracy.

Configuration options:

Parameter	Default	Description
penalty	0.1	Per-step penalty (positive value, applied as negative reward)
_name	"step_penalty"	Signal name
_reward_type	RewardType.STEP	Signal type
_weight	1.0	Weight for aggregation

Custom Reward Analysis

from llenvs.core import RewardType

for result in batch_result.trajectory_results:
    for transition in result.trajectory.transitions:
        # Get signals by type
        outcome_signals = transition.rewards.by_type(RewardType.OUTCOME)
        format_signals = transition.rewards.by_type(RewardType.FORMAT)

        # Get specific signal
        correctness = transition.rewards.by_name("correctness")
        if correctness:
            print(f"Correctness: {correctness.reward}")

        # Get only signals with numeric rewards (skips feedback-only)
        numeric = transition.rewards.numeric_signals()

        # Collect textual feedback from all signals
        feedback = transition.rewards.feedback_texts()
        if feedback:
            print(f"Feedback: {feedback}")