aiml
SearchMy learning
Menu

Lesson 2 of 6

Measure RAG and tool-call quality

Evaluate a LangChain docs agent stage by stage with DeepEval, score its tool calls, retrieval and answer faithfulness, and report the first stage that fails.

Prerequisites

Python 3.11 or later, and the pytest and record/replay ideas from the eval harness lesson. Familiarity with embeddings and vector search helps. Live runs need a Gemini API key in GOOGLE_API_KEY, or Ollama with gemma4:e2b and qwen3-embedding:0.6b pulled and DOCS_MODEL=gemma4:e2b DOCS_EMBEDDER=qwen3-embedding:0.6b set; replaying the recordings needs neither.

By the end, you can

  • Record an agent's trajectory, tool calls, retrieved page IDs and answer, as the input to its evaluation.
  • Score tool use and page-ID retrieval separately from answer faithfulness and contextual recall, and name the first stage that fails.
  • Calibrate a judged metric with known-bad answers and choose its settings from the result.
Runtime
Python 3.11+; a Gemini API key for live runs (or Ollama with gemma4:e2b and qwen3-embedding:0.6b)
Dependencies
DeepEval 4.3.1, LangChain 1.4.6, langchain-google-genai 4.4.2 and pytest 9.1.1; langchain-ollama 1.1.0 for the local alternative; pinned in requirements-lock.txt

Lab code, tests and fixtures (ZIP, 35 KB)Quiz check your understanding

Progress is saved in this browser.

A docs assistant that retrieves pages and calls tools can fail in several places: it calls the wrong tool, it retrieves the wrong page, it answers beyond what the page says, or it answers a different question. One overall score says that something failed, not where. You will evaluate a LangChain agent that answers questions about tracker, a sample task-tracker CLI, and score each stage of every answer separately: tool calls with DeepEval’s ToolCorrectnessMetric, retrieval with rank metrics from page IDs, source coverage with contextual recall, and the answer with DeepEval’s faithfulness and answer relevancy metrics. The report names the first stage that fails.

When you need this. Your docs assistant gave a wrong answer, and you need to know whether to fix the retriever, the tool descriptions or the prompt. A single quality score can’t tell you. A per-stage report identifies the first failed check in its chosen check order, giving you a place to investigate.

The saved report evaluates gemini-3.8-flash with gemini-3.8-flash as the judge. Step 6 shows how to rebuild and inspect it.

python evaluate.py · one trace per case, five stages per trace
cases.json10 questionsexpected tools, pagesand a referencequestionassistant.py · create_agentagentgemini-3.8-flashpicks toolsquerypagessearch_docsquerytop 3InMemoryVectorStore10 pages, docs.jsongemini-embedding-2numberissueget_issuelookupissues.jsonissue state, fix versionmessagestrace from run()tool calls, tool outputs,retrieved page IDs, answertracetoolsToolCorrectness+ unexpected callsretrievalhit@3, recall@3from page IDsfaithfulnessclaims supportedby the context?contextual recallreference foundin the context?answer relevancystatements onthe question?scoresfirst_failure()the first stage that fails, in orderrowruns/<model>.mdand <model>.jsondata flowreturncases.json10 questions with expectationsquestionagentgemini-3.8-flash · create_agentquerypagessearch_docsquerytop 3Vector store10 pagesnumberissueget_issuelookupissues.jsonmessagestracecalls, outputs, page IDs, answertracetoolsToolCorrectness + unexpectedretrievalhit@3, recall@3, RRfaithfulnessjudge: claims supported?contextual recalljudge: reference found?answer relevancyjudge: on the question?scoresfirst_failure()first failing stagerowruns/<model>.mddata flowreturn
Every stage is scored on every trace; first_failure() then picks the earliest failing stage, so a missed page is reported as a retrieval failure even when the judge also marks the answer unfaithful. Agent traces and judge verdicts are recorded, so the report reruns without a model. Contextual precision is reported but does not gate. Values come from runs/gemini-3.8-flash.json.

This lesson evaluates a docs agent, while the first harness, robustness tests and regression-gate lab use issue triage. Their case schemas and reports are different. The shared method is versioned inputs, recorded calls and explicit graders; this stage report is not an input to the triage pytest gate.

Step 1: Build the system under test

Download the complete lab, extract it and open a terminal in build-retrieval-generation-and-tool-execution-evaluation-reports/. The README lists its runtime and recorded examples.

Prepare the local labTerminal · bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt
assistant.pySource · pythonDownload
"""Docs assistant: a LangChain agent on Gemini that searches the tracker docs and looks up issues.
Set DOCS_MODEL=gemma4:e2b and DOCS_EMBEDDER=qwen3-embedding:0.6b to run it on a local Ollama server."""
import json
import os
from pathlib import Path

from langchain.agents import create_agent
from langchain_core.documents import Document
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_core.vectorstores import InMemoryVectorStore

