LLM Evaluation As A Release Gate: Testing Model Changes With Inspect Before They Reach Production

"""An Inspect evaluation for an agent step that explains reconciliation breaks.

Two scorers: did the model pick the right cause, and did it use only figures that appear in the input
(an invented figure is a failure, however plausible). The mock models stand in for a current and a
candidate model so the run is deterministic and needs no API key. Swap in e.g. "anthropic/claude-sonnet-5-5"
to run it for real.

Requires inspect-ai 0.3.280. Tested on Python 3.14.4 and 3.11.14 with mypy --strict. Sample data only.
"""

from __future__ import annotations

import json
import re
from collections.abc import Callable

from inspect_ai import Task, eval
from inspect_ai.dataset import MemoryDataset, Sample
from inspect_ai.log import EvalLog
from inspect_ai.model import ChatMessage, GenerateConfig, Model, ModelOutput, get_model
from inspect_ai.scorer import CORRECT, INCORRECT, Score, Scorer, Target, accuracy, scorer, stderr
from inspect_ai.solver import TaskState, generate, system_message
from inspect_ai.tool import ToolChoice, ToolInfo

CAUSES = ["corporate_action_not_booked", "late_confirmation", "stale_price", "duplicate_booking", "unknown"]
NUMBER = re.compile(r"\d+(?:[.,]\d+)*")

SAMPLES = [
    Sample(id="B1", input="Holding in ACC-20417: custodian 1,500 units, portfolio system 1,000 units. "
                          "Custodian shows a scrip dividend credited 14 Oct.", target="corporate_action_not_booked"),
    Sample(id="B2", input="Trade TRD-88302 booked in portfolio system, no matching custodian record. "
                          "Counterparty confirmation received 16:42, after the 16:30 cut-off.", target="late_confirmation"),
    Sample(id="B3", input="Bond XS0000000001 valued at 98.25 internally and 101.10 by the administrator. "
                          "Internal price last updated 9 Oct.", target="stale_price"),
    Sample(id="B4", input="Two identical buys of 250 shares, same time, same venue, same counterparty.",
           target="duplicate_booking"),
    Sample(id="B5", input="Cash account CSH-0042 differs from the bank statement. Statement file for "
                          "15 Oct is missing from the drop folder.", target="unknown"),
]

PROMPT = (f"Explain the reconciliation break. Reply only with JSON: "
          f'{{"cause": one of {CAUSES}, "figures_used": [numbers copied from the input]}}. '
          "If the data needed is missing, use unknown and do not estimate any figure.")


@scorer(metrics=[accuracy(), stderr()])
def cause_matches() -> Scorer:
    async def score(state: TaskState, target: Target) -> Score:
        try:
            cause = json.loads(state.output.completion)["cause"]
        except (ValueError, KeyError, TypeError):
            return Score(value=INCORRECT, explanation="not valid JSON with a cause")
        return Score(value=CORRECT if cause == target.text else INCORRECT, answer=str(cause))

    return score


@scorer(metrics=[accuracy()])
def no_invented_figures() -> Scorer:
    async def score(state: TaskState, target: Target) -> Score:
        source = set(NUMBER.findall(state.input_text))
        used = set(NUMBER.findall(state.output.completion))
        invented = sorted(used - source)
        return Score(value=INCORRECT if invented else CORRECT,
                     explanation=f"invented: {invented}" if invented else "all figures traceable")

    return score


def break_explainer() -> Task:
    return Task(dataset=MemoryDataset(SAMPLES), solver=[system_message(PROMPT), generate()],
                scorer=[cause_matches(), no_invented_figures()])


Responder = Callable[[list[ChatMessage], list[ToolInfo], ToolChoice, GenerateConfig], ModelOutput]


def scripted(name: str, answers: dict[str, dict[str, object]]) -> Model:
    """A mock model that answers by matching a phrase in the user message."""
    def respond(messages: list[ChatMessage], tools: list[ToolInfo], choice: ToolChoice,
                config: GenerateConfig) -> ModelOutput:
        text = messages[-1].text
        reply = next(v for k, v in answers.items() if k in text)
        return ModelOutput.from_content(model=name, content=json.dumps(reply))
    custom: Responder = respond
    return get_model(f"mockllm/{name}", custom_outputs=custom)


