# ParsBench Documentation

> ParsBench provides toolkits for benchmarking LLMs and evaluating AI applications in the Persian language.

ParsBench provides toolkits for benchmarking LLMs and evaluating AI applications (chatbots, agents) in the Persian language. It ships ready-made Persian benchmark tasks with datasets, and an app evaluation layer with Persian-aware normalization (digit scripts, rial/toman amounts, Jalali/Gregorian dates, ZWNJ spacing) for testing your own Persian AI product in CI.

# Getting started

# ParsBench

## Overview

ParsBench is a toolkit for making AI work well in Persian. It has two pillars:

- **App evaluation.** Test *your own* Persian chatbot or agent, whatever framework it is built with, and wire the checks into CI. The Persian-aware matching layer understands digit scripts, rial/toman amounts, Jalali dates, and ZWNJ spacing, so checks that would silently fail in an English-shaped eval harness hold up on Persian text.
- **Model benchmarking.** Evaluate and rank LLMs on ready-made Persian tasks (ParsiNLU, Persian MMLU, FarsTail, Persian Math, and more), each bundled with its dataset, prompt templates, and scorer.

Both share the same class-based API: create an evaluator, hand it the thing under test, read the result. No YAML, no metric registry, one import.

## Show me

Evaluating an app. The app is any function that takes a message and returns an answer:

```
from parsbench.appeval import AppEvaluator, Golden, ToolCall

evaluator = AppEvaluator(goldens=[
    Golden(
        input="بلیط تهران-مشهد برای ۵ مهر ۱۴۰۵ می‌خوام. قیمتش چنده؟",
        tools=[ToolCall("search_flights", date="2026-09-27")],  # Jalali == Gregorian
        contains=["250 هزار تومان"],                            # rials == tomans
        forbidden_tools=["book_flight"],
    ),
])
evaluator.evaluate(my_bot).assert_passed()
```

Benchmarking models:

```
from parsbench.benchmarks import CustomBenchmark
from parsbench.models import OpenAIModel
from parsbench.tasks import ParsiNLUMultipleChoice, PersianMath

benchmark = CustomBenchmark(
    models=[OpenAIModel(...)],
    tasks=[ParsiNLUMultipleChoice, PersianMath],
)
result = benchmark.run(prompt_lang="fa", prompt_shots=[0, 3])
result.show_radar_plot()
```

