Skip to content

Evaluation Reference

Everything evaluatorq() accepts, and the patterns built on top of it. If you have not run an evaluation yet, start with Getting Started.

evaluatorq()

async def evaluatorq(
    name: str,
    params: EvaluatorParams | dict[str, Any] | None = None,
    *,
    data: DatasetIdInput | ExperimentInput | Sequence[Awaitable[DataPoint] | DataPointInput] | None = None,
    jobs: list[Job] | None = None,
    evaluators: list[Evaluator] | None = None,
    parallelism: int = 10,
    print_results: bool = True,
    description: str | None = None,
    path: str | None = None,
    inference: bool = True,
) -> EvaluatorqResult
Parameter Type Default Description
data list[DataPoint \| dict] | list[Awaitable[DataPoint]] | DatasetIdInput | ExperimentInput required Data to evaluate — local rows (a DataPoint or a plain dict with the same keys), an Orq dataset, or an existing experiment
jobs list[Job] required Jobs to run on each data point
evaluators list[Evaluator] | None None Evaluators that score job outputs
parallelism int (≥1) 10 Number of concurrent jobs
print_results bool True Display the progress and results table
description str | None None Optional evaluation description
path str | None None Path for organizing results on the Orq dashboard (e.g. "Project/Category")
inference bool True Run the jobs; set False to score data that already has outputs

Parameters can also be passed positionally as an EvaluatorParams model or a plain dict — the three forms below are equivalent:

await evaluatorq("my-eval", data=[...], jobs=[...], parallelism=5)
await evaluatorq("my-eval", {"data": [...], "jobs": [...], "parallelism": 5})
await evaluatorq("my-eval", EvaluatorParams(data=[...], jobs=[...], parallelism=5))

Full type signatures live in the API Reference.

Jobs

The @job() decorator

@job() names a job. The name shows up in the results table, in traces, and — crucially — in error messages:

from evaluatorq import job

@job("risky-job")
async def risky_operation(data: DataPoint, row: int):
    return await potentially_failing_operation(data)

# Error output: "Job 'risky-job' failed: <error details>"
# Without @job:  "<error details>"

It also wraps plain callables, which is handy for one-liners:

uppercase_job = job("uppercase", lambda data, row: data.inputs["text"].upper())
word_count_job = job("word-count", lambda data, row: len(data.inputs["text"].split()))

Multiple jobs per data point

Every job runs against every data point, which is how you compare variants (two prompts, two models, preprocessing on and off) on identical inputs:

await evaluatorq(
    "multi-job-eval",
    data=[...],
    jobs=[preprocessor, analyzer, transformer],
    evaluators=[...],
)

Data sources

data accepts inline DataPoints, an Orq dataset, or awaitables that resolve to DataPoints — the last of which lets you stream rows in from a slow source without blocking the run:

async def get_data_point(i: int) -> DataPoint:
    await asyncio.sleep(0.01)  # e.g. a network fetch
    return DataPoint(inputs={"value": i})

await evaluatorq(
    "async-eval",
    data=[get_data_point(i) for i in range(1000)],
    jobs=[...],
)

For Orq-hosted datasets, pass data=DatasetIdInput(dataset_id="..."). That path requires ORQ_API_KEY — see Configuration.

Built-in evaluators

from evaluatorq import exact_match_evaluator, string_contains_evaluator

string_contains_evaluator()                        # case-insensitive by default
string_contains_evaluator(case_insensitive=False)  # case-sensitive
string_contains_evaluator(name="my-contains-check")  # custom name in the table
exact_match_evaluator()                            # case-sensitive by default

Both compare the job output against the data point's expected_output. For LLM-graded evaluators see LLM as a Jury; for structured, multi-dimensional scores see Structured Results.

Custom evaluators

An evaluator is a {"name": ..., "scorer": ...} pair whose scorer receives the data point and the job output and returns a score:

async def accuracy_scorer(params):
    data, output = params["data"], params["output"]
    score = calculate_score(output, data.expected_output)
    return {"value": score, "explanation": "High accuracy match" if score > 0.8 else "Partial match"}


await evaluatorq(
    "dataset-evaluation",
    data=DatasetIdInput(dataset_id="your-dataset-id"),
    jobs=[processor],
    evaluators=[{"name": "accuracy", "scorer": accuracy_scorer}],
)

Pass/fail and CI

An evaluator that returns pass_ turns the run into a gate:

async def quality_scorer(params):
    score = calculate_quality(params["output"])
    return {
        "value": score,
        "pass_": score >= 0.8,
        "explanation": f"Quality score: {score}",
    }

When any evaluator returns pass_: False, the process exits with code 1 — drop the script into a CI job and it fails the build. The results table gains a pass rate row — Pass Rate | 75% (3/4).

Controlling the run

Parallelism

await evaluatorq("parallel-eval", data=[...], jobs=[...], parallelism=10)

Start low (3–5) when jobs call rate-limited external APIs.

Organizing results on Orq

await evaluatorq(
    "my-evaluation",
    data=[...],
    jobs=[...],
    path="MyProject/Evaluations/Unit Tests",
)

path groups runs in the Orq dashboard — e.g. "Team/Sprint-42/Feature-X".

Documenting a run

await evaluatorq(
    "model-comparison",
    description="Compare GPT-4o vs Claude on customer support responses",
    data=[...],
    jobs=[...],
)

Suppressing terminal output

results = await evaluatorq("silent-eval", data=[...], jobs=[...], print_results=False)

for result in results:
    for job_result in result.job_results or []:
        print(f"{job_result.job_name}: {job_result.output}")

Where to next