diff --git a/agent-evaluation/README.md b/agent-evaluation/README.md new file mode 100644 index 0000000000..fa91eba44b --- /dev/null +++ b/agent-evaluation/README.md @@ -0,0 +1,40 @@ +# AI Agent Evaluation: Build an Eval Harness in Python + +This folder contains the sample code for the Real Python tutorial +[AI Agent Evaluation: Build an Eval Harness in Python](https://realpython.com/agent-evaluation/). + +It holds a small helpdesk agent for internal Python packages (`agent.py`, +`tools.py`) and an eval harness (`agent_evals.py`, `models.py`) that scores +each run's final answer, trajectory, and tool selection. + +- The top-level folder holds the finished project. +- The `follow_along/` folder holds the starter project. The functions you + write in the tutorial are stubs there. + +## Setup + +```console +$ python -m venv .venv +$ source .venv/bin/activate +(.venv) $ python -m pip install -r requirements.txt +(.venv) $ export OPENAI_API_KEY="sk-..." +``` + +The agent uses `gpt-5.4-mini-2026-03-17`. Set the `AGENT_MODEL` environment +variable to try another model. + +## Usage + +```console +(.venv) $ python agent_evals.py run --prompt prompts/helpdesk_v1.txt +(.venv) $ python agent_evals.py report runs/helpdesk_v1.jsonl +(.venv) $ python agent_evals.py compare \ + runs/helpdesk_v1.jsonl runs/helpdesk_v2_docs_first.jsonl +``` + +The optional DeepEval example needs its own requirements file: + +```console +(.venv) $ python -m pip install -r requirements-deepeval.txt +(.venv) $ python deepeval_tools.py +``` diff --git a/agent-evaluation/agent.py b/agent-evaluation/agent.py new file mode 100644 index 0000000000..46efb4f0ad --- /dev/null +++ b/agent-evaluation/agent.py @@ -0,0 +1,48 @@ +"""A small helpdesk agent for internal Python packages.""" + +import json +import os +import sys +from pathlib import Path + +from openai import OpenAI + +from tools import TOOLS, call_tool + +MODEL = os.environ.get("AGENT_MODEL", "gpt-5.4-mini-2026-03-17") +MAX_TURNS = 6 +client = OpenAI() + + +def run_agent(question: str, instructions: str) -> list[dict]: + """Run the agent loop and return the full message history.""" + messages = [ + {"role": "system", "content": instructions}, + {"role": "user", "content": question}, + ] + for _ in range(MAX_TURNS): + response = client.chat.completions.create( + model=MODEL, + messages=messages, + tools=TOOLS, + ) + message = response.choices[0].message + messages.append(message.to_dict()) + if not message.tool_calls: + break + for call in message.tool_calls: + arguments = json.loads(call.function.arguments) + messages.append( + { + "role": "tool", + "tool_call_id": call.id, + "content": call_tool(call.function.name, arguments), + } + ) + return messages + + +if __name__ == "__main__": + prompt = Path("prompts/helpdesk_v1.txt").read_text() + history = run_agent(sys.argv[1], prompt) + print(history[-1]["content"]) diff --git a/agent-evaluation/agent_evals.py b/agent-evaluation/agent_evals.py new file mode 100644 index 0000000000..3fbb7fbc81 --- /dev/null +++ b/agent-evaluation/agent_evals.py @@ -0,0 +1,318 @@ +"""An eval harness for a tool-calling agent.""" + +import argparse +import json +import re +import sys +from collections import Counter +from pathlib import Path + +from models import ( + AgentCase, + AgentRun, + AgentTrace, + MetricResult, + ToolCall, + Trial, + Verdict, +) + + +def load_cases(path: Path = Path("cases.jsonl")) -> list[AgentCase]: + with path.open() as file: + return [AgentCase.model_validate_json(line) for line in file] + + +def load_runs(path: Path) -> list[AgentRun]: + with path.open() as file: + return [AgentRun.model_validate_json(line) for line in file] + + +def code_result( + criterion_id: str, problems: list[str], score: float | None = None +) -> MetricResult: + return MetricResult( + criterion_id=criterion_id, + result_type="categorical", + verdict="FAIL" if problems else "PASS", + score=score, + source="code", + reason="; ".join(problems) if problems else "all checks passed", + ) + + +def capture_trace(question: str, messages: list[dict]) -> AgentTrace: + """Turn an agent's message history into a structured trace.""" + results = { + message["tool_call_id"]: message["content"] + for message in messages + if message["role"] == "tool" + } + steps = [ + ToolCall( + name=call["function"]["name"], + arguments=json.loads(call["function"]["arguments"]), + result=results.get(call["id"], ""), + ) + for message in messages + if message["role"] == "assistant" + for call in message.get("tool_calls") or [] + ] + last = messages[-1] + finished = last["role"] == "assistant" and not last.get("tool_calls") + return AgentTrace( + question=question, + steps=steps, + answer=(last.get("content") or "") if finished else "", + turns=sum(message["role"] == "assistant" for message in messages), + ) + + +def run_suite(prompt: Path, cases: list[AgentCase], trials: int) -> Path: + """Run every case through the agent and save the traces.""" + from agent import run_agent + + instructions = prompt.read_text() + output = Path("runs") / f"{prompt.stem}.jsonl" + output.parent.mkdir(exist_ok=True) + with output.open("w") as file: + for case in cases: + for trial in range(1, trials + 1): + messages = run_agent(case.user_message, instructions) + run = AgentRun( + case_id=case.id, + prompt_id=prompt.stem, + trial=trial, + trace=capture_trace(case.user_message, messages), + ) + file.write(run.model_dump_json() + "\n") + return output + + +def grade_answer(case: AgentCase, trace: AgentTrace) -> MetricResult: + """Check the final answer against required and forbidden patterns.""" + if not trace.answer.strip(): + return code_result("final_answer", ["the agent gave no answer"]) + expected = case.expected + problems = [ + f"missing {pattern!r}" + for pattern in expected.required_patterns + if not re.search(pattern, trace.answer, re.IGNORECASE) + ] + problems += [ + f"contains {pattern!r}" + for pattern in expected.forbidden_patterns + if re.search(pattern, trace.answer, re.IGNORECASE) + ] + return code_result("final_answer", problems) + + +def follows(called: list[str], expected: list[str], match: str) -> bool: + """Compare the called tools with the expected tools.""" + if match == "strict": + return called == expected + if match == "in_order": + remaining = iter(called) + return all(name in remaining for name in expected) + return Counter(expected) <= Counter(called) + + +def grade_trajectory(case: AgentCase, trace: AgentTrace) -> MetricResult: + """Check the order and number of tool calls.""" + called = [step.name for step in trace.steps] + expected = case.expected + problems = [] + if not follows(called, expected.tools, expected.match): + problems.append(f"expected {expected.tools} ({expected.match})") + if len(called) > expected.max_tool_calls: + problems.append( + f"{len(called)} > {expected.max_tool_calls} tool calls" + ) + if problems: + problems.insert(0, f"called {called}") + return code_result("trajectory", problems) + + +def grade_tool_selection(case: AgentCase, trace: AgentTrace) -> MetricResult: + """Check that the agent picked the right tools with the right arguments.""" + called = {step.name for step in trace.steps} + expected = set(case.expected.tools) + problems = [] + if extra := called - expected: + problems.append(f"unneeded {sorted(extra)}") + if missing := expected - called: + problems.append(f"missing {sorted(missing)}") + for step in trace.steps: + patterns = case.expected.arguments.get(step.name, {}) + for argument, pattern in patterns.items(): + value = str(step.arguments.get(argument, "")) + if not re.search(pattern, value, re.IGNORECASE): + problems.append(f"{step.name}({argument}={value!r})") + hits = len(called & expected) + precision = hits / len(called) if called else 1.0 + return code_result("tool_selection", problems, score=precision) + + +def tool_stats(path: Path, cases: list[AgentCase]) -> tuple[float, float]: + """Compute tool-selection precision and recall over all runs.""" + cases_by_id = {case.id: case for case in cases} + picked = expected = hits = 0 + for run in load_runs(path): + called_tools = {step.name for step in run.trace.steps} + expected_tools = set(cases_by_id[run.case_id].expected.tools) + picked += len(called_tools) + expected += len(expected_tools) + hits += len(called_tools & expected_tools) + precision = hits / picked if picked else 1.0 + recall = hits / expected if expected else 1.0 + return precision, recall + + +GRADERS = [grade_answer, grade_trajectory, grade_tool_selection] + + +def grade_runs(path: Path, cases: list[AgentCase]) -> list[Trial]: + cases_by_id = {case.id: case for case in cases} + trials = [] + for run in load_runs(path): + case = cases_by_id[run.case_id] + trials.append( + Trial( + case_id=case.id, + prompt_id=run.prompt_id, + critical=case.critical, + response=run.trace.answer, + metrics=[grade(case, run.trace) for grade in GRADERS], + ) + ) + return trials + + +def case_verdicts(trials: list[Trial]) -> dict[tuple[str, str], Verdict]: + """Merge repeated trials into one verdict per case and criterion.""" + seen: dict[tuple[str, str], set[str]] = {} + for trial in trials: + for metric in trial.metrics: + key = (trial.case_id, metric.criterion_id) + seen.setdefault(key, set()).add(metric.verdict) + return { + key: verdicts.pop() if len(verdicts) == 1 else "REVIEW" + for key, verdicts in seen.items() + } + + +def keys_with(trials: list[Trial], verdict: Verdict) -> set[tuple[str, str]]: + """Collect the (case, criterion) pairs with a given verdict.""" + verdicts = case_verdicts(trials) + return {key for key, value in verdicts.items() if value == verdict} + + +def release_decision( + new_critical_failures: set[str], + new_failures: set[str], + candidate_reviews: set[str], +) -> tuple[str, str]: + """Apply the release rules in order of severity.""" + if new_critical_failures: + return "FAIL", "The candidate broke a blocker on a critical case." + if new_failures: + return "FAIL", "The candidate introduced a new blocker failure." + if candidate_reviews: + return "REVIEW", "A blocker could not be settled from the evidence." + return "PASS", "The candidate introduced no new blocker failures." + + +def compare(baseline: Path, candidate: Path, cases: list[AgentCase]) -> str: + """Gate a candidate run against a baseline run.""" + before = grade_runs(baseline, cases) + after = grade_runs(candidate, cases) + critical_ids = {case.id for case in cases if case.critical} + new = keys_with(after, "FAIL") - keys_with(before, "FAIL") + fixed = keys_with(before, "FAIL") - keys_with(after, "FAIL") + new_ids = {case_id for case_id, _ in new} + review_ids = {case_id for case_id, _ in keys_with(after, "REVIEW")} + decision, reason = release_decision( + new_ids & critical_ids, new_ids, review_ids + ) + print(f"Baseline: {before[0].prompt_id}") + print(f"Candidate: {after[0].prompt_id}\n") + print(f"New failures: {len(new)}") + for case_id, criterion_id in sorted(new): + print(f" {case_id}: {criterion_id}") + print(f"Fixed failures: {len(fixed)}") + print(f"\nRelease decision: {decision}\nReason: {reason}") + return decision + + +def report(path: Path, cases: list[AgentCase]) -> None: + trials = grade_runs(path, cases) + verdicts = case_verdicts(trials) + criteria = list(dict.fromkeys(key[1] for key in verdicts)) + case_count = len({key[0] for key in verdicts}) + print(f"Prompt: {trials[0].prompt_id} | cases: {case_count}\n") + print(f"{'Criterion':<16} {'Pass rate':>9} P/F/R") + print("-" * 35) + for criterion in criteria: + values = [v for (_, c), v in verdicts.items() if c == criterion] + passed, failed, review = ( + values.count("PASS"), + values.count("FAIL"), + values.count("REVIEW"), + ) + rate = f"{passed / len(values):.0%}" + print(f"{criterion:<16} {rate:>9} {passed}/{failed}/{review}") + if "tool_selection" in criteria: + precision, recall = tool_stats(path, cases) + print(f"\nTool precision: {precision:.2f} | recall: {recall:.2f}") + printed = set() + for trial in trials: + for metric in trial.metrics: + key = (trial.case_id, metric.criterion_id) + if metric.verdict != "PASS" and key not in printed: + printed.add(key) + print(f"\n{verdicts[key]} {trial.case_id}: {key[1]}") + for line in metric.reason.split("; "): + print(f" {line}") + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + + trace_cmd = commands.add_parser("trace", help="Run once, print trace") + trace_cmd.add_argument("question") + trace_cmd.add_argument("--prompt", type=Path, required=True) + + run_cmd = commands.add_parser("run", help="Run every case") + run_cmd.add_argument("--prompt", type=Path, required=True) + run_cmd.add_argument("--trials", type=int, default=1) + + report_cmd = commands.add_parser("report", help="Grade saved runs") + report_cmd.add_argument("runs", type=Path) + + compare_cmd = commands.add_parser("compare", help="Gate a candidate") + compare_cmd.add_argument("baseline", type=Path) + compare_cmd.add_argument("candidate", type=Path) + + args = parser.parse_args() + cases = load_cases() + if args.command == "trace": + from agent import run_agent + + messages = run_agent(args.question, args.prompt.read_text()) + trace = capture_trace(args.question, messages) + print(trace.model_dump_json(indent=2)) + elif args.command == "run": + output = run_suite(args.prompt, cases, args.trials) + count = len(load_runs(output)) + print(f"Saved {count} runs to {output}") + elif args.command == "report": + report(args.runs, cases) + else: + decision = compare(args.baseline, args.candidate, cases) + return {"PASS": 0, "FAIL": 1, "REVIEW": 2}[decision] + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/agent-evaluation/cases.jsonl b/agent-evaluation/cases.jsonl new file mode 100644 index 0000000000..22f50af8fb --- /dev/null +++ b/agent-evaluation/cases.jsonl @@ -0,0 +1,5 @@ +{"schema_version":"1.0","id":"billing-version","user_message":"What's the latest billing-client and its minimum Python?","expected":{"required_patterns":["\\b3\\.2\\.0\\b","\\b3\\.11\\b"],"forbidden_patterns":["\\b2\\.4\\b","\\b3\\.9\\b"],"tools":["lookup_package"],"match":"strict","max_tool_calls":1,"arguments":{"lookup_package":{"name":"^billing-client$"}}},"tags":["registry"],"critical":true} +{"schema_version":"1.0","id":"auth-install","user_message":"How do I install auth-client from our internal index?","expected":{"required_patterns":["pypi\\.internal\\.example","pip install"],"forbidden_patterns":[],"tools":["search_docs"],"match":"in_order","max_tool_calls":2,"arguments":{}},"tags":["docs"],"critical":false} +{"schema_version":"1.0","id":"metrics-owner-batching","user_message":"Who owns metrics-client, and how do I batch metrics with it?","expected":{"required_patterns":["Platform","flush_every"],"forbidden_patterns":[],"tools":["lookup_package","search_docs"],"match":"unordered","max_tool_calls":3,"arguments":{"lookup_package":{"name":"^metrics-client$"}}},"tags":["registry","docs"],"critical":false} +{"schema_version":"1.0","id":"unknown-package","user_message":"What's the latest version of payments-sdk?","expected":{"required_patterns":["not found|couldn.t find|could not find|no such package|does(n.t| not) exist"],"forbidden_patterns":["\\b\\d+\\.\\d+\\.\\d+\\b"],"tools":["lookup_package"],"match":"strict","max_tool_calls":1,"arguments":{"lookup_package":{"name":"^payments-sdk$"}}},"tags":["registry","negative"],"critical":true} +{"schema_version":"1.0","id":"general-python","user_message":"How do I install a package in editable mode with pip?","expected":{"required_patterns":["pip install -e"],"forbidden_patterns":[],"tools":[],"match":"strict","max_tool_calls":0,"arguments":{}},"tags":["general"],"critical":false} diff --git a/agent-evaluation/deepeval_tools.py b/agent-evaluation/deepeval_tools.py new file mode 100644 index 0000000000..3ccfe6d4f9 --- /dev/null +++ b/agent-evaluation/deepeval_tools.py @@ -0,0 +1,17 @@ +from deepeval.metrics import ToolCorrectnessMetric +from deepeval.test_case import LLMTestCase, ToolCall + +test_case = LLMTestCase( + input="What's the latest billing-client and its minimum Python?", + actual_output="Version 3.2.0 is the latest. It needs Python 3.11+.", + tools_called=[ + ToolCall(name="search_docs"), + ToolCall(name="lookup_package"), + ], + expected_tools=[ToolCall(name="lookup_package")], +) + +for exact in (False, True): + metric = ToolCorrectnessMetric(should_exact_match=exact) + metric.measure(test_case) + print(f"should_exact_match={exact}: score {metric.score}") diff --git a/agent-evaluation/follow_along/agent.py b/agent-evaluation/follow_along/agent.py new file mode 100644 index 0000000000..46efb4f0ad --- /dev/null +++ b/agent-evaluation/follow_along/agent.py @@ -0,0 +1,48 @@ +"""A small helpdesk agent for internal Python packages.""" + +import json +import os +import sys +from pathlib import Path + +from openai import OpenAI + +from tools import TOOLS, call_tool + +MODEL = os.environ.get("AGENT_MODEL", "gpt-5.4-mini-2026-03-17") +MAX_TURNS = 6 +client = OpenAI() + + +def run_agent(question: str, instructions: str) -> list[dict]: + """Run the agent loop and return the full message history.""" + messages = [ + {"role": "system", "content": instructions}, + {"role": "user", "content": question}, + ] + for _ in range(MAX_TURNS): + response = client.chat.completions.create( + model=MODEL, + messages=messages, + tools=TOOLS, + ) + message = response.choices[0].message + messages.append(message.to_dict()) + if not message.tool_calls: + break + for call in message.tool_calls: + arguments = json.loads(call.function.arguments) + messages.append( + { + "role": "tool", + "tool_call_id": call.id, + "content": call_tool(call.function.name, arguments), + } + ) + return messages + + +if __name__ == "__main__": + prompt = Path("prompts/helpdesk_v1.txt").read_text() + history = run_agent(sys.argv[1], prompt) + print(history[-1]["content"]) diff --git a/agent-evaluation/follow_along/agent_evals.py b/agent-evaluation/follow_along/agent_evals.py new file mode 100644 index 0000000000..8f20fba4ee --- /dev/null +++ b/agent-evaluation/follow_along/agent_evals.py @@ -0,0 +1,224 @@ +"""An eval harness for a tool-calling agent.""" + +import argparse +import sys +from pathlib import Path + +from models import ( + AgentCase, + AgentRun, + AgentTrace, + MetricResult, + Trial, + Verdict, +) + + +def load_cases(path: Path = Path("cases.jsonl")) -> list[AgentCase]: + with path.open() as file: + return [AgentCase.model_validate_json(line) for line in file] + + +def load_runs(path: Path) -> list[AgentRun]: + with path.open() as file: + return [AgentRun.model_validate_json(line) for line in file] + + +def code_result( + criterion_id: str, problems: list[str], score: float | None = None +) -> MetricResult: + return MetricResult( + criterion_id=criterion_id, + result_type="categorical", + verdict="FAIL" if problems else "PASS", + score=score, + source="code", + reason="; ".join(problems) if problems else "all checks passed", + ) + + +def capture_trace(question: str, messages: list[dict]) -> AgentTrace: + """Turn an agent's message history into a structured trace.""" + # TODO: Write this function in Step 1. + raise NotImplementedError + + +def run_suite(prompt: Path, cases: list[AgentCase], trials: int) -> Path: + """Run every case through the agent and save the traces.""" + from agent import run_agent + + instructions = prompt.read_text() + output = Path("runs") / f"{prompt.stem}.jsonl" + output.parent.mkdir(exist_ok=True) + with output.open("w") as file: + for case in cases: + for trial in range(1, trials + 1): + messages = run_agent(case.user_message, instructions) + run = AgentRun( + case_id=case.id, + prompt_id=prompt.stem, + trial=trial, + trace=capture_trace(case.user_message, messages), + ) + file.write(run.model_dump_json() + "\n") + return output + + +def grade_answer(case: AgentCase, trace: AgentTrace) -> MetricResult: + """Check the final answer against required and forbidden patterns.""" + # TODO: Write this function in Step 2. + raise NotImplementedError + + +def follows(called: list[str], expected: list[str], match: str) -> bool: + """Compare the called tools with the expected tools.""" + # TODO: Write this function in Step 3. + raise NotImplementedError + + +def grade_trajectory(case: AgentCase, trace: AgentTrace) -> MetricResult: + """Check the order and number of tool calls.""" + # TODO: Write this function in Step 3. + raise NotImplementedError + + +def grade_tool_selection(case: AgentCase, trace: AgentTrace) -> MetricResult: + """Check that the agent picked the right tools with the right arguments.""" + # TODO: Write this function in Step 4. + raise NotImplementedError + + +def tool_stats(path: Path, cases: list[AgentCase]) -> tuple[float, float]: + """Compute tool-selection precision and recall over all runs.""" + # TODO: Write this function in Step 4. + raise NotImplementedError + + +# TODO (Steps 2 to 4): Add your graders to this list. +GRADERS = [] + + +def grade_runs(path: Path, cases: list[AgentCase]) -> list[Trial]: + cases_by_id = {case.id: case for case in cases} + trials = [] + for run in load_runs(path): + case = cases_by_id[run.case_id] + trials.append( + Trial( + case_id=case.id, + prompt_id=run.prompt_id, + critical=case.critical, + response=run.trace.answer, + metrics=[grade(case, run.trace) for grade in GRADERS], + ) + ) + return trials + + +def case_verdicts(trials: list[Trial]) -> dict[tuple[str, str], Verdict]: + """Merge repeated trials into one verdict per case and criterion.""" + seen: dict[tuple[str, str], set[str]] = {} + for trial in trials: + for metric in trial.metrics: + key = (trial.case_id, metric.criterion_id) + seen.setdefault(key, set()).add(metric.verdict) + return { + key: verdicts.pop() if len(verdicts) == 1 else "REVIEW" + for key, verdicts in seen.items() + } + + +def keys_with(trials: list[Trial], verdict: Verdict) -> set[tuple[str, str]]: + """Collect the (case, criterion) pairs with a given verdict.""" + verdicts = case_verdicts(trials) + return {key for key, value in verdicts.items() if value == verdict} + + +def release_decision( + new_critical_failures: set[str], + new_failures: set[str], + candidate_reviews: set[str], +) -> tuple[str, str]: + """Apply the release rules in order of severity.""" + # TODO: Write this function in Step 5. + raise NotImplementedError + + +def compare(baseline: Path, candidate: Path, cases: list[AgentCase]) -> str: + """Gate a candidate run against a baseline run.""" + # TODO: Write this function in Step 5. + raise NotImplementedError + + +def report(path: Path, cases: list[AgentCase]) -> None: + trials = grade_runs(path, cases) + verdicts = case_verdicts(trials) + criteria = list(dict.fromkeys(key[1] for key in verdicts)) + case_count = len({key[0] for key in verdicts}) + print(f"Prompt: {trials[0].prompt_id} | cases: {case_count}\n") + print(f"{'Criterion':<16} {'Pass rate':>9} P/F/R") + print("-" * 35) + for criterion in criteria: + values = [v for (_, c), v in verdicts.items() if c == criterion] + passed, failed, review = ( + values.count("PASS"), + values.count("FAIL"), + values.count("REVIEW"), + ) + rate = f"{passed / len(values):.0%}" + print(f"{criterion:<16} {rate:>9} {passed}/{failed}/{review}") + if "tool_selection" in criteria: + precision, recall = tool_stats(path, cases) + print(f"\nTool precision: {precision:.2f} | recall: {recall:.2f}") + printed = set() + for trial in trials: + for metric in trial.metrics: + key = (trial.case_id, metric.criterion_id) + if metric.verdict != "PASS" and key not in printed: + printed.add(key) + print(f"\n{verdicts[key]} {trial.case_id}: {key[1]}") + for line in metric.reason.split("; "): + print(f" {line}") + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + + trace_cmd = commands.add_parser("trace", help="Run once, print trace") + trace_cmd.add_argument("question") + trace_cmd.add_argument("--prompt", type=Path, required=True) + + run_cmd = commands.add_parser("run", help="Run every case") + run_cmd.add_argument("--prompt", type=Path, required=True) + run_cmd.add_argument("--trials", type=int, default=1) + + report_cmd = commands.add_parser("report", help="Grade saved runs") + report_cmd.add_argument("runs", type=Path) + + compare_cmd = commands.add_parser("compare", help="Gate a candidate") + compare_cmd.add_argument("baseline", type=Path) + compare_cmd.add_argument("candidate", type=Path) + + args = parser.parse_args() + cases = load_cases() + if args.command == "trace": + from agent import run_agent + + messages = run_agent(args.question, args.prompt.read_text()) + trace = capture_trace(args.question, messages) + print(trace.model_dump_json(indent=2)) + elif args.command == "run": + output = run_suite(args.prompt, cases, args.trials) + count = len(load_runs(output)) + print(f"Saved {count} runs to {output}") + elif args.command == "report": + report(args.runs, cases) + else: + decision = compare(args.baseline, args.candidate, cases) + return {"PASS": 0, "FAIL": 1, "REVIEW": 2}[decision] + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/agent-evaluation/follow_along/cases.jsonl b/agent-evaluation/follow_along/cases.jsonl new file mode 100644 index 0000000000..22f50af8fb --- /dev/null +++ b/agent-evaluation/follow_along/cases.jsonl @@ -0,0 +1,5 @@ +{"schema_version":"1.0","id":"billing-version","user_message":"What's the latest billing-client and its minimum Python?","expected":{"required_patterns":["\\b3\\.2\\.0\\b","\\b3\\.11\\b"],"forbidden_patterns":["\\b2\\.4\\b","\\b3\\.9\\b"],"tools":["lookup_package"],"match":"strict","max_tool_calls":1,"arguments":{"lookup_package":{"name":"^billing-client$"}}},"tags":["registry"],"critical":true} +{"schema_version":"1.0","id":"auth-install","user_message":"How do I install auth-client from our internal index?","expected":{"required_patterns":["pypi\\.internal\\.example","pip install"],"forbidden_patterns":[],"tools":["search_docs"],"match":"in_order","max_tool_calls":2,"arguments":{}},"tags":["docs"],"critical":false} +{"schema_version":"1.0","id":"metrics-owner-batching","user_message":"Who owns metrics-client, and how do I batch metrics with it?","expected":{"required_patterns":["Platform","flush_every"],"forbidden_patterns":[],"tools":["lookup_package","search_docs"],"match":"unordered","max_tool_calls":3,"arguments":{"lookup_package":{"name":"^metrics-client$"}}},"tags":["registry","docs"],"critical":false} +{"schema_version":"1.0","id":"unknown-package","user_message":"What's the latest version of payments-sdk?","expected":{"required_patterns":["not found|couldn.t find|could not find|no such package|does(n.t| not) exist"],"forbidden_patterns":["\\b\\d+\\.\\d+\\.\\d+\\b"],"tools":["lookup_package"],"match":"strict","max_tool_calls":1,"arguments":{"lookup_package":{"name":"^payments-sdk$"}}},"tags":["registry","negative"],"critical":true} +{"schema_version":"1.0","id":"general-python","user_message":"How do I install a package in editable mode with pip?","expected":{"required_patterns":["pip install -e"],"forbidden_patterns":[],"tools":[],"match":"strict","max_tool_calls":0,"arguments":{}},"tags":["general"],"critical":false} diff --git a/agent-evaluation/follow_along/models.py b/agent-evaluation/follow_along/models.py new file mode 100644 index 0000000000..01f2e9c761 --- /dev/null +++ b/agent-evaluation/follow_along/models.py @@ -0,0 +1,71 @@ +"""Records for test cases, agent runs, and evaluation results.""" + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict + +Verdict = Literal["PASS", "FAIL", "REVIEW"] +ResultType = Literal["categorical", "ordinal"] + + +class StrictModel(BaseModel): + """Reject unknown fields in checked-in evaluation artifacts.""" + + model_config = ConfigDict(extra="forbid") + + +class ExpectedBehavior(StrictModel): + required_patterns: list[str] + forbidden_patterns: list[str] + tools: list[str] + match: Literal["strict", "in_order", "unordered"] + max_tool_calls: int + arguments: dict[str, dict[str, str]] + + +class AgentCase(StrictModel): + schema_version: str + id: str + user_message: str + expected: ExpectedBehavior + tags: list[str] + critical: bool + + +class ToolCall(StrictModel): + name: str + arguments: dict[str, Any] + result: str + + +class AgentTrace(StrictModel): + question: str + steps: list[ToolCall] + answer: str + turns: int + + +class AgentRun(StrictModel): + case_id: str + prompt_id: str + trial: int + trace: AgentTrace + + +class MetricResult(StrictModel): + criterion_id: str + result_type: ResultType + verdict: Verdict + score: float | None = None + source: Literal["code", "reference", "llm_judge"] + reason: str + response_quote: str = "" + policy_quote: str = "" + + +class Trial(StrictModel): + case_id: str + prompt_id: str + critical: bool + response: str + metrics: list[MetricResult] diff --git a/agent-evaluation/follow_along/prompts/helpdesk_v1.txt b/agent-evaluation/follow_along/prompts/helpdesk_v1.txt new file mode 100644 index 0000000000..8ffdb76f6c --- /dev/null +++ b/agent-evaluation/follow_along/prompts/helpdesk_v1.txt @@ -0,0 +1,6 @@ +You answer developer questions about our internal Python packages. + +- Use lookup_package for versions, supported Python versions, and owners. +- Use search_docs for how-to questions. +- Answer general Python questions directly, without tools. +- Keep answers short and specific. diff --git a/agent-evaluation/follow_along/prompts/helpdesk_v2_docs_first.txt b/agent-evaluation/follow_along/prompts/helpdesk_v2_docs_first.txt new file mode 100644 index 0000000000..2ca319e605 --- /dev/null +++ b/agent-evaluation/follow_along/prompts/helpdesk_v2_docs_first.txt @@ -0,0 +1,6 @@ +You answer developer questions about our internal Python packages. + +- Always search the docs with search_docs before you answer, + so that every answer is grounded in our documentation. +- Use lookup_package for versions, supported Python versions, and owners. +- Keep answers short and specific. diff --git a/agent-evaluation/follow_along/prompts/helpdesk_v3_scoped.txt b/agent-evaluation/follow_along/prompts/helpdesk_v3_scoped.txt new file mode 100644 index 0000000000..e6532d6423 --- /dev/null +++ b/agent-evaluation/follow_along/prompts/helpdesk_v3_scoped.txt @@ -0,0 +1,7 @@ +You answer developer questions about our internal Python packages. + +- Use lookup_package for versions, supported Python versions, and owners. + The registry is the source of truth for these facts. +- Use search_docs for how-to questions, and name the page you used. +- Answer general Python questions directly, without tools. +- Keep answers short and specific. diff --git a/agent-evaluation/follow_along/requirements-deepeval.txt b/agent-evaluation/follow_along/requirements-deepeval.txt new file mode 100644 index 0000000000..8fe931eede --- /dev/null +++ b/agent-evaluation/follow_along/requirements-deepeval.txt @@ -0,0 +1 @@ +deepeval==4.3.2 diff --git a/agent-evaluation/follow_along/requirements.txt b/agent-evaluation/follow_along/requirements.txt new file mode 100644 index 0000000000..81c63d3df2 --- /dev/null +++ b/agent-evaluation/follow_along/requirements.txt @@ -0,0 +1,2 @@ +openai==3.40.0 +pydantic==2.14.1 diff --git a/agent-evaluation/follow_along/tools.py b/agent-evaluation/follow_along/tools.py new file mode 100644 index 0000000000..e3f39eac9e --- /dev/null +++ b/agent-evaluation/follow_along/tools.py @@ -0,0 +1,99 @@ +"""Tools for the helpdesk agent: a package registry and a docs search.""" + +import json + +REGISTRY = { + "billing-client": { + "latest": "3.2.0", + "python": ">=3.11", + "owner": "Payments", + }, + "auth-client": { + "latest": "1.8.1", + "python": ">=3.10", + "owner": "Identity", + }, + "metrics-client": { + "latest": "0.9.4", + "python": ">=3.10", + "owner": "Platform", + }, +} + +DOCS = { + "billing-client-overview": ( + "billing-client: current release is 2.4. Supports Python 3.9 " + "and later. (This page is out of date.)" + ), + "auth-client-install": ( + "Install auth-client from the team index with: " + "pip install --index-url https://pypi.internal.example/simple " + "auth-client" + ), + "metrics-client-batching": ( + "metrics-client batching: call Batcher(flush_every=50) to send " + "metrics in groups of 50 instead of one request per metric." + ), +} + +TOOLS = [ + { + "type": "function", + "function": { + "name": "lookup_package", + "description": ( + "Look up a package in the team registry. Returns its " + "latest version, supported Python versions, and owner." + ), + "parameters": { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "search_docs", + "description": ( + "Keyword search over the internal documentation pages." + ), + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + }, + }, +] + + +def lookup_package(name: str) -> str: + info = REGISTRY.get(name.lower()) + if info is None: + return f"{name}: not found in the registry" + return ( + f"{name}: latest {info['latest']}, Python {info['python']}, " + f"owner {info['owner']}" + ) + + +def search_docs(query: str) -> str: + words = query.lower().split() + hits = [ + f"[{page}] {text}" + for page, text in DOCS.items() + if any(word in f"{page} {text}".lower() for word in words) + ] + return "\n".join(hits) or "no matching pages" + + +def call_tool(name: str, arguments: dict) -> str: + functions = { + "lookup_package": lookup_package, + "search_docs": search_docs, + } + if name not in functions: + return json.dumps({"error": f"unknown tool {name}"}) + return functions[name](**arguments) diff --git a/agent-evaluation/models.py b/agent-evaluation/models.py new file mode 100644 index 0000000000..01f2e9c761 --- /dev/null +++ b/agent-evaluation/models.py @@ -0,0 +1,71 @@ +"""Records for test cases, agent runs, and evaluation results.""" + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict + +Verdict = Literal["PASS", "FAIL", "REVIEW"] +ResultType = Literal["categorical", "ordinal"] + + +class StrictModel(BaseModel): + """Reject unknown fields in checked-in evaluation artifacts.""" + + model_config = ConfigDict(extra="forbid") + + +class ExpectedBehavior(StrictModel): + required_patterns: list[str] + forbidden_patterns: list[str] + tools: list[str] + match: Literal["strict", "in_order", "unordered"] + max_tool_calls: int + arguments: dict[str, dict[str, str]] + + +class AgentCase(StrictModel): + schema_version: str + id: str + user_message: str + expected: ExpectedBehavior + tags: list[str] + critical: bool + + +class ToolCall(StrictModel): + name: str + arguments: dict[str, Any] + result: str + + +class AgentTrace(StrictModel): + question: str + steps: list[ToolCall] + answer: str + turns: int + + +class AgentRun(StrictModel): + case_id: str + prompt_id: str + trial: int + trace: AgentTrace + + +class MetricResult(StrictModel): + criterion_id: str + result_type: ResultType + verdict: Verdict + score: float | None = None + source: Literal["code", "reference", "llm_judge"] + reason: str + response_quote: str = "" + policy_quote: str = "" + + +class Trial(StrictModel): + case_id: str + prompt_id: str + critical: bool + response: str + metrics: list[MetricResult] diff --git a/agent-evaluation/prompts/helpdesk_v1.txt b/agent-evaluation/prompts/helpdesk_v1.txt new file mode 100644 index 0000000000..8ffdb76f6c --- /dev/null +++ b/agent-evaluation/prompts/helpdesk_v1.txt @@ -0,0 +1,6 @@ +You answer developer questions about our internal Python packages. + +- Use lookup_package for versions, supported Python versions, and owners. +- Use search_docs for how-to questions. +- Answer general Python questions directly, without tools. +- Keep answers short and specific. diff --git a/agent-evaluation/prompts/helpdesk_v2_docs_first.txt b/agent-evaluation/prompts/helpdesk_v2_docs_first.txt new file mode 100644 index 0000000000..2ca319e605 --- /dev/null +++ b/agent-evaluation/prompts/helpdesk_v2_docs_first.txt @@ -0,0 +1,6 @@ +You answer developer questions about our internal Python packages. + +- Always search the docs with search_docs before you answer, + so that every answer is grounded in our documentation. +- Use lookup_package for versions, supported Python versions, and owners. +- Keep answers short and specific. diff --git a/agent-evaluation/prompts/helpdesk_v3_scoped.txt b/agent-evaluation/prompts/helpdesk_v3_scoped.txt new file mode 100644 index 0000000000..e6532d6423 --- /dev/null +++ b/agent-evaluation/prompts/helpdesk_v3_scoped.txt @@ -0,0 +1,7 @@ +You answer developer questions about our internal Python packages. + +- Use lookup_package for versions, supported Python versions, and owners. + The registry is the source of truth for these facts. +- Use search_docs for how-to questions, and name the page you used. +- Answer general Python questions directly, without tools. +- Keep answers short and specific. diff --git a/agent-evaluation/requirements-deepeval.txt b/agent-evaluation/requirements-deepeval.txt new file mode 100644 index 0000000000..8fe931eede --- /dev/null +++ b/agent-evaluation/requirements-deepeval.txt @@ -0,0 +1 @@ +deepeval==4.3.2 diff --git a/agent-evaluation/requirements.txt b/agent-evaluation/requirements.txt new file mode 100644 index 0000000000..81c63d3df2 --- /dev/null +++ b/agent-evaluation/requirements.txt @@ -0,0 +1,2 @@ +openai==3.40.0 +pydantic==2.14.1 diff --git a/agent-evaluation/tools.py b/agent-evaluation/tools.py new file mode 100644 index 0000000000..e3f39eac9e --- /dev/null +++ b/agent-evaluation/tools.py @@ -0,0 +1,99 @@ +"""Tools for the helpdesk agent: a package registry and a docs search.""" + +import json + +REGISTRY = { + "billing-client": { + "latest": "3.2.0", + "python": ">=3.11", + "owner": "Payments", + }, + "auth-client": { + "latest": "1.8.1", + "python": ">=3.10", + "owner": "Identity", + }, + "metrics-client": { + "latest": "0.9.4", + "python": ">=3.10", + "owner": "Platform", + }, +} + +DOCS = { + "billing-client-overview": ( + "billing-client: current release is 2.4. Supports Python 3.9 " + "and later. (This page is out of date.)" + ), + "auth-client-install": ( + "Install auth-client from the team index with: " + "pip install --index-url https://pypi.internal.example/simple " + "auth-client" + ), + "metrics-client-batching": ( + "metrics-client batching: call Batcher(flush_every=50) to send " + "metrics in groups of 50 instead of one request per metric." + ), +} + +TOOLS = [ + { + "type": "function", + "function": { + "name": "lookup_package", + "description": ( + "Look up a package in the team registry. Returns its " + "latest version, supported Python versions, and owner." + ), + "parameters": { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "search_docs", + "description": ( + "Keyword search over the internal documentation pages." + ), + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + }, + }, +] + + +def lookup_package(name: str) -> str: + info = REGISTRY.get(name.lower()) + if info is None: + return f"{name}: not found in the registry" + return ( + f"{name}: latest {info['latest']}, Python {info['python']}, " + f"owner {info['owner']}" + ) + + +def search_docs(query: str) -> str: + words = query.lower().split() + hits = [ + f"[{page}] {text}" + for page, text in DOCS.items() + if any(word in f"{page} {text}".lower() for word in words) + ] + return "\n".join(hits) or "no matching pages" + + +def call_tool(name: str, arguments: dict) -> str: + functions = { + "lookup_package": lookup_package, + "search_docs": search_docs, + } + if name not in functions: + return json.dumps({"error": f"unknown tool {name}"}) + return functions[name](**arguments)