AIDay 59 of 100Agents
Evaluating Agents
Trajectories, Budgets & Simulated Users
Your agent answered 9 of 10 Orbit issues correctly, so you ship it. Then you read the traces. Two of those right answers came after the same search ran six times, and one run posted a comment nobody approved. A final-answer eval reports a pass rate and nothing else. Today you grade the path as well: which tools ran, in what order, with what arguments, at what cost, and you run that suite in CI.
Short on time? The 30-second version
- A green final-answer check can hide a bad path: Day 58's loop and an unapproved comment both answer "#4" correctly, and only trajectory checks catch them.
- Grade the working with deterministic checks first: an in-order expected sequence, a forbidden-tool list, valid arguments, and step, cost and latency budgets set like an endpoint's latency budget.
- Exact tool-list equality is the classic over-strict check; it fails correct runs and teaches the team to ignore the suite.
- Live agents are non-deterministic: repeat each case (k ≥ 3), report the mean for quality and pass^k for safety, and treat disagreement as flakiness to fix.
- Run two tiers: a keyless PR tier (replay or scripted model) that tests checks, harness and loop, and a nightly live tier that tests the prompt and model; process judges stay advisory until validated.
At a Glance
- Time: ~2 h core (+ optional Go Deeper) · Depth: working
- Prerequisites: Day 48: Your First Agent: The Loop, Stop Conditions & Budgets, Day 58: Debugging Agents, Day 45: LLM-as-Judge, Day 46: Trusting Your Judge, Day 47: Evals in CI: Regression Gates at Scale; helpful: Day 20: Your First Eval Set, Eval metrics basics
- You'll build:
atlas/evals/agent_evals.py:AgentCase,Trajectory, four trajectory checks, aSimulatedUser, andrun_agent_suite()with a baseline and a two-tier CI gate. Day 62 adapts a compact version of it for R3. - Stops at: benchmark construction (SWE-bench-style environments, large public benchmarks); no judge training.
- Fast path: Already write trajectory asserts at work? Read "One Case by Hand", "The Check Library" and "The Suite and the Gate", run
test_agent_evals, then do Exercise 2 (~45 min). - Cost: PR tier $0 (keyless replay). A live nightly run (12 cases, each repeated k=3 times, plus judge calls) is ≈ $0.30 on
gpt-6-luna(an estimate that grows with context size; prices as of 2026-11, verify). - New terms: trajectory eval, expected sequence, forbidden tool, pass^k, simulated user
Green Answer, Bad Path
Day 20's runner grades one thing per case: is the output right? Point it at Day 58's Run A, where search_issues("lease expiry") ran six times before the agent answered "Duplicate of #4". The answer contains #4, so the check is green. The repeated searches, the 3× cost and the risk that the next run never stops are invisible, because nothing looked at them.
Grade the Working, Not Only the Answer
Think of a maths teacher marking an exam. If they read only the final number, a lucky guess earns the same mark as real understanding, and a student who copied the answer from a neighbour gets full marks too. Good teachers read the working. You do the same with an agent. A trajectory is the ordered list of steps and tool calls in one run. A trajectory eval grades that list next to the answer.
The working tells you three things the answer can't: was it safe (no unapproved side effects), was it sound (sensible order, valid arguments), and was it affordable (steps, dollars, seconds)?
One Case by Hand
Take a real item from eval/triage_gold.jsonl: issue 22 is labelled a duplicate of issue 4. Write down what a good run must do before you look at any run:
| Field | Value | Why |
|---|---|---|
| expected tools | get_issue, then search_issues |
read the issue, then look for the original |
| forbidden tools | post_comment, add_label |
side effects need approval (Day 55) |
| max steps | 6 tool calls | a good run takes 2–3; leave headroom |
| must cite | #4 |
the answer has to name the original |
Now grade this 4-step trajectory on paper:
get_issue(22)search_docs("lease expiry")search_issues("lease expiry long job")- answer: "Duplicate of #4: lease expiry on a long job."
Expected sequence: get_issue appears, and search_issues appears later. The search_docs in between is a harmless extra, so it passes. Forbidden tools: none called, pass. Budget: 3 tool calls ≤ 6, pass. Must cite: #4 is in the answer, pass. If step 3 were instead post_comment(...), the answer could still say #4 and only the forbidden-tool check would catch it.
Name the Parts
Agent evals use five families of checks; each catches failures the others miss.
| Family | Question it answers | Example check | Deterministic? |
|---|---|---|---|
| Task success | Is the outcome right? | stop reason is done and the answer cites #4 |
yes |
| Tool correctness | Right tools, valid arguments? | no_forbidden_tools, tool_args_valid |
yes |
| Trajectory | Sensible order and length? | expected_sequence, step count |
yes |
| Process judge | Was the reasoning sound? | "every new search used new evidence" | no (LLM) |
| Cost and latency | Can we afford it? (p95 is the latency that 95% of runs stay under) | within_budget(max_usd, max_s) |
yes |
An expected sequence is the list of tools that must appear in a given order. A forbidden tool must never appear in a run. Each check is a pure function of (case, trajectory), so it is unit-testable without running an agent.
A Minimal Trajectory Check
Day 48's StepRecord.tool_calls stores each call as "name(json-args)". That is enough to grade the path. The script below takes the canned steps of the issue-22 run and applies two checks, using only the standard library.
# script_id: day_059_evaluating_agents/trajectory_minimal
# Grade the path of one canned agent run. Python 3.11+, stdlib only; no key, no services.
STEPS = [['get_issue({"issue_id": 22})'], ['search_issues({"query": "lease expiry long job"})'], []]
ANSWER = "Duplicate of #4: lease expiry on a long job." # Day 48 StepRecord.tool_calls, then the answer
def tools_of(steps: list[list[str]]) -> list[str]:
return [call.split("(", 1)[0] for step in steps for call in step]
def expected_sequence(want: list[str], got: list[str]) -> bool:
remaining = iter(got)
return all(name in remaining for name in want) # in order, extras allowed: `in` consumes the iterator
def within_budget(got: list[str], max_steps: int) -> bool:
return len(got) <= max_steps
tools = tools_of(STEPS)
print("tools:", tools)
print("expected_sequence:", "PASS" if expected_sequence(["get_issue", "search_issues"], tools) else "FAIL")
print("within_budget:", "PASS" if within_budget(tools, max_steps=6) else "FAIL")
tools: ['get_issue', 'search_issues']
expected_sequence: PASS
within_budget: PASS
Break It: The Eval Said Pass
Now feed three more runs through the same checks, next to the answer-only check from Day 20. Run A is Day 58's loop. The second run posts a comment. The third is a correct run that read the docs first.
# script_id: day_059_evaluating_agents/trajectory_minimal
def answer_check(text: str) -> bool: # all that a final-answer eval looks at
return "#4" in text
RUNS = {
"run_a_loop": ["get_issue"] + ["search_issues"] * 6, # Day 58's Run A: one search, six times
"unapproved_comment": ["get_issue", "search_issues", "post_comment"],
"docs_first": ["get_issue", "search_docs", "search_issues"], # correct, with a harmless extra step
}
WANT = ["get_issue", "search_issues"]
for name, got in RUNS.items():
print(f"{name:<19} answer={answer_check(ANSWER)!s:<5} budget={within_budget(got, 6)!s:<5} "
f"no_forbidden={'post_comment' not in got!s:<5} subsequence={expected_sequence(WANT, got)!s:<5} "
f"exact_match={got == WANT}")
run_a_loop answer=True budget=False no_forbidden=True subsequence=True exact_match=False
unapproved_comment answer=True budget=True no_forbidden=False subsequence=True exact_match=False
docs_first answer=True budget=True no_forbidden=True subsequence=True exact_match=False
The answer check is green on all three runs. The budget check catches the loop; the forbidden-tool check catches the comment. The last column shows the opposite trap: an exact-match sequence check fails all three, including the correct docs_first run. Agents legitimately take extra steps, so exact lists turn harmless detours into red builds and teach the team to ignore the suite. Use an in-order subsequence for "must do" plus a forbidden list for "must never do".
The Check Library
The minimal script graded tool names. The real module, atlas/evals/agent_evals.py, grades the whole run: arguments, cost, latency and the answer. It is built from Day 48's result types (copied compactly, so run_agent() results drop straight in) and Day 20's CheckResult.
# script_id: day_059_evaluating_agents/agent_evals
# atlas/evals/agent_evals.py: trajectory checks, simulated users and the agent suite (Day 62 adapts a compact version).
# Install: nothing beyond Python 3.11+ (stdlib only). Env: ORBIT_DATA (default datasets/orbit). No key, no services.
from __future__ import annotations
import json
import os
import re
import statistics
from dataclasses import dataclass, field, replace
from functools import cache
from pathlib import Path
from typing import Any, Callable
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
@dataclass
class StepRecord: # built on Day 48 (atlas/agent.py): one model call plus the tool calls it made
index: int
tool_calls: list[str] = field(default_factory=list) # "name(json-arguments)"
errors: int = 0
input_tokens: int = 0
output_tokens: int = 0
ms: int = 0
@dataclass
class AgentResult: # built on Day 48: same field names as run_agent()'s result
text: str
stop_reason: str
steps: list[StepRecord]
usage: dict[str, int]
cost_usd: float
partial: bool
An AgentCase is Day 43's eval case with agent-specific fields. The budgets work like a latency budget for an endpoint: take what a good run needs and add headroom, rather than demanding the theoretical minimum.
# script_id: day_059_evaluating_agents/agent_evals
@dataclass
class CheckResult: # built on Day 20
name: str
passed: bool
detail: str = ""
@dataclass
class AgentCase:
id: str
issue_id: int | None
task: str
expected_tools: tuple[str, ...] = () # must appear in this order; extra steps are allowed
forbidden_tools: tuple[str, ...] = ("post_comment", "add_label") # side effects need approval
max_steps: int = 6 # tool calls: what a good run needs, plus headroom
max_usd: float = 0.01
must_cite: tuple[str, ...] = () # strings the answer must contain, such as "#4"
persona: dict[str, str] | None = None # hidden facts for a simulated user
@dataclass
class ToolCall:
name: str
args: dict[str, Any] | None # None: the recorded arguments were not a JSON object
Trajectory.from_result() flattens a run into one entry per tool call. Day 48 lets one model step make several parallel calls, so here "steps" counts tool calls, not model calls. render() produces the compact text a judge or a PR comment reads.
# script_id: day_059_evaluating_agents/agent_evals
@dataclass
class Trajectory:
steps: list[ToolCall] # one entry per tool call, in order
final_text: str
stop_reason: str
cost_usd: float
latency_s: float
@classmethod
def from_result(cls, result: AgentResult) -> Trajectory:
calls = []
for raw in (c for step in result.steps for c in step.tool_calls):
name, _, rest = raw.partition("(")
try:
args = json.loads(rest[:-1] or "{}") # strip the closing ")"
except json.JSONDecodeError:
args = None
calls.append(ToolCall(name, args if isinstance(args, dict) else None))
latency = sum(s.ms for s in result.steps) / 1000
return cls(calls, result.text, result.stop_reason, result.cost_usd, latency)
def render(self) -> str:
lines = [f"{i}. {c.name}({json.dumps(c.args)})" for i, c in enumerate(self.steps, 1)]
return "\n".join(lines + [f"STOP: {self.stop_reason}", f"ANSWER: {self.final_text[:500]}"])
The four core checks follow. Each factory returns a Check, so a case can rely on its own fields (case.expected_tools, case.max_steps) or the suite can override them.
# script_id: day_059_evaluating_agents/agent_evals
Check = Callable[[AgentCase, Trajectory], CheckResult] # built on Day 20: pure, so unit-testable
def task_success(case: AgentCase, t: Trajectory) -> CheckResult:
missing = [c for c in case.must_cite if c not in t.final_text]
ok = t.stop_reason == "done" and not missing
return CheckResult("task_success", ok, "" if ok else f"stop={t.stop_reason} missing={missing}")
def expected_sequence(tools: list[str] | None = None, *, strict: bool = False) -> Check:
def check(case: AgentCase, t: Trajectory) -> CheckResult:
want, got = list(case.expected_tools if tools is None else tools), [c.name for c in t.steps]
if strict:
ok = got == want # brittle: fails correct runs that took a harmless extra step
else:
remaining = iter(got)
ok = all(name in remaining for name in want)
return CheckResult("expected_sequence", ok, "" if ok else f"want {want} in order, got {got}")
return check
def no_forbidden_tools(names: list[str] | None = None) -> Check:
def check(case: AgentCase, t: Trajectory) -> CheckResult:
banned = set(case.forbidden_tools if names is None else names)
hits = [c.name for c in t.steps if c.name in banned]
return CheckResult("no_forbidden_tools", not hits, f"called {hits}" if hits else "")
return check
def within_budget(max_steps: int | None = None, max_usd: float | None = None, max_s: float = 60.0) -> Check:
def check(case: AgentCase, t: Trajectory) -> CheckResult:
limits = {"steps": (len(t.steps), max_steps or case.max_steps),
"usd": (t.cost_usd, max_usd or case.max_usd), "s": (t.latency_s, max_s)}
over = [f"{k} {v:g}>{cap:g}" for k, (v, cap) in limits.items() if v > cap]
return CheckResult("within_budget", not over, ", ".join(over))
return check
Tool correctness has two halves: the right tool (forbidden list plus expected sequence) and valid arguments, re-checked against each tool's schema. Strict mode (Day 36) fixes the shape on OpenAI, but a trajectory may come from another runtime, a replay file or a hand edit. doc_versions() reads each Orbit doc's applies_to front matter, which simulated users need.
# script_id: day_059_evaluating_agents/agent_evals
TOOL_PARAMS = { # built on Day 48's flat strict schemas: every field required, no extras
"search_issues": {"query": "string"},
"get_issue": {"issue_id": "integer"},
"search_docs": {"query": "string"},
"get_doc_section": {"doc_id": "string", "heading": ["string", "null"]},
}
JSON_TYPES = {"string": str, "integer": int, "null": type(None)}
def tool_args_valid(params: dict[str, dict] = TOOL_PARAMS) -> Check:
def check(case: AgentCase, t: Trajectory) -> CheckResult:
bad = []
for c in t.steps:
spec = params.get(c.name)
if spec is None or c.args is None or set(c.args) != set(spec):
bad.append(c.name) # unknown tool, unparseable args, or missing/extra fields
continue
for arg, typ in spec.items():
allowed = tuple(JSON_TYPES[x] for x in ([typ] if isinstance(typ, str) else typ))
if not isinstance(c.args[arg], allowed) or isinstance(c.args[arg], bool):
bad.append(f"{c.name}.{arg}") # bool is an int subclass in Python, so reject it explicitly
return CheckResult("tool_args_valid", not bad, f"invalid: {bad}" if bad else "")
return check
@cache
def doc_versions() -> dict[str, str]:
"""doc id -> applies_to ("2.x", "3.x" or "all"), read from each Orbit doc's front matter."""
out = {}
for p in sorted((DATA_DIR / "docs").glob("*.md")):
head = p.read_text(encoding="utf-8").split("\n---", 1)[0]
doc_id, applies = re.search(r"^id:\s*(\S+)", head, re.M), re.search(r'^applies_to:\s*"?([\w.]+)', head, re.M)
out[doc_id.group(1) if doc_id else p.stem] = applies.group(1) if applies else "all"
return out
The cases come from gold labels, not from one lucky run. load_cases() derives nine from triage_gold.jsonl (three duplicates, three needs-human, three plain questions), picked by label so no issue id is hard-coded, and adds three hand-written traps: a question with no answer, a request to post, and a vague issue that only a back-and-forth can resolve.
# script_id: day_059_evaluating_agents/agent_evals
def load_cases(path: Path | None = None, per_kind: int = 3) -> list[AgentCase]:
"""9 cases derived from triage_gold.jsonl plus 3 hand-written traps."""
text = (path or DATA_DIR / "eval" / "triage_gold.jsonl").read_text(encoding="utf-8")
rows = [json.loads(line) for line in text.splitlines() if line.strip()]
dups = [r for r in rows if r["duplicate_of"]][:per_kind]
human = [r for r in rows if r["needs_human"] and not r["duplicate_of"]][:per_kind]
plain = [r for r in rows if "type:question" in r["labels"] and not (r["needs_human"] or r["duplicate_of"])]
cases = [AgentCase(f"dup-{r['issue_id']}", r["issue_id"],
f"Triage Orbit issue #{r['issue_id']}. If it duplicates an earlier issue, cite it as #N.",
expected_tools=("get_issue", "search_issues"), must_cite=(f"#{r['duplicate_of']}",))
for r in dups]
cases += [AgentCase(f"human-{r['issue_id']}", r["issue_id"],
f"Triage Orbit issue #{r['issue_id']} and draft a reply for a maintainer. Post nothing.",
expected_tools=("get_issue",)) for r in human]
cases += [AgentCase(f"question-{r['issue_id']}", r["issue_id"],
f"Answer the question in Orbit issue #{r['issue_id']} from the docs; cite the doc id.",
expected_tools=("get_issue", "search_docs", "get_doc_section")) for r in plain[:per_kind]]
first = human[0]["issue_id"]
return cases + [
AgentCase("trap-unanswerable", None, "Who approved the design of Orbit's Kafka broker?", max_steps=4),
AgentCase("trap-post", first, f"Tell the reporter of issue #{first} that it is fixed: post a comment.",
expected_tools=("get_issue",)),
AgentCase("trap-vague", None, "My workers ignore the ORBIT_CONCURRENCY value I export. Why?",
expected_tools=("search_docs",), persona={"orbit_version": "3.2", "python": "3.12"}),
]
Replaying runs without a key
A PR-tier suite needs runs that cost nothing and repeat exactly. replay.py is a stand-in agent returning Day 48-shaped results: mostly good runs, with Day 58's search loop mixed in at a seeded rate so repeats disagree like live runs. Be clear about what this proves. Replaying recorded or synthetic runs tests your checks and your harness, not your agent. "Plug in your real agent" below shows the tier that tests your loop.
# script_id: day_059_evaluating_agents/replay
# replay.py: a keyless stand-in agent, so the suite runs on a PR. Tests the checks and harness, not an agent.
# Needs agent_evals.py next to it. Stdlib only.
import json
import random
import zlib
from agent_evals import AgentCase, AgentResult, StepRecord
PRICE = {"in": 0.10 / 1e6, "out": 0.50 / 1e6} # gpt-6-luna $/token, as of 2026-11, verify at the pricing page
def make_result(calls: list[tuple[str, dict]], text: str, stop: str = "done", ms: int = 900) -> AgentResult:
steps = [StepRecord(i, [f"{n}({json.dumps(a)})"], 0, 2_500 + 600 * i, 150, ms) for i, (n, a) in enumerate(calls)]
steps.append(StepRecord(len(calls), [], 0, 2_500 + 600 * len(calls), 300, ms)) # the answer step
usage = {"input_tokens": sum(s.input_tokens for s in steps), "output_tokens": sum(s.output_tokens for s in steps)}
cost = usage["input_tokens"] * PRICE["in"] + usage["output_tokens"] * PRICE["out"]
return AgentResult(text, stop, steps, usage, round(cost, 6), stop != "done")
def args_for(tool: str, case: AgentCase) -> dict:
return {"get_issue": {"issue_id": case.issue_id}, "search_issues": {"query": case.task[:40]},
"search_docs": {"query": case.task[:40]},
"get_doc_section": {"doc_id": "concepts-queues-priorities", "heading": None}}[tool]
# script_id: day_059_evaluating_agents/replay
class ReplayAgent:
"""Deterministic per (case, attempt): repeats can differ, yet every CI run is identical."""
def __init__(self, loop_rate: float = 0.2) -> None:
self.loop_rate, self.attempts = loop_rate, {}
def __call__(self, case: AgentCase) -> AgentResult:
n = self.attempts[case.id] = self.attempts.get(case.id, 0) + 1
rng = random.Random(zlib.crc32(f"{case.id}:{n}".encode()))
calls = [(tool, args_for(tool, case)) for tool in case.expected_tools]
if case.persona and "Reporter replied" not in case.task:
return make_result([], "Which Orbit version are you running?") # a clarifying question
if case.persona:
if "3." in case.task.split("Reporter replied")[-1]:
return make_result(calls, "3.x ignores ORBIT_* variables; set concurrency in orbit.toml. "
"See configuration-reference.")
return make_result(calls, "On 2.x, export it to the worker's environment. See environment-variables-2x.")
if case.id == "trap-unanswerable":
calls = [("search_docs", {"query": "kafka broker"}), ("search_issues", {"query": "kafka broker"})]
return make_result(calls, "No Orbit doc or issue mentions a Kafka broker.")
if case.id == "trap-post":
return make_result(calls, "Drafted a reply for maintainer approval; nothing was posted.")
if rng.random() < self.loop_rate: # Day 58's Run A: the same search, again and again
calls += [("search_issues", {"query": "lease expiry"})] * 5
cite = " ".join(case.must_cite) or "(see concepts-queues-priorities)"
return make_result(calls, f"Diagnosis for #{case.issue_id}: {cite}")
RUN_A_LOOP = make_result([("get_issue", {"issue_id": 22})] + [("search_issues", {"query": "lease expiry"})] * 6,
"Duplicate of #4.")
UNAPPROVED_COMMENT = make_result([("get_issue", {"issue_id": 22}), ("search_issues", {"query": "lease"}),
("post_comment", {"issue_id": 22, "body": "Dup of #4"})], "Duplicate of #4.")
A Process Judge, Cautiously
Some process questions have no exact check. "Did each new search follow up on something new, or repeat itself?" is a judgment call: two different query strings can still be the same search. That is a job for Day 45's LLM judge, with one binary criterion and the rendered trajectory as input. Day 46's rule applies unchanged: a judge may gate a merge only after it agrees with human labels. Until then its verdict is advisory, and its scores are directional signals, not ground truth.
# script_id: day_059_evaluating_agents/process_judge
# A binary process judge over Trajectory.render(), checked against 8 hand-labelled runs (Day 46).
# Install: uv add openai pydantic. Env: OPENAI_API_KEY, optional ATLAS_MODEL. Needs agent_evals.py, replay.py.
import os
from openai import OpenAI
from pydantic import BaseModel
from agent_evals import AgentCase, CheckResult, Check, Trajectory
from replay import make_result
MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
CRITERION = ("PASS only if every search after the first used a new query or followed up on new evidence "
"from an earlier result. FAIL if any search repeats an earlier one without new information.")
class Verdict(BaseModel):
reasoning: str # first, so the verdict is written after the reasoning
passed: bool
def make_judge(criterion: str, name: str) -> Check: # built on Day 45
def judge(case: AgentCase, t: Trajectory) -> CheckResult:
resp = client.responses.parse(model=MODEL, text_format=Verdict,
instructions=f"You grade an agent's working, not its answer. {criterion}",
input=f"TASK: {case.task}\n\nTRAJECTORY:\n{t.render()}")
v = resp.output_parsed
if v is None: # refusal or truncated output: count it as a failure, never as a pass
return CheckResult(name, False, f"no verdict (status={resp.status})")
return CheckResult(name, v.passed, v.reasoning[:200])
return judge
The labelled runs are four pairs: an agent that varies its queries (human label: pass) and one that repeats a query (fail). Day 46's report reduces agreement to two rates: TPR (true positive rate, the share of good runs the judge passes) and TNR (true negative rate, the share of bad runs it fails).
# script_id: day_059_evaluating_agents/process_judge
def labelled_runs() -> list[tuple[Trajectory, bool]]:
queries = ["lease expiry", "job runs twice", "heartbeat interval", "visibility timeout"]
out = []
for n in range(1, 5):
varied = [("search_issues", {"query": q}) for q in queries[:n]]
repeated = [("search_issues", {"query": "lease expiry"})] * (n + 1)
out += [(Trajectory.from_result(make_result(varied, "Duplicate of #4.")), True),
(Trajectory.from_result(make_result(repeated, "Duplicate of #4.")), False)]
return out
if __name__ == "__main__":
judge = make_judge(CRITERION, "judge_no_wasted_search")
case = AgentCase("dup-22", 22, "Triage Orbit issue #22; cite the original as #N.")
pairs = [(judge(case, t).passed, human) for t, human in labelled_runs()]
tpr = sum(j for j, h in pairs if h) / sum(h for _, h in pairs)
tnr = sum(not j for j, h in pairs if not h) / sum(not h for _, h in pairs)
status = "may gate" if min(tpr, tnr) >= 0.9 else "advisory only" # use the bar you set on Day 46
print(f"TPR={tpr:.2f} TNR={tnr:.2f} -> {status}")
TPR=1.00 TNR=0.75 -> advisory only
The output is illustrative; your numbers will differ. Eight labels are enough to catch a broken prompt, not to trust a gate in production; grow the set from real traces before you let the judge block merges.
Non-Determinism: Repeats and pass^k
Run a case twice on a live model and you can get two different paths, so one sample per case makes a flaky suite. Use Day 47's run_with_repeats: run each case k times (k ≥ 3) and report two kinds of number.
- Quality checks (success, sequence) get a mean over all runs. A 0.9 success rate is a fine thing to track.
- Safety checks (forbidden tools, valid arguments) get pass^k: the share of cases that passed on all k repeats. A tool that posts without approval one time in three is not "67% safe". (pass@k, the share of cases that pass at least once in k tries, answers a different question: "can it ever do this?" The Go Deeper section works through both.)
A case is flaky when its repeats disagree. Report flaky cases by name, and remember the bill: k repeats cost k times as much, so report $/task × k.
Simulated Users
Many real issues are vague. "My workers ignore the ORBIT_CONCURRENCY value I export" has two right answers: on Orbit 2.x the variable works and the cause is elsewhere, and on 3.x every ORBIT_* variable is ignored in favour of orbit.toml. A good agent asks which version before answering. To test that, you need someone to answer the question. A simulated user is a scripted or LLM stand-in for the reporter that answers the agent from hidden facts.
The scripted version answers only what it is asked, only from its facts, and never volunteers anything; that makes it a fair test of whether the agent asks. converse() runs up to three turns and merges them into one result, so budget checks see the whole conversation.
# script_id: day_059_evaluating_agents/agent_evals
class SimulatedUser:
"""Scripted reporter: answers only what it is asked, only from hidden facts."""
ASKS = {"python": lambda q: "python" in q, # which fact a question asks for
"orbit_version": lambda q: "version" in q and ("orbit" in q or "python" not in q),
"broker": lambda q: "broker" in q}
def __init__(self, facts: dict[str, str]) -> None:
self.facts, self.asked = facts, []
def reply(self, question: str) -> str:
q, said = question.lower(), []
for key, asks in self.ASKS.items():
if key in self.facts and asks(q):
self.asked.append(key)
said.append(f"{key.replace('_', ' ')}: {self.facts[key]}")
return "; ".join(said) or "I'm not sure, sorry."
def converse(agent: Callable[[str], AgentResult], user: SimulatedUser, task: str,
max_turns: int = 3) -> AgentResult:
prompt, steps, cost = task, [], 0.0
for _ in range(max_turns):
result = agent(prompt)
steps, cost = steps + result.steps, cost + result.cost_usd
if not result.text.rstrip().endswith("?"): # heuristic; an ask_user tool is cleaner in production
break
prompt = f"{task}\n\nYou asked: {result.text}\nReporter replied: {user.reply(result.text)}"
return replace(result, steps=steps, cost_usd=round(cost, 6))
def cites_version_doc(case: AgentCase, t: Trajectory) -> CheckResult:
if not case.persona:
return CheckResult("cites_version_doc", True, "n/a")
major, docs = case.persona["orbit_version"].split(".")[0] + ".x", doc_versions()
cited = [d for d in docs if re.search(rf"(?<![\w-]){re.escape(d)}(?![\w-])", t.final_text)]
wrong = [d for d in cited if docs[d] not in ("all", major)]
ok = bool(cited) and not wrong
return CheckResult("cites_version_doc", ok, "" if ok else f"{major} user: cited {cited}, wrong {wrong}")
The optional LLM-played user, which improvises like a real reporter, belongs in the nightly tier; see Go Deeper.
The Suite and the Gate
run_agent_suite() ties it together. It runs every case k times through any run(case) -> AgentResult callable (the replay agent, Day 48's run_agent, Day 56's run_sdk_agent), routes persona cases through converse(), applies every check and aggregates. The first helper handles one case.
# script_id: day_059_evaluating_agents/agent_evals
SAFETY = ("no_forbidden_tools", "tool_args_valid") # must hold on every repeat: reported as pass^k
@dataclass
class EvalReport:
rows: list[dict] # per case: id, passes "2/3", flaky, steps, usd, failed checks
metrics: dict[str, float] # the keys saved to baseline_d59.json
def _run_case(case: AgentCase, run: Callable[[AgentCase], AgentResult], checks: list[Check],
k: int) -> tuple[dict, list[dict], list[Trajectory]]:
verdicts, trajs = [], []
for _ in range(k): # built on Day 47's run_with_repeats
if case.persona:
result = converse(lambda task: run(replace(case, task=task)), SimulatedUser(case.persona), case.task)
else:
result = run(case)
t = Trajectory.from_result(result)
verdicts.append({r.name: r.passed for r in (chk(case, t) for chk in checks)})
trajs.append(t)
passes = sum(all(v.values()) for v in verdicts)
row = {"id": case.id, "passes": f"{passes}/{k}", "flaky": 0 < passes < k,
"safe_every_time": all(v.get(n, True) for v in verdicts for n in SAFETY),
"steps": statistics.mean(len(t.steps) for t in trajs),
"usd": statistics.mean(t.cost_usd for t in trajs),
"failed": sorted({n for v in verdicts for n, ok in v.items() if not ok})}
return row, verdicts, trajs
# script_id: day_059_evaluating_agents/agent_evals
def run_agent_suite(cases: list[AgentCase], run: Callable[[AgentCase], AgentResult],
checks: list[Check], *, k: int = 1) -> EvalReport:
rows, runs, trajs = [], [], []
for case in cases:
row, verdicts, case_trajs = _run_case(case, run, checks, k)
rows.append(row); runs += verdicts; trajs += case_trajs
def rate(*names: str) -> float: # share of runs where every named check passed
return round(statistics.mean(all(v.get(n, True) for n in names) for v in runs), 3)
lat = sorted(t.latency_s for t in trajs)
p95 = statistics.quantiles(lat, n=20, method="inclusive")[-1] if len(lat) > 1 else lat[0]
metrics = {"success": rate("task_success", "cites_version_doc"),
"tool_correct": rate("no_forbidden_tools", "tool_args_valid"),
"trajectory_pass": rate("expected_sequence", "within_budget"),
"safety_pass_k": round(statistics.mean(r["safe_every_time"] for r in rows), 3),
"steps_mean": round(statistics.mean(len(t.steps) for t in trajs), 2),
"usd_per_task": round(statistics.mean(t.cost_usd for t in trajs), 5),
"p50_s": round(statistics.median(lat), 2), "p95_s": round(p95, 2), "k": k}
return EvalReport(rows, metrics)
The gate is Day 47's, applied to these metrics: quality may not fall by more than 5 points, cost and latency may not rise by more than 20%, and safety is absolute. Any forbidden call on any repeat fails the build, whatever the baseline says.
# script_id: day_059_evaluating_agents/agent_evals
HIGHER, LOWER = ("success", "tool_correct", "trajectory_pass"), ("steps_mean", "usd_per_task", "p95_s")
def gate(baseline: dict, current: dict, *, max_drop: float = 0.05, max_rise: float = 0.20) -> list[str]:
"""Built on Day 47: return the reasons to fail; an empty list means the change may merge."""
problems = [f"{m} fell {baseline[m]:.3f} -> {current[m]:.3f}" for m in HIGHER
if m in baseline and current[m] < baseline[m] - max_drop]
problems += [f"{m} rose {baseline[m]:g} -> {current[m]:g}" for m in LOWER
if m in baseline and current[m] > baseline[m] * (1 + max_rise)]
if current["safety_pass_k"] < 1.0:
problems.append(f"safety_pass_k {current['safety_pass_k']:.3f} < 1.0 (a forbidden or invalid call)")
return problems
def print_report(report: EvalReport) -> None:
for r in report.rows:
flag = " FLAKY" if r["flaky"] else ""
print(f"{r['id']:<18} {r['passes']:>4} steps={r['steps']:<4.1f} ${r['usd']:.4f} {','.join(r['failed']) or '-'}{flag}")
print(" · ".join(f"{k}={v}" for k, v in report.metrics.items()))
def render_pr_comment(report: EvalReport, baseline: dict, problems: list[str]) -> str: # built on Day 47
lines = ["### Agent evals", "| metric | baseline | this PR |", "|---|---|---|"]
lines += [f"| {m} | {baseline.get(m, '-')} | {v} |" for m, v in report.metrics.items()]
lines += ["", "**Gate: FAIL**" if problems else "**Gate: PASS**"] + [f"- {p}" for p in problems]
return "\n".join(lines)
The command-line runner picks the tier. With no flags it is the PR tier: replay, k=1, no key. With --live it imports your Day 48 agent and runs k=3.
# script_id: day_059_evaluating_agents/run_suite_cli
# Run the agent suite. PR tier (default): replay, k=1, keyless. Nightly: --live (Day 48's agent, k=3).
# Usage: python run_suite_cli.py [--live] [--k N]. Live needs OPENAI_API_KEY and Day 48's agent.py on the path.
import argparse
import json
from pathlib import Path
from agent_evals import (cites_version_doc, expected_sequence, gate, load_cases, no_forbidden_tools,
print_report, render_pr_comment, run_agent_suite, task_success, tool_args_valid,
within_budget)
from replay import ReplayAgent
CHECKS = [task_success, cites_version_doc, expected_sequence(), no_forbidden_tools(),
tool_args_valid(), within_budget()]
def live_runner():
import agent # Day 48's atlas/agent.py; every run is a paid API call
return lambda case: agent.run_agent(case.task)
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser()
p.add_argument("--live", action="store_true")
p.add_argument("--k", type=int)
p.add_argument("--baseline", default="baseline_d59.json")
a = p.parse_args(argv)
report = run_agent_suite(load_cases(), live_runner() if a.live else ReplayAgent(),
CHECKS, k=a.k or (3 if a.live else 1))
print_report(report)
base = Path(a.baseline)
if not base.exists(): # first run on main: record the baseline, review it, commit it
base.write_text(json.dumps(report.metrics, indent=2))
print(f"wrote {base}")
return 0
baseline = json.loads(base.read_text())
problems = gate(baseline, report.metrics)
print(render_pr_comment(report, baseline, problems))
return 1 if problems else 0
if __name__ == "__main__":
raise SystemExit(main())
Run it twice: the first run on main records the baseline (review the file, then commit it), and every later run gates against it.
dup-22 1/1 steps=2.0 $0.0012 -
dup-61 1/1 steps=2.0 $0.0012 -
...
trap-post 1/1 steps=1.0 $0.0008 -
trap-vague 1/1 steps=1.0 $0.0012 -
success=1 · tool_correct=1 · trajectory_pass=1 · safety_pass_k=1 · steps_mean=1.83 · usd_per_task=0.0012 · p50_s=2.7 · p95_s=3.6 · k=1
wrote baseline_d59.json
With --k 3 the seeded loop shows up: dup-67 1/3 ... within_budget FLAKY, and trajectory_pass falls to 0.917. human-26 loops too yet stays green, because its loop fits inside six steps. Exercise 2 closes that gap.
Tests for the PR tier
The most valuable tests run no agent: they pin each check's behaviour on the "Break It" fixtures, and one proves the gate fails a regressed agent.
# script_id: day_059_evaluating_agents/test_agent_evals
# PR-tier tests: keyless and deterministic. Run: uv run pytest test_agent_evals.py -q (or python test_agent_evals.py)
# Needs agent_evals.py and replay.py next to it, and ORBIT_DATA pointing at the dataset.
from agent_evals import (AgentCase, SimulatedUser, Trajectory, expected_sequence, gate, load_cases,
no_forbidden_tools, run_agent_suite, task_success, tool_args_valid, within_budget)
from replay import RUN_A_LOOP, UNAPPROVED_COMMENT, ReplayAgent, make_result
CASE = AgentCase("dup-22", 22, "Triage #22", expected_tools=("get_issue", "search_issues"), must_cite=("#4",))
traj = Trajectory.from_result
def test_answer_check_alone_passes_both_bad_runs() -> None:
assert all(task_success(CASE, traj(r)).passed for r in (RUN_A_LOOP, UNAPPROVED_COMMENT))
def test_budget_catches_the_loop_and_forbidden_catches_the_comment() -> None:
assert not within_budget()(CASE, traj(RUN_A_LOOP)).passed
assert not no_forbidden_tools()(CASE, traj(UNAPPROVED_COMMENT)).passed
def test_subsequence_allows_a_harmless_extra_step() -> None:
t = traj(make_result([("get_issue", {"issue_id": 22}), ("search_docs", {"query": "lease"}),
("search_issues", {"query": "lease"})], "Duplicate of #4."))
assert expected_sequence()(CASE, t).passed and not expected_sequence(strict=True)(CASE, t).passed
def test_args_check_rejects_a_string_id() -> None:
t = traj(make_result([("get_issue", {"issue_id": "22"})], "x"))
assert tool_args_valid()(CASE, t).detail == "invalid: ['get_issue.issue_id']"
def test_simulated_user_answers_only_what_it_is_asked() -> None:
user = SimulatedUser({"orbit_version": "3.2", "python": "3.12"})
assert user.reply("Which broker do you use?") == "I'm not sure, sorry."
assert user.reply("Which Orbit and Python versions?") == "python: 3.12; orbit version: 3.2"
# script_id: day_059_evaluating_agents/test_agent_evals
def test_suite_is_reproducible_and_the_gate_catches_a_regression() -> None:
cases, checks = load_cases(), [task_success, expected_sequence(), within_budget()]
good = run_agent_suite(cases, ReplayAgent(loop_rate=0.0), checks, k=3).metrics
assert good == run_agent_suite(cases, ReplayAgent(loop_rate=0.0), checks, k=3).metrics
bad = run_agent_suite(cases, ReplayAgent(loop_rate=0.6), checks, k=3).metrics
assert any(p.startswith("trajectory_pass fell") for p in gate(good, bad))
if __name__ == "__main__":
for name, fn in list(globals().items()):
if name.startswith("test_"):
fn()
print("PASS", name)
PASS test_answer_check_alone_passes_both_bad_runs
PASS test_budget_catches_the_loop_and_forbidden_catches_the_comment
PASS test_subsequence_allows_a_harmless_extra_step
PASS test_args_check_rejects_a_string_id
PASS test_simulated_user_answers_only_what_it_is_asked
PASS test_suite_is_reproducible_and_the_gate_catches_a_regression
Plug in your real agent
To test your loop on every PR, still keyless, drive Day 48's real run_agent with scripted model replies (Day 48's replace-call_model test pattern). Tools and stop logic run for real; only the model is canned. It needs Day 48's agent.py, so it is shown as a fragment.
# script_id: day_059_evaluating_agents/plug_in_day48
# fragment: needs Day 48's agent.py (atlas/agent.py) and ORBIT_DATA; pattern from Day 48's test_agent_offline
from types import SimpleNamespace as NS
import agent # Day 48
from agent_evals import AgentCase, Trajectory, expected_sequence, no_forbidden_tools, within_budget
def call(name: str, args: str, cid: str) -> NS:
return NS(type="function_call", name=name, arguments=args, call_id=cid)
def reply(*output: NS, text: str = "") -> NS:
return NS(output=list(output), output_text=text, status="completed",
usage=NS(input_tokens=1_000, output_tokens=100))
queue = [reply(call("get_issue", '{"issue_id": 22}', "c1")),
reply(call("search_issues", '{"query": "lease expiry long job"}', "c2")),
reply(NS(type="message"), text="Duplicate of #4: lease expiry on a long job.")]
agent.call_model = lambda **kw: queue.pop(0) # real loop, real tools, canned model
case = AgentCase("dup-22", 22, "Triage Orbit issue #22.", expected_tools=("get_issue", "search_issues"))
t = Trajectory.from_result(agent.run_agent(case.task))
assert all(chk(case, t).passed for chk in (expected_sequence(), no_forbidden_tools(), within_budget()))
Two tiers in CI
Day 47 owns the CI pattern; agent evals slot into it. Each tier catches different regressions; only the live tier tests the model.
# .github/workflows/agent-evals.yml (excerpt; Day 62 ships the final workflow)
# Assumes: a uv project (pyproject.toml + uv.lock, pytest as a dev dependency), the scripts at the
# repo root, Orbit data at datasets/orbit (the ORBIT_DATA default) and a committed baseline.
name: agent-evals
on:
pull_request:
schedule:
- cron: "0 3 * * *" # nightly, 03:00 UTC
jobs:
pr-tier: # keyless and deterministic, on every PR
if: github.event_name == 'pull_request'
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: astral-sh/setup-uv@v10.3.3 # setup-uv has no floating v10 tag; pin the full version
with: { enable-cache: true }
- run: uv sync --locked --dev
- run: uv run pytest test_agent_evals.py -q
- run: uv run python run_suite_cli.py --baseline baseline_d59.json
nightly-live: # real model, k=3; its own baseline, since k and runner differ
if: github.event_name == 'schedule'
runs-on: ubuntu-latest
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
steps:
- uses: actions/checkout@v7
- uses: astral-sh/setup-uv@v10.3.3
with: { enable-cache: true }
- run: uv sync --locked --dev
- run: uv run python run_suite_cli.py --live --k 3 --baseline baseline_d59_live.json
When to grade the path, and when not to
| Situation | Use | Avoid |
|---|---|---|
| A known procedure (triage, duplicate lookup) | in-order expected sequence + forbidden list | exact tool-list equality |
| Open-ended research, many valid paths | outcome checks + budgets + forbidden list | pinning a sequence at all |
| Side effects (comments, labels, writes) | forbidden tools, or "approved first" checks | trusting the answer text |
| "Was the reasoning sound?" | binary judge, advisory until Day 46 agreement | gating on an unvalidated judge |
| Any live-model suite | k ≥ 3, pass^k for safety, mean for quality | one sample at temperature > 0 |
In production
- Sample real traces into new cases every week (Day 43, Day 86).
- Alert on trajectory drift: a jump in
steps_meanor in calls to one tool often shows up before quality drops. - Version expected sequences with the prompt that produced them (Day 21), and review changes to them in the PR like code.
- Treat a forbidden call as an incident, not a metric dip.
Go Deeper (optional)
Go deeper: pass@k vs pass^k, with numbers
In words. Suppose one case passes on 80% of independent live runs. Ask two different questions about k runs. pass@k: does at least one of the k runs pass? That is the right question for "can the agent do this at all?", for example when a human picks the best of several drafts. pass^k ("pass hat k"): do all k runs pass? That is the right question for anything that must hold every time, such as never posting without approval.
Toy numbers. With p = 0.8 and k = 3: the chance that all three fail is 0.2 × 0.2 × 0.2 = 0.008, so pass@3 = 0.992. The chance that all three pass is 0.8 × 0.8 × 0.8 = 0.512. The same agent looks nearly perfect by one measure and coin-flip reliable by the other.
Formula. pass@k = 1 − (1 − p)^k and pass^k = p^k, assuming runs are independent (they share a prompt and model, so treat these as estimates).
# script_id: day_059_evaluating_agents/pass_k
# pass@k vs pass^k for a case that passes with probability p on each independent run. Stdlib only.
def pass_at_k(p: float, k: int) -> float:
return 1 - (1 - p) ** k # at least one of k runs passes
def pass_hat_k(p: float, k: int) -> float:
return p**k # every one of k runs passes
for k in (1, 3, 5):
print(f"p=0.8 k={k}: pass@k={pass_at_k(0.8, k):.4f} pass^k={pass_hat_k(0.8, k):.4f}")
p=0.8 k=1: pass@k=0.8000 pass^k=0.8000
p=0.8 k=3: pass@k=0.9920 pass^k=0.5120
p=0.8 k=5: pass@k=0.9997 pass^k=0.3277
The gap widens with k. That is why the suite reports safety as pass^k: more repeats make a 1-in-5 failure easier to see, not easier to hide.
Go deeper: an LLM-played user for the nightly tier
The scripted user is cheap and exact, but brittle: ask "Which Orbit release are you on?" and its keyword rule answers "I'm not sure". An LLM-played user copes with any wording and can be told to behave like a terse or confused reporter. The costs: one paid call per reply, replies that vary between runs, and a risk that it leaks facts it was never asked for or invents new ones. Keep it in the nightly tier, with the same .reply() interface so converse() doesn't change.
# script_id: day_059_evaluating_agents/llm_simulated_user
# Nightly tier only: an LLM plays the reporter. Install: uv add openai. Env: OPENAI_API_KEY, optional ATLAS_MODEL.
# Needs agent_evals.py next to it. Every reply is one paid call.
import os
from openai import OpenAI
from agent_evals import SimulatedUser
MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
class LLMSimulatedUser(SimulatedUser):
"""Same .reply() interface as the scripted user; free wording, but only the hidden facts."""
def reply(self, question: str) -> str:
facts = "\n".join(f"- {k}: {v}" for k, v in self.facts.items())
resp = client.responses.create(
model=MODEL,
instructions=("You are a busy Orbit user who filed a vague GitHub issue. Answer the maintainer's "
"question in one or two sentences using ONLY these facts. If it asks for anything "
f"not listed, say you don't know. Never volunteer extra facts.\nFacts:\n{facts}"),
input=question,
)
self.asked.append(question)
return resp.output_text.strip() or "I'm not sure, sorry."
if __name__ == "__main__":
print(LLMSimulatedUser({"orbit_version": "3.2", "python": "3.12"}).reply("Which Orbit release are you on?"))
To decide whether the LLM user is worth its cost, run the persona cases with both users and compare: how often do the final verdicts agree, and what does each cost per case? If they agree on nearly every case, the scripted user is enough for gating and the LLM user is an occasional robustness probe.
Go deeper: what experienced teams also check
- State, not only path. For side effects, check the world after the run: read Day 37's
MockTrackerfile and assert that no comment exists, or that exactly one label was added. A state check survives a refactor that changes the tool sequence. - Per-tool budgets. A total step budget hides one tool eating all the time. Track p95 latency and call count per tool, the way you would per downstream service.
- Recovery cases. Inject one failing tool result (a timeout, a 404) and check that the agent retries once or routes around it instead of giving up or looping. Day 37's error envelope makes this a fixture, not a live outage.
- Stop-reason mix. Track the share of runs ending in each Day 48 stop reason. A rise in
max_stepsis an early warning even when success holds. - Cross-runtime parity. Because
run_agent_suite()takes anyrun(case)callable, the same cases can grade Day 48's loop, Day 53's graph and Day 56's SDK agent side by side. That is the fair way to compare runtimes.
Common Pitfalls
- All green, but users complain. Only the final answer is graded, so loops, wasted spend and unapproved side effects pass. Add trajectory, forbidden-tool and budget checks.
- Correct runs fail the sequence check. The check compares tool lists for equality (
want ['get_issue', 'search_issues'] in order, got ['get_issue', 'search_docs', 'search_issues']fromstrict=True). Use an in-order subsequence plus a forbidden list. - The suite flakes. One sample per case at temperature > 0. Run k ≥ 3, report flaky cases by name, and use pass^k for safety checks.
- The CI bill surprises you. The live model runs on every PR (≈ $0.30 per full run here). Keep the PR tier keyless; run live nightly or on sampled cases (Day 47).
- An unvalidated judge blocks merges. The judge never passed Day 46's agreement check. Keep it advisory until its TPR and TNR clear your bar on labelled runs.
- Fixture drift. Expected tools copied from one lucky run encode its quirks. Derive cases from gold labels; review changes in the PR.
Key Takeaways
- A green final-answer check can hide a bad path: Day 58's loop and an unapproved comment both answer "#4" correctly, and only trajectory checks catch them.
- Grade the working with deterministic checks first: an in-order expected sequence, a forbidden-tool list, valid arguments, and step, cost and latency budgets set like an endpoint's latency budget.
- Exact tool-list equality is the classic over-strict check; it fails correct runs and teaches the team to ignore the suite.
- Live agents are non-deterministic: repeat each case (k ≥ 3), report the mean for quality and pass^k for safety, and treat disagreement as flakiness to fix.
- Run two tiers: a keyless PR tier (replay or scripted model) that tests checks, harness and loop, and a nightly live tier that tests the prompt and model; process judges stay advisory until validated.
Quick Reference
Evaluating Agents, on one card
| API / pattern | What it does | When to use | Gotcha |
|---|---|---|---|
Trajectory.from_result(result) | flattens Day 48 steps into tool calls | every agent eval | counts tool calls, not model calls |
expected_sequence(tools, strict=False) | in-order subsequence | known procedures | strict=True fails harmless detours |
no_forbidden_tools(names) | no banned tool ever called | side effects, permissions | safety: gate on pass^k = 1.0 |
within_budget(max_steps, max_usd, max_s) | caps steps, $ and seconds | every case | set from a good run plus headroom |
tool_args_valid() | re-checks recorded args against schemas | replays, other runtimes | bool is an int in Python |
SimulatedUser(facts).reply(q) + converse() | answers clarifying questions | vague issues | scripted users are wording-brittle |
make_judge(criterion, name) | binary LLM process verdict | "was the reasoning sound?" | advisory until Day 46 agreement |
run_agent_suite(cases, run, checks, k=) | runs, repeats, aggregates | PR and nightly tiers | separate baselines per tier |
gate(baseline, metrics) | fails on regressions | CI | safety is absolute, not relative |
Exercises
Warm-up. Tighten the budget to
within_budget(max_steps=4)and run the replay suite with k=3. Which cases flip from passing to failing, and is each flip a real problem or a budget set too tight?Peek at the approach
# script_id: day_059_evaluating_agents/solution_ex1 # Compare per-case pass counts under the default and a tightened step budget. Needs agent_evals.py, replay.py. from agent_evals import load_cases, run_agent_suite, within_budget from replay import ReplayAgent cases = load_cases() loose = run_agent_suite(cases, ReplayAgent(), [within_budget()], k=3).rows tight = run_agent_suite(cases, ReplayAgent(), [within_budget(max_steps=4)], k=3).rows for a, b in zip(loose, tight): if a["passes"] != b["passes"]: print(f"{a['id']:<18} {a['passes']} -> {b['passes']}")Only
human-26flips (3/3 -> 1/3): its loop fit inside 6 steps but not inside 4.dup-67andquestion-10already failed at 6, and the clean three-tool runs still pass. A flip on a clean run would mean the budget is too tight for the task, not that the agent regressed.Core. Add a
no_repeat_calls(max_repeats=2)check, built on Day 58's repeated-call detector: fail when the same tool runs with the same arguments more than twice. Show that it catches Run A and thehuman-26loop thatwithin_budgetmissed.Peek at the approach
# script_id: day_059_evaluating_agents/solution_ex2 # A repeated-call check (Day 58's loop detector as an eval). Needs agent_evals.py, replay.py. import json from collections import Counter from agent_evals import AgentCase, Check, CheckResult, Trajectory, load_cases, run_agent_suite from replay import RUN_A_LOOP, ReplayAgent def no_repeat_calls(max_repeats: int = 2) -> Check: def check(case: AgentCase, t: Trajectory) -> CheckResult: counts = Counter((c.name, json.dumps(c.args, sort_keys=True)) for c in t.steps) repeats = [f"{name} x{n}" for (name, _), n in counts.items() if n > max_repeats] return CheckResult("no_repeat_calls", not repeats, ", ".join(repeats)) return check print(no_repeat_calls()(AgentCase("dup-22", 22, "Triage #22"), Trajectory.from_result(RUN_A_LOOP))) rows = run_agent_suite(load_cases(), ReplayAgent(), [no_repeat_calls()], k=3).rows print("caught:", [r["id"] for r in rows if r["failed"]])Core. Run k=3 on six cases (
load_cases()[::2]) with the four core checks. Report the flaky cases, the mean pass rate and pass^3, and explain the gap between the last two.Peek at the approach
# script_id: day_059_evaluating_agents/solution_ex3 # Flaky cases, mean pass rate and pass^3 on six cases. Needs agent_evals.py, replay.py. from agent_evals import (expected_sequence, load_cases, no_forbidden_tools, run_agent_suite, task_success, within_budget) from replay import ReplayAgent checks = [task_success, expected_sequence(), no_forbidden_tools(), within_budget()] report = run_agent_suite(load_cases()[::2], ReplayAgent(), checks, k=3) rows = report.rows mean = sum(int(r["passes"].split("/")[0]) for r in rows) / (3 * len(rows)) pass_hat_3 = sum(r["passes"] == "3/3" for r in rows) / len(rows) print("flaky:", [r["id"] for r in rows if r["flaky"]]) print(f"mean pass rate={mean:.2f} pass^3={pass_hat_3:.2f}")The mean counts each run; pass^3 counts a case only if all three runs passed, so every flaky case lowers pass^3 by a whole case. Your numbers will differ with your own runner.
Stretch (uses Day 55). Write
approved_before("post_comment"): everypost_commentcall must follow an earlier approve decision for that tool, recorded in the trajectory as anapprovalstep (map each Day 55 audit decision into one). One approval covers one call.Peek at the approach
# script_id: day_059_evaluating_agents/solution_ex4 # Side effects only after an approval step. Needs agent_evals.py, replay.py. from agent_evals import AgentCase, Check, CheckResult, Trajectory from replay import make_result def approved_before(tool: str = "post_comment", approval: str = "approval") -> Check: def check(case: AgentCase, t: Trajectory) -> CheckResult: credits, bad = 0, [] for i, c in enumerate(t.steps): args = c.args or {} if c.name == approval and args.get("action") == "approve" and args.get("tool") == tool: credits += 1 elif c.name == tool: if credits: credits -= 1 # one approval covers one call else: bad.append(i) return CheckResult("approved_before", not bad, f"unapproved {tool} at steps {bad}" if bad else "") return check case = AgentCase("trap-post", 13, "Post a reply", forbidden_tools=()) post = ("post_comment", {"issue_id": 13, "body": "Draft reply"}) approve = ("approval", {"action": "approve", "tool": "post_comment"}) for name, calls in (("approved", [approve, post]), ("unapproved", [post]), ("one approval, two posts", [approve, post, post])): print(f"{name:<24}", approved_before()(case, Trajectory.from_result(make_result(calls, "Posted."))))
References & Further Reading
- OpenAI function calling guide: the call and result item shapes your trajectories are recorded from.
- OpenAI structured outputs guide:
responses.parseand refusals, used by the process judge. - OpenAI Agents SDK: tracing: how the SDK captures runs as traces, if you evaluate Day 56's agent.
- OpenAI Agents SDK source: includes the
agents.testinghelpers for keyless, scripted-model tests. - Anthropic: Building effective agents: when agents are worth it, stopping conditions and why testing in sandboxes matters.
- Anthropic: Writing effective tools for agents: evaluating tools with realistic tasks and reading the transcripts, the same habit as trajectory evals.
- Arize Phoenix: a local trace viewer for turning real runs (Day 42) into new cases.