Start with [Getting Started](https://parsbench.github.io/ParsBench/getting-started/index.md), then go deeper with [App Evaluation](https://parsbench.github.io/ParsBench/app-eval/index.md) or the [Benchmarking tutorial](https://parsbench.github.io/ParsBench/tutorial/models/index.md).

## Key features

- **App evaluation for Persian products**: goldens with tool-call, content, format, budget, and judge-based checks; multi-turn user simulation with an Iranian-user persona; golden generation from your docs; judge calibration.
- **Persian-aware normalization**: «۲۵۰ هزار تومان» matches «۲٬۵۰۰٬۰۰۰ ریال», «۱۴۰۵/۰۷/۰۵» matches `2026-09-27`, and «می‌روم» matches «می روم», in every check.
- **Framework agnostic**: adapters for the OpenAI Agents SDK, LangGraph, Pydantic AI, and Agno, plus a universal OpenTelemetry collector for everything else (CrewAI, LlamaIndex, Google ADK, ...).
- **A local run viewer**: `parsbench view` shows live runs, a goldens × checks matrix, full traces, RTL simulation replay, and run-vs-run diffs.
- **Ready-made Persian benchmarks**: 13 tasks with datasets and prompt templates, benchmarking tools to compare and rank models, radar/bar plots, and a leaderboard builder.
- **Customizable API**: create custom models, tasks, scores, checks, and benchmarks with plain Python classes.

## Motivation

I was trying to fine-tune an open-source LLM for the Persian language and needed a way to measure whether it was actually any good. That led me to [this paper](https://arxiv.org/abs/2404.02403), great work preparing datasets and evaluation methods for testing ChatGPT on Persian, with the code shared in [this repository](https://github.com/Ipouyall/Benchmarking_ChatGPT_for_Persian).

So I built a handy framework that packages various tasks and datasets for evaluating LLMs on Persian, reusing parts of their work (datasets, metrics, basic prompt templates). ParsBench powered the [Open Persian LLM Leaderboard](https://huggingface.co/spaces/ParsBench/leaderboard), and has since grown a second pillar: helping teams that *build* Persian AI products ship them with confidence, not just ranking base models.

## Example notebooks

- Benchmark [Aya](https://huggingface.co/CohereForAI) models:
- Benchmark [Ava](https://huggingface.co/MehdiHosseiniMoghadam) models:
- Benchmark [Dorna](https://huggingface.co/PartAI) models:
- Benchmark [MaralGPT](https://huggingface.co/MaralGPT) models:

Runnable app-evaluation examples, one per framework plus industry scenarios (banking, e-commerce, healthcare, telecom, RAG), live in [`examples/`](https://github.com/ParsBench/ParsBench/tree/main/examples); see the [Examples](https://parsbench.github.io/ParsBench/examples/index.md) page.

## For LLMs

These docs are also published in LLM-friendly form: [`llms.txt`](https://parsbench.github.io/ParsBench/llms.txt) (index) and [`llms-full.txt`](https://parsbench.github.io/ParsBench/llms-full.txt) (everything inlined). Paste either into your assistant to give it the whole API.

## Sponsors

Here are the companies/people who have helped keep this project maintained. If you want to support the project, see the [donation page](https://parsbench.github.io/ParsBench/donation/index.md).

- [AvalAI](https://avalai.ir/): gave us free OpenAI API credit several times through their "AvalAward" program, which funded R&D and benchmarking GPT models.
- [Basalam](https://basalam.com/): voluntarily helped run the benchmarks on open-weight models and build the [ParsBench Leaderboard](https://huggingface.co/spaces/ParsBench/leaderboard).

## Contributing

Contributions are welcome! Please refer to the [contribution guidelines](https://parsbench.github.io/ParsBench/contribution/index.md) for how to get involved.

## License

ParsBench is distributed under the Apache-2.0 license.

## Contact

For support or questions, contact [shahriarshm81@gmail.com](mailto:shahriarshm81@gmail.com) or open an issue on [GitHub](https://github.com/ParsBench/ParsBench/issues).

# Getting Started

## Installation

> **Requires Python ≥ 3.12.** ParsBench 0.2+ targets current library versions (transformers 5, datasets 5, numpy 2), which need Python 3.12+. If you're on Python 3.10/3.11, pin the previous release: `pip install "parsbench==0.1.7"`.

Install ParsBench using pip:

```
pip install parsbench
```

Two optional extras, depending on what you're doing:

- `pip install 'parsbench[test]'` also installs pytest, so you can run golden suites with `parsbench test` in CI.
- The [Persian Math](https://github.com/hendrycks/math) benchmark task additionally needs the Math Equivalence package, installed manually: `pip install git+https://github.com/hendrycks/math.git`

## Pick your path

ParsBench does two different jobs. Pick the one you came for:

- **"I'm building a Persian chatbot/agent and want to test it"** → [evaluate your app](#evaluate-your-app-60-seconds-no-api-key) below, then the [App Evaluation](https://parsbench.github.io/ParsBench/app-eval/index.md) docs.
- **"I have a model and want to know how good it is at Persian"** → [benchmark a model](#benchmark-a-model) below, then the [Benchmarking tutorial](https://parsbench.github.io/ParsBench/tutorial/models/index.md).

## Evaluate your app (60 seconds, no API key)

Your app is any function that takes a user message and returns a string, a `Trace`, or an OpenAI-format message list. Write down what a good answer looks like as `Golden` objects, and evaluate:

```
from parsbench.appeval import AppEvaluator, Golden, ToolCall

def my_bot(message):          # stand-in for your app
    return [
        {"role": "assistant", "tool_calls": [
            {"id": "1", "function": {"name": "search_flights",
                                     "arguments": '{"date": "1405-07-05"}'}}]},
        {"role": "tool", "tool_call_id": "1", "content": "پرواز PY-101"},
        {"role": "assistant", "content": "پرواز ساعت ۸ صبح، قیمت ۲٬۵۰۰٬۰۰۰ ریال"},
    ]

evaluator = AppEvaluator(goldens=[
    Golden(
        input="بلیط تهران-مشهد برای ۵ مهر ۱۴۰۵ می‌خوام. قیمتش چنده؟",
        tools=[ToolCall("search_flights", date="2026-09-27")],  # Jalali == Gregorian
        contains=["250 هزار تومان"],                            # rials == tomans
        forbidden_tools=["book_flight"],
    ),
])
result = evaluator.evaluate(my_bot)
print(result)
```

Every filled `Golden` field switches on its check. There is no separate metric configuration and no YAML. The deterministic checks (tools, contains, format, budgets) run with no API key at all; add a [judge model](https://parsbench.github.io/ParsBench/app-eval/judge/index.md) when you want LLM-judged correctness, faithfulness, and refusal checks too.

Then open the local viewer to browse the run, traces included:

```
parsbench view
```

Continue with the [App Evaluation overview](https://parsbench.github.io/ParsBench/app-eval/index.md).

## Benchmark a model

### Evaluating a pre-trained model

Load a model and tokenizer from HuggingFace, then evaluate on a task, here Persian Math:

```
from transformers import AutoModelForCausalLM, AutoTokenizer

from parsbench.models import PreTrainedTransformerModel
from parsbench.tasks import PersianMath

model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2-72B-Instruct",
    torch_dtype="auto",
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-72B-Instruct")

tf_model = PreTrainedTransformerModel(model=model, tokenizer=tokenizer)

with PersianMath() as task:
    results = task.evaluate(tf_model)
```

### Benchmarking multiple models on multiple tasks

Any OpenAI-compatible API works, so local models via Ollama are one line each:

```
ollama run qwen2
ollama run aya
```

Then benchmark them:

```
from parsbench.benchmarks import CustomBenchmark
from parsbench.models import OpenAIModel
from parsbench.tasks import ParsiNLUMultipleChoice, PersianMath, ParsiNLUReadingComprehension

qwen2_model = OpenAIModel(
    api_base_url="http://localhost:11434/v1/",
    api_secret_key="ollama",
    model="qwen2:latest",
)
aya_model = OpenAIModel(
    api_base_url="http://localhost:11434/v1/",
    api_secret_key="ollama",
    model="aya:latest",
)

benchmark = CustomBenchmark(
    models=[qwen2_model, aya_model],
    tasks=[
        ParsiNLUMultipleChoice,
        ParsiNLUReadingComprehension,
        PersianMath,
    ],
)
result = benchmark.run(
    prompt_lang="fa",
    prompt_shots=[0, 3],
    n_first=100,
    sort_by_score=True,
)
result.show_radar_plot()
```

Continue with the [Benchmarking tutorial](https://parsbench.github.io/ParsBench/tutorial/models/index.md).

## Troubleshooting

**"I can't reach the OpenAI API from Iran."** Every API-based piece of ParsBench (models, the judge, the user simulator) speaks the OpenAI protocol, so point it at any OpenAI-compatible gateway: [AvalAI](https://avalai.ir/), [OpenRouter](https://openrouter.ai/), or a local [Ollama](https://ollama.com/).

```
export OPENAI_BASE_URL=https://api.avalai.ir/v1
export OPENAI_API_KEY=...
```

**Judge checks show as "skipped".** Judge-based checks (`output=`, `context=`, `refuses=`) need a judge model and skip gracefully without one. Set `PARSBENCH_JUDGE` to any model name your gateway serves; see [The Judge](https://parsbench.github.io/ParsBench/app-eval/judge/index.md).

**`import parsbench` fails on `math_equivalence`.** You only need that package for the Persian Math benchmark task; install it as shown under [Installation](#installation).

**Dataset downloads fail.** Benchmark task datasets come from the HuggingFace Hub; set `HF_ENDPOINT` to a mirror if the hub is unreachable from your network.
# App evaluation

# Evaluating your AI app

ParsBench's app evaluation layer tests **your Persian chatbot or agent**, whatever framework it is built with. It is the part of ParsBench you wire into CI so a prompt tweak that breaks tool calls, leaks a forbidden answer, or quotes the wrong price never reaches production.

English-shaped eval harnesses silently fail on Persian apps: they treat «۲۵۰ هزار تومان» and «۲٬۵۰۰٬۰۰۰ ریال» as different answers, «۱۴۰۵/۰۷/۰۵» and `2026-09-27` as different dates, and «می‌روم» / «می روم» / «میروم» as different words. ParsBench's [normalization layer](https://parsbench.github.io/ParsBench/app-eval/normalization/index.md) makes those equivalences hold in every check, on both text answers and tool-call arguments.

Like the rest of ParsBench, the API is class-based: an `AppEvaluator` holds your goldens the way a `Task` holds its dataset, and `evaluate()` takes the thing under test, your app instead of a model.

## Sixty seconds, no API key

Your app is any function that takes a user message and returns a string, a `Trace`, or an OpenAI-format message list:

```
from parsbench.appeval import AppEvaluator, Golden, ToolCall

def my_bot(message):          # stand-in for your app
    return [
        {"role": "assistant", "tool_calls": [
            {"id": "1", "function": {"name": "search_flights",
                                     "arguments": '{"date": "1405-07-05"}'}}]},
        {"role": "tool", "tool_call_id": "1", "content": "پرواز PY-101"},
        {"role": "assistant", "content": "پرواز ساعت ۸ صبح، قیمت ۲٬۵۰۰٬۰۰۰ ریال"},
    ]

evaluator = AppEvaluator(goldens=[
    Golden(
        input="بلیط تهران-مشهد برای ۵ مهر ۱۴۰۵ می‌خوام. قیمتش چنده؟",
        tools=[ToolCall("search_flights", date="2026-09-27")],  # Jalali == Gregorian
        contains=["250 هزار تومان"],                            # rials == tomans
        forbidden_tools=["book_flight"],
    ),
])
result = evaluator.evaluate(my_bot)
print(result)
```

Every filled `Golden` field switches on its check. There is no separate metric configuration and no YAML.

## Two ways in

```
# 1. run-for-me: hand evaluate() your function (sync or async)
result = evaluator.evaluate(bot)

# 2. traces mode: run the app yourself, score what happened
result = evaluator.score_traces([trace_or_message_list])
```

A crash inside your app is reported as a failing `app_error` check on that golden, so one broken case never aborts the suite. Async apps work everywhere, including notebooks and servers with a running event loop.

## Where to go next

- [Goldens & Checks](https://parsbench.github.io/ParsBench/app-eval/goldens/index.md): every `Golden` field, what its check asserts, and the `metrics=` filter.
- [The Judge](https://parsbench.github.io/ParsBench/app-eval/judge/index.md): configuring the judge model for LLM-judged checks, and calibrating it against human labels.
- [Persian Normalization](https://parsbench.github.io/ParsBench/app-eval/normalization/index.md): what the matching layer equates and why, with examples.
- [Framework Integrations](https://parsbench.github.io/ParsBench/app-eval/integrations/index.md): OpenAI Agents SDK, LangGraph, Pydantic AI, Agno, and any OTel-instrumented app.
- [Multi-turn Simulation](https://parsbench.github.io/ParsBench/app-eval/simulation/index.md): an LLM plays an Iranian user against your bot; a judge scores the conversation.
- [Generating Goldens](https://parsbench.github.io/ParsBench/app-eval/generating-goldens/index.md): bootstrap a test suite from your product docs or knowledge base.
- [CI & Regression Tracking](https://parsbench.github.io/ParsBench/app-eval/ci/index.md): pytest integration, concurrency, flaky-agent consistency (`pass^k`), diffs, Langfuse export.
- [The Viewer](https://parsbench.github.io/ParsBench/app-eval/viewer/index.md): `parsbench view`, the local UI over recorded runs.

Runnable end-to-end examples for each framework and several industries live in [`examples/`](https://github.com/ParsBench/ParsBench/tree/main/examples); see the [Examples](https://parsbench.github.io/ParsBench/examples/index.md) page.

# CI and regression tracking

The point of goldens is that they run on every commit. `assert_passed()` raises with the failing checks, which makes goldens plain pytest cases with readable failure messages:

```
import pytest
from parsbench.appeval import AppEvaluator

@pytest.mark.parametrize("golden", GOLDENS, ids=lambda g: g.label)
def test_bot(golden):
    AppEvaluator(goldens=[golden]).evaluate(bot).assert_passed()
```

Run them with `pytest`, or with `parsbench test`, a thin pytest wrapper that ships in the `parsbench[test]` extra and passes its arguments straight through:

```
pip install 'parsbench[test]'
parsbench test tests/ -k booking -x
```

A complete working file is [`examples/ci_with_pytest.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/ci_with_pytest.py). In CI you'll usually also set `PARSBENCH_NO_RECORD=1` so runs don't write into the [local run store](https://parsbench.github.io/ParsBench/app-eval/viewer/index.md).

## Flaky agents: n_runs and pass^k

Agents are sampled, so a golden that passes once may fail the next run. Repeat each golden and measure consistency instead of luck:

```
result = evaluator.evaluate(bot, n_runs=5)
result.pass_hat_k()      # pass^k: probability all k sampled runs pass
result.pass_hat_k(3)     # same, for a subsample of k=3
```

`pass_hat_k` is the unbiased pass^k estimator: for each golden, the probability that `k` randomly chosen runs out of `n_runs` all pass, averaged over goldens. `pass_hat_k()` with no argument uses `k = n_runs`. A bot with `average_score` 0.9 but pass^5 of 0.4 works most of the time and fails somebody every day; the second number is the one your support team feels.

## Concurrency

Independent goldens fan out over threads:

```
result = evaluator.evaluate(bot, prefer_concurrency=True, n_workers=8)
```

Your app and judge callables must then be thread-safe. Most stateful bots are not, which is why this is off by default. `JudgeCalibrator.calibrate` takes the same two arguments.

## Tracking regressions between runs

Results are plain dataclasses with the same conveniences as benchmark results:

```
result.to_pandas()                       # one row per check
result.save("out/")                      # writes out/app_evaluation.jsonl
result = evaluator.evaluate(bot, save_evaluation=True, output_path="out/")  # same
```

`diff()` compares against a saved baseline and prints how each check's mean score moved:

```
result.diff("baseline/app_evaluation.jsonl")
# contains: 0.90 -> 0.80 (-0.10)
```

Unchanged checks print nothing. For a per-golden view of what regressed, use the [viewer's compare page](https://parsbench.github.io/ParsBench/app-eval/viewer/index.md).

A practical CI recipe: save the result on every main-branch run, and in PR builds diff against the latest main artifact before `assert_passed()`, so the log answers "what moved" and not just "something broke".

## Exporting results

`result.to_langfuse()` pushes per-check scores to a Langfuse instance, so eval scores land next to your production traces:

```
export LANGFUSE_HOST=...
export LANGFUSE_PUBLIC_KEY=...
export LANGFUSE_SECRET_KEY=...
```

```
result.to_langfuse()          # returns the created trace's id
```

For anything else, `result.to_dict()` / `to_pandas()` serialize the whole run, and the [viewer](https://parsbench.github.io/ParsBench/app-eval/viewer/index.md) exports JSON, CSV, and Markdown reports.

# Generating goldens

Writing the first thirty goldens by hand is the boring part of adopting any eval tool, and it's where most teams stall. `GoldenGenerator` bootstraps a suite from documentation you already have: your product FAQ, knowledge base, or policy pages.

```
from parsbench.appeval import AppEvaluator, GoldenGenerator

goldens = GoldenGenerator(model="gpt-4.1-mini", adversarial=True).generate("kb/", n=30)
result = AppEvaluator(goldens).evaluate(my_bot)
```

`generate()` accepts a file path, a glob pattern, a directory, or a list of those. Text-like files only (`.txt`, `.md`, `.rst`, `.html`, `.json`). It chunks the documents, asks the generator model for questions a real user would ask about each chunk, and returns ready `Golden` objects.

Three things make the output more than generic QA pairs:

- Questions rotate through registers, formal and colloquial Persian by default. Pass `registers=[...]` to change the rotation.
- `adversarial=True` mixes digit scripts, Jalali dates, and Finglish into some questions, the same traps the [user simulator](https://parsbench.github.io/ParsBench/app-eval/simulation/index.md) plays.
- Each generated golden carries its source chunk as `context=`, so the [faithfulness check](https://parsbench.github.io/ParsBench/app-eval/goldens/index.md) runs automatically: a judge verifies the bot's answer is supported by the document the question came from.

The generator model resolves like the judge does: `model=` argument, then the `PARSBENCH_GENERATOR` env var, then `PARSBENCH_JUDGE`. Unlike judge checks, generation raises without a model, since there is nothing useful to do without one.

## Review before you trust

Generated goldens are a starting point, not ground truth. Skim them, delete the bad ones, and tighten the good ones with explicit `contains=` or `tools=` expectations where you know the right answer. A practical loop:

```
import json
from dataclasses import asdict

goldens = GoldenGenerator(model="gpt-4.1-mini").generate("kb/", n=50)
with open("goldens.json", "w") as f:
    json.dump([asdict(g) for g in goldens], f, ensure_ascii=False, indent=2, default=str)
```

Edit the JSON by hand, then load it back. `AppEvaluator` accepts dicts directly:

```
with open("goldens.json") as f:
    result = AppEvaluator(goldens=json.load(f)).evaluate(my_bot)
```

See [`examples/industry/rag_faq_support.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/industry/rag_faq_support.py) for a full knowledge-base RAG evaluation built this way.

# Goldens and checks

A `Golden` is one expectation: the user message you send, plus what a good response looks like. Only `input` is required. Every other field you fill switches on its corresponding check. Leave a field empty and that check doesn't run.

```
from pydantic import BaseModel
from parsbench.appeval import AppEvaluator, Golden, ToolCall

class FlightAnswer(BaseModel):
    flight_no: str
    price_toman: int

golden = Golden(
    input="بلیط تهران-مشهد برای ۵ مهر ۱۴۰۵ می‌خوام. قیمتش چنده؟",
    name="flight search, Jalali date",
    output="پرواز ساعت ۸ صبح به قیمت ۲۵۰ هزار تومان موجود است.",
    contains=["250 هزار تومان"],
    not_contains=["رزرو شد"],
    tools=[ToolCall("search_flights", date="2026-09-27")],
    forbidden_tools=["book_flight"],
    max_steps=4,
    tags=["booking"],
)
```

## Field reference

| Field                                    | Check             | Passes when                                                         |
| ---------------------------------------- | ----------------- | ------------------------------------------------------------------- |
| `output=`                                | `correctness`     | a judge scores the answer consistent with the reference             |
| `contains=[...]`                         | `contains`        | each needle appears, normalized, money- and date-equivalent         |
| `not_contains=[...]`                     | `not_contains`    | no needle appears (same equivalences)                               |
| `format=Model`                           | `format`          | the output parses and validates against a pydantic schema           |
| `tools=[ToolCall(...)]`                  | `tools`           | expected calls happened; args compared date, then number, then text |
| `forbidden_tools=[...]`                  | `forbidden_tools` | none of these tools were called                                     |
| `context=[...]`                          | `faithfulness`    | a judge finds every claim supported by the context                  |
| `refuses=True`                           | `refusal`         | a judge confirms the request was declined                           |
| `max_steps` / `max_latency` / `max_cost` | `budget:*`        | the trace stayed within the budget                                  |
| `check=fn`                               | `custom`          | your `Trace -> bool \| float` returns truthy / 1.0                  |

Notes on individual fields:

- `name=` labels the golden in reports and in `parsbench view`. Without it, the first 40 characters of the input are used.
- List fields accept a bare string, so `contains="۲۵۰ هزار تومان"` works.
- `tools=` also accepts plain dicts. `ToolCall("search", date="...")` is shorthand for `ToolCall(name="search", arguments={"date": "..."})`. A golden's expected call matches an observed call when the names match and every expected argument is present and equivalent. Extra observed arguments are allowed.
- `check=` is the escape hatch. It receives the full `Trace` (messages, tool calls, latency, cost) and returns a bool or a 0..1 score. Use it for anything the built-in checks don't cover.
- Goldens can also be plain dicts. `AppEvaluator` promotes them via `Golden.from_dict`, which accepts `in`/`out` as short aliases for `input`/`output`. Handy when goldens live in a JSON file.

`output=`, `context=`, and `refuses=` are judged by an LLM and need a [judge model](https://parsbench.github.io/ParsBench/app-eval/judge/index.md). The rest are deterministic and run with no API key.

## Filtering checks with metrics=

By default every check implied by the filled fields runs. `metrics=` restricts or re-modes them:

```
AppEvaluator(goldens, metrics=["tools:strict", "contains"])
```

- Only the named checks run; here, tool calls and `contains`.
- `tools:strict` fails on any tool call that wasn't expected, not just on missing ones.
- Unknown names raise a `ValueError` instead of silently passing. Golden field names (`max_steps`, `refuses`, ...) work as aliases for their check names.

## Reading results

`evaluate()` and `score_traces()` return an `AppEvaluationResult`:

```
result = evaluator.evaluate(my_bot)

result.passed              # True when every check on every golden passed
result.average_score       # mean score over all checks
result.score("contains")   # mean score for one check
print(result)              # readable failure-first summary
```

Each golden's entry holds a `CheckResult` per check with `name`, `passed`, `score`, and `reason`; judge checks carry the judge's reasoning in Persian. `result.to_pandas()` gives one row per check for analysis, and `result.assert_passed()` raises with the failing checks, which makes goldens [plain pytest cases](https://parsbench.github.io/ParsBench/app-eval/ci/index.md).

# Framework integrations

The evaluator doesn't care what your app is built with. It scores a `Trace`, and there are three ways to produce one:

1. Return a string or an OpenAI-format message list from your app function. `evaluate()` builds the trace for you. No integration needed.
1. Use an adapter to convert your framework's run object into a `Trace`.
1. Collect spans via OpenTelemetry for frameworks without an adapter.

## Adapters

Every adapter is duck-typed. Importing it never requires the framework to be installed, so `parsbench` stays dependency-light:

```
from parsbench.integrations import openai_agents, langgraph, pydantic_ai, agno

trace = openai_agents.to_trace(run_result)     # OpenAI Agents SDK RunResult
trace = langgraph.to_trace(state)              # LangGraph graph state
trace = pydantic_ai.to_trace(result)           # Pydantic AI AgentRunResult
trace = agno.to_trace(run)                     # Agno RunOutput
```

Typical use inside `evaluate()`:

```
from agents import Runner
from parsbench.integrations import openai_agents

def app(message):
    run_result = Runner.run_sync(my_agent, message)
    return openai_agents.to_trace(run_result)

result = evaluator.evaluate(app)
```

Or run the app yourself and score afterwards:

```
result = evaluator.score_traces([openai_agents.to_trace(r) for r in run_results])
```

## OpenTelemetry: the universal path

Anything OTel-instrumented (CrewAI, LlamaIndex, Google ADK, instrumented LangChain) feeds through the universal collector. No adapter needed:

```
from parsbench.integrations.otel import TraceCollector

collector = TraceCollector()
tracer_provider.add_span_processor(collector)
...   # run your app
result = evaluator.score_traces([collector.to_trace()])
```

The collector reads the GenAI semantic-convention attributes that these frameworks emit (messages, tool calls, model usage) and assembles them into a `Trace`. See [`examples/with_crewai_otel.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_crewai_otel.py) for a complete CrewAI setup.

## Building a Trace by hand

When nothing above fits (a bot behind an HTTP API, logs replayed from production), construct traces directly:

```
from parsbench.appeval import Message, ToolCall, Trace

trace = Trace(
    messages=[
        Message(role="user", content="..."),
        Message(role="assistant", tool_calls=[ToolCall("search_flights", date="1405-07-05")]),
        Message(role="assistant", content="..."),
    ],
    final_output="...",
    latency=1.9,          # feeds the max_latency budget check
    cost=0.0004,          # feeds the max_cost budget check
)

result = evaluator.score_traces([trace])
```

`Trace.from_messages([...])` builds one from OpenAI-format dicts, which is also what `evaluate()` does when your app returns a message list.

## Google ADK

For Google ADK agents, collect via OTel as above. The `parsbench.integrations.adk` module additionally provides `persian_response_match`, a drop-in replacement for ADK's own `response_match_score`. ADK's default is ROUGE-1 with an ASCII-oriented tokenizer, which collapses on Perso-Arabic script; the replacement computes token-level F1 over Persian-normalized tokens.

## Langfuse

`result.to_langfuse()` pushes per-check scores to a Langfuse instance for dashboarding alongside your production traces. See [CI & regression tracking](https://parsbench.github.io/ParsBench/app-eval/ci/#exporting-results).

# The judge

Deterministic checks always run. Judge checks (`output=`, `context=`, `refuses=`) need a judge model, and they **skip gracefully** without one, so a suite with no API key still runs its deterministic checks.

## Configuration

The quickest setup is environment variables:

```
export PARSBENCH_JUDGE=gpt-4.1-mini      # any OpenAI-compatible model name
export OPENAI_BASE_URL=...               # AvalAI, OpenRouter, Ollama, ...
export OPENAI_API_KEY=...
```

When the judge should use different credentials than your app (common when the app under test and the judge run through different gateways), use the judge-specific variables. They win over the `OPENAI_*` ones:

```
export PARSBENCH_JUDGE_BASE_URL=...
export PARSBENCH_JUDGE_API_KEY=...
```

`judge=` on the evaluator overrides the environment and accepts three forms:

```
# 1. a model name string, resolved against the env vars above
AppEvaluator(goldens, judge="gpt-4.1-mini")

# 2. any parsbench Model
from parsbench.models import OpenAIModel
AppEvaluator(goldens, judge=OpenAIModel(api_base_url=..., api_secret_key=..., model=...))

# 3. a plain callable, prompt -> str  (your own client, a local model, a stub in tests)
AppEvaluator(goldens, judge=lambda prompt: my_client.complete(prompt))
```

The built-in client retries transient failures and bounds each call. `PARSBENCH_MAX_RETRIES` and `PARSBENCH_TIMEOUT` override the defaults of 5 retries and 120 seconds.

## The prompts

Judge prompts are authored in Persian, because a judge reasoning about Persian text in Persian makes fewer normalization mistakes than one translating on the fly. They're importable, so you can read or edit the rubrics:

```
from parsbench.appeval import prompts_fa
print(prompts_fa.CORRECTNESS)
```

The judge's verdict and its reasoning land in each `CheckResult.reason`, and `parsbench view` shows them one click away from the checks matrix.

## Calibrating the judge

An LLM judge is a measurement instrument, and you should know its error rate before you trust it. `JudgeCalibrator` measures judge-vs-human agreement on a labeled sample:

```
from parsbench.appeval import Golden, JudgeCalibrator

calibration = JudgeCalibrator(judge="gpt-4.1-mini").calibrate(
    [
        {"golden": Golden(input="...", output="..."), "output": "...", "human": True},
        {"golden": Golden(input="...", output="..."), "output": "...", "human": False},
        # ... 30-50 labeled cases is a reasonable start
    ],
    prefer_concurrency=True, n_workers=8,
)
print(calibration)     # agreement rate, Cohen's kappa, readable disagreements
```

Each item is a golden, the app output the judge will score, and your human verdict. The printed report lists every disagreement so you can see whether the judge is too strict, too lenient, or confused by a specific phrasing. Kappa near zero means the judge agrees with humans no more than chance would; fix the rubric or pick a stronger judge model before publishing scores.

# Persian normalization

This layer is the reason ParsBench exists as an app-eval tool. A generic harness compares strings; Persian answers rarely lose on meaning, they lose on encoding. The same answer can arrive in three digit scripts, two calendars, two currency units, and with or without ZWNJ. ParsBench applies one equivalence chain under every text check and every tool-argument comparison, always on, with nothing to configure.

## What is equated

**Codepoints and digits.** Arabic twins map to their Persian forms (`ي` → `ی`, `ك` → `ک`, `ة` → `ه`), and all three digit scripts unify: `۱۲۳` (Persian), `١٢٣` (Arabic-Indic), and `123` compare equal. Thousands separators (`٬` and `,`) drop out, so `۲٬۵۰۰٬۰۰۰` equals `2500000`. The Arabic decimal separator reads as a point: `۲٫۵` equals `2.5`.

**ZWNJ and spacing.** ZWNJ reads as a space and whitespace collapses, and a space inside a needle also matches the fully joined form. «می‌روم», «می روم», and «میروم» all match each other. Needles without spaces never merge words, so short needles can't false-positive across word boundaries.

**Money.** Amounts parse into rials before comparing, across scale words and units. «۲۵۰ هزار تومان», «۲٬۵۰۰٬۰۰۰ ریال», and «۲/۵ میلیون تومان» all state the same amount, including compound forms like «۲ میلیون و ۵۰۰ هزار» and both Persian decimal notations (`۲٫۵` and `۲/۵`). When one side has no currency word, either the rial or the toman reading is accepted. A bare number in the text only matches at the exact rial value, so order ids and phone numbers can't satisfy a money check.

**Dates.** Numeric dates parse in either calendar; a year below 1600 reads as Jalali. So a tool argument of `"1405-07-05"` equals `"2026-09-27"`, and `contains=["1405/07/05"]` matches an answer that states the Gregorian date. Times after the date (`2026-09-27T08:00`) are tolerated. Persian month names («۵ مهر») are not parsed by this layer; the [user simulator](https://parsbench.github.io/ParsBench/app-eval/simulation/index.md) uses them as a trap precisely because most bots must convert them to a numeric date before calling a tool, which this layer then checks.

## Where it applies

- `contains=` and `not_contains=`: each needle tries, in order, normalized substring, then money equivalence, then date equivalence.
- Tool arguments: each expected argument compares as date, then number, then normalized text.
- Symmetrically. `not_contains=["رزرو شد"]` also catches «رزرو شد» with odd spacing, and a forbidden amount is caught in any unit.

## Using it directly

The functions are importable if you want the same behavior in your own checks or tests:

```
from parsbench.appeval.normalize import (
    normalize,             # canonical form of a string
    contains_normalized,   # substring check under normalization
    numbers_equal,         # '۲۵۰ هزار تومان' == '2500000 ریال' -> True
    amount_in,             # is this amount stated anywhere in the text?
    dates_equal,           # '1405-07-05' == '2026-09-27' -> True
    date_in,               # is this date mentioned, either calendar?
    values_equal,          # the tool-argument chain: date -> number -> text
)

amount_in("قیمت ۲٬۵۰۰٬۰۰۰ ریال است", "250 هزار تومان")   # True
```

The module is dependency-free by design (no hazm import). It runs inside every check, so it stays a handful of table lookups and small regexes.

# Multi-turn simulation

Single-turn goldens catch a lot, but real users don't speak in goldens. They open with taarof, quote prices in the wrong unit, switch to Finglish mid-conversation, and change their mind. `SimulationEvaluator` drives your bot with an LLM playing an Iranian user, then judges the finished conversation against the goal.

```
from parsbench.appeval import SimulationEvaluator

evaluator = SimulationEvaluator(
    goal="خرید بسته اینترنت یک‌ماهه و دانستن قیمت آن",
    user="محاوره‌ای+toman_rial_confusion+finglish_switch",
    criteria=["قیمت به کاربر اعلام شود"],
    simulator_model="gpt-4.1-mini",           # the user simulator
    judge="gpt-4.1-mini",
)
result = evaluator.evaluate(my_bot)           # fn(message) or fn(message, history)
```

Your bot is a callable again. `fn(message)` works for stateful bots that track their own history; `fn(message, history)` receives the conversation so far as OpenAI-format dicts.

Simulation needs two models: the simulator (falls back to the `PARSBENCH_SIMULATOR` env var, then `PARSBENCH_JUDGE`) and the [judge](https://parsbench.github.io/ParsBench/app-eval/judge/index.md) (`PARSBENCH_JUDGE`).

## The simulated user

`user=` is a string mixing a register with trap names, joined by `+`. Free text in the string becomes extra instructions for the simulated user, so `"محاوره‌ای+typos+اهل اصفهان است"` works. The named traps live in `parsbench.appeval.TRAPS`:

| Trap                   | The simulated user will                                                  |
| ---------------------- | ------------------------------------------------------------------------ |
| `taarof_opening`       | open with taarof and hold back the real request in the first message     |
| `toman_rial_confusion` | quote amounts in toman even when the system looks rial-based             |
| `jalali_date`          | give dates in Jalali with Persian month names («۵ مهر»)                  |
| `finglish_switch`      | write some mid-conversation messages in Finglish («merci, hamin khoobe») |
| `typos`                | make natural typos now and then                                          |
| `impatient`            | complain when answers are slow or vague                                  |
| `iran_formats`         | use Iranian phone/address formats (۰۹۱۲…، خیابان/کوچه/پلاک)              |

For finer control, pass a `PersianUser(style=..., persona=..., traps=[...])` object instead of the string.

## Multiple scenarios

`goal=` is shorthand for a single scenario. A real suite is a list of `ConversationGolden` objects:

```
from parsbench.appeval import ConversationGolden, SimulationEvaluator

evaluator = SimulationEvaluator(
    goldens=[
        ConversationGolden(
            goal="خرید بسته اینترنت یک‌ماهه",
            scenario="کاربر قبلاً یک بسته دارد که هفتهٔ بعد منقضی می‌شود.",
            expected_outcome="بسته مناسب پیشنهاد و قیمت اعلام شود.",
            criteria=["قیمت به کاربر اعلام شود", "بدون تأیید کاربر خریدی انجام نشود"],
            max_turns=8,
        ),
        ConversationGolden(goal="لغو اشتراک", criteria=["فرایند لغو کامل توضیح داده شود"]),
    ],
    user="محاوره‌ای+impatient",
)
```

Each criterion is judged as its own check, so the result shows exactly which behavior failed. `max_turns` caps the conversation per golden; `evaluate(app, max_turns=...)` overrides it for a run. Hitting the turn cap does not fail a conversation the goal judge scored as successful. A chatty simulator that never stops talking shouldn't punish the app.

## Reading the run

The result is the same `AppEvaluationResult` as everywhere else, so `assert_passed()`, `to_pandas()`, and `n_runs=` for [consistency scoring](https://parsbench.github.io/ParsBench/app-eval/ci/index.md) all apply. In [`parsbench view`](https://parsbench.github.io/ParsBench/app-eval/viewer/index.md), simulation runs get a replay page that shows the conversation as an RTL chat with the goal and active traps alongside.

# The viewer: parsbench view

Every `evaluate()` and `score_traces()` call records its run, full traces included, into a project-local `.parsbench/` store. The store is self-gitignored plain files (`run.json` + `events.jsonl` per run), and the viewer is a local web UI over it:

```
parsbench view
```

By default it serves `127.0.0.1:1404` (walking upward if the port is busy) and opens your browser. Options:

```
parsbench view path/to/project     # a .parsbench store, or a directory containing one
parsbench view --port 8080
parsbench view --host 0.0.0.0      # e.g. to view from another machine
parsbench view --no-open
```

## What you get

- Live runs. Runs stream into the UI while they execute; finished runs stay as the archive.
- A checks matrix: goldens × checks, failure-first, with the judge's Persian reasons one click away.
- Trace detail: messages, tool calls (arguments, results, errors), latency and steps, plus per-run tabs when `n_runs > 1`.
- Simulation replay: the conversation as an RTL chat, with the goal and traps in play.
- Compare: pick a baseline run and see which goldens regressed and how each check's mean moved.
- Export: any run as JSON (full traces), CSV (one row per check, Excel-safe UTF-8), or a paste-ready Markdown report.
- Optional charts: score per check, and the app's score history across runs.
- Dark and light themes, toggled in the top bar.

The viewer has zero extra dependencies and no build step. It ships inside the `parsbench` package.

## Recording

Recording is on by default. Turn it off per call or globally:

```
evaluator.evaluate(bot, record=False)
```

```
export PARSBENCH_NO_RECORD=1     # e.g. in CI
```

Delete `.parsbench/` whenever you like. It is only the viewer's data; nothing else reads it.
# Benchmarking

# Advanced tutorial

This section is for implementing your own tasks, or using the framework's building blocks (scores, data loaders, prompt templates) for other purposes.

## Scores

Scores measure how good a completion is compared to the expected answer.

### Available scores

| Score Name            | Description                                                              |
| --------------------- | ------------------------------------------------------------------------ |
| Exact Match           | `1` if the completion and target are equal, otherwise `0`.               |
| English Sentence Bleu | Bleu n-gram score with NLTK English word tokenizer. Between `0` and `1`. |
| Persian Sentence Bleu | Bleu n-gram score with Hazm Persian word tokenizer. Between `0` and `1`. |
| English Rouge         | Rouge score with NLTK English word tokenizer. Between `0` and `1`.       |
| Persian Rouge         | Rouge score with Hazm Persian word tokenizer. Between `0` and `1`.       |

### Make your own score

Write a plain function and wrap it with `wrap_scorer`:

```
from parsbench.scores.base import wrap_scorer

@wrap_scorer
def my_exact_match(completion: str, target: str) -> float:
    return float(completion.strip() == target.strip())
```

The function's name becomes the score's name in results.

## Data loaders

A data loader loads the dataset a task evaluates on. All three load from a local path or a URL.

### JSONLine

`JSONLineDataLoader` reads jsonlines files (`.jsonl`):

```
from parsbench.tasks.base import JSONLineDataLoader

data_loader = JSONLineDataLoader(data_path="dataset.jsonl")
data = data_loader.load()
```

### CSV

`CSVDataLoader` reads CSV files:

```
from parsbench.tasks.base import CSVDataLoader

data_loader = CSVDataLoader(data_path="dataset.csv")
data = data_loader.load()
```

### HuggingFace

`HuggingFaceDataLoader` uses HuggingFace's `datasets` library to load from disk or download from the Hub:

```
from parsbench.tasks.base import HuggingFaceDataLoader

data_loader = HuggingFaceDataLoader(
    data_path="persiannlp/parsinlu_entailment",
    split="validation",
)
data = data_loader.load()
```

## Prompt templates

`PromptTemplate` defines the prompt for each language, shot templates, shot examples, and variable mappings.

### With a shot template

A sentiment-analysis template where few-shot examples are rendered from the dataset:

```
FA_TEMPLATE = """
جمله زیر نظر یک شخص است. این جمله به زبان فارسی است. بار یا احساس موجود در این جمله را شناسایی کن.
پاسخ‌ های ممکن حالت‌های روبرو هستند:
SAD
NEUTRAL
HAPPY

فقط کلمه مربوط به احساس نظر داده شده را خروجی بده.

{example_shots}

نظر: {review}
احساس:
"""

FA_SHOT_TEMPLATE = """
نظر: {review}
احساس: {label}
"""
```

```
from parsbench.tasks.base import PromptTemplate

prompt_template = PromptTemplate(
    language_templates={"fa": FA_TEMPLATE},
    prompt_shot_templates={"fa": FA_SHOT_TEMPLATE},
    prompt_variables_mapping={"review": "review"},
    target_variables_mapping={"label": "label"},
)

prompt = prompt_template.get_prompt(
    prompt_lang="fa",
    data={"review": "غذا خیلی بد بود", "label": "SAD"},
    n_shots=3,
    sample_data=[
        {"review": "خوشمزه بود ممنونم", "label": "HAPPY"},
        {"review": "غذا خوب بود فقط کاش زودتر می‌رسید.", "label": "NEUTRAL"},
        {"review": "نوشابه گرم بود. پیتزا هم خیلی بد مزه بود.", "label": "SAD"},
    ],
)
```

The variable mappings translate between prompt placeholders and dataset columns: `prompt_variables_mapping={"review": "review"}` fills `{review}` in the template from the `review` column, and `target_variables_mapping` does the same for the expected answer.

### With static shot examples

For complicated tasks where you want hand-written few-shot examples (for instance chain-of-thought prompting), use static shot examples instead of a shot template:

```
from parsbench.tasks.base import PromptTemplate

prompt_template = PromptTemplate(
    language_templates={"fa": FA_TEMPLATE},
    prompt_shot_examples={"fa": {1: FA_1_SHOT, 3: FA_3_SHOT, 5: FA_5_SHOT}},
)
prompt = prompt_template.get_prompt(n_shots=5, ...)
```

### Load templates from files

`LazyLoadTemplates` reads templates from text files on first use, which keeps long prompts out of your Python code:

```
from parsbench.tasks.base import PromptTemplate, LazyLoadTemplates

prompt_template = PromptTemplate(
    language_templates=LazyLoadTemplates(
        fa="fa_math.txt",
        en="en_math.txt",
    ),
    ...
)
```

### Constant prompt variables

To fill a placeholder with a fixed value rather than a dataset column, use `ConstantPromptVariable`:

```
from parsbench.tasks.base import PromptTemplate, ConstantPromptVariable

prompt_template = PromptTemplate(
    language_templates={"fa": FA_TEMPLATE},
    prompt_shot_templates={"fa": FA_SHOT_TEMPLATE},
    prompt_variables_mapping={
        "input": "input",
        "first_name": ConstantPromptVariable("شهریار"),
    },
    target_variables_mapping={"label": "label"},
)
```

## Tasks

The task is the primary unit of the framework: a batteries-included evaluator that runs the whole pipeline from loading data to scoring.

### Task data provider

Each task carries its dataset. `task.get_data()` returns it:

```
from parsbench.tasks import ParsiNLUEntailment

with ParsiNLUEntailment() as task:  # the context manager loads the data
    data = task.get_data()
```

### Task match generator

A `TaskMatch` holds a prompt, the target answer, and (once generated) the model's completion and its score. Fresh matches have `completion` and `score` set to `None`:

```
from parsbench.tasks import ParsiNLUEntailment

with ParsiNLUEntailment() as task:
    matches = task.generate_matches(prompt_lang="fa", n_shots=0, n_first=100)
```

### Generate completions and score

You can drive the pipeline step by step, which is useful when you want to inspect or modify matches between steps:

```
from parsbench.tasks import ParsiNLUEntailment

with ParsiNLUEntailment() as task:
    matches = task.generate_matches(prompt_lang="fa", n_shots=0, n_first=100)
    model.generate_completions(matches)   # the model fills in completions
    task.score_matches(matches)           # the task's scorer fills in scores
```

### Make your own task

For a task with your own dataset, prompts, and scoring, inherit `Task` (or one of the existing tasks). Put prompt templates in text files and load them with `LazyLoadTemplates`:

```
from parsbench.scores.base import Scorer, wrap_scorer
from parsbench.tasks.base import (
    HuggingFaceDataLoader,
    LazyLoadTemplates,
    PromptTemplate,
    Task,
    TaskCategory,
    TaskMatchGroup,
)

@wrap_scorer
def my_custom_score(completion: str, target: str) -> float:
    return float(completion.strip() == target.strip())

class CustomTask(Task):
    task_name: str = "Custom Task"
    task_category: TaskCategory = TaskCategory.REASONING

    data_loader: HuggingFaceDataLoader = HuggingFaceDataLoader(
        data_path="org/custom_dataset",
        split="test",
    )
    data_target_key: str = "target"

    prompt_template: PromptTemplate = PromptTemplate(
        language_templates=LazyLoadTemplates(
            en="path/to/en_template.txt",
            fa="path/to/fa_template.txt",
        ),
        prompt_shot_templates=LazyLoadTemplates(
            en="path/to/en_shot_template.txt",
            fa="path/to/fa_shot_template.txt",
        ),
        prompt_variables_mapping={"prompt_variable1": "variable1", "prompt_variable2": "variable2"},
        target_variables_mapping={"prompt_target": "target"},
    )

    scorer: Scorer = my_custom_score

    def score_matches(self, matches: TaskMatchGroup) -> TaskMatchGroup:
        matches.format_completions(
            lambda c: c.strip().strip("'").lower()
        )
        return super().score_matches(matches)

    def get_overall_score(cls, matches: TaskMatchGroup) -> float:
        return sum(match.score for match in matches) / len(matches)
```

Any data loader, prompt template, and scorer combination works.

#### Sub-tasks

If your dataset covers several categories, declare them and ParsBench reports a score per category:

```
class CustomTask(Task):
    ...
    sub_task_key: str = "category"  # dataset column that holds the sub task
    sub_tasks: list[str] = ["math_and_logic", "common_knowledge", "literature"]
    ...
```

Both evaluation and benchmarks can then run a subset:

```
with CustomTask() as task:
    results = task.evaluate(..., sub_tasks=["math_and_logic"])

# or inside a benchmark
benchmark = CustomBenchmark(
    ...,
    tasks=[
        PersianMath,
        CustomTask.select_sub_tasks(["math_and_logic"]),
    ],
)
```

# Benchmarks

A benchmark evaluates multiple models on multiple tasks and compares their scores. It is a loop over `task.evaluate(model)` plus a result object that knows how to rank, pivot, plot, merge, and save.

## Custom benchmark

`CustomBenchmark` takes your models and tasks. Interfaces mix freely; here a local transformers checkpoint runs against an API model:

```
from transformers import AutoModelForCausalLM, AutoTokenizer

from parsbench.benchmarks import CustomBenchmark
from parsbench.models import OpenAIModel, PreTrainedTransformerModel
from parsbench.tasks import ParsiNLUMultipleChoice, PersianMath, ParsiNLUReadingComprehension

# Create models
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2-72B-Instruct",
    torch_dtype="auto",
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-72B-Instruct")
qwen2_model = PreTrainedTransformerModel(model=model, tokenizer=tokenizer)

aya_model = OpenAIModel(
    api_base_url="http://localhost:11434/v1/",
    api_secret_key="ollama",
    model="aya:latest",
)

# Run the benchmark
benchmark = CustomBenchmark(
    models=[qwen2_model, aya_model],
    tasks=[
        ParsiNLUMultipleChoice,
        ParsiNLUReadingComprehension,
        PersianMath,
    ],
)
result = benchmark.run(
    prompt_lang="fa",
    prompt_shots=[0, 3],
    n_first=100,
    sort_by_score=True,
)
```

`run()` accepts the same evaluation parameters as `task.evaluate()` (`n_first`, `skip_existing_matches`, `prefer_concurrency`, `n_workers`, the save flags), plus `sort_by_score=` to rank models by average score in the result.

There is also `ParsiNLUBenchmark`, a `CustomBenchmark` subclass hard-wired to the ParsiNLU task set.

## Full benchmark

To benchmark on every task in the framework, use `load_all_tasks`:

```
from parsbench.benchmarks import CustomBenchmark
from parsbench.models import OpenAIModel
from parsbench.tasks.utils import load_all_tasks

aya_model = OpenAIModel(
    api_base_url="http://localhost:11434/v1/",
    api_secret_key="ollama",
    model="aya:latest",
)

benchmark = CustomBenchmark(
    models=[aya_model],
    tasks=load_all_tasks(),
)
result = benchmark.run(
    prompt_lang="fa",
    prompt_shots=[0, 3],
    n_first=100,
    sort_by_score=True,
)
```

## Benchmark result

`BenchmarkResult` holds every evaluation result for every model. Convert it to a pandas DataFrame with `to_pandas()`; `to_pandas(pivot=True)` gives the models-as-columns pivot table:

```
print(result.to_pandas(pivot=True))
```

Output:

```
                                                                                     score
model_name                                                                     qwen2:latest
n_shots                                                                                   0         3
task_category task_name                      sub_task         score_name
classic       ParsiNLU Reading Comprehension NaN              Common Tokens         0.46231  0.588274
knowledge     ParsiNLU Multiple Choice       common_knowledge Exact Match           0.30000  0.000000
                                             literature       Exact Match           0.20000  0.428571
                                             math_and_logic   Exact Match           0.60000  0.285714
math          Persian Math                   NaN              Math Equivalence      0.00000  0.142857
```

It renders best in a Jupyter notebook.

### Plots

`show_radar_plot()` compares models across task categories; `show_bar_plot()` shows the same data as bars:

```
result.show_radar_plot()
result.show_bar_plot()
```

### Saving results

Set `save_matches`, `save_evaluation`, and `save_benchmark` to write matches, per-task evaluations, and the combined benchmark file during the run:

```
benchmark = CustomBenchmark(
    models=[aya_model, qwen2_model],
    tasks=[PersianMath, FarsTailEntailment],
)
result = benchmark.run(
    prompt_lang="fa",
    prompt_shots=[0, 5],
    n_first=100,
    save_matches=True,
    save_evaluation=True,
    save_benchmark=True,
    output_path="results",
    sort_by_score=True,
)
```

The output directory structure:

```
results
├── aya:latest
│   ├── FarsTail_Entailment
│   │   ├── evaluation.jsonl
│   │   ├── matches_0_shot.jsonl
│   │   └── matches_5_shot.jsonl
│   └── Persian_Math
│       ├── evaluation.jsonl
│       ├── matches_0_shot.jsonl
│       └── matches_5_shot.jsonl
├── qwen2:latest
│   ├── FarsTail_Entailment
│   │   ├── evaluation.jsonl
│   │   ├── matches_0_shot.jsonl
│   │   └── matches_5_shot.jsonl
│   └── Persian_Math
│       ├── evaluation.jsonl
│       ├── matches_0_shot.jsonl
│       └── matches_5_shot.jsonl
└── benchmark.jsonl
```

### Rebuilding and merging results

Three helpers cover the "I ran benchmarks last week and want to work with them now" cases:

```
from parsbench.benchmarks import BenchmarkResult, merge_benchmark_results

# rebuild a result from saved matches files (rescore=True re-runs the scorers)
result = BenchmarkResult.from_matches_files("results/", rescore=False)

# combine runs done at different times / on different machines
merged = merge_benchmark_results([result_a, result_b], sort=True)
```

`merge_benchmark_results` drops duplicate model names by default; pass `keep_duplicates=True` to keep them all.

### Building a leaderboard

`build_leaderboard_from_benchmark` writes the request/result file layout used by the [ParsBench Leaderboard](https://huggingface.co/spaces/ParsBench/leaderboard) HuggingFace space:

```
from parsbench.benchmarks import build_leaderboard_from_benchmark

build_leaderboard_from_benchmark(result, "leaderboard_data/")
```

# Models

Models are the interfaces that generate completions from LLMs. A benchmark or task doesn't care where the completion comes from; it calls the model interface and scores what comes back. Three interfaces ship with ParsBench, and writing your own takes four methods.

## OpenAIModel

`OpenAIModel` speaks the OpenAI chat-completions protocol, which in practice means most of the ecosystem: OpenAI itself, gateways like [AvalAI](https://avalai.ir/) and [OpenRouter](https://openrouter.ai/), and local runtimes like [Ollama](https://ollama.com/) and vLLM.

```
from parsbench.models import OpenAIModel

model = OpenAIModel(
    api_base_url="https://api.openai.com/v1/",
    api_secret_key="{SECRET_KEY}",
    model="gpt-4.1",
)
```

A local model through Ollama is the same interface with a different base URL:

```
ollama run llama3
```

```
model = OpenAIModel(
    api_base_url="http://localhost:11434/v1/",
    api_secret_key="ollama",
    model="llama3:latest",
)
```

Useful optional parameters, shared with `AnthropicModel`:

- `instruction_prompt=` replaces the default system prompt.
- `completion_parameters=` passes sampling parameters through to the API, e.g. `{"temperature": 0}`.
- `retry_on_ratelimit=True` with `cooldown_interval=` (seconds, default 10) and `max_retries=` (default 1) retries rate-limited calls instead of failing the run.

## AnthropicModel

`AnthropicModel` is the same idea for Anthropic-style APIs:

```
from parsbench.models import AnthropicModel

model = AnthropicModel(
    api_secret_key="{SECRET_KEY}",
    model="claude-sonnet-4-5",
)
```

`api_base_url=` is optional and lets you point it at an Anthropic-compatible gateway.

## PreTrainedTransformerModel

`PreTrainedTransformerModel` wraps a `PreTrainedModel` from the [transformers](https://huggingface.co/docs/transformers) library, so you can evaluate a checkpoint directly, including one you just fine-tuned, without serving it behind an API:

```
from transformers import AutoModelForCausalLM, AutoTokenizer

from parsbench.models import PreTrainedTransformerModel
from parsbench.tasks import PersianMath

model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2-72B-Instruct",
    torch_dtype="auto",
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-72B-Instruct")

tf_model = PreTrainedTransformerModel(model=model, tokenizer=tokenizer)

with PersianMath() as task:
    results = task.evaluate(tf_model)
```

## Create your own interface

Inherit the `Model` abstract class and implement four methods:

```
from parsbench.models import Model

class CustomModel(Model):
    @property
    def model_name(self) -> str:
        return "My Custom Model"

    def get_prompt_completion(self, prompt: str) -> str:
        # call your API / model here
        return f"Response to {prompt}"

    def prompt_formatter(self, prompt: str) -> str | list[dict]:
        return prompt  # or wrap into chat messages

    def completion_formatter(self, completion: str) -> str:
        return completion.strip().replace("'", "")
```

- `model_name` labels the model in results and output directories.
- `get_prompt_completion` does the actual call.
- `prompt_formatter` turns the task's prompt string into whatever your API expects (a raw string, or an OpenAI-style message list).
- `completion_formatter` cleans the raw completion before scoring, stripping quotes, whitespace, or chatter that would break exact-match scores.

# Tasks

A task evaluates model responses on one dataset. It ships with the data, a prompt template per language, and a scorer, so evaluating a model is: build prompts from the data, get the model's completions, score them against the targets.

## Available tasks

| Task Name                             | Score Name                  | Dataset                                                                               |
| ------------------------------------- | --------------------------- | ------------------------------------------------------------------------------------- |
| ParsiNLU Sentiment Analysis           | Exact Match (F1)            | [ParsiNLU](https://huggingface.co/datasets/persiannlp/parsinlu_sentiment)             |
| ParsiNLU Entailment                   | Exact Match (F1)            | [ParsiNLU](https://huggingface.co/datasets/persiannlp/parsinlu_entailment)            |
| ParsiNLU Machine Translation En -> Fa | Bleu                        | [ParsiNLU](https://huggingface.co/datasets/persiannlp/parsinlu_translation_en_fa)     |
| ParsiNLU Machine Translation Fa -> En | Bleu                        | [ParsiNLU](https://huggingface.co/datasets/persiannlp/parsinlu_translation_fa_en)     |
| ParsiNLU Multiple Choice              | Exact Match (Accuracy)      | [ParsiNLU](https://github.com/persiannlp/parsinlu)                                    |
| ParsiNLU Reading Comprehension        | Common Tokens (F1)          | [ParsiNLU](https://huggingface.co/datasets/persiannlp/parsinlu_reading_comprehension) |
| Persian NER                           | NER Exact Match (F1)        | [PersianNER](https://github.com/HaniehP/PersianNER)                                   |
| Persian Math                          | Math Equivalence (Accuracy) | [Source](https://github.com/Ipouyall/Benchmarking_ChatGPT_for_Persian)                |
| ConjNLI Entailment                    | Exact Match (F1)            | [Source](https://github.com/Ipouyall/Benchmarking_ChatGPT_for_Persian)                |
| Persian MMLU (Khayyam Challenge)      | Exact Match (Accuracy)      | [Khayyam Challenge](https://huggingface.co/datasets/raia-center/khayyam-challenge)    |
| FarsTail Entailment                   | Exact Match (F1)            | [FarsTail](https://github.com/dml-qom/FarsTail)                                       |
| Persian News Summary                  | Rouge                       | [PNSummary](https://huggingface.co/datasets/HooshvareLab/pn_summary)                  |
| XL-Sum                                | Rouge                       | [XLSum](https://huggingface.co/datasets/csebuetnlp/xlsum)                             |

Import any of them from `parsbench.tasks`, or get instances of all of them with `parsbench.tasks.utils.load_all_tasks()`.

## Evaluation

The evaluation process has 6 steps:

1. Loading data
1. Loading the prompt template
1. Generating matches (prompt-answer pairs)
1. Generating completions
1. Scoring completions
1. Storing the result (optional)

`evaluate()` runs all of them:

```
from parsbench.models import OpenAIModel
from parsbench.tasks import ParsiNLUMultipleChoice

model = OpenAIModel(
    api_base_url="http://localhost:11434/v1/",
    api_secret_key="ollama",
    model="qwen2:latest",
)

with ParsiNLUMultipleChoice() as task:
    results = task.evaluate(
        model=model,
        prompt_lang="fa",
        prompt_shots=[0, 5],
    )
```

Use the task in a context manager. It loads the dataset on enter and frees it on exit.

The parameters you'll actually reach for:

- `prompt_lang=` selects the prompt template language, `"fa"` (default) or `"en"` where a task ships both.
- `prompt_shots=[0, 5]` evaluates zero-shot and 5-shot in one run; each shot count produces its own result.
- `n_first=100` evaluates only the first 100 samples (default 200). Handy for cheap smoke runs before a full evaluation.
- `sub_tasks=["math_and_logic"]` restricts a task with sub-tasks (Persian MMLU, ParsiNLU Multiple Choice) to a subset.
- `skip_existing_matches=True` resumes an interrupted run: matches already generated and scored under `output_path` are not re-run.
- `prefer_concurrency=` (default True) fans completion calls out over threads when the model supports it; tune with `n_workers=` (default 4).

## Evaluation result

`evaluate()` returns a list of `EvaluationResult` objects, one per sub-task, each holding the overall score per shot count. Use them directly or convert to a pandas DataFrame:

```
eval_result = results[0]
print(eval_result.to_pandas())
```

Output:

```
     model_name                 task_name task_category        sub_task  n_shots   score_name     score
0  qwen2:latest  ParsiNLU Multiple Choice     knowledge  math_and_logic        0  Exact Match  0.600000
1  qwen2:latest  ParsiNLU Multiple Choice     knowledge  math_and_logic        3  Exact Match  0.285714
```

## Saving results

Save manually with the `save` method of `EvaluationResult`, or pass `save_evaluation=True` to `evaluate()`. `save_matches=True` also writes every match (prompt, completion, target, score), which is what you want when you need to inspect *why* a score is low:

```
with PersianMath() as task:
    results = task.evaluate(
        model=model,
        prompt_lang="fa",
        prompt_shots=[0, 5],
        save_matches=True,
        save_evaluation=True,
        output_path="results/",
    )
```

The output directory structure:

```
results
└── qwen2:latest
    └── Persian_Math
        ├── evaluation.jsonl
        ├── matches_0_shot.jsonl
        └── matches_5_shot.jsonl
```
# Examples

# Examples

The repository's [`examples/`](https://github.com/ParsBench/ParsBench/tree/main/examples) directory holds runnable, end-to-end app evaluations. Every framework example runs the **same real case**: «پروازیار», a Persian flight-booking bot with a `search_flights` tool and a `book_flight` tool it must not call without confirmation. The golden expects a Gregorian date and a price in toman with Latin digits; the bot answers with a Jalali date and rials in Persian digits. ParsBench's [normalization layer](https://parsbench.github.io/ParsBench/app-eval/normalization/index.md) matches them. Only the framework changes between files, so you can diff the integrations.

| File                                                                                                       | Framework                                                                   | Needs API key |
| ---------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------- | ------------- |
| [`quickstart.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/quickstart.py)                 | none, a plain function returning OpenAI messages                            | no            |
| [`with_openai_sdk.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_openai_sdk.py)       | OpenAI SDK (manual tool loop)                                               | yes           |
| [`with_openai_agents.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_openai_agents.py) | OpenAI Agents SDK                                                           | yes           |
| [`with_langgraph.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_langgraph.py)         | LangGraph / LangChain                                                       | yes           |
| [`with_pydantic_ai.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_pydantic_ai.py)     | Pydantic AI                                                                 | yes           |
| [`with_agno.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_agno.py)                   | Agno                                                                        | yes           |
| [`with_crewai_otel.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/with_crewai_otel.py)     | CrewAI via OpenTelemetry, the universal path for any instrumented framework | yes           |
| [`ci_with_pytest.py`](https://github.com/ParsBench/ParsBench/blob/main/examples/ci_with_pytest.py)         | pytest / CI (`result.assert_passed()`)                                      | no            |

## Industry scenarios

[`examples/industry/`](https://github.com/ParsBench/ParsBench/tree/main/examples/industry) holds vertical-specific evaluations, each showing the checks that matter for that product. Start here to see what evaluating *your* product looks like:

- `banking_support.py` (offline): rial/toman equivalence in answers and tool args, OTP-gated `forbidden_tools`, compliance `not_contains`, credential refusal.
- `ecommerce_orders.py` (offline): Jalali/Gregorian delivery dates, order ids compared as text (a wrong id fails, proven), a `max_steps` budget.
- `medical_triage.py` (offline): booking args across calendars and digit scripts, dosage-advice refusal, emergency escalation to «۱۱۵».
- `telecom_sales_simulation.py` (needs an API key): multi-turn [Iranian-user simulation](https://parsbench.github.io/ParsBench/app-eval/simulation/index.md) with toman/rial confusion and Finglish, goal and criteria judging.
- `rag_faq_support.py` (needs an API key): goldens [generated](https://parsbench.github.io/ParsBench/app-eval/generating-goldens/index.md) from `faq_netyar.md`, correctness and faithfulness judged in Persian.

The offline examples use scripted stand-in bots on purpose: swap the stand-in for the function that calls your real bot and the goldens keep working.

## Running them

```
pip install parsbench            # plus the framework of the example you run
export OPENAI_API_KEY=...        # any OpenAI-compatible gateway works:
export OPENAI_BASE_URL=...       # AvalAI, OpenRouter, Ollama, ...
export MODEL=gpt-4o-mini         # optional override

python examples/quickstart.py    # works offline, start here
```

Judge-based checks (`output=`, `context=`, `refuses=`) need a judge model and skip gracefully without one; see [The Judge](https://parsbench.github.io/ParsBench/app-eval/judge/index.md). The offline examples run in the project's CI, so they stay working.

## Benchmark notebooks

For the model-benchmarking side, these Colab notebooks benchmark real Persian-capable models with ParsBench:

- [Aya](https://huggingface.co/CohereForAI): [notebook](https://colab.research.google.com/drive/1aPayB9AaheDxT7zS4A_4SAMH3a7mIDFX?usp=sharing)
- [Ava](https://huggingface.co/MehdiHosseiniMoghadam): [notebook](https://drive.google.com/file/d/1ToJ8gTQz1ifU70EBAM7fZG2LIOY4zAp0/view?usp=sharing)
- [Dorna](https://huggingface.co/PartAI): [notebook](https://drive.google.com/file/d/1f64d0GnmcQIZ-tlN8cg49pPdiwlVlWvi/view?usp=sharing)
- [MaralGPT](https://huggingface.co/MaralGPT): [notebook](https://drive.google.com/file/d/1ZfjxPa4CfAZdQgtPaEt3nnX180A825ZF/view?usp=sharing)
# API reference

# App Evaluation

AppEvaluator evaluates a Persian AI app — any framework — against a suite of Golden expectations. Each filled Golden field switches on its check; there is no separate metric configuration.

Attributes:

| Name      | Type           | Description                                                             |
| --------- | -------------- | ----------------------------------------------------------------------- |
| `goldens` | `list[Golden]` | The golden expectations to evaluate.                                    |
| `judge`   | \`Model        | Callable                                                                |
| `metrics` | `list[str]`    | Filter/override of which checks run, e.g. ["tools:strict", "contains"]. |

Methods:

| Name           | Description                                             |
| -------------- | ------------------------------------------------------- |
| `evaluate`     | Runs each golden against the app and scores the traces. |
| `score_traces` | Scores pre-captured traces instead of running the app.  |

Source code in `parsbench/appeval/evaluator.py`

```
class AppEvaluator:
    """
    AppEvaluator evaluates a Persian AI app — any framework — against a suite
    of Golden expectations. Each filled Golden field switches on its check;
    there is no separate metric configuration.

    Attributes:
        goldens (list[Golden]): The golden expectations to evaluate.
        judge (Model | Callable | str, optional): The judge for judge-based
            checks — a parsbench Model, a callable `prompt -> str`, or a model
            name string (client built from PARSBENCH_JUDGE_* / OPENAI_* env).
            Defaults to the PARSBENCH_JUDGE env var; judge checks skip
            gracefully without one.
        metrics (list[str], optional): Filter/override of which checks run,
            e.g. ["tools:strict", "contains"].

    Methods:
        evaluate: Runs each golden against the app and scores the traces.
        score_traces: Scores pre-captured traces instead of running the app.
    """

    def __init__(
        self,
        goldens: list[Golden | dict],
        judge: Any = None,
        metrics: list[str] | None = None,
    ):
        self.goldens = [
            g if isinstance(g, Golden) else Golden.from_dict(g) for g in goldens
        ]
        if not self.goldens:
            raise ValueError("goldens is empty. You should provide at least one Golden.")
        self.judge = judge
        self.metrics = metrics

    def evaluate(
        self,
        app: Callable,
        n_runs: int = 1,
        prefer_concurrency: bool = False,
        n_workers: int = 4,
        save_evaluation: bool = False,
        output_path: str | None = None,
        record: bool = True,
    ) -> AppEvaluationResult:
        """
        Run each golden against the app and score the resulting traces.

        Parameters:
            app (Callable): The app under evaluation — a plain callable (sync
                or async), `input -> str | messages | Trace`. A crash inside
                the app is reported as a failing `app_error` check on that
                golden instead of aborting the whole run.
            n_runs (int, optional): Repeated runs per golden (default is 1);
                see `AppEvaluationResult.pass_hat_k`.
            prefer_concurrency (bool, optional): Evaluate goldens in parallel
                over a thread pool (default is False). The app and judge
                callables must then be thread-safe — most stateful bots are not,
                which is why this is off by default.
            n_workers (int, optional): The number of workers for concurrent
                processing (default is 4).
            save_evaluation (bool, optional): Flag to save the evaluation
                result (default is False).
            output_path (str, optional): The output path to save the
                evaluation result.
            record (bool, optional): Record this run into the local
                `.parsbench` store for `parsbench view` (default is True;
                also disabled by the PARSBENCH_NO_RECORD env var).

        Returns:
            AppEvaluationResult: The evaluation result over all goldens.
        """
        if n_runs < 1:
            raise ValueError("n_runs must be at least 1.")

        def runner(golden: Golden) -> Trace:
            return _run_app(app, golden)

        return self._score(
            runner,
            n_runs=n_runs,
            prefer_concurrency=prefer_concurrency,
            n_workers=n_workers,
            save_evaluation=save_evaluation,
            output_path=output_path,
            record=record,
            kind="evaluation",
            app_name=getattr(app, "__name__", type(app).__name__),
        )

    def score_traces(
        self,
        traces: list[Trace | list[dict]],
        save_evaluation: bool = False,
        output_path: str | None = None,
        record: bool = True,
    ) -> AppEvaluationResult:
        """
        Score pre-captured traces instead of running the app — run the app
        yourself (or in production) and evaluate what happened.

        Parameters:
            traces (list[Trace | list[dict]]): One trace per golden — a Trace
                or an OpenAI-format message list.
            save_evaluation (bool, optional): Flag to save the evaluation
                result (default is False).
            output_path (str, optional): The output path to save the
                evaluation result.
            record (bool, optional): Record this run into the local
                `.parsbench` store for `parsbench view` (default is True;
                also disabled by the PARSBENCH_NO_RECORD env var).

        Returns:
            AppEvaluationResult: The evaluation result over all goldens.
        """
        if len(traces) != len(self.goldens):
            raise ValueError(f"{len(traces)} traces for {len(self.goldens)} goldens.")
        # pair positionally — an id()-keyed dict would collapse when the same
        # Golden object appears twice and silently score the wrong trace
        traces_iter = iter([_to_trace(trace) for trace in traces])
        return self._score(
            lambda golden: next(traces_iter),
            n_runs=1,
            prefer_concurrency=False,
            n_workers=1,
            save_evaluation=save_evaluation,
            output_path=output_path,
            record=record,
            kind="score_traces",
            app_name="traces",
        )

    def _score(
        self,
        runner: Callable[[Golden], Trace],
        n_runs: int,
        prefer_concurrency: bool,
        n_workers: int,
        save_evaluation: bool,
        output_path: str | None,
        record: bool = True,
        kind: str = "evaluation",
        app_name: str = "app",
    ) -> AppEvaluationResult:
        if save_evaluation and not output_path:
            raise Exception("You should set the output path to save the evaluation.")

        judge = resolve_judge(self.judge)
        recorder = (
            RunRecorder.start(
                kind=kind,
                app_name=app_name,
                n_goldens=len(self.goldens),
                n_runs=n_runs,
                metrics=self.metrics,
            )
            if record
            else None
        )

        def evaluate_golden(item: tuple[int, Golden]) -> GoldenEvaluationResult:
            golden_index, golden = item
            run_passes = []
            detail: list[CheckResult] | None = None
            for run_index in range(n_runs):
                trace: Trace | None = None
                try:
                    trace = runner(golden)
                except Exception as exc:  # the app crashing is a finding, not a crash
                    run = [
                        CheckResult(
                            check="app_error",
                            passed=False,
                            reason=f"{type(exc).__name__}: {exc}",
                        )
                    ]
                else:
                    run = run_checks(golden, trace, judge=judge, only=self.metrics)
                if recorder:
                    recorder.record_event(
                        golden, golden_index, run_index, run, trace=trace
                    )
                detail = detail or run  # first run carries the readable detail
                run_passes.append(all(r.passed for r in run if not r.skipped))
            return GoldenEvaluationResult(
                golden_name=golden.label,
                check_results=detail or [],
                run_passes=run_passes,
            )

        try:
            items = list(enumerate(self.goldens))
            if prefer_concurrency and n_workers > 1:
                from concurrent.futures import ThreadPoolExecutor

                with ThreadPoolExecutor(max_workers=n_workers) as pool:
                    golden_results = list(
                        tqdm(
                            pool.map(evaluate_golden, items),
                            total=len(items),
                            desc="Evaluating goldens",
                        )
                    )
            else:
                golden_results = [
                    evaluate_golden(item)
                    for item in tqdm(items, desc="Evaluating goldens")
                ]

            evaluation_result = AppEvaluationResult(golden_results=golden_results)
        except BaseException as exc:
            if recorder:
                recorder.crashed(exc)
            raise

        if recorder:
            recorder.finish(evaluation_result)

        if save_evaluation and output_path:
            evaluation_result.save(output_path)

        return evaluation_result
```

## `evaluate(app, n_runs=1, prefer_concurrency=False, n_workers=4, save_evaluation=False, output_path=None, record=True)`

Run each golden against the app and score the resulting traces.

Parameters:

| Name                 | Type       | Description                                                                                                                                                                                 | Default  |
| -------------------- | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
| `app`                | `Callable` | The app under evaluation — a plain callable (sync or async), input -> str                                                                                                                   | messages |
| `n_runs`             | `int`      | Repeated runs per golden (default is 1); see AppEvaluationResult.pass_hat_k.                                                                                                                | `1`      |
| `prefer_concurrency` | `bool`     | Evaluate goldens in parallel over a thread pool (default is False). The app and judge callables must then be thread-safe — most stateful bots are not, which is why this is off by default. | `False`  |
| `n_workers`          | `int`      | The number of workers for concurrent processing (default is 4).                                                                                                                             | `4`      |
| `save_evaluation`    | `bool`     | Flag to save the evaluation result (default is False).                                                                                                                                      | `False`  |
| `output_path`        | `str`      | The output path to save the evaluation result.                                                                                                                                              | `None`   |
| `record`             | `bool`     | Record this run into the local .parsbench store for parsbench view (default is True; also disabled by the PARSBENCH_NO_RECORD env var).                                                     | `True`   |

Returns:

| Name                  | Type                  | Description                             |
| --------------------- | --------------------- | --------------------------------------- |
| `AppEvaluationResult` | `AppEvaluationResult` | The evaluation result over all goldens. |

Source code in `parsbench/appeval/evaluator.py`

```
def evaluate(
    self,
    app: Callable,
    n_runs: int = 1,
    prefer_concurrency: bool = False,
    n_workers: int = 4,
    save_evaluation: bool = False,
    output_path: str | None = None,
    record: bool = True,
) -> AppEvaluationResult:
    """
    Run each golden against the app and score the resulting traces.

    Parameters:
        app (Callable): The app under evaluation — a plain callable (sync
            or async), `input -> str | messages | Trace`. A crash inside
            the app is reported as a failing `app_error` check on that
            golden instead of aborting the whole run.
        n_runs (int, optional): Repeated runs per golden (default is 1);
            see `AppEvaluationResult.pass_hat_k`.
        prefer_concurrency (bool, optional): Evaluate goldens in parallel
            over a thread pool (default is False). The app and judge
            callables must then be thread-safe — most stateful bots are not,
            which is why this is off by default.
        n_workers (int, optional): The number of workers for concurrent
            processing (default is 4).
        save_evaluation (bool, optional): Flag to save the evaluation
            result (default is False).
        output_path (str, optional): The output path to save the
            evaluation result.
        record (bool, optional): Record this run into the local
            `.parsbench` store for `parsbench view` (default is True;
            also disabled by the PARSBENCH_NO_RECORD env var).

    Returns:
        AppEvaluationResult: The evaluation result over all goldens.
    """
    if n_runs < 1:
        raise ValueError("n_runs must be at least 1.")

    def runner(golden: Golden) -> Trace:
        return _run_app(app, golden)

    return self._score(
        runner,
        n_runs=n_runs,
        prefer_concurrency=prefer_concurrency,
        n_workers=n_workers,
        save_evaluation=save_evaluation,
        output_path=output_path,
        record=record,
        kind="evaluation",
        app_name=getattr(app, "__name__", type(app).__name__),
    )
```

## `score_traces(traces, save_evaluation=False, output_path=None, record=True)`

Score pre-captured traces instead of running the app — run the app yourself (or in production) and evaluate what happened.

Parameters:

| Name              | Type          | Description                                                                                                                             | Default                                                          |
| ----------------- | ------------- | --------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------- |
| `traces`          | \`list\[Trace | list[dict]\]\`                                                                                                                          | One trace per golden — a Trace or an OpenAI-format message list. |
| `save_evaluation` | `bool`        | Flag to save the evaluation result (default is False).                                                                                  | `False`                                                          |
| `output_path`     | `str`         | The output path to save the evaluation result.                                                                                          | `None`                                                           |
| `record`          | `bool`        | Record this run into the local .parsbench store for parsbench view (default is True; also disabled by the PARSBENCH_NO_RECORD env var). | `True`                                                           |

Returns:

| Name                  | Type                  | Description                             |
| --------------------- | --------------------- | --------------------------------------- |
| `AppEvaluationResult` | `AppEvaluationResult` | The evaluation result over all goldens. |

Source code in `parsbench/appeval/evaluator.py`

```
def score_traces(
    self,
    traces: list[Trace | list[dict]],
    save_evaluation: bool = False,
    output_path: str | None = None,
    record: bool = True,
) -> AppEvaluationResult:
    """
    Score pre-captured traces instead of running the app — run the app
    yourself (or in production) and evaluate what happened.

    Parameters:
        traces (list[Trace | list[dict]]): One trace per golden — a Trace
            or an OpenAI-format message list.
        save_evaluation (bool, optional): Flag to save the evaluation
            result (default is False).
        output_path (str, optional): The output path to save the
            evaluation result.
        record (bool, optional): Record this run into the local
            `.parsbench` store for `parsbench view` (default is True;
            also disabled by the PARSBENCH_NO_RECORD env var).

    Returns:
        AppEvaluationResult: The evaluation result over all goldens.
    """
    if len(traces) != len(self.goldens):
        raise ValueError(f"{len(traces)} traces for {len(self.goldens)} goldens.")
    # pair positionally — an id()-keyed dict would collapse when the same
    # Golden object appears twice and silently score the wrong trace
    traces_iter = iter([_to_trace(trace) for trace in traces])
    return self._score(
        lambda golden: next(traces_iter),
        n_runs=1,
        prefer_concurrency=False,
        n_workers=1,
        save_evaluation=save_evaluation,
        output_path=output_path,
        record=record,
        kind="score_traces",
        app_name="traces",
    )
```

One expectation for the evaluated app. Every field except `input` is optional; each filled field switches on its corresponding check.

Attributes:

| Name              | Type             | Description                                                                                 |
| ----------------- | ---------------- | ------------------------------------------------------------------------------------------- |
| `input`           | `str`            | The user message sent to the app.                                                           |
| `name`            | `str`            | A readable name for reports.                                                                |
| `output`          | `str`            | Reference answer; judged by the correctness check.                                          |
| `contains`        | `list[str]`      | Substrings the answer must state — normalized, money- and calendar-equivalent.              |
| `not_contains`    | `list[str]`      | Substrings the answer must not state.                                                       |
| `format`          | `type`           | A pydantic-style model class the answer must validate against (checked via model_validate). |
| `tools`           | `list[ToolCall]` | Tool calls the app must make; arguments are compared date -> number -> normalized text.     |
| `forbidden_tools` | `list[str]`      | Tools the app must not call.                                                                |
| `context`         | `list[str]`      | Grounding context; judged by the faithfulness check.                                        |
| `refuses`         | `bool`           | Whether the app must decline the request (refusal check).                                   |
| `max_steps`       | `int`            | Budget on assistant steps.                                                                  |
| `max_latency`     | `float`          | Budget on wall-clock seconds.                                                               |
| `max_cost`        | `float`          | Budget on cost.                                                                             |
| `check`           | `Callable`       | Custom Trace -> bool                                                                        |
| `tags`            | `list[str]`      | Free-form labels.                                                                           |

Methods:

| Name        | Description                                                                       |
| ----------- | --------------------------------------------------------------------------------- |
| `from_dict` | Builds a Golden from a dict, accepting the short in/out aliases for input/output. |

Source code in `parsbench/appeval/golden.py`

```
@dataclass
class Golden:
    """
    One expectation for the evaluated app. Every field except `input` is
    optional; each filled field switches on its corresponding check.

    Attributes:
        input (str): The user message sent to the app.
        name (str, optional): A readable name for reports.
        output (str, optional): Reference answer; judged by the `correctness` check.
        contains (list[str]): Substrings the answer must state — normalized,
            money- and calendar-equivalent.
        not_contains (list[str]): Substrings the answer must not state.
        format (type, optional): A pydantic-style model class the answer must
            validate against (checked via `model_validate`).
        tools (list[ToolCall]): Tool calls the app must make; arguments are
            compared date -> number -> normalized text.
        forbidden_tools (list[str]): Tools the app must not call.
        context (list[str]): Grounding context; judged by the `faithfulness` check.
        refuses (bool): Whether the app must decline the request (`refusal` check).
        max_steps (int, optional): Budget on assistant steps.
        max_latency (float, optional): Budget on wall-clock seconds.
        max_cost (float, optional): Budget on cost.
        check (Callable, optional): Custom `Trace -> bool | float` check.
        tags (list[str]): Free-form labels.

    Methods:
        from_dict: Builds a Golden from a dict, accepting the short `in`/`out`
            aliases for `input`/`output`.
    """

    input: str
    name: str | None = None
    # answer expectations
    output: str | None = None
    contains: list[str] = field(default_factory=list)
    not_contains: list[str] = field(default_factory=list)
    format: type | None = None
    # behavior expectations
    tools: list[ToolCall] = field(default_factory=list)
    forbidden_tools: list[str] = field(default_factory=list)
    context: list[str] = field(default_factory=list)
    refuses: bool = False
    # budgets
    max_steps: int | None = None
    max_latency: float | None = None
    max_cost: float | None = None
    # escape hatch
    check: Callable[[Trace], bool | float] | None = None
    tags: list[str] = field(default_factory=list)

    def __post_init__(self):
        # a bare string for a list field must not be iterated character by
        # character — wrap it; dict tool specs are promoted to ToolCall
        self.contains = _as_list(self.contains)
        self.not_contains = _as_list(self.not_contains)
        self.forbidden_tools = _as_list(self.forbidden_tools)
        self.context = _as_list(self.context)
        self.tags = _as_list(self.tags)
        self.tools = [
            tc if isinstance(tc, ToolCall) else ToolCall(**tc) for tc in self.tools
        ]

    @classmethod
    def from_dict(cls, data: dict) -> "Golden":
        data = dict(data)
        if "in" in data:
            data["input"] = data.pop("in")
        if "out" in data:
            data["output"] = data.pop("out")
        return cls(**data)

    @property
    def label(self) -> str:
        return self.name or (self.input[:40] + "…" if len(self.input) > 40 else self.input)
```

A multi-turn expectation, consumed by SimulationEvaluator.

Attributes:

| Name               | Type        | Description                                          |
| ------------------ | ----------- | ---------------------------------------------------- |
| `goal`             | `str`       | What the simulated user wants from the conversation. |
| `name`             | `str`       | A readable name for reports.                         |
| `scenario`         | `str`       | Extra situation description for the user simulator.  |
| `expected_outcome` | `str`       | What "done well" looks like, for the judge.          |
| `criteria`         | `list[str]` | Behaviors judged each as their own check.            |
| `max_turns`        | `int`       | Turn cap for the conversation.                       |

Source code in `parsbench/appeval/golden.py`

```
@dataclass
class ConversationGolden:
    """
    A multi-turn expectation, consumed by SimulationEvaluator.

    Attributes:
        goal (str): What the simulated user wants from the conversation.
        name (str, optional): A readable name for reports.
        scenario (str): Extra situation description for the user simulator.
        expected_outcome (str): What "done well" looks like, for the judge.
        criteria (list[str]): Behaviors judged each as their own check.
        max_turns (int): Turn cap for the conversation.
    """

    goal: str
    name: str | None = None
    scenario: str = ""
    expected_outcome: str = ""
    criteria: list[str] = field(default_factory=list)
    max_turns: int = 8

    def __post_init__(self):
        self.criteria = _as_list(self.criteria)

    @classmethod
    def from_dict(cls, data: dict) -> "ConversationGolden":
        return cls(**data)

    @property
    def label(self) -> str:
        return self.name or (self.goal[:40] + "…" if len(self.goal) > 40 else self.goal)
```

What the evaluated app actually did in response to one input.

Attributes:

| Name           | Type            | Description                               |
| -------------- | --------------- | ----------------------------------------- |
| `messages`     | `list[Message]` | The conversation messages, OpenAI-style.  |
| `final_output` | `str`           | The final assistant answer.               |
| `latency`      | `float`         | Wall-clock seconds for the run.           |
| `cost`         | `float`         | Cost of the run, if known.                |
| `raw`          | `Any`           | The original framework object, untouched. |

Methods:

| Name            | Description                                      |
| --------------- | ------------------------------------------------ |
| `from_messages` | Builds a Trace from OpenAI-format message dicts. |

Source code in `parsbench/appeval/trace.py`

```
@dataclass
class Trace:
    """
    What the evaluated app actually did in response to one input.

    Attributes:
        messages (list[Message]): The conversation messages, OpenAI-style.
        final_output (str): The final assistant answer.
        latency (float, optional): Wall-clock seconds for the run.
        cost (float, optional): Cost of the run, if known.
        raw (Any, optional): The original framework object, untouched.

    Methods:
        from_messages: Builds a Trace from OpenAI-format message dicts.
    """

    messages: list[Message] = field(default_factory=list)
    final_output: str = ""
    latency: float | None = None
    cost: float | None = None
    raw: Any = None

    @property
    def tool_calls(self) -> list[ToolCall]:
        return [tc for m in self.messages if m.role == "assistant" for tc in m.tool_calls]

    @property
    def n_steps(self) -> int:
        return sum(1 for m in self.messages if m.role == "assistant")

    @classmethod
    def from_messages(
        cls, messages: list[dict], *, final_output: Any = None, raw: Any = None
    ) -> "Trace":
        """
        Build a Trace from OpenAI-format message dicts.

        Parameters:
            messages (list[dict]): OpenAI chat-format messages.
            final_output (Any, optional): Overrides the derived final answer,
                but only when it is a non-empty string.
            raw (Any, optional): The original framework object to keep around.

        Returns:
            Trace: The parsed trace.
        """
        parsed: list[Message] = []
        calls_by_id: dict[str, ToolCall] = {}
        for m in messages:
            calls = []
            for tc in m.get("tool_calls") or []:
                fn = tc.get("function", tc)
                args = fn.get("arguments", {})
                if isinstance(args, str):
                    try:
                        args = json.loads(args)
                    except json.JSONDecodeError:
                        args = {"_raw": args}
                call = ToolCall(fn.get("name", ""), arguments=args)
                calls.append(call)
                if tc.get("id"):
                    calls_by_id[tc["id"]] = call
            role = m.get("role", "assistant")
            role = _ROLE_ALIASES.get(role, role if role in _ROLES else "assistant")
            parsed.append(
                Message(
                    role=role,
                    content=m.get("content"),
                    tool_calls=calls,
                    tool_call_id=m.get("tool_call_id"),
                )
            )
            if role == "tool" and m.get("tool_call_id") in calls_by_id:
                calls_by_id[m["tool_call_id"]].result = m.get("content")
        final = next(
            (m.content for m in reversed(parsed) if m.role == "assistant" and m.content),
            "",
        )
        if isinstance(final_output, str) and final_output:
            final = final_output
        return cls(messages=parsed, final_output=final, raw=raw)
```

## `from_messages(messages, *, final_output=None, raw=None)`

Build a Trace from OpenAI-format message dicts.

Parameters:

| Name           | Type         | Description                                                                 | Default    |
| -------------- | ------------ | --------------------------------------------------------------------------- | ---------- |
| `messages`     | `list[dict]` | OpenAI chat-format messages.                                                | *required* |
| `final_output` | `Any`        | Overrides the derived final answer, but only when it is a non-empty string. | `None`     |
| `raw`          | `Any`        | The original framework object to keep around.                               | `None`     |

Returns:

| Name    | Type    | Description       |
| ------- | ------- | ----------------- |
| `Trace` | `Trace` | The parsed trace. |

Source code in `parsbench/appeval/trace.py`

```
@classmethod
def from_messages(
    cls, messages: list[dict], *, final_output: Any = None, raw: Any = None
) -> "Trace":
    """
    Build a Trace from OpenAI-format message dicts.

    Parameters:
        messages (list[dict]): OpenAI chat-format messages.
        final_output (Any, optional): Overrides the derived final answer,
            but only when it is a non-empty string.
        raw (Any, optional): The original framework object to keep around.

    Returns:
        Trace: The parsed trace.
    """
    parsed: list[Message] = []
    calls_by_id: dict[str, ToolCall] = {}
    for m in messages:
        calls = []
        for tc in m.get("tool_calls") or []:
            fn = tc.get("function", tc)
            args = fn.get("arguments", {})
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except json.JSONDecodeError:
                    args = {"_raw": args}
            call = ToolCall(fn.get("name", ""), arguments=args)
            calls.append(call)
            if tc.get("id"):
                calls_by_id[tc["id"]] = call
        role = m.get("role", "assistant")
        role = _ROLE_ALIASES.get(role, role if role in _ROLES else "assistant")
        parsed.append(
            Message(
                role=role,
                content=m.get("content"),
                tool_calls=calls,
                tool_call_id=m.get("tool_call_id"),
            )
        )
        if role == "tool" and m.get("tool_call_id") in calls_by_id:
            calls_by_id[m["tool_call_id"]].result = m.get("content")
    final = next(
        (m.content for m in reversed(parsed) if m.role == "assistant" and m.content),
        "",
    )
    if isinstance(final_output, str) and final_output:
        final = final_output
    return cls(messages=parsed, final_output=final, raw=raw)
```

A single message in a conversation trace.

Attributes:

| Name           | Type             | Description                                                 |
| -------------- | ---------------- | ----------------------------------------------------------- |
| `role`         | `str`            | One of "system", "user", "assistant", or "tool".            |
| `content`      | `str`            | The text content of the message.                            |
| `tool_calls`   | `list[ToolCall]` | Tool calls made in this message.                            |
| `tool_call_id` | `str`            | For tool messages, the id of the call this message answers. |

Source code in `parsbench/appeval/trace.py`

```
@dataclass
class Message:
    """
    A single message in a conversation trace.

    Attributes:
        role (str): One of "system", "user", "assistant", or "tool".
        content (str, optional): The text content of the message.
        tool_calls (list[ToolCall]): Tool calls made in this message.
        tool_call_id (str, optional): For tool messages, the id of the call
            this message answers.
    """

    role: str
    content: str | None = None
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_call_id: str | None = None
```

Represents an expected or observed tool call.

Extra keyword arguments become tool arguments, so the short form `ToolCall("search", date="1405-07-05")` is equivalent to `ToolCall(name="search", arguments={"date": "1405-07-05"})`.

Attributes:

| Name        | Type   | Description                                            |
| ----------- | ------ | ------------------------------------------------------ |
| `name`      | `str`  | The name of the tool.                                  |
| `arguments` | `dict` | The arguments the tool was (or should be) called with. |
| `result`    | `Any`  | The observed tool result, if any.                      |
| `error`     | `str`  | The observed tool error, if any.                       |

Source code in `parsbench/appeval/trace.py`

```
@dataclass(init=False)
class ToolCall:
    """
    Represents an expected or observed tool call.

    Extra keyword arguments become tool arguments, so the short form
    `ToolCall("search", date="1405-07-05")` is equivalent to
    `ToolCall(name="search", arguments={"date": "1405-07-05"})`.

    Attributes:
        name (str): The name of the tool.
        arguments (dict): The arguments the tool was (or should be) called with.
        result (Any, optional): The observed tool result, if any.
        error (str, optional): The observed tool error, if any.
    """

    name: str
    arguments: dict[str, Any]
    result: Any
    error: str | None

    def __init__(
        self,
        name: str = "",
        arguments: dict[str, Any] | None = None,
        result: Any = None,
        error: str | None = None,
        **extra_arguments: Any,
    ):
        self.name = name
        self.arguments = {**(arguments or {}), **extra_arguments}
        self.result = result
        self.error = error
```

The result of evaluating an app against a suite of goldens.

Attributes:

| Name             | Type                           | Description            |
| ---------------- | ------------------------------ | ---------------------- |
| `golden_results` | `list[GoldenEvaluationResult]` | One result per golden. |

Methods:

| Name            | Description                                                      |
| --------------- | ---------------------------------------------------------------- |
| `score`         | Mean score, optionally restricted to one check name.             |
| `pass_hat_k`    | tau2-style pass^k consistency over repeated runs.                |
| `save`          | Writes the result to app_evaluation.jsonl in the given path.     |
| `diff`          | Prints per-check mean deltas vs a previously saved result.       |
| `to_langfuse`   | Pushes per-check scores to a Langfuse instance.                  |
| `assert_passed` | Raises AssertionError with the failing checks (pytest-friendly). |

Source code in `parsbench/appeval/evaluation_result.py`

```
@dataclass
class AppEvaluationResult:
    """
    The result of evaluating an app against a suite of goldens.

    Attributes:
        golden_results (list[GoldenEvaluationResult]): One result per golden.

    Methods:
        score: Mean score, optionally restricted to one check name.
        pass_hat_k: tau2-style pass^k consistency over repeated runs.
        save: Writes the result to `app_evaluation.jsonl` in the given path.
        diff: Prints per-check mean deltas vs a previously saved result.
        to_langfuse: Pushes per-check scores to a Langfuse instance.
        assert_passed: Raises AssertionError with the failing checks (pytest-friendly).
    """

    golden_results: list[GoldenEvaluationResult]

    @property
    def passed(self) -> bool:
        return all(gr.passed for gr in self.golden_results)

    @property
    def average_score(self) -> float:
        return self.score()

    def score(self, check: str | None = None) -> float:
        scores = [
            cr.score
            for gr in self.golden_results
            for cr in gr.check_results
            if not cr.skipped and (check is None or cr.check.startswith(check))
        ]
        return sum(scores) / len(scores) if scores else 0.0

    def pass_hat_k(self, k: int | None = None) -> float:
        """
        tau2-style pass^k over repeated runs: C(c,k)/C(n,k) averaged over
        goldens, where n = runs done and c = runs passed.

        Parameters:
            k (int, optional): The consistency exponent (default is all runs).

        Returns:
            float: The pass^k score.
        """
        values = []
        for gr in self.golden_results:
            runs = gr.run_passes or [gr.passed]
            n, c = len(runs), sum(runs)
            kk = k or n
            if kk > n:
                raise ValueError(
                    f"k={kk} but only {n} runs were done (evaluate with n_runs={kk})."
                )
            values.append(math.comb(c, kk) / math.comb(n, kk))
        return sum(values) / len(values) if values else 0.0

    @classmethod
    def from_file(cls, path: str) -> "AppEvaluationResult":
        import jsonlines

        with jsonlines.open(path, "r") as reader:
            golden_results = [
                GoldenEvaluationResult.from_dict(row) for row in reader.iter(type=dict)
            ]
        return cls(golden_results=golden_results)

    @classmethod
    def from_dict(cls, data: dict) -> "AppEvaluationResult":
        golden_results = [
            GoldenEvaluationResult.from_dict(gr) for gr in data.pop("golden_results")
        ]
        return cls(**data, golden_results=golden_results)

    def to_dict(self) -> dict:
        return {"golden_results": [gr.to_dict() for gr in self.golden_results]}

    def to_pandas(self) -> pd.DataFrame:
        import pandas as pd

        return pd.concat([gr.to_pandas() for gr in self.golden_results])

    def save(self, path: str):
        evaluation_path = Path(path) / EVALUATION_FILE_NAME
        # create the directory up front — failing here after a full (paid,
        # judge-calling) evaluation would lose the finished result
        evaluation_path.parent.mkdir(parents=True, exist_ok=True)
        import jsonlines

        with jsonlines.open(evaluation_path, "w") as writer:
            for gr in self.golden_results:
                writer.write(gr.to_dict())

    def diff(self, path: str):
        """
        Print per-check mean score deltas vs a previously saved result.

        Parameters:
            path (str): Path to a saved `app_evaluation.jsonl` file.
        """
        old = AppEvaluationResult.from_file(path)
        checks = {cr.check for gr in self.golden_results for cr in gr.check_results}
        for check in sorted(checks):
            delta = self.score(check) - old.score(check)
            if abs(delta) > 1e-9:
                print(f"{check}: {old.score(check):.2f} -> {self.score(check):.2f} ({delta:+.2f})")

    def to_langfuse(self, **kwargs) -> str:
        """Push per-check scores into Langfuse. Returns the created trace id."""
        from parsbench.integrations import langfuse

        return langfuse.push(self, **kwargs)

    def assert_passed(self):
        """Raise AssertionError listing every failing check, for pytest/CI."""
        failed = [
            f"{gr.golden_name} — {cr.check}: {cr.reason or f'score={cr.score:.2f}'}"
            for gr in self.golden_results
            for cr in gr.check_results
            if not cr.passed and not cr.skipped
        ]
        if failed:
            raise AssertionError("ParsBench checks failed:\n  " + "\n  ".join(failed))

    def __str__(self) -> str:
        text = ""
        for gr in self.golden_results:
            text += str(gr) + "\n"
        total = sum(len(gr.check_results) for gr in self.golden_results)
        failed = sum(
            1
            for gr in self.golden_results
            for cr in gr.check_results
            if not cr.passed and not cr.skipped
        )
        text += f"score={self.score():.2f}  checks={total}  failed={failed}"
        return text
```

## `assert_passed()`

Raise AssertionError listing every failing check, for pytest/CI.

Source code in `parsbench/appeval/evaluation_result.py`

```
def assert_passed(self):
    """Raise AssertionError listing every failing check, for pytest/CI."""
    failed = [
        f"{gr.golden_name} — {cr.check}: {cr.reason or f'score={cr.score:.2f}'}"
        for gr in self.golden_results
        for cr in gr.check_results
        if not cr.passed and not cr.skipped
    ]
    if failed:
        raise AssertionError("ParsBench checks failed:\n  " + "\n  ".join(failed))
```

## `diff(path)`

Print per-check mean score deltas vs a previously saved result.

Parameters:

| Name   | Type  | Description                                | Default    |
| ------ | ----- | ------------------------------------------ | ---------- |
| `path` | `str` | Path to a saved app_evaluation.jsonl file. | *required* |

Source code in `parsbench/appeval/evaluation_result.py`

```
def diff(self, path: str):
    """
    Print per-check mean score deltas vs a previously saved result.

    Parameters:
        path (str): Path to a saved `app_evaluation.jsonl` file.
    """
    old = AppEvaluationResult.from_file(path)
    checks = {cr.check for gr in self.golden_results for cr in gr.check_results}
    for check in sorted(checks):
        delta = self.score(check) - old.score(check)
        if abs(delta) > 1e-9:
            print(f"{check}: {old.score(check):.2f} -> {self.score(check):.2f} ({delta:+.2f})")
```

## `pass_hat_k(k=None)`

tau2-style pass^k over repeated runs: C(c,k)/C(n,k) averaged over goldens, where n = runs done and c = runs passed.

Parameters:

| Name | Type  | Description                                     | Default |
| ---- | ----- | ----------------------------------------------- | ------- |
| `k`  | `int` | The consistency exponent (default is all runs). | `None`  |

Returns:

| Name    | Type    | Description       |
| ------- | ------- | ----------------- |
| `float` | `float` | The pass^k score. |

Source code in `parsbench/appeval/evaluation_result.py`

```
def pass_hat_k(self, k: int | None = None) -> float:
    """
    tau2-style pass^k over repeated runs: C(c,k)/C(n,k) averaged over
    goldens, where n = runs done and c = runs passed.

    Parameters:
        k (int, optional): The consistency exponent (default is all runs).

    Returns:
        float: The pass^k score.
    """
    values = []
    for gr in self.golden_results:
        runs = gr.run_passes or [gr.passed]
        n, c = len(runs), sum(runs)
        kk = k or n
        if kk > n:
            raise ValueError(
                f"k={kk} but only {n} runs were done (evaluate with n_runs={kk})."
            )
        values.append(math.comb(c, kk) / math.comb(n, kk))
    return sum(values) / len(values) if values else 0.0
```

## `to_langfuse(**kwargs)`

Push per-check scores into Langfuse. Returns the created trace id.

Source code in `parsbench/appeval/evaluation_result.py`

```
def to_langfuse(self, **kwargs) -> str:
    """Push per-check scores into Langfuse. Returns the created trace id."""
    from parsbench.integrations import langfuse

    return langfuse.push(self, **kwargs)
```

The evaluation result for one golden: its check results and, when the golden was run more than once, the pass/fail of each repeated run.

Attributes:

| Name            | Type                | Description                                                                          |
| --------------- | ------------------- | ------------------------------------------------------------------------------------ |
| `golden_name`   | `str`               | The label of the evaluated golden.                                                   |
| `check_results` | `list[CheckResult]` | The results of each check.                                                           |
| `run_passes`    | `list[bool]`        | Pass/fail of each repeated run (n_runs > 1); a single-element list for a single run. |
| `transcript`    | `str`               | The rendered conversation (simulation only).                                         |

Source code in `parsbench/appeval/evaluation_result.py`

```
@dataclass
class GoldenEvaluationResult:
    """
    The evaluation result for one golden: its check results and, when the
    golden was run more than once, the pass/fail of each repeated run.

    Attributes:
        golden_name (str): The label of the evaluated golden.
        check_results (list[CheckResult]): The results of each check.
        run_passes (list[bool]): Pass/fail of each repeated run (n_runs > 1);
            a single-element list for a single run.
        transcript (str, optional): The rendered conversation (simulation only).
    """

    golden_name: str
    check_results: list[CheckResult] = field(default_factory=list)
    run_passes: list[bool] = field(default_factory=list)
    transcript: str | None = None

    @property
    def passed(self) -> bool:
        return all(r.passed for r in self.check_results if not r.skipped)

    @classmethod
    def from_dict(cls, data: dict) -> "GoldenEvaluationResult":
        check_results = [CheckResult.from_dict(cr) for cr in data.pop("check_results")]
        return cls(**data, check_results=check_results)

    def to_dict(self) -> dict:
        return {
            **asdict(self),
            "check_results": [cr.to_dict() for cr in self.check_results],
        }

    def to_pandas(self) -> pd.DataFrame:
        import pandas as pd

        return pd.DataFrame(
            [
                {"golden_name": self.golden_name, **cr.to_dict()}
                for cr in self.check_results
            ]
        )

    def __str__(self) -> str:
        mark = "PASS" if self.passed else "FAIL"
        text = f"[{mark}] {self.golden_name}\n"
        for cr in self.check_results:
            status = "skip" if cr.skipped else ("ok  " if cr.passed else "FAIL")
            text += f"  {status} {cr.check:<16} {cr.score:.2f}"
            if cr.reason and not cr.passed:
                text += f"  — {cr.reason}"
            text += "\n"
        return text.strip("\n")
```

The outcome of a single check on a single golden.

Attributes:

| Name      | Type    | Description                                               |
| --------- | ------- | --------------------------------------------------------- |
| `check`   | `str`   | The name of the check (e.g. "contains", "tools:subset").  |
| `score`   | `float` | The check score between 0 and 1.                          |
| `passed`  | `bool`  | Whether the check passed.                                 |
| `skipped` | `bool`  | Whether the check was skipped (e.g. no judge configured). |
| `reason`  | `str`   | A readable explanation for failures/skips.                |

Source code in `parsbench/appeval/evaluation_result.py`

```
@dataclass
class CheckResult:
    """
    The outcome of a single check on a single golden.

    Attributes:
        check (str): The name of the check (e.g. "contains", "tools:subset").
        score (float): The check score between 0 and 1.
        passed (bool): Whether the check passed.
        skipped (bool): Whether the check was skipped (e.g. no judge configured).
        reason (str, optional): A readable explanation for failures/skips.
    """

    check: str
    score: float = 0.0
    passed: bool = False
    skipped: bool = False
    reason: str | None = None

    @classmethod
    def from_dict(cls, data: dict) -> "CheckResult":
        return cls(**data)

    def to_dict(self) -> dict:
        return asdict(self)
```

SimulationEvaluator simulates Persian users against a bot and judges the finished conversations against each goal and its criteria.

Attributes:

| Name              | Type                       | Description                                                                                                                                                                  |
| ----------------- | -------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `goldens`         | `list[ConversationGolden]` | The multi-turn expectations.                                                                                                                                                 |
| `user`            | `PersianUser`              | The simulated user's personality. Also accepts a string like "محاوره‌ای+finglish_switch" mixing a register with trap names (see TRAPS); free text becomes extra instructions. |
| `simulator_model` | \`Model                    | Callable                                                                                                                                                                     |
| `judge`           | \`Model                    | Callable                                                                                                                                                                     |

Methods:

| Name       | Description                                                  |
| ---------- | ------------------------------------------------------------ |
| `evaluate` | Simulates the conversations against the app and scores them. |

Source code in `parsbench/appeval/simulation.py`

```
class SimulationEvaluator:
    """
    SimulationEvaluator simulates Persian users against a bot and judges the
    finished conversations against each goal and its criteria.

    Attributes:
        goldens (list[ConversationGolden]): The multi-turn expectations.
        user (PersianUser): The simulated user's personality. Also accepts a
            string like "محاوره‌ای+finglish_switch" mixing a register with trap
            names (see TRAPS); free text becomes extra instructions.
        simulator_model (Model | Callable | str, optional): The user-simulator
            LLM (falls back to PARSBENCH_SIMULATOR, then PARSBENCH_JUDGE env).
        judge (Model | Callable | str, optional): The conversation judge
            (falls back to the PARSBENCH_JUDGE env var).

    Methods:
        evaluate: Simulates the conversations against the app and scores them.
    """

    def __init__(
        self,
        goldens: list[ConversationGolden | dict] | None = None,
        goal: str | None = None,
        criteria: list[str] | None = None,
        user: PersianUser | str | None = None,
        simulator_model: Any = None,
        judge: Any = None,
    ):
        self.goldens = [
            g if isinstance(g, ConversationGolden) else ConversationGolden.from_dict(g)
            for g in (goldens or [])
        ]
        if goal:
            self.goldens.append(ConversationGolden(goal=goal, criteria=criteria or []))
        if not self.goldens:
            raise ValueError(
                "goldens is empty. You should provide a goal or at least one "
                "ConversationGolden."
            )
        self.user = _parse_user_spec(user)
        self.simulator_model = simulator_model
        self.judge = judge

    def evaluate(
        self,
        app: Callable,
        n_runs: int = 1,
        max_turns: int | None = None,
        save_evaluation: bool = False,
        output_path: str | None = None,
        record: bool = True,
    ) -> AppEvaluationResult:
        """
        Simulate each golden's conversation against the app and judge it.

        Parameters:
            app (Callable): The bot under evaluation — `fn(message)` for
                stateful bots or `fn(message, history)`.
            n_runs (int, optional): Repeated conversations per golden
                (default is 1); see `AppEvaluationResult.pass_hat_k`.
            max_turns (int, optional): Turn cap override; defaults to each
                golden's own `max_turns`.
            save_evaluation (bool, optional): Flag to save the evaluation
                result (default is False).
            output_path (str, optional): The output path to save the
                evaluation result.
            record (bool, optional): Record this run into the local
                `.parsbench` store for `parsbench view` (default is True;
                also disabled by the PARSBENCH_NO_RECORD env var).

        Returns:
            AppEvaluationResult: The evaluation result over all conversations.
        """
        if n_runs < 1:
            raise ValueError("n_runs must be at least 1.")
        if save_evaluation and not output_path:
            raise Exception("You should set the output path to save the evaluation.")

        # the user simulator stays at the provider's default temperature —
        # pinning it to 0 would replay near-identical conversations and blind
        # pass_hat_k
        simulator = resolve_model(
            self.simulator_model, "PARSBENCH_SIMULATOR", "PARSBENCH_JUDGE",
            temperature=None,
        )
        if simulator is None:
            raise ValueError(
                "simulator model not configured — pass simulator_model= or set "
                "PARSBENCH_SIMULATOR."
            )
        judge = resolve_model(self.judge, "PARSBENCH_JUDGE")

        recorder = (
            RunRecorder.start(
                kind="simulation",
                app_name=getattr(app, "__name__", type(app).__name__),
                n_goldens=len(self.goldens),
                n_runs=n_runs,
                extra_meta={
                    "user": {
                        "style": self.user.style,
                        "persona": self.user.persona,
                        "traps": self.user.traps,
                    }
                },
            )
            if record
            else None
        )

        arity = _app_arity(app)
        golden_results = []
        try:
            for golden_index, golden in enumerate(
                tqdm(self.goldens, desc="Simulating conversations")
            ):
                turn_cap = max_turns or golden.max_turns
                run_passes: list[bool] = []
                detail: list[CheckResult] | None = None
                transcript = None
                for run_index in range(n_runs):
                    rendered: str | None = None
                    run_trace: Trace | None = None
                    converged: bool | None = None
                    try:
                        history, converged = self._run_conversation(
                            app, arity, simulator, golden, turn_cap
                        )
                    except Exception as exc:  # a crashing bot is a finding
                        checks = [
                            CheckResult(
                                check="app_error",
                                passed=False,
                                reason=f"{type(exc).__name__}: {exc}",
                            )
                        ]
                        converged = None
                    else:
                        checks = self._score_conversation(
                            history, converged, golden, judge
                        )
                        rendered = _render(history)
                        run_trace = Trace.from_messages(history)
                        if transcript is None:
                            transcript = rendered
                    if recorder:
                        recorder.record_event(
                            golden, golden_index, run_index, checks,
                            trace=run_trace, transcript=rendered,
                            converged=converged,
                        )
                    if detail is None:
                        detail = checks
                    run_passes.append(all(c.passed for c in checks if not c.skipped))
                golden_results.append(
                    GoldenEvaluationResult(
                        golden_name=golden.label,
                        check_results=detail or [],
                        run_passes=run_passes,
                        transcript=transcript,
                    )
                )

            evaluation_result = AppEvaluationResult(golden_results=golden_results)
        except BaseException as exc:
            if recorder:
                recorder.crashed(exc)
            raise

        if recorder:
            recorder.finish(evaluation_result)

        if save_evaluation and output_path:
            evaluation_result.save(output_path)

        return evaluation_result

    def _run_conversation(
        self,
        app: Callable,
        arity: int,
        simulator: Callable,
        golden: ConversationGolden,
        max_turns: int,
    ) -> tuple[list[dict], bool]:
        system = self.user.system_prompt(golden)
        history: list[dict] = []
        converged = False
        for _ in range(max_turns):
            prompt = system
            if history:
                prompt += "\n\nگفتگو تا این لحظه:\n" + _render(history)
            prompt += "\n\nپیام بعدی کاربر:"
            user_msg = str(simulator(prompt)).strip()
            if DONE in user_msg:
                converged = True
                break
            history.append({"role": "user", "content": user_msg})
            history.append(
                {"role": "assistant",
                 "content": _call_app(app, arity, user_msg, history[:-1])}
            )
        return history, converged

    def _score_conversation(
        self,
        history: list[dict],
        converged: bool,
        golden: ConversationGolden,
        judge: Any,
    ) -> list[CheckResult]:
        transcript = _render(history)
        specs = [
            ("goal", GOAL_EVAL_FA.format(
                transcript=transcript, goal=golden.goal,
                expected=f"نتیجهٔ مطلوب: {golden.expected_outcome}\n"
                if golden.expected_outcome else "",
            ))
        ]
        specs += [
            (f"criterion:{c[:24]}",
             CRITERION_EVAL_FA.format(transcript=transcript, criterion=c))
            for c in golden.criteria
        ]
        judged = run_judge_specs(judge, specs)
        goal = judged[0]
        # a chatty simulator that never emits DONE must not fail a conversation
        # the judge scored as successful — the turn cap merely cut the chat short
        reached = converged or (not goal.skipped and goal.passed)
        if not reached:
            reason = "شبیه‌ساز کاربر به هدف نرسید (سقف نوبت‌ها)."
        elif not converged:
            reason = "سقف نوبت‌ها پر شد ولی داور هدف را برآورده‌شده ارزیابی کرد."
        else:
            reason = None
        out = [CheckResult(check="converged", score=float(reached), passed=reached,
                           reason=reason)]
        out.extend(judged)
        return out
```

## `evaluate(app, n_runs=1, max_turns=None, save_evaluation=False, output_path=None, record=True)`

Simulate each golden's conversation against the app and judge it.

Parameters:

| Name              | Type       | Description                                                                                                                             | Default    |
| ----------------- | ---------- | --------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `app`             | `Callable` | The bot under evaluation — fn(message) for stateful bots or fn(message, history).                                                       | *required* |
| `n_runs`          | `int`      | Repeated conversations per golden (default is 1); see AppEvaluationResult.pass_hat_k.                                                   | `1`        |
| `max_turns`       | `int`      | Turn cap override; defaults to each golden's own max_turns.                                                                             | `None`     |
| `save_evaluation` | `bool`     | Flag to save the evaluation result (default is False).                                                                                  | `False`    |
| `output_path`     | `str`      | The output path to save the evaluation result.                                                                                          | `None`     |
| `record`          | `bool`     | Record this run into the local .parsbench store for parsbench view (default is True; also disabled by the PARSBENCH_NO_RECORD env var). | `True`     |

Returns:

| Name                  | Type                  | Description                                   |
| --------------------- | --------------------- | --------------------------------------------- |
| `AppEvaluationResult` | `AppEvaluationResult` | The evaluation result over all conversations. |

Source code in `parsbench/appeval/simulation.py`

```
def evaluate(
    self,
    app: Callable,
    n_runs: int = 1,
    max_turns: int | None = None,
    save_evaluation: bool = False,
    output_path: str | None = None,
    record: bool = True,
) -> AppEvaluationResult:
    """
    Simulate each golden's conversation against the app and judge it.

    Parameters:
        app (Callable): The bot under evaluation — `fn(message)` for
            stateful bots or `fn(message, history)`.
        n_runs (int, optional): Repeated conversations per golden
            (default is 1); see `AppEvaluationResult.pass_hat_k`.
        max_turns (int, optional): Turn cap override; defaults to each
            golden's own `max_turns`.
        save_evaluation (bool, optional): Flag to save the evaluation
            result (default is False).
        output_path (str, optional): The output path to save the
            evaluation result.
        record (bool, optional): Record this run into the local
            `.parsbench` store for `parsbench view` (default is True;
            also disabled by the PARSBENCH_NO_RECORD env var).

    Returns:
        AppEvaluationResult: The evaluation result over all conversations.
    """
    if n_runs < 1:
        raise ValueError("n_runs must be at least 1.")
    if save_evaluation and not output_path:
        raise Exception("You should set the output path to save the evaluation.")

    # the user simulator stays at the provider's default temperature —
    # pinning it to 0 would replay near-identical conversations and blind
    # pass_hat_k
    simulator = resolve_model(
        self.simulator_model, "PARSBENCH_SIMULATOR", "PARSBENCH_JUDGE",
        temperature=None,
    )
    if simulator is None:
        raise ValueError(
            "simulator model not configured — pass simulator_model= or set "
            "PARSBENCH_SIMULATOR."
        )
    judge = resolve_model(self.judge, "PARSBENCH_JUDGE")

    recorder = (
        RunRecorder.start(
            kind="simulation",
            app_name=getattr(app, "__name__", type(app).__name__),
            n_goldens=len(self.goldens),
            n_runs=n_runs,
            extra_meta={
                "user": {
                    "style": self.user.style,
                    "persona": self.user.persona,
                    "traps": self.user.traps,
                }
            },
        )
        if record
        else None
    )

    arity = _app_arity(app)
    golden_results = []
    try:
        for golden_index, golden in enumerate(
            tqdm(self.goldens, desc="Simulating conversations")
        ):
            turn_cap = max_turns or golden.max_turns
            run_passes: list[bool] = []
            detail: list[CheckResult] | None = None
            transcript = None
            for run_index in range(n_runs):
                rendered: str | None = None
                run_trace: Trace | None = None
                converged: bool | None = None
                try:
                    history, converged = self._run_conversation(
                        app, arity, simulator, golden, turn_cap
                    )
                except Exception as exc:  # a crashing bot is a finding
                    checks = [
                        CheckResult(
                            check="app_error",
                            passed=False,
                            reason=f"{type(exc).__name__}: {exc}",
                        )
                    ]
                    converged = None
                else:
                    checks = self._score_conversation(
                        history, converged, golden, judge
                    )
                    rendered = _render(history)
                    run_trace = Trace.from_messages(history)
                    if transcript is None:
                        transcript = rendered
                if recorder:
                    recorder.record_event(
                        golden, golden_index, run_index, checks,
                        trace=run_trace, transcript=rendered,
                        converged=converged,
                    )
                if detail is None:
                    detail = checks
                run_passes.append(all(c.passed for c in checks if not c.skipped))
            golden_results.append(
                GoldenEvaluationResult(
                    golden_name=golden.label,
                    check_results=detail or [],
                    run_passes=run_passes,
                    transcript=transcript,
                )
            )

        evaluation_result = AppEvaluationResult(golden_results=golden_results)
    except BaseException as exc:
        if recorder:
            recorder.crashed(exc)
        raise

    if recorder:
        recorder.finish(evaluation_result)

    if save_evaluation and output_path:
        evaluation_result.save(output_path)

    return evaluation_result
```

The simulated user's personality. Editable, composable.

Attributes:

| Name      | Type        | Description                                    |
| --------- | ----------- | ---------------------------------------------- |
| `style`   | `str`       | The speech register (e.g. "محاوره‌ای", "رسمی"). |
| `persona` | `str`       | Extra free-text persona description.           |
| `traps`   | `list[str]` | Keys of TRAPS or free-text instructions.       |

Methods:

| Name            | Description                                       |
| --------------- | ------------------------------------------------- |
| `system_prompt` | Renders the simulator system prompt for a golden. |

Source code in `parsbench/appeval/simulation.py`

```
@dataclass
class PersianUser:
    """
    The simulated user's personality. Editable, composable.

    Attributes:
        style (str): The speech register (e.g. "محاوره‌ای", "رسمی").
        persona (str): Extra free-text persona description.
        traps (list[str]): Keys of TRAPS or free-text instructions.

    Methods:
        system_prompt: Renders the simulator system prompt for a golden.
    """

    style: str = "محاوره‌ای"
    persona: str = ""
    traps: list[str] = field(default_factory=list)

    def system_prompt(self, golden: ConversationGolden) -> str:
        trap_lines = "\n".join(f"- {TRAPS.get(t, t)}" for t in self.traps)
        parts = [
            "تو نقش یک کاربر واقعی ایرانی را بازی می‌کنی که با یک دستیار هوشمند گفتگو می‌کند.",
            f"هدف تو از این گفتگو: {golden.goal}",
            f"موقعیت: {golden.scenario}" if golden.scenario else "",
            f"سبک گفتار: {self.style}. {self.persona}".strip(),
            trap_lines,
            "قواعد: هر بار فقط پیامِ بعدیِ کاربر را بنویس، کوتاه و طبیعی. "
            "به محض این که به هدفت رسیدی یا مطمئن شدی به نتیجه نمی‌رسی، دیگر "
            f"سؤال تازه‌ای نپرس و فقط بنویس: {DONE}",
        ]
        return "\n".join(p for p in parts if p)
```

GoldenGenerator turns the user's own documentation into Golden expectations, ready to feed an AppEvaluator. Generated goldens carry their source chunk as `context`, so faithfulness is judged automatically.

Attributes:

| Name          | Type        | Description                                                                           |
| ------------- | ----------- | ------------------------------------------------------------------------------------- |
| `model`       | \`Model     | Callable                                                                              |
| `registers`   | `list[str]` | Question registers to rotate through (default is formal and colloquial Persian).      |
| `adversarial` | `bool`      | Mix digit scripts, Jalali dates, and Finglish into some questions (default is False). |

Methods:

| Name       | Description                       |
| ---------- | --------------------------------- |
| `generate` | Generates goldens from documents. |

Source code in `parsbench/appeval/generator.py`

```
class GoldenGenerator:
    """
    GoldenGenerator turns the user's own documentation into Golden
    expectations, ready to feed an AppEvaluator. Generated goldens carry
    their source chunk as `context`, so faithfulness is judged automatically.

    Attributes:
        model (Model | Callable | str, optional): The generator LLM (falls
            back to PARSBENCH_GENERATOR, then PARSBENCH_JUDGE env).
        registers (list[str], optional): Question registers to rotate through
            (default is formal and colloquial Persian).
        adversarial (bool): Mix digit scripts, Jalali dates, and Finglish
            into some questions (default is False).

    Methods:
        generate: Generates goldens from documents.
    """

    def __init__(
        self,
        model: Any = None,
        registers: list[str] | None = None,
        adversarial: bool = False,
    ):
        self.model = model
        self.registers = registers or ["رسمی", "محاوره‌ای"]
        self.adversarial = adversarial

    def generate(self, docs, n: int = 20) -> list[Golden]:
        """
        Generate goldens from the given documents.

        Parameters:
            docs (str | Path | list): A file path, glob pattern, directory, or
                a list of those. Text-like files only (.txt, .md, .rst, .html,
                .json).
            n (int, optional): The number of goldens to generate (default is 20).

        Returns:
            list[Golden]: The generated goldens.
        """
        llm = resolve_model(self.model, "PARSBENCH_GENERATOR", "PARSBENCH_JUDGE")
        if llm is None:
            raise ValueError(
                "generator model not configured — pass model= or set PARSBENCH_JUDGE."
            )
        chunks = [c for text in _read_docs(docs) for c in _chunks(text)]
        per_chunk = max(1, -(-n // len(chunks)))
        goldens: list[Golden] = []
        for i, chunk in enumerate(chunks):
            if len(goldens) >= n:
                break
            prompt = TESTGEN_FA.format(
                chunk=chunk,
                count=min(per_chunk, n - len(goldens)),
                register=self.registers[i % len(self.registers)],
                adversarial=_ADVERSARIAL_FA if self.adversarial else "",
            )
            reply = str(llm(prompt))
            match = re.search(r"\[.*\]", reply, re.DOTALL)
            if not match:
                continue
            try:
                rows = json.loads(match.group(0))
            except json.JSONDecodeError:
                continue
            for row in rows:
                if not isinstance(row, dict) or not row.get("input"):
                    continue
                goldens.append(
                    Golden(
                        input=row["input"],
                        output=row.get("output"),
                        contains=row.get("contains") or [],
                        context=[chunk],
                        tags=["generated"],
                    )
                )
        return goldens[:n]
```

## `generate(docs, n=20)`

Generate goldens from the given documents.

Parameters:

| Name   | Type  | Description                                        | Default |
| ------ | ----- | -------------------------------------------------- | ------- |
| `docs` | \`str | Path                                               | list\`  |
| `n`    | `int` | The number of goldens to generate (default is 20). | `20`    |

Returns:

| Type           | Description                            |
| -------------- | -------------------------------------- |
| `list[Golden]` | list\[Golden\]: The generated goldens. |

Source code in `parsbench/appeval/generator.py`

```
def generate(self, docs, n: int = 20) -> list[Golden]:
    """
    Generate goldens from the given documents.

    Parameters:
        docs (str | Path | list): A file path, glob pattern, directory, or
            a list of those. Text-like files only (.txt, .md, .rst, .html,
            .json).
        n (int, optional): The number of goldens to generate (default is 20).

    Returns:
        list[Golden]: The generated goldens.
    """
    llm = resolve_model(self.model, "PARSBENCH_GENERATOR", "PARSBENCH_JUDGE")
    if llm is None:
        raise ValueError(
            "generator model not configured — pass model= or set PARSBENCH_JUDGE."
        )
    chunks = [c for text in _read_docs(docs) for c in _chunks(text)]
    per_chunk = max(1, -(-n // len(chunks)))
    goldens: list[Golden] = []
    for i, chunk in enumerate(chunks):
        if len(goldens) >= n:
            break
        prompt = TESTGEN_FA.format(
            chunk=chunk,
            count=min(per_chunk, n - len(goldens)),
            register=self.registers[i % len(self.registers)],
            adversarial=_ADVERSARIAL_FA if self.adversarial else "",
        )
        reply = str(llm(prompt))
        match = re.search(r"\[.*\]", reply, re.DOTALL)
        if not match:
            continue
        try:
            rows = json.loads(match.group(0))
        except json.JSONDecodeError:
            continue
        for row in rows:
            if not isinstance(row, dict) or not row.get("input"):
                continue
            goldens.append(
                Golden(
                    input=row["input"],
                    output=row.get("output"),
                    contains=row.get("contains") or [],
                    context=[chunk],
                    tags=["generated"],
                )
            )
    return goldens[:n]
```

JudgeCalibrator measures how well a judge model agrees with human labels on a labeled sample, so judge scores can be trusted (or fixed) before they are published.

Attributes:

| Name    | Type    | Description |
| ------- | ------- | ----------- |
| `judge` | \`Model | Callable    |

Methods:

| Name        | Description                                                |
| ----------- | ---------------------------------------------------------- |
| `calibrate` | Scores the labeled items and computes agreement and kappa. |

Source code in `parsbench/appeval/calibration.py`

```
class JudgeCalibrator:
    """
    JudgeCalibrator measures how well a judge model agrees with human labels
    on a labeled sample, so judge scores can be trusted (or fixed) before
    they are published.

    Attributes:
        judge (Model | Callable | str, optional): The judge to calibrate
            (falls back to the PARSBENCH_JUDGE env var).

    Methods:
        calibrate: Scores the labeled items and computes agreement and kappa.
    """

    def __init__(self, judge: Any = None):
        self.judge = judge

    def calibrate(
        self,
        items: list[dict],
        prefer_concurrency: bool = False,
        n_workers: int = 4,
    ) -> CalibrationResult:
        """
        Score each labeled item with the judge and compare to the human label.

        Parameters:
            items (list[dict]): Items of the form
                `{"golden": Golden(...), "output": "...", "human": True}` where
                the golden triggers at least one judge check (output=, context=
                or refuses=).
            prefer_concurrency (bool, optional): Fan judge calls out over a
                thread pool (default is False); the judge callable must then
                be thread-safe.
            n_workers (int, optional): The number of workers for concurrent
                processing (default is 4).

        Returns:
            CalibrationResult: Agreement, Cohen's kappa, and disagreements.
        """
        if not items:
            raise ValueError("items is empty. You should provide at least one labeled item.")
        judge = resolve_model(self.judge, "PARSBENCH_JUDGE")
        if judge is None:
            raise ValueError(
                "calibrate needs a judge — pass judge= or set PARSBENCH_JUDGE."
            )

        def score_item(item) -> tuple[str, bool, bool, str | None]:
            golden = item["golden"]
            golden = golden if isinstance(golden, Golden) else Golden.from_dict(golden)
            trace = Trace(final_output=str(item["output"]))
            results = [
                r for r in run_checks(golden, trace, judge=judge)
                if r.check in JUDGE_CHECKS and not r.skipped
            ]
            if not results:
                raise ValueError(
                    f"golden {golden.label!r} triggers no judge check — it needs "
                    "output=, context= or refuses=."
                )
            judged = all(r.passed for r in results)
            reason = next((r.reason for r in results if not r.passed), results[0].reason)
            return golden.label, judged, bool(item["human"]), reason

        if prefer_concurrency and n_workers > 1:
            from concurrent.futures import ThreadPoolExecutor

            with ThreadPoolExecutor(max_workers=n_workers) as pool:
                scored = list(pool.map(score_item, items))
        else:
            scored = [score_item(item) for item in items]

        pairs = [(judged, human) for _, judged, human, _ in scored]
        disagreements = [
            f"{label}: judge={'pass' if judged else 'fail'} "
            f"human={'pass' if human else 'fail'} — {reason}"
            for label, judged, human, reason in scored
            if judged != human
        ]

        n = len(pairs)
        agreement = sum(j == h for j, h in pairs) / n
        # Cohen's kappa from marginals
        judge_yes = sum(j for j, _ in pairs) / n
        human_yes = sum(h for _, h in pairs) / n
        expected = judge_yes * human_yes + (1 - judge_yes) * (1 - human_yes)
        kappa = 0.0 if expected == 1.0 else (agreement - expected) / (1 - expected)
        return CalibrationResult(
            n=n, agreement=agreement, kappa=kappa, disagreements=disagreements
        )
```

## `calibrate(items, prefer_concurrency=False, n_workers=4)`

Score each labeled item with the judge and compare to the human label.

Parameters:

| Name                 | Type         | Description                                                                                                                                                   | Default    |
| -------------------- | ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `items`              | `list[dict]` | Items of the form {"golden": Golden(...), "output": "...", "human": True} where the golden triggers at least one judge check (output=, context= or refuses=). | *required* |
| `prefer_concurrency` | `bool`       | Fan judge calls out over a thread pool (default is False); the judge callable must then be thread-safe.                                                       | `False`    |
| `n_workers`          | `int`        | The number of workers for concurrent processing (default is 4).                                                                                               | `4`        |

Returns:

| Name                | Type                | Description                                  |
| ------------------- | ------------------- | -------------------------------------------- |
| `CalibrationResult` | `CalibrationResult` | Agreement, Cohen's kappa, and disagreements. |

Source code in `parsbench/appeval/calibration.py`

```
def calibrate(
    self,
    items: list[dict],
    prefer_concurrency: bool = False,
    n_workers: int = 4,
) -> CalibrationResult:
    """
    Score each labeled item with the judge and compare to the human label.

    Parameters:
        items (list[dict]): Items of the form
            `{"golden": Golden(...), "output": "...", "human": True}` where
            the golden triggers at least one judge check (output=, context=
            or refuses=).
        prefer_concurrency (bool, optional): Fan judge calls out over a
            thread pool (default is False); the judge callable must then
            be thread-safe.
        n_workers (int, optional): The number of workers for concurrent
            processing (default is 4).

    Returns:
        CalibrationResult: Agreement, Cohen's kappa, and disagreements.
    """
    if not items:
        raise ValueError("items is empty. You should provide at least one labeled item.")
    judge = resolve_model(self.judge, "PARSBENCH_JUDGE")
    if judge is None:
        raise ValueError(
            "calibrate needs a judge — pass judge= or set PARSBENCH_JUDGE."
        )

    def score_item(item) -> tuple[str, bool, bool, str | None]:
        golden = item["golden"]
        golden = golden if isinstance(golden, Golden) else Golden.from_dict(golden)
        trace = Trace(final_output=str(item["output"]))
        results = [
            r for r in run_checks(golden, trace, judge=judge)
            if r.check in JUDGE_CHECKS and not r.skipped
        ]
        if not results:
            raise ValueError(
                f"golden {golden.label!r} triggers no judge check — it needs "
                "output=, context= or refuses=."
            )
        judged = all(r.passed for r in results)
        reason = next((r.reason for r in results if not r.passed), results[0].reason)
        return golden.label, judged, bool(item["human"]), reason

    if prefer_concurrency and n_workers > 1:
        from concurrent.futures import ThreadPoolExecutor

        with ThreadPoolExecutor(max_workers=n_workers) as pool:
            scored = list(pool.map(score_item, items))
    else:
        scored = [score_item(item) for item in items]

    pairs = [(judged, human) for _, judged, human, _ in scored]
    disagreements = [
        f"{label}: judge={'pass' if judged else 'fail'} "
        f"human={'pass' if human else 'fail'} — {reason}"
        for label, judged, human, reason in scored
        if judged != human
    ]

    n = len(pairs)
    agreement = sum(j == h for j, h in pairs) / n
    # Cohen's kappa from marginals
    judge_yes = sum(j for j, _ in pairs) / n
    human_yes = sum(h for _, h in pairs) / n
    expected = judge_yes * human_yes + (1 - judge_yes) * (1 - human_yes)
    kappa = 0.0 if expected == 1.0 else (agreement - expected) / (1 - expected)
    return CalibrationResult(
        n=n, agreement=agreement, kappa=kappa, disagreements=disagreements
    )
```

The result of calibrating a judge against human labels.

Attributes:

| Name            | Type        | Description                                    |
| --------------- | ----------- | ---------------------------------------------- |
| `n`             | `int`       | The number of labeled items.                   |
| `agreement`     | `float`     | Fraction where judge pass/fail == human label. |
| `kappa`         | `float`     | Cohen's kappa vs human labels.                 |
| `disagreements` | `list[str]` | Readable descriptions for error analysis.      |

Source code in `parsbench/appeval/calibration.py`

```
@dataclass
class CalibrationResult:
    """
    The result of calibrating a judge against human labels.

    Attributes:
        n (int): The number of labeled items.
        agreement (float): Fraction where judge pass/fail == human label.
        kappa (float): Cohen's kappa vs human labels.
        disagreements (list[str]): Readable descriptions for error analysis.
    """

    n: int
    agreement: float
    kappa: float
    disagreements: list[str]

    @classmethod
    def from_dict(cls, data: dict) -> "CalibrationResult":
        return cls(**data)

    def to_dict(self) -> dict:
        from dataclasses import asdict

        return asdict(self)

    def __str__(self) -> str:
        return (
            f"judge-vs-human on {self.n} items: agreement={self.agreement:.2f}, "
            f"kappa={self.kappa:.2f}, disagreements={len(self.disagreements)}"
        )
```

Unify codepoints/digits, read ZWNJ as a space, drop thousands commas, collapse whitespace.

Source code in `parsbench/appeval/normalize.py`

```
def normalize(text: str) -> str:
    """Unify codepoints/digits, read ZWNJ as a space, drop thousands commas,
    collapse whitespace."""
    text = str(text).translate(_CHAR_MAP).replace(ZWNJ, " ")
    text = _DIGIT_COMMA.sub("", text)
    return re.sub(r"\s+", " ", text).strip()
```

Substring check that survives digit scripts, ZWNJ and spacing variants.

Source code in `parsbench/appeval/normalize.py`

```
def contains_normalized(haystack: str, needle: str) -> bool:
    """Substring check that survives digit scripts, ZWNJ and spacing variants."""
    return _contains_norm(normalize(haystack), needle)
```

Parse '۲۵۰ هزار تومان' → (2_500_000.0, 'rial'). Returns (value, unit|None).

Source code in `parsbench/appeval/normalize.py`

```
def parse_number(value) -> tuple[float, str | None] | None:
    """Parse '۲۵۰ هزار تومان' → (2_500_000.0, 'rial'). Returns (value, unit|None)."""
    parsed = _parse_amount(value)
    return None if parsed is None else (parsed[0], parsed[1])
```

Source code in `parsbench/appeval/normalize.py`

```
def numbers_equal(a, b) -> bool:
    pa, pb = _parse_amount(a), _parse_amount(b)
    if pa is None or pb is None:
        return False
    (va, ua, _), (vb, ub, _) = pa, pb
    if ua and ub:
        return va == vb
    # unit missing on one side — accept either rial/toman reading
    if ua or ub:
        return va == vb or va == vb * 10 or vb == va * 10
    return va == vb
```

Does the text state this amount, in any unit/scale/digit-script? amount_in('قیمت ۲٬۵۰۰٬۰۰۰ ریال است', '250 هزار تومان') → True.

Source code in `parsbench/appeval/normalize.py`

```
def amount_in(haystack: str, needle) -> bool:
    """Does the text state this amount, in any unit/scale/digit-script?
    amount_in('قیمت ۲٬۵۰۰٬۰۰۰ ریال است', '250 هزار تومان') → True."""
    return _amount_in_norm(normalize(haystack), needle)
```

Parse a date(-time) string to a Gregorian (y, m, d). Year \<1600 → Jalali. The whole string must be the date — ranges and prose return None.

Source code in `parsbench/appeval/normalize.py`

```
def parse_date(value) -> tuple[int, int, int] | None:
    """Parse a date(-time) string to a Gregorian (y, m, d). Year <1600 → Jalali.
    The whole string must be the date — ranges and prose return None."""
    m = _DATE_ONLY.match(normalize(value))
    if not m:
        return None
    return _to_gregorian(int(m.group(1)), int(m.group(2)), int(m.group(3)))
```

Source code in `parsbench/appeval/normalize.py`

```
def dates_equal(a, b) -> bool:
    pa, pb = parse_date(a), parse_date(b)
    return pa is not None and pa == pb
```

Does the text mention this date, in either calendar?

Source code in `parsbench/appeval/normalize.py`

```
def date_in(haystack: str, needle) -> bool:
    """Does the text mention this date, in either calendar?"""
    return _date_in_norm(normalize(haystack), needle)
```

Equivalence chain for text expectations (contains / not_contains): normalized substring, then money-equivalence, then calendar-equivalence. Pass normalized=True when the haystack is already normalize()d.

Source code in `parsbench/appeval/normalize.py`

```
def text_matches(haystack: str, needle, *, normalized: bool = False) -> bool:
    """Equivalence chain for text expectations (contains / not_contains):
    normalized substring, then money-equivalence, then calendar-equivalence.
    Pass normalized=True when the haystack is already normalize()d."""
    h = haystack if normalized else normalize(haystack)
    n = needle if isinstance(needle, str) else str(needle)
    if _contains_norm(h, n):
        return True
    if _amount_in_norm(h, n):
        return True
    return _date_in_norm(h, n)
```

Equivalence chain for tool arguments: date → number → normalized string.

Source code in `parsbench/appeval/normalize.py`

```
def values_equal(a, b) -> bool:
    """Equivalence chain for tool arguments: date → number → normalized string."""
    da, db = parse_date(a), parse_date(b)
    if da or db:
        return da == db
    if numbers_equal(a, b):
        return True
    return normalize(a) == normalize(b)
```

# Benchmarks

Bases: `ABC`

This abstract class defines the structure for a benchmarking task. Subclasses of Benchmark must implement the 'run' method, which takes in various parameters related to the benchmarking task and returns a BenchmarkResult object.

Methods:

| Name  | Description                                                                                                                                           |
| ----- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `run` | Abstract method that must be implemented by subclasses. It runs the benchmarking task with the given parameters and returns a BenchmarkResult object. |

Source code in `parsbench/benchmarks/base.py`

```
class Benchmark(ABC):
    """
    This abstract class defines the structure for a benchmarking task. Subclasses of Benchmark must implement the 'run' method, which takes in various parameters related to the benchmarking task and returns a BenchmarkResult object.

    Methods:
        run: Abstract method that must be implemented by subclasses. It runs the benchmarking task with the given parameters and returns a BenchmarkResult object.
    """

    @abstractmethod
    def run(
        self,
        prompt_lang: str = "fa",
        prompt_shots: list[int] | None = None,
        n_first: int | None = None,
        sort_by_score: bool = True,
        save_matches: bool = False,
        save_evaluation: bool = False,
        save_benchmark: bool = False,
        output_path: str = None,
        skip_existing_matches: bool = False,
        prefer_concurrency: bool = True,
        n_workers: int = 4,
    ) -> BenchmarkResult:
        """
        Abstract method that must be implemented by subclasses. It runs the benchmarking task with the given parameters and returns a BenchmarkResult object.

        Parameters:
            prompt_lang (str, optional): The language of the prompt (default is "fa").
            prompt_shots (list[int], optional): The list of prompt shots to evaluate (default is None).
            n_first (int, optional): The number of initial prompts to consider (default is 200).
            sort_by_score (bool, optional): Whether to sort the model benchmarks by average score (default is True).
            save_matches (bool, optional): Flag to save the generated matches (default is False).
            save_evaluation (bool, optional): Flag to save the evaluation results (default is False).
            skip_existing_matches (bool, optional): Flag to skip already generated matches in the output path (default is False).
            output_path (str, optional): The output path to save the matches and evaluation results.
            prefer_concurrency (bool, optional): The flag to use concurrent processing if the model and task support that (default is True).
            n_workers (int, optional): The number of workers for concurrent processing (default is 4).

        Returns:
            BenchmarkResult: An object containing the benchmarking results.
        """
        pass
```

## `run(prompt_lang='fa', prompt_shots=None, n_first=None, sort_by_score=True, save_matches=False, save_evaluation=False, save_benchmark=False, output_path=None, skip_existing_matches=False, prefer_concurrency=True, n_workers=4)`

Abstract method that must be implemented by subclasses. It runs the benchmarking task with the given parameters and returns a BenchmarkResult object.

Parameters:

| Name                    | Type        | Description                                                                                 | Default |
| ----------------------- | ----------- | ------------------------------------------------------------------------------------------- | ------- |
| `prompt_lang`           | `str`       | The language of the prompt (default is "fa").                                               | `'fa'`  |
| `prompt_shots`          | `list[int]` | The list of prompt shots to evaluate (default is None).                                     | `None`  |
| `n_first`               | `int`       | The number of initial prompts to consider (default is 200).                                 | `None`  |
| `sort_by_score`         | `bool`      | Whether to sort the model benchmarks by average score (default is True).                    | `True`  |
| `save_matches`          | `bool`      | Flag to save the generated matches (default is False).                                      | `False` |
| `save_evaluation`       | `bool`      | Flag to save the evaluation results (default is False).                                     | `False` |
| `skip_existing_matches` | `bool`      | Flag to skip already generated matches in the output path (default is False).               | `False` |
| `output_path`           | `str`       | The output path to save the matches and evaluation results.                                 | `None`  |
| `prefer_concurrency`    | `bool`      | The flag to use concurrent processing if the model and task support that (default is True). | `True`  |
| `n_workers`             | `int`       | The number of workers for concurrent processing (default is 4).                             | `4`     |

Returns:

| Name              | Type              | Description                                    |
| ----------------- | ----------------- | ---------------------------------------------- |
| `BenchmarkResult` | `BenchmarkResult` | An object containing the benchmarking results. |

Source code in `parsbench/benchmarks/base.py`

```
@abstractmethod
def run(
    self,
    prompt_lang: str = "fa",
    prompt_shots: list[int] | None = None,
    n_first: int | None = None,
    sort_by_score: bool = True,
    save_matches: bool = False,
    save_evaluation: bool = False,
    save_benchmark: bool = False,
    output_path: str = None,
    skip_existing_matches: bool = False,
    prefer_concurrency: bool = True,
    n_workers: int = 4,
) -> BenchmarkResult:
    """
    Abstract method that must be implemented by subclasses. It runs the benchmarking task with the given parameters and returns a BenchmarkResult object.

    Parameters:
        prompt_lang (str, optional): The language of the prompt (default is "fa").
        prompt_shots (list[int], optional): The list of prompt shots to evaluate (default is None).
        n_first (int, optional): The number of initial prompts to consider (default is 200).
        sort_by_score (bool, optional): Whether to sort the model benchmarks by average score (default is True).
        save_matches (bool, optional): Flag to save the generated matches (default is False).
        save_evaluation (bool, optional): Flag to save the evaluation results (default is False).
        skip_existing_matches (bool, optional): Flag to skip already generated matches in the output path (default is False).
        output_path (str, optional): The output path to save the matches and evaluation results.
        prefer_concurrency (bool, optional): The flag to use concurrent processing if the model and task support that (default is True).
        n_workers (int, optional): The number of workers for concurrent processing (default is 4).

    Returns:
        BenchmarkResult: An object containing the benchmarking results.
    """
    pass
```

Bases: `Benchmark`

CustomBenchmark class represents a custom benchmarking task that extends the Benchmark abstract class. It defines the run method to execute the benchmarking process for a given list of models and tasks.

Attributes:

| Name     | Type          | Description                                              |
| -------- | ------------- | -------------------------------------------------------- |
| `models` | `list[Model]` | The list of models to evaluate in the benchmarking task. |
| `tasks`  | `list[Task]`  | The list of tasks to evaluate with the models.           |

Methods:

| Name  | Description                                                                                                                      |
| ----- | -------------------------------------------------------------------------------------------------------------------------------- |
| `run` | Executes the benchmarking process for the specified models and tasks, generating evaluation results for each model on each task. |

Source code in `parsbench/benchmarks/custom_benchmark.py`

```
class CustomBenchmark(Benchmark):
    """
    CustomBenchmark class represents a custom benchmarking task that extends the Benchmark abstract class. It defines the run method to execute the benchmarking process for a given list of models and tasks.

    Attributes:
        models (list[Model]): The list of models to evaluate in the benchmarking task.
        tasks (list[Task]): The list of tasks to evaluate with the models.

    Methods:
        run: Executes the benchmarking process for the specified models and tasks, generating evaluation results for each model on each task.
    """

    def __init__(
        self,
        models: list[Model],
        tasks: list[Task],
    ):
        self.models = models
        self.tasks = tasks

    def run(
        self,
        prompt_lang: str = "fa",
        prompt_shots: list[int] | None = None,
        n_first: int | None = None,
        sort_by_score: bool = True,
        save_matches: bool = False,
        save_evaluation: bool = False,
        save_benchmark: bool = False,
        output_path: str = None,
        skip_existing_matches: bool = False,
        prefer_concurrency: bool = True,
        n_workers: int = 4,
    ) -> BenchmarkResult:
        """
        Run the benchmarking process for the given models and tasks.

        Parameters:
            prompt_lang (str, optional): The language of the prompt (default is "fa").
            prompt_shots (list[int], optional): The list of prompt shots to evaluate (default is None).
            n_first (int, optional): The number of initial prompts to consider (default is 200).
            sort_by_score (bool, optional): Whether to sort the model benchmarks by average score (default is True).
            save_matches (bool, optional): Flag to save the generated matches (default is False).
            save_evaluation (bool, optional): Flag to save the evaluation results (default is False).
            output_path (str, optional): The output path to save the matches and evaluation results.
            skip_existing_matches (bool, optional): Flag to skip already generated matches in the output path (default is False).
            prefer_concurrency (bool, optional): The flag to use concurrent processing if the model and task support that (default is True).
            n_workers (int, optional): The number of workers for concurrent processing (default is 4).

        Returns:
            BenchmarkResult: The result of the benchmarking process.
        """

        model_evaluations: dict[str, list[EvaluationResult]] = defaultdict(list)

        for task in self.tasks:
            print(f"Evaluating {task.task_name}:")

            if inspect.isclass(task):
                if issubclass(task, Task):
                    task: Task = task()
                else:
                    raise TypeError(
                        f"{task} is not a subclass/instance of the Task class."
                    )

            with task:
                for model in self.models:
                    print(f"Model: {model.model_name}")

                    evaluation_results = task.evaluate(
                        model=model,
                        prompt_lang=prompt_lang,
                        prompt_shots=prompt_shots,
                        n_first=n_first,
                        save_matches=save_matches,
                        save_evaluation=save_evaluation,
                        output_path=output_path,
                        skip_existing_matches=skip_existing_matches,
                        prefer_concurrency=prefer_concurrency,
                        n_workers=n_workers,
                    )
                    model_evaluations[model.model_name].extend(evaluation_results)

        model_benchmarks = [
            ModelBenchmarkResult(
                model_name=model_name,
                evaluation_results=evaluation_results,
            )
            for model_name, evaluation_results in model_evaluations.items()
        ]

        if sort_by_score:
            model_benchmarks.sort(key=lambda mb: mb.average_score, reverse=True)

        benchmark_result = BenchmarkResult(model_benchmarks=model_benchmarks)

        if save_benchmark:
            benchmark_result.save(output_path)

        return benchmark_result
```

## `run(prompt_lang='fa', prompt_shots=None, n_first=None, sort_by_score=True, save_matches=False, save_evaluation=False, save_benchmark=False, output_path=None, skip_existing_matches=False, prefer_concurrency=True, n_workers=4)`

Run the benchmarking process for the given models and tasks.

Parameters:

| Name                    | Type        | Description                                                                                 | Default |
| ----------------------- | ----------- | ------------------------------------------------------------------------------------------- | ------- |
| `prompt_lang`           | `str`       | The language of the prompt (default is "fa").                                               | `'fa'`  |
| `prompt_shots`          | `list[int]` | The list of prompt shots to evaluate (default is None).                                     | `None`  |
| `n_first`               | `int`       | The number of initial prompts to consider (default is 200).                                 | `None`  |
| `sort_by_score`         | `bool`      | Whether to sort the model benchmarks by average score (default is True).                    | `True`  |
| `save_matches`          | `bool`      | Flag to save the generated matches (default is False).                                      | `False` |
| `save_evaluation`       | `bool`      | Flag to save the evaluation results (default is False).                                     | `False` |
| `output_path`           | `str`       | The output path to save the matches and evaluation results.                                 | `None`  |
| `skip_existing_matches` | `bool`      | Flag to skip already generated matches in the output path (default is False).               | `False` |
| `prefer_concurrency`    | `bool`      | The flag to use concurrent processing if the model and task support that (default is True). | `True`  |
| `n_workers`             | `int`       | The number of workers for concurrent processing (default is 4).                             | `4`     |

Returns:

| Name              | Type              | Description                             |
| ----------------- | ----------------- | --------------------------------------- |
| `BenchmarkResult` | `BenchmarkResult` | The result of the benchmarking process. |

Source code in `parsbench/benchmarks/custom_benchmark.py`

```
def run(
    self,
    prompt_lang: str = "fa",
    prompt_shots: list[int] | None = None,
    n_first: int | None = None,
    sort_by_score: bool = True,
    save_matches: bool = False,
    save_evaluation: bool = False,
    save_benchmark: bool = False,
    output_path: str = None,
    skip_existing_matches: bool = False,
    prefer_concurrency: bool = True,
    n_workers: int = 4,
) -> BenchmarkResult:
    """
    Run the benchmarking process for the given models and tasks.

    Parameters:
        prompt_lang (str, optional): The language of the prompt (default is "fa").
        prompt_shots (list[int], optional): The list of prompt shots to evaluate (default is None).
        n_first (int, optional): The number of initial prompts to consider (default is 200).
        sort_by_score (bool, optional): Whether to sort the model benchmarks by average score (default is True).
        save_matches (bool, optional): Flag to save the generated matches (default is False).
        save_evaluation (bool, optional): Flag to save the evaluation results (default is False).
        output_path (str, optional): The output path to save the matches and evaluation results.
        skip_existing_matches (bool, optional): Flag to skip already generated matches in the output path (default is False).
        prefer_concurrency (bool, optional): The flag to use concurrent processing if the model and task support that (default is True).
        n_workers (int, optional): The number of workers for concurrent processing (default is 4).

    Returns:
        BenchmarkResult: The result of the benchmarking process.
    """

    model_evaluations: dict[str, list[EvaluationResult]] = defaultdict(list)

    for task in self.tasks:
        print(f"Evaluating {task.task_name}:")

        if inspect.isclass(task):
            if issubclass(task, Task):
                task: Task = task()
            else:
                raise TypeError(
                    f"{task} is not a subclass/instance of the Task class."
                )

        with task:
            for model in self.models:
                print(f"Model: {model.model_name}")

                evaluation_results = task.evaluate(
                    model=model,
                    prompt_lang=prompt_lang,
                    prompt_shots=prompt_shots,
                    n_first=n_first,
                    save_matches=save_matches,
                    save_evaluation=save_evaluation,
                    output_path=output_path,
                    skip_existing_matches=skip_existing_matches,
                    prefer_concurrency=prefer_concurrency,
                    n_workers=n_workers,
                )
                model_evaluations[model.model_name].extend(evaluation_results)

    model_benchmarks = [
        ModelBenchmarkResult(
            model_name=model_name,
            evaluation_results=evaluation_results,
        )
        for model_name, evaluation_results in model_evaluations.items()
    ]

    if sort_by_score:
        model_benchmarks.sort(key=lambda mb: mb.average_score, reverse=True)

    benchmark_result = BenchmarkResult(model_benchmarks=model_benchmarks)

    if save_benchmark:
        benchmark_result.save(output_path)

    return benchmark_result
```

Bases: `CustomBenchmark`

This benchmark class includes all existing tasks which use ParsiNLU datasets.

Attributes:

| Name     | Type          | Description                                              |
| -------- | ------------- | -------------------------------------------------------- |
| `models` | `list[Model]` | The list of models to evaluate in the benchmarking task. |

Methods:

| Name  | Description                                                                                                            |
| ----- | ---------------------------------------------------------------------------------------------------------------------- |
| `run` | Executes the benchmarking process for the specified models, generating evaluation results for each model on each task. |

Source code in `parsbench/benchmarks/parsinlu_benchmark.py`

```
class ParsiNLUBenchmark(CustomBenchmark):
    """
    This benchmark class includes all existing tasks which use ParsiNLU datasets.

    Attributes:
        models (list[Model]): The list of models to evaluate in the benchmarking task.

    Methods:
        run: Executes the benchmarking process for the specified models, generating evaluation results for each model on each task.
    """

    def __init__(self, models: list[Model]):
        tasks = [
            ParsiNLUEntailment,
            ParsiNLUMachineTranslationEnFa,
            ParsiNLUMachineTranslationFaEn,
            ParsiNLUMultipleChoice,
            ParsiNLUReadingComprehension,
            ParsiNLUSentimentAnalysis,
        ]
        super().__init__(models, tasks)
```

Represents the results of benchmarking a model across multiple evaluations.

Attributes:

| Name                 | Type                     | Description                                                                           |
| -------------------- | ------------------------ | ------------------------------------------------------------------------------------- |
| `model_name`         | `str`                    | The name of the model being benchmarked.                                              |
| `evaluation_results` | `list[EvaluationResult]` | A list of EvaluationResult objects representing the evaluation results for the model. |

Source code in `parsbench/benchmarks/benchmark_result.py`

```
@dataclass
class ModelBenchmarkResult:
    """
    Represents the results of benchmarking a model across multiple evaluations.

    Attributes:
        model_name (str): The name of the model being benchmarked.
        evaluation_results (list[EvaluationResult]): A list of EvaluationResult objects representing the evaluation results for the model.
    """

    model_name: str
    evaluation_results: list[EvaluationResult]

    @classmethod
    def from_dict(cls, data: dict) -> "ModelBenchmarkResult":
        evaluation_results = [
            EvaluationResult.from_dict(task) for task in data.pop("evaluation_results")
        ]
        return cls(**data, evaluation_results=evaluation_results)

    def to_dict(self) -> dict:
        return {
            **asdict(self),
            "evaluation_results": [e.to_dict() for e in self.evaluation_results],
        }

    def to_pandas(self) -> pd.DataFrame:
        return pd.concat([er.to_pandas() for er in self.evaluation_results])

    def __str__(self) -> str:
        text = f"Model: {self.model_name}\nEvaluation Results:\n"
        for er in self.evaluation_results:
            text += f"- {er.task_name}"
            if er.sub_task:
                text += f" ({er.sub_task}):\n"
            else:
                text += ":\n"
            for psr in er.prompt_shot_results:
                text += f"  - {psr.n_shots}-shot prompt: {psr.score:.4f}\n"
        return text.strip("\n")

    @property
    def average_score(self) -> float:
        return sum([er.average_score for er in self.evaluation_results]) / len(
            self.evaluation_results
        )
```

Represents the results of benchmarking multiple models across various evaluations.

Attributes:

| Name               | Type                         | Description                                                                               |
| ------------------ | ---------------------------- | ----------------------------------------------------------------------------------------- |
| `model_benchmarks` | `list[ModelBenchmarkResult]` | A list of ModelBenchmarkResult objects representing the benchmark results for each model. |

Source code in `parsbench/benchmarks/benchmark_result.py`

```
@dataclass
class BenchmarkResult:
    """
    Represents the results of benchmarking multiple models across various evaluations.

    Attributes:
        model_benchmarks (list[ModelBenchmarkResult]): A list of ModelBenchmarkResult objects representing the benchmark results for each model.
    """

    model_benchmarks: list[ModelBenchmarkResult]

    @classmethod
    def from_file(cls, path: str) -> "BenchmarkResult":
        with jsonlines.open(path, "r") as reader:
            model_benchmarks: list[ModelBenchmarkResult] = []
            for row in reader.iter(type=dict, skip_invalid=True):
                model_benchmarks.append(ModelBenchmarkResult.from_dict(row))

        return cls(model_benchmarks=model_benchmarks)

    @classmethod
    def from_evaluation_files(cls, path: str) -> "BenchmarkResult":
        models = [(d.name, d.path) for d in os.scandir(path) if d.is_dir()]
        model_benchmarks = []

        for model_name, model_path in models:
            eval_paths = [d.path for d in os.scandir(model_path) if d.is_dir()]

            evaluation_results = []

            for eval_path in eval_paths:
                eval_files = [
                    f
                    for f in os.scandir(eval_path)
                    if f.is_file() and f.name.startswith("evaluation")
                ]
                evaluation_results.extend(
                    [
                        EvaluationResult.from_file(eval_file.path)
                        for eval_file in eval_files
                    ]
                )

            model_benchmarks.append(
                ModelBenchmarkResult(
                    model_name=model_name,
                    evaluation_results=evaluation_results,
                )
            )

        return BenchmarkResult(model_benchmarks=model_benchmarks)

    @classmethod
    def from_matches_files(cls, path: str, rescore: bool = False) -> "BenchmarkResult":
        task_cls_mapping = {
            task_cls.task_name.replace("-", " "): task_cls
            for task_cls in load_all_tasks()
        }

        _with_subtask_pattern = re.compile(r"matches_([\w\s]+)_(\d+)_shot\.jsonl")
        _without_subtask_pattern = re.compile(r"matches_(\d+)_shot\.jsonl")

        matches_paths = glob.glob(f"{path}/*/*/matches*.jsonl")

        model_evals: list[tuple[str, str, str, TaskMatchGroup]] = []

        for match_path in matches_paths:
            match_file = os.path.basename(match_path)
            task_name = os.path.basename(os.path.dirname(match_path)).replace("_", " ")
            model_name = os.path.basename(os.path.dirname(os.path.dirname(match_path)))
            sub_task = None
            n_shots = 0

            if m := _with_subtask_pattern.match(match_file):
                sub_task = m.group(1)
                n_shots = int(m.group(2))
            elif m := _without_subtask_pattern.match(match_file):
                n_shots = int(m.group(1))
            else:
                raise Exception(
                    f"Matches file '{match_file}' doesn't match the expected pattern."
                )

            task_matches = TaskMatchGroup.from_file(
                Path(match_path).parent, n_shots=n_shots, sub_task=sub_task
            )
            assert (
                task_name in task_cls_mapping
            ), f"No task class found for '{task_name}'."

            model_evals.append((model_name, task_name, sub_task, task_matches))

        # groupby only groups consecutive keys, so sort by (model, task) first.
        model_evals.sort(key=lambda t: (t[0], t[1]))

        model_benchmarks: list[ModelBenchmarkResult] = []

        for model_name, task_evals in itertools.groupby(
            model_evals, key=lambda t: t[0]
        ):
            print(f"Model: {model_name}")
            evaluation_results: list[EvaluationResult] = []

            for task_name, task_matches_group in itertools.groupby(
                task_evals, key=lambda t: t[1]
            ):
                print(f"Re-scoring {task_name}:")
                task = task_cls_mapping[task_name]()
                prompt_shot_evals = defaultdict(list)

                for _, _, sub_task, task_matches in task_matches_group:
                    print(f"{sub_task} {task_matches.n_shots}-shot prompt:")
                    if rescore:
                        task_matches = task.score_matches(task_matches)

                    score = task.get_overall_score(task_matches)

                    prompt_shot_evals[sub_task].append(
                        PromptShotEvaluationResult(
                            n_shots=task_matches.n_shots, score=score
                        )
                    )

                evaluation_results.extend(
                    EvaluationResult(
                        model_name=model_name,
                        task_name=task_name,
                        task_category=task.task_category,
                        score_name=task.score_name,
                        prompt_shot_results=prompt_shot_results,
                        sub_task=sub_task,
                    )
                    for sub_task, prompt_shot_results in prompt_shot_evals.items()
                )

            model_benchmarks.append(
                ModelBenchmarkResult(
                    model_name=model_name, evaluation_results=evaluation_results
                )
            )
            print("-" * 10)

        return BenchmarkResult(model_benchmarks=model_benchmarks)

    @classmethod
    def from_dict(cls, data: dict) -> "BenchmarkResult":
        model_benchmarks = [
            ModelBenchmarkResult.from_dict(mbr) for mbr in data.pop("model_benchmarks")
        ]
        return cls(**data, model_benchmarks=model_benchmarks)

    def to_dict(self) -> dict:
        return {
            **asdict(self),
            "model_benchmarks": [mb.to_dict() for mb in self.model_benchmarks],
        }

    def to_pandas(self, pivot: bool = False) -> pd.DataFrame:
        df = pd.concat([mb.to_pandas() for mb in self.model_benchmarks])
        if pivot:
            return df.pivot(
                index=["task_category", "task_name", "sub_task", "score_name"],
                columns=["model_name", "n_shots"],
                values=["score"],
            )
        return df

    def show_radar_plot(self, title="Radar Plot"):
        data = []
        categories = set()

        for mb in self.model_benchmarks:
            values = []
            for _, evals in groupby(mb.evaluation_results, key=lambda e: e.task_name):
                evals = list(evals)
                score = sum(e.average_score for e in evals) / len(evals)
                values.append(score)

            data.append({"name": mb.model_name, "values": values})
            categories |= set(e.task_name for e in mb.evaluation_results)

        _radar_plot(data, categories, title)

    def show_bar_plot(self, title="Bar Plot"):
        data = []
        categories = set()

        for mb in self.model_benchmarks:
            values = []
            for _, evals in groupby(mb.evaluation_results, key=lambda e: e.task_name):
                evals = list(evals)
                score = sum(e.average_score for e in evals) / len(evals)
                values.append(score)

            data.append({"name": mb.model_name, "values": values})
            categories |= set(e.task_name for e in mb.evaluation_results)

        _bar_plot(data, categories, title)

    def save(self, path: str):
        benchmark_path = Path(path) / "benchmark.jsonl"

        with jsonlines.open(benchmark_path, "w") as writer:
            for mb in self.model_benchmarks:
                writer.write(mb.to_dict())

    def __str__(self) -> str:
        text = ""
        for mb in self.model_benchmarks:
            text += str(mb) + "\n" + "-" * 10 + "\n"
        return text.strip("\n")
```

Merge multiple BenchmarkResult objects into a single BenchmarkResult object.

Parameters:

| Name              | Type                    | Description                                                                              | Default    |
| ----------------- | ----------------------- | ---------------------------------------------------------------------------------------- | ---------- |
| `benchmarks`      | `list[BenchmarkResult]` | A list of BenchmarkResult objects to merge.                                              | *required* |
| `sort`            | `bool`                  | Whether to sort the merged ModelBenchmarkResult list by average score. Defaults to True. | `True`     |
| `keep_duplicates` | `bool`                  | Whether to keep duplicate model names in the merged list. Defaults to False.             | `False`    |

Returns:

| Name              | Type              | Description                                                                   |
| ----------------- | ----------------- | ----------------------------------------------------------------------------- |
| `BenchmarkResult` | `BenchmarkResult` | A new BenchmarkResult object containing the merged ModelBenchmarkResult list. |

Source code in `parsbench/benchmarks/benchmark_result.py`

```
def merge_benchmark_results(
    benchmarks: list[BenchmarkResult], sort: bool = True, keep_duplicates: bool = False
) -> BenchmarkResult:
    """
    Merge multiple BenchmarkResult objects into a single BenchmarkResult object.

    Parameters:
        benchmarks (list[BenchmarkResult]): A list of BenchmarkResult objects to merge.
        sort (bool, optional): Whether to sort the merged ModelBenchmarkResult list by average score. Defaults to True.
        keep_duplicates (bool, optional): Whether to keep duplicate model names in the merged list. Defaults to False.

    Returns:
        BenchmarkResult: A new BenchmarkResult object containing the merged ModelBenchmarkResult list.
    """
    model_benchmarks: list[ModelBenchmarkResult] = []
    for benchmark in benchmarks:
        model_benchmarks.extend(benchmark.model_benchmarks)

    if not keep_duplicates:
        model_names = set()
        deduped: list[ModelBenchmarkResult] = []
        for mbr in model_benchmarks:
            if mbr.model_name not in model_names:
                model_names.add(mbr.model_name)
                deduped.append(mbr)
        model_benchmarks = deduped

    if sort:
        model_benchmarks.sort(key=lambda m: m.average_score, reverse=True)

    return BenchmarkResult(model_benchmarks=model_benchmarks)
```

This function generates leaderboard data from the benchmark result object.

Parameters:

| Name               | Type              | Description                         | Default    |
| ------------------ | ----------------- | ----------------------------------- | ---------- |
| `benchmark_result` | `BenchmarkResult` | BenchmarkResult object.             | *required* |
| `leaderboard_path` | `str`             | Path to store the leaderboard data. | *required* |

Source code in `parsbench/benchmarks/benchmark_result.py`

```
def build_leaderboard_from_benchmark(
    benchmark_result: BenchmarkResult, leaderboard_path: str
):
    """
    This function generates leaderboard data from the benchmark result object.

    Parameters:
        benchmark_result (BenchmarkResult): BenchmarkResult object.
        leaderboard_path (str): Path to store the leaderboard data.
    """
    requests_path = Path(leaderboard_path) / "requests"
    results_path = Path(leaderboard_path) / "results"

    requests_path.mkdir(exist_ok=True)
    results_path.mkdir(exist_ok=True)

    now = datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")

    for mb in benchmark_result.model_benchmarks:
        model_name = mb.model_name

        os.makedirs(results_path / model_name, exist_ok=True)

        request = {
            "model": model_name,
            "base_model": "",
            "revision": "main",
            "private": False,
            "precision": "?",
            "weight_type": "Original",
            "status": "FINISHED",
            "submitted_time": now,
            "model_type": "\ud83d\udfe2 : pretrained",
            "likes": 0,
            "params": 0.1,
            "license": "custom",
        }
        with open(
            requests_path / f"{model_name}_eval_request_nshot.json", "wt"
        ) as writer:
            writer.write(json.dumps(request))

        result = {
            "config": {"model_dtype": "", "model_name": model_name, "model_sha": ""},
            "results": {
                er.task_name: {er.score_name: round(er.max_score, 3)}
                for er in mb.evaluation_results
            },
        }

        with open(results_path / model_name / f"results_{now}.json", "wt") as writer:
            writer.write(json.dumps(result))
```

# Integrations

Source code in `parsbench/integrations/openai_agents.py`

```
def to_trace(run_result) -> Trace:
    items = (
        run_result.to_input_list()
        if hasattr(run_result, "to_input_list")
        else list(run_result)
    )
    messages: list[dict] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        kind = item.get("type")
        if kind == "function_call":
            messages.append(
                {
                    "role": "assistant",
                    "tool_calls": [
                        {
                            "id": item.get("call_id"),
                            "function": {
                                "name": item.get("name", ""),
                                "arguments": item.get("arguments", "{}"),
                            },
                        }
                    ],
                }
            )
        elif kind == "function_call_output":
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": item.get("call_id"),
                    "content": str(item.get("output", "")),
                }
            )
        elif "role" in item:
            content = item.get("content")
            if isinstance(content, list):  # Responses-format content parts
                content = "".join(
                    p.get("text", "") for p in content if isinstance(p, dict)
                )
            messages.append({"role": item["role"], "content": content})

    return Trace.from_messages(
        messages, final_output=getattr(run_result, "final_output", None),
        raw=run_result,
    )
```

Source code in `parsbench/integrations/langgraph.py`

```
def to_trace(state) -> Trace:
    msgs = (state.get("messages") or []) if isinstance(state, dict) else state
    out: list[dict] = []
    for m in msgs:
        if isinstance(m, dict):
            out.append(m)
            continue
        content = getattr(m, "content", "")
        entry: dict = {
            "role": _ROLES.get(getattr(m, "type", ""), "assistant"),
            "content": content if isinstance(content, str) else str(content),
        }
        tool_calls = getattr(m, "tool_calls", None) or []
        if tool_calls:
            entry["tool_calls"] = [
                {
                    "id": tc.get("id"),
                    "function": {"name": tc.get("name", ""), "arguments": tc.get("args", {})},
                }
                for tc in tool_calls
            ]
        if getattr(m, "tool_call_id", None):
            entry["tool_call_id"] = m.tool_call_id
        out.append(entry)
    return Trace.from_messages(out, raw=state)
```

Source code in `parsbench/integrations/pydantic_ai.py`

```
def to_trace(result) -> Trace:
    msgs = result.all_messages() if hasattr(result, "all_messages") else result
    out: list[dict] = []
    for m in msgs:
        for part in getattr(m, "parts", []):
            kind = getattr(part, "part_kind", "")
            if kind == "user-prompt":
                out.append({"role": "user", "content": str(part.content)})
            elif kind == "system-prompt":
                out.append({"role": "system", "content": str(part.content)})
            elif kind == "text":
                out.append({"role": "assistant", "content": part.content})
            elif kind == "tool-call":
                args = getattr(part, "args", {})
                if not isinstance(args, (dict, str)):
                    args = json.loads(getattr(part, "args_as_json_str", lambda: "{}")())
                out.append(
                    {
                        "role": "assistant",
                        "tool_calls": [
                            {
                                "id": getattr(part, "tool_call_id", None),
                                "function": {"name": part.tool_name, "arguments": args},
                            }
                        ],
                    }
                )
            elif kind == "tool-return":
                out.append(
                    {
                        "role": "tool",
                        "tool_call_id": getattr(part, "tool_call_id", ""),
                        "content": str(part.content),
                    }
                )
    return Trace.from_messages(out, final_output=getattr(result, "output", None),
                               raw=result)
```

Source code in `parsbench/integrations/agno.py`

```
def to_trace(run) -> Trace:
    msgs = getattr(run, "messages", None) or (run if isinstance(run, list) else [])
    out = []
    for m in msgs:
        if isinstance(m, dict):
            out.append(m)
            continue
        content = getattr(m, "content", None)
        entry = {
            "role": getattr(m, "role", "assistant"),
            "content": content if content is None or isinstance(content, str) else str(content),
        }
        if getattr(m, "tool_calls", None):
            entry["tool_calls"] = m.tool_calls  # already OpenAI-format dicts
        if getattr(m, "tool_call_id", None):
            entry["tool_call_id"] = m.tool_call_id
        out.append(entry)
    return Trace.from_messages(out, final_output=getattr(run, "content", None),
                               raw=run)
```

Duck-typed OTel SpanProcessor that records finished spans.

Source code in `parsbench/integrations/otel.py`

```
class TraceCollector:
    """Duck-typed OTel SpanProcessor that records finished spans."""

    def __init__(self):
        self.spans = []

    def on_start(self, span, parent_context=None):
        pass

    def on_end(self, span):
        self.spans.append(span)

    def shutdown(self):
        pass

    def force_flush(self, timeout_millis: int = 30000):
        return True

    def to_trace(self) -> Trace:
        return spans_to_trace(self.spans)
```

Source code in `parsbench/integrations/otel.py`

```
def spans_to_trace(spans) -> Trace:
    messages = []
    final = ""
    for span in spans:
        attrs = _attrs(span)
        op = attrs.get("gen_ai.operation.name")
        kind = (attrs.get("openinference.span.kind") or "").upper()

        if op == "execute_tool" or kind == "TOOL":
            name = attrs.get("gen_ai.tool.name") or attrs.get("tool.name") or _name(span)
            args = _json(
                attrs.get("gen_ai.tool.call.arguments")
                or attrs.get("input.value")
                or {}
            )
            if not isinstance(args, dict):
                args = {"_raw": args}
            result = attrs.get("gen_ai.tool.call.result") or attrs.get("output.value")
            call_id = attrs.get("gen_ai.tool.call.id") or _name(span)
            messages.append(
                {
                    "role": "assistant",
                    "tool_calls": [
                        {"id": call_id, "function": {"name": name, "arguments": args}}
                    ],
                }
            )
            if result is not None:
                messages.append(
                    {"role": "tool", "tool_call_id": call_id, "content": str(result)}
                )
        elif op in ("chat", "invoke_agent") or kind in ("LLM", "AGENT", "CHAIN"):
            text = _last_text(
                attrs.get("gen_ai.output.messages") or attrs.get("output.value")
            )
            if text:
                final = text

    if final:
        messages.append({"role": "assistant", "content": final})
    return Trace.from_messages(messages, raw=list(spans))
```

Source code in `parsbench/integrations/adk.py`

```
def persian_response_match(reference: str, response: str) -> float:
    ref = normalize(reference).split()
    got = normalize(response).split()
    if not ref or not got:
        return float(ref == got)
    common: dict[str, int] = {}
    for token in ref:
        common[token] = common.get(token, 0) + 1
    overlap = 0
    for token in got:
        if common.get(token, 0) > 0:
            common[token] -= 1
            overlap += 1
    if overlap == 0:
        return 0.0
    precision = overlap / len(got)
    recall = overlap / len(ref)
    return 2 * precision * recall / (precision + recall)
```

Source code in `parsbench/integrations/langfuse.py`

```
def push(evaluation_result, *, name: str = "parsbench", host: str | None = None,
         public_key: str | None = None, secret_key: str | None = None,
         _post=None):
    host = (host or os.environ["LANGFUSE_HOST"]).rstrip("/")
    public_key = public_key or os.environ["LANGFUSE_PUBLIC_KEY"]
    secret_key = secret_key or os.environ["LANGFUSE_SECRET_KEY"]

    now = datetime.now(timezone.utc).isoformat()
    trace_id = str(uuid.uuid4())
    batch = [
        {
            "id": str(uuid.uuid4()),
            "type": "trace-create",
            "timestamp": now,
            "body": {"id": trace_id, "name": name},
        }
    ]
    for golden_result in evaluation_result.golden_results:
        for check_result in golden_result.check_results:
            if check_result.skipped:
                continue
            batch.append(
                {
                    "id": str(uuid.uuid4()),
                    "type": "score-create",
                    "timestamp": now,
                    "body": {
                        "id": str(uuid.uuid4()),
                        "traceId": trace_id,
                        "name": check_result.check,
                        "value": check_result.score,
                        "comment": f"{golden_result.golden_name}"
                        + (f" — {check_result.reason}" if check_result.reason else ""),
                    },
                }
            )

    if _post is None:  # pragma: no cover - exercised via injection in tests
        import requests

        _post = requests.post
    response = _post(
        f"{host}/api/public/ingestion",
        json={"batch": batch},
        auth=(public_key, secret_key),
        timeout=30,
    )
    response.raise_for_status()
    return trace_id
```

# Models

Bases: `ABC`

An abstract base class representing a model.

Attributes:

| Name                  | Type       | Description                                          |
| --------------------- | ---------- | ---------------------------------------------------- |
| `model_name`          | `property` | A property representing the name of the model.       |
| `support_concurrency` | `bool`     | A flag indicating if the model supports concurrency. |

Methods:

| Name                    | Description                                                                                                                                                                 |
| ----------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `model_name`            | Abstract method to return the name of the model.                                                                                                                            |
| `get_prompt_completion` | str) -> str: Abstract method to generate completion for a given prompt.                                                                                                     |
| `prompt_formatter`      | str) -> Union\[str, List[Dict]\]: Abstract method to format a prompt.                                                                                                       |
| `completion_formatter`  | str) -> str: Method to format the model completion.                                                                                                                         |
| `generate_completions`  | TaskMatchGroup, prefer_concurrency: bool = True, n_workers: int = 4) -> TaskMatchGroup: Method to generate completions for a list of matches, optionally using concurrency. |

Note

This class should be subclassed to implement the abstract methods.

Source code in `parsbench/models/base.py`

```
class Model(ABC):
    """
    An abstract base class representing a model.

    Attributes:
        model_name (property): A property representing the name of the model.
        support_concurrency (bool): A flag indicating if the model supports concurrency.

    Methods:
        model_name(self) -> str: Abstract method to return the name of the model.
        get_prompt_completion (self, prompt: str) -> str: Abstract method to generate completion for a given prompt.
        prompt_formatter (self, prompt: str) -> Union[str, List[Dict]]: Abstract method to format a prompt.
        completion_formatter (self, completion: str) -> str: Method to format the model completion.
        generate_completions (self, matches: TaskMatchGroup, prefer_concurrency: bool = True, n_workers: int = 4) -> TaskMatchGroup: Method to generate completions for a list of matches, optionally using concurrency.

    Note:
        This class should be subclassed to implement the abstract methods.
    """

    support_concurrency: bool = False

    @property
    @abstractmethod
    def model_name(self) -> str:
        pass

    @abstractmethod
    def get_prompt_completion(self, prompt: str) -> str:
        pass

    @abstractmethod
    def prompt_formatter(self, prompt: str) -> str | list[dict]:
        pass

    def completion_formatter(self, completion: str) -> str:
        return completion

    def generate_completions(
        self,
        matches: "TaskMatchGroup",
        prefer_concurrency: bool = True,
        skip_existing: bool = False,
        n_workers: int = 4,
    ) -> "TaskMatchGroup":
        if prefer_concurrency and self.support_concurrency:
            matches = self._gen_with_concurrency(
                matches, n_workers=n_workers, skip_existing=skip_existing
            )
        else:
            for match in tqdm(
                matches, total=len(matches), desc="Generating completions"
            ):
                if match.completion is not None and skip_existing:
                    continue
                match.completion = self.completion_formatter(
                    self.get_prompt_completion(match.prompt)
                )
        return matches

    def _gen_with_concurrency(
        self,
        matches: "TaskMatchGroup",
        n_workers: int = 4,
        skip_existing: bool = False,
    ) -> "TaskMatchGroup":
        def _gen_single_match_completion(match: "TaskMatch") -> "TaskMatch":
            match.completion = self.completion_formatter(
                self.get_prompt_completion(match.prompt)
            )
            return match

        with ThreadPoolExecutor(max_workers=n_workers) as executor:
            futures = []

            for match in matches:
                if match.completion is not None and skip_existing:
                    continue
                future = executor.submit(
                    _gen_single_match_completion,
                    match,
                )
                futures.append(future)

            for future in tqdm(
                as_completed(futures), total=len(futures), desc="Generating completions"
            ):
                future.result()

        matches.matches.sort(key=lambda m: m.id)
        return matches
```

Bases: `Model`

A model interface for OpenAI-like APIs.

Attributes:

| Name                    | Type     | Description                                            |
| ----------------------- | -------- | ------------------------------------------------------ |
| `api_base_url`          | `str`    | The base URL for the OpenAI API.                       |
| `api_secret_key`        | `str`    | The secret key for accessing the OpenAI API.           |
| `model`                 | `str`    | The specific model being used for processing.          |
| `instruction_prompt`    | `str`    | The default instruction prompt for the model.          |
| `model_parameters`      | `dict`   | Additional parameters specific to the model.           |
| `completion_parameters` | `dict`   | Parameters for completion generation.                  |
| `retry_on_ratelimit`    |          | bool = False,                                          |
| `cooldown_interval`     |          | int = 10,                                              |
| `max_retries`           |          | int = 1,                                               |
| `client`                | `OpenAI` | An instance of the OpenAI client for API interactions. |

Methods:

| Name                    | Description                                                                     |
| ----------------------- | ------------------------------------------------------------------------------- |
| `model_name`            | Returns the name of the model.                                                  |
| `prompt_formatter`      | Formats a given prompt into a list of messages. Could be overloaded.            |
| `completion_formatter`  | Method to format the model completion. Could be overloaded.                     |
| `get_prompt_completion` | Generates completion for a given prompt using the OpenAI API.                   |
| `generate_completions`  | Generates completions for a list of TaskMatch objects using ThreadPoolExecutor. |

Source code in `parsbench/models/openai_interface.py`

```
class OpenAIModel(Model):
    """
    A model interface for OpenAI-like APIs.

    Attributes:
        api_base_url (str): The base URL for the OpenAI API.
        api_secret_key (str): The secret key for accessing the OpenAI API.
        model (str): The specific model being used for processing.
        instruction_prompt (str): The default instruction prompt for the model.
        model_parameters (dict): Additional parameters specific to the model.
        completion_parameters (dict): Parameters for completion generation.
        retry_on_ratelimit: bool = False,
        cooldown_interval: int = 10,
        max_retries: int = 1,
        client (OpenAI): An instance of the OpenAI client for API interactions.

    Methods:
        model_name: Returns the name of the model.
        prompt_formatter: Formats a given prompt into a list of messages. Could be overloaded.
        completion_formatter: Method to format the model completion. Could be overloaded.
        get_prompt_completion: Generates completion for a given prompt using the OpenAI API.
        generate_completions: Generates completions for a list of TaskMatch objects using ThreadPoolExecutor.
    """

    support_concurrency: bool = True

    def __init__(
        self,
        api_base_url: str,
        api_secret_key: str,
        model: str,
        instruction_prompt: str = DEFAULT_INSTRUCTION_PROMPT,
        model_parameters: dict = None,
        completion_parameters: dict = None,
        retry_on_ratelimit: bool = False,
        cooldown_interval: int = 10,
        max_retries: int = 1,
        **kwargs
    ):
        self.api_base_url = api_base_url
        self.api_secret_key = api_secret_key
        self.model = model
        self.instruction_prompt = instruction_prompt
        self.model_parameters = model_parameters or dict()
        self.completion_parameters = completion_parameters or dict(temperature=0.7)
        self.retry_on_ratelimit = retry_on_ratelimit
        self.cooldown_interval = cooldown_interval
        self.max_retries = max_retries

        self.client = OpenAI(
            base_url=self.api_base_url,
            api_key=self.api_secret_key,
            **self.model_parameters,
        )

    @property
    def model_name(self) -> str:
        return self.model

    def prompt_formatter(self, prompt: str) -> list[dict]:
        messages = [
            {"role": "system", "content": self.instruction_prompt},
            {"role": "user", "content": prompt},
        ]
        return messages

    def get_prompt_completion(self, prompt: str) -> str:
        messages = self.prompt_formatter(prompt)

        retries = 0
        while retries < self.max_retries:
            try:
                completion = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    **self.completion_parameters,
                    stream=False,  # Always override this parameter.
                )
                return completion.choices[0].message.content
            except RateLimitError as exc:
                if self.retry_on_ratelimit:
                    retries += 1
                    time.sleep(self.cooldown_interval)
                else:
                    raise exc

        raise Exception("Max retries exceeded.")
```

Bases: `Model`

A model interface for Anthropic-like APIs.

Attributes:

| Name                    | Type        | Description                                               |
| ----------------------- | ----------- | --------------------------------------------------------- |
| `api_base_url`          | `str`       | The base URL for the Anthropic API.                       |
| `api_secret_key`        | `str`       | The secret key for accessing the Anthropic API.           |
| `model`                 | `str`       | The name of the model.                                    |
| `instruction_prompt`    | `str`       | The default instruction prompt for the model.             |
| `model_parameters`      | `dict`      | Additional parameters specific to the model.              |
| `completion_parameters` | `dict`      | Parameters for generating completions.                    |
| `retry_on_ratelimit`    |             | bool = False,                                             |
| `cooldown_interval`     |             | int = 10,                                                 |
| `max_retries`           |             | int = 1,                                                  |
| `client`                | `Anthropic` | An instance of the Anthropic client for API interactions. |

Methods:

| Name                    | Description                                                                                                                          |
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
| `model_name`            | Returns the name of the model.                                                                                                       |
| `prompt_formatter`      | str) -> list\[dict\]: Formats the prompt into a list of messages.                                                                    |
| `get_prompt_completion` | str) -> str: Generates completion for a given prompt.                                                                                |
| `generate_completions`  | TaskMatchGroup, prefer_concurrency: bool = True, n_workers: int = 4) -> TaskMatchGroup: Generates completions for a list of matches. |

Source code in `parsbench/models/anthropic_interface.py`

```
class AnthropicModel(Model):
    """
    A model interface for Anthropic-like APIs.

    Attributes:
        api_base_url (str): The base URL for the Anthropic API.
        api_secret_key (str): The secret key for accessing the Anthropic API.
        model (str): The name of the model.
        instruction_prompt (str): The default instruction prompt for the model.
        model_parameters (dict): Additional parameters specific to the model.
        completion_parameters (dict): Parameters for generating completions.
        retry_on_ratelimit: bool = False,
        cooldown_interval: int = 10,
        max_retries: int = 1,
        client (Anthropic): An instance of the Anthropic client for API interactions.

    Methods:
        model_name(self) -> str: Returns the name of the model.
        prompt_formatter(self, prompt: str) -> list[dict]: Formats the prompt into a list of messages.
        get_prompt_completion(self, prompt: str) -> str: Generates completion for a given prompt.
        generate_completions(self, matches: TaskMatchGroup, prefer_concurrency: bool = True, n_workers: int = 4) -> TaskMatchGroup: Generates completions for a list of matches.
    """

    support_concurrency: bool = True

    def __init__(
        self,
        api_secret_key: str,
        model: str,
        api_base_url: str | None = None,
        instruction_prompt: str = DEFAULT_INSTRUCTION_PROMPT,
        model_parameters: dict = None,
        completion_parameters: dict = None,
        retry_on_ratelimit: bool = False,
        cooldown_interval: int = 10,
        max_retries: int = 1,
        **kwargs
    ):
        self.api_base_url = api_base_url
        self.api_secret_key = api_secret_key
        self.model = model
        self.instruction_prompt = instruction_prompt
        self.model_parameters = model_parameters or dict()
        self.completion_parameters = completion_parameters or dict(
            max_tokens=1024, temperature=0.7
        )
        self.retry_on_ratelimit = retry_on_ratelimit
        self.cooldown_interval = cooldown_interval
        self.max_retries = max_retries

        self.client = Anthropic(
            base_url=self.api_base_url,
            api_key=self.api_secret_key,
            **self.model_parameters,
        )

    @property
    def model_name(self) -> str:
        return self.model

    def prompt_formatter(self, prompt: str) -> list[dict]:
        messages = [
            {"role": "user", "content": prompt},
        ]
        return messages

    def get_prompt_completion(self, prompt: str) -> str:
        messages = self.prompt_formatter(prompt)

        retries = 0
        while retries < self.max_retries:
            try:
                message = self.client.messages.create(
                    model=self.model,
                    messages=messages,
                    system=self.instruction_prompt,
                    **self.completion_parameters,
                    stream=False,  # Always override this parameter.
                )
                return message.content[0].text
            except RateLimitError as exc:
                if self.retry_on_ratelimit:
                    retries += 1
                    time.sleep(self.cooldown_interval)
                else:
                    raise exc

        raise Exception("Max retries exceeded.")
```

Bases: `Model`

A model interface for pre-trained transformer models.

Attributes:

| Name                      | Type                     | Description                                       |
| ------------------------- | ------------------------ | ------------------------------------------------- |
| `model`                   | `PreTrainedModel`        | The pre-trained transformer model.                |
| `tokenizer`               | `PreTrainedTokenizer`    | The tokenizer associated with the model.          |
| `generation_config`       | `GenerationConfig`       | The generation configuration for text generation. |
| `instruction_prompt`      | `str`                    | The default instruction prompt for the model.     |
| `custom_prompt_formatter` | \`Callable\[[str], str\] | None\`                                            |

Methods:

| Name                    | Description                                                                           |
| ----------------------- | ------------------------------------------------------------------------------------- |
| `model_name`            | Returns the base model prefix of the transformer model.                               |
| `prompt_formatter`      | Formats a prompt by combining system instruction and user input. Could be overloaded. |
| `completion_formatter`  | Method to format the model completion. Could be overloaded.                           |
| `get_prompt_completion` | Generates a completion for a given prompt using the model and tokenizer.              |

Source code in `parsbench/models/transformers_interface.py`

```
class PreTrainedTransformerModel(Model):
    """
    A model interface for pre-trained transformer models.

    Attributes:
        model (PreTrainedModel): The pre-trained transformer model.
        tokenizer (PreTrainedTokenizer): The tokenizer associated with the model.
        generation_config (GenerationConfig): The generation configuration for text generation.
        instruction_prompt (str): The default instruction prompt for the model.
        custom_prompt_formatter (Callable[[str], str] | None): A custom prompt formatter function.

    Methods:
        model_name: Returns the base model prefix of the transformer model.
        prompt_formatter: Formats a prompt by combining system instruction and user input. Could be overloaded.
        completion_formatter: Method to format the model completion. Could be overloaded.
        get_prompt_completion: Generates a completion for a given prompt using the model and tokenizer.
    """

    support_concurrency: bool = False  # TODO: should support later.

    def __init__(
        self,
        model: PreTrainedModel,
        tokenizer: PreTrainedTokenizer,
        generation_config: GenerationConfig = DEFAULT_GENERATION_CONFIG,
        instruction_prompt: str = DEFAULT_INSTRUCTION_PROMPT,
        custom_prompt_formatter: Callable[[str], str] | None = None,
    ):
        self.model = model
        self.tokenizer = tokenizer
        self.generation_config = generation_config
        self.instruction_prompt = instruction_prompt
        self.custom_prompt_formatter = custom_prompt_formatter

    @property
    def model_name(self) -> str:
        return self.model.config.name_or_path or "model"

    def prompt_formatter(self, prompt: str) -> str:
        messages = [
            {"role": "system", "content": self.instruction_prompt},
            {"role": "user", "content": prompt},
        ]
        text = self.tokenizer.apply_chat_template(
            messages, tokenize=False, add_generation_prompt=True
        )
        return text

    def get_prompt_completion(self, prompt: str) -> str:
        if self.custom_prompt_formatter:
            input_text = self.custom_prompt_formatter(prompt)
        else:
            input_text = self.prompt_formatter(prompt)

        model_inputs = self.tokenizer([input_text], return_tensors="pt").to(
            self.model.device
        )

        generated_ids = self.model.generate(
            model_inputs.input_ids,
            generation_config=self.generation_config,
            attention_mask=model_inputs.attention_mask,
        )
        generated_ids = [
            output_ids[len(input_ids) :]
            for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
        ]

        response = self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[
            0
        ]
        return response
```

# Scores

Class representing a Scorer object.

Attributes:

| Name   | Type                          | Description                         |
| ------ | ----------------------------- | ----------------------------------- |
| `func` | `Callable[[str, str], float]` | The scoring function to be wrapped. |

Methods:

| Name      | Description                                                                                                                    |
| --------- | ------------------------------------------------------------------------------------------------------------------------------ |
| `measure` | str, target: str) -> float: Calculates the score between the completion and target strings using the wrapped scoring function. |
| `name`    | Returns the name of the wrapped scoring function with underscores replaced by spaces and title-cased.                          |

Source code in `parsbench/scores/base.py`

```
class Scorer:
    """
    Class representing a Scorer object.

    Attributes:
        func (Callable[[str, str], float]): The scoring function to be wrapped.

    Methods:
        measure(completion: str, target: str) -> float:
            Calculates the score between the completion and target strings using the wrapped scoring function.

        name() -> str:
            Returns the name of the wrapped scoring function with underscores replaced by spaces and title-cased.
    """

    def __init__(self, func: Callable[[str, str], float]):
        self.func = func

    def measure(self, completion: str, target: str) -> float:
        return self.func(completion, target)

    @property
    def name(self) -> str:
        return self.func.__name__.replace("_", " ").title()
```

Wraps a scorer function inside the Scorer class.

Source code in `parsbench/scores/base.py`

```
def wrap_scorer(func):
    """Wraps a scorer function inside the Scorer class."""
    return Scorer(func)
```

Source code in `parsbench/scores/common.py`

```
@wrap_scorer
def exact_match(completion: str, target: str) -> int:
    return int(completion == target)
```

Source code in `parsbench/scores/machine_translation.py`

```
@wrap_scorer
def english_sentence_bleu(completion: str, target: str) -> float:
    _ensure_punkt()

    reference_translation = [nltk.word_tokenize(target)]
    model_translation = nltk.word_tokenize(completion)
    bleu_score = nltk.translate.bleu(
        reference_translation, model_translation, weights=(1,)
    )
    return bleu_score
```

Source code in `parsbench/scores/machine_translation.py`

```
@wrap_scorer
def persian_sentence_bleu(completion: str, target: str) -> float:
    reference_translation = [hazm.word_tokenize(target)]
    model_translation = hazm.word_tokenize(completion)
    bleu_score = nltk.translate.bleu(
        reference_translation, model_translation, weights=(1,)
    )
    return bleu_score
```

Source code in `parsbench/scores/summarization.py`

```
@wrap_scorer
def english_rouge(completion: str, target: str) -> float:
    scores = _english_scorer().score(target, completion)
    return scores["rouge1"].fmeasure
```

Source code in `parsbench/scores/summarization.py`

```
@wrap_scorer
def persian_rouge(completion: str, target: str) -> float:
    scores = _persian_scorer().score(target, completion)
    return scores["rouge1"].fmeasure
```

# Tasks

Bases: `TaskMatchGenerator`, `TaskScorer`

Task class represents a task that combines functionality from TaskMatchGenerator and TaskScorer.

Attributes:

| Name            | Type           | Description               |
| --------------- | -------------- | ------------------------- |
| `task_name`     | `str`          | The name of the task.     |
| `task_category` | `TaskCategory` | The category of the task. |

Methods:

| Name       | Description                                                                              |
| ---------- | ---------------------------------------------------------------------------------------- |
| `evaluate` | Method to evaluate the task by generating matches, scoring them, and saving the results. |

Source code in `parsbench/tasks/base/task.py`

```
class Task(TaskMatchGenerator, TaskScorer, metaclass=ABCMeta):
    """
    Task class represents a task that combines functionality from TaskMatchGenerator and TaskScorer.

    Attributes:
        task_name (str): The name of the task.
        task_category (TaskCategory): The category of the task.

    Methods:
        evaluate: Method to evaluate the task by generating matches, scoring them, and saving the results.
    """

    task_name: str
    task_category: TaskCategory

    def evaluate(
        self,
        model: Model,
        prompt_lang: str = "fa",
        prompt_shots: list[int] = None,
        n_first: int = 200,
        sub_tasks: list[str] | None = None,
        save_matches: bool = False,
        save_evaluation: bool = False,
        output_path: str = None,
        skip_existing_matches: bool = False,
        prefer_concurrency: bool = True,
        n_workers: int = 4,
    ) -> list[EvaluationResult]:
        """
        Method to evaluate the task by generating matches, scoring them, and saving the results.

        Parameters:
            model (Model): The model to be evaluated.
            prompt_lang (str, optional): The language of the prompt (default is "fa").
            prompt_shots (list[int], optional): The list of prompt shots to evaluate (default is None).
            n_first (int, optional): The number of initial prompts to consider (default is 200).
            sub_tasks (list[str], optional): The list of sub-tasks to evaluate (default is None).
            save_matches (bool, optional): Flag to save the generated matches (default is False).
            save_evaluation (bool, optional): Flag to save the evaluation results (default is False).
            output_path (str, optional): The output path to save the matches and evaluation results.
            skip_existing_matches (bool, optional): Flag to skip already generated matches in the output path (default is False).
            prefer_concurrency (bool, optional): The flag to use concurrent processing if the model and task support that (default is True).
            n_workers (int, optional): The number of workers for concurrent processing (default is 4).

        Returns:
            list[EvaluationResult]: A list of EvaluationResult objects representing the evaluation results.

        Raises:
            Exception: If output_path is not provided when saving matches or evaluation.
            Exception: If output_path is not provided when skipping existing matches.
            Exception: If sub tasks are not defined or if invalid sub tasks are provided.

        """
        if (save_matches or save_evaluation) and not output_path:
            raise Exception(
                "You should set the output path to save matches/evaluation."
            )

        if skip_existing_matches and not output_path:
            raise Exception(
                "Cannot find already generated matches when output_path is not set."
            )

        task_path = None
        if output_path:
            task_path = get_task_path(output_path, model.model_name, self.task_name)

        prompt_shots = [0] if prompt_shots is None else prompt_shots

        if sub_tasks:
            if not self.sub_tasks:
                raise Exception("Sub tasks are not defined.")

            invalid_sub_tasks = set(sub_tasks) - set(self.sub_tasks)
            if invalid_sub_tasks:
                raise Exception(f"Sub tasks {invalid_sub_tasks} are not defined.")

        sub_tasks = sub_tasks or self._selected_sub_tasks or self.sub_tasks

        evaluation_results: list[EvaluationResult] = []

        for sub_task in sub_tasks or [None]:
            match_groups: list[TaskMatchGroup] = []

            for shots in prompt_shots:
                if skip_existing_matches and check_task_matches_exists(
                    task_path, shots, sub_task=sub_task
                ):
                    match_group = TaskMatchGroup.from_file(
                        task_path, shots, sub_task=sub_task
                    )
                    match_group._loaded_locally = True
                else:
                    match_group = self.generate_matches(
                        prompt_lang,
                        n_shots=shots,
                        n_first=n_first,
                        sub_task=sub_task,
                    )
                match_groups.append(match_group)

            has_error = False
            for match_group in match_groups:
                eval_desc = f"{match_group.n_shots}-shot"
                if sub_task:
                    eval_desc = f"sub task '{sub_task}' with " + eval_desc
                desc = f"Evaluating {eval_desc} prompt:"
                print(desc)

                is_loaded_locally = getattr(match_group, "_loaded_locally", False)

                if is_loaded_locally:
                    total_skipped = sum(m.completion is not None for m in match_group)
                    print(
                        f"{total_skipped} of {len(match_group)} match completions will be loaded from local."
                    )

                try:
                    model.generate_completions(
                        match_group,
                        prefer_concurrency=prefer_concurrency,
                        skip_existing=is_loaded_locally,
                        n_workers=n_workers,
                    )
                    self.score_matches(match_group)
                except Exception:
                    has_error = True
                finally:
                    if save_matches:
                        match_group.save(task_path, sub_task=sub_task)

            if not has_error:
                evaluation_result = EvaluationResult(
                    model_name=model.model_name,
                    task_name=self.task_name,
                    task_category=self.task_category,
                    score_name=self.score_name,
                    sub_task=sub_task,
                    prompt_shot_results=[
                        PromptShotEvaluationResult(
                            n_shots=m.n_shots,
                            score=self.get_overall_score(m),
                        )
                        for m in match_groups
                    ],
                )
                evaluation_results.append(evaluation_result)

                if save_evaluation:
                    evaluation_result.save(task_path)

        return evaluation_results

    def __enter__(self) -> "Task":
        self.load_data()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self._data = None
```

## `evaluate(model, prompt_lang='fa', prompt_shots=None, n_first=200, sub_tasks=None, save_matches=False, save_evaluation=False, output_path=None, skip_existing_matches=False, prefer_concurrency=True, n_workers=4)`

Method to evaluate the task by generating matches, scoring them, and saving the results.

Parameters:

| Name                    | Type        | Description                                                                                 | Default    |
| ----------------------- | ----------- | ------------------------------------------------------------------------------------------- | ---------- |
| `model`                 | `Model`     | The model to be evaluated.                                                                  | *required* |
| `prompt_lang`           | `str`       | The language of the prompt (default is "fa").                                               | `'fa'`     |
| `prompt_shots`          | `list[int]` | The list of prompt shots to evaluate (default is None).                                     | `None`     |
| `n_first`               | `int`       | The number of initial prompts to consider (default is 200).                                 | `200`      |
| `sub_tasks`             | `list[str]` | The list of sub-tasks to evaluate (default is None).                                        | `None`     |
| `save_matches`          | `bool`      | Flag to save the generated matches (default is False).                                      | `False`    |
| `save_evaluation`       | `bool`      | Flag to save the evaluation results (default is False).                                     | `False`    |
| `output_path`           | `str`       | The output path to save the matches and evaluation results.                                 | `None`     |
| `skip_existing_matches` | `bool`      | Flag to skip already generated matches in the output path (default is False).               | `False`    |
| `prefer_concurrency`    | `bool`      | The flag to use concurrent processing if the model and task support that (default is True). | `True`     |
| `n_workers`             | `int`       | The number of workers for concurrent processing (default is 4).                             | `4`        |

Returns:

| Type                     | Description                                                                                       |
| ------------------------ | ------------------------------------------------------------------------------------------------- |
| `list[EvaluationResult]` | list\[EvaluationResult\]: A list of EvaluationResult objects representing the evaluation results. |

Raises:

| Type        | Description                                                        |
| ----------- | ------------------------------------------------------------------ |
| `Exception` | If output_path is not provided when saving matches or evaluation.  |
| `Exception` | If output_path is not provided when skipping existing matches.     |
| `Exception` | If sub tasks are not defined or if invalid sub tasks are provided. |

Source code in `parsbench/tasks/base/task.py`

```
def evaluate(
    self,
    model: Model,
    prompt_lang: str = "fa",
    prompt_shots: list[int] = None,
    n_first: int = 200,
    sub_tasks: list[str] | None = None,
    save_matches: bool = False,
    save_evaluation: bool = False,
    output_path: str = None,
    skip_existing_matches: bool = False,
    prefer_concurrency: bool = True,
    n_workers: int = 4,
) -> list[EvaluationResult]:
    """
    Method to evaluate the task by generating matches, scoring them, and saving the results.

    Parameters:
        model (Model): The model to be evaluated.
        prompt_lang (str, optional): The language of the prompt (default is "fa").
        prompt_shots (list[int], optional): The list of prompt shots to evaluate (default is None).
        n_first (int, optional): The number of initial prompts to consider (default is 200).
        sub_tasks (list[str], optional): The list of sub-tasks to evaluate (default is None).
        save_matches (bool, optional): Flag to save the generated matches (default is False).
        save_evaluation (bool, optional): Flag to save the evaluation results (default is False).
        output_path (str, optional): The output path to save the matches and evaluation results.
        skip_existing_matches (bool, optional): Flag to skip already generated matches in the output path (default is False).
        prefer_concurrency (bool, optional): The flag to use concurrent processing if the model and task support that (default is True).
        n_workers (int, optional): The number of workers for concurrent processing (default is 4).

    Returns:
        list[EvaluationResult]: A list of EvaluationResult objects representing the evaluation results.

    Raises:
        Exception: If output_path is not provided when saving matches or evaluation.
        Exception: If output_path is not provided when skipping existing matches.
        Exception: If sub tasks are not defined or if invalid sub tasks are provided.

    """
    if (save_matches or save_evaluation) and not output_path:
        raise Exception(
            "You should set the output path to save matches/evaluation."
        )

    if skip_existing_matches and not output_path:
        raise Exception(
            "Cannot find already generated matches when output_path is not set."
        )

    task_path = None
    if output_path:
        task_path = get_task_path(output_path, model.model_name, self.task_name)

    prompt_shots = [0] if prompt_shots is None else prompt_shots

    if sub_tasks:
        if not self.sub_tasks:
            raise Exception("Sub tasks are not defined.")

        invalid_sub_tasks = set(sub_tasks) - set(self.sub_tasks)
        if invalid_sub_tasks:
            raise Exception(f"Sub tasks {invalid_sub_tasks} are not defined.")

    sub_tasks = sub_tasks or self._selected_sub_tasks or self.sub_tasks

    evaluation_results: list[EvaluationResult] = []

    for sub_task in sub_tasks or [None]:
        match_groups: list[TaskMatchGroup] = []

        for shots in prompt_shots:
            if skip_existing_matches and check_task_matches_exists(
                task_path, shots, sub_task=sub_task
            ):
                match_group = TaskMatchGroup.from_file(
                    task_path, shots, sub_task=sub_task
                )
                match_group._loaded_locally = True
            else:
                match_group = self.generate_matches(
                    prompt_lang,
                    n_shots=shots,
                    n_first=n_first,
                    sub_task=sub_task,
                )
            match_groups.append(match_group)

        has_error = False
        for match_group in match_groups:
            eval_desc = f"{match_group.n_shots}-shot"
            if sub_task:
                eval_desc = f"sub task '{sub_task}' with " + eval_desc
            desc = f"Evaluating {eval_desc} prompt:"
            print(desc)

            is_loaded_locally = getattr(match_group, "_loaded_locally", False)

            if is_loaded_locally:
                total_skipped = sum(m.completion is not None for m in match_group)
                print(
                    f"{total_skipped} of {len(match_group)} match completions will be loaded from local."
                )

            try:
                model.generate_completions(
                    match_group,
                    prefer_concurrency=prefer_concurrency,
                    skip_existing=is_loaded_locally,
                    n_workers=n_workers,
                )
                self.score_matches(match_group)
            except Exception:
                has_error = True
            finally:
                if save_matches:
                    match_group.save(task_path, sub_task=sub_task)

        if not has_error:
            evaluation_result = EvaluationResult(
                model_name=model.model_name,
                task_name=self.task_name,
                task_category=self.task_category,
                score_name=self.score_name,
                sub_task=sub_task,
                prompt_shot_results=[
                    PromptShotEvaluationResult(
                        n_shots=m.n_shots,
                        score=self.get_overall_score(m),
                    )
                    for m in match_groups
                ],
            )
            evaluation_results.append(evaluation_result)

            if save_evaluation:
                evaluation_result.save(task_path)

    return evaluation_results
```

Bases: `ABC`

An abstract base class for defining data loaders.

Attributes:

| Name        | Type  | Description                  |
| ----------- | ----- | ---------------------------- |
| `data_path` | `str` | The path to the data source. |

Methods:

| Name   | Description                                                       |
| ------ | ----------------------------------------------------------------- |
| `load` | Abstract method to be implemented by subclasses for loading data. |

Source code in `parsbench/tasks/base/data_loader.py`

```
class DataLoader(ABC):
    """
    An abstract base class for defining data loaders.

    Attributes:
        data_path (str): The path to the data source.

    Methods:
        load(self) -> list[dict]: Abstract method to be implemented by subclasses for loading data.
    """

    def __init__(self, data_path: str, **kwargs) -> None:
        self.data_path = data_path

    @abstractmethod
    def load(self) -> list[dict]:
        pass
```

Bases: `DataLoader`

A data loader class for loading JSON line data from either a local file or a URL.

Attributes:

| Name        | Type  | Description                            |
| ----------- | ----- | -------------------------------------- |
| `data_path` | `str` | The path to the JSON line data source. |

Methods:

| Name   | Description                                         |
| ------ | --------------------------------------------------- |
| `load` | Loads the JSON line data from the specified source. |

Source code in `parsbench/tasks/base/data_loader.py`

```
class JSONLineDataLoader(DataLoader):
    """
    A data loader class for loading JSON line data from either a local file or a URL.

    Attributes:
        data_path (str): The path to the JSON line data source.

    Methods:
        load(self) -> list[dict]: Loads the JSON line data from the specified source.
    """

    def load(self) -> list[dict]:
        content = _fetch_text_file(self.data_path)

        reader = jsonlines.Reader(content.split("\n"))
        return list(reader.iter(type=dict, skip_invalid=True, skip_empty=True))
```

Bases: `DataLoader`

A data loader class for loading datasets using the Hugging Face library.

Attributes:

| Name        | Type  | Description                  |
| ----------- | ----- | ---------------------------- |
| `data_path` | `str` | The path to the data source. |
| `split`     | \`str | None\`                       |

Methods:

| Name          | Description                                                                                                |
| ------------- | ---------------------------------------------------------------------------------------------------------- |
| `load`        | Loads the dataset from the specified data path and split.                                                  |
| `with_filter` | Callable[..., bool]) -> "HuggingFaceDataLoader": Adds a filter function to apply when loading the dataset. |

Source code in `parsbench/tasks/base/data_loader.py`

```
class HuggingFaceDataLoader(DataLoader):
    """
    A data loader class for loading datasets using the Hugging Face library.

    Attributes:
        data_path (str): The path to the data source.
        split (str | None): The split of the dataset to load.

    Methods:
        load(self) -> list[dict]: Loads the dataset from the specified data path and split.
        with_filter(self, func: Callable[..., bool]) -> "HuggingFaceDataLoader": Adds a filter function to apply when loading the dataset.
    """

    def __init__(
        self,
        data_path: str,
        split: str | None = None,
        **optional_parameters: dict[str, Any],
    ) -> None:
        super().__init__(data_path)
        self.split = split
        self.optional_parameters = optional_parameters
        self._filters = []

    def load(self) -> list[dict]:
        dataset = datasets.load_dataset(
            self.data_path, split=self.split, **self.optional_parameters
        )
        if len(self._filters):
            for filter_ in self._filters:
                dataset = dataset.filter(filter_)
        return dataset.to_list()

    def with_filter(self, func: Callable[..., bool]) -> "HuggingFaceDataLoader":
        self._filters.append(func)
        return self
```

Bases: `DataLoader`

A data loader class for loading CSV line data from either a local file or a URL.

Attributes:

| Name        | Type  | Description                           |
| ----------- | ----- | ------------------------------------- |
| `data_path` | `str` | The path to the CSV line data source. |

Methods:

| Name   | Description                                        |
| ------ | -------------------------------------------------- |
| `load` | Loads the CSV line data from the specified source. |

Source code in `parsbench/tasks/base/data_loader.py`

```
class CSVDataLoader(DataLoader):
    """
    A data loader class for loading CSV line data from either a local file or a URL.

    Attributes:
        data_path (str): The path to the CSV line data source.

    Methods:
        load(self) -> list[dict]: Loads the CSV line data from the specified source.
    """

    def __init__(self, data_path: str, csv_arguments: dict | None = None, **kwargs):
        super().__init__(data_path)
        self.csv_arguments = csv_arguments or {}

    def load(self) -> list[dict]:
        content = _fetch_text_file(self.data_path)

        csv_reader = csv.DictReader(content.split("\n"), **self.csv_arguments)
        return list(csv_reader)
```

A class representing a prompt template.

Attributes:

| Name                       | Type                          | Description                                                            |
| -------------------------- | ----------------------------- | ---------------------------------------------------------------------- |
| `language_templates`       | `dict[str, str]`              | A dictionary mapping language codes to prompt templates.               |
| `prompt_variables_mapping` | `dict[str, str]`              | A dictionary mapping prompt variable names to corresponding data keys. |
| `target_variables_mapping` | `dict[str, str]`              | A dictionary mapping target variable names to corresponding data keys. |
| `prompt_shot_templates`    | \`dict[str, str]              | None\`                                                                 |
| `prompt_shot_examples`     | \`dict\[str, dict[int, str]\] | None\`                                                                 |

Source code in `parsbench/tasks/base/prompt_template.py`

```
class PromptTemplate:
    """
    A class representing a prompt template.

    Attributes:
        language_templates (dict[str, str]): A dictionary mapping language codes to prompt templates.
        prompt_variables_mapping (dict[str, str]): A dictionary mapping prompt variable names to corresponding data keys.
        target_variables_mapping (dict[str, str]): A dictionary mapping target variable names to corresponding data keys.
        prompt_shot_templates (dict[str, str] | None): A dictionary mapping prompt shot templates to language codes, or None if not provided.
        prompt_shot_examples (dict[str, dict[int, str]] | None): A dictionary mapping prompt shot examples to language codes and shot numbers, or None if not provided.
    """

    def __init__(
        self,
        language_templates: dict[str, str],
        prompt_variables_mapping: dict[str, str],
        target_variables_mapping: dict[str, str],
        prompt_shot_templates: dict[str, str] | None = None,
        prompt_shot_examples: dict[str, dict[int, str]] | None = None,
    ):
        self.language_templates = language_templates
        self.prompt_variables_mapping = prompt_variables_mapping
        self.target_variables_mapping = target_variables_mapping

        if prompt_shot_templates is not None and prompt_shot_examples is not None:
            raise ValueError("Cannot provide both prompt shot templates and examples")

        if prompt_shot_templates is None and prompt_shot_examples is None:
            raise ValueError("Must provide either prompt shot templates or examples")

        self.prompt_shot_templates = prompt_shot_templates
        self.prompt_shot_examples = prompt_shot_examples

    def get_prompt(
        self,
        prompt_lang: str,
        data: dict,
        n_shots: int = 0,
        sample_data: list[dict] | None = None,
    ):
        prompt_template = self.language_templates.get(prompt_lang, None)
        if not prompt_template:
            raise RuntimeError(
                f"There is no prompt template for language {prompt_lang}."
            )

        if n_shots > 0:
            if sample_data:
                example_text = self._gen_example_text(prompt_lang, n_shots, sample_data)
            else:
                example_text = self._get_static_example_text(prompt_lang, n_shots)
        else:
            example_text = ""

        prompt = prompt_template.format(
            example_shots=example_text, **self.get_prompt_variables(data)
        )
        prompt = prompt.replace("\n\n\n", "\n")

        return prompt

    def get_prompt_variables(self, data: dict) -> dict:
        mapped_data = {}
        for pk, dk in self.prompt_variables_mapping.items():
            if isinstance(dk, ConstantPromptVariable):
                mapped_data[pk] = dk.value
            else:
                if dk not in data:
                    raise ValueError(f"Key {dk} not in data.")
                mapped_data[pk] = data[dk]
        return mapped_data

    def get_target_variables(self, data: dict) -> dict:
        mapped_data = {}
        for tk, dk in self.target_variables_mapping.items():
            if dk not in data:
                raise ValueError(f"Key {dk} not in data.")
            mapped_data[tk] = data[dk]
        return mapped_data

    def _get_static_example_text(self, prompt_lang: str, n_shots: int) -> str:
        shot_examples = self.prompt_shot_examples.get(prompt_lang, None)
        if not shot_examples:
            raise RuntimeError(f"There is no shot example for language {prompt_lang}.")

        example_text = shot_examples.get(n_shots, "")
        if not example_text:
            raise RuntimeError(
                f"There is no {n_shots}-shot example for langauge {prompt_lang}. "
                f"You can only use {', '.join(map(str, shot_examples.keys()))} shot examples."
            )

        return example_text

    def _gen_example_text(
        self, prompt_lang: str, n_shots: int, sample_data: list[dict]
    ) -> str:
        if len(sample_data) != n_shots:
            raise RuntimeError(
                f"The number of samples ({len(sample_data)}) is not equal to the number of shots ({n_shots})."
            )

        if shot_template := self.prompt_shot_templates.get(prompt_lang):
            example_text = "\n".join(
                shot_template.format(
                    **self.get_prompt_variables(sample),
                    **self.get_target_variables(sample),
                )
                for sample in sample_data
            )
        else:
            sample_variables = [
                {
                    **self.get_prompt_variables(sample),
                    **self.get_target_variables(sample),
                }
                for sample in sample_data
            ]
            example_text = "\n".join(
                "\n".join(f"{k.capitalize()}:\n{v}" for k, v in variables)
                for variables in sample_variables
            )

        return example_text

    @property
    def has_shot_templates(self) -> bool:
        return bool(self.prompt_shot_templates)

    @property
    def has_shot_examples(self) -> bool:
        return bool(self.prompt_shot_examples)
```

Bases: `Mapping`

A class representing lazy loading of templates.

Inherits from Mapping.

Attributes:

| Name             | Type             | Description                                       |
| ---------------- | ---------------- | ------------------------------------------------- |
| `template_paths` | `dict[str, str]` | A dictionary mapping template keys to file paths. |

Source code in `parsbench/tasks/base/prompt_template.py`

```
class LazyLoadTemplates(Mapping):
    """
    A class representing lazy loading of templates.

    Inherits from Mapping.

    Attributes:
        template_paths (dict[str, str]): A dictionary mapping template keys to file paths.
    """

    def __init__(self, template_paths: dict[str, str] | None = None, **kwargs):
        super().__init__()
        self.template_paths = template_paths or kwargs or {}
        self._contents: dict[str, str | None] = {
            key: None for key in self.template_paths
        }

    def _load_content(self, key):
        if key in self.template_paths:
            with open(self.template_paths[key], "r") as file:
                self._contents[key] = file.read()
        else:
            raise KeyError(f"Key '{key}' not found in template_paths")

    def __getitem__(self, key) -> str:
        if key not in self._contents:
            raise KeyError(f"Key '{key}' not found")
        if self._contents[key] is None:
            self._load_content(key)
        return self._contents[key]

    def __getattr__(self, key) -> str:
        try:
            return self.__getitem__(key)
        except KeyError:
            raise AttributeError(f"Attribute '{key}' not found")

    def __iter__(self):
        return iter(self.template_paths)

    def __len__(self):
        return len(self.template_paths)
```

A data class representing the evaluation result for a prompt shot, including the number of shots and the corresponding score.

Attributes:

| Name      | Type    | Description                                        |
| --------- | ------- | -------------------------------------------------- |
| `n_shots` | `int`   | The number of shots for the evaluation.            |
| `score`   | `float` | The score obtained for the prompt shot evaluation. |

Source code in `parsbench/tasks/base/evaluation_result.py`

```
@dataclass
class PromptShotEvaluationResult:
    """
    A data class representing the evaluation result for a prompt shot, including the number of shots and the corresponding score.

    Attributes:
        n_shots (int): The number of shots for the evaluation.
        score (float): The score obtained for the prompt shot evaluation.
    """

    n_shots: int
    score: float

    @classmethod
    def from_dict(cls, data: dict) -> "PromptShotEvaluationResult":
        return cls(**data)

    def to_dict(self) -> dict:
        return asdict(self)

    def to_pandas(self) -> pd.DataFrame:
        return pd.DataFrame([self])

    def __str__(self) -> str:
        return f"{self.n_shots}-shot score: {self.score:.4f}"
```

A data class representing the evaluation result for a model on a specific task, including the model name, task name, task category, score name, prompt shot results, and optional sub-task.

Attributes:

| Name                  | Type                               | Description                                                                                        |
| --------------------- | ---------------------------------- | -------------------------------------------------------------------------------------------------- |
| `model_name`          | `str`                              | The name of the model being evaluated.                                                             |
| `task_name`           | `str`                              | The name of the task for which the model is being evaluated.                                       |
| `task_category`       | `TaskCategory`                     | The category of the task (e.g., CLASSIC, REASONING, MATH, KNOWLEDGE).                              |
| `score_name`          | `str`                              | The name of the score obtained for the evaluation.                                                 |
| `prompt_shot_results` | `list[PromptShotEvaluationResult]` | A list of PromptShotEvaluationResult objects representing the evaluation results for prompt shots. |
| `sub_task`            | `str`                              | The name of the sub-task being evaluated, if applicable.                                           |

Source code in `parsbench/tasks/base/evaluation_result.py`

```
@dataclass
class EvaluationResult:
    """
    A data class representing the evaluation result for a model on a specific task, including the model name, task name, task category, score name, prompt shot results, and optional sub-task.

    Attributes:
        model_name (str): The name of the model being evaluated.
        task_name (str): The name of the task for which the model is being evaluated.
        task_category (TaskCategory): The category of the task (e.g., CLASSIC, REASONING, MATH, KNOWLEDGE).
        score_name (str): The name of the score obtained for the evaluation.
        prompt_shot_results (list[PromptShotEvaluationResult]): A list of PromptShotEvaluationResult objects representing the evaluation results for prompt shots.
        sub_task (str, optional): The name of the sub-task being evaluated, if applicable.
    """

    model_name: str
    task_name: str
    task_category: TaskCategory
    score_name: str
    prompt_shot_results: list[PromptShotEvaluationResult]
    sub_task: str | None = None

    @classmethod
    def from_file(cls, path: str) -> "EvaluationResult":
        with jsonlines.open(path, "r") as reader:
            data = reader.read(type=dict)
            return cls.from_dict(data)

    @classmethod
    def from_dict(cls, data: dict) -> "EvaluationResult":
        prompt_shot_results = [
            PromptShotEvaluationResult.from_dict(psr)
            for psr in data.pop("prompt_shot_results")
        ]
        data["task_category"] = TaskCategory[data["task_category"].upper()]
        return cls(**data, prompt_shot_results=prompt_shot_results)

    def to_dict(self) -> dict:
        return {
            **asdict(self),
            "prompt_shot_results": [e.to_dict() for e in self.prompt_shot_results],
        }

    def to_pandas(self) -> pd.DataFrame:
        data = [
            {
                "model_name": self.model_name,
                "task_name": self.task_name,
                "task_category": self.task_category.value,
                "sub_task": self.sub_task,
                "n_shots": psr.n_shots,
                "score_name": self.score_name,
                "score": psr.score,
            }
            for psr in self.prompt_shot_results
        ]
        return pd.DataFrame(data)

    def save(self, path: str):
        file_name = (
            f"evaluation_{self.sub_task}.jsonl" if self.sub_task else "evaluation.jsonl"
        )
        task_path = path / file_name

        with jsonlines.open(task_path, "w") as writer:
            writer.write(self.to_dict())

    def __str__(self) -> str:
        text = f"Model: {self.model_name}\nTask: {self.task_name}"

        if self.sub_task:
            text += f" ({self.sub_task})"

        text += "\nScore:\n"

        for psr in self.prompt_shot_results:
            text += f" - {psr.n_shots}-shot prompt: {psr.score:.4f}\n"
        return text.strip("\n")

    @property
    def average_score(self) -> float:
        return sum([psr.score for psr in self.prompt_shot_results]) / len(
            self.prompt_shot_results
        )

    @property
    def max_score(self) -> float:
        return max([psr.score for psr in self.prompt_shot_results])
```

Source code in `parsbench/tasks/base/task_match.py`

```
@dataclass
class TaskMatch:
    id: int
    prompt: str
    target: str
    completion: str | None = None
    formatted_completion: str | None = None
    score: int | None = None

    @classmethod
    def from_dict(cls, data: dict) -> "TaskMatch":
        return cls(**data)

    def format_completion(self, formatter: Callable[[str], str]):
        self.formatted_completion = formatter(self.completion)

    def format_prompt(self, formatter: Callable[[str], str]):
        self.prompt = formatter(self.prompt)

    def format_target(self, formatter: Callable[[str], str]):
        self.target = formatter(self.target)

    def to_dict(self) -> dict:
        return asdict(self)

    def to_pandas(self) -> pd.DataFrame:
        return pd.DataFrame([self])

    @property
    def cleaned_completion(self) -> str | None:
        if self.formatted_completion is not None:
            return self.formatted_completion
        return self.completion
```

Source code in `parsbench/tasks/base/task_match.py`

```
@dataclass
class TaskMatchGroup:
    n_shots: int
    matches: list[TaskMatch]

    def __iter__(self):
        yield from iter(self.matches)

    def __len__(self) -> int:
        return len(self.matches)

    @classmethod
    def from_file(
        cls, path: str, n_shots: int, sub_task: str | None
    ) -> "TaskMatchGroup":
        if sub_task:
            matches_path = path / f"matches_{sub_task}_{n_shots}_shot.jsonl"
        else:
            matches_path = path / f"matches_{n_shots}_shot.jsonl"

        with jsonlines.open(matches_path, "r") as reader:
            matches: list[TaskMatch] = []
            for row in reader.iter(type=dict, skip_invalid=True):
                matches.append(TaskMatch.from_dict(row))

        return cls(n_shots=n_shots, matches=matches)

    @classmethod
    def from_dict(cls, data: dict) -> "TaskMatchGroup":
        matches = [TaskMatch.from_dict(m) for m in data.pop("matches")]
        return cls(**data, matches=matches)

    def format_completions(self, formatter: Callable[[str], str]):
        for m in self.matches:
            m.format_completion(formatter)

    def format_prompts(self, formatter: Callable[[str], str]):
        for m in self.matches:
            m.format_prompt(formatter)

    def format_targets(self, formatter: Callable[[str], str]):
        for m in self.matches:
            m.format_target(formatter)

    def to_dict(self) -> dict:
        return {
            **asdict(self),
            "matches": [match.to_dict() for match in self.matches],
        }

    def to_pandas(self) -> pd.DataFrame:
        df = pd.DataFrame(
            [
                {
                    **asdict(match),
                    "n_shots": self.n_shots,
                }
                for match in self.matches
            ]
        )
        return df

    def save(self, path: str, sub_task: str | None):
        if sub_task:
            matches_path = path / f"matches_{sub_task}_{self.n_shots}_shot.jsonl"
        else:
            matches_path = path / f"matches_{self.n_shots}_shot.jsonl"

        with jsonlines.open(matches_path, "w") as writer:
            writer.write_all(self.to_dict()["matches"])

    @property
    def prompts(self) -> list[str]:
        return [m.prompt for m in self.matches]

    @property
    def targets(self) -> list[str]:
        return [m.target for m in self.matches]

    @property
    def completions(self) -> list[str | None]:
        return [m.cleaned_completion for m in self.matches]

    @property
    def scores(self) -> list[int | None]:
        return [m.score for m in self.matches]
```

Load all tasks from the 'parsbench.tasks' package and return a list of Task objects.

Returns:

| Type         | Description                                                                                         |
| ------------ | --------------------------------------------------------------------------------------------------- |
| `list[Task]` | list\[Task\]: A list of Task objects representing all tasks found in the 'parsbench.tasks' package. |

Source code in `parsbench/tasks/utils.py`

```
def load_all_tasks() -> list[Task]:
    """
    Load all tasks from the 'parsbench.tasks' package and return a list of Task objects.

    Returns:
        list[Task]: A list of Task objects representing all tasks found in the 'parsbench.tasks' package.

    """
    tasks: list[Task] = []

    module = importlib.import_module("parsbench.tasks")
    for attr_name in dir(module):
        attr = getattr(module, attr_name)
        if isinstance(attr, type) and issubclass(attr, Task) and attr is not Task:
            tasks.append(attr)

    return tasks
```
# Optional

# Contribution

Thank you for considering contributing to ParsBench! Contributions of code, tasks, datasets, and benchmark results all help. Here are the guidelines.

## Ways to contribute

### 1. Fix bugs

If you find a bug in ParsBench:

- Report it in the [issue tracker](https://github.com/ParsBench/ParsBench/issues) with steps to reproduce, the expected behavior, and the actual behavior.
- Or submit a pull request with a fix. Please include relevant tests and documentation updates where applicable.

### 2. Add features

If you have an idea for a new feature:

- Propose it in the [issue tracker](https://github.com/ParsBench/ParsBench/issues) first, describing its purpose and impact, so we can discuss the design before you invest time in it.
- Then submit a PR with the implementation, tests, and documentation with usage examples.

### 3. Add new tasks with new datasets

New Persian evaluation tasks expand what ParsBench can measure:

- Propose the task in the [issue tracker](https://github.com/ParsBench/ParsBench/issues), including its objective, its relevance to Persian language processing, and the dataset's source, format, and any preprocessing.
- Submit a PR that adds the task, integrates the dataset, includes tests, and documents how to use it. The [advanced tutorial](https://parsbench.github.io/ParsBench/tutorial/advanced/index.md) shows the anatomy of a task.

### 4. Run benchmarks on new models

Benchmark results on state-of-the-art or fine-tuned open-weight models are contributions too:

- Run the benchmarks with ParsBench and share the results in the [issue tracker](https://github.com/ParsBench/ParsBench/issues), including details about the model and any fine-tuning, the saved matches, and your observations.

## Development setup

ParsBench uses [Poetry](https://python-poetry.org/) and requires Python 3.12+:

```
git clone https://github.com/<your-username>/ParsBench.git
cd ParsBench
poetry install

poetry run pytest              # run the test suite
poetry run mkdocs serve        # preview the documentation locally
```

## Contribution process

1. Fork the [ParsBench repository](https://github.com/ParsBench/ParsBench) to your GitHub account and clone it.
1. Create a branch for your change:

```
git checkout -b feature/your-feature-name
```

1. Make your changes, commit with a meaningful message, and push:

```
git commit -m "Description of your changes"
git push origin feature/your-feature-name
```

1. Open a pull request against the main repository with a description of what changed and why.

## Code style and testing

- Follow the existing code style and conventions (the project uses `black` and `isort`).
- Include tests for your changes and make sure the suite passes before submitting.
- Update the documentation when behavior changes.

## Getting help

If you have questions or get stuck, open an issue in the [issue tracker](https://github.com/ParsBench/ParsBench/issues).

# Changelog

## 0.3.0 - 2026-08-17

### Added

- App evaluation (`parsbench.appeval`), in the same class-based style as tasks and benchmarks: `AppEvaluator(goldens).evaluate(app)` scores your own Persian AI app with `Golden` expectations — tool calls, contains/not_contains, structured format, budgets, custom checks, and Persian-prompted judge checks (correctness, faithfulness, refusal) — and `score_traces()` evaluates pre-captured traces instead, with `pass^k` for flaky agents. Results are plain dataclasses (`AppEvaluationResult`) with the familiar `to_dict`/`to_pandas`/`save` conveniences.
- Persian-aware matching layer: Arabic/Persian codepoint and digit-script unification, ZWNJ/spacing tolerance, rial/toman + هزار/میلیون amounts (including compound «۲ میلیون و ۵۰۰ هزار» and the ۲٫۵ / ۲/۵ decimal forms), and Jalali/Gregorian date equivalence — applied symmetrically to `contains`, `not_contains`, and tool-argument comparison.
- Multi-turn simulation (`SimulationEvaluator`) with an Iranian-user simulator (taarof, Finglish, toman/rial confusion, Jalali dates, …), test generation from your docs (`GoldenGenerator`), and judge calibration against human labels (`JudgeCalibrator`).
- Framework integrations: OpenAI Agents SDK, LangGraph/LangChain, Pydantic AI, Agno, Google ADK, and any OTel-instrumented app via `TraceCollector`; Langfuse score export via `result.to_langfuse()`.
- `parsbench view`: a local, ParsBench-themed viewer over recorded runs — live progress, failure-first run pages, trace detail, RTL simulation replay, run-vs-run diffs, dark/light themes, JSON/CSV/Markdown export, and optional score charts. Zero new dependencies, no build step.
- Evaluations now record into a project-local `.parsbench/` store by default (full traces included; opt out with `record=False` or `PARSBENCH_NO_RECORD=1`).
- `parsbench test` CLI (pytest wrapper; install with `parsbench[test]`) and runnable examples for every supported framework under `examples/`, plus industry scenarios (banking, e-commerce, healthcare, telecom simulation, knowledge-base RAG) under `examples/industry/` — the offline ones run in CI.
- Production hardening: a crash inside the evaluated app is reported as a failing `app_error` check instead of aborting the run, async apps work inside running event loops (notebooks, servers), evaluation and calibration accept `prefer_concurrency=`/`n_workers=`, and the built-in judge client retries transient failures with a bounded per-call timeout (`PARSBENCH_MAX_RETRIES` / `PARSBENCH_TIMEOUT`). The package ships `py.typed`, the app-eval surface is mypy-clean, and CI runs the suite on Python 3.12/3.13.

### Changed

- Simulation: hitting the turn cap no longer fails a conversation the goal judge scored as successful — a chatty user simulator that never emits the stop token was punishing the app for the simulator's behavior.

## 0.2.0 - 2026-07-15

### Changed

- Support current library versions: transformers 5.x, datasets 5.x, openai 2.x, anthropic, and numpy 2.
- **Breaking:** drop Python 3.10/3.11 support; Python >= 3.12 is now required (needed by hazm >= 0.11 and numpy 2).
- Bump hazm to 0.12, drop the unused `scipy` pin, and declare the `numpy`/`pandas`/`requests`/`tqdm`/`nltk` dependencies that were previously only installed transitively.

### Fixed

- Fix evaluation and merge correctness bugs, and cache the summarization/NER scorers.
- Use cleaned completions when scoring matches.
- Replace the undeclared `pytz` dependency with the standard library (`pandas` 3 no longer ships it).
- Remove format targets in the Persian Math task.
- Import the optional `math_equivalence` package lazily so `import parsbench` no longer fails when it is not installed.

### Added

- Add `show_bar_plot` to `BenchmarkResult`.
- Add a mechanism to skip evaluation results on error.

## 0.1.7 - 2024-08-15

### Fixed

- Fix typos in prompt templates.
- Fix error on using formatted targets while scoring matches.
- Improve `from_matches_files` function speed in BenchmarkResult.
- Fix returning list in AnthropicModel completion function.

### Added

- Add `formatted_completion` field to task matches.
- Update completion formatters in tasks.
- Add re-score option to `from_matches_files` in BenchmarkResult.
- Add max retries exceeded error in API-based models.
- Add snapshot functionality to save matches on error.
- Add leaderboard builder function.
- Add build from file functions to the BenchmarkResult class.

## 0.1.6 - 2024-07-25

### Fixed

- Fix misspell in ParsiNLUMultipleChoice task name.
- Fix wrong target key in the XLSummary.
- Add org prefix to the sentiment analysis task.
- Fix FarsTailEntailment prompt target key.

### Added

- Add `attention_mask` to the transformer model `generate` function.

## 0.1.5 - 2024-07-18

### Added

- Add FarsTail entailment task.
- Add Persian News Summary task.
- Add XL-Sum task.
- Add ParsiNLUBenchmark. It is a sub class of CustomBenchmark with hard-coded ParsiNLU tasks.

## 0.1.4 - 2024-07-12

### Fixed

- Fix `load_all_tasks` returning empty list.

### Added

- Add Anthropic model interface.
- Add retry on rate limit to API-based models.
- Add `skip_existing_matches` to the task evaluate function. It skips matches that are already generated and scored.

## 0.1.3 - 2024-07-06

### Fixed

- Fix `model_name` property in PreTrainedTransformerModel.

### Added

- Add Persian MMLU (Khayyam Challenge) task.
- Add `select_sub_tasks` to the task class.

## 0.1.2 - 2024-07-06

### Fixed

- Fix misspells and typos.
- Use`name_or_path` parameter as the `model_name` in PreTrainedTransformerModel.

### Changed

- Update sentiment analysis task prompt template.

### Added

- Add `completion_formatter` to the model interfaces.

## 0.1.1 - 2024-07-05

### Added

- Add support for Python >= 3.10
- Add `prefer_concurrency` to the benchmark, task and models.

## 0.1.0 - 2024-07-03

ParsBench got alive!

# Donation

ParsBench is a self-funded project I build on my weekends, carried this far with the volunteer help of companies and individuals. If you want to support it, here are some ways.

## Support me

You can donate any amount you want. I will spend it on paying for APIs and GPUs :)

- [Hamibash](https://hamibash.com/parsbench)

## Buy me a coffee

Coffee is fuel for my internal engine. It makes the coding go faster.

- [Coffeete](https://www.coffeete.ir/shahriarshm)

## Crypto

The most valuable donation for me is crypto.

- BTC (Bitcoin): ‍`bc1q5sfazp0a9ls8rdg0ql6u70fr5pss0krvmctp5f`
- ETH (Ethereum): `0x58a64E699a75D4c2CAE31411036819A91C397693`
- USDT (Tron): `TDZnM5qKyGFBNmuDuAicAkTtHrVT78i515`

## Tell me you've donated

If you'd like, tell me you've donated and I will add your name to the list of donors on the ParsBench pages. Reach me at [shahriarshm81@gmail.com](mailto:shahriarshm81@gmail.com).