HERE = Path(__file__).parent
CHAT_MODEL = os.environ.get("DOCS_MODEL", "gemini-3.8-flash")
EMBED_MODEL = os.environ.get("DOCS_EMBEDDER", "gemini-embedding-2")
K = 3
PROMPT = """You answer questions about `tracker`, an open-source task-tracker CLI.
Use search_docs for questions about how tracker works and get_issue for a numbered issue.
Answer only from what the tools return, in at most three sentences.
If a message needs no lookup, reply without calling a tool."""


def models():
    if CHAT_MODEL.startswith("gemini"):
        from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
        return (ChatGoogleGenerativeAI(model=CHAT_MODEL, thinking_level="low"),
                GoogleGenerativeAIEmbeddings(model=EMBED_MODEL))
    from langchain_ollama import ChatOllama, OllamaEmbeddings
    return ChatOllama(model=CHAT_MODEL, temperature=0, reasoning=False), OllamaEmbeddings(model=EMBED_MODEL)


def build_agent(docs: list[dict], issues: dict):
    chat, embeddings = models()
    store = InMemoryVectorStore.from_documents(
        [Document(page_content=f"{d['title']}\n{d['text']}", id=d["id"]) for d in docs], embeddings)

    @tool
    def search_docs(query: str) -> str:
        """Search the tracker documentation and return the three closest pages."""
        return json.dumps([{"id": d.id, "text": d.page_content} for d in store.similarity_search(query, k=K)])

    @tool
    def get_issue(number: int) -> str:
        """Look up a tracker GitHub issue by number: its title, state and the version that fixed it."""
        issue = issues.get(str(number))
        return json.dumps({"number": number, **issue}) if issue else f"issue {number} does not exist"

    return create_agent(chat, tools=[search_docs, get_issue], system_prompt=PROMPT)


def run(agent, question: str) -> dict:
    """Run one question and keep the trajectory the evaluation needs."""
    messages = agent.invoke({"messages": [{"role": "user", "content": question}]})["messages"]
    calls = [{"name": c["name"], "args": c["args"]} for m in messages if isinstance(m, AIMessage) for c in m.tool_calls]
    outputs = [{"tool": m.name, "content": m.content} for m in messages if isinstance(m, ToolMessage)]
    retrieved = [d["id"] for o in outputs if o["tool"] == "search_docs" for d in json.loads(o["content"])]
    return {"answer": messages[-1].text, "tool_calls": calls, "tool_outputs": outputs, "retrieved": retrieved}


if __name__ == "__main__":
    import sys
    agent = build_agent(json.loads((HERE / "docs.json").read_text()), json.loads((HERE / "issues.json").read_text()))
    print(json.dumps(run(agent, sys.argv[1]), indent=2))

create_agent builds the tool-calling loop: the model either answers or asks for a tool, LangChain runs the tool and returns its output as a ToolMessage, and the loop repeats until the model answers. search_docs queries an InMemoryVectorStore of the 10 documentation pages, embedded with gemini-embedding-2. An in-memory store does exact search over every vector, which is right for a few dozen pages and a test fixture; a production index of many thousands of pages belongs in pgvector or another persistent store. Each Document keeps the page’s ID, so retrieval can be scored by ID instead of by reading text.

run() is the part that matters for evaluation. It keeps the whole trajectory from the agent’s messages: every tool call with its arguments, every tool output, the IDs of the retrieved pages and the final answer. An evaluation that sees only the final answer can’t tell a wrong tool from a wrong page from a wrong summary.

System prompt edited. Replay stops with prompt_sha changed; record the reviewed prompt before comparing reports.

Recording has a key field the code no longer sends. Replay stops and names the field; regenerate through the current recording code rather than deleting fields by hand.

Step 2: Write cases for every stage

Each case states what a correct trajectory looks like: the tools that must be called, the pages that must be retrieved, and a reference answer.

