Skip to content

aaa.eval

EvalSuite 

EvalSuite(
    dataset: Dataset,
    algo: Algorithm,
    metrics: Sequence[Metric],
    *,
    name: str | None = None,
    concurrency: int = 4,
    repeats: int = 1,
    budget: Budget | None = None,
    tenant: str | None = None,
    extract: Callable[[Any], str] = default_extract,
    make_input: Callable[[Case], dict[str, Any]]
    | None = None,
)

EvalSuite(dataset, algo, [metrics]).run(engine) -> EvalReport.

  • Each case is a real run (engine.run) with meta={"aaa_eval": True}; the trajectory (spans), steps, tokens and cost are read back from its log, so tool-use and budget metrics see what really happened.
  • repeats runs each case several times (agents are non-deterministic); the report keeps every run.
  • A metric that raises is recorded as a failed score, not a crashed suite; a run the harness could not start is error.
  • budget bounds each case, so one runaway case cannot spend the whole eval.

Dataset 

Dataset(cases: Iterable[Case], name: str = 'dataset')

from_list classmethod 

from_list(
    rows: Sequence[dict[str, Any]], name: str = "dataset"
) -> Dataset

Rows are {"input": ..., "expected": ...}; input may be a plain string (becomes {"task": ...}).

from_traces async classmethod 

from_traces(engine: Any, **kw: Any) -> Dataset

Cases from finished production runs (scrubbed by default). See aaa.eval.traces.dataset_from_traces.

Case

Bases: BaseModel

CaseResult dataclass 

CaseResult(
    case: Case,
    outcome: Outcome[Any] | None,
    text: str = "",
    run_id: str | None = None,
    spans: list[Span] = list(),
    steps: int = 0,
    tokens: int = 0,
    cost: float = 0.0,
    latency_s: float = 0.0,
    error: str | None = None,
    repeat: int = 0,
    scores: list[Score] = list(),
    turns: list[tuple[str, str]] = list(),
)

EvalReport dataclass 

EvalReport(name: str, results: list[CaseResult] = list())

pass_rate property 

pass_rate: float

Share of case runs where every pass/fail-like score passed (and the harness ran the case).

regressions 

regressions(
    baseline: dict[str, Any] | str | Path,
    *,
    tolerance: float = 0.02,
) -> list[Regression]

Metrics whose mean fell by more than tolerance against a saved baseline, plus a fall in the overall pass rate.

assert_ok 

assert_ok(
    *,
    min_pass_rate: float = 1.0,
    baseline: dict[str, Any] | str | Path | None = None,
    tolerance: float = 0.02,
) -> None

For CI: raise AssertionError with the failing cases if the run is below the bar or regressed.

OnlineEvaluator 

OnlineEvaluator(
    engine: Engine,
    metrics: Sequence[Metric],
    *,
    sample: float = 1.0,
    sinks: Sequence[Any] = (),
    max_cost: float | None = None,
    tenant: str | None = None,
    group: str = "aaa.eval",
)

Metric

Bases: Protocol

LLMJudge 

LLMJudge(
    client: LLMClient,
    rubric: str,
    *,
    model: str | None = None,
    name: str = "judge",
    use_expected: bool = True,
)