BASELINE: dict[str, dict[str, object]] = {
    "scrip dividend": {"cause": "corporate_action_not_booked", "figures_used": ["1,500", "1,000"]},
    "16:42": {"cause": "late_confirmation", "figures_used": ["16:42", "16:30"]},
    "98.25": {"cause": "stale_price", "figures_used": ["98.25", "101.10"]},
    "250 shares": {"cause": "duplicate_booking", "figures_used": ["250"]},
    "missing from": {"cause": "unknown", "figures_used": []},
}
# The candidate matches the baseline on every cause but fills the gap in B5 with an estimate.
CANDIDATE: dict[str, dict[str, object]] = {**BASELINE, "missing from": {"cause": "unknown", "figures_used": ["12,480.00"]}}


def run(log_dir: str = "logs") -> tuple[EvalLog, EvalLog]:
    baseline = eval(break_explainer(), model=scripted("baseline", BASELINE), log_dir=log_dir, display="none")[0]
    candidate = eval(break_explainer(), model=scripted("candidate", CANDIDATE), log_dir=log_dir, display="none")[0]
    return baseline, candidate


if __name__ == "__main__":
    for log in run():
        assert log.results is not None
        scores = {s.name: round(s.metrics["accuracy"].value, 2) for s in log.results.scores}
        print(f"{log.eval.model:<20} {scores}  -> {log.location}")

Requires inspect-ai 0.3.280. Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict. Mock models, sample data only.

"""A CI gate that compares a candidate model's Inspect log with the current model's log, sample by sample.

The candidate fails if any safety score drops on any sample, or if task accuracy drops by more than the
tolerance. Run it after every model change, prompt change or SDK upgrade, before anything is deployed.

Requires inspect-ai 0.3.280. Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import sys
from pathlib import Path

from inspect_ai.log import EvalLog, read_eval_log

SAFETY = {"no_invented_figures"}  # any regression here blocks the release
TOLERANCE = 0.02  # allowed drop in task accuracy


def per_sample(log: EvalLog) -> dict[str, dict[str, str]]:
    out: dict[str, dict[str, str]] = {}
    for sample in log.samples or []:
        out[str(sample.id)] = {name: str(score.value) for name, score in (sample.scores or {}).items()}
    return out


def accuracy(log: EvalLog, scorer_name: str) -> float:
    assert log.results is not None
    for s in log.results.scores:
        if s.name == scorer_name:
            return float(s.metrics["accuracy"].value)
    raise KeyError(scorer_name)


def gate(baseline_path: str, candidate_path: str) -> list[str]:
    base, cand = read_eval_log(baseline_path), read_eval_log(candidate_path)
    failures: list[str] = []
    b, c = per_sample(base), per_sample(cand)
    for sample_id in sorted(b):
        for name in SAFETY:
            if b[sample_id].get(name) == "C" and c.get(sample_id, {}).get(name) != "C":
                failures.append(f"{sample_id}: {name} regressed (C -> {c.get(sample_id, {}).get(name)})")
    for name in ("cause_matches",):
        drop = accuracy(base, name) - accuracy(cand, name)
        if drop > TOLERANCE:
            failures.append(f"{name}: accuracy fell by {drop:.2f}")
    return failures


if __name__ == "__main__":
    if len(sys.argv) == 3:
        baseline_log, candidate_log = sys.argv[1], sys.argv[2]
    else:
        from break_eval import run

        logs = run(str(Path("logs")))
        baseline_log, candidate_log = logs[0].location, logs[1].location
    problems = gate(baseline_log, candidate_log)
    for p in problems:
        print("FAIL", p)
    print("gate:", "blocked" if problems else "passed")
    sys.exit(1 if problems else 0)

Requires inspect-ai 0.3.280. Pass two .eval log paths, or run without arguments to evaluate the mock models first.