cases.jsonSource · jsonDownload
{
  "name": "docs-assistant",
  "version": 1,
  "cases": [
    {"id": "backup", "question": "Where does tracker keep its data, and what should I back up?", "expected_docs": ["storage"], "expected_tools": ["search_docs"], "reference": "Tasks live in tasks.db inside the list folder, ~/.tracker by default; back up that file."},
    {"id": "two-laptops", "question": "Is it safe to sync from two laptops at the same time?", "expected_docs": ["sync"], "expected_tools": ["search_docs"], "reference": "Yes. Sync locks the remote, so the second machine waits up to 30 seconds and then exits with 'remote is locked'."},
    {"id": "restore", "question": "How do I bring back a task I archived?", "expected_docs": ["archive"], "expected_tools": ["search_docs"], "reference": "Run tracker restore ID."},
    {"id": "config-keys", "question": "Which keys can I put in config.toml?", "expected_docs": ["config"], "expected_tools": ["search_docs"], "reference": "default_project, week_start, color and editor."},
    {"id": "pipe-colors", "question": "My grep fails because tracker list prints color codes into the pipe. How do I stop that?", "expected_docs": ["output", "config"], "expected_tools": ["search_docs"], "reference": "With color = auto, tracker turns color off when output is not a terminal; set color = never in config.toml to turn it off always."},
    {"id": "dry-run-import", "question": "How can I check a CSV import without writing any tasks?", "expected_docs": ["import"], "expected_tools": ["search_docs"], "reference": "Run tracker import FILE.csv --dry-run."},
    {"id": "issue-212", "question": "Is issue 212 fixed? Which version has the fix?", "expected_docs": [], "expected_tools": ["get_issue"], "reference": "Yes, issue 212 is closed and fixed in 2.4.1."},
    {"id": "issue-231", "question": "What is the status of issue 231?", "expected_docs": [], "expected_tools": ["get_issue"], "reference": "Issue 231, import drops rows, is still open."},
    {"id": "token-leak", "question": "Issue 218 says --debug leaks my API token. Which version fixes it, and where does debug output go?", "expected_docs": ["debug"], "expected_tools": ["get_issue", "search_docs"], "reference": "2.4.1 fixes it by redacting tokens; debug output goes to debug.log in the list folder."},
    {"id": "thanks", "question": "Thanks, tracker has been great for my team!", "expected_docs": [], "expected_tools": [], "reference": "A short thank-you reply with no lookup."}
  ]
}

The cases cover the agent’s real paths: docs questions, issue lookups, one question that needs both tools (token-leak), one that needs two pages (pipe-colors) and a thank-you that needs no tool at all. Expected pages are empty where no search should happen, so those cases skip the retrieval stage instead of counting as a miss.

Case, page or issue edited without a version bump. Replay still stops with cases_sha changed or docs_sha changed; record the new inputs after reviewing the edit.

Step 3: Score tool calls and retrieval

Now we score the trajectory, starting with the two stages that need no judge.

evaluate.pySource · pythonDownload
"""Evaluate the docs assistant per stage (tools, retrieval, grounding) and write a report."""
import argparse
import functools
import hashlib
import json
import os
import sys
from pathlib import Path

os.environ.setdefault("DEEPEVAL_TELEMETRY_OPT_OUT", "1")
from deepeval.metrics import (  # noqa: E402
    AnswerRelevancyMetric, ContextualPrecisionMetric, ContextualRecallMetric, FaithfulnessMetric, ToolCorrectnessMetric)
from deepeval.models import GeminiModel, OllamaModel  # noqa: E402
from deepeval.test_case import LLMTestCase, ToolCall  # noqa: E402

from assistant import CHAT_MODEL, EMBED_MODEL, K, PROMPT, build_agent, run  # noqa: E402
from cassette import Cassette  # noqa: E402

HERE = Path(__file__).parent
THRESHOLD = 0.7
STAGES = ["tools", "retrieval", "faithfulness", "contextual_recall", "answer_relevancy"]
# calibrate.py: the default faithfulness metric passed an answer with an unsupported claim; strict mode caught it.
STRICT_FAITHFULNESS = True
JUDGE_MODEL = "gemini-3.8-flash"  # fixed, so scores stay comparable when the assistant's model changes


def sha(value) -> str:
    """Content hash: editing a case, a page, an issue or the prompt changes it."""
    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()[:12]


def recording_key(cases: dict, docs: list[dict], issues: dict) -> dict:
    """Everything a recorded trace or verdict depends on. A replay with any other value is refused."""
    return {"cases": f"{cases['name']}@{cases['version']}", "cases_sha": sha(cases),
            "docs_sha": sha({"docs": docs, "issues": issues}), "prompt_sha": sha(PROMPT), "k": K,
            "model": CHAT_MODEL, "embedder": EMBED_MODEL, "judge": JUDGE_MODEL,
            "strict_faithfulness": STRICT_FAITHFULNESS}


def make_judge(name: str = JUDGE_MODEL):
    if name.startswith("gemini"):
        return GeminiModel(model=name)
    return OllamaModel(model=name, base_url="http://localhost:11434", temperature=0)


def retrieval_scores(retrieved: list[str], expected: list[str]) -> dict | None:
    if not expected:
        return None
    ranks = [retrieved.index(doc) + 1 for doc in expected if doc in retrieved]
    return {"hit": bool(ranks), "recall": len(ranks) / len(expected),
            "reciprocal_rank": 1 / min(ranks) if ranks else 0.0}


def tool_scores(question: str, calls: list[dict], expected: list[str], judge) -> dict:
    case = LLMTestCase(
        input=question,
        tools_called=[ToolCall(name=c["name"], input_parameters=c["args"]) for c in calls],
        expected_tools=[ToolCall(name=name) for name in expected],
    )
    # Without available_tools the metric only compares names and never calls the judge,
    # but it still requires a model object.
    metric = ToolCorrectnessMetric(model=judge, async_mode=False)
    metric.measure(case)
    unexpected = [c["name"] for c in calls if c["name"] not in expected]
    return {"score": metric.score, "unexpected": unexpected, "pass": metric.score == 1.0 and not unexpected}


