assay-evals

Record what your AI system does, and how it went, in Assay: runs, agent steps, user feedback and test results. Standard library only, Python 3.9+.

pip install assay-evals

With an Assay server, events go there (see the setup guide). Without one, they're recorded to a local file, so you can start with no account and no server (see "No server" below).

import assay_sdk as assay

assay.init("https://assay.example.com", key="ak_...")   # or set ASSAY_URL / ASSAY_KEY

# An agent
with assay.run("refund_request", input=message, version={"prompt": "support@v5", "model": "claude-sonnet-5"}) as run:
    run.llm(model="claude-sonnet-5", tokens_in=620, tokens_out=180, cost_usd=0.0024)
    order = run.call("get_order", get_order, order_id="O-17")    # runs it; records the result or the error
    run.state("refund:O-17", "create", {"amount": order["price"]})
    run.answer(f"Refunded ${order['price']}.")

# A pipeline
with assay.run("invoice", kind="pipeline", input_ref="s3://inbox/inv-9.pdf") as run:
    with run.stage("extract", prompt="extract_fields@v13") as s:
        run.llm(model="claude-sonnet-5", cost_usd=0.01)          # nested under the stage
        s.outputs.update(fields)

# Outcomes, whenever they're known
assay.feedback(run.id, "thumbs_down")
p = assay.prompt("support", "13", template=text, note="faster refunds")   # "support@13", for run.llm(prompt=p)
assay.feedback(run.id, "edited")      # the quiet ones: they fixed the answer, or "redone": did it themselves
assay.correction(run.id, "total", expected="1240.00", observed="1204.00")
assay.check("nightly-0924", "case-17", "fail", run_id=run.id, field="total", expected="1240.00", actual="1204.00")
assay.check("nightly-0924", "case-17", "pass", run_id=run.id, field="helpful", evaluator="helpful@1",
            inputs={"query": question, "generation": graded})   # what the judge saw, checked against the trace
assay.expect("case-17", calls=[{"tool": "get_order", "args": {"order_id": "O-17"}}], answer="27.61")

These options go to init():

• redact: a function applied to inputs, arguments, results, text and outputs before they leave the process.
• sample=0.1: record one run in ten. A run is recorded whole or not at all, and outcomes are always sent.
• strict=True: raise send errors while developing. Otherwise the SDK never raises into your code.
• enabled=False: the SDK does nothing, e.g. in unit tests.
• path: where to record when there's no server (see below).

Events follow the Assay event schema v1. They stream to POST /v1/ingest in the background, so a run that crashes still shows every step up to the crash.

No server: record locally

Call assay.init() with no URL, and with ASSAY_URL unset. Events are then appended to .assay/events.jsonl, one per line, in the same form the server takes. Set path= or ASSAY_PATH to use another file. Several processes can record to the same file, e.g. pytest -n 4.

When the SDK creates the .assay folder, it adds a .gitignore there, so recorded inputs don't end up in git.

To look at a recording, load it into a local Assay server (assay-server) and open the dashboard:

pip install assay-server
assay load            # reads .assay/events.jsonl into the tenant "local"
assay serve           # http://127.0.0.1:8400, source events:local

Loading the same file twice changes nothing, because every event has an id.

Attach with a few lines: @assay.step, @assay.tool, assay.instrument()

import assay_sdk as assay

assay.init()
assay.instrument()                     # Anthropic and OpenAI calls are recorded, in the step they're in

@assay.step("classification")          # a step of the pipeline; a dict it returns is its outputs
def classify(doc): ...

@assay.tool                            # a tool the agent calls: arguments, result or error, timing
def get_order(order_id): ...

@assay.pipeline("invoice", id_from="document_id")   # one run per call; @assay.agent for an agent
def handle(document_id, pdf):
    return graph.invoke({"pdf": pdf})

The steps, tools and model calls made inside @assay.pipeline, @assay.agent or with assay.run(...) are recorded into that run. A step called outside any run starts one of its own. With no run at all, a tool just runs. assay.instrument() never changes what a call returns, and recording never breaks your code. Everything works on async functions too. assay connect code proposes these lines for your code, as a diff.

Any model provider, one shape: Judge and normalize()

from assay_sdk import Judge, normalize

judge = Judge(provider="openai", model="gpt-5")        # anthropic, gemini, ollama, openai-compatible
r = judge.ask("Rate this answer…", system=RUBRIC, schema=VERDICT, temperature=0)
r.text, r.structured, r.tool_calls, r.usage, r.reasoning, r.finish_reason, r.error

r = normalize(response)   # a response you already have, from any of them (or LiteLLM)

Whatever answered, the fields are the same:

