Evals

loomflow.eval is the offline + online evaluation harness. Offline: run a dataset of labelled cases through the agent, score each run’s full event trace, gate CI on the aggregates. Online: sample a fraction of live traffic and score it with ground-truth-free metrics.

It’s a tier-2 submodule — 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) — 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.

Running: EvalHarness

harness = EvalHarness(agent, metrics=[...], concurrency=4)
report = await harness.run(dataset)

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.

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

report.passed                      # True when no case raised
report.mean("exact_match")         # per-metric mean (None if no scores)
report.summary()                   # {metric: {"mean", "min", "count"}}
print(report.to_json())            # full report: cases + aggregates

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), and any errored case. 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. 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
    result = await agent.run(prompt, user_id=uid, emit=capture.append)
    scorer.submit(prompt, result, capture)   # fire-and-forget, never blocks
 
print(scorer.rollup())   # {"seen": ..., "sampled": ..., "days": {...}}

• Sampling uses a seeded random.Random — 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.

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