def judged_scores(case: dict, trace: dict, judge) -> dict:
    """DeepEval's judged metrics. Context metrics run only when a tool returned context."""
    context = [d["text"] for o in trace["tool_outputs"] if o["tool"] == "search_docs" for d in json.loads(o["content"])]
    context += [o["content"] for o in trace["tool_outputs"] if o["tool"] == "get_issue"]
    test_case = LLMTestCase(input=case["question"], actual_output=trace["answer"],
                            expected_output=case["reference"], retrieval_context=context or None)
    metrics = {"answer_relevancy": AnswerRelevancyMetric(THRESHOLD, model=judge, async_mode=False)}
    if context:
        metrics |= {"faithfulness": FaithfulnessMetric(THRESHOLD, model=judge, async_mode=False,
                                                       penalize_ambiguous_claims=STRICT_FAITHFULNESS),
                    "contextual_precision": ContextualPrecisionMetric(THRESHOLD, model=judge, async_mode=False),
                    "contextual_recall": ContextualRecallMetric(THRESHOLD, model=judge, async_mode=False)}
    scores = {}
    for name, metric in metrics.items():
        metric.measure(test_case)
        scores[name] = {"score": metric.score, "reason": metric.reason}
    return scores


def first_failure(row: dict) -> str | None:
    checks = {"tools": row["tools"]["pass"],
              "retrieval": row["retrieval"] is None or row["retrieval"]["recall"] == 1.0,
              **{n: row["judged"][n]["score"] >= THRESHOLD for n in row["judged"]}}
    return next((stage for stage in STAGES if checks.get(stage) is False), None)


def evaluate(cases: dict, cassette: Cassette, judge) -> dict:
    """`judge` is a callable that builds the judge model, so a replay needs no API key."""
    agent = None
    rows = []
    for case in cases["cases"]:
        def live_run():
            nonlocal agent
            agent = agent or build_agent(json.loads((HERE / "docs.json").read_text()),
                                         json.loads((HERE / "issues.json").read_text()))
            return run(agent, case["question"])
        trace = cassette.get(f"agent/{case['id']}", live_run)
        row = {"id": case["id"], "answer": trace["answer"], "calls": [c["name"] for c in trace["tool_calls"]],
               "tools": cassette.get(f"tools/{case['id']}", lambda: tool_scores(
                   case["question"], trace["tool_calls"], case["expected_tools"], judge())),
               "retrieval": retrieval_scores(trace["retrieved"], case["expected_docs"]),
               "judged": cassette.get(f"judge/{case['id']}", lambda: judged_scores(case, trace, judge()))}
        rows.append({**row, "first_failure": first_failure(row)})
        print(f"evaluated {case['id']} ({len(rows)}/{len(cases['cases'])})", file=sys.stderr, flush=True)
    return {"key": cassette.key, "k": K, "threshold": THRESHOLD, "rows": rows, "summary": summarize(rows)}


def summarize(rows: list[dict]) -> dict:
    ranked = [r["retrieval"] for r in rows if r["retrieval"]]
    faith = [r["judged"]["faithfulness"]["score"] for r in rows if "faithfulness" in r["judged"]]
    return {"cases": len(rows), "passed": sum(r["first_failure"] is None for r in rows),
            "tool_pass": sum(r["tools"]["pass"] for r in rows),
            f"hit_at_{K}": f"{sum(r['hit'] for r in ranked)}/{len(ranked)}",
            f"recall_at_{K}": round(sum(r["recall"] for r in ranked) / len(ranked), 3),
            "mrr": round(sum(r["reciprocal_rank"] for r in ranked) / len(ranked), 3),
            "mean_faithfulness": round(sum(faith) / len(faith), 3)}


def markdown(report: dict) -> str:
    fmt = lambda x: "-" if x is None else f"{x:.2f}"  # noqa: E731
    lines = [f"| case | tools called | tools | hit@{report['k']} | recall@{report['k']} | RR | ctx precision | "
             "faithful | ctx recall | relevancy | first failure |",
             "|---|---|---|---|---|---|---|---|---|---|---|"]
    for r in report["rows"]:
        ret, j = r["retrieval"], r["judged"]
        lines.append(f"| {r['id']} | {', '.join(r['calls']) or 'none'} | {'pass' if r['tools']['pass'] else 'FAIL'} | "
                     f"{'-' if ret is None else ('yes' if ret['hit'] else 'NO')} | {fmt(ret and ret['recall'])} | "
                     f"{fmt(ret and ret['reciprocal_rank'])} | {fmt(j.get('contextual_precision', {}).get('score'))} | "
                     f"{fmt(j.get('faithfulness', {}).get('score'))} | {fmt(j.get('contextual_recall', {}).get('score'))} | "
                     f"{fmt(j.get('answer_relevancy', {}).get('score'))} | "
                     f"{r['first_failure'] or '-'} |")
    s, k = report["summary"], report["k"]
    lines.append(f"\n{s['passed']}/{s['cases']} cases pass every stage. Tools {s['tool_pass']}/{s['cases']}, "
                 f"hit@{k} {s[f'hit_at_{k}']}, recall@{k} {s[f'recall_at_{k}']}, MRR {s['mrr']}, "
                 f"mean faithfulness {s['mean_faithfulness']}.")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--live", action="store_true", help="call the models and rewrite the recording")
    live = parser.parse_args().live
    cases = json.loads((HERE / "cases.json").read_text())
    key = recording_key(cases, json.loads((HERE / "docs.json").read_text()), json.loads((HERE / "issues.json").read_text()))
    name = CHAT_MODEL.replace(":", "-")  # one recording and one report per model under test
    cassette = Cassette(HERE / "recordings" / f"{name}.json", key, live)
    report = evaluate(cases, cassette, functools.cache(make_judge))
    if live:
        cassette.save()
    (HERE / "runs" / f"{name}.json").write_text(json.dumps(report, indent=2) + "\n")
    (HERE / "runs" / f"{name}.md").write_text(markdown(report) + "\n")
    print(markdown(report))