• tool_calls: [{"name", "arguments", "id"}], with arguments always a dict. A JSON string (OpenAI's) is parsed; anything else is kept as {"_raw": value}. Nothing is dropped.
• usage: tokens input, output, cached, reasoning.
• finish_reason: stop, length, tool_call, refusal, content_filter or error.
• structured: the answer parsed as JSON, and checked against schema. When it doesn't fit, it's None with error_kind="invalid", never an empty stand-in.
• error and error_kind (timeout, rate_limited, unavailable, invalid, error), when there's no answer. ask() never raises for a provider's failure.

Every other parameter goes to the provider unchanged: nothing is filtered or renamed. Credentials are the provider SDK's own (its key variables, or a CLI login it supports). Ollama and OpenAI-compatible servers (vLLM, LM Studio, a LiteLLM proxy) need no SDK: base_url, and api_key if the server wants one. evaluate() takes a Judge's answer as it is: evaluate(judge, prompt, schema=VERDICT, judge_kwargs={"schema": VERDICT}). judge_kwargs is how arguments reach your judge when their names are also evaluate()'s own (schema, field, run, ...); evaluate() warns when one looks misrouted.

A judge's few-shot examples come from the golden set's train split, never the items it's measured on: golden_examples(split="train", k=8) returns them with their labels and critiques, and assay calibrate fails a judge that read dev or test (see calibration).

with assay.faults(get_order="error"): breaks a tool on purpose, and assay.faults(get_supplier={"returns": [assay.REAL, {"status": "on_hold"}]}) changes what it returns between calls (a state that moves while the agent works); expect(run).handles_failure(), .no_false_success(), .well_formed_arguments() and .checkpoint(name, ...) check what the agent did about it (see agents).

assay.claim_review(run_id, claim, verdict, evidence=..., correction=..., by=...) records an expert's decision on one claim (supported, wrong, a conflict between sources resolved), made while they use the product. A run tagged consent="shared" is one its user agreed to share. See sensitive data.

with assay.tagged(origin="synthetic"): gives every run started inside its tags, including runs the code opens itself; assay synth run uses it so generated traffic is never counted as production.

assay.instrument() records Anthropic, OpenAI, Gemini, Ollama and LiteLLM calls through the same reading, with the tool calls asked for and why each call stopped.

Your own evaluators: results whose validity is explicit

An LLM judge that answers with something that isn't a verdict, or a metric that divides by zero, shouldn't become a score of 0: that reads as a real failure of your AI. evaluate() calls your evaluator and says whether what came back is a verdict at all:

from assay_sdk import evaluate

VERDICT = {"type": "object", "required": ["score", "reason"],
           "properties": {"score": {"type": "number", "minimum": 0, "maximum": 1}, "reason": {"type": "string"}}}

result = evaluate(my_judge, question, answer, schema=VERDICT, threshold=0.7, retries=2,
                  run=assay_case, field="helpful", evaluator="helpful@2")
result.status             # PASS, FAIL, INVALID, ERROR, TIMEOUT or RATE_LIMITED
result.score              # only for PASS and FAIL: an invalid result has none
result.reason, result.error, result.attempts, result.raw_judge_output

• INVALID: it answered, but not with a verdict: not JSON (a fenced JSON block is fine), not the schema, a score that's None, NaN, infinite or outside score_range (0 to 1 by default), or a score with no threshold to decide by.
• TIMEOUT, RATE_LIMITED, ERROR: it raised. The kind comes from the exception's type, its status_code (429, 5xx) or its message.
• Retries: INVALID, timeouts, rate limits and an unavailable service (connection error, 5xx) are tried again, retries times, with a pause that doubles (backoff seconds first). Any other exception is an ERROR at once: asking a bug again doesn't fix it.
• Recorded: with run=, the result becomes a check. PASS and FAIL are passes and fails; the rest are errors with their kind, what the judge said, and how many tries it took. So they're INVALID, TIMEOUT, RATE_LIMITED, INFRA_ERROR or EVALUATOR_ERROR in Assay, and never count against the AI.

The judge may return a bool, a number, a dict (score, passed or pass, reason), JSON text, or an object with those attributes. aevaluate() is the same for an async judge. A category in the verdict ({"score": 0.2, "category": "grounding"}) names the kind of failure: it's result.category, it's recorded with the check, and an acknowledged failure wakes when it changes. judge_model= and judge_prompt="helpful@3" say which judge it was (a Judge's answer gives its model on its own): a check judged by another model or prompt than its baseline's isn't compared with it as a regression.

Is it the AI or the judge? rejudge=3 asks the judge about the same answer three times and keeps the majority (a tie keeps the first verdict). result.votes is (passed, judged), and it's recorded with the check. When a case varies across attempts and the judge also split on an answer it was asked about again, Assay reports it as judge unstable, not flaky: more attempts of the AI won't settle it, a steadier judge will. See repeated attempts.

result = evaluate(my_judge, question, answer, threshold=0.7, rejudge=3, run=assay_case, field="helpful")
result.votes              # (1, 3): one of three judgements passed the same answer

Sending your own checks, run.check(..., judgements=3, judgements_passed=1) says the same.

Many samples at once: EvalRuntime

Evaluating a dataset is where evaluation tools break down. A 429 cancels the whole run, a retry loop never ends, a library retries out of sight and doubles the bill, one bad sample crashes the job. EvalRuntime runs evaluate() over every sample within limits, and says what the run cost:

import assay_sdk as assay

rt = assay.EvalRuntime(concurrency=10, retries=3, timeout=30, rate_limit=50,
                       retry_on=[429, 500, 502, 503], max_time=600, budget_usd=5,
                       prices={"claude-opus-5": (5, 25)})      # dollars per million tokens
judge = assay.Judge("anthropic", "claude-opus-5")
report = rt.run(judge, prompts, schema=VERDICT, threshold=0.7)   # or: await rt.arun(...)
print(report)

50 samples in 1m12s

46 judged (41 passed, 5 failed)
 2 invalid judge outputs
 1 rate limited
 1 timed out

LLM calls:      57
Retries:        7
Tokens:         68,400 in, 4,560 out
Estimated cost: $0.46

• One retry layer, counted. Through a Judge, the provider SDK's own retries are off inside the runtime, so every request is one LLM call in the report, with its tokens. For a judge of your own, an attempt counts as one call unless its calls are seen (through a Judge, or with assay.instrument() on). An attempt that called the model more than once is reported, because that's where doubled costs hide.
• One sample's failure is its own. An exception, a timeout, even a CancelledError that a client raises on a 429, is that sample's result, and the others go on. Cancelling the run itself does stop it: rt.report has what was judged so far, and run() returns it after Ctrl-C.
• Samples: each is the judge's one argument, and a tuple is spread over its arguments. assay.Sample(question, answer, id="q7", run=case) says exactly what goes where, and records the result as a check on that case's run.
• Results: report.results[i].result is the evaluate() Result, with calls, retries, seconds and cost_usd for that sample. report.to_dict() has the totals. rt.map(fn, items) runs a function of your own per item under the same limits and accounting.

assay test --judge judges with this runtime too (Agents).

With pytest

Take the assay_case fixture. It's a pytest plugin that comes with this package, so there's nothing to configure:

def test_refund(assay_case):
    assay_case.expect(calls=[{"tool": "get_order", "args": {"order_id": "O-17"}}], answer="27.61")
    reply = my_agent("Refund O-17", run=assay_case)   # record steps on it
    assert "27.61" in reply

The fixture wraps the test in assay.run(<test name>, test=<test id>). It also records the test's own outcome as a check on the field pytest, so failing asserts count too. Tests that don't take the fixture are left alone.

With assay-server installed, the test also fails when its run fails Assay's checks: its expect(...), and the contracts and PII rules in assay.toml. So plain pytest goes red on an unsafe tool call:

The run failed Assay's checks:
  Safety: Unsafe action: Broke “delete_order never runs”: ran delete_order (step 2, order_id='O-2').

expect(run) declares everything a run should do, checked together when the test ends: expect(run).must_call("get_order").must_not_call("delete_order").max_steps(8) .must_get_approval_before("refund").max_cost(0.05).max_latency(8).max_tools_exposed(10) .max_context_tokens(8000).must_resolve().

Each model call also records what its input was made of: the system prompt, tool definitions, history and user message in tokens, images and video (count, bytes, size), and settings (temperature, max tokens, reasoning effort). assay.instrument() reads them from the request; run.llm(context=, media=, settings=) takes them otherwise.

What retrieval put into the prompt is recorded with run.retrieve(query, docs, used=4): the fragments (text, dicts, LangChain Documents, LlamaIndex nodes), their tokens, and which went into the prompt. assay.instrument() records LangChain and LlamaIndex retrievers on its own. simulate(agent, Persona(goal=..., traits=..., facts=...), user=Judge(...), run=assay_case) tests a whole conversation against a simulated user (Simulated users).

faithfulness(judge, question, answer, fragments) checks the answer claim by claim against them, with the judge's quotes checked (Faithfulness). Fragments and retrieved tokens per query are compared with the baseline, the whole run's totals are too, and [behavior] max_fragments and max_retrieved_tokens are limits (Retrieved context).

For the test body, assay_sdk.testing has assert_called(run, tool, **args), assert_not_called, assert_called_before(run, first, then), assert_max_steps(run, n), assert_answer_contains and assert_no_pii. Each fails with what the run actually did. assay_sdk.documents scores document extraction field by field: score_document(run, expected, extracted, schema, rules) (Document extraction). assay_sdk.frameworks runs DeepEval and RAGAS metrics as checks: check(run, metric, test_case), and assert_test, a drop-in for DeepEval's (DeepEval and RAGAS). @rewordings("Can I get a refund for O-18?", "refund O-18 pls") runs a test once per wording, and pytest --assay checks each rewording does what the original does (Rewordings).

pytest --assay (with assay-server installed) also compares each test with its last passing run, prints Assay's report in pytest's summary, and exits 1 only when something got worse.

To test with it, assay test (in assay-server) runs your code with the SDK recording, checks each run against its assay.expect(...), and compares with the last run that passed. See Test your AI app locally.