Evals

Evals measure how well an agent does its job, so you can tell whether a prompt, model or tool change made it better or worse before users notice. loomflow.eval covers both sides. Offline: run a dataset of labelled cases through the agent, score each run’s full event trace, and gate CI on the aggregates. Online: sample a fraction of live traffic and score it with metrics that need no ground truth.

Import from loomflow.eval explicitly; nothing is re-exported from the top-level package:

from loomflow.eval import (
    Case, Dataset, EvalHarness, ToolSelectionAccuracy, LLMJudge,
)
 
harness = EvalHarness(agent, metrics=[
    ToolSelectionAccuracy(),
    LLMJudge(model=judge_model),      # a DIFFERENT model than the agent's
])
report = await harness.run(dataset)
report.assert_thresholds({"tool_selection_accuracy": 0.9})

Everything runs offline against in-process fakes (ScriptedModel / EchoModel), so no network is required to eval, or to test the eval.

Cases and datasets

A Case is one labelled example; a Dataset is an ordered list with JSONL round-trip persistence, so eval suites live as flat files next to the code they gate.

from loomflow.eval import Case, Dataset
 
dataset = Dataset([
    Case(input="What is 2+2?", expected="4"),
    Case(input="Refund order 812", expected_tools=["lookup_order", "issue_refund"]),
    Case(input="Summarise our SLA"),          # unlabelled: judge-only
])
dataset.to_jsonl("cases.jsonl")
dataset = Dataset.from_jsonl("cases.jsonl")   # one Case object per line

expected (reference output) and expected_tools (tool names the agent should call) are both optional. Metrics that need ground truth skip cases that don’t carry it, via each metric’s applies(case) gate. id is auto-generated when not supplied. A dataset can carry a name (Dataset(cases, name="billing-faq"); from_jsonl uses the file name), which Studio groups experiments by.

Running: EvalHarness

harness = EvalHarness(agent, metrics=[...], concurrency=4)
report = await harness.run(dataset)
report = await harness.run(dataset, max_cost_usd=2.0)   # stop starting cases after $2

Each case runs with a fresh session_id and user_id="__eval__" (EVAL_USER_ID), so eval traffic stays in its own memory partition, invisible to real users’ recall. The full event trace is captured per case, cases run concurrently under a capacity limiter, and a crashing case is recorded as an error rather than aborting the suite. A metric that raises on a case is recorded too, in that case’s CaseResult.metric_errors (metric name → error), and the other metrics still score it.

max_cost_usd= stops starting new cases once the runs so far have cost that much. Cases already running finish, so it can go over by up to concurrency - 1 runs. The cases that didn’t run are listed in report.skipped, and report.passed is then False.

The EvalReport gives you per-case scores and per-metric aggregates:

report.passed                      # True when no case and no metric raised
report.results[0].metric_errors    # {"llm_judge": "TimeoutError: ..."} when a metric raised
report.mean("exact_match")         # per-metric mean (None if no scores)
report.summary()                   # {metric: {"mean", "min", "count"}}
report.cost_usd                    # what the cases' runs cost; per case: results[0].cost_usd
report.skipped                     # ids of the cases max_cost_usd left out
print(report.to_json())            # full report: cases + aggregates

With Studio recording, each run is also saved as an experiment (harness.run(dataset, experiment="billing v2") names it), with every case’s trace and scores, so you can compare it case by case with an earlier run. loom eval ... --studio does the same from the command line, and Studio can run evals from its page (Running evals from the page).

CI gating: assert_thresholds

report.assert_thresholds({
    "exact_match": 0.9,
    "tool_execution_success": 0.95,
})

Raises AssertionError listing every failure: any metric whose mean is below its threshold, any metric that produced no scores at all (a silent skip must not pass a gate), any thresholded metric that raised on any case, and any errored case. A mean over the cases a metric managed to score says nothing about the rest, so a judge that scored one case in ten no longer passes the gate on that one score. Wire it into pytest or a CI step, or use loom eval for the one-liner CLI form (--threshold exact_match=0.9, exit 1 when unmet).

Built-in metrics

Every metric scores one (case, run) pair in [0.0, 1.0] from the case, the final RunResult, and the captured event trace.

Custom metrics implement the Metric protocol: a name and an async score(case, result, events) -> float; add applies(case) to skip unlabelled cases. tool_calls_from_events / tool_results_from_events extract the trace’s tool channel for you.

LLMJudge

from loomflow.eval import LLMJudge
from loomflow.model import OpenAIModel
 
judge = LLMJudge(
    OpenAIModel("gpt-4.1-mini"),          # separate model instance!
    rubric="Grade factual accuracy only. Ignore style.",
)

The judge must reply with an explicit score: <0–1> line; free-form numbers in prose are deliberately not accepted. The line may be formatted (**Score:** 0.8, Final score: 0.8.), and a fraction or percentage is scaled (score: 4/5 and score: 80% both read as 0.8). A bare score outside 0–1 isn’t clamped; it counts as unparseable. On a parse failure it retries once with corrective feedback; on a second failure it warns and returns the neutral score 0.5 rather than silently reporting 0.0 (which would conflate “judge misbehaved” with “agent failed”).

Online: sampling live traffic

harness.online(...) returns an OnlineScorer your application calls after its own runs; nothing is wrapped or monkeypatched:

scorer = harness.online(sample_rate=0.1, seed=7)
 
async with scorer:                 # task group for background scoring
    events = []
 
    async def capture(event):
        events.append(event)
 
    result = await agent.run(prompt, user_id=uid, emit=capture)
    scorer.submit(prompt, result, events)   # fire-and-forget, never blocks
 
print(scorer.rollup())   # {"seen": ..., "sampled": ..., "days": {...}}

• Sampling uses a seeded random.Random, so it is deterministic for a fixed seed and call sequence.
• Only ground-truth-free metrics score (each metric’s applies gate skips the rest, since live prompts are unlabelled). So ToolExecutionSuccess, MultiStepCoherence, and LLMJudge run; ExactMatch silently doesn’t.
• submit schedules scoring (including judge calls) on the internal task group so the response path never blocks; await scorer.maybe_score(...) is the awaited variant that returns the scores (or None when the run wasn’t sampled).
• rollup() aggregates per UTC day: per-metric mean / min / count.
• With Studio recording, the scores are also attached to the run they scored.

Aggregates are in-process; export them to your telemetry sink on whatever cadence suits your dashboarding.