if __name__ == "__main__":
    main()

ToolCorrectnessMetric compares the names of the tools called with the expected ones. Without available_tools it makes no model call, but it still needs a model object, which is why tool_scores() takes the judge. The metric scores the share of expected tools that were called, so an extra call does not lower it. tool_scores() adds an unexpected list and fails the stage on any extra call: a search the question didn’t need costs latency and can pull in context that misleads the answer.

Retrieval needs no judge, because every case lists its pages by ID. retrieval_scores() computes three numbers at the depth the agent sees, 3 pages: hit@3 (was any expected page retrieved), recall@3 (what share of them was) and reciprocal rank (1 divided by the rank of the first expected page). Hit alone hides a question that needs two pages and gets one; recall shows it.

Extra tool call. DeepEval’s score stays 1.0; tool_scores() lists the call and fails the stage.

Missing tool call. The score drops to the share of expected tools called.

Tool call when none was expected. The score is 0.0.

Hit hides a missing second page. hit@3 passes and recall@3 is 0.5.

Step 4: Score the answer with DeepEval’s judged metrics

Page-ID recall above counts expected documents found by the retriever. Contextual recall below asks a judge which reference-answer statements are supported by the retrieved text. Faithfulness asks which generated-answer claims that text supports. Correct page IDs can coexist with a poor generated answer; these are different denominators and different failure investigations.

judged_scores() builds one DeepEval LLMTestCase per case, with the question, the answer, the reference and the tool outputs as retrieval_context, and runs four metrics on it with gemini-3.8-flash as the judge:

  • Faithfulness: the judge extracts the claims in the answer and the facts in the context, then rules on each claim; the score is the share of claims the context supports.
  • Contextual recall: the share of the reference answer’s sentences that the retrieved context supports. A low score means retrieval missed something the answer needed.
  • Contextual precision: whether the relevant context chunks rank above the irrelevant ones. It is reported but does not gate, because the agent reads all 3 pages anyway.
  • Answer relevancy: the share of the answer’s statements that address the question.

The context metrics run only when a tool returned something; the thank-you case gets answer relevancy alone. Each verdict comes with the judge’s reason, and the reasons are recorded with the scores. For token-leak, contextual precision is 0.75, and the reason explains it: the issue record that confirms the fix is the fourth context item, behind two search results that don’t mention it. It is fourth because judged_scores() puts search results first and the get_issue output last. Precision scores order, so the way you assemble the context changes it, which is one more reason it reports here rather than gates.

Step 5: Calibrate the judge

Here is what happens when we check the judge itself before trusting its scores.

A judged metric is a model making a decision, so test it like one. calibration-cases.json holds answers and contexts with a known defect: an invented storage engine, a claim that no retrieved page makes, and contexts that are missing the page the answer needs. calibrate.py scores each one three times; a defect is caught only if all three scores fall below the threshold.

calibrate.pySource · pythonDownload
"""Calibrate the judge: answers and contexts with a known defect must score below the threshold."""
import argparse
import json
import os
from pathlib import Path

os.environ.setdefault("DEEPEVAL_TELEMETRY_OPT_OUT", "1")
from deepeval.metrics import ContextualRecallMetric, FaithfulnessMetric  # noqa: E402
from deepeval.test_case import LLMTestCase  # noqa: E402

from cassette import Cassette  # noqa: E402
from evaluate import JUDGE_MODEL, THRESHOLD, make_judge, sha  # noqa: E402

HERE = Path(__file__).parent
REPEATS = 3
METRICS = {
    "faithfulness": lambda judge: FaithfulnessMetric(THRESHOLD, model=judge, async_mode=False),
    # Borderline verdicts count as unsupported instead of supported.
    "faithfulness_strict": lambda judge: FaithfulnessMetric(THRESHOLD, model=judge, async_mode=False,
                                                            penalize_ambiguous_claims=True),
    "contextual_recall": lambda judge: ContextualRecallMetric(THRESHOLD, model=judge, async_mode=False),
}


