# Setup for Eval Protocol Development
Source: https://evalprotocol.io/authoring-with-ai-agents
Configuring MCP servers for AI coding agents
Modern AI coding agents like Claude Code, Cursor, and GitHub Copilot can dramatically accelerate evaluation development, but they need the right context to be effective. This guide shows you how to supercharge your AI coding assistant with Model Context Protocol (MCP) servers that provide real-time access to Eval Protocol documentation and examples.
### Recommended MCP Servers
* **Documentation Server**: `https://evalprotocol.io/mcp` - Complete EP documentation, tutorials, and API references
* **Deep Wiki Server**: `https://mcp.deepwiki.com/mcp` - GitHub repository analysis and code search across EP projects
## Claude Code Integration
```bash theme={null}
claude mcp add --transport http eval-protocol-docs https://evalprotocol.io/mcp
claude mcp add --transport http eval-protocol-deep-wiki https://mcp.deepwiki.com/mcp
```
Create an `mcp.json` file in your project root:
```json theme={null}
{
"mcpServers": {
"eval-protocol-docs": {
"url": "https://evalprotocol.io/mcp"
},
"eval-protocol-deep-wiki": {
"url": "https://mcp.deepwiki.com/mcp"
}
}
}
```
**Enable Web Access**: Since the prompt references GitHub URLs, enable web access in Claude Code's settings by adding the WebFetch tool permission in `.claude/settings.json`.
## Cursor Integration
Install both MCP servers with one click each:
Access complete EP documentation, tutorials, and API references
[](https://cursor.com/en/install-mcp?name=eval-protocol-docs\&config=eyJ1cmwiOiJodHRwczovL2V2YWxwcm90b2NvbC5pby9tY3AifQ%3D%3D)
GitHub repository analysis and code search across EP projects
[](https://cursor.com/en/install-mcp?name=eval-protocol-deep-wiki\&config=eyJ1cmwiOiJodHRwczovL21jcC5kZWVwd2lraS5jb20vbWNwIn0%3D)
After clicking the install button above, you'll need to press the "Install" button in Cursor to complete the setup:
![Cursor MCP Server Install Dialog]()
Create an `mcp.json` file in your workspace root:
```json theme={null}
{
"mcpServers": {
"eval-protocol-docs": {
"url": "https://evalprotocol.io/mcp"
},
"eval-protocol-deep-wiki": {
"url": "https://mcp.deepwiki.com/mcp"
}
}
}
```
## Example Prompt to Develop With
With your MCP environment configured, the next step is to give your AI coding agent its mission. We start with a general prompt that defines its role and how to use Eval Protocol. Below that, you append the specific instructions for your project. This "meta-prompting" approach is crucial for guiding the agent effectively.
Here is a prompt template to use:
```markdown theme={null}
You are an applied AI engineer whose job is to write tests called "evals" in the form of code. An "eval" helps determines whether an AI application is working as expected by programmatically assessing the output of the model. To do this, you will use a library called "eval-protocol" (aka EP) that helps you easily author, run, and review evals. Evals accept whats called an EvaluationRow and outputs an EvaluationRow (or multiple EvaluationRows depending on the mode of the @evaluation_test decorator). In the eval, a score from 0 to 1 is generated based on the output of the model. Use the provided tools to help you understand more about eval-protocol so that you can generate evals for the given task. The tools can help provide examples of evals, guide you with tutorials on how to use eval-protocol, retrieve reference documentation for the eval-protocol API, and ask questions about the source code of eval-protocol.
GitHub source repos to query with the eval-protocol-deep-wiki tool:
- for the docs, see eval-protocol/eval-protocol
- for the Python SDK, see eval-protocol/python-sdk
Please follow the below instructions to write our evals:
{insert your prompt here}
```
## Next Steps
With MCP-enhanced AI coding agents, you can:
1. **Build faster**: Leverage real-time EP knowledge for accurate code generation
2. **Reduce errors**: Get context-aware suggestions that follow EP best practices
3. **Learn efficiently**: Ask specific questions and get authoritative answers
4. **Scale confidently**: Use proven patterns from the EP codebase and documentation
The combination of Eval Protocol's robust framework with MCP-enhanced AI agents creates a powerful development experience. Your coding assistant becomes an expert in EP, dramatically accelerating evaluation development while maintaining code quality and best practices.
# Common Errors
Source: https://evalprotocol.io/common-errors
Quick fixes for the most frequent evaluation issues
## Rate Limit Errors (429)
**Error**: `Error code: 429 - too_many_requests`
**Cause**: Hitting API rate limits during rollout generation or LLM judging.
**Solution**: Reduce concurrency in your `@evaluation_test`:
```python theme={null}
@evaluation_test(
max_concurrent_rollouts=8, # If error during model responses
max_concurrent_evaluations=2, # If error during judging ("impartial judge" prompts)
# ... other parameters
)
```
## Database Errors
**Error**: `sqlite3.OperationalError: disk I/O error` or `peewee.OperationalError`
**Solution**: Delete the corrupted database file:
```bash theme={null}
rm ~/.eval_protocol/logs.db
```
The database will be recreated automatically on next run.
## Model Not Found
**Error**: `ValueError: Model 'your-model' not found`
**Solution**: Use correct LiteLLM format:
```python theme={null}
completion_params=[
{"model": "openai/gpt-4o"}, # OpenAI
{"model": "anthropic/claude-3-sonnet"}, # Anthropic
{"model": "fireworks_ai/accounts/fireworks/models/llama-v3p1-8b-instruct"}, # Fireworks
]
```
## Import Errors
**Error**: `ImportError: Langfuse not installed`
**Solution**: Install platform dependencies:
```bash theme={null}
pip install 'eval-protocol[langfuse]' # For Langfuse
pip install 'eval-protocol[braintrust]' # For Braintrust
pip install 'eval-protocol[langsmith]' # For LangSmith
```
## Authentication Errors
**Error**: `AuthenticationError: Invalid API key`
**Solution**: Set environment variables:
```bash theme={null}
export OPENAI_API_KEY="your_key"
export FIREWORKS_API_KEY="your_key"
export LANGFUSE_PUBLIC_KEY="your_key"
export LANGFUSE_SECRET_KEY="your_secret"
```
## No Data Found
**Error**: `❌ No evaluation rows provided`
**Solution**: Double check your adapter is returning the rows. You may want to debug by adding print statements around the number of rows being returned, as well as the content of the rows.
```python theme={null}
# Test with minimal filters
rows = adapter.get_evaluation_rows(limit=10, hours_back=168)
print(f"Found {len(rows)} rows")
for row in rows:
print(f"Messages: {row.messages}")
```
## Fireworks Persistence Warning
**Warning**: `❌ Experiment past-story-29: No Fireworks account ID found`
**Cause**: Missing `FIREWORKS_ACCOUNT_ID` environment variable.
**Solution**:
* **To persist results to Fireworks** (easier sharing): `export FIREWORKS_ACCOUNT_ID="your_account_id"`
* **To use locally only**: Ignore this warning - everything still works, results just stay local
## UI Not Updating
**Problem**: Local UI at [http://localhost:8000](http://localhost:8000) not showing new results
**Solution**: Restart the EP logs server:
```bash theme={null}
# Stop current ep logs (Ctrl+C)
# Then restart:
ep logs
```
## Getting Help
For other issues, join the [Discord](https://discord.com/channels/1137072072808472616/1400975572405850155) for support with a Fireworks engineer.
# GEPA Prompt Optimizer
Source: https://evalprotocol.io/integrations/gepa-trainer
Automatically optimize prompts using your existing evaluations
Eval Protocol integrates with [GEPA](https://arxiv.org/abs/2507.19457) (via [DSPy](https://github.com/stanfordnlp/dspy)) to automatically optimize your prompts using the evaluations you've already written. GEPA analyzes which examples pass or fail, proposes structured edits to the prompt, and keeps changes that improve your metric.
## How It Works
GEPA treats your `@evaluation_test` as the optimization objective. It:
1. Extracts the system prompt from your dataset
2. Splits your data into training and validation sets
3. Runs your evaluation function on candidate prompts
4. Uses a reflection LLM to propose improvements based on failure patterns
5. Returns the best-performing prompt
The key insight is that your evaluation's `reason` field (in [`EvaluateResult`](/specification#evaluateresult)) tells GEPA *why* examples failed, enabling targeted improvements.
## Prerequisites
Install eval-protocol with the `dspy` extra:
```bash theme={null}
pip install eval-protocol[dspy]
```
Set your API key:
```bash theme={null}
export FIREWORKS_API_KEY="your-fireworks-key"
```
## Basic Usage
Write a normal `@evaluation_test`, then wrap it with `GEPATrainer`.
### Step 1: Define Your Evaluation Test
```python my_eval.py theme={null}
from eval_protocol.models import EvaluationRow, EvaluateResult
from eval_protocol.pytest.evaluation_test import evaluation_test
from eval_protocol.pytest.default_single_turn_rollout_process import SingleTurnRolloutProcessor
@evaluation_test(
input_dataset=["datasets/my_dataset.jsonl"],
dataset_adapter=my_adapter,
completion_params=[{
"model": "fireworks_ai/accounts/fireworks/models/llama-v3p1-70b-instruct",
"max_tokens": 4096,
}],
rollout_processor=SingleTurnRolloutProcessor(),
mode="pointwise",
)
def test_my_task(row: EvaluationRow) -> EvaluationRow:
predicted = row.get_last_message_content()
expected = row.ground_truth
is_correct = predicted.strip() == expected.strip()
# Feedback tells GEPA why this example failed
if is_correct:
feedback = "Correct answer."
else:
feedback = f"Incorrect. Expected '{expected}', got '{predicted}'."
row.evaluation_result = EvaluateResult(
score=1.0 if is_correct else 0.0,
reason=feedback,
is_score_valid=True,
)
return row
```
### Step 2: Add GEPA Training
```python my_eval.py theme={null}
from eval_protocol.training import GEPATrainer, build_reflection_lm
if __name__ == "__main__":
trainer = GEPATrainer(
test_my_task,
train_ratio=0.7,
val_ratio=0.3,
)
reflection_lm = build_reflection_lm(
"fireworks_ai/accounts/fireworks/models/llama-v3p1-70b-instruct"
)
optimized_program = trainer.train(
reflection_lm=reflection_lm,
max_metric_calls=1500,
num_threads=4,
)
print(trainer.evaluate(optimized_program))
print(trainer.get_optimized_system_prompt(optimized_program))
```
### Step 3: Run
```bash theme={null}
python my_eval.py
```
**Single System Prompt Requirement**
GEPA extracts and optimizes **only the first system prompt** found in your dataset. All rows must share the same system prompt for GEPA to work correctly.
If your dataset contains different system prompts per row (e.g., different personas or task variations), GEPA will only optimize the first one and apply it to all examples, which may produce unexpected results. Consider splitting such datasets into separate optimization runs.
## Case Study: Text-to-SQL
The [text-to-sql-quickstart](https://github.com/eval-protocol/text-to-sql-quickstart) repository demonstrates GEPA on a text-to-SQL benchmark.
### Repository Structure
```text theme={null}
text-to-sql-quickstart/
├── data/
│ └── synthetic_openflights.db # DuckDB database with airlines/airports/routes
├── datasets/
│ ├── final_rft_sql_train_data.jsonl # Training examples
│ └── final_rft_sql_test_data.jsonl # Held-out test set
├── mcp_server/ # HTTP server that executes SQL queries
├── evaluator/
│ └── sql_gepa_training.py # GEPA training script
└── scripts/
└── eval_baseline.py # Evaluate prompts on test set
```
### The Database
The benchmark uses a synthetic OpenFlights database (`synthetic_openflights.db`) containing tables for airlines, airports, countries, planes, and routes. This database is shared between:
1. Ground truth generation (SQL queries executed to create expected results)
2. The MCP server (executes model-generated SQL during evaluation)
### The MCP Server
The MCP server is a simple HTTP service that accepts SQL queries and returns results from the DuckDB database:
```python theme={null}
# mcp_server/run_mcp_server.py
DB = os.environ.get("DB_PATH", "data/synthetic_openflights.db")
```
When you run the training script, the MCP server starts automatically. It receives the model's generated SQL, executes it against the database, and returns the results for comparison with ground truth.
### The Evaluation Function
The evaluation compares the model's SQL output against ground truth by executing both and checking if they return the same data:
```python theme={null}
def test_sql_generation(row: EvaluationRow) -> EvaluationRow:
generated_sql = row.get_last_message_content()
# Ground truth is stored as list[dict] from the original query
expected_results = row.ground_truth
# Execute the model's SQL via MCP server
actual_results = execute_sql_via_mcp(generated_sql)
# Compare results semantically (values match, ignoring column order)
is_correct = compare_results(expected_results, actual_results)
# Build detailed feedback for GEPA
if is_correct:
feedback = "Query returned correct results."
else:
feedback = analyze_mismatch(expected_results, actual_results)
row.evaluation_result = EvaluateResult(
score=1.0 if is_correct else 0.0,
reason=feedback,
is_score_valid=True,
)
return row
```
The feedback function provides specific details about failures:
```python theme={null}
def analyze_mismatch(expected, actual):
issues = []
if len(expected) != len(actual):
issues.append(f"Row count: expected {len(expected)}, got {len(actual)}")
expected_cols = set(expected[0].keys()) if expected else set()
actual_cols = set(actual[0].keys()) if actual else set()
missing = expected_cols - actual_cols
if missing:
issues.append(f"Missing columns: {missing}")
return " | ".join(issues)
```
### Running the Example
1. Clone the repository:
```bash theme={null}
git clone https://github.com/eval-protocol/text-to-sql-quickstart
cd text-to-sql-quickstart
pip install -r requirements.txt
pip install eval-protocol[dspy] # Required for GEPA training
```
2. Set your API key:
```bash theme={null}
export FIREWORKS_API_KEY="your-key"
```
3. The repository includes pre-generated data. To run GEPA training:
```bash theme={null}
python evaluator/sql_gepa_training.py
```
This starts the MCP server automatically, runs GEPA optimization, and prints the optimized prompt.
4. To compare original vs optimized prompts on the test set:
```bash theme={null}
python scripts/eval_baseline.py --prompt both
```
### Data Generation (Optional)
If you want to regenerate the synthetic data from scratch:
```bash theme={null}
make all-data
```
This runs:
1. Download real OpenFlights data
2. Generate synthetic rows using an LLM
3. Generate SQL queries
4. Execute queries to get ground truth results
5. Generate natural language questions from SQL
The `scripts/08_regenerate_balanced_data.py` script generates data with consistent column naming for better train/test distribution.
### Results
On this benchmark, GEPA discovered that failures clustered around column alias mismatches (`avg_altitude` vs `average_altitude`) and missing columns. It rewrote the prompt to include explicit naming conventions and a validation checklist.
| Metric | Original | Optimized |
| ---------- | -------- | --------- |
| Test Set | 38.3% | 48.3% |
| Validation | 30.9% | 36.4% |
## Configuration
### GEPATrainer Parameters
The @evaluation\_test decorated function to optimize
Proportion of data for training
Proportion of data for validation
Random seed for dataset splits
Name of the input field in DSPy signature
Name of the output field in DSPy signature
DSPy module type: PREDICT, CHAIN\_OF\_THOUGHT, or PROGRAM\_OF\_THOUGHT
### train() Parameters
DSPy LM for proposing prompt improvements
Total budget of LLM calls for optimization
Budget preset: "light", "medium", or "heavy". Alternative to max\_metric\_calls.
Number of examples shown to reflection LLM per iteration
Parallel threads for running evaluations
## Tips
**Provide specific feedback.** GEPA learns from your `evaluation_result.reason`. Instead of "Incorrect", say "Missing column 'airport\_count'".
**Choose appropriate budget.** For small datasets (under 50 examples), use `auto="light"`. For larger datasets, increase to `"medium"` or `"heavy"`.
**Use the right module type.** `PREDICT` for simple tasks, `CHAIN_OF_THOUGHT` for reasoning tasks, `PROGRAM_OF_THOUGHT` for code generation.
**Keep a held-out test set.** GEPA should never see your final test data during optimization.
## Troubleshooting
**GEPA finds no improvement:** Add more detailed feedback, increase `reflection_minibatch_size`, or increase budget.
**API timeouts:** Reduce `num_threads` or use a faster model.
**Memory issues:** Reduce `num_threads` or process smaller batches.
**Dataset has multiple system prompts:** GEPA only optimizes the first system prompt found. If your dataset uses different prompts for different tasks, split it into separate datasets with consistent prompts and run GEPA on each.
## Resources
[GEPA Paper](https://arxiv.org/abs/2507.19457) ·
[DSPy Documentation](https://dspy.ai/) ·
[Text-to-SQL Quickstart](https://github.com/eval-protocol/text-to-sql-quickstart)
# Klavis MCP Environments
Source: https://evalprotocol.io/integrations/klavis-mcp
How to use Klavis with Eval Protocol
[Klavis AI](https://klavis.ai/) provides hosted Model Context Protocol (MCP) servers and managed sandbox environments that integrate with Eval Protocol. This guide covers two ways to use Klavis:
1. **Klavis MCP Sandbox** - Fully managed isolated environments for model training and evaluation at scale
2. **Klavis MCP Server** - Direct MCP server connections using your own accounts
## Which Option Should You Choose?
### Use Klavis MCP Sandbox
Use Klavis MCP Sandbox if you **only have input data and ground truth** for your RL work. Klavis Sandbox handles all the tooling infrastructure for you:
* **Hosted MCP Servers** - hundreds of pre-built servers ready to use
* **Authentication** - OAuth and session management handled automatically
* **Isolated Concurrency Environments** - run 64+ models in parallel without interference
* **Tooling State Management** - automatic initialization, reset, and cleanup
* **Scaling** - dedicated QPS per instance with automatic account pooling
This is the **turnkey solution** for model training and reinforcement learning with tools.
### Use Klavis MCP Server
Use Klavis MCP Server if you **already have your own tooling infrastructure** (authentication, isolated environments, state management, scaling) but only need Klavis hosted MCP servers to perform tool calls for your RL or model training work.
This option allows you to connect directly to 100+ external applications through Klavis MCP while maintaining full control over your evaluation and training pipeline.
***
## Use with Klavis MCP Sandbox
Klavis MCP Sandbox provides fully managed, isolated sandbox environments designed for training and evaluating models at scale. Each sandbox has dedicated accounts, automatic state initialization, and cleanup - allowing you to focus on model interaction without managing sandbox environments.
### Key Features
* **Isolated Environments**: Each sandbox gets dedicated, authenticated sessions with automatic token management
* **Account Pooling**: Dynamic pool of test accounts supporting 64+ concurrent models
* **State Management**: Built-in `initialize`, `dump`, and `reset` APIs for environment lifecycle
* **Supported Services**: Gmail, Jira, Salesforce, Slack, Linear, Google Calendar, and 100+ more
### Setup
Set up your API keys:
```bash theme={null}
export KLAVIS_API_KEY=your_klavis_api_key
export FIREWORKS_API_KEY=your_fireworks_api_key
```
### Step 1: Define Your Input Data and Ground Truth
Create a JSONL dataset file with your test cases. Each row should include:
* `initialize_data`: Initial state to seed the sandbox
* `messages`: The task instruction for your model
* `ground_truth`: Expected final state after the model completes the task
Example dataset structure:
```json theme={null}
{
"initialize_data": {
"messages": [
{
"subject": "Project Update",
"to": "zihao@klavisai.com",
"body": "The project is progressing well.",
"from": "sarah@klavisai.com",
"labels": ["INBOX"]
}
],
"drafts": []
},
"messages": "Please delete the email with subject 'Spam Newsletter' from my inbox.",
"ground_truth": {
"messages": [
{
"subject": "Project Update",
"to": "zihao@klavisai.com",
"body": "The project is progressing well.",
"from": "sarah@klavisai.com",
"labels": ["INBOX"]
}
],
"drafts": []
}
}
```
See [full example dataset](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/datasets/klavis_gmail_sandbox_test.jsonl) for more test cases.
### Step 2: Implement Your RolloutProcessor
Use the `KlavisSandboxRolloutProcessor` to handle sandbox lifecycle management. The processor will:
1. Create an isolated sandbox instance
2. Initialize the sandbox with your input data
3. Run your model with MCP tools from the sandbox
4. Dump the final state after model interaction
5. Clean up and return sandbox to pool
```python theme={null}
from eval_protocol.pytest import KlavisSandboxRolloutProcessor
rollout_processor = KlavisSandboxRolloutProcessor(
server_name="gmail", # or "jira", "salesforce", "slack", etc.
)
```
For custom initialization logic, you can extend `KlavisSandboxRolloutProcessor`:
```python theme={null}
from typing import Dict, Any
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import KlavisSandboxRolloutProcessor
def custom_initialize_data(row: EvaluationRow) -> Dict[str, Any]:
# Custom logic to transform your row data
# into sandbox initialization format
return {
"messages": row.input_metadata.session_data.get("emails", []),
"drafts": []
}
rollout_processor = KlavisSandboxRolloutProcessor(
server_name="gmail",
initialize_data_factory=custom_initialize_data
)
```
See the [full implementation](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/default_klavis_sandbox_rollout_processor.py) for advanced customization.
### Step 3: Evaluate by Comparing State with Ground Truth
Create your evaluation test that compares the final sandbox state with your ground truth:
```python theme={null}
from eval_protocol.pytest import evaluation_test, KlavisSandboxRolloutProcessor
from eval_protocol.models import EvaluationRow, EvaluateResult
from openai import AsyncOpenAI
@evaluation_test(
input_dataset=["datasets/klavis_gmail_sandbox_test.jsonl"],
completion_params=[{"model": "fireworks_ai/accounts/fireworks/models/llama-v3p3-70b-instruct"}],
rollout_processor=KlavisSandboxRolloutProcessor(server_name="gmail"),
mode="pointwise",
)
async def test_gmail_sandbox(row: EvaluationRow) -> EvaluationRow:
# Extract final sandbox state and ground truth
sandbox_data = row.execution_metadata.extra.get("sandbox_data", {})
ground_truth = row.ground_truth
# Use LLM judge to evaluate
async with AsyncOpenAI(api_key=os.environ["FIREWORKS_API_KEY"]) as client:
response = await client.chat.completions.create(
model="accounts/fireworks/models/kimi-k2-thinking",
messages=[{"role": "user", "content": f"Compare final state {sandbox_data} with expected {ground_truth}. Return score 0-1."}],
response_format={"type": "json_schema", ...},
)
score = json.loads(response.choices[0].message.content).get("score", 0.0)
row.evaluation_result = EvaluateResult(score=score)
return row
```
The final sandbox state is available in `row.execution_metadata.extra["sandbox_data"]`. Use an LLM judge to semantically compare it with your ground truth.
See the [complete test implementation](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/test_pytest_klavis_sandbox.py) for the full example.
## Use with Klavis MCP Server
### Setting Up Klavis MCP Server
Login to your Klavis AI account, then find the applications you want to connect with Eval Protocol and enable MCP for those applications. Follow the auth flow to authorize Klavis MCP to access those applications on your behalf. You can follow the Klavis quickstart guide [here](https://www.klavis.ai/docs/quickstart) to set up your MCP.
In the Klavis dashboard, click **Add to Other Clients**, and generate the access token. Save the access token in `.env` file as `KLAVIS_API_KEY`.
The Klavis MCP is defined as follows in Eval Protocol configuration:
```json theme={null}
{
"mcpServers": {
"klavis-strata": {
"url": "https://strata.klavis.ai/mcp/",
"authorization": "Bearer ${KLAVIS_API_KEY}"
}
}
}
```
### Using Klavis MCP Server in Eval Protocol
We've set up an example in Eval Protocol to use Klavis MCP Server. You can also use it to connect to more applications and add more use cases.
Here is the example [test file](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/test_pytest_klavis_mcp.py). In this example, we connect to Gmail, Notion and Outlook Calendar using Klavis MCP, and have a few example test cases. To run this example workflow, you need to set up the test cases in those applications.
#### Gmail
No particular setup. You should have at least 5 emails in your Gmail inbox.
#### Notion
Copy this [Notion page template](https://painted-tennis-ebc.notion.site/MCPMark-Source-Hub-23181626b6d7805fb3a7d59c63033819) (credit to [MCPMark](https://mcpmark.ai/)) to your Notion workspace. And when you authorize Klavis MCP to access Notion, make sure to give access to this page.
#### Outlook Calendar
You should set up the following calendar events in your Outlook calendar. It's recommended to create a new outlook account with a clean calendar for testing.
1. Create 3 events today. It's better one starting at 12 am today, and one ending at 12 am tomorrow.
2. Create an event that covers the whole workding hours except the first and last hour of your next working day. Outlook calendar default working hour is Monday to Friday, 8 am to 5 pm. In this case, you should create an event from 9 am to 4 pm on your next working day.
3. Create total 8 events on this week's working days. It should include the above events if they are on working days.
4. Follow step 1, create 2 events on next week's Thursday.
5. Follow step 3, create total 5 events on next week's working days.
6. Follow step 1, create 4 events on Oct 15 2025.
7. Follow step 3, create total 9 events from Oct 13 to Oct 17, 2025.
## Resources
Learn about tooling infrastructure for LLM training, RL and evaluation.
Browse all high quality MCP servers written and evaluated by Klavis AI.
Create sandboxes, seed data, run an agent, then dump and clean up.
Manage isolated sandbox environments for training/eval: pooling, init, export, teardown.
# OpenAI RFT Trainer
Source: https://evalprotocol.io/integrations/openai-rft-trainer
Reuse Eval Protocol evaluation tests as Python graders for OpenAI Reinforcement Fine-Tuning (RFT)
The OpenAI RFT adapter lets you **reuse Eval Protocol evaluation tests as Python graders** for OpenAI Reinforcement Fine-Tuning (RFT). Because your grading logic lives in an Eval Protocol `@evaluation_test`, you can reuse the exact same code as an OpenAI Python grader—making it easy to start with OpenAI RFT and later move to other Eval-Protocol supported training workflows (or vice versa) without rewriting your evals.
For a minimal working example, clone the [`openai-rft-quickstart`](https://github.com/eval-protocol/openai-rft-quickstart) repository, which contains the `example_rapidfuzz.py` and `test_openai_grader.py` files used in the examples below.
## High Level Overview
The core helper function lives in:
```python theme={null}
from eval_protocol.integrations.openai_rft import build_python_grader_from_evaluation_test
```
Under the hood, `build_python_grader_from_evaluation_test`:
* **Takes** your Eval Protocol `@evaluation_test` function that operates on an `EvaluationRow`.
* **Wraps** it into a self-contained `{"type": "python", "source": ...}` grader module with a `grade(sample, item)` entrypoint.
* **Builds** a minimal `EvaluationRow` from the OpenAI RFT inputs by:
* Mapping `item["reference_answer"]` to `row.ground_truth`
* Mapping `item["messages"]` (if present) to `row.messages`
* Mapping `sample["output_text"]` to the last assistant message
* **Removes** any runtime dependency on `eval-protocol` inside the grader by using simple duck-typed stand-ins for `EvaluationRow`, `EvaluateResult`, and `Message`.
* **Normalizes** whatever your evaluation returns (e.g., `EvaluateResult`, `EvaluationRow` with `.evaluation_result`, or a bare number) into a single float score.
You can inspect the full implementation in [`eval_protocol/integrations/openai_rft.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/integrations/openai_rft.py).
## Grader Constraints
When you convert an `@evaluation_test` into an OpenAI Python grader, it must satisfy OpenAI’s runtime limits, i.e. no network access, fixed set of packages (e.g., `numpy`, `pandas`, `rapidfuzz`, etc.). For more details, see [OpenAI graders documentation](https://platform.openai.com/docs/guides/graders#technical-constraints).
## Basic Usage
### 1. Write an Eval Protocol `@evaluation_test`
In `example_rapidfuzz.py` (from the `openai-rft-quickstart` repo) we define a simple evaluation test that uses `rapidfuzz` to score how close a model’s answer is to the ground truth:
```python theme={null}
@evaluation_test(
input_rows=[DEMO_ROWS],
rollout_processor=NoOpRolloutProcessor(),
aggregation_method="mean",
mode="pointwise",
)
def rapidfuzz_eval(row: EvaluationRow, **kwargs: Any) -> EvaluationRow:
"""
Example @evaluation_test that scores a row using rapidfuzz.WRatio and
attaches an EvaluateResult.
"""
from rapidfuzz import fuzz, utils
# For EP evals, we compare the EvaluationRow's ground_truth to the last assistant message.
reference = row.ground_truth
assistant_msgs = [m for m in row.messages if m.role == "assistant"]
last_assistant_content = assistant_msgs[-1].content if assistant_msgs else ""
prediction = last_assistant_content if isinstance(last_assistant_content, str) else ""
score = float(
fuzz.WRatio(
str(prediction),
str(reference),
processor=utils.default_process,
)
/ 100.0
)
row.evaluation_result = EvaluateResult(score=score)
return row
```
### 2. Convert to a Python grader and call `/graders/*`
In `test_openai_grader.py` (also in the `openai-rft-quickstart` repo) we show how to:
* Build a Python grader spec from `rapidfuzz_eval`
* Validate it via `/fine_tuning/alpha/graders/validate`
* Run it once via `/fine_tuning/alpha/graders/run`
```python theme={null}
import os
import requests
from eval_protocol.integrations.openai_rft import build_python_grader_from_evaluation_test
from examples.openai_rft.example_rapidfuzz import rapidfuzz_eval
api_key = os.environ["OPENAI_API_KEY"]
headers = {"Authorization": f"Bearer {api_key}"}
grader = build_python_grader_from_evaluation_test(rapidfuzz_eval) # {"type": "python", "source": "..."}
# validate the grader
resp = requests.post(
"https://api.openai.com/v1/fine_tuning/alpha/graders/validate",
json={"grader": grader},
headers=headers,
)
print("validate response:", resp.text)
# run the grader once with a dummy item/sample
payload = {
"grader": grader,
"item": {"reference_answer": "fuzzy wuzzy had no hair"},
"model_sample": "fuzzy wuzzy was a bear",
}
resp = requests.post(
"https://api.openai.com/v1/fine_tuning/alpha/graders/run",
json=payload,
headers=headers,
)
print("run response:", resp.text)
```
## End-to-End Example
To see an end-to-end example that takes an `@evaluation_test` (`rapidfuzz_eval`), converts it into a `{"type": "python", "source": ...}` grader spec with `build_python_grader_from_evaluation_test`, and validates/runs it against the OpenAI `/graders/*` HTTP APIs, clone the quickstart repo and run:
```bash theme={null}
git clone git@github.com:eval-protocol/openai-rft-quickstart.git
cd openai-rft-quickstart
pytest example_rapidfuzz.py -vs # Shows that this works as an EP evaluation_test
python test_openai_grader.py # Validates and runs the Python grader via OpenAI's /graders/* APIs
```
You can expect an output like:
```bash theme={null}
validate response: {
"grader": {
"type": "python",
"source": "def _ep_eval(row, **kwargs):\n \"\"\"\n Example @evaluation_test that scores a row using rapidfuzz.WRatio and\n attaches an EvaluateResult.\n \"\"\"\n reference = row.ground_truth\n assistant_msgs = [m for m in row.messages if m.role == 'assistant']\n last_assistant_content = assistant_msgs[-1].content if assistant_msgs else ''\n prediction = last_assistant_content if isinstance(last_assistant_content, str) else ''\n from rapidfuzz import fuzz, utils\n score = float(fuzz.WRatio(str(prediction), str(reference), processor=utils.default_process) / 100.0)\n row.evaluation_result = EvaluateResult(score=score)\n return row\n\n\nfrom typing import Any, Dict\nfrom types import SimpleNamespace\n\n\nclass EvaluationRow(SimpleNamespace):\n \"\"\"Minimal duck-typed stand-in for an evaluation row.\n\n Extend this with whatever attributes your eval logic uses.\n \"\"\"\n pass\n\n\nclass EvaluateResult(SimpleNamespace):\n \"\"\"Simple stand-in for Eval Protocol's EvaluateResult.\n\n This lets evaluation-style functions that construct EvaluateResult(score=...)\n run inside the Python grader sandbox without importing eval_protocol.\n \"\"\"\n\n def __init__(self, score: float, **kwargs: Any) -> None:\n super().__init__(score=score, **kwargs)\n\n\nclass Message(SimpleNamespace):\n \"\"\"Duck-typed stand-in for eval_protocol.models.Message (role/content).\"\"\"\n pass\n\n\ndef _build_row(sample: Dict[str, Any], item: Dict[str, Any]) -> EvaluationRow:\n # Start from any item-provided messages (EP-style), defaulting to [].\n raw_messages = item.get(\"messages\") or []\n normalized_messages = []\n for m in raw_messages:\n if isinstance(m, dict):\n normalized_messages.append(\n Message(\n role=m.get(\"role\"),\n content=m.get(\"content\"),\n )\n )\n else:\n # Already Message-like; rely on duck typing (must have role/content)\n normalized_messages.append(m)\n\n reference = item.get(\"reference_answer\")\n prediction = sample.get(\"output_text\")\n\n # EP-style: ensure the model prediction is present as the last assistant message\n if prediction is not None:\n normalized_messages = list(normalized_messages) # shallow copy\n normalized_messages.append(Message(role=\"assistant\", content=prediction))\n\n return EvaluationRow(\n ground_truth=reference,\n messages=normalized_messages,\n item=item,\n sample=sample,\n )\n\n\ndef grade(sample: Dict[str, Any], item: Dict[str, Any]) -> float:\n row = _build_row(sample, item)\n result = _ep_eval(row=row)\n\n # Try to normalize different result shapes into a float score\n try:\n from collections.abc import Mapping\n\n if isinstance(result, (int, float)):\n return float(result)\n\n # EvaluateResult-like object with .score\n if hasattr(result, \"score\"):\n return float(result.score)\n\n # EvaluationRow-like object with .evaluation_result.score\n eval_res = getattr(result, \"evaluation_result\", None)\n if eval_res is not None:\n if isinstance(eval_res, Mapping):\n if \"score\" in eval_res:\n return float(eval_res[\"score\"])\n elif hasattr(eval_res, \"score\"):\n return float(eval_res.score)\n\n # Dict-like with score\n if isinstance(result, Mapping) and \"score\" in result:\n return float(result[\"score\"])\n except Exception:\n pass\n\n return 0.0\n",
"name": "grader-R5FhpA6BFQlo"
}
}
run response: {
"reward": 0.7555555555555555,
"metadata": {
"name": "grader-5XXSBZ9B1OJj",
"type": "python",
"errors": {
"formula_parse_error": false,
"sample_parse_error": false,
"sample_parse_error_details": null,
"truncated_observation_error": false,
"unresponsive_reward_error": false,
"invalid_variable_error": false,
"invalid_variable_error_details": null,
"other_error": false,
"python_grader_server_error": false,
"python_grader_server_error_type": null,
"python_grader_runtime_error": false,
"python_grader_runtime_error_details": null,
"model_grader_server_error": false,
"model_grader_refusal_error": false,
"model_grader_refusal_error_details": null,
"model_grader_parse_error": false,
"model_grader_parse_error_details": null,
"model_grader_exceeded_max_tokens_error": false,
"model_grader_server_error_details": null,
"endpoint_grader_internal_error": false,
"endpoint_grader_internal_error_details": null,
"endpoint_grader_server_error": false,
"endpoint_grader_server_error_details": null,
"endpoint_grader_safety_check_error": false
},
"execution_time": 6.831332206726074,
"scores": {},
"token_usage": null,
"sampled_model_name": null
},
"sub_rewards": {},
"model_grader_token_usage_per_model": {}
}
```
This confirms that:
* Your Eval Protocol `@evaluation_test` (`rapidfuzz_eval`) runs as a normal eval via `pytest`.
* The same function can be converted into a `type: "python"` grader spec and validated / run through the OpenAI RFT graders API.
Now that you have your grader, see OpenAI’s docs on [preparing your dataset and creating a reinforcement fine-tuning job](https://platform.openai.com/docs/guides/reinforcement-fine-tuning#prepare-your-dataset).
# OpenEnv Environments
Source: https://evalprotocol.io/integrations/openenv-rollout-processor
Use any OpenEnv HTTP environment with Eval Protocol via a single rollout processor
## Overview
[OpenEnv](https://github.com/meta-pytorch/OpenEnv/tree/main) is an open-source framework from Meta’s PyTorch team for defining, deploying, and interacting with environments in RL and agentic workflows. It gives you **Gym-style APIs** (`reset()`, `step()`, `state()`) wrapped in HTTP clients (for example `BrowserGymEnv`, `EchoEnv`, `TextArenaEnv`), and lets you run those environments:
* As local Python processes.
* Inside Docker containers.
* As hosted Hugging Face Spaces.
Eval Protocol integrates with OpenEnv by talking only to the **environment client**. Once an environment is exposed as an OpenEnv client, Eval Protocol can drive episodes without any environment-specific code in your tests.
`OpenEnvRolloutProcessor` is the component that **runs the OpenEnv loop for you**:
* It calls `env.reset()` to start an episode for each `EvaluationRow`.
* For each step, it builds a user message from the observation, calls your model, parses the model’s response into an action, and calls `env.step(action)`.
* It appends a sentinel system message with per-step rewards so your `@evaluation_test` can compute a final score in a single place.
You can use the **same pattern** to write evals for any OpenEnv environment (BrowserGym, Echo, TextArena, Atari-style games, etc.) by changing only:
* Which OpenEnv client you pass (`BrowserGymEnv`, `EchoEnv`, `TextArenaEnv`, …).
* How you build prompts (`prompt_builder`).
* How you parse actions (`action_parser`).
## How to use OpenEnvRolloutProcessor
At a high level:
1. **Pick an OpenEnv client** for your environment (see the [OpenEnv environments](https://github.com/meta-pytorch/OpenEnv/tree/main/src/envs) for a full list):
* BrowserGym: `from envs.browsergym_env import BrowserGymEnv, BrowserGymAction`
* Echo: `from envs.echo_env import EchoEnv, EchoAction`
* TextArena: `from envs.textarena_env import TextArenaEnv, TextArenaAction`
2. **Write a `prompt_builder(observation, step, history)`** that turns the current observation into a user-facing prompt string (or chat messages).
3. **Write an `action_parser(response_text)`** that converts model output into the environment’s `Action` type.
4. **Instantiate `OpenEnvRolloutProcessor`** with the right constructor kwargs:
* `env_client_cls` or `env_factory` (how to construct the client).
* `prompt_builder` and `action_parser`.
* Environment wiring:
* `docker_image` and `env_vars` for Docker-based envs (BrowserGym, TextArena).
* `hub_repo_id` to launch from Hugging Face Hub (for example `"openenv/echo-env"`).
* `env_base_url` when connecting to an already running server or remote Space.
* Optional task routing:
* `tasks` and `task_var` if you want to rotate across multiple tasks (for example multiple MiniWoB levels).
5. **Use it in an `@evaluation_test`**:
* Set `rollout_processor=OpenEnvRolloutProcessor(...)`.
* In the test body, read the step rewards sentinel from `row.messages` and set `row.evaluation_result` based on whatever scoring you want.
Concrete examples of `prompt_builder` and `action_parser` can be found in the Eval Protocol Python SDK:
* BrowserGym: [`tests.pytest.test_openenv_browsergym_eval`](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/test_openenv_browsergym_eval.py)
* Echo: [`tests.pytest.test_openenv_echo_hub`](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/test_openenv_echo_hub.py)
* TextArena: [`tests.pytest.test_openenv_textarena_docker`](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/test_openenv_textarena_docker.py)
## BrowserGym example (MiniWoB via Docker)
```python openenv_browsergym_eval.py theme={null}
from typing import Any, Dict, List
import os
import re
import pytest
from eval_protocol.models import EvaluationRow, Message, EvaluateResult
from eval_protocol.pytest import evaluation_test
from eval_protocol.pytest.openenv_rollout_processor import OpenEnvRolloutProcessor
def browsergym_dataset_to_rows(data: List[Dict[str, Any]]) -> List[EvaluationRow]:
"""Adapt simple dict rows into EvaluationRow objects."""
rows: List[EvaluationRow] = []
for row in data:
prompt = str(row.get("prompt", "start"))
rows.append(EvaluationRow(messages=[Message(role="user", content=prompt)]))
return rows
ACTION_PATTERN = re.compile(r"[A-Za-z_]+\s*\(.*\)", re.DOTALL)
def prompt_builder(observation: Any, step: int, history: List[str]) -> str:
"""Turn a BrowserGym observation into a text prompt."""
goal = getattr(observation, "goal", "") or ""
url = getattr(observation, "url", "") or "(unknown)"
error_note = "Yes" if getattr(observation, "last_action_error", False) else "No"
text = (getattr(observation, "text", "") or "")[:2048]
return (
f"Step: {step}\n"
f"Goal: {goal}\n"
f"Current URL: {url}\n"
f"Previous steps:\n" + ("\n".join(history[-4:]) if history else "None") + "\n"
f"Last action error: {error_note}\n\n"
"Reply with a single BrowserGym action, e.g., click('13') or noop().\n\n"
f"Page excerpt:\n{text}\n\n"
"Reply with exactly one BrowserGym action string."
).strip()
def action_parser(response_text: str):
"""Parse model output into a BrowserGym action."""
try:
from envs.browsergym_env import BrowserGymAction # provided by OpenEnv
except Exception:
pytest.skip("OpenEnv (envs.browsergym_env) is not installed; skipping BrowserGym test.")
raise
if not response_text:
return BrowserGymAction(action_str="noop()")
for raw in response_text.splitlines():
line = raw.strip()
if not line:
continue
m = ACTION_PATTERN.search(line)
if m:
return BrowserGymAction(action_str=m.group(0))
m = ACTION_PATTERN.search(response_text)
if m:
return BrowserGymAction(action_str=m.group(0))
return BrowserGymAction(action_str="noop()")
try:
from envs.browsergym_env import BrowserGymEnv # provided by OpenEnv
_HAS_BROWSERGYM = True
except Exception:
_HAS_BROWSERGYM = False
BROWSERGYM_INLINE_DATA: List[Dict[str, Any]] = [
{"id": "click-test", "prompt": "start"},
]
@evaluation_test( # type: ignore[misc]
input_rows=[browsergym_dataset_to_rows(BROWSERGYM_INLINE_DATA)],
completion_params=[
{
"temperature": 0.0,
"max_tokens": 512,
"model": "fireworks_ai/accounts/fireworks/models/kimi-k2-instruct",
}
],
num_runs=1,
max_concurrent_rollouts=1,
mode="pointwise",
rollout_processor=(
OpenEnvRolloutProcessor(
env_client_cls=BrowserGymEnv if _HAS_BROWSERGYM else None,
prompt_builder=prompt_builder,
action_parser=action_parser,
tasks=["click-test"],
task_var="BROWSERGYM_TASK_NAME",
miniwob_url=os.getenv("MINIWOB_URL", "http://host.docker.internal:8888/miniwob/"),
docker_image="browsergym-env:latest",
benchmark="miniwob",
timeout_ms=10000,
num_generations=1,
env_vars={
"BROWSERGYM_BENCHMARK": "miniwob",
"BROWSERGYM_HEADLESS": "true",
"BROWSERGYM_VIEWPORT_WIDTH": "1280",
"BROWSERGYM_VIEWPORT_HEIGHT": "720",
"BROWSERGYM_TIMEOUT": "10000",
"BROWSERGYM_OBS_AXTREE": "1",
"BROWSERGYM_OBS_PRUNED_HTML": "1",
"BROWSERGYM_RETURN_INFO": "1",
"MINIWOB_URL": os.getenv("MINIWOB_URL", "http://host.docker.internal:8888/miniwob/"),
},
)
if _HAS_BROWSERGYM
else None
),
)
def test_openenv_browsergym_eval(row: EvaluationRow) -> EvaluationRow:
"""
Example: run a BrowserGym MiniWoB environment via OpenEnvRolloutProcessor.
"""
if not _HAS_BROWSERGYM:
pytest.skip("OpenEnv (envs.browsergym_env) is not installed; skipping BrowserGym test.")
# The rollout processor appends per-step rewards in a sentinel system message:
# "__ep_step_rewards__:[r0, r1, ...]".
step_rewards: List[float] = []
try:
for msg in row.messages or []:
if (
msg.role == "system"
and isinstance(msg.content, str)
and msg.content.startswith("__ep_step_rewards__:")
):
import json as _json
payload = msg.content.split(":", 1)[1]
step_rewards = _json.loads(payload) or []
break
except Exception:
step_rewards = []
total = float(sum(step_rewards)) if step_rewards else 0.0
# Map total reward into [0, 1]
score = max(0.0, min(1.0, total))
reason = f"Total reward={total:.2f} across {len(step_rewards)} steps"
row.evaluation_result = EvaluateResult(score=score, reason=reason)
return row
```
This pattern generalizes to **any OpenEnv client**:
* Swap `BrowserGymEnv` / `BrowserGymAction` for `EchoEnv` / `EchoAction`, `TextArenaEnv` / `TextArenaAction`, or your own environment class.
* Keep `prompt_builder` and `action_parser` aligned with the environment’s observation and action types.
* Reuse the same `@evaluation_test` file across offline evals, dashboards, and RL integrations that call Eval Protocol.
## Echo / TextArena and connection modes
`OpenEnvRolloutProcessor` can construct environments in three main ways, all driven by `env_client_cls`:
* **From Hugging Face Hub (recommended)** — `from_hub`:
```python theme={null}
from envs.echo_env import EchoEnv
processor = OpenEnvRolloutProcessor(
env_client_cls=EchoEnv,
hub_repo_id="openenv/echo-env", # HF Space repo_id
prompt_builder=prompt_builder,
action_parser=action_parser,
timeout_ms=5000,
)
```
When you use `EchoEnv.from_hub("openenv/echo-env")`, OpenEnv will pull and start the container for you locally. Internally it runs a command similar to:
```bash theme={null}
docker run -d -p 8001:8000 --platform linux/amd64 registry.hf.space/openenv-echo-env:latest
```
You typically do **not** need to run this yourself; it is shown here so you know what OpenEnv is doing under the hood and can debug or run it manually if needed.
* **Local / Docker image (TextArena, BrowserGym, custom)** — `from_docker_image`:
```python theme={null}
from envs.textarena_env import TextArenaEnv
processor = OpenEnvRolloutProcessor(
env_client_cls=TextArenaEnv,
docker_image="textarena-env:latest",
env_vars={
"TEXTARENA_ENV_ID": "Wordle-v0",
"TEXTARENA_NUM_PLAYERS": "1",
},
task_var="TEXTARENA_ENV_ID",
tasks=None, # single env id via TEXTARENA_ENV_ID
prompt_builder=textarena_prompt_builder,
action_parser=textarena_action_parser,
)
```
* **Existing HTTP server / remote Space** — `base_url`:
```python theme={null}
from envs.echo_env import EchoEnv
# Local or Docker-mapped port
local_client = EchoEnv(base_url="http://0.0.0.0:8001")
# Remote Hugging Face Space
space_client = EchoEnv(base_url="https://openenv-echo-env.hf.space")
```
With `OpenEnvRolloutProcessor`, you can pass a factory instead of `env_client_cls`:
```python theme={null}
def make_echo_env():
return EchoEnv(base_url="https://openenv-echo-env.hf.space")
processor = OpenEnvRolloutProcessor(
env_factory=make_echo_env,
prompt_builder=prompt_builder,
action_parser=action_parser,
)
```
Once your OpenEnv client is wired into `OpenEnvRolloutProcessor`, all Eval Protocol tooling (evaluation tests, logs UI, and integrations like TRL/rLLM) can reuse the same environment + reward logic by simply pointing at your `@evaluation_test` function via its module path.
# rLLM Trainer
Source: https://evalprotocol.io/integrations/rllm-trainer
Reuse Eval Protocol environments and evaluation tests as workflows inside the rLLM reinforcement learning framework
This adapter lets you **run Eval Protocol environments and evaluation tests as rLLM workflows** for reinforcement learning training. It does this by pointing rLLM at an Eval Protocol `@evaluation_test`, which uses Eval Protocol’s rollout processor to generate trajectories, calls the same evaluation function you use for offline evals, and converts the result into rLLM’s abstractions. This makes it easy to start with rLLM and later move to other Eval-Protocol supported training workflows (or vice versa) without rewriting your evals.
For an end to end example, see the [FrozenLake Eval Protocol example](https://github.com/rllm-org/rllm/tree/main/examples/eval_protocol).
## High Level Overview
The core integration lives in rLLM’s `EvalProtocolWorkflow` (implemented in [`rllm/workflows/eval_protocol_workflow.py`](https://github.com/rllm-org/rllm/blob/main/rllm/workflows/eval_protocol_workflow.py)):
```python theme={null}
from rllm.workflows.eval_protocol_workflow import EvalProtocolWorkflow
```
You typically use it together with rLLM’s workflow engine. Under the hood, `EvalProtocolWorkflow`:
* **Takes** an Eval Protocol `@evaluation_test` (found via its module path, e.g. `"eval_protocol.benchmarks.test_frozen_lake"`).
* **Reads** the test’s metadata (attached by `@evaluation_test`), including:
* `rollout_processor` (e.g., `MCPGymRolloutProcessor`)
* `server_script_path` / `mcp_config_path`
* rollout kwargs, mode, etc.
* **Builds** a rollout config combining:
* Eval Protocol metadata, and
* rLLM’s config (model id, temperature, max tokens, number of steps).
* **Runs** rollouts through Eval Protocol’s `rollout_processor`, then calls the evaluation function (your `@evaluation_test`) to produce an `EvaluationRow` with an `evaluation_result`.
* **Converts** the resulting `EvaluationRow` into an rLLM `Episode` / `Trajectory` / `Step`, attaching the final score and metrics.
This design means you can reuse the exact same Eval Protocol tests and MCP environments in rLLM with minimal extra glue code.
## Basic Usage
### 1. Define an Eval Protocol `@evaluation_test`
Start with a normal Eval Protocol test. For example, a FrozenLake environment that uses an MCP rollout processor:
```python test_frozen_lake.py theme={null}
@evaluation_test(
input_dataset=["tests/pytest/data/frozen_lake_dataset.jsonl"],
dataset_adapter=frozen_lake_to_evaluation_row,
completion_params=[
{
"temperature": 0.0,
"max_tokens": 4096,
"model": "fireworks_ai/accounts/fireworks/models/kimi-k2-instruct",
}
],
rollout_processor=MCPGymRolloutProcessor(),
passed_threshold=0.66,
num_runs=1,
max_concurrent_rollouts=3,
mode="pointwise",
server_script_path="examples/frozen_lake_mcp/server.py",
)
def test_frozen_lake_evaluation(row: EvaluationRow) -> EvaluationRow:
"""
Evaluate how well the model plays FrozenLake by checking if it reaches the
goal while avoiding holes.
"""
score = row.get_total_reward()
if score == 1.0:
reason = "Agent reached the goal"
else:
reason = "Agent did not reach the goal"
row.evaluation_result = EvaluateResult(
score=score,
reason=reason,
)
return row
```
This is a regular Eval Protocol test: it describes how to roll out (via `rollout_processor`) and how to score (via the body of `test_frozen_lake_evaluation`).
### 2. Prepare a dataset for rLLM
On the rLLM side, you typically build a small dataset of task dicts that `EvalProtocolWorkflow` can map into `EvaluationRow`s. For FrozenLake, rLLM uses a script like:
```python prepare_frozen_lake_data.py theme={null}
# examples/eval_protocol/prepare_frozen_lake_data.py (in rLLM)
from datasets import Dataset
from rllm.data.dataset import DatasetRegistry
def prepare_frozen_lake_data(train_size: int, test_size: int):
system_prompt = "..." # explains the FrozenLake rules and tool usage
user_prompt_template = "Current game state grid:\n{observation}\n\n..."
def create_row(idx, seed):
return {
"id": f"run_{idx}",
"system_prompt": system_prompt,
"user_prompt_template": user_prompt_template,
"environment_context": {
"game": "FrozenLake",
"map_name": "4x4",
"seed": seed,
},
}
# build HF datasets and register with DatasetRegistry under "frozen_lake_eval_protocol"
...
```
Each task row includes:
* `id`
* `system_prompt`
* `user_prompt_template` (e.g., uses `{observation}`)
* `environment_context` (whatever your Eval Protocol test expects)
Those fields are converted to an `EvaluationRow` by `EvalProtocolWorkflow`’s `_task_to_evaluation_row`.
### 3. Run Eval Protocol tests through `AgentWorkflowEngine`
To run evals (no training), rLLM uses `AgentWorkflowEngine` with `EvalProtocolWorkflow`:
```python run_frozen_lake_flow.py theme={null}
# examples/eval_protocol/run_frozen_lake_flow.py (in rLLM)
from rllm.data.dataset import DatasetRegistry
from rllm.engine.agent_workflow_engine import AgentWorkflowEngine
from rllm.engine.rollout.openai_engine import OpenAIEngine
from rllm.workflows.eval_protocol_workflow import EvalProtocolWorkflow
async def main():
model_id = "accounts/fireworks/models/kimi-k2-instruct"
rollout_engine = OpenAIEngine(
model=model_id,
base_url="https://api.fireworks.ai/inference/v1",
api_key=os.getenv("FIREWORKS_API_KEY"),
)
engine = AgentWorkflowEngine(
workflow_cls=EvalProtocolWorkflow,
workflow_args={
"env_path": "eval_protocol.benchmarks.test_frozen_lake",
"lite_llm_prefix": "fireworks_ai/",
"steps": 30,
"temperature": 1.0,
"max_tokens": 16384,
},
rollout_engine=rollout_engine,
n_parallel_tasks=4,
retry_limit=1,
)
test_dataset = DatasetRegistry.load_dataset("frozen_lake_eval_protocol", "test")
tasks = [test_dataset[i] for i in range(4)]
episodes = await engine.execute_tasks(tasks)
...
```
Key points:
* `workflow_cls=EvalProtocolWorkflow` tells rLLM to use the Eval Protocol adapter.
* `env_path="eval_protocol.benchmarks.test_frozen_lake"` points to the module containing your `@evaluation_test`.
* `EvalProtocolWorkflow` imports that module, finds the decorated test with its metadata, and wires everything together.
### 4. Train with `AgentTrainer` + `EvalProtocolWorkflow`
For reinforcement learning, rLLM plugs the same workflow into its trainer:
```python train_frozen_lake_flow.py theme={null}
# examples/eval_protocol/train_frozen_lake_flow.py (in rLLM)
import hydra
from rllm.data.dataset import DatasetRegistry
from rllm.trainer.agent_trainer import AgentTrainer
from rllm.workflows.eval_protocol_workflow import EvalProtocolWorkflow
@hydra.main(config_path="pkg://rllm.trainer.config", config_name="agent_ppo_trainer", version_base=None)
def main(config):
train_dataset = DatasetRegistry.load_dataset("frozen_lake_eval_protocol", "train")
test_dataset = DatasetRegistry.load_dataset("frozen_lake_eval_protocol", "test")
trainer = AgentTrainer(
workflow_class=EvalProtocolWorkflow,
workflow_args={
"env_path": "eval_protocol.benchmarks.test_frozen_lake",
"lite_llm_prefix": "fireworks_ai/",
"steps": 30,
"temperature": 1.0,
"max_tokens": 32768,
},
config=config,
train_dataset=train_dataset,
val_dataset=test_dataset,
backend="fireworks",
)
trainer.train()
```
Here, `AgentTrainer`:
* Uses `EvalProtocolWorkflow` as its sampler/workflow.
* Collects Episodes from Eval Protocol rollouts.
* Uses those Episodes as input to the underlying PPO/GRPO trainer.
## End-to-End FrozenLake Example
To see this in action:
1. Clone the rLLM repository.
2. Prepare the FrozenLake Eval Protocol dataset:
```bash theme={null}
cd examples/eval_protocol
python prepare_frozen_lake_data.py
```
3. Run the FrozenLake Eval Protocol workflow through rLLM:
```bash theme={null}
python run_frozen_lake_flow.py
```
4. Start training:
```bash theme={null}
bash train_frozen_lake_flow.sh
```
The same pattern applies to any other Eval Protocol test:
* Change `env_path` to the module containing your `@evaluation_test`.
* Prepare a matching dataset for rLLM (id, system prompt, user prompt template, environment context).
* Reuse `EvalProtocolWorkflow` with `AgentWorkflowEngine` and/or `AgentTrainer` to run or train on that environment.
# Training with TRL
Source: https://evalprotocol.io/integrations/trl-trainer
Connect environments to TRL to train language models
Eval Protocol makes it easy to connect your environments to open‑source trainers like [TRL](https://huggingface.co/docs/trl/en/index) and train language models on interactive environments such as web navigation, text games, and custom simulators. Our TRL integration lets Eval Protocol handle rollouts with any [OpenEnv](https://github.com/meta-pytorch/OpenEnv/tree/main) environment (and other Eval Protocol rollout processors), while TRL handles optimization, gradients, and checkpoints.
**Supported environments today:**
* **OpenEnv environments** via `OpenEnvRolloutProcessor` (BrowserGym, Echo, TextArena, Atari-style games, coding envs, and more).
* **Other Eval Protocol tests** that expose token IDs and rewards through a rollout processor (for example `SingleTurnRolloutProcessor`), with more trainers and environments added over time.
## Why Use Eval Protocol for TRL Training?
**Eval Protocol handles the rollouts:**
* Environment management (Docker containers, lifecycle)
* Rollout execution (observation → LLM → action → environment)
* Task Rotation
* Reward collection and formatting
* Concurrency control
**TRL handles training:**
* GRPO optimization
* Model updates
* Gradient computation
* Checkpointing
**You just need to:**
* Define how to build prompts from observations
* Define how to parse LLM outputs into actions
* Configure your environment and training parameters
## Architecture
At a high level, TRL and Eval Protocol split responsibilities:
* **TRL (GRPOTrainer)** owns the training loop: it calls a `rollout_func`, computes losses and gradients, and updates the model.
* **Eval Protocol** owns the rollout loop: it turns TRL prompts into `EvaluationRow`s, runs environments (for example via `OpenEnvRolloutProcessor`), calls your model through vLLM, and returns token IDs and rewards.
* **Environments** (OpenEnv or other Eval Protocol tests) are configured once in a `@evaluation_test` file, which Eval Protocol reuses both for offline evals and for TRL training.
When you pass a `rollout_func` created by `create_openenv_vllm_rollout_func` into `GRPOTrainer`, each training step looks like:
1. TRL calls `rollout_func(prompts, trainer)`.
2. Eval Protocol builds `EvaluationRow`s and runs rollouts using the configured rollout processor.
3. The `@evaluation_test` is executed to compute `evaluation_result.score` for each row.
4. Eval Protocol returns token IDs and scores to TRL, which computes gradients and updates the model.
## Prerequisites
### 1. Install Dependencies
```bash theme={null}
# Recommended: Eval Protocol with TRL + OpenEnv extras
pip install "eval-protocol[trl,openenv]"
# This installs:
# - TRL + friends: trl, transformers, peft, accelerate, torch (as needed)
# - OpenEnv packages: openenv-core, openenv, openenv-browsergym-env
# You do NOT need to clone the OpenEnv repo just to use hub or remote environments.
# Or install pieces separately
pip install "eval-protocol[trl]"
pip install openenv-core
pip install "openenv @ git+https://github.com/meta-pytorch/OpenEnv.git"
pip install "openenv-browsergym-env @ git+https://github.com/meta-pytorch/OpenEnv.git#subdirectory=src/envs/browsergym_env"
```
### 2. (Optional) Build local BrowserGym Docker images
You only need this step if you want to run **BrowserGym locally in Docker**.\
If you are using environments from the **Hugging Face Hub** (for example `EchoEnv.from_hub(...)`) or a **remote HTTP server/Space** via `base_url=...`, you can skip this section.
```bash theme={null}
# Clone OpenEnv (only needed for building local images)
git clone https://github.com/meta-pytorch/OpenEnv.git
cd OpenEnv
# Build OpenEnv base image
docker build -t openenv-base:latest -f src/core/containers/images/Dockerfile .
# Build BrowserGym environment
docker build -t browsergym-env:latest -f src/envs/browsergym_env/server/Dockerfile .
```
### 3. Start vLLM Server
Start TRL's vLLM server on a separate GPU:
```bash theme={null}
# Use an INSTRUCT model for better instruction following
CUDA_VISIBLE_DEVICES=0 trl vllm-serve \
--model Qwen/Qwen2.5-7B-Instruct \
--port 8000
```
Use a separate GPU for vLLM inference (GPU 0) and training (GPU 1) for best performance.
### 4. Setup Environment (BrowserGym + MiniWoB++ example)
This step is only required if you are training on **MiniWoB++ BrowserGym tasks** locally. For other environments (Echo, TextArena, remote BrowserGym on the hub/Spaces), you can skip it.
For MiniWoB++ tasks, serve the HTML locally:
```bash theme={null}
# Clone MiniWoB++ (if you don't have it yet)
git clone https://github.com/Farama-Foundation/miniwob-plusplus.git
# From the cloned repo root:
cd miniwob-plusplus/miniwob/html
python -m http.server 8888 --bind 0.0.0.0
```
Set environment variables:
```bash theme={null}
export MINIWOB_URL="http://host.docker.internal:8888/miniwob/" # macOS
# or
export MINIWOB_URL="http://172.17.0.1:8888/miniwob/" # Linux
```
## Reusing Eval Protocol tests with TRL
The biggest value of this integration is that you can **reuse your existing `@evaluation_test` files** for training:
* The Eval Protocol test owns the **environment wiring** (for example `OpenEnvRolloutProcessor` + BrowserGym/Echo/TextArena config).
* The test body owns the **reward logic** (it sets `row.evaluation_result`).
* TRL just points at that test by module path and reuses both environment and scoring.
You can see a concrete example in the Eval Protocol Python SDK repo at:
[tests.pytest.test\_openenv\_browsergym\_eval](https://github.com/eval-protocol/python-sdk/blob/main/tests/pytest/test_openenv_browsergym_eval.py).
Eval Protocol’s `create_openenv_vllm_rollout_func` helper:
* Looks up your `@evaluation_test` via its `env_path` (for example `"tests.pytest.test_openenv_browsergym_eval"`).
* Reuses the attached `OpenEnvRolloutProcessor` configuration (env client, tasks, env vars, timeouts, etc.).
* Runs the test function itself to populate `row.evaluation_result`.
* Returns a `rollout_func` that produces token IDs and rewards in the format TRL expects.
This means you can:
* Add a new environment by writing a **single `@evaluation_test`**.
* Use the same test in:
* Offline evals and dashboards (`ep logs`).
* TRL training, by pointing `env_path` at that test.
## Inspecting TRL rollouts in the Logs UI
Because all rollouts go through Eval Protocol, every TRL training step that uses this integration is also visible in the **Eval Protocol Logs UI**:
* Each call to `rollout_func` creates one or more `EvaluationRow`s, just like a normal eval run.
* Those rows are logged with `EvalMetadata` so you can filter by eval name, time, and status.
* You can inspect the full message history, actions, rewards, and token usage for each rollout.
![TRL rollouts in Eval Protocol logs]()
## Quick Start: BrowserGym + TRL (reusing an eval test)
Below is a minimal example that trains on BrowserGym MiniWoB++ tasks by reusing an existing Eval Protocol test (`tests.pytest.test_openenv_browsergym_eval`) that already configures `OpenEnvRolloutProcessor` and reward logic.
```python train_browsergym_trl.py theme={null}
from typing import Any, List
from datasets import Dataset
from transformers import AutoTokenizer
from trl import GRPOConfig, GRPOTrainer
from peft import LoraConfig
from eval_protocol.pytest.integrations.openenv_trl_vllm import create_openenv_vllm_rollout_func
from envs.browsergym_env import BrowserGymAction
MODEL = "Qwen/Qwen2.5-7B-Instruct"
VLLM_URL = "http://localhost:8000"
# Module path to the Eval Protocol @evaluation_test we want to reuse.
EVAL_ENV_PATH = "tests.pytest.test_openenv_browsergym_eval"
# 1. Define prompt builder (observation → text for LLM)
def build_prompt(obs: Any, step: int, history: List[str]) -> str:
goal = getattr(obs, "goal", "") or ""
url = getattr(obs, "url", "") or "(unknown)"
text = (getattr(obs, "text", "") or "")[:1500]
history_block = "\n".join(history[-4:]) if history else "None"
return (
f"Step {step}\n"
f"Goal: {goal}\n"
f"URL: {url}\n"
f"Previous steps:\n{history_block}\n\n"
f"Page excerpt:\n{text}\n\n"
"Reply with a single BrowserGym action, e.g., click('13') or noop()."
)
# 2. Define action parser (LLM text → environment action)
def parse_action(text: str) -> BrowserGymAction:
import re
match = re.search(r"[A-Za-z_]+\\s*\\(.*\\)", text)
if match:
return BrowserGymAction(action_str=match.group(0))
return BrowserGymAction(action_str="noop()")
# 3. Define reward function (uses eval_protocol evaluation scores)
def reward_func(completions, **kwargs):
"""
Reward per episode taken from eval_protocol's evaluation_result.score.
The rollout_func runs the @evaluation_test for each EvaluationRow and
exposes the score as `eval_score`.
"""
eval_scores = kwargs.get("eval_score") or []
if eval_scores:
return [float(s) for s in eval_scores]
return [0.0] * len(completions)
# 4. Create rollout function (Eval Protocol handles OpenEnv + vLLM)
rollout_func = create_openenv_vllm_rollout_func(
env_factory=None,
env_client_cls=None, # taken from the @evaluation_test
prompt_builder=build_prompt,
action_parser=parse_action,
vllm_base_url=VLLM_URL,
vllm_model=MODEL,
env_path=EVAL_ENV_PATH, # reuse OpenEnvRolloutProcessor config + rewards
max_steps=6,
completion_params={
"temperature": 0.7,
"max_tokens": 1024,
},
concurrency=2,
)
# 5. Setup TRL trainer
tokenizer = AutoTokenizer.from_pretrained(MODEL)
dataset = Dataset.from_dict({"prompt": ["Start task"] * 6})
training_args = GRPOConfig(
output_dir="outputs/browsergym",
per_device_train_batch_size=2,
num_generations=2,
num_train_epochs=1,
learning_rate=5e-6,
max_completion_length=100,
max_prompt_length=4096,
logging_steps=1,
use_vllm=True,
vllm_mode="colocate", # or "server" if you use a separate vLLM server
vllm_gpu_memory_utilization=0.5,
)
trainer = GRPOTrainer(
model=MODEL,
args=training_args,
train_dataset=dataset,
processing_class=tokenizer,
reward_funcs=reward_func,
rollout_func=rollout_func, # ← Eval Protocol handles OpenEnv + vLLM here
peft_config=LoraConfig(r=16, lora_alpha=16, target_modules="all-linear"),
)
def main():
trainer.train()
if __name__ == "__main__":
main()
```
## Running the Training
```bash theme={null}
# Start vLLM server (GPU 0)
CUDA_VISIBLE_DEVICES=0 trl vllm-serve --model Qwen/Qwen2.5-7B-Instruct --port 8000
# In another terminal, start MiniWoB server
cd miniwob-plusplus/miniwob/html
python -m http.server 8888 --bind 0.0.0.0
# In another terminal, run training (GPU 1)
CUDA_VISIBLE_DEVICES=1 PYTHONUNBUFFERED=1 python train_browsergym.py
```
## How It Works
When you call `create_openenv_vllm_rollout_func()`, eval-protocol creates a function that TRL's trainer will call during training. Here's what happens:
1. **TRL calls `rollout_func(prompts, trainer)`** with a batch of prompts
2. **eval-protocol creates evaluation rows** from the prompts (one row per generation)
3. **OpenEnvRolloutProcessor executes rollouts**:
* Creates Docker containers for environments
* Runs the agent loop: observation → LLM → action → environment
* Collects rewards and tokens from each step
4. **eval-protocol formats results** into TRL-compatible format (token IDs + rewards)
5. **TRL uses the results** to compute policy gradients and update the model
You don't need to worry about Docker management, concurrency, or reward collection—eval-protocol handles it all!
## Configuration Parameters
### Rollout Function Parameters
OpenEnv environment client class (e.g., `BrowserGymEnv`, `TextArenaEnv`)
Function that converts observation to text prompt for the LLM
Function that converts LLM text output to environment action
URL of the TRL vLLM server
Model name on the vLLM server
List of tasks to rotate through during training
Environment variable name for task selection (required when `tasks` is provided)
Environment variables to pass to Docker containers
Docker image for the environment
Maximum steps per episode
LLM sampling parameters (temperature, max\_tokens, etc.)
Maximum concurrent rollouts (defaults to batch size)
### GRPO Training Parameters
Batch size per device
Number of rollouts per prompt (must divide evenly into batch size)
Learning rate for training
Sampling temperature for generation
Maximum tokens per generation
Must be `True` to use vLLM server
Must be `"server"` to use separate vLLM server
URL of the vLLM server
## Best Practices
### 1. Use Instruct Models
Use instruction-tuned models (e.g., `Qwen2.5-7B-Instruct`) rather than base models for better instruction following:
```python theme={null}
MODEL = "Qwen/Qwen2.5-7B-Instruct" # ✅ Good
# MODEL = "Qwen/Qwen2.5-7B" # ❌ Base model may not follow instructions well
```
### 2. Separate GPUs for Inference and Training
Run vLLM inference on one GPU and training on another:
```bash theme={null}
# GPU 0: vLLM inference
CUDA_VISIBLE_DEVICES=0 trl vllm-serve --model MODEL --port 8000
# GPU 1: Training
CUDA_VISIBLE_DEVICES=1 python train.py
```
### 3. Use LoRA for Efficiency
LoRA reduces memory usage and speeds up training:
```python theme={null}
peft_config = LoraConfig(
r=16, # Rank (higher = more parameters)
lora_alpha=16,
target_modules="all-linear", # Apply to all linear layers
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM",
)
```
### 4. Balance Batch Size and Generations
Ensure `per_device_train_batch_size` is divisible by `num_generations`:
```python theme={null}
per_device_train_batch_size=4 # ✅ Divisible by num_generations
num_generations=2
```
### 5. Monitor Rewards
Track average rewards to ensure learning progress:
```python theme={null}
def reward_func(completions, **kwargs):
step_rewards = kwargs.get("step_rewards", [])
avg_reward = sum(step_rewards) / len(step_rewards) if step_rewards else 0.0
print(f"Average reward: {avg_reward:.2f}")
return [float(r) for r in step_rewards]
```
## Troubleshooting
### vLLM Server Not Found
**Error**: `Connection refused to http://localhost:8000`
**Solution**: Ensure vLLM server is running:
```bash theme={null}
CUDA_VISIBLE_DEVICES=0 trl vllm-serve --model MODEL --port 8000
```
### Docker Container Fails
**Error**: `RuntimeError: Failed to start Docker container`
**Solution**:
* Verify Docker image exists: `docker images | grep browsergym-env`
* Check container logs: `docker logs `
* Ensure environment variables are correct
### Out of Memory
**Error**: `CUDA out of memory`
**Solution**:
* Use LoRA instead of full fine-tuning
* Reduce `per_device_train_batch_size`
* Reduce `max_completion_length`
* Enable `gradient_checkpointing=True`
### Low Rewards
If rewards remain low:
* Verify your reward function is correct
* Check that environment tasks are solvable
* Review LLM outputs in rollout logs
* Adjust temperature (lower = more deterministic)
* Improve prompt engineering
## Advanced: Custom Reward Functions
You can implement custom reward shaping:
```python theme={null}
def custom_reward_func(completions, **kwargs):
"""Custom reward with shaping."""
step_rewards = kwargs.get("step_rewards", [])
shaped_rewards = []
for reward in step_rewards:
# Reward shaping: bonus for positive rewards
shaped = reward
if reward > 0:
shaped += 0.1 # Bonus for any success
shaped_rewards.append(shaped)
return shaped_rewards
```
## Resources
* [TRL Documentation](https://huggingface.co/docs/trl)
* [GRPO Paper](https://arxiv.org/abs/2402.03300)
* [OpenEnv Documentation](https://meta-pytorch.org/OpenEnv/)
* [eval-protocol TRL Integration](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/integrations/openenv_trl_vllm.py)
* [Example Training Script](https://github.com/eval-protocol/python-sdk/blob/main/examples/trl/train_browsergym.py)
# Introduction to Eval Protocol (EP)
Source: https://evalprotocol.io/introduction
**Eval Protocol (EP) is an open solution for doing reinforcement learning fine-tuning on existing agents — across any language, container, or framework.**
![]()
Most teams already have complex agents running in production — often across remote services with heavy dependencies, Docker containers, or TypeScript backends deployed on Vercel. When they try to train or fine-tune these agents with reinforcement learning, connecting them to a trainer quickly becomes painful.
Eval Protocol makes this possible in two ways:
1. **Expose Your Agent Through a Simple API**
Wrap your existing agent (Python, TypeScript, Docker, etc.) in a simple HTTP service using EP’s rollout interface. EP handles the rollout orchestration, metadata passing, and trace storage automatically.
2. **Connect With Any Trainer**
Once your agent speaks the EP standard, it can be fine-tuned or evaluated with any supported trainer — Fireworks RFT, TRL, Unsloth, or your own — with no environment rewrites.
The result: RL that works out-of-the-box for existing production agents.
## Who This Is For
* **Applied AI teams** adding RL to existing production agents.
* **Research engineers** experimenting with fine-tuning complex, multi-turn or tool-using agents.
* **MLOps teams** building reproducible, language-agnostic rollout pipelines.
## Getting Started
Try our Quickstart to see how we built and trained an SVGAgent end-to-end using the RemoteRolloutProcessor — including full Fireworks Tracing integration.
# MCP Control/Data Planes
Source: https://evalprotocol.io/mcp-extensions
EP adopts a clear split between the data plane (MCP calls that carry observations) and the control plane (HTTP endpoints for rewards, termination, and lifecycle). This separation improves reproducibility, session awareness, and failure recovery.
EP separates agent evaluation into two independent planes: data plane (MCP tool calls carrying observations) and control plane (HTTP endpoints for rewards and termination). This architectural split prevents observation/reward coupling that breaks caching and session isolation, while enabling graceful failure recovery—if reward calculation fails, agents still receive observations and evaluations continue with safe defaults.
## The Split
* Data plane (MCP): `list_tools`, `call_tool`, `list_resources`/`read_resource`.
* Purpose: tool schemas and observations only--what the agent receives.
* Control plane (HTTP): `/control/*` endpoints with `mcp-session-id` header.
* Purpose: initial state, reward, status (terminated/truncated), and reset.
Separation rules:
* Observations never come from control plane endpoints.
* Rewards/termination never come from tool results.
### Sequence diagram (data vs control planes)
```mermaid theme={null}
sequenceDiagram
participant LLMPolicy as LLM Policy (Fireworks)
participant MCPClient as MCP Client
participant MCPServer as MCP-Gym Server
participant ControlPlane as HTTP Control Plane
Note over LLMPolicy,MCPClient: CONVERSATION FLOW
LLMPolicy->>LLMPolicy: Clean messages (strip metadata)
LLMPolicy->>LLMPolicy: Generate tool calls via LLM API
Note over MCPClient,MCPServer: DATA PLANE (MCP Protocol)
MCPClient->>MCPServer: MCP Tool Call (lake_move)
MCPServer->>MCPServer: Execute environment step
MCPServer->>ControlPlane: Update session state (reward, terminated)
MCPServer-->>MCPClient: MCP Response (observation only)
Note over MCPClient,ControlPlane: CONTROL PLANE (HTTP Endpoints)
MCPClient->>ControlPlane: GET /control/reward (session-id header)
ControlPlane-->>MCPClient: {"reward": 1.0}
MCPClient->>ControlPlane: GET /control/status (session-id header)
ControlPlane-->>MCPClient: {"terminated": true}
Note over LLMPolicy,MCPClient: TRAJECTORY RECORDING
MCPClient->>LLMPolicy: Add tool response + metadata
LLMPolicy->>LLMPolicy: Record with control plane data
```
EP’s client enforces this separation in [MCPConnectionManager](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/client/connection.py) and [GeneralMCPVectorEnv](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/session/manager.py).
## Control Plane Endpoints
Servers should implement the following endpoints alongside their MCP transport (e.g., at `https://your-server.example/mcp` for MCP, and `https://your-server.example/control/...` for control):
* `POST /control/reset_session`
* Headers: `mcp-session-id: `
* Body: `{ "seed": }`
* Use: cleanup/reseed before a rollout or at close.
* `GET /control/initial_state`
* Headers: `mcp-session-id: `
* Returns: JSON initial observation/state used to seed the first user prompt.
* `GET /control/reward`
* Headers: `mcp-session-id: `
* Returns: `{ "reward": }` for the most recent step.
* `GET /control/status`
* Headers: `mcp-session-id: `
* Returns: `{ "terminated": , "truncated": }` to indicate episode end.
Notes:
* EP generates a stable `session_id` by hashing dataset row values and the model ID via `gen_session_id(...)` and passes it in MCP `clientInfo` and as the control-plane header. Heads up: it does not use run ID, so between runs, the MCP server needs to be restarted. This is automatically done in the current implementation of `MCPGymRolloutProcessor()`.
* The simulator framework ([SimulationServerBase](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/simulation_server.py)) demonstrates session-aware design but you still need to expose the `/control/*` endpoints in your production server. Note: the EP client does not depend on `SimulationServerBase`; it is provided as a reference pattern only.
## End-to-End Flows
### 1) Initialization
1. EP opens a streamable MCP session and sends `clientInfo` with `session_id`, `seed`, `config`, and `model_id`.
2. EP pre-warms tool schemas via `list_tools` (data plane) and caches them.
3. EP fetches initial state via `GET /control/initial_state` (control plane); if that times out or fails, it falls back to `list_resources`/`read_resource` (data plane) heuristics.
4. The initial observation seeds the first user prompt with your `user_prompt_template`.
Key guarantees:
* Initial state is session-aware (derived from control plane when available).
* Tool schemas are cached per `base_url` to avoid thundering herds.
### 2) Step Execution (per agent turn)
1. Policy returns one or more MCP tool calls based on tool schemas and conversation history.
2. EP executes the tool call via `call_tool` (data plane) and parses the observation from tool content.
3. EP queries control plane for reward and status:
* `GET /control/reward` → scalar reward
* `GET /control/status` → `terminated`/`truncated`
4. EP attaches a control-plane step summary to the conversation for logging, including reward, termination, and tool calls.
Separation:
* Observations never come from control plane endpoints
* Rewards/termination never come from tool results.
### 3) Termination
An episode ends when any of the following occurs:
* Control plane status reports `terminated` (environment signaled end) or `truncated` (cutoff).
* The policy returns `_no_tool_call` or `_playback_terminate` (e.g., model finished or playback hit the end).
* The simulated user signals stop; EP maps this to `termination_reason = user_stop`.
EP maps LLM finish reasons into `TerminationReason` values: `stop`, `length`, `tool_calls`, plus environment-driven `control_plane_signal`, `max_steps`, `user_stop`, `error`.
### 4) Failure Recovery
EP is defensive at the boundaries between planes:
* Initial state: If `/control/initial_state` fails or times out, EP falls back to `read_resource` (and ultimately a default observation) so rollouts can proceed.
* Tool responses: If a tool returns invalid/empty JSON, EP wraps it into a structured observation with an error tag instead of failing hard.
* Control queries: `/control/reward` and `/control/status` use short timeouts; absent data yields defaults (0.0 reward, not-terminated) and the step continues.
* Session re-init: Re-initialization closes any existing session handles and re-opens cleanly before retrying.
### 5) Cleanup
* At `close`, EP calls `POST /control/reset_session` and then closes the MCP transport.
## Minimal Client Example
```python theme={null}
import eval_protocol as ep
from eval_protocol.models import EvaluationRow, Message
rows = [
EvaluationRow(
messages=[Message(role="system", content="Use tools to help the user.")],
input_metadata={
"dataset_info": {
"user_prompt_template": "Observation: {observation}",
"environment_context": {"seed": 123}
}
},
)
]
envs = ep.make("https://your-server.example/mcp", evaluation_rows=rows, model_id="my-model")
policy = ep.OpenAIPolicy(model_id="gpt-4o-mini")
async def run():
async for row in ep.rollout(envs, policy=policy, steps=64, openai_format_log_file="terminated.jsonl"):
print(row.rollout_status.status, row.rollout_status.termination_reason)
```
## Multi-Server Aggregation (Optional)
If you need to aggregate tools from multiple MCP servers, EP provides [MCPMultiClient](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/mcp_multi_client.py) that connects to both stdio and remote servers and exposes all tools under one client.
```json theme={null}
{
"mcpServers": {
"local": { "command": "python", "args": ["-m", "my_mcp_server"], "env": ["API_KEY"] },
"remote": { "url": "https://your-server.example/mcp" }
}
}
```
This is independent from the control plane split; each server still implements its own `/control/*` endpoints.
## Record/Playback
Set `EP_PLAYBACK_FILE` to enable deterministic record/playback. During playback, the policy is stepped to match prior turns, and `_playback_terminate` ends the episode at the recorded boundary. Control-plane step summaries and an optional OpenAI-format log are emitted for terminated trajectories.
## Server Implementation Checklist
Use this as a reference when building the control plane alongside your MCP server.
* Headers: include `mcp-session-id` on every control request; return `Content-Type: application/json`.
* Session ID: treat as opaque but stable per dataset row + model; do not coalesce across different seeds/config.
* Idempotency: make `POST /control/reset_session` safe to call multiple times; ignore duplicate resets.
* Initialization:
* `GET /control/initial_state` returns the initial observation JSON for this session, derived from `seed` and `config` (from MCP `clientInfo`).
* Keep this response free of reward/termination fields; it seeds the first user prompt only.
* Step reporting:
* `GET /control/reward` returns `{ "reward": }` for the most recent applied action.
* `GET /control/status` returns `{ "terminated": , "truncated": }` for the episode state.
* Do not include observation content here; that stays in the data plane.
* Timeouts and SLAs:
* EP uses \~15s timeout for initial\_state under high concurrency (3s in playback) and \~3s for reward/status.
* Aim for sub-1s responses; if computation is heavy, cache per `session_id`.
* Errors:
* Use `4xx` for client mistakes (missing/invalid `mcp-session-id`), `5xx` for server errors.
* On faults, respond with a minimal JSON error body; EP will default to `reward=0.0` and `terminated=false` on non-200s.
* Concurrency:
* Expect many concurrent sessions; isolate per `session_id` and avoid global mutable state.
* Ensure tool results (data plane) and control updates are applied atomically in your environment loop.
* Security:
* You may authenticate control endpoints; keep auth orthogonal to `mcp-session-id` routing.
* Validate reasonable `session_id` lengths to prevent abuse.
Example responses
```http theme={null}
GET /control/initial_state
200 OK
Content-Type: application/json
{
"observation": "initial_state",
"grid_layout": "...",
"session_id": ""
}
```
```http theme={null}
GET /control/reward
200 OK
Content-Type: application/json
{ "reward": 1.0 }
```
```http theme={null}
GET /control/status
200 OK
Content-Type: application/json
{ "terminated": false, "truncated": false }
```
```http theme={null}
POST /control/reset_session
200 OK
Content-Type: application/json
{ "ok": true }
```
## Reading clientInfo on the Server
Servers using the low-level MCP server can extract `clientInfo` extras to create stable, session-aware environments. Example:
```python theme={null}
from mcp.server.lowlevel import Server
app = Server("MyServer")
@app.call_tool()
async def call_tool(name: str, arguments: dict):
# Access per-request context
ctx = app.request_context
session_id = None
seed = None
config = {}
if hasattr(ctx, "session") and hasattr(ctx.session, "client_params"):
client_params = ctx.session.client_params
if hasattr(client_params, "clientInfo"):
client_info = client_params.clientInfo
if client_info and hasattr(client_info, "_extra"):
extra = client_info._extra or {}
session_id = extra.get("session_id")
seed = extra.get("seed")
config = extra.get("config", {})
env = get_or_create_env(session_id=session_id, seed=seed, config=config)
# Apply action and return observation (data plane only)
observation = env.step(name, arguments)
return [{"type": "text", "text": json.dumps(observation)}]
```
Notes:
* Use `session_id` as the key for per-session state. Seed and config should shape the initial state.
* Keep observations on the data plane; publish reward and termination via `/control/*`.
## GitHub References
* Client: MCP connection manager (control/data split)
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/mcp/client/connection.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/client/connection.py)
* Client: Vector env/session manager
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/mcp/session/manager.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/session/manager.py)
* Server: MCP-Gym base with control-plane endpoints
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/mcp/mcpgym.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/mcpgym.py)
* Server: Simulation server base (session-aware patterns)
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/mcp/simulation\_server.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/simulation_server.py)
* Example servers implementing McpGym
* Frozen Lake: [https://github.com/eval-protocol/python-sdk/blob/main/examples/frozen\_lake\_mcp/frozen\_lake\_mcp.py](https://github.com/eval-protocol/python-sdk/blob/main/examples/frozen_lake_mcp/frozen_lake_mcp.py)
* Lunar Lander: [https://github.com/eval-protocol/python-sdk/blob/main/examples/lunar\_lander\_mcp/lunar\_lander\_mcp.py](https://github.com/eval-protocol/python-sdk/blob/main/examples/lunar_lander_mcp/lunar_lander_mcp.py)
* Cliff Walking: [https://github.com/eval-protocol/python-sdk/blob/main/examples/cliff\_walking\_mcp/cliff\_walking\_mcp.py](https://github.com/eval-protocol/python-sdk/blob/main/examples/cliff_walking_mcp/cliff_walking_mcp.py)
* Blackjack: [https://github.com/eval-protocol/python-sdk/blob/main/examples/blackjack\_mcp/blackjack\_mcp.py](https://github.com/eval-protocol/python-sdk/blob/main/examples/blackjack_mcp/blackjack_mcp.py)
* Tau2 domains: [https://github.com/eval-protocol/python-sdk/blob/main/examples/tau2\_mcp/tau2\_mcp.py](https://github.com/eval-protocol/python-sdk/blob/main/examples/tau2_mcp/tau2_mcp.py)
# Fine Tuning an SVGAgent with Eval Protocol
Source: https://evalprotocol.io/quickstart
Train and improve an SVG generation agent using reinforcement fine tuning with Eval Protocol
## Introduction
This repo demonstrates building an SVG generation agent using reinforcement fine tuning, with the parts:
* **Eval Protocol** - Orchestrates the rollout execution and evaluation framework
* **Vercel Typescript Server** - Remote server that handles SVG code generation rollouts
* **Fireworks RFT** - Reinforcement fine tuning trainer
A big thank you to [SVGBench](https://github.com/johnbean393/SVGBench) for the dataset. SVGBench is a comprehensive benchmark that evaluates language models on their ability to generate SVG code that meets specific visual requirements. Each prompt includes detailed criteria (like "draw a red circle in the top-left corner") that the generated SVG must fulfill.
**The Evaluation Process**: The model generates SVG code from text prompts, we render the SVGs to images, and then use GPT-4.1 as a visual judge to count how many requirements were fulfilled. This gives us concrete scores to measure improvement and lets you see dramatic before/after visual comparisons as your model gets better through training.
![SVG Agent Training Overview]()
## Quick Start
### Installation
1. **Create a Fireworks account**: [https://app.fireworks.ai/account/home](https://app.fireworks.ai/account/home)
2. **Clone the quickstart repo**: [https://github.com/eval-protocol/quickstart](https://github.com/eval-protocol/quickstart)
```bash theme={null}
git clone git@github.com:eval-protocol/quickstart.git
cd quickstart
```
3. **Install Eval Protocol**:
```bash theme={null}
pip install "eval-protocol[svgbench]"
```
4. **Environment Setup**:
The `env.example` file is located in the `evaluator/` directory. Make a copy of it in the same directory, name it `.env`, and fill in your API keys:
```bash theme={null}
cp evaluator/env.example evaluator/.env
```
Then edit `evaluator/.env` with your API keys:
```
FIREWORKS_API_KEY=your-fireworks-key-here
OPENAI_API_KEY=your-openai-key-here
```
The create process below automatically reads and uploads these secrets to Fireworks.
## Running Locally
**Terminal 1** - Start the local UI server to view results:
```bash theme={null}
ep logs
```
**Terminal 2** - Test locally:
```bash theme={null}
cd evaluator
ep local-test
```
This command discovers and runs your `@evaluation_test` with pytest. In this case, it builds an image and runs the test in Docker, because a `Dockerfile` is present.
The test automatically uses our Vercel remote server:
```
rollout_processor=RemoteRolloutProcessor(
remote_base_url="https://vercel-svg-server-ts.vercel.app",
)
```
If you want to use a local development Vercel server instead, see [Local Development Server](#local-development-server)
**Note:**
* If your evaluation setup has custom dependencies, for example Chromium, you will need containerize it using `Dockerfile`
* Then, when you run `ep local-test`, we will build an image and run pytest inside Docker
* If not, `ep local-test` will just run pytest on your host machine
* You can also ignore the `Dockerfile` and run on the host Python env using `ep local-test --ignore-docker`
### Expected Test Output:
Navigate to [http://localhost:8000](http://localhost:8000) to see the Eval Protocol UI.
```
INFO:eval_protocol.pytest.remote_rollout_processor:Found status log for rollout democratic-way-12: Rollout democratic-way-12 completed
INFO:eval_protocol.pytest.remote_rollout_processor:Found Fireworks log for rollout democratic-way-12 with status code 100.0
INFO:eval_protocol.adapters.fireworks_tracing:Successfully converted 1 traces to evaluation rows | 3/8 [00:19<00:22, 4.52s/rollout]
...
Runs (Parallel): 100%|████████████████████████████████████████████| 1/1 [00:31<00:00, 31.07s/run]
PASSED
```
![Eval Protocol Logs Interface]()
If you're interested in understanding how Remote Rollout Processing works and how it communicates with the remote server, see [How Remote Rollout Processing Works](#how-remote-rollout-processing-works).
## Single Command to Train
To kickoff training, simply do:
```bash theme={null}
eval-protocol create rft \
--base-model accounts/fireworks/models/qwen3-0p6b \
--chunk-size 10
```
This command:
1. **🔐 Uploads Secrets** - Automatically reads your `.env` file and uploads API keys as Fireworks secrets
2. **📦 Uploads Evaluator** - Packages and uploads your evaluation code
3. **⏳ Waits for Build** - Polls evaluator status every 10 seconds until ACTIVE (timeout: 10 minutes)
4. **📊 Creates Dataset** - Automatically uploads your `svgbench_dataset.jsonl`
5. **🚀 Launches RFT Job** - Starts reinforcement fine-tuning with your evaluator
### Configuration & Troubleshooting
**Training Parameters**: We use Eval Protocol's default values for training parameters (batch size, epochs, learning rate, LoRA rank, accelerator count, etc.). For a complete list of available RFT flags you can customize, see [Fireworks RFT Command Documentation](https://docs.fireworks.ai/tools-sdks/firectl/commands/create-reinforcement-fine-tuning-job).
**Changing Evaluators**: If you've made changes to your evaluator code and want to upload a new version:
```bash theme={null}
eval-protocol create rft \
--base-model accounts/fireworks/models/qwen3-0p6b \
--chunk-size 10 \
--force
```
**Evaluator Upload Timing Out**: If your evaluator takes longer than 10 minutes to build, you'll see:
```
⏰ Timeout after 10.0m - evaluator is not yet ACTIVE
❌ Evaluator is not ready within the timeout period.
📊 Please check the evaluator status at: https://app.fireworks.ai/dashboard/evaluators/test-svgagent-test-svg-generation-evaluation
Wait for it to become ACTIVE, then run 'eval-protocol create rft' again.
```
In this case, monitor the evaluator upload at the link, and run the command again when ACTIVE.
### Monitor Training Progress
After successful job creation, you'll see:
```
✅ Created Reinforcement Fine-tuning Job
name: accounts/pyroworks/reinforcementFineTuningJobs/sdnld4yn
📊 Dashboard Links:
Evaluator: https://app.fireworks.ai/dashboard/evaluators/test-svgagent-test-svg-generation-evaluation
Dataset: https://app.fireworks.ai/dashboard/datasets/svgbench-dataset
RFT Job: https://app.fireworks.ai/dashboard/fine-tuning/reinforcement/sdnld4yn
```
Click on the **RFT Job** link to view real-time training progress, epoch counts, and rollout data.
### Training Results
After successful training, you should see performance improvements reflected in the training metrics:
![SVG Agent Training Progress]()
### SVG Quality Improvement
You can inspect individual rollouts to see the dramatic improvement in SVG generation quality. Below is a comparison between the first epoch and the final 8th epoch:
**Before (1st Epoch):**
![SVG Generation - Before Training]()
**After (8th Epoch):**
![SVG Generation - After Training]()
The reinforcement fine tuning process significantly improves the model's ability to generate accurate, detailed SVG graphics that better match the input descriptions.
## Debugging Tips
When your training is running, you have several powerful tools to debug and monitor your rollouts:
### Rollout Overview
Clicking on any **Epoch** or **Step** in the training dashboard, then clicking the **table icon** to the right, will show you a comprehensive table of all rollouts. It's a good high-level overview to see if any rollouts failed and for what reason.
![Rollout Overview Table]()
### Individual Rollout Details
If you click on a specific row in the rollout table, you can see exactly what the prompt was and how the model responded. You can even copy and paste out the SVG code generated and render it yourself to see what the model did. This is how we got the results above in the before and after comparison.
![Individual Rollout Details]()
### Live Log Streaming
Clicking on **View Logs** takes you to a page of logs being streamed in. Here, you can see precisely what errors are happening to the rollouts. This is useful to debug and fix any issues with your rollouts.
![Live Log Streaming]()
## Contact Us / Learn More
* [Discord Server](https://discord.gg/mMqQxvFD9A). Come talk to us in the #eval-protocol channel!
* [Eval Protocol Documentation](https://evalprotocol.io/introduction)
* [Remote Rollout Processor Tutorial](https://evalprotocol.io/tutorial/remote-rollout-processor)
* [SVGBench Dataset](https://github.com/johnbean393/SVGBench) - The original benchmark this project is based on
* [Fireworks AI Platform](https://fireworks.ai)
## Appendix
### How Remote Rollout Processing Works
Eval Protocol enables **reinforcement learning that meets you where you are**. Instead of forcing you to rewrite your agent in a specific framework, you can implement a lightweight remote server wherever your codebase and infrastructure already live.
Your remote server is only responsible for:
* **Executing rollouts** - Run your agent logic (in this case, SVG generation from text prompts)
* **Logging to tracing** - Send structured logs to `tracing.fireworks.ai` for evaluation (see the below linked docs for more information)
In this example, we showcase a **Vercel TypeScript server** that executes single-turn SVG code generation.
**📖 Learn More**: For a complete deep-dive into Remote Rollout Processing, see the [Remote Rollout Processor Tutorial](https://evalprotocol.io/tutorial/remote-rollout-processor).
### Local Development Server
```bash theme={null}
cd vercel_svg_server_ts
vercel dev
```
Then swap out the `remote_base_url` to point to the local server you just started:
```
rollout_processor=RemoteRolloutProcessor(
remote_base_url="http://localhost:3000",
)
```
And in a third terminal, run the evaluation:
```bash theme={null}
ep local-test
```
See [Vercel CLI documentation](https://vercel.com/docs/cli/dev) for more information on local development.
# CLI
Source: https://evalprotocol.io/reference/cli
The `ep` command-line interface can inspect evaluation runs locally, upload evaluators, and create reinforcement fine-tuning jobs on Fireworks.
```bash theme={null}
ep [global options] [command options]
```
## Global Options
These options can be used with any command:
Enable verbose logging (Aliases: `-v`)
Fireworks API server hostname or URL (e.g., dev.api.fireworks.ai or [https://dev.api.fireworks.ai](https://dev.api.fireworks.ai))
## Commands
### `ep logs`
Serve logs with file watching and real-time updates
Port to bind to (default: 8000)
Enable debug mode
Disable Elasticsearch setup
Use env vars for Elasticsearch config (requires ELASTICSEARCH\_URL, ELASTICSEARCH\_API\_KEY, ELASTICSEARCH\_INDEX\_NAME)
Force Fireworks tracing backend for logs UI (overrides env auto-detection)
Force Elasticsearch backend for logs UI (overrides env auto-detection)
### `ep upload`
Scan for evaluation tests, select, and upload as Fireworks evaluators
Path to search for evaluation tests (default: current directory)
Entrypoint of evaluation test to upload (module:function or path::function). For multiple, separate by commas.
Non-interactive: upload all discovered evaluation tests (Aliases: `-y`)
Path to .env file containing secrets to upload (default: .env in current directory)
Overwrite existing evaluator with the same ID
Default dataset to use with this evaluator (Aliases: `--default-dataset`)
Description for evaluator (Aliases: `--description`)
Display name for evaluator (defaults to ID) (Aliases: `--name`, `--display-name`)
Pytest-style entrypoint (e.g., test\_file.py::test\_func). Auto-detected if not provided. (Aliases: `--entry-point`)
Requirements for evaluator (auto-detected from requirements.txt if not provided) (Aliases: `--requirements`)
Evaluator ID to use (if multiple selections, a numeric suffix is appended) (Aliases: `--id`)
### `ep create rft`
Create a Reinforcement Fine-tuning Job on Fireworks
Non-interactive mode (Aliases: `-y`)
Print planned SDK call without sending
Overwrite existing evaluator with the same ID
Skip local dataset/evaluator validation
Ignore Dockerfile even if present; run pytest on host during evaluator validation
Extra flags to pass to 'docker build' when validating evaluator (quoted string, e.g. "--no-cache --pull --progress=plain")
Extra flags to pass to 'docker run' when validating evaluator (quoted string, e.g. "--env-file .env --memory=8g")
Path to .env file containing secrets to upload to Fireworks (default: .env in project root)
The source reinforcement fine-tuning job to copy configuration from. If other flags are set, they will override the source job's configuration.
If set, only errors will be printed.
The name of the dataset used for training.
The evaluator resource name to use for RLOR fine-tuning job.
ID of the reinforcement fine-tuning job, a random UUID will be generated if not specified. (Aliases: `--job-id`)
Data chunking for rollout, default size 200, enabled when dataset > 300. Valid range is 1-10,000.
Whether to auto-carve the dataset for eval.
The name of a separate dataset to use for evaluation.
Additional parameters for the inference request as a JSON string. For example:
"\{"stop": \["\n"]}". (Aliases: `--extra-body`)
Maximum number of tokens to generate per response. (Aliases: `--max-output-tokens`)
Number of response candidates to generate per input. (Aliases: `--response-candidates-count`)
Sampling temperature, typically between 0 and 2. (Aliases: `--temperature`)
Top-k sampling parameter, limits the token selection to the top k tokens. (Aliases: `--top-k`)
Top-p sampling parameter, typically between 0 and 1. (Aliases: `--top-p`)
KL coefficient (beta) override for GRPO-like methods. If unset, the trainer
default is used. (Aliases: `--rl-kl-beta`, `--kl-beta`)
RL loss method for underlying trainers. One of \{grpo,dapo}. (Aliases: `--rl-loss-method`, `--method`)
The MCP server resource name to use for the reinforcement fine-tuning job. (Optional)
The number of nodes to use for the fine-tuning job. If not specified, the default is 1. (Aliases: `--nodes`)
The name of the base model to be fine-tuned Only one of 'base\_model' or
'warm\_start\_from' should be specified. (Aliases: `--base-model`)
The maximum packed number of tokens per batch for training in sequence packing. (Aliases: `--batch-size`)
The number of epochs to train for. (Aliases: `--epochs`)
The number of batches to accumulate gradients before updating the model parameters. The effective batch size will be batch-size multiplied by this value. (Aliases: `--gradient-accumulation-steps`)
The learning rate used for training. (Aliases: `--learning-rate`)
The number of learning rate warmup steps for the reinforcement fine-tuning job. (Aliases: `--learning-rate-warmup-steps`)
The rank of the LoRA layers. (Aliases: `--lora-rank`)
The maximum context length to use with the model. (Aliases: `--max-context-length`)
The model ID to be assigned to the resulting fine-tuned model.
If not specified, the job ID will be used. (Aliases: `--output-model`)
The PEFT addon model in Fireworks format to be fine-tuned from Only one of
'base\_model' or 'warm\_start\_from' should be specified. (Aliases: `--warm-start-from`)
The API key for the wandb service. (Aliases: `--wandb-api-key`, `--api-key`)
Whether to enable wandb logging. (Aliases: `--wandb`, `--enabled`)
The entity name for the wandb service. (Aliases: `--wandb-entity`, `--entity`)
The project name for the wandb service. (Aliases: `--wandb-project`, `--project`)
### `ep local-test`
Select an evaluation test and run it locally. If a Dockerfile exists, build and run via Docker; otherwise run on host.
Entrypoint to run (path::function or path). If not provided, a selector will be shown (unless --yes).
Ignore Dockerfile even if present; run pytest on host
Non-interactive: if multiple tests exist and no --entry, fails with guidance (Aliases: `-y`)
Extra flags to pass to 'docker build' (quoted string, e.g. "--no-cache --pull --progress=plain")
Extra flags to pass to 'docker run' (quoted string, e.g. "--env-file .env --memory=8g")
# Data Loader
Source: https://evalprotocol.io/reference/data-loader
Load evaluation data using DynamicDataLoader and InlineDataLoader for reusable, parameterized inputs
The data loader module provides a standard way to feed evaluation data into tests. Use it to:
* Build reusable input sources (adapters, files, generators)
* Parameterize datasets with clear variant labeling
* Preprocess inputs consistently (e.g., expand multi-turn data)
## Components
### DynamicDataLoader
Uses callables that return lists of `EvaluationRow`. Each callable becomes a labeled variant.
```python theme={null}
from eval_protocol import DynamicDataLoader
from eval_protocol.models import EvaluationRow
def my_generator() -> list[EvaluationRow]:
# Fetch or generate rows here (adapters, DB, etc.)
return []
data_loader = DynamicDataLoader(
generators=[my_generator],
)
```
### InlineDataLoader
Use when you have rows or raw messages inline.
```python theme={null}
from eval_protocol import InlineDataLoader
from eval_protocol.models import EvaluationRow, Message
inline_rows = [
EvaluationRow(messages=[
Message(role="user", content="Hello"),
Message(role="assistant", content="Hi there!"),
])
]
loader = InlineDataLoader(rows=inline_rows, id="demo", description="Two-turn chat")
```
## Preprocessing
All loaders support an optional `preprocess_fn` applied before returning rows. For example, expand multi-turn traces into multiple test cases:
```python theme={null}
from eval_protocol import DynamicDataLoader, multi_turn_assistant_to_ground_truth
DynamicDataLoader(
generators=[my_generator],
preprocess_fn=multi_turn_assistant_to_ground_truth,
)
```
## Using with evaluation\_test
```python theme={null}
from eval_protocol import evaluation_test, SingleTurnRolloutProcessor
@evaluation_test(
data_loaders=data_loader,
rollout_processor=SingleTurnRolloutProcessor(),
)
async def test_llm_judge(row: EvaluationRow) -> EvaluationRow:
return await aha_judge(row)
```
## Metadata and Variants
Each loader emits one or more variants. For each variant, Eval Protocol stores metadata on every row under `row.input_metadata.dataset_info`:
* `data_loader_type`: loader class (e.g., `DynamicDataLoader`)
* `data_loader_variant_id`: callable name or inline id
* `data_loader_variant_description`: docstring/description
* `data_loader_num_rows`: original count before preprocessing
* `data_loader_num_rows_after_preprocessing`: final count
This enables clear tracking of which inputs produced which results in the UI.
## Example with an Adapter
```python theme={null}
from eval_protocol import evaluation_test, aha_judge, DynamicDataLoader, SingleTurnRolloutProcessor
from eval_protocol.adapters.langfuse import create_langfuse_adapter
def langfuse_data_generator():
adapter = create_langfuse_adapter()
return adapter.get_evaluation_rows(limit=50, sample_size=10)
@evaluation_test(
data_loaders=DynamicDataLoader(generators=[langfuse_data_generator]),
rollout_processor=SingleTurnRolloutProcessor(),
)
async def test_llm_judge(row: EvaluationRow) -> EvaluationRow:
return await aha_judge(row)
```
## API Reference
### DynamicDataLoader
```python theme={null}
class DynamicDataLoader(EvaluationDataLoader):
generators: Sequence[Callable[[], list[EvaluationRow]]]
```
### InlineDataLoader
```python theme={null}
class InlineDataLoader(EvaluationDataLoader):
rows: list[EvaluationRow] | None
messages: Sequence[list[Message]] | None
id: str
description: str | None
```
### EvaluationDataLoader
```python theme={null}
class EvaluationDataLoader(ABC):
preprocess_fn: Callable[[list[EvaluationRow]], list[EvaluationRow]] | None
def variants(self) -> Sequence[DataLoaderVariant]: ...
def load(self) -> list[DataLoaderResult]: ...
```
## Source Code
See the Python source for full details: [eval\_protocol/data\_loader/models.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/data_loader/models.py)
# @evaluation_test
Source: https://evalprotocol.io/reference/evaluation-test
Create pytest-based evaluation tests for AI model evaluation with support for pointwise, groupwise, and all modes
The [`@evaluation_test`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/evaluation_test.py) decorator is the core component for creating pytest-based evaluation tests in the Evaluation Protocol. It enables you to evaluate AI models by running rollouts and applying evaluation criteria to measure performance.
## Key Concepts
Before diving into the API, it's important to understand the terminology used in the Evaluation Protocol:
* **Invocation**: A single execution of a test function that can generate 1 or more experiments
* **Experiment**: One per unique combination of input parameters (e.g., `completion_params`). `num_runs` creates multiple runs within the same experiment, not multiple experiments.
* **Run**: A group of rollouts (multiple run IDs if `num_runs > 1`)
* **Rollout**: The execution/process that produces a trajectory
* **Trajectory**: The result produced by a rollout — a list of OpenAI Chat Completion messages
* **Row**: Both input and output of an evaluation (e.g., a task within a dataset)
* **Dataset**: A collection of rows (List\[EvaluationRow])
* **Eval**: A rubric implemented in the test function body that produces a score from 0 to 1
Each of these entities has a unique ID for easy grouping and identification.
## Basic Usage
```python theme={null}
import pytest
from typing import List
from eval_protocol.pytest import evaluation_test
from eval_protocol.models import EvaluationRow
@pytest.mark.parametrize(
"completion_params",
[
{"model": "openai/gpt-4o", "temperature": 0.1},
{"model": "openai/gpt-3.5-turbo", "temperature": 0.1},
],
)
@evaluation_test(
input_dataset=["path/to/dataset.jsonl"],
passed_threshold=0.8,
mode="all"
)
def test_math_reasoning(rows: List[EvaluationRow]) -> List[EvaluationRow]:
"""Evaluate mathematical reasoning capabilities."""
for row in rows:
# Your evaluation logic here
score = evaluate_math_reasoning(row.messages)
row.evaluation_result.score = score
return rows
```
## Parameters
No single parameter is strictly required. Provide `completion_params` whenever your rollout processor performs model calls (e.g., `SingleTurnRolloutProcessor`).
Generation parameters for the rollout. Recommended to set via `@pytest.mark.parametrize("completion_params", [...])` together with `@evaluation_test`. The required fields depend on the rollout processor used:
**For SingleTurnRolloutProcessor and AgentRolloutProcessor:**
* Must include a `model` field using a LiteLLM-compatible provider route (e.g., `openai/gpt-4o`, `anthropic/claude-3-sonnet`, `fireworks_ai/*`)
* Optional: `temperature`, `max_tokens`, `extra_body`, etc.
* See the LiteLLM providers list for supported prefixes and models: [https://docs.litellm.ai/docs/providers](https://docs.litellm.ai/docs/providers)
**For PydanticAgentRolloutProcessor:**
* Must include `model` field (the canonical way to pass model names to LLM clients)
* Optional `provider` field (defaults to "openai" if not specified)
* Example: `{"model": "accounts/fireworks/models/kimi-k2-instruct", "provider": "fireworks"}`
* The agent factory uses the `model` field to create the appropriate Pydantic AI model
**For MCPGymRolloutProcessor:**
* Must include a `model` field using a LiteLLM-compatible provider route
* Used to create the policy for environment interaction
**For NoOpRolloutProcessor:**
* Can be any value (not used for actual model calls)
* Often set to `{"model": "not-used-offline"}` for clarity
Data loaders to produce evaluation rows. Preferred for reusable, parameterized inputs. Each loader may emit multiple variants; rows inherit metadata describing the loader, variant ID, and preprocessing state. Cannot be combined with `input_dataset`, `input_messages`, or `input_rows`.
See [Data Loader](/reference/data-loader) for details.
Messages to send to the model. Useful when you don't have a dataset but can hard-code messages. Will be passed as "input\_dataset" to the test function.
Paths to JSONL datasets that will be loaded using `load_jsonl()`. Each path can be either a local file path or an HTTP/HTTPS URL. Provide a `dataset_adapter` to convert the raw JSONL data to EvaluationRows.
**Behavior:**
* Files are loaded using `load_jsonl()` which reads JSONL format (one JSON object per line)
* **Supports both local files and HTTP URLs**: Local file paths and HTTP/HTTPS URLs are both supported
* **Robust parsing**: Automatically skips blank or whitespace-only lines to handle trailing newlines gracefully
* **Error handling**: Provides detailed error messages including line numbers and row IDs when JSON parsing fails
* **Timeout support**: HTTP requests have a 30-second timeout
* When multiple paths are provided and `combine_datasets=True` (default), files are concatenated into one dataset
* When `combine_datasets=False`, each path is parameterized into separate test invocations
* Raw JSONL data is passed to the `dataset_adapter` function for conversion to `EvaluationRow` format
**Supported formats:**
* Local files: `"path/to/dataset.jsonl"`
* HTTP URLs: `"http://example.com/dataset.jsonl"`
* HTTPS URLs: `"https://example.com/dataset.jsonl"`
**Example:**
```python theme={null}
import pytest
@pytest.mark.parametrize("completion_params", [{"model": "gpt-4"}])
@evaluation_test(
input_dataset=[
"path/to/local_dataset.jsonl",
"https://example.com/remote_dataset.jsonl"
],
dataset_adapter=my_adapter,
)
```
Pre-constructed EvaluationRow objects to use directly. Useful when you already have messages and/or metadata prepared. Will be passed as "input\_dataset" to the test function.
Note: cannot be combined with `data_loaders`.
Function to convert input dataset to a list of EvaluationRows. Defaults to `default_dataset_adapter`.
Function used to perform the rollout. Defaults to `NoOpRolloutProcessor()`.
Additional keyword arguments for the evaluation function.
Additional keyword arguments for the rollout processor.
How to aggregate scores across runs. One of: "mean", "max", "min", "bootstrap". Defaults to "mean".
Notes:
* With "mean", a 95% CI and standard error are computed for valid scores.
* With "bootstrap", a bootstrap mean score is computed (no CI output).
Optional preprocessing function applied to rows before rollout. Use this to expand multi-turn conversations (e.g., `multi_turn_assistant_to_ground_truth`) or filter/transform rows.
Note: when using `data_loaders`, pass `preprocess_fn` to the loader itself (e.g., `DynamicDataLoader(preprocess_fn=...)`). When `data_loaders` is provided, the decorator-level `preprocess_fn` is not applied to avoid double-processing.
Threshold configuration for test success. Can be a float or EvaluationThreshold object. Success rate must be above `success`, and if set, standard error must be below `standard_error`.
Number of times to repeat the rollout and evaluations. Defaults to 1.
Evaluate only rows whose `row.input_metadata.row_id` is in this list.
Limit dataset to the first N rows.
Path to MCP config file that follows MCPMultiClientConfiguration schema.
Maximum number of concurrent rollouts to run in parallel. Defaults to 8.
Maximum number of concurrent evaluations to run in parallel. Defaults to 64.
Path to the MCP server script to run. Defaults to "examples/tau2\_mcp/server.py".
Number of rollout steps to execute. Defaults to 30.
Evaluation mode. "pointwise" (default) applies test function to each row individually. "groupwise" applies test function to a group of rollout results from the same original row (for use cases such as DPO/GRPO). "all" applies test function to the whole dataset.
Whether to combine multiple datasets. Defaults to True.
DatasetLogger to use for logging. If not provided, a default logger will be used.
Configuration for exception handling and backoff retry logic. If not provided, a default configuration will be used with common retryable exceptions. See the ExceptionHandlerConfig section below for detailed configuration options.
## ExceptionHandlerConfig
The `ExceptionHandlerConfig` parameter allows you to customize exception handling and retry logic for your evaluation tests. This configuration is defined in [`eval_protocol/pytest/exception_config.ExceptionHandlerConfig`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/exception_config.py).
### Key Features
* **Retryable Exceptions**: Configure which exceptions should trigger retry attempts
* **Backoff Strategies**: Choose between exponential or constant backoff with configurable delays
* **Environment Variable Overrides**: Automatically respect `EP_MAX_RETRY` and `EP_FAIL_ON_MAX_RETRY` settings
* **Custom Giveup Logic**: Define custom conditions for when to stop retrying
### Configuration Classes
#### ExceptionHandlerConfig
The main configuration class that controls exception handling behavior:
```python theme={null}
@dataclass
class ExceptionHandlerConfig:
# Exceptions that should be retried using backoff
retryable_exceptions: Set[Type[Exception]] = DEFAULT_RETRYABLE_EXCEPTIONS
# Backoff configuration
backoff_config: BackoffConfig = BackoffConfig()
```
#### BackoffConfig
Controls the retry backoff behavior:
```python theme={null}
@dataclass
class BackoffConfig:
strategy: str = "expo" # "expo" or "constant"
base_delay: float = 1.0 # Base delay in seconds
max_delay: float = 60.0 # Maximum delay in seconds
max_tries: int = 3 # Maximum number of retry attempts
jitter: Union[None, Callable] = None # Jitter function for randomization
factor: float = 2.0 # Factor for exponential backoff
raise_on_giveup: bool = True # Whether to raise exception when giving up
giveup_func: Callable[[Exception], bool] = lambda e: False # Custom giveup logic
```
### Default Configuration
By default, the following exceptions are considered retryable:
* **Standard library exceptions**: `ConnectionError`, `TimeoutError`, `OSError`
* **Requests library exceptions**: `requests.exceptions.ConnectionError`, `requests.exceptions.Timeout`, `requests.exceptions.HTTPError`, `requests.exceptions.RequestException`
* **HTTPX library exceptions**: `httpx.ConnectError`, `httpx.TimeoutException`, `httpx.NetworkError`, `httpx.RemoteProtocolError`
### Backoff Strategies
#### Exponential Backoff (Default)
* Starts with `base_delay` and multiplies by `factor` each retry
* Good for transient failures that may resolve quickly
* Example: 1s → 2s → 4s → 8s → 16s (capped at `max_delay`)
#### Constant Backoff
* Uses the same delay (`base_delay`) for all retries
* Good for predictable, consistent retry timing
* Example: 2s → 2s → 2s → 2s
### Environment Variable Integration
The configuration automatically respects these environment variables:
* `EP_MAX_RETRY`: Overrides `max_tries` in BackoffConfig
* `EP_FAIL_ON_MAX_RETRY`: Controls `raise_on_giveup` behavior
### Example Usage
#### Basic Custom Configuration
```python theme={null}
from eval_protocol.pytest.exception_config import ExceptionHandlerConfig, BackoffConfig
# Custom exception handling configuration
custom_config = ExceptionHandlerConfig(
backoff_config=BackoffConfig(
strategy="expo",
base_delay=2.0,
max_delay=120.0,
max_tries=5,
jitter=None
)
)
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
exception_handler_config=custom_config
)
def test_with_custom_retry_logic(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# Your evaluation logic here
return rows
```
#### Aggressive Retry Strategy
```python theme={null}
# Aggressive retry for unreliable networks
aggressive_config = ExceptionHandlerConfig(
backoff_config=BackoffConfig(
strategy="expo",
base_delay=0.5, # Start with 0.5s delay
max_delay=30.0, # Cap at 30s
max_tries=10, # Try up to 10 times
jitter=None # No jitter for predictable timing
)
)
```
#### Conservative Retry Strategy
```python theme={null}
# Conservative retry for stable networks
conservative_config = ExceptionHandlerConfig(
backoff_config=BackoffConfig(
strategy="constant",
base_delay=5.0, # 5 second constant delay
max_tries=3, # Only 3 attempts
jitter=None
)
)
```
#### Custom Exception Handling
```python theme={null}
from typing import Set, Type
# Only retry on specific exceptions
custom_exceptions: Set[Type[Exception]] = {
ConnectionError,
TimeoutError,
# Add your custom exceptions here
}
custom_config = ExceptionHandlerConfig(
retryable_exceptions=custom_exceptions,
backoff_config=BackoffConfig(
strategy="expo",
base_delay=1.0,
max_tries=3
)
)
```
## Evaluation Modes
### Pointwise Mode (Default)
In pointwise mode, your test function processes each row individually, enabling pipelined evaluation:
```python theme={null}
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
input_dataset=["dataset.jsonl"],
mode="pointwise"
)
def test_pointwise_evaluation(row: EvaluationRow) -> EvaluationRow:
"""Process each row individually."""
# Evaluate single row
score = evaluate_single_row(row)
row.evaluation_result.score = score
return row
```
**Requirements:**
* Function must have a parameter named `row` of type `EvaluationRow`
* Function must return `EvaluationRow`
### Groupwise Mode
In groupwise mode, your test function processes groups of rollout results from the same original row, useful for comparing different models or parameters:
```python theme={null}
@evaluation_test(
completion_params=[
{"model": "gpt-4", "temperature": 0.1},
{"model": "gpt-3.5-turbo", "temperature": 0.1},
],
input_dataset=["dataset.jsonl"],
mode="groupwise"
)
def test_groupwise_evaluation(rows: List[EvaluationRow]) -> List[EvaluationRow]:
"""Process groups of rows from the same original input."""
# Compare results across different models/parameters
scores = compare_model_outputs(rows)
for i, row in enumerate(rows):
row.evaluation_result.score = scores[i]
return rows
```
**Requirements:**
* Function must have a parameter named `rows` of type `List[EvaluationRow]`
* Function must return `List[EvaluationRow]`
* Must provide at least 2 completion parameters
### All Mode
In all mode, your test function receives the entire dataset and processes all rows together:
```python theme={null}
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
input_dataset=["dataset.jsonl"],
mode="all"
)
def test_all_evaluation(rows: List[EvaluationRow]) -> List[EvaluationRow]:
"""Process all rows together."""
# Access to full dataset for cross-row analysis
for row in rows:
# Evaluate each row
score = evaluate_single_row(row)
row.evaluation_result.score = score
return rows
```
**Requirements:**
* Function must have a parameter named `rows` of type `List[EvaluationRow]`
* Function must return `List[EvaluationRow]`
## Threshold Configuration
You can set thresholds for test success using the `passed_threshold` parameter:
```python theme={null}
# Simple threshold (just success rate)
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
passed_threshold=0.8
)
# Advanced threshold with standard error
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
passed_threshold={
"success": 0.8,
"standard_error": 0.05
}
)
# Using EvaluationThreshold object
from eval_protocol.models import EvaluationThreshold
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
passed_threshold=EvaluationThreshold(success=0.8, standard_error=0.05)
)
```
## Multiple Runs and Aggregation
Set `num_runs > 1` to run multiple evaluations and aggregate results:
```python theme={null}
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
input_dataset=["dataset.jsonl"],
num_runs=5,
aggregation_method="mean"
)
def test_with_multiple_runs(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# This function will be called 5 times
# Results will be aggregated using the mean
return rows
```
## Environment Variables
The decorator supports several environment variables for configuration:
* `EP_MAX_DATASET_ROWS`: Override `max_dataset_rows` parameter. Applies to both datasets and `input_messages` (slices to first N rows).
* `EP_NUM_RUNS`: Override the number of runs for evaluation\_test.
* `EP_MAX_CONCURRENT_ROLLOUTS`: Override the maximum number of concurrent rollouts.
* `EP_INPUT_PARAMS_JSON`: JSON object deep-merged into `completion_params`. Example: `{"temperature":0,"extra_body":{"reasoning":{"effort":"low"}}}`.
* `EP_COMPLETION_PARAMS`: JSON array that fully replaces `completion_params` (e.g., `[{"model":"openai/gpt-4o","temperature":0}]`).
* `EP_PASSED_THRESHOLD`: Float for success threshold (e.g., `0.8`). Equivalent to `passed_threshold=0.8`.
* `EP_JSONL_PATH`: When set, automatically constructs a `DynamicDataLoader` that loads rows from the given JSONL path.
* `EP_INVOCATION_ID`: Override the auto-generated invocation ID.
* `EP_PRINT_SUMMARY`: Set to "1" to print a one-line evaluation summary to stdout.
* `EP_SUMMARY_JSON`: File or directory path to write a JSON summary artifact. See "Summary artifacts" for naming behavior.
* Retry-related environment variables are documented in the [Retries and failure policy](#retries-and-failure-policy) section.
## Return Values
Your test function must return the appropriate type based on the mode:
* **Pointwise mode**: `EvaluationRow`
* **Groupwise mode**: `List[EvaluationRow]`
* **All mode**: `List[EvaluationRow]`
Each returned row should have:
* `evaluation_result.score`: A float between 0 and 1
* Optional `evaluation_result.metrics`: Additional metric scores
* Optional `execution_metadata.cost_metrics`: Automatically populated when token usage and model info are present (input, output, total costs).
## Dataset loading and input formats
* **Data loaders (`data_loaders`)**: Preferred for reusable and parameterized inputs. Accepts one or more `EvaluationDataLoader` instances (e.g., `DynamicDataLoader`, `InlineDataLoader`). Each loader can emit multiple variants and apply `preprocess_fn` internally. Cannot be combined with `input_dataset`, `input_messages`, or `input_rows`.
* **Datasets (`input_dataset`)**: You can pass a single path or a list of paths to JSONL files. Files are loaded using `load_jsonl()` which supports both local files and HTTP/HTTPS URLs. The function reads JSONL format (one JSON object per line) with robust error handling, automatically skips blank lines, and provides detailed error messages with line numbers and row IDs. When a list is provided and `combine_datasets=True` (default), files are concatenated into one dataset; when `combine_datasets=False`, each path is parameterized into separate test invocations.
* **Input messages (`input_messages`)**: Accepts either a single row as `List[Message]` or many rows as `List[List[Message]]`. When `EP_MAX_DATASET_ROWS` is set, the list is sliced before parameterization.
* **Input rows (`input_rows`)**: Similar to input\_messages, when `EP_MAX_DATASET_ROWS` is set, the list is sliced before parameterization.
* **Dataset adapter (`dataset_adapter`)**: Receives raw JSONL rows (as loaded by `load_jsonl()`) and must return `List[EvaluationRow]`.
Important: Provide exactly one of `data_loaders`, `input_dataset`, `input_messages`, or `input_rows`. Supplying more than one will raise an error.
## Error Handling
The decorator handles errors gracefully:
* Failed rollouts are still evaluated (you can choose to give them a score of 0)
* Assertion errors are logged with status "finished"
* Other exceptions are logged with status "error"
* Summary generation failures don't cause test failures
* For retry behavior and configuration, see [ExceptionHandlerConfig](#exceptionhandlerconfig) and [Retries and failure policy](#retries-and-failure-policy).
## Row IDs and metadata
* Stable `row_id` values are generated for rows missing `row.input_metadata.row_id`, using a deterministic hash of row content. This ensures consistent IDs across processes and runs.
* `EvalMetadata` is created for each evaluation with: `name` (test function name), `description` (docstring), `num_runs`, `aggregation_method`, and threshold info. Its `status` transitions from "running" to "finished" or "error".
* `completion_params` used for a row are recorded in `row.input_metadata.completion_params`.
## Dataset combination and parameterization
* Parameter combinations are generated across `data_loaders`, `input_dataset`, `completion_params`, `input_messages`, `input_rows`, and `evaluation_test_kwargs`.
* Pytest parameter names (in order when present): `dataset_path`, `completion_params`, `input_messages`, `input_rows`, `data_loaders`, `evaluation_test_kwargs`.
* Set `combine_datasets=False` to parameterize each dataset path separately. With `True` (default), multiple paths are combined into a single logical dataset per invocation.
### Recommended parameterization style
Use `@pytest.mark.parametrize("completion_params", [...])` with `@evaluation_test`. The decorator integrates with pytest's parameterization and will align the function signature accordingly.
```python theme={null}
import pytest
from typing import List
from eval_protocol.models import EvaluationRow, Message
from eval_protocol.pytest import evaluation_test, SingleTurnRolloutProcessor
@pytest.mark.parametrize(
"completion_params",
[
{"model": "openai/gpt-4o", "temperature": 0.1},
{"model": "openai/gpt-4o-mini", "temperature": 0},
],
)
@evaluation_test(
input_messages=[
[
[Message(role="user", content="What is the capital of France?")]
]
],
rollout_processor=SingleTurnRolloutProcessor(),
mode="all",
)
def test_parametrized_input_messages(rows: List[EvaluationRow]) -> List[EvaluationRow]:
return rows
```
You can use the same pattern with datasets:
```python theme={null}
import pytest
from typing import List
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import evaluation_test
@pytest.mark.parametrize("completion_params", [{"model": "openai/gpt-4o"}])
@evaluation_test(
input_dataset=["dataset.jsonl"],
mode="pointwise",
)
def test_parametrized_dataset(row: EvaluationRow) -> EvaluationRow:
return row
```
Note: Passing `completion_params` inside `@evaluation_test(...)` still works; the decorator will auto-generate `pytest.mark.parametrize` under the hood. However, explicitly using `@pytest.mark.parametrize` for `completion_params` is the recommended style for clarity and IDE tooling.
## Summary artifacts
When `EP_SUMMARY_JSON` is set:
* If a directory or a non-`.json` path is provided, a file is written inside with the base name: `"{suite}__{model}__{mode}__runs{num_runs}.json"`, where `suite` is the test function name and `model` is a sanitized slug.
* If a file path is provided, it writes that file. If an "effort" tag is detected in `completion_params` (e.g., via `extra_body.reasoning.effort` or `extra_body.reasoning_effort`), a variant suffixed with `__effort-{effort}` is written instead.
* The summary includes: `suite`, `model`, `agg_score`, `num_runs`, `rows`, and a `timestamp`. When `aggregation_method` is `"mean"`, it also includes `standard_error` and 95% CI (`agg_ci_low`, `agg_ci_high`).
* When per-row metric scores are present, `metrics_agg` contains per-metric mean and, when available, CI bounds.
* In `groupwise` mode, summaries are generated per `completion_params` group.
## Retries and failure policy
* Rollouts are retried up to `EP_MAX_RETRY` times using the `rollout_processor_with_retry` wrapper.
* Permanent failures are, by default, raised immediately to fail the test. Override with `EP_FAIL_ON_MAX_RETRY=false` to continue and include errored rows (you can score them as 0 in your evaluation).
* Exception handling and retry logic can be customized via `exception_handler_config`.
### Environment Variables
The following environment variables control retry behavior:
* `EP_MAX_RETRY`: Maximum number of retry attempts (default: 0, meaning no retries)
* `EP_FAIL_ON_MAX_RETRY`: Whether to fail the test after max retries (default: "true")
### Retry Implementation Details
The retry logic is implemented in the `rollout_processor_with_retry` function which:
* Wraps the rollout processor with configurable backoff retry
* Handles both retryable and non-retryable exceptions
* Uses the Python `backoff` library for exponential/constant backoff strategies
* Processes rows concurrently while handling retries transparently
* Logs all results (success or failure) through the configured logger
### Custom Retry Configuration
For advanced retry logic, you can provide a custom `ExceptionHandlerConfig`:
```python theme={null}
from eval_protocol.pytest.exception_config import ExceptionHandlerConfig, BackoffConfig
# Aggressive retry strategy for unreliable networks
aggressive_retry = ExceptionHandlerConfig(
backoff_config=BackoffConfig(
strategy="expo",
base_delay=0.5, # Start with 0.5s delay
max_delay=30.0, # Cap at 30s
max_tries=10, # Try up to 10 times
jitter=None # No jitter for predictable timing
)
)
@evaluation_test(
completion_params=[{"model": "gpt-4"}],
exception_handler_config=aggressive_retry
)
def test_with_aggressive_retries(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# Your evaluation logic here
return rows
```
## [Rollout processors](/reference/rollout-processors)
* A rollout processor turns input rows into completed rows (e.g., by calling a model). The decorator passes a `RolloutProcessorConfig` containing `completion_params`, `mcp_config_path`, `server_script_path`, `max_concurrent_rollouts`, and `steps`.
* Built-ins include:
* `NoOpRolloutProcessor()`: passes rows through unchanged (useful for offline evaluation of pre-generated outputs).
* `SingleTurnRolloutProcessor()`: performs a single chat completion via LiteLLM and appends the assistant message.
* `AgentRolloutProcessor()`: runs multi-turn agent loops with MCP tool calling.
* `PydanticAgentRolloutProcessor()`: runs Pydantic AI agents with structured tool calling.
* `MCPGymRolloutProcessor()`: runs interactive environments via MCP servers.
* All processors are wrapped with `rollout_processor_with_retry` for automatic retry handling.
Note: With `MCPGymRolloutProcessor`, repeated runs (`num_runs > 1`) are executed sequentially to avoid port conflicts; other processors run runs in parallel with concurrency controlled by the shared semaphore.
## RolloutProcessorConfig
The `RolloutProcessorConfig` is passed to all rollout processors and contains the configuration needed to execute rollouts. It's defined in [`eval_protocol/pytest/types.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/types.py).
### Configuration Fields
Model and generation parameters for the rollout. The structure and required fields depend on the rollout processor:
**SingleTurnRolloutProcessor & AgentRolloutProcessor:**
* Must include `model` field with LiteLLM-compatible provider route
* Supports standard LiteLLM parameters: `temperature`, `max_tokens`, `extra_body`, etc.
**PydanticAgentRolloutProcessor:**
* Must include `model` field (canonical way to pass model names)
* Optional `provider` field (defaults to "openai" if not specified)
* Used to create Pydantic AI model instances via the agent factory
**MCPGymRolloutProcessor:**
* Must include `model` field for environment policy creation
* Additional parameters passed to the policy constructor
**NoOpRolloutProcessor:**
* Can contain any values (not used for actual model calls)
* Often set to placeholder values for clarity
Path to an MCP client configuration file that follows the MCPMultiClientConfiguration schema. Used by agent and tool-based rollout processors to enumerate available tools and capabilities.
Shared semaphore for unified concurrency control across all rollout processors. Controls the maximum number of concurrent rollouts that can run simultaneously.
Path to an MCP server script to run. Used by gym-like processors (e.g., `MCPGymRolloutProcessor`) to launch interactive environments. Defaults to `None`.
Maximum number of rollout steps to execute. Used by multi-turn processors to limit the length of agent conversations. Defaults to `30`.
Logger to use for capturing mid-rollout logs and debugging information. Defaults to `default_logger`.
Additional keyword arguments specific to the rollout processor. This is where processor-specific configuration is passed, such as:
* `usage_limits` for Pydantic AI agents
* `agent` for pre-configured agents
* Custom tool configurations
* Environment-specific settings
Configuration for exception handling and backoff retry logic. If not provided, a default configuration will be used with common retryable exceptions. See the [ExceptionHandlerConfig](#exceptionhandlerconfig) section for detailed configuration options.
### Usage in Custom Rollout Processors
When implementing custom rollout processors, you can access these configuration values:
```python theme={null}
from eval_protocol.pytest.rollout_processor import RolloutProcessor
from eval_protocol.pytest.types import RolloutProcessorConfig
from eval_protocol.models import EvaluationRow
import asyncio
class CustomRolloutProcessor(RolloutProcessor):
def __call__(self, rows: list[EvaluationRow], config: RolloutProcessorConfig) -> list[asyncio.Task[EvaluationRow]]:
# Access model configuration
model = config.completion_params.get("model")
temperature = config.completion_params.get("temperature", 0.0)
# Access concurrency control
semaphore = config.semaphore
# Access custom configuration
custom_setting = config.kwargs.get("custom_setting", "default_value")
# Access MCP configuration
mcp_config_path = config.mcp_config_path
# Access step limits
max_steps = config.steps
# Access logger
logger = config.logger
# Your rollout logic here
async def process_row(row: EvaluationRow) -> EvaluationRow:
async with semaphore:
# Process the row
return row
return [asyncio.create_task(process_row(row)) for row in rows]
```
### Environment Variable Integration
Several configuration values can be overridden at runtime using environment variables:
* `EP_MAX_CONCURRENT_ROLLOUTS`: Overrides the semaphore limit
* `EP_NUM_RUNS`: Affects the number of runs for evaluation\_test
* `EP_MAX_RETRY`: Controls retry behavior via exception\_handler\_config
* `EP_FAIL_ON_MAX_RETRY`: Controls failure behavior after max retries
### Processor-Specific Configuration
Different rollout processors use the `kwargs` field for their specific needs:
#### AgentRolloutProcessor
```python theme={null}
config = RolloutProcessorConfig(
completion_params={
"model": "openai/gpt-4",
"temperature": 0.1,
"max_tokens": 1000
},
mcp_config_path="./mcp_config.json",
semaphore=asyncio.Semaphore(8),
kwargs={
"custom_tool_config": {...},
"agent_instructions": "You are a helpful assistant"
}
)
```
#### PydanticAgentRolloutProcessor
```python theme={null}
config = RolloutProcessorConfig(
completion_params={
"model": "accounts/fireworks/models/kimi-k2-instruct",
"provider": "fireworks" # Optional: defaults to "openai"
},
mcp_config_path="./mcp_config.json",
semaphore=asyncio.Semaphore(8),
kwargs={
"agent": my_pydantic_agent,
"usage_limits": UsageLimits(max_tokens=1000)
}
)
```
#### MCPGymRolloutProcessor
```python theme={null}
config = RolloutProcessorConfig(
completion_params={
"model": "openai/gpt-4",
"temperature": 0.0
},
mcp_config_path="./mcp_config.json",
semaphore=asyncio.Semaphore(8),
server_script_path="./gym_server.py",
kwargs={
"environment_config": {...},
"gym_timeout": 300
}
)
```
## Direct invocation (dual-mode)
Decorated functions can be called directly in addition to running under pytest:
* Pointwise mode: `await test_fn(row)` or `await test_fn(row=...)`
* Groupwise mode: `await test_fn(rows)` or `await test_fn(rows=[...])`
* All mode: `await test_fn(rows)` or `await test_fn(rows=[...])`
When using `data_loaders`, direct invocation works the same way; the decorator resolves loaders into rows before calling your function. If a decorated function is called directly with `row`/`rows` arguments, those are used as-is.
## Examples
### Basic Math Evaluation (Pointwise Mode)
```python theme={null}
import pytest
@pytest.mark.parametrize("completion_params", [{"model": "gpt-4"}])
@evaluation_test(
input_messages=[
[Message(role="user", content="What is 2 + 2?")]
],
passed_threshold=0.9,
mode="pointwise"
)
def test_basic_math(row: EvaluationRow) -> EvaluationRow:
# Simple correctness check
response = row.messages[-1].content
if "4" in response:
row.evaluation_result.score = 1.0
else:
row.evaluation_result.score = 0.0
return row
```
### Multi-Model Comparison (All Mode)
```python theme={null}
import pytest
@pytest.mark.parametrize(
"completion_params",
[
{"model": "gpt-4", "temperature": 0.1},
{"model": "gpt-3.5-turbo", "temperature": 0.1},
{"model": "claude-3-sonnet", "temperature": 0.1},
],
)
@evaluation_test(
input_dataset=["reasoning_tasks.jsonl"],
passed_threshold=0.7,
num_runs=3,
mode="all"
)
def test_reasoning_capabilities(rows: List[EvaluationRow]) -> List[EvaluationRow]:
for row in rows:
# Complex evaluation logic
score = evaluate_reasoning_quality(row.messages)
row.evaluation_result.score = score
# Add additional metrics
row.evaluation_result.metrics = {
"clarity": evaluate_clarity(row.messages),
"correctness": evaluate_correctness(row.messages)
}
return rows
```
### Groupwise Evaluation for Model Comparison
```python theme={null}
import pytest
@pytest.mark.parametrize(
"completion_params",
[
{"model": "gpt-4", "temperature": 0.1},
{"model": "gpt-3.5-turbo", "temperature": 0.1},
],
)
@evaluation_test(
input_dataset=["comparison_tasks.jsonl"],
mode="groupwise"
)
def test_model_comparison(rows: List[EvaluationRow]) -> List[EvaluationRow]:
"""Compare outputs from different models on the same input."""
# Group rows by their original input
for row in rows:
# Evaluate relative to other models or absolute quality
score = evaluate_model_output(row.messages, row.input_metadata.completion_params)
row.evaluation_result.score = score
return rows
```
### Pointwise Evaluation with Custom Dataset
```python theme={null}
def custom_dataset_adapter(data: List[Dict[str, Any]]) -> List[EvaluationRow]:
"""Convert custom format to EvaluationRows."""
rows = []
for item in data:
messages = [
Message(role="user", content=item["question"]),
Message(role="assistant", content=item["answer"])
]
row = EvaluationRow(messages=messages)
rows.append(row)
return rows
import pytest
@pytest.mark.parametrize("completion_params", [{"model": "gpt-4"}])
@evaluation_test(
input_dataset=["custom_format.jsonl"],
dataset_adapter=custom_dataset_adapter,
mode="pointwise"
)
def test_custom_format(row: EvaluationRow) -> EvaluationRow:
# Process individual row
score = evaluate_custom_metric(row.messages)
row.evaluation_result.score = score
return row
```
### Complete runnable example (offline, no model calls)
This example evaluates pre-generated assistant messages using the no-op rollout processor.
```python theme={null}
from typing import Any, Dict, List
from eval_protocol.models import EvaluationRow, Message, EvaluateResult
from eval_protocol.pytest.evaluation_test import evaluation_test
from eval_protocol.pytest.default_no_op_rollout_processor import NoOpRolloutProcessor
def adapter(json_rows: List[Dict[str, Any]]) -> List[EvaluationRow]:
rows: List[EvaluationRow] = []
for r in json_rows:
# Expect fields: question, model_answer, ground_truth
rows.append(
EvaluationRow(
messages=[
Message(role="user", content=str(r["question"])) ,
Message(role="assistant", content=str(r["model_answer"]))
],
ground_truth=str(r.get("ground_truth", ""))
)
)
return rows
@evaluation_test(
input_dataset=["offline_answers.jsonl"],
dataset_adapter=adapter,
completion_params=[{"model": "not-used-offline"}],
rollout_processor=NoOpRolloutProcessor(),
mode="all"
)
def test_offline_eval(rows: List[EvaluationRow]) -> List[EvaluationRow]:
for row in rows:
pred = (row.get_assistant_messages()[-1].content or "").strip()
gt = (row.ground_truth or "").strip()
score = 1.0 if pred == gt else 0.0
row.evaluation_result = EvaluateResult(score=score, reason="exact match")
return rows
```
### Complete runnable example (single-turn online via LiteLLM)
### Using data\_loaders with DynamicDataLoader
```python theme={null}
from eval_protocol import evaluation_test, DynamicDataLoader, SingleTurnRolloutProcessor
from eval_protocol.adapters.langfuse import create_langfuse_adapter
def langfuse_data_generator():
adapter = create_langfuse_adapter()
return adapter.get_evaluation_rows(limit=20, sample_size=5)
import pytest
@pytest.mark.parametrize("completion_params", [{"model": "openai/gpt-4o"}])
@evaluation_test(
data_loaders=DynamicDataLoader(generators=[langfuse_data_generator]),
rollout_processor=SingleTurnRolloutProcessor(),
mode="pointwise",
)
def test_with_loader(row: EvaluationRow) -> EvaluationRow:
# Evaluate row here
return row
```
Requires `pip install litellm` and provider credentials configured.
```python theme={null}
import pytest
from typing import List
from eval_protocol.models import EvaluationRow, Message, EvaluateResult
from eval_protocol.pytest.evaluation_test import evaluation_test
from eval_protocol.pytest.default_single_turn_rollout_process import SingleTurnRolloutProcessor
@pytest.mark.parametrize("completion_params", [{"model": "openai/gpt-4o-mini", "temperature": 0}])
@evaluation_test(
input_messages=[[Message(role="user", content="What is 2 + 2?")]],
rollout_processor=SingleTurnRolloutProcessor(),
passed_threshold=0.8,
mode="pointwise"
)
def test_online_math(row: EvaluationRow) -> EvaluationRow:
answer = (row.get_assistant_messages()[-1].content or "").strip()
score = 1.0 if "4" in answer else 0.0
row.evaluation_result = EvaluateResult(score=score, reason="contains 4")
return row
```
## Integration with pytest
The decorator automatically creates pytest-compatible test functions:
```bash theme={null}
# Run all evaluation tests
pytest test_file.py
# Run specific test
pytest test_file.py::test_math_reasoning
# Run with specific parameters
pytest test_file.py::test_math_reasoning[dataset_path0-completion_params0]
```
Tip: Prefer explicit `@pytest.mark.parametrize("completion_params", [...])` together with `@evaluation_test` for clearer parameter control and readable test IDs.
## Programmatic Usage
Decorated functions can be called directly in addition to running under pytest. See [Direct invocation (dual-mode)](#direct-invocation-dual-mode) for patterns by mode.
## Best Practices
1. **Clear Documentation**: Always include docstrings explaining what your evaluation measures
2. **Error Handling**: Handle edge cases gracefully and provide meaningful scores for failed rollouts
3. **Metric Design**: Design metrics that are objective and reproducible
4. **Reason**: Include a `reason` field in the `evaluation_result` to explain the score
5. **Threshold Setting**: Set realistic thresholds based on your use case
6. **Multiple Runs**: Use `num_runs > 1` for more reliable results when possible
7. **Resource Management**: Consider `max_concurrent_rollouts` and `max_concurrent_evaluations` based on your system capabilities
8. **Mode Selection**: Choose the appropriate mode for your evaluation needs:
* Use "pointwise" for simple per-row evaluation
* Use "groupwise" for comparing multiple models/parameters on the same inputs
* Use "all" for batch processing with cross-row analysis
## Troubleshooting
### Common Issues
* **"No combinations of parameters found"**: Ensure you provide both `completion_params` and either `input_dataset`, `input_messages`, or `input_rows`
* **"No model provided"**: Check that your `CompletionParams` includes a `model` field
* **Signature validation errors**: Ensure your function signature matches the mode requirements:
* Pointwise mode: `def func(row: EvaluationRow) -> EvaluationRow`
* Groupwise mode: `def func(rows: List[EvaluationRow]) -> List[EvaluationRow]`
* All mode: `def func(rows: List[EvaluationRow]) -> List[EvaluationRow]`
* **Return type errors**: Verify you're returning the correct type based on your mode
* **"In groupwise mode, you must provide at least 2 completion parameters"**: Groupwise mode requires multiple completion parameters to compare
### Debug Tips
* Set `EP_PRINT_SUMMARY=1` to see evaluation results in console
* Use `EP_SUMMARY_JSON` to save detailed results to a file
* Check the generated pytest parameterization for complex setups
* Use `max_dataset_rows` to limit dataset size during development
* Monitor `max_concurrent_rollouts` and `max_concurrent_evaluations` for performance tuning
* Set `EP_DEBUG_SERIALIZATION=1` to print compact per-row message previews (roles, lengths, tool call counts).
# Rollout Processors
Source: https://evalprotocol.io/reference/rollout-processors
Overview of built-in rollout processors, their configs, and when to use each
Rollout processors are classes that implement a common interface to turn input `EvaluationRow`s into completed rows (e.g., by calling a model once, running a tool-using agent loop, or interacting with an MCP "gym"). They all implement the same Python interface:
```python theme={null}
from typing import List
import asyncio
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest.types import RolloutProcessorConfig
class RolloutProcessor:
def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig) -> List[asyncio.Task[EvaluationRow]]:
... # return asyncio Tasks that resolve to completed rows
def cleanup(self) -> None:
... # optional; release external resources (servers, temp files)
```
The config object is defined in [`eval_protocol/pytest/types.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/types.py) as `RolloutProcessorConfig` and includes the most common knobs for evaluation runs. The interface lives in [`eval_protocol/pytest/rollout_processor.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/rollout_processor.py). The evaluation framework awaits these tasks with per-row retries and finally calls `cleanup()` for you.
## Config: RolloutProcessorConfig
* **completion\_params**: model and generation parameters (provider-agnostic via LiteLLM). Must include `model`.
* **mcp\_config\_path**: path to an MCP client configuration file (used by agent/tool processors).
* **server\_script\_path**: path to an MCP server script (used by gym-like processors).
* **max\_concurrent\_rollouts**: maximum number of rows processed in parallel (default 8).
* **steps**: maximum rollout steps for multi-turn processors (default 30).
* **logger**: `DatasetLogger` to capture mid-rollout logs.
* **kwargs**: extra, processor-specific options.
* **exception\_handler\_config**: controls automatic backoff/retry for rollout errors. See ExceptionHandlerConfig in the `@evaluation_test` reference.
Tip: You can override certain input parameters at runtime with the pytest plugin flags (see below), e.g., `--ep-reasoning-effort` or `--ep-input-param`.
## Built-in processors
### NoOpRolloutProcessor
* **What it does**: Pass-through. Returns tasks that immediately resolve to the same rows, so you can handle rollout yourself inside the evaluation function.
* **When to use**: You already have model outputs precomputed or you want to implement rollout logic in the test body.
* **Module**: [`eval_protocol/pytest/default_no_op_rollout_processor.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/default_no_op_rollout_processor.py)
Usage with `@evaluation_test`:
```python theme={null}
from eval_protocol.pytest import evaluation_test, NoOpRolloutProcessor
@evaluation_test(
completion_params=[{"model": "openai/gpt-4o-mini"}],
rollout_processor=NoOpRolloutProcessor(),
)
def my_eval(rows):
# rows are unchanged; compute scores here
return rows
```
### SingleTurnRolloutProcessor
* **What it does**: Issues a single LiteLLM `completion` per row and appends the assistant message (and any tool\_calls) to `row.messages`.
* **When to use**: Single-turn prompts, static QA, or benchmarks that only need the model's immediate reply.
* **Respects**: `completion_params` and forwards `reasoning_effort` under `extra_body` when present.
* **Module**: [`eval_protocol/pytest/default_single_turn_rollout_process.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/default_single_turn_rollout_process.py)
Usage:
```python theme={null}
from eval_protocol.pytest import evaluation_test, SingleTurnRolloutProcessor
@evaluation_test(
completion_params=[{
"model": "fireworks_ai/accounts/fireworks/models/gpt-oss-120b",
"temperature": 0.0,
"extra_body": {"reasoning_effort": "low"}, # forwarded to providers that support it
}],
rollout_processor=SingleTurnRolloutProcessor(),
)
def single_turn_eval(rows):
# each row now contains the assistant's reply; compute scores
return rows
```
### AgentRolloutProcessor
* **What it does**: Runs a simple multi-turn agent that can call MCP tools. The agent:
* Calls the model with current `messages` and available tools.
* Executes any returned tool calls in parallel.
* Appends tool results then calls the model again, until there are no more tool calls.
* **When to use**: Tool-augmented tasks, function-calling, or scenarios requiring iterative reasoning via tools.
* **Requires**: `mcp_config_path` to enumerate available tools via `MCPMultiClient`.
* **Honors**: `max_concurrent_rollouts` for dataset-level parallelism; tool calls within a single row are also executed in parallel.
* **Module**: [`eval_protocol/pytest/default_agent_rollout_processor.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/default_agent_rollout_processor.py)
Usage:
```python theme={null}
from eval_protocol.pytest import evaluation_test, AgentRolloutProcessor
@evaluation_test(
completion_params=[{"model": "openai/gpt-4o"}],
rollout_processor=AgentRolloutProcessor(),
mcp_config_path="./path/to/mcp.config.json",
max_concurrent_rollouts=8,
steps=30, # upper bound; the agent stops earlier if no tools are requested
)
def agent_eval(rows):
return rows
```
### PydanticAgentRolloutProcessor
* **What it does**: Runs Pydantic AI agents with automatic message format conversion between eval-protocol and Pydantic AI formats.
* **When to use**: **ONLY for Pydantic AI framework.** Multi-turn conversations, tool usage scenarios, and complex agent workflows.
* **Requires**: `agent_factory` parameter - a callable that creates a Pydantic AI `Agent` instance from `RolloutProcessorConfig`.
* **Module**: [`eval_protocol/pytest/default_pydantic_ai_rollout_processor.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/default_pydantic_ai_rollout_processor.py)
* **See also**: [Pydantic AI integration guide](/integrations/pydantic-ai) for detailed examples and agent factory patterns.
The processor automatically converts message formats and handles concurrency control:
| Eval-Protocol Role | Pydantic AI Conversion |
| ------------------ | ---------------------- |
| `user` | `UserPromptPart` |
| `system` | `SystemPromptPart` |
| `assistant` | `ChatCompletion` |
| `tool` | `ToolReturnPart` |
Example agent factory:
The examples assume a `setup_agent` function exists that creates and configures your Pydantic AI agent.
```python theme={null}
from eval_protocol.pytest import evaluation_test, PydanticAgentRolloutProcessor
from pydantic_ai.usage import UsageLimits
def agent_factory(config: RolloutProcessorConfig) -> Agent:
model_name = config.completion_params["model"]
# Provider is optional - defaults to "openai" if not specified
provider = config.completion_params.get("provider", "openai")
model = OpenAIChatModel(model_name, provider=provider)
return setup_agent(model)
@evaluation_test(
input_messages=[Message(role="user", content="Hello, how are you?")],
completion_params=[{
"model": "accounts/fireworks/models/gpt-oss-120b",
"provider": "fireworks" # Optional: defaults to "openai"
}],
rollout_processor=PydanticAgentRolloutProcessor(
agent_factory=agent_factory,
usage_limits=UsageLimits(max_tokens=1000)
),
mode="pointwise"
)
def test_pydantic_agent(row: EvaluationRow) -> EvaluationRow:
return row
```
Multi-agent scenario:
```python theme={null}
from eval_protocol.pytest import evaluation_test, PydanticAgentRolloutProcessor
from pydantic_ai import Agent, RunContext
from pydantic_ai.models import Model
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.usage import UsageLimits
import pytest
def setup_agent(joke_generation_model: Model, joke_selection_model: Model) -> Agent:
"""Setup multi-agent system with joke generation and selection."""
joke_selection_agent = Agent(
model=joke_selection_model,
system_prompt="Use the `joke_factory` to generate some jokes, then choose the best. You must return just a single joke."
)
joke_generation_agent = Agent(joke_generation_model, output_type=list[str])
@joke_selection_agent.tool
async def joke_factory(ctx: RunContext[None], count: int) -> list[str]:
r = await joke_generation_agent.run(
f"Please generate {count} jokes.",
usage=ctx.usage,
)
return r.output
return joke_selection_agent
def agent_factory(config: RolloutProcessorConfig) -> Agent:
joke_generation_model = OpenAIChatModel(
config.completion_params["model"]["joke_generation_model"], provider="fireworks"
)
joke_selection_model = OpenAIChatModel(
config.completion_params["model"]["joke_selection_model"], provider="fireworks"
)
return setup_agent(joke_generation_model, joke_selection_model)
@pytest.mark.asyncio
@evaluation_test(
input_messages=[[[Message(role="user", content="Tell me a joke.")]]],
completion_params=[{
"model": {
"joke_generation_model": "accounts/fireworks/models/kimi-k2-instruct",
"joke_selection_model": "accounts/fireworks/models/deepseek-v3p1"
}
}],
rollout_processor=PydanticAgentRolloutProcessor(
agent_factory=agent_factory,
usage_limits=UsageLimits(request_limit=5, total_tokens_limit=1000)
),
mode="pointwise"
)
async def test_pydantic_multi_agent(row: EvaluationRow) -> EvaluationRow:
return row
```
### MCPGymRolloutProcessor
* **What it does**: Spins up an MCP server (e.g., tau-bench style), creates environments, and runs rollouts through `eval_protocol.rollout(...)`.
* **When to use**: Interactive environments or "gym" tasks exposed over MCP.
* **Requires**: `server_script_path` to launch the MCP server. Binds `localhost:9700` by default.
* **Module**: [`eval_protocol/pytest/default_mcp_gym_rollout_processor.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/default_mcp_gym_rollout_processor.py)
Usage:
```python theme={null}
from eval_protocol.pytest import evaluation_test, MCPGymRolloutProcessor
@evaluation_test(
completion_params=[{"model": "openai/gpt-4o"}],
rollout_processor=MCPGymRolloutProcessor(),
server_script_path="examples/tau2_mcp/server.py",
steps=30,
)
def gym_eval(rows):
return rows
```
### RemoteRolloutProcessor (HTTP)
Using a remote HTTP service to perform the rollout is advanced. See [Remote
Rollout Processor](/tutorial/remote-rollout-processor) for more details.
## Pytest plugin helpers (CLI flags)
The pytest plugin in [`eval_protocol/pytest/plugin.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/plugin.py) adds flags to make evaluations CI-friendly:
* `--ep-max-rows=N|all`: limit dataset rows processed.
* `--ep-num-runs=N`: override the number of runs for evaluation\_test.
* `--ep-max-concurrent-rollouts=N`: override the maximum number of concurrent rollouts.
* `--ep-print-summary`: print a concise summary line at end of each run.
* `--ep-summary-json=PATH`: write a JSON artifact for CI.
* `--ep-input-param key=value` or `--ep-input-param @params.json`: ad-hoc overrides of `completion_params`.
* `--ep-reasoning-effort low|medium|high|none`: sets `extra_body.reasoning_effort` via LiteLLM.
* `--ep-max-retry=N`: set maximum retry attempts for failed rollouts.
* `--ep-fail-on-max-retry true|false`: whether to fail the entire rollout when permanent failures occur after max retries.
Example:
```bash theme={null}
pytest -k my_eval --ep-print-summary --ep-summary-json artifacts/my_eval.json --ep-max-rows 50 --ep-max-concurrent-rollouts 16
```
## Choosing a processor
* Use **single-turn** for simple QA and classification.
* Use **agent** when you need tool calls or iterative reasoning.
* Use **Pydantic AI agent** only for Pydantic AI framework.
* Use **MCP gym** for interactive environments hosted as MCP servers.
* Use **no-op** if you want full control inside your test body.
All processors stream results as they complete with bounded concurrency, so large datasets can run efficiently.
# Simulated Users
Source: https://evalprotocol.io/simulated-users
Evaluating conversational agents typically requires expensive human participants or pre-recorded dialogues that don't adapt to agent behavior. EP can simulate end-users in multi-turn evaluations, enabling full conversational loops without a human in the loop. This is powered by a lightweight user simulator derived from 𝜏²-bench and integrated into EP’s rollout manager. EP can simulate end-users in multi-turn evaluations, enabling full conversational loops without a human in the loop. This is powered by a lightweight user simulator derived from 𝜏²-bench and integrated into EP’s rollout manager.
## What It Does
* Generates realistic user turns based on scenario instructions and global guidelines.
* Interleaves with the agent’s tool-using turns to create full conversations.
* Signals when to stop (e.g., task complete, transfer, or out-of-scope) via a special termination token.
Under the hood, EP uses [UserSimulator](https://github.com/eval-protocol/python-sdk/blob/main/vendor/tau2/user/user_simulator.py). Rollout orchestration is handled by [ExecutionManager](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/execution/manager.py). The simulator:
* Builds a system prompt from global guidelines + your scenario instructions.
* Optionally uses tool schemas to steer requests.
* Provides a `is_stop(...)` check that EP maps to `termination_reason = "user_stop"`.
## Enabling Simulation
Provide `dataset_info.user_simulation` in your `EvaluationRow` (or dataset) to turn on the simulator for that row.
```json theme={null}
{
"messages": [
{ "role": "system", "content": "You are an assistant that uses tools." }
],
"input_metadata": {
"dataset_info": {
"user_prompt_template": "Observation: {observation}",
"environment_context": { "seed": 42 },
"user_simulation": {
"enabled": true,
"system_prompt": "You are a shopper trying to find a red jacket under $100.",
"llm": "gpt-4.1",
"llm_args": { "temperature": 0.0 }
}
}
}
}
```
Fields and defaults:
* `enabled`: boolean flag; if true, EP uses the simulator for the conversation.
* `system_prompt`: scenario instructions appended to global guidelines.
* `llm`: backing model for the user simulation (default: `gpt-4.1`).
* `llm_args`: sampling args for the simulator (default: `{ "temperature": 0.0 }`).
## Conversation Flow
When `user_simulation.enabled` is true:
* EP seeds the conversation with the simulator’s first user message.
* The agent policy receives tool schemas and responds with tool calls or a final answer.
* After each agent turn, the simulator may produce the next user message.
* If the simulator emits a stop intent, EP ends the episode with `termination_reason = user_stop`.
Step counting:
* Without simulation: each tool call increments the step counter.
* With simulation: EP increments the step counter after a full agent↔user turn, and records a consolidated control-plane step (reward, termination, tool calls).
## Minimal End-to-End
```python theme={null}
import eval_protocol as ep
from eval_protocol.models import EvaluationRow, Message
rows = [
EvaluationRow(
messages=[Message(role="system", content="Use tools to help the user.")],
input_metadata={
"dataset_info": {
"user_prompt_template": "Obs: {observation}",
"environment_context": {"seed": 7},
"user_simulation": {
"enabled": True,
"system_prompt": "Book a table for two tonight at 7pm.",
"llm": "gpt-4.1",
"llm_args": {"temperature": 0.0}
}
}
},
)
]
envs = ep.make("http://localhost:8000/mcp", evaluation_rows=rows, model_id="my-model")
policy = ep.OpenAIPolicy(model_id="gpt-4o-mini")
async def run():
async for row in ep.rollout(envs, policy=policy, steps=64):
print(row.rollout_status.termination_reason)
```
## Tips
* Keep scenario instructions specific and outcome-oriented to guide the simulator.
* Set `temperature` low for reproducible behavior (or use record/playback).
* Use rewards and control-plane summaries to assess task success rather than only length of the dialogue.
## Troubleshooting
* Simulator does nothing: ensure `user_simulation.enabled` is `true` and you have at least a system message.
* Episode never ends: check that your environment’s rewards/termination are wired, or set a sensible `steps` limit.
* Unexpected termination: the simulator may have emitted a stop intent; inspect `termination_reason` and conversation history.
## GitHub References
* User simulation integration in rollouts (ExecutionManager):
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/mcp/execution/manager.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/execution/manager.py)
* Backing user simulator (𝜏²-bench):
* [https://github.com/eval-protocol/python-sdk/blob/main/vendor/tau2/user/user\_simulator.py](https://github.com/eval-protocol/python-sdk/blob/main/vendor/tau2/user/user_simulator.py)
* Convenience facade and types:
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/mcp\_env.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp_env.py)
* [https://github.com/eval-protocol/python-sdk/blob/main/eval\_protocol/types/types.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/types/types.py)
# Specification
Source: https://evalprotocol.io/specification
## Core Execution Concepts
The following concepts define the lifecycle and data units of an evaluation. These match the semantics used by the `@evaluation_test` decorator in the Python SDK.
### invocation
A single execution of a test function. One invocation can generate one or more experiments.
### experiment
A group of runs for a specific combination of parameters (e.g., model x dataset x generation params). Each new execution of the test function produces a new experiment.
### run
A group of rollouts produced when repeating the same experiment multiple times. When `num_runs > 1`, each repetition has a unique `run_id`.
### rollout
The process that produces a `trajectory` for a single row. Each rollout has a unique `rollout_id`.
### trajectory
The sequence of chat `messages` (and optional tool calls) produced during a rollout.
### row
The atomic evaluation unit. A row contains the conversation `messages`, optional
`ground_truth`, and the evaluator's `evaluation_result`. Every row is uniquely
identified by its `row_id`. If not provided by the dataset, a stable hash is
generated based on the row's content.
### dataset
A collection (list) of rows. When stored, it is a JSONL file where each line is an `EvaluationRow`.
### eval
The rubric implemented in the body of an `@evaluation_test`-decorated function. It computes a `score` in
\[0, 1] and writes it to the row’s `evaluation_result`.
## Execution IDs and how they relate
Eval Protocol uses a small set of IDs to make rollouts traceable across systems (log stores, tracing UIs, dataset artifacts, etc.). These IDs are **correlation identifiers**, not “job objects” by themselves.
### ID hierarchy (recommended mental model)
Each rollout/trajectory can be identified by the tuple:
* **`invocation_id`** → **`experiment_id`** → **`run_id`** → **`row_id`** → **`rollout_id`**
In other words:
* An **invocation** contains one or more **experiments**
* An **experiment** contains one or more **runs** (repetitions)
* A **run** contains many **rows**
* For each **row**, there may be one or more **rollouts** (e.g., (N) samples per prompt)
### What each ID means
| Field | Stored on | Meaning | Uniqueness / stability expectations |
| --------------- | -------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `invocation_id` | `row.execution_metadata.invocation_id` | Identifier for one *invocation* (one top-level evaluation execution). All rows produced by the same invocation should share this value. | Stable within an invocation. Commonly set by an orchestrator (CI job id, workflow run id, etc.). |
| `experiment_id` | `row.execution_metadata.experiment_id` | Identifier for one *experiment* (a specific combination of eval code + dataset + completion params). | Stable for all rows within that experiment; changes when you change the parameterization. |
| `run_id` | `row.execution_metadata.run_id` | Identifier for one *run* (one repetition of an experiment when `num_runs > 1`). | Stable for the repetition. May be `null`/unset when `num_runs == 1` (implementation default). |
| `row_id` | `row.input_metadata.row_id` | Identifier for one *dataset row* (one prompt / episode seed / task instance). | Should be stable across reruns so you can compare the “same row” across experiments/runs. If omitted, implementations may deterministically generate one from row content. |
| `rollout_id` | `row.execution_metadata.rollout_id` | Identifier for one *rollout* (one concrete trajectory for a given row in a given run/experiment/invocation). | Unique per rollout. If you sample multiple trajectories per row, each sampled trajectory should have its own `rollout_id`. |
### Remote rollout processors: why all 5 IDs appear together
When using remote rollout processors, Eval Protocol passes a `RolloutMetadata` object that includes `invocation_id`, `experiment_id`, `run_id`, `row_id`, and `rollout_id` so the remote system can:
* tag logs/traces (e.g., `rollout_id:*`) for retrieval,
* correlate failures to a specific row and rollout,
* emit artifacts that can be merged back into the evaluation dataset.
### Mapping these IDs to “RL job” and “RL step”
Eval Protocol itself does not define an “RL job” concept; it defines **rollouts and evaluations**. In RL training pipelines, we recommend:
* **RL job id**: set/use `invocation_id` as the top-level grouping id (e.g., your trainer job id, CI run id, or workflow id).
* **RL step / epoch / iteration**: encode this in either:
* `run_id` (if your “step” is conceptually a repetition of the same experiment), or
* `execution_metadata.extra` / `input_metadata.session_data` (preferred when you need multiple axes like epoch + step + shard).
The key requirement is consistency: pick a convention so that all systems (tracing, dataset artifacts, training code) can join on the same keys.
## Foundational Types
### JSONType
```python theme={null}
JSONType = Union[Dict[str, Any], List[Any], str, int, float, bool, None]
```
### Message
Represents a chat message with trajectory evaluation support. `content` supports either a string or OpenAI content parts.
```python theme={null}
class ChatCompletionContentPartTextParam(BaseModel):
text: str
type: Literal["text"] = "text"
class Message(BaseModel):
role: str # assistant, user, system, tool
content: Optional[Union[str, List[ChatCompletionContentPartTextParam]]] = ""
reasoning_content: Optional[str] = None
name: Optional[str] = None
tool_call_id: Optional[str] = None
tool_calls: Optional[List[ChatCompletionMessageToolCall]] = None
function_call: Optional[FunctionCall] = None
control_plane_step: Optional[Dict[str, Any]] = None
```
### CompletionParams
```python theme={null}
CompletionParams = Dict[str, Any]
"""
Provider-agnostic completion parameters.
Required:
- model: str
Common fields:
- temperature: Optional[float]
- max_tokens: Optional[int]
- top_p: Optional[float]
Extra provider-specific fields are allowed and passed through (e.g., max_tool_calls).
"""
```
### InputMetadata
```python theme={null}
class InputMetadata(BaseModel):
# Accepts additional keys for future extensibility
# (model_config = ConfigDict(extra="allow") in implementation)
row_id: Optional[str] # defaulted to a generated ID
completion_params: CompletionParams = Field(default_factory=dict)
dataset_info: Optional[Dict[str, Any]] # seed, system_prompt, environment_context, etc.
session_data: Optional[Dict[str, Any]]
```
### ErrorInfo (AIP-193)
Structured error detail used inside `Status.details` per Google's AIP-193.
```python theme={null}
class ErrorInfo(BaseModel):
reason: str
domain: str
metadata: Dict[str, Any] = {}
```
### Status (AIP-193)
```python theme={null}
class Status(BaseModel):
class Code(int, Enum):
OK = 0
CANCELLED = 1
UNKNOWN = 2
INVALID_ARGUMENT = 3
DEADLINE_EXCEEDED = 4
NOT_FOUND = 5
ALREADY_EXISTS = 6
PERMISSION_DENIED = 7
RESOURCE_EXHAUSTED = 8
FAILED_PRECONDITION = 9
ABORTED = 10
OUT_OF_RANGE = 11
UNIMPLEMENTED = 12
INTERNAL = 13
UNAVAILABLE = 14
DATA_LOSS = 15
UNAUTHENTICATED = 16
# Custom codes used by Eval Protocol
FINISHED = 100
RUNNING = 101
SCORE_INVALID = 102
code: Code
message: str
details: List[Dict[str, Any]] = []
```
### TerminationReason
```python theme={null}
class TerminationReason(str, Enum):
MAX_STEPS = "max_steps"
CONTROL_PLANE_SIGNAL = "control_plane_signal"
USER_STOP = "user_stop"
SKIPPABLE_ERROR = "skippable_error"
NON_SKIPPABLE_ERROR = "non_skippable_error"
STOP = "stop"
LENGTH = "length"
TOOL_CALLS = "tool_calls"
```
### MetricResult
Result of a single metric evaluation:
```python theme={null}
class MetricResult(BaseModel):
is_score_valid: bool = True
score: float # Between 0.0 and 1.0
reason: str # Explanation for the score
data: Dict[str, Any] = Field(default_factory=dict) # Optional extra metric data
```
### StepOutput
Defines the base reward and other metrics for a single conceptual step within a rollout:
```python theme={null}
class StepOutput(BaseModel):
step_index: Union[int, str] # User-defined index for the step
base_reward: float # Base reward calculated by the user's reward function
terminated: bool = False # Whether the environment signaled termination
control_plane_info: Optional[Dict[str, Any]] # Structured info from environment
metrics: Dict[str, Any] = Field(default_factory=dict) # Optional custom metrics
reason: Optional[str] # Optional explanation for the step's base reward
```
### EvaluationThreshold
```python theme={null}
class EvaluationThreshold(BaseModel):
success: float # Minimum success rate threshold (0.0 to 1.0)
standard_error: Optional[float] # Optional maximum standard error threshold
```
### EvalMetadata
```python theme={null}
class EvalMetadata(BaseModel):
name: str
description: Optional[str]
version: str # PEP 440 version string (auto-populated)
status: Optional[Status]
num_runs: int
aggregation_method: str
passed_threshold: Optional[EvaluationThreshold]
passed: Optional[bool]
```
### CostMetrics
```python theme={null}
class CostMetrics(BaseModel):
input_cost: Optional[float]
output_cost: Optional[float]
total_cost_dollar: Optional[float]
```
### ExecutionMetadata
```python theme={null}
class ExecutionMetadata(BaseModel):
invocation_id: Optional[str]
experiment_id: Optional[str]
rollout_id: Optional[str]
run_id: Optional[str]
usage: Optional[CompletionUsage]
cost_metrics: Optional[CostMetrics]
duration_seconds: Optional[float]
experiment_duration_seconds: Optional[float]
```
## EvaluateResult
The `EvaluateResult` represents the complete result of an evaluator, providing an overall score and component metrics.
```python theme={null}
class EvaluateResult(BaseModel):
# Core evaluation data
score: float # Overall evaluation score (0.0 to 1.0)
is_score_valid: bool # Whether the overall score is valid (defaults to True)
reason: Optional[str] # Optional explanation for the overall score
# Component metrics
metrics: Dict[str, MetricResult] # Dictionary of component metrics
# RL-specific fields
step_outputs: Optional[List[StepOutput]] # Per-step base rewards for RL
# Error handling
error: Optional[str] # Optional error message if evaluation failed
# Trajectory information
trajectory_info: Optional[Dict[str, Any]] # Additional trajectory-level information
final_control_plane_info: Optional[Dict[str, Any]] # Final control plane state
# Aggregation across runs
agg_score: Optional[float] # Aggregated score across runs
standard_error: Optional[float] # Standard error across runs
```
**Key Features:**
* **Unified Model**: Serves both per-turn and per-trajectory evaluation scenarios
* **Component Metrics**: Detailed breakdown through `MetricResult` objects
* **RL Support**: Per-step base rewards via `step_outputs` for reinforcement learning
* **Error Handling**: Graceful error reporting and validation
* **Trajectory Info**: Additional metadata for trajectory-based evaluations
* **Aggregation**: Optional `agg_score` and `standard_error` for multi-run summaries
## EvaluationRow
The `EvaluationRow` is the canonical JSON-serializable unit of data used for both single-turn and trajectory evaluations. It contains the conversation, tool context, evaluation results, and metadata needed for reproducibility and analysis.
```python theme={null}
class EvaluationRow(BaseModel):
# Core conversation (trajectory) data
messages: List[Message]
# Tool and function call information
tools: Optional[List[Dict[str, Any]]] = None
# Input-related metadata
input_metadata: InputMetadata = Field(default_factory=InputMetadata)
# Rollout status (AIP-193)
rollout_status: Status = Field(default_factory=Status.rollout_running)
# Optional ground truth reference
ground_truth: Optional[JSONType] = None
# Unified evaluation result
evaluation_result: Optional[EvaluateResult] = None
# Correlation identifiers grouped under execution metadata
execution_metadata: ExecutionMetadata = Field(default_factory=lambda: ExecutionMetadata(run_id=None))
# Timestamps and evaluation metadata
created_at: datetime = Field(default_factory=datetime.now)
eval_metadata: Optional[EvalMetadata] = None
# Process info for watchdogs
pid: Optional[int] = None
```
**Key Features:**
* **Unified Format**: Canonical row format for both pointwise and trajectory evaluations
* **Explicit Status**: `rollout_status` captures running/finished/error
* **Reproducibility**: `input_metadata`, seeds, and identifiers support traceability
* **Usage Tracking**: Captures token usage statistics from LLM calls
## Dataset
A list of `EvaluationRow`s. When saved to file, it is a JSONL file where each
line is a JSON-encoded `EvaluationRow`.
### JSONL example
```json expandable theme={null}
{
"messages": [
{ "role": "system", "content": "You are a helpful assistant." },
{ "role": "user", "content": "Add 2 and 3." },
{ "role": "assistant", "content": "5" }
],
"tools": null,
"input_metadata": {
"row_id": "row_123",
"completion_params": {
"model": "openai/gpt-4o",
"temperature": 0.0,
"max_tokens": 256,
"max_tool_calls": 0
},
"dataset_info": {
"seed": 42,
"system_prompt": "You are a helpful assistant.",
"environment_context": {}
},
"session_data": {
"mode": "pointwise"
}
},
"rollout_status": {
"code": 100,
"message": "Rollout finished",
"details": []
},
"ground_truth": "5",
"evaluation_result": {
"score": 1.0,
"is_score_valid": true,
"reason": "Exact match",
"metrics": {
"exact_match": {
"is_score_valid": true,
"score": 1.0,
"reason": "assistant output matches ground truth"
}
},
"step_outputs": null,
"error": null,
"trajectory_info": null,
"final_control_plane_info": null,
"agg_score": 1.0,
"standard_error": 0.0
},
"execution_metadata": {
"invocation_id": "ivk_abcd",
"experiment_id": "exp_efgh",
"rollout_id": "rll_ijkl",
"run_id": null,
"usage": {
"prompt_tokens": 10,
"completion_tokens": 1,
"total_tokens": 11
},
"cost_metrics": { "total_cost_dollar": 0.0002 },
"duration_seconds": 0.012,
"experiment_duration_seconds": 0.045
},
"created_at": "2025-01-01T12:00:00",
"eval_metadata": {
"name": "basic_addition",
"description": "Verify simple arithmetic",
"version": "0.1.0",
"status": { "code": 100, "message": "Evaluation finished", "details": [] },
"num_runs": 1,
"aggregation_method": "mean",
"passed_threshold": { "success": 0.95 },
"passed": true
},
"pid": 12345
}
```
## EvaluationTest
The `EvaluationTest` represents a test configuration for evaluating models.
While not explicitly defined as a separate class in the current implementation,
evaluation tests are configured through the `evaluation_test` decorator. The decorator
can be used to configure the following:
* **Dataset Configuration**: JSONL files containing test cases or hard-coded `input_messages`
* **Model Configuration**: Completion parameters (must include `model`) and generation settings via `completion_params`
* **Evaluation Criteria**: Success thresholds (via `passed_threshold`), with optional standard deviation constraint
* **Environment Configuration**: MCP config, rollout steps, server path, and concurrency
* **Rollout Processor**: Class to execute rollouts (e.g., `SingleTurnRolloutProcessor()`)
* **Number of Runs**: Number of times to repeat the rollout (e.g., `num_runs=1`)
* **Mode**: Evaluation mode (`pointwise`, `groupwise`, or `all`)
* **Aggregation**: Aggregation method (e.g., `mean`) and optional env overrides for summaries
```python theme={null}
from eval_protocol.pytest import evaluation_test, SingleTurnRolloutProcessor
@evaluation_test(
input_dataset=["tests/pytest/data/markdown_dataset.jsonl"],
dataset_adapter=markdown_dataset_to_evaluation_row,
completion_params=[{
"model": "fireworks_ai/accounts/fireworks/models/llama-v3p1-8b-instruct",
"temperature": 0.0,
"max_tokens": 4096,
}],
passed_threshold={"success": 0.5},
rollout_processor=SingleTurnRolloutProcessor(),
num_runs=1,
mode="pointwise",
)
def test_markdown_highlighting_evaluation(row: EvaluationRow) -> EvaluationRow:
...
```
## MCP Gym
`McpGym` is the base class for building environments that an LLM can interact with via MCP tool calls (data plane) while exposing rewards and episode status via HTTP control-plane endpoints. This enables reproducible RL-style rollouts with clean separation of concerns.
Key concepts:
* **Data plane**: Tool calls and JSON responses used by the model to act and observe state
* **Control plane**: Session-scoped endpoints for rewards, termination, and info
* **Multi-session**: Stable `session_id` keys route control-plane queries to the right episode
Core API surface:
* `control_plane_endpoint(path)`: Decorator to register a session-aware endpoint
* `_register_tools()`: Register domain tools with `self.mcp.tool()`
* `format_observation(obs, env) -> Dict[str, Any]`: Return JSON-serializable observation payloads
* `run(transport="streamable-http")`: Start the FastMCP server with high-concurrency settings
* Standard control-plane endpoints on subclasses: `/control/reward`, `/control/status`, `/control/info`, `/control/initial_state`
Example stub:
```python theme={null}
class McpGym(ABC):
def __init__(self, server_name: str, adapter: EnvironmentAdapter, seed: Optional[int] = None, max_workers: Optional[int] = None):
...
@abstractmethod
def _register_tools(self):
...
def format_observation(self, obs: Any, env: Any) -> Dict[str, Any]:
...
def run(self, transport: str = "streamable-http", **kwargs):
...
```
See [`python-sdk/eval_protocol/mcp/mcpgym.py`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/mcp/mcpgym.py) for the full implementation including the `control_plane_endpoint` decorator and session handling.
## Environment
The `EnvironmentAdapter` class provides the interface for connecting environments to the MCP framework.
```python theme={null}
class EnvironmentAdapter:
"""
Environment adapter with default implementations.
Users can either use this class directly by providing an env_class,
or inherit from it to customize specific methods for their environment.
This provides a clean separation between the MCP protocol layer
and the environment implementation.
"""
```
**Key Features:**
* **Default Implementations**: Works with most gymnasium-style and complex environments
* **Flexible Configuration**: Supports custom configuration dictionaries
* **Seed Support**: Reproducible environments through seed-based initialization
* **Clean Interface**: Separates MCP protocol layer from environment implementation
**Core Methods:**
* `create_environment()`: Create and return a new environment instance
* `create_environment_with_seed()`: Create environment with specific seed for reproducibility
* `reset_environment()`: Reset environment to initial state
* `step_environment()`: Execute one step in the environment
* `close_environment()`: Clean up environment resources
* `parse_action()`: Parse action string to environment-specific format
* `format_observation()`: Format observation for MCP transmission
## Policy
A policy is a model such as `gpt-4o` or `llama-3.1-8b`. In more advanced scenarios, a policy can be your own custom fine-tuned model.
The `LiteLLMPolicy` class provides a unified implementation that works with ANY MCP environment via tool calling:
```python theme={null}
class LiteLLMPolicy(LLMBasePolicy):
"""
Unified LiteLLM policy implementation that works with ANY MCP environment via tool calling.
Supports OpenAI, Anthropic, Fireworks AI
Includes built-in retry logic and caching.
NO environment-specific logic - everything comes from MCP tools and dataset prompts.
"""
```
**Key Features:**
* **Provider Agnostic**: Supports OpenAI, Anthropic, Fireworks AI, and other providers
* **Built-in Caching**: Multiple cache types (memory, Redis, dual, S3, disk)
* **Retry Logic**: Robust retry strategies with exponential backoff
* **Tool Calling**: Native support for MCP tool calling
* **Environment Agnostic**: No environment-specific logic - everything from MCP tools
**Specialized Implementations:**
* `OpenAIPolicy`: OpenAI-specific policy implementation
* `AnthropicPolicy`: Anthropic Claude-specific policy implementation
* `FireworksPolicy`: Fireworks AI-specific policy implementation
* `LocalPolicy`: Local model policy implementation
**Core Capabilities:**
* **Multi-Tool Support**: Handle multiple tool calls per turn
* **Conversation History**: Maintain context across interactions
* **Error Handling**: Graceful handling of API failures and retries
* **Caching**: Response caching for improved performance and cost reduction
* **Logging**: Comprehensive logging for debugging and analysis
## Additional Core Classes
### MCPSession
Represents a single MCP session with an environment:
```python theme={null}
@dataclass
class MCPSession:
session_id: str
base_url: str
seed: Optional[int]
model_id: str
dataset_row: Optional[DatasetRow] = None
terminated: bool = False
last_observation: Any = None
_exit_stack: Optional[AsyncExitStack] = None # persistent connection resources
_mcp_session: Optional[ClientSession] = None # persistent MCP client session
```
### Trajectory
Represents a complete rollout trajectory:
```python theme={null}
@dataclass
class Trajectory:
session: MCPSession
observations: List[Any]
actions: List[str]
rewards: List[float]
terminated: bool
total_reward: float
steps: int
duration: float
control_plane_steps: List[Dict[str, Any]]
control_plane_summary: Dict[str, Any]
termination_reason: str
conversation_history: List[Dict[str, Any]]
usage: Dict[str, int] = field(default_factory=dict)
```
# Evaluation Tests (Getting Started)
Source: https://evalprotocol.io/tutorial/evaluation-tests-getting-started
Write your first @evaluation_test in a few minutes, then scale up to real benchmarks.
The [`@evaluation_test`](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/evaluation_test.py) decorator is the core component for creating pytest-based evaluation tests in the Evaluation Protocol. It enables you to evaluate AI models by running rollouts and applying evaluation criteria to measure performance.
## What is an `@evaluation_test`?
An `@evaluation_test` is a **pytest test with superpowers**:
* **Takes in rows** (from a dataset, loader, or hard‑coded messages)
* **Runs rollouts via a [Rollout Processor](/tutorial/rollout-processors-getting-started)** (more on this on the next page)
* **Evaluates and writes scores** onto each row
* **Aggregates results** and surfaces a pass/fail signal for CI
You can think of it as:
* pytest for orchestration
* Eval Protocol for rollouts, retries, and logging
* Your function body for scoring
For the full API, see the [reference page](/reference/evaluation-test). This guide focuses on the **shortest path to something useful**.
## Smallest useful example (pointwise, single model)
This example:
* Loads rows from a JSONL dataset
* Calls a model once per row using the default rollout processor
* Scores each row from 0–1
```python test_math_reasoning.py theme={null}
import pytest
from typing import List
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import evaluation_test, SingleTurnRolloutProcessor
def evaluate_math_reasoning(messages) -> float:
# Dummy scoring: in a real eval, parse the model's answer and check it
text = (messages[-1].content or "").lower()
return 1.0 if "4" in text else 0.0
@pytest.mark.parametrize(
"completion_params",
[
{"model": "openai/gpt-4o", "temperature": 0.0},
],
)
@evaluation_test(
input_dataset=["path/to/dataset.jsonl"],
rollout_processor=SingleTurnRolloutProcessor(),
passed_threshold=0.8,
mode="pointwise",
)
def test_math_reasoning(row: EvaluationRow) -> EvaluationRow:
"""Score a single row between 0 and 1."""
score = evaluate_math_reasoning(row.messages)
row.evaluation_result.score = score
return row
```
Run it with:
```bash theme={null}
pytest test_math_reasoning.py
```
> The decorator handles dataset loading, concurrency, aggregation, and summary generation for you.
## Minimal mental model
* **Input**: rows come from **one** of:
* `input_dataset` (JSONL paths)
* `input_messages` (inline messages)
* `input_rows` (pre‑built `EvaluationRow`s)
* `data_loaders` (dynamic loaders)
* **Rollouts**: controlled by:
* `completion_params` (model + generation settings)
* `rollout_processor` (how to talk to the model / environment)
* **Scoring**:
* Your function body writes `row.evaluation_result.score` in \[0, 1]
* Optionally add `evaluation_result.reason` and `evaluation_result.metrics`
If you remember **"rows in → rollouts → scores out"**, you're 80% of the way there.
## Using `is_score_valid` for Training
The `is_score_valid` field controls whether a rollout's score should be used in training algorithms. When set to `False`, the rollout is excluded from training data (e.g., RFT, GRPO) while still being logged for analysis.
**When to set `is_score_valid=False`:**
* No assistant response exists (malformed rollout)
* The evaluation cannot produce a meaningful score
* External dependencies failed (e.g., tool execution errors)
* The response is unparseable or invalid
**Example: Excluding rollouts without assistant responses**
```python theme={null}
def evaluate(row: EvaluationRow) -> EvaluationRow:
# Check if the last message is from the assistant
if not row.messages or row.messages[-1].role != "assistant":
row.evaluation_result = EvaluateResult(
score=0.0,
reason="No assistant response",
is_score_valid=False # Exclude from training
)
return row
# Normal evaluation logic
score = compute_score(row)
row.evaluation_result = EvaluateResult(
score=score,
reason="Evaluation completed",
is_score_valid=True # Include in training (default)
)
return row
```
Training pipelines (TRL, rLLM, OpenAI RFT) use `is_score_valid` to filter rollouts before computing gradients. A rollout with `is_score_valid=False` will not contribute to the loss function, preventing noisy or invalid samples from affecting model updates.
## When to jump to the full reference
Stay on this page until you need:
* `data_loaders` or custom adapters
* Advanced **aggregation** configs
* Custom **exception handling** and backoff strategies
* Detailed environment variable behavior
When you hit those needs, the [full `@evaluation_test` reference](/reference/evaluation-test) covers every parameter and edge case.
# Running Rollouts with GitHub Actions
Source: https://evalprotocol.io/tutorial/github-actions-rollout
If you already have an agent, you can integrate it with Eval Protocol by
using the [GithubActionRolloutProcessor](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/github_action_rollout_processor.py).
`GithubActionRolloutProcessor` delegates rollout execution to GitHub Actions workflows
that you control. It's useful for implementing rollouts with your existing agent
codebase by wrapping it in a GitHub Actions workflow.
## High Level Flow
![Github Action Rollout Processor flow]()
1. **/init triggers one rollout**: Eval Protocol dispatches a GitHub Actions workflow with `completion_params`, `metadata` (incl. `rollout_id`), and `model_base_url`.
2. **Polling to check rollout status**: The processor finds the `rollout:` run and polls GitHub Actions until it completes.
3. **Send chat completions and store as trace**: The workflow executes your agent and sends completions/logs to Fireworks with the rollout’s correlation tags.
4. **Once rollout finished, pull full trace and evaluate**: Eval Protocol fetches the Fireworks trace by `rollout_id` and scores the result.
Everything inside the dotted box is handled by Eval Protocol — you only need to implement the GitHub Actions workflow, more on this below.
## Setup
For the GitHub token, create a [Personal Access Token (classic)](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens#personal-access-tokens-classic) with permissions for `repo` and `workflow`.
```bash theme={null}
GITHUB_TOKEN="ghp_..."
```
## GitHub Actions Contract
We expect the GitHub Actions workflow to accept the following inputs:
JSON completion parameters (model, temperature, etc.)
JSON string containing rollout execution metadata
Base URL for the model API (e.g., "[https://tracing.fireworks.ai](https://tracing.fireworks.ai)")
API key for the model API
You must add `run-name: rollout:${{ fromJSON(inputs.metadata).rollout_id }}` to your workflow. The GitHub API doesn't return the run ID when dispatching, so this allows us to find and monitor the correct workflow run.
```yaml theme={null}
name: Eval Protocol Rollout
run-name: rollout:${{ fromJSON(inputs.metadata).rollout_id }}
on:
workflow_dispatch:
inputs:
completion_params:
description: 'JSON completion params'
required: true
type: string
metadata:
description: 'JSON serialized metadata object'
required: true
type: string
model_base_url:
description: 'Base URL for the model API'
required: true
type: string
api_key:
description: 'API key for the model API'
required: true
type: string
```
## Metadata Correlation
When making model calls in your GitHub Actions workflow, include the following metadata
in your traces and logs so that `eval-protocol` can correlate them with the
corresponding `EvaluationRow`s during result collection. `GithubActionRolloutProcessor`
automatically generates this and sends it to the server, so you don't need to worry
about wrangling metadata.
* `invocation_id`
* `experiment_id`
* `rollout_id`
* `run_id`
* `row_id`
## Example
See the following repo for a simple end to end example:
* [Github Action Rollout Processor Hello World](https://github.com/eval-protocol/github-action-rollout-processor-hello-world)
# GSM8K Fine-tuning Quickstart (Small Model)
Source: https://evalprotocol.io/tutorial/gsm8k-finetuning-quickstart
Run pytest to materialize the evaluator and dataset, then launch a local Reinforcement Fine-Tuning job on a small model.
Run GSM8K locally end-to-end:
* Materialize a GSM8K evaluator and dataset with `pytest`
* Kick off a Reinforcement Fine-Tuning (RFT) job for a small base model
* Track accuracy improvements by re-running the evaluator
Running the GSM8K tutorial in Google Colab requires a Google account with billing enabled (credit card on file). Fireworks usage also bills against your account once you supply `FIREWORKS_API_KEY`. 👉 [Run the GSM8K Fine-tuning Colab](https://colab.research.google.com/drive/16xrb9rx6AoAEOtrDXumzo71HjhunaoPi#scrollTo=CP18QX4tgi-0)
## Prerequisites
* Python 3.10+
* Local Python environment with Jupyter support (VS Code, JupyterLab, or classic notebook)
* `FIREWORKS_API_KEY` with permissions to launch RFT jobs (stored in your shell or `.env`)
* Basic familiarity with GSM8K-style math reasoning tasks
Install the latest `eval-protocol` SDK directly from the main branch and make sure `pytest` is on the path. Upgrade `pip` first to avoid resolver issues.
```bash theme={null}
python -m pip install --upgrade pip
python -m pip install pytest git+https://github.com/eval-protocol/python-sdk.git
```
Download the evaluation assets we will use to kick off the job. Copy the GSM8K pytest script and sample dataset into a working directory (here `gsm8k_artifacts/`). The snippet below is safe to run inside a notebook cell or standalone script and ensures the files land where later steps expect them.
```python tutorial/download_gsm8k_assets.py theme={null}
from pathlib import Path
import requests
ARTIFACT_ROOT = Path("gsm8k_artifacts")
TEST_PATH = ARTIFACT_ROOT / "tests" / "pytest" / "gsm8k" / "test_pytest_math_example.py"
DATASET_PATH = ARTIFACT_ROOT / "development" / "gsm8k_sample.jsonl"
files_to_download = {
TEST_PATH: "https://raw.githubusercontent.com/eval-protocol/python-sdk/main/tests/pytest/gsm8k/test_pytest_math_example.py",
DATASET_PATH: "https://raw.githubusercontent.com/eval-protocol/python-sdk/main/development/gsm8k_sample.jsonl",
}
for local_path, url in files_to_download.items():
local_path.parent.mkdir(parents=True, exist_ok=True)
response = requests.get(url, timeout=30)
response.raise_for_status()
local_path.write_bytes(response.content)
print(f"Saved {url} -> {local_path}")
```
Expected output:
```
Saved https://raw.githubusercontent.com/.../test_pytest_math_example.py -> gsm8k_artifacts/tests/pytest/gsm8k/test_pytest_math_example.py
Saved https://raw.githubusercontent.com/.../gsm8k_sample.jsonl -> gsm8k_artifacts/development/gsm8k_sample.jsonl
```
Execute the evaluation that materializes the evaluator and dataset. Point the test at the artifacts folder you created in the previous step.
```bash theme={null}
cd gsm8k_artifacts
ep local-test
```
This command discovers and runs your `@evaluation_test` with pytest.
You should see log output for each rollout and navigate to [http://localhost:8000](http://localhost:8000) to see the Eval Protocol UI and inspect results.
![Screenshot of the local GSM8K evaluation UI showing aggregate scores and trajectories panels.]()
Store your Fireworks API key in the environment so the CLI can authenticate. The command below keeps the key confined to the current shell session.
```bash theme={null}
export FIREWORKS_API_KEY=""
```
Alternatively, load it from a secrets manager or `.env` file if your workflow already manages credentials securely.
Trigger the `eval-protocol` CLI to start a Reinforcement Fine-Tuning job using the evaluator and dataset registered above. Replace the base model to experiment with other policies.
```bash theme={null}
cd ..
eval-protocol create rft --base-model accounts/fireworks/models/qwen3-0p6b
```
The CLI reports dashboard links for the evaluator, dataset, and RFT job so you can monitor rollouts.
![Screenshot of the Fireworks RFT dashboard highlighting final accuracy metrics for the GSM8K job.]()
## Track accuracy over time
* Re-run the `ep local-test` command periodically to evaluate the latest checkpoint against the GSM8K slice.
* Adjust reward shaping or parsing logic inside `test_pytest_math_example.py` to fit your formatting expectations.
* Swap in a custom dataset JSONL by editing the local artifact or passing `--dataset-jsonl` when creating the RFT job.
## What’s happening under the hood
* The evaluation tests a small GSM8K slice with a numeric-check reward and registers an evaluator plus dataset with your local API.
* The `create rft` command wires those resources into a Reinforcement Fine-Tuning job for the specified base model.
* As training progresses, evaluation scores reflect improved accuracy on the held-out set, letting you iterate quickly before scaling up.
## Next steps
* Parameterize base models and dataset paths in scripts or notebooks to make repeated experiments easier.
* Automate the evaluation loop in CI so new policies are validated before deployment.
* Promote successful evaluators and datasets to shared registries once the workflow is stable.
Ready to try your own? For a ready-to-run project with all files included, clone the [GSM8K Quickstart Repository](https://github.com/eval-protocol/quickstart-gsm8k) and get started now!
# Running Rollouts with a Remote Server
Source: https://evalprotocol.io/tutorial/remote-rollout-processor
If you already have an agent, you can integrate it with Eval Protocol by
using the [RemoteRolloutProcessor](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/pytest/remote_rollout_processor.py).
`RemoteRolloutProcessor` delegates rollout execution to a remote HTTP service
that you control. It's useful for implementing rollouts with your existing agent
codebase by wrapping it in an HTTP service.
## High Level Flow
![Remote Rollout Processor flow]()
1. **/init triggers one rollout**: Eval Protocol calls your service’s POST `/init` with the row payload and correlation metadata.
2. **Send logs via `FireworksTracingHttpHandler`**: Your service emits structured logs tagged with the rollout’s correlation fields.
3. **Send chat completions and store as trace**: Your agent’s calls are recorded as traces in Fireworks.
4. **Once rollout finished, pull full trace and evaluate**: Eval Protocol polls Fireworks for a completion signal, then loads the trace and scores it.
Everything inside the dotted box is handled by Eval Protocol — you only need to implement the Remote Server, more on this below.
## API Contract
**POST /init:**
We expect the remote service to implement a single /init endpoint that accepts an `InitRequest` with the following fields:
Dictionary containing model and optional parameters like temperature, max\_tokens, etc.
Array of conversation messages
Array of available tools for the model
Base URL for the remote server to make LLM calls
Rollout execution metadata for correlation
API key to be used by the remote server
```json init_request.json theme={null}
{
"completion_params": {
"model": "accounts/fireworks/models/gpt-oss-120b",
"temperature": 0.7,
"max_tokens": 2048
},
"messages": [
{ "role": "user", "content": "What is the weather in San Francisco?" }
],
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the weather for a city",
"parameters": {
"type": "object",
"properties": {
"city": { "type": "string" }
}
}
}
}
],
"model_base_url": "https://tracing.fireworks.ai/rollout_id/brave-night-42/invocation_id/wise-ocean-15/experiment_id/calm-forest-28/run_id/quick-river-07/row_id/bright-star-91",
"metadata": {
"invocation_id": "wise-ocean-15",
"experiment_id": "calm-forest-28",
"rollout_id": "brave-night-42",
"run_id": "quick-river-07",
"row_id": "bright-star-91"
},
"api_key": "fw_your_api_key"
}
```
## Metadata Correlation
When making model calls in your remote server, include the following metadata
in your traces and logs so that `eval-protocol` can correlate them with the
corresponding `EvaluationRow`s during result collection. `RemoteRolloutProcessor`
automatically generates this and sends it to the server, so you don't need to worry
about wrangling metadata.
* `invocation_id`
* `experiment_id`
* `rollout_id`
* `run_id`
* `row_id`
## Handling `InitRequest` in your Server
This section shows how to **parse the `InitRequest` fields and call your model**. Note: the `model_base_url` is a `tracing.fireworks.ai` URL that proxies your model calls so Fireworks can capture full traces for each rollout.
Below is a minimal FastAPI server showing how to wire this together.
```python remote_server.py theme={null}
@app.post("/init")
def init(req: InitRequest):
if not req.messages:
raise ValueError("messages is required")
model = req.completion_params.get("model")
if not model:
raise ValueError("model is required in completion_params")
# Spread all completion_params (model, temperature, max_tokens, etc.)
completion_kwargs = {"messages": req.messages, **req.completion_params}
if req.tools:
completion_kwargs["tools"] = req.tools
# Build OpenAI client from InitRequest
# You can also use req.api_key instead of an environment variable if preferred.
client = OpenAI(
base_url=req.model_base_url,
api_key=os.environ.get("FIREWORKS_API_KEY"),
)
completion = client.chat.completions.create(**completion_kwargs)
```
## Signaling Rollout Completion
The `RemoteRolloutProcessor` detects rollout completion by polling **structured logs** sent to Fireworks Tracing. Your remote server should use the appropriate SDK for your language to emit structured completion statuses.
The `eval-protocol` JS/TS SDK provides equivalent helpers:
* **`withFireworksLogging`**: Wraps your handler to automatically send structured logs to Fireworks Tracing.
* **`createRolloutLogger`**: Creates a rollout-scoped logger tagged with the current `rollout_id`.
* **`Status` / `mapOpenAIErrorToStatus`**: Helpers for emitting structured completion and error statuses.
```ts api/init.ts highlight={14-15, 20-21, 23-24, 29} theme={null}
import type { VercelRequest, VercelResponse } from '@vercel/node';
import {
initRequestSchema,
type InitRequest,
Status,
createRolloutLogger,
withFireworksLogging,
} from 'eval-protocol';
async function handler(req: VercelRequest, res: VercelResponse) {
const initRequest: InitRequest = initRequestSchema.parse(req.body);
// Extract rollout id from InitRequest payload and create rollout-specific logger
const rolloutId = initRequest.metadata.rollout_id;
const logger = createRolloutLogger(rolloutId);
try {
// Execute your rollout here
const status = Status.rolloutFinished();
logger.info(`Rollout ${rolloutId} completed`, { status });
} catch (error: any) {
const status = Status.rolloutInternalError(error.message);
logger.error(`Rollout ${rolloutId} failed: ${error.message}`, { status });
}
}
// Export wrapped handler so all logs go to Fireworks Tracing
export default withFireworksLogging(handler);
```
Add `FireworksTracingHttpHandler` as the logging handler, a `RolloutIdFilter`, and log completion status using structured `Status` objects:
```python remote_server.py highlight={5-6, 11-12, 18-21, 25-28} theme={null}
import logging
from eval_protocol import Status, InitRequest, FireworksTracingHttpHandler, RolloutIdFilter
# Configure Fireworks tracing handler
fireworks_handler = FireworksTracingHttpHandler()
logging.getLogger().addHandler(fireworks_handler)
@app.post("/init")
def init(request: InitRequest):
# Create rollout-specific logger with filter
rollout_logger = logging.getLogger(f"eval_server.{request.metadata.rollout_id}")
rollout_logger.addFilter(RolloutIdFilter(request.metadata.rollout_id))
try:
# Execute your rollout here
# Then log successful completion with structured status
rollout_logger.info(
f"Rollout {request.metadata.rollout_id} completed",
extra={"status": Status.rollout_finished()}
)
except Exception as e:
# Log errors with structured status
rollout_logger.error(
f"Rollout {request.metadata.rollout_id} failed: {e}",
extra={"status": Status.rollout_error(str(e))}
)
```
### Alternative: Environment Variable Approach
For the following setups, you can use the `EP_ROLLOUT_ID` environment variable instead of manual filters:
1. One rollout is processed per server instance
```python remote_server.py highlight={6} theme={null}
import os
import logging
from eval_protocol import Status, InitRequest, FireworksTracingHttpHandler
# Configure Fireworks tracing handler
os.environ["EP_ROLLOUT_ID"] = request.metadata.rollout_id
fireworks_handler = FireworksTracingHttpHandler()
logging.getLogger().addHandler(fireworks_handler)
logger = logging.getLogger(__name__)
@app.post("/init")
def init(request: InitRequest):
...
```
2. `/init` spawns separate Python processes
```python remote_server.py highlight={8-9} theme={null}
import os
import logging
import multiprocessing
from eval_protocol import FireworksTracingHttpHandler, InitRequest
def execute_rollout_step_sync(request):
# Set in the CHILD process
os.environ["EP_ROLLOUT_ID"] = rollout_id
logging.getLogger().addHandler(FireworksTracingHttpHandler())
# Execute your rollout here
@app.post("/init")
async def init(request: InitRequest):
# Do NOT set EP_ROLLOUT_ID here; set it in the child
p = multiprocessing.Process(
target=execute_rollout_step_sync,
args=(request),
)
p.start()
```
### How `RemoteRolloutProcessor` uses Fireworks Tracing
1. **Remote server logs completion**: Uses `Status.rollout_finished()` or `Status.rollout_error()`
2. **RemoteRolloutProcessor polls**: Searches logs by `rollout_id` tag until completion found
3. **Status extraction**: Reads structured status fields (`code`, `message`, `details`)
## Multi-Agent Setup Fine-tuning / Artifact Storage
One other use-case the `RemoteRolloutProcessor` enables is fine-tuning on multi-agent setups by storing artifacts. For example, let's say you have a Deep Research multi-agent setup and you want to fine-tune the first subagent, but the evaluation is on the artifact produced at the end of this entire multi-agent pipeline, e.g. the final deep research output. To accomplish this, we must store the artifact and correlate it with the subagent's output.
Pass custom data in the `extras` field when logging completion status:
Only log `Status.rolloutFinished()` **after the entire pipeline completes** and the final artifact is ready—not immediately when the subagent finishes. The completion log signals that evaluation can begin, so it must include all artifacts needed for scoring.
```ts api/init.ts theme={null}
// Wait for full pipeline to complete, then log with extras
const status = Status.rolloutFinished();
logger.info(`Rollout ${rolloutId} completed`, {
status,
extras: {
messages: result.messages,
research_report: result.report,
sources: result.citations,
}
});
```
```python remote_server.py theme={null}
# Wait for full pipeline to complete, then log with extras
rollout_logger.info(
f"Rollout {request.metadata.rollout_id} completed",
extra={
"status": Status.rollout_finished(),
"extras": {
"messages": result.messages,
"research_report": result.report,
"sources": result.citations,
}
}
)
```
The `RemoteRolloutProcessor` automatically extracts `extras` from the completion log and stores them in `row.execution_metadata.extra`. Access them in your evaluation function:
```python test_deep_research.py theme={null}
@evaluation_test(
input_dataset=[str(Path(__file__).parent / "research_dataset.jsonl")],
completion_params=[{"model": "accounts/fireworks/models/gpt-oss-120b"}],
rollout_processor=RemoteRolloutProcessor(
remote_base_url="https://your-server.vercel.app",
),
)
async def deep_research_evaluation(row: EvaluationRow) -> EvaluationRow:
# Access artifacts stored by your remote server
research_report = row.execution_metadata.extra["research_report"]
sources = row.execution_metadata.extra["sources"]
# Evaluate the final artifact
row.evaluation_result = evaluate_report_quality(research_report, sources)
return row
```
## Handling Rollout Failures
Rollouts can fail for various reasons: your remote server might crash, tracing might fail, or the model might not produce an assistant response. The `RemoteRolloutProcessor` automatically detects these failures and sets `row.rollout_status` accordingly.
The most common failure is when the rollout produces no assistant response. The SDK detects this and sets `row.rollout_status` to `Internal (13)` with the message "Rollout finished with the same number of messages as the original row".
Even when a rollout fails, your evaluation function is still called—giving you control over how to handle errors.
**Best practice:** Check `row.rollout_status.is_error()` at the start of your evaluation function to catch failed rollouts. This method returns `True` when the status code is `INTERNAL`:
```python test_my_eval.py theme={null}
from eval_protocol import EvaluateResult, EvaluationRow, evaluation_test
from eval_protocol.pytest import RemoteRolloutProcessor
@evaluation_test(
input_dataset=["dataset.jsonl"],
rollout_processor=RemoteRolloutProcessor(
remote_base_url="https://your-server.vercel.app",
),
completion_params=[{"model": "accounts/fireworks/models/gpt-oss-120b"}],
)
def test_my_evaluation(row: EvaluationRow) -> EvaluationRow:
# Check if rollout failed with internal error (e.g., no assistant response)
if row.rollout_status.is_error():
row.evaluation_result = EvaluateResult(
score=0.0,
reason=f"Rollout failed: {row.rollout_status.message}",
is_score_valid=False,
)
return row
# Proceed with normal evaluation logic
...
```
This catches the most common failure mode—when your remote server fails to produce an assistant response or encounters an internal error.
**Alternative:** Check the messages directly instead of relying on the SDK's error detection:
```python test_my_eval.py theme={null}
def test_my_evaluation(row: EvaluationRow) -> EvaluationRow:
# Check if no assistant response was produced
if not row.messages or row.messages[-1].role != "assistant":
row.evaluation_result = EvaluateResult(
score=0.0,
reason="No assistant response - rollout may have failed",
is_score_valid=False,
)
return row
# Proceed with normal evaluation logic
...
```
This approach doesn't depend on the SDK's status detection and directly validates that an assistant response exists.
## Example
See the following repos for end to end examples:
* [Remote Rollout Processor Hello World](https://github.com/eval-protocol/remote-rollout-processor-hello-world)
* [Typescript Vercel Example Server](https://github.com/eval-protocol/quickstart)
# Rollout Processors (Getting Started)
Source: https://evalprotocol.io/tutorial/rollout-processors-getting-started
Pick the right rollout processor for your eval and wire it up with minimal boilerplate.
## What is a rollout processor?
A rollout processor is **how Eval Protocol turns input rows into trajectories**:
* Takes a batch of `EvaluationRow`s
* Calls a model, agent, or environment as needed
* Returns updated rows with new messages attached
You choose **one** rollout processor per `@evaluation_test`, and Eval Protocol handles:
* Concurrency and retries
* Logging and cost tracking
* Cleanup of external resources (e.g., MCP servers)
For the full catalog and configuration options, see the [reference page](/reference/rollout-processors). This guide focuses on **choosing and using a processor quickly**.
## Quick decision guide
* **Already have model outputs?** → `NoOpRolloutProcessor`
* **Single chat completion per row?** → `SingleTurnRolloutProcessor`
* **Tools / function calling via MCP?** → `AgentRolloutProcessor`
* **Interactive MCP “gym” environment?** → `MCPGymRolloutProcessor`
* **Already have an in‑production agent/service you want to eval or train?** → `RemoteRolloutProcessor` (see the next page: [Remote Rollout Processor](/tutorial/remote-rollout-processor))
The examples below all assume you are inside an `@evaluation_test`.
## Single-turn model calls
Use `SingleTurnRolloutProcessor` for classic “prompt → answer” tasks:
```python tutorial_single_turn.py theme={null}
import pytest
from typing import List
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import evaluation_test, SingleTurnRolloutProcessor
@pytest.mark.parametrize(
"completion_params",
[
{
"model": "openai/gpt-4o",
"temperature": 0.0,
}
],
)
@evaluation_test(
input_dataset=["dataset.jsonl"],
rollout_processor=SingleTurnRolloutProcessor(),
mode="pointwise",
)
def test_single_turn(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# Each row now has the assistant's reply appended to messages
for row in rows:
# Read row.messages and write row.evaluation_result.score
...
return rows
```
* Good for **QA, grading, and static benchmarks**
* For more knobs (e.g., `extra_body.reasoning_effort`), see the [full reference](/reference/rollout-processors#singleturnrolloutprocessor).
## No-op processor for offline evaluation
If you have **pre-generated model outputs**, use `NoOpRolloutProcessor`:
```python tutorial_noop.py theme={null}
from typing import List
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import evaluation_test, NoOpRolloutProcessor
@evaluation_test(
input_dataset=["offline_answers.jsonl"],
completion_params=[{"model": "not-used-offline"}],
rollout_processor=NoOpRolloutProcessor(),
mode="all",
)
def test_offline(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# rows are passed through unchanged; just score them
for row in rows:
...
return rows
```
This is ideal when you **don’t want Eval Protocol to call any models**.
## Agents and tools via MCP
Use `AgentRolloutProcessor` when your eval requires **tools or function calling**:
```python tutorial_agent.py theme={null}
from typing import List
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import evaluation_test, AgentRolloutProcessor
@evaluation_test(
input_dataset=["tasks.jsonl"],
completion_params=[{"model": "openai/gpt-4o"}],
rollout_processor=AgentRolloutProcessor(),
mcp_config_path="./mcp.config.json",
steps=30,
)
def test_agent(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# Each row reflects the full tool-using conversation
...
return rows
```
* The agent will:
* Call the model with available tools
* Execute any returned tool calls
* Loop until there are no more tools to call or `steps` is reached
## MCP gym environments
Use `MCPGymRolloutProcessor` for **interactive environments** exposed via MCP:
```python tutorial_gym.py theme={null}
from typing import List
from eval_protocol.models import EvaluationRow
from eval_protocol.pytest import evaluation_test, MCPGymRolloutProcessor
@evaluation_test(
input_dataset=["env_tasks.jsonl"],
completion_params=[{"model": "openai/gpt-4o"}],
rollout_processor=MCPGymRolloutProcessor(),
server_script_path="examples/tau2_mcp/server.py",
steps=30,
)
def test_env(rows: List[EvaluationRow]) -> List[EvaluationRow]:
# Each row includes the full trajectory through the environment
...
return rows
```
This is the pattern used by benchmarks like **TauBench** or custom gym‑style environments.
## When to read the full reference
Stay on this page until you need:
* Fine‑grained `RolloutProcessorConfig` usage
* Pydantic AI–specific integrations
* Detailed concurrency and retry behavior
* CLI flags (pytest plugin) for CI tuning
When you do, jump to the [Rollout Processors reference](/reference/rollout-processors) for complete details and edge cases.
# Starting the UI
Source: https://evalprotocol.io/tutorial/ui/getting-started
Reviewing model outputs by hand is an important part of evaluating quality. EP
makes this process simple by offering an easy-to-use, locally hosted UI you can
set up in minutes.
To start the UI, simply run the following command and open your browser to [http://localhost:8000](http://localhost:8000).
```bash CLI theme={null}
ep logs
```
Ensure the `eval-protocol` package is installed so the `ep` CLI is available.
```bash theme={null}
# Add to your project (installs the `ep` CLI in your environment)
uv add eval-protocol
# Verify installation
uv run ep --version
```
```bash theme={null}
# Install or upgrade
pip install -U eval-protocol
# Verify installation
ep --version
```
Once you navigate to the UI at [http://localhost:8000](http://localhost:8000), you will see a table of
evaluation rows that you can click to inspect.
![Table View]()
Whenever you run an `@evaluation_test`—whether from the VSCode Test
Explorer/Debugger or from the CLI via `pytest`—the UI automatically shows
`running` tests and you can watch rollouts live in the chat interface. When a
test finishes, detailed evaluation results appear to the right of the chat.
Tests are stored under a SQLite database on your local device at
`.eval_protocol/logs.db` in the root of your Python project.
To run your tests in CLI, you use the `pytest` command directly.
```bash theme={null}
# Run your tests (UI will reflect live rollouts)
uv run pytest
```
```bash theme={null}
# Run your tests (UI will reflect live rollouts)
pytest
```
You can also run tests in your IDE. Once you have your tests running, you can
open the UI at [http://localhost:8000](http://localhost:8000) to monitor
rollouts live.
## Next Steps
Checkout the [Table View](/tutorial/ui/table) and [Pivot View](/tutorial/ui/pivot) for
more information on how to use the UI.
# Pivot View
Source: https://evalprotocol.io/tutorial/ui/pivot
EP's convenient and locally hosted UI offers a pivot view to help you analyze
your data. If you a familiar with Excel's Pivot Tables, you will feel right at
home. For those who are unfamiliar, pivot tables are an easy way to summarize
and analyze data without having to write formulas or code.
Using a pivot table, you can easily compute aggregate metrics across your data to answer questions like:
* Which model performs best for my application?
* Which prompt performs best for my application?
* How does each model perform on this evaluation and dataset?
* What impact does temperature have on model performance?
* Which tasks in my dataset are the most challenging?
* Is my fine-tuned model outperforming the base model?
* What is the average score across multiple runs?
* Which set of completion parameters yields the best results?
### How to open the pivot view
To know if you are in the pivot view, check that the `Pivot` tab is selected in
the top left corner of the UI.
![Pivot tab]()
### Configuring the pivot table
In the pivot view, you will see a section at the top where you configure your
pivot table.
![Pivot table configuration]()
The pivot table configuration section has five parts:
* **Pivot Rows**: The rows that will be used to group the data.
* **Pivot Columns**: The columns that will be used to group the data.
* **Pivot Values**: The values that will be used to aggregate the data.
* **Pivot Aggregation**: The aggregation function to use for the values.
* **Pivot Filters**: The filters that will be used to filter the data.
### Default configurations
There are three default configurations that you can easily select to help you
get started:
* `Quality (agg_score)`: This will show the average score of the data.
* `Cost (total_cost_dollar)`: This will show the total cost of the data.
* `Speed (duration_seconds)`: This will show the average duration of the data.
![Default configurations]()
### Viewing the data
Once you have configured the pivot table, you can view the data either by chart or table.
#### Chart
A chart will be automatically rendered based on the pivot table you generate.
![Pivot table chart]()
You can also click `Export as Image` to download the chart as an image.
#### Table
You can also see the exact computed values in the table view below the chart.
![Pivot table table]()
You can also click `Export as CSV` to download the table as a CSV file.
### Example (Picking the best model for math problems)
An common example of how to use the pivot view to analyze the data is to compare the performance of different models on a given dataset. For our example, we will compare the performance of
1. gpt-oss-120b (on Fireworks)
2. kimi-k2-instruct (on Fireworks)
3. gpt-4o (on OpenAI)
4. gpt-4o-mini (on OpenAI)
An implementation of this eval is publicly available in EP at
[test\_aime25.py](https://github.com/eval-protocol/python-sdk/blob/main/eval_protocol/benchmarks/test_aime25.py).
To run this eval with 4 different models, you can modify the `completion_params`
parameter in the `evaluation_test` decorator to the following value:
```python focus={7-23} expandable theme={null}
@evaluation_test(
input_dataset=[
"https://huggingface.co/datasets/opencompass/AIME2025/raw/main/aime2025-I.jsonl",
"https://huggingface.co/datasets/opencompass/AIME2025/raw/main/aime2025-II.jsonl",
],
dataset_adapter=aime2025_dataset_adapter,
completion_params=[
{
"extra_body": {"reasoning_effort": "low"},
"model": "fireworks_ai/accounts/fireworks/models/gpt-oss-120b",
},
{
"extra_body": {"reasoning_effort": "low"},
"model": "fireworks_ai/accounts/fireworks/models/kimi-k2-instruct",
},
{
"model": "openai/gpt-4o",
},
{
"model": "openai/gpt-4o-mini",
},
],
rollout_processor=SingleTurnRolloutProcessor(),
aggregation_method="mean",
passed_threshold=None,
num_runs=8,
max_dataset_rows=2,
max_concurrent_rollouts=4,
mode="pointwise",
)
def test_aime25_pointwise(row: EvaluationRow) -> EvaluationRow:
assistant_msgs = [m for m in row.messages if m.role == "assistant"]
content = assistant_msgs[-1].content if assistant_msgs else ""
extracted_text = _extract_boxed_text(content or "")
extracted_int = _normalize_to_int_or_none(extracted_text)
gt_int = _normalize_to_int_or_none(row.ground_truth or "")
is_valid = extracted_int is not None and gt_int is not None
score = 1.0 if (is_valid and extracted_int == gt_int) else 0.0
metrics = {
"exact_match": MetricResult(
score=score,
is_score_valid=is_valid,
reason=(
"Parsed both integers and they matched"
if score == 1.0
else ("Parsed integers did not match" if is_valid else "Failed to parse integer")
),
data={
"extracted_text": extracted_text,
"extracted_int": extracted_int,
"ground_truth_int": gt_int,
},
)
}
row.evaluation_result = EvaluateResult(
score=score,
reason=("Answer correct" if score == 1.0 else "Answer incorrect"),
is_score_valid=is_valid,
metrics=metrics,
)
return row
```
Then looking at the pivot view, after filtering for the `invocation_id` of the
execution, you can see the following chart using the default pivot view
configuration.
![Pivot view after running the AIME 2025 eval]()
We can see that `gpt-oss-120b` out-performs the rest of the models using `"reasoning_effort": "low"`.
# Table View
Source: https://evalprotocol.io/tutorial/ui/table
The table view is the default view when you open the UI. It shows a list of
evaluation rows which you can click to inspect. Every evaluation [row](/specification#row) corresponds
to a single [rollout](/specification#rollout).
### How to open the table view
To know if you are in the table view, check that the `Table` tab is selected in
the top left corner of the UI.
![Table View]()
## Inspecting a row
In the table view, you will see a list of evaluation rows. For each row, you can see:
* **Name**: the test function name
* **Status**: either `running`, `finished`, `stopped`, or `error`
* **Invocation ID**: auto-generated by EP for every [invocation](/specification#invocation)
* **Rollout ID**: auto-generated by EP for every [rollout](/specification#rollout)
* **Model**: the model used for the evaluation
* **Score**: found in `evaluation_result.score`
* **Created**: the timestamp of when the row was created
![Table View]()
To inspect a row, hover over the row you want to inspect and click to expand.
When you expand a row, you can see the trajectory of the rollout as well as
other metadata like evaluation results, IDs, input metadata, and eval metadata.
### Chat Interface
On the left side of an expanded row, you can see the chat interface. This is
where you can see the trajectory of the rollout to inspect the model's responses
and tool calls.
![Chat Interface]()
### Metadata
On the right side of an expanded row, you can see the metadata. This is where
you can see the evaluation results, IDs, input metadata, and eval metadata.
![Metadata]()
## Filtering
Above the table, you can see a section for configuring filters. You can filter based on any attribute of the evaluation row.
![Filtering]()
### Filtering based on Invocation ID
Often times you just want to see the rollouts for a single
[invocation](/specification#invocation). To do this, you can easily click on the
funnel icon next to the invocation ID in the table. This will automatically add
a filter for the invocation ID to the table.
### Custom filters
You can also create custom filters by clicking on the `+ Add Filter Group`
button above the table. Then you can choose to filter by `AND` or `OR` and add
filters to the group by clicking on the `+ Add Filter to Group` button.
## Viewing live rollouts
When it takes a long time to run an eval, it can be helpful to see the live
rollouts so you can track the progress of an eval and catch unexpected errors or problems.
Whenever you run an `@evaluation_test`, the UI automatically shows `running`
tests and you can watch rollouts live in the chat interface. When a test finishes,
detailed evaluation results appear to the right of the chat.
Checkout this example of a test running in VSCode and the UI updating with the
rollout.
## Next Steps
Often times you want to ask questions like "how did the model perform on this
eval across this dataset?" or "which model should I use for my application?".
Creating and running evals helps you answer these questions, but answering these
questions requires computing some aggregate metrics across a set of evaluation rows.
To do this, you can use the [Pivot View](/tutorial/ui/pivot) to pivot the data
and see the data in a different way.