A model grades the answer against a rubric. The judge runs OUTSIDE the evaluated run (its calls are not part of that run's trace or budget); what it cost is reported on the Score and summed in the report.

Budgeted 

Budgeted(
    *,
    steps: int | None = None,
    tokens: int | None = None,
    cost: float | None = None,
    latency_s: float | None = None,
    name: str = "within_budget",
)

Within a cost/step/token/latency allowance; the score falls linearly past the limit to 0 at twice the limit.

AutoevalsMetric 

AutoevalsMetric(
    scorer: Any,
    *,
    name: str | None = None,
    pass_threshold: float | None = None,
)

Run one autoevals scorer on the case: output is the agent's answer, expected the case's, input the task text.

openai_shim 

openai_shim(
    client: LLMClient, *, model: str | None = None
) -> Any

An OpenAI-SDK-shaped object backed by an AAA LLMClient (async only: use autoevals' eval_async).

DeepEvalMetric 

DeepEvalMetric(
    metric: Any,
    *,
    name: str | None = None,
    retrieval_context_key: str = "retrieval_context",
    context_key: str = "context",
)

Run one DeepEval metric (single-turn LLMTestCase metrics) on a case result.

DeepEvalJudge 

DeepEvalJudge(
    client: LLMClient,
    model: str | None = None,
    **defaults: Any,
) -> Any

A DeepEvalBaseLLM over client. Totals of what it used are on .calls, .tokens and .cost.

RagasMetric 

RagasMetric(metric: Any, *, name: str | None = None)

RagasJudge 

RagasJudge(
    client: LLMClient,
    model: str | None = None,
    **params: Any,
) -> Any

An InstructorBaseRagasLLM over client: structured replies are requested with the pydantic model's JSON Schema.

aaa_solver 

aaa_solver(
    setup: Callable[
        [],
        AbstractAsyncContextManager[
            tuple[Engine, Algorithm]
        ],
    ],
    *,
    make_input: Callable[
        [str], dict[str, Any]
    ] = lambda text: {"task": text},
    extract: Callable[[Any], str] = default_extract,
    budget: Budget | None = None,
) -> Any

An Inspect Solver that runs one AAA run per sample and puts its answer in state.output.

Traces, red team and simulation

dataset_from_traces async 

dataset_from_traces(
    engine: Engine,
    *,
    filter: Callable[[RunInfo], bool] | None = None,
    algo: str | None = None,
    tenant: str | None = None,
    outcomes: tuple[str, ...] | None = (
        OutcomeKind.SUCCESS.value,
    ),
    since: datetime | None = None,
    limit: int = 100,
    scrub: Redactor | bool = True,
    expected: str | None = None,
    include_tenant: bool = False,
    name: str = "traces",
) -> Dataset

Build a Dataset from finished top-level runs in engine's store.

  • Skipped always: child runs, runs started by an eval (aaa_eval), runs still open, runs whose payloads were erased, session runs (their turns are runs of the inner algorithm; build session cases by hand).
  • algo / tenant / outcomes / since / filter narrow the selection; the newest limit matching runs are kept.
  • scrub=True redacts each input with standard_redactor() (emails, cards, tokens ...); pass your own Redactor, or False to keep inputs verbatim (only for data you are allowed to copy).
  • expected="output" records what the run answered as the case's expected value: a golden file for regression testing, which pins down CURRENT behaviour, not correct behaviour. Review it before trusting it.
  • The case id is the run id (so a case can be traced back); include_tenant also copies the tenant into the metadata.

evaluate_guard async 

evaluate_guard(
    guard: Guardrail,
    *,
    stage: Stage = "input",
    attack_set: Dataset | None = None,
    benign_set: Dataset | None = None,
) -> RedTeamReport

Run the attacks and the benign prompts through guard.check at stage and tabulate the actions it took.

RedTeamReport dataclass 

RedTeamReport(
    guard: str,
    attacks: list[Row] = list(),
    benign: list[Row] = list(),
)

attack_success_rate property 

attack_success_rate: float

The share of attacks the guard did NOT stop (allowed, or only flagged).

false_positive_rate property 

false_positive_rate: float

The share of harmless prompts the guard stopped.

assert_ok 

assert_ok(
    *,
    max_attack_success: float = 1.0,
    max_false_positive: float = 1.0,
) -> None

Raise AssertionError (listing the missed attacks) when the rates are worse than your thresholds: for CI.

attack_suite 

attack_suite(
    algo: Algorithm,
    *,
    canary: str | None = None,
    categories: Sequence[str] | None = None,
    judge: Any = None,
    budget: Budget | None = None,
    **kw: Any,
) -> EvalSuite

An EvalSuite that throws the attacks at algo. Metrics: resisted, and no_leak when you give a canary. Read a failure as "this attack worked on this agent today", then fix the agent and keep the case.

NoLeak 

NoLeak(canary: str, *, name: str = 'no_leak')

Fails if the agent's answer contains a planted canary (a secret string you put in its system prompt or tools).

Resisted 

Resisted(*, judge: Any = None, name: str = 'resisted')

Passes if the attack did not work: a guardrail blocked the run, or the answer is a refusal (matched by phrases, or by judge, a metric such as LLMJudge, for a stronger reading). A heuristic: use a judge for anything that matters.

SimulatedUser 

SimulatedUser(
    client: Any,
    *,
    persona: str = "a polite customer",
    goal: str = "get the question answered",
    opening: str | None = None,
    model: str | None = None,
    max_cost: float | None = None,
)

An LLM-driven user. client is any LLMClient; its calls are counted in spent/calls and capped by max_cost.

for_case 

for_case(case: Case) -> SimulatedUser

A copy using the case's own persona/goal/opening where it gives them (the cost counters stay shared).

next_message async 

next_message(
    transcript: Sequence[tuple[str, str]],
) -> str | None

The user's next message given the (user, assistant) pairs so far, or None when the conversation is over.

SessionEvalRunner 

SessionEvalRunner(
    manager: SessionManager,
    user: SimulatedUser,
    metrics: Sequence[Metric],
    *,
    max_turns: int = 6,
    concurrency: int = 2,
    turn_timeout_s: float = 120.0,
    tenant: str | None = None,
    name: str = "conversations",
)

Run each case of a dataset as a conversation between user and a session of manager.