def score(metric: str, case: dict, judge) -> dict:
    m = METRICS[metric](judge)
    m.measure(LLMTestCase(input=case["question"], actual_output=case["answer"],
                          expected_output=case["reference"], retrieval_context=case["context"]))
    return {"score": m.score, "reason": m.reason}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--live", action="store_true", help="call the judge and rewrite the recording")
    cases = json.loads((HERE / "calibration-cases.json").read_text())
    key = {"cases": f"{cases['name']}@{cases['version']}", "cases_sha": sha(cases), "judge": JUDGE_MODEL, "repeats": REPEATS}
    tape = Cassette(HERE / "recordings" / "calibration.json", key, parser.parse_args().live)
    judge = make_judge() if tape.live else None
    rows = []
    for case in cases["cases"]:
        scores = [tape.get(f"{case['id']}/{i}", lambda: score(case["metric"], case, judge))["score"] for i in range(REPEATS)]
        rows.append({"id": case["id"], "metric": case["metric"], "defect": case["defect"], "scores": scores,
                     "caught": all(s < THRESHOLD for s in scores)})
    if tape.live:
        tape.save()
    (HERE / "runs" / "calibration.json").write_text(json.dumps({"threshold": THRESHOLD, "rows": rows}, indent=2) + "\n")
    for r in rows:
        print(f"{r['id']:<26}{r['metric']:<21}{[round(x, 2) for x in r['scores']]}  {'caught' if r['caught'] else 'MISSED'}")


if __name__ == "__main__":
    main()
Terminal commandsTerminal · bash
python calibrate.py --live
Recorded outputRecorded output · textDownload
invented-storage          faithfulness         [0.0, 0.5, 0.5]  caught
csv-removes-color         faithfulness         [1.0, 1.0, 1.0]  MISSED
config-page-missing       contextual_recall    [0.0, 0.0, 0.0]  caught
debug-page-missing        contextual_recall    [0.0, 0.0, 0.0]  caught
invented-storage-strict   faithfulness_strict  [0.0, 0.0, 0.0]  caught
csv-removes-color-strict  faithfulness_strict  [0.6, 0.33, 0.33]  caught

Two results change how evaluate.py is configured. The default faithfulness metric scored the answer that claims --format csv removes color codes 1, 1, 1: it passed an unsupported claim every time. DeepEval counts a claim that the judge rules borderline as supported unless penalize_ambiguous_claims=True, and with that setting the same answer scored 0.6, 0.33, 0.33. That is why evaluate.py sets STRICT_FAITHFULNESS = True and records the setting in the key. The second result is variance: the invented-storage answer scored 0, 0.5, 0.5 on three identical calls. Gemini runs the judge at its default temperature, so a score near the threshold can pass on one run and fail on the next. Recording the verdicts makes the report reproducible; it doesn’t make one verdict certain.

Step 6: Report the first failing stage, and replay

first_failure() checks the stages in a fixed order, tools, retrieval, faithfulness, contextual recall and answer relevancy, and names the first one that fails. The order helps you choose where to investigate: missing source text can also lead to a faithfulness failure. Inspect the retrieved text and answer before deciding which component to change; the first failed check does not establish the cause. evaluate.py records every agent trace, tool score and judge verdict. recording_key() keys the recording to the cases’ name and version, a content hash of the cases (cases_sha), a hash of the pages and issues the tools read (docs_sha), a hash of the system prompt (prompt_sha), k, the models, the judge and the faithfulness setting. A replay with any different value stops with recording is stale, so an edited question or page can’t reuse old traces and verdicts. The hashes cover data and the prompt, not code: after a change to a tool description or to other code in assistant.py, record again with --live. When the inputs match, the report rebuilds offline:

Terminal commandsTerminal · bash
python evaluate.py
Saved Gemini evaluation reportRecorded output · markdownDownload
| case | tools called | tools | hit@3 | recall@3 | RR | ctx precision | faithful | ctx recall | relevancy | first failure |
|---|---|---|---|---|---|---|---|---|---|---|
| backup | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| two-laptops | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| restore | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| config-keys | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| pipe-colors | search_docs | pass | yes | 0.50 | 1.00 | 1.00 | 0.67 | 0.00 | 1.00 | retrieval |
| dry-run-import | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| issue-212 | get_issue | pass | - | - | - | 1.00 | 1.00 | 1.00 | 1.00 | - |
| issue-231 | get_issue | pass | - | - | - | 1.00 | 1.00 | 1.00 | 1.00 | - |
| token-leak | get_issue, search_docs | pass | yes | 1.00 | 1.00 | 0.75 | 1.00 | 1.00 | 1.00 | - |
| thanks | none | pass | - | - | - | - | - | - | 1.00 | - |

9/10 cases pass every stage. Tools 10/10, hit@3 7/7, recall@3 0.929, MRR 1.0, mean faithfulness 0.963.

In the run above, 9 of 10 cases pass every stage, and the one that fails is reported as a retrieval failure although three later stages also scored it low. The recorded examples below trace that case and compare the local-model run.

First failing stage. Retrieval is named before faithfulness when both fail.

Recording made with another judge. Replay stops with judge changed; keep one judge for the comparison or rerun both sides with the new judge.

Replay differs from the saved report. The rebuilt report must equal the saved JSON for both models.

A page that retrieval missed. For pipe-colors, the agent searched for “disable color output pipe tracker list” and got the pages output, install, storage. The output page is there; config, the page that documents disabling color with color = never, is not. hit@3 is yes and recall@3 is 0.5, which is exactly the case that hit alone hides. The answer that followed:

Color output follows the `color` setting. Set `color = auto` so that tracker automatically turns color off when standard output is not a terminal (such as when piped to `grep`). Alternatively, you can use `tracker list --format csv` for comma-separated output without table formatting.

Without the config page, the agent recommended color = auto, which is the behavior the reporter already has, and contextual recall scored 0 because no retrieved text supports the reference answer. Faithfulness scored 0.67 in strict mode. These judged scores flag the answer, while the missing expected page gives a concrete retrieval defect to address first. Try a larger K or a page that names the problem in the user’s words, then record again to check both retrieval and the answer.

The local model on the same cases. With DOCS_MODEL=gemma4:e2b and qwen3-embedding:0.6b, 7 of 10 cases pass every stage:

Recorded report, gemma4:e2bRecorded output · markdownDownload
| case | tools called | tools | hit@3 | recall@3 | RR | ctx precision | faithful | ctx recall | relevancy | first failure |
|---|---|---|---|---|---|---|---|---|---|---|
| backup | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 0.67 | 1.00 | 1.00 | faithfulness |
| two-laptops | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| restore | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| config-keys | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| pipe-colors | search_docs | pass | yes | 0.50 | 0.50 | 0.50 | 0.00 | 0.00 | 1.00 | retrieval |
| dry-run-import | search_docs | pass | yes | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | - |
| issue-212 | get_issue | pass | - | - | - | 1.00 | 1.00 | 1.00 | 1.00 | - |
| issue-231 | get_issue | pass | - | - | - | 1.00 | 1.00 | 1.00 | 1.00 | - |
| token-leak | get_issue | FAIL | NO | 0.00 | 0.00 | 1.00 | 0.50 | 0.00 | 1.00 | tools |
| thanks | none | pass | - | - | - | - | - | - | 1.00 | - |

7/10 cases pass every stage. Tools 9/10, hit@3 6/7, recall@3 0.786, MRR 0.786, mean faithfulness 0.796.
CaseFirst failureWhat happened
token-leaktoolsCalled only get_issue and answered “The issue with `--debug` leaking the API token is fixed in version 2.4.1. The documentation does not specify where debug output goes.” The docs do say where debug output goes; this recorded trajectory contains no docs search.
pipe-colorsretrievalMissed the config page, then claimed that –format csv stops the color codes, which no page says. Strict faithfulness scored it 0: the same defect that the calibration case csv-removes-color tests.
backupfaithfulnessFaithfulness 0.67: “The score is 0.67 because the output incorrectly prescribes backing up tasks.db for every task, tag, and comment, whereas the context only notes that backing it up preserves these items.”

The token-leak failure is the one only a trajectory shows: the answer reads as a reasonable “the docs don’t say”, and the missing search_docs call shows that the agent did not check the docs before saying so. The backup failure is a judgment call. Strict mode turned a paraphrase into an unsupported claim; a reviewer would likely accept the answer. Strict faithfulness trades a missed defect for an occasional false alarm like this one, and the calibration run is the measurement behind choosing that trade.

Offline evaluation checks. test_evaluate.py has 13 offline tests that replay the recordings and need no model:

Terminal commandsTerminal · bash
pytest -q test_evaluate.py
Inspect test_evaluate.py
test_evaluate.pySource · pythonDownload
"""Failure cases for the evaluation code. Replays the recording; no model call."""
import json
from pathlib import Path

import pytest
from deepeval.models import GeminiModel

from cassette import Cassette, RecordingError
import evaluate as evaluate_module
from evaluate import evaluate, first_failure, recording_key, retrieval_scores, tool_scores

HERE = Path(__file__).parent
# ToolCorrectnessMetric compares names here and never calls this model; the key is a placeholder.
JUDGE = GeminiModel(model="gemini-3.8-flash", api_key="offline")


def calls(*names):
    return [{"name": n, "args": {}} for n in names]


def test_hit_and_reciprocal_rank():
    assert retrieval_scores(["init", "storage", "sync"], ["storage"]) == {"hit": True, "recall": 1.0, "reciprocal_rank": 0.5}
    assert retrieval_scores(["init", "sync", "tokens"], ["storage"]) == {"hit": False, "recall": 0.0, "reciprocal_rank": 0.0}


def test_hit_hides_a_missing_second_page_and_recall_shows_it():
    scores = retrieval_scores(["import", "output", "init"], ["output", "config"])
    assert scores["hit"] and scores["recall"] == 0.5


def test_case_without_expected_pages_skips_retrieval():
    assert retrieval_scores(["init"], []) is None


def test_extra_tool_call_scores_one_in_deepeval_but_fails_here():
    scores = tool_scores("q", calls("get_issue", "search_docs"), ["get_issue"], JUDGE)
    assert scores["score"] == 1.0
    assert scores["unexpected"] == ["search_docs"] and not scores["pass"]


def test_missing_tool_call_fails():
    assert tool_scores("q", calls("search_docs"), ["get_issue", "search_docs"], JUDGE)["score"] == 0.5


def test_tool_call_when_none_expected_fails():
    assert tool_scores("q", calls("search_docs"), [], JUDGE)["score"] == 0.0


def test_first_failing_stage_is_named():
    row = {"tools": {"pass": True}, "retrieval": {"hit": True, "recall": 0.5},
           "judged": {"faithfulness": {"score": 0.2}, "contextual_recall": {"score": 0.0}}}
    assert first_failure(row) == "retrieval"
    row["retrieval"]["recall"] = 1.0
    assert first_failure(row) == "faithfulness"


TAPE = HERE / "recordings" / "gemini-3.8-flash.json"


def test_stale_recording_is_refused():
    recorded = json.loads(TAPE.read_text())["key"]
    with pytest.raises(RecordingError, match="judge changed"):
        Cassette(TAPE, {**recorded, "judge": "gemma4:e2b"}, live=False)


HASHED = ("cases_sha", "docs_sha", "prompt_sha", "k")


def inputs() -> dict:
    return {name: json.loads((HERE / f"{name}.json").read_text()) for name in ("cases", "docs", "issues")}


def hashed_key(cases, docs, issues) -> dict:
    key = recording_key(cases, docs, issues)
    return {k: key[k] for k in HASHED}


@pytest.mark.parametrize("name", ["gemini-3.8-flash", "gemma4-e2b"])
def test_recording_is_keyed_to_the_current_inputs(name):
    recorded = json.loads((HERE / "recordings" / f"{name}.json").read_text())["key"]
    assert {k: recorded[k] for k in HASHED} == hashed_key(**inputs())


def edit_case(i):
    i["cases"]["cases"][0]["question"] += " Answer in French."


def edit_page(i):
    i["docs"][0]["text"] += " Deprecated."


def edit_issue(i):
    next(iter(i["issues"].values()))["state"] = "open"


@pytest.mark.parametrize("edit, field", [(edit_case, "cases_sha"), (edit_page, "docs_sha"), (edit_issue, "docs_sha")])
def test_edited_input_without_a_version_bump_makes_the_recording_stale(edit, field):
    edited = inputs()
    edit(edited)
    recorded = json.loads(TAPE.read_text())["key"]
    with pytest.raises(RecordingError, match=f"{field} changed"):
        Cassette(TAPE, {**recorded, **hashed_key(**edited)}, live=False)


def test_edited_prompt_makes_the_recording_stale(monkeypatch):
    monkeypatch.setattr(evaluate_module, "PROMPT", evaluate_module.PROMPT + "\nAlways cite the page id.")
    recorded = json.loads(TAPE.read_text())["key"]
    with pytest.raises(RecordingError, match="prompt_sha changed"):
        Cassette(TAPE, {**recorded, **hashed_key(**inputs())}, live=False)


def test_recording_with_a_field_the_code_no_longer_sends_is_stale():
    recorded = json.loads(TAPE.read_text())["key"]
    with pytest.raises(RecordingError, match="docs_sha changed"):
        Cassette(TAPE, {k: v for k, v in recorded.items() if k != "docs_sha"}, live=False)


@pytest.mark.parametrize("name", ["gemini-3.8-flash", "gemma4-e2b"])
def test_replay_reproduces_the_saved_report(name):
    tape = HERE / "recordings" / f"{name}.json"
    cassette = Cassette(tape, json.loads(tape.read_text())["key"], live=False)
    report = evaluate(json.loads((HERE / "cases.json").read_text()), cassette, judge=lambda: JUDGE)
    saved = json.loads((HERE / "runs" / f"{name}.json").read_text())
    assert report == saved

A case retrieves the right page, the answer is grounded in it, and the report’s first failure is answer_relevancy. Decide where to look first before you answer.

A case retrieves the right page, the answer is grounded in it, and the report's first failure is answer_relevancy. Where do you look first?

Choose one answer, then check your reasoning.

Read the explanation

The report names the earliest configured check that failed. A later first failure means earlier checks passed, not that every possible defect in those stages was ruled out.

Write robustness tests for an LLM app checks whether the answers survive typos, pasted logs and prompt injection.

References

Written by Arjun Prabhulal.

Course

LLM Evals