Evaluation & Observability

# agent.py

"""agent.py - the order-status agent the rest of this lesson observes."""
import json

import anthropic

client = anthropic.Anthropic()

ORDERS = {
    "A-1042": {"order_id": "A-1042", "status": "delivered",
               "delivered_on": "2026-08-02", "carrier": "UPS"},
    "A-1099": {"order_id": "A-1099", "status": "in_transit",
               "eta": "2026-08-19", "carrier": "FedEx"},
}

TOOLS = [{
    "name": "get_order",
    "description": "Look up the status of a customer order by its id.",
    "input_schema": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
}]


def get_order(order_id):
    return ORDERS.get(order_id, {"error": "not_found", "order_id": order_id})


def run_agent(question, model="claude-haiku-4-5"):
    """The same loop as below, without the instrumentation."""
    messages = [{"role": "user", "content": question}]
    while True:
        resp = client.messages.create(
            model=model, max_tokens=400, tools=TOOLS, messages=messages,
        )
        if resp.stop_reason != "tool_use":
            return "".join(b.text for b in resp.content if b.type == "text")
        messages.append({"role": "assistant", "content": resp.content})
        # tool_result content must be a string, never a raw dict.
        messages.append({"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": b.id,
             "content": json.dumps(get_order(**b.input))}
            for b in resp.content if b.type == "tool_use"
        ]})


if __name__ == "__main__":
    print(run_agent("Where is order A-1042?"))

# Same loop as run_agent(), wrapped in spans. Durations and token counts vary run to run.

import json
import time
from contextlib import contextmanager

from agent import TOOLS, client, get_order

SPANS = []


@contextmanager
def span(name, **attrs):
    start = time.monotonic()
    rec = {"name": name, "attrs": attrs}
    SPANS.append(rec)
    try:
        yield rec
    finally:
        rec["ms"] = round((time.monotonic() - start) * 1000)


messages = [{"role": "user", "content": "Where is order A-1042?"}]

with span("agent.run", question=messages[0]["content"]):
    while True:
        with span("model.request", model="claude-haiku-4-5") as s:
            resp = client.messages.create(
                model="claude-haiku-4-5", max_tokens=400,
                tools=TOOLS, messages=messages,
            )
            s["attrs"]["stop_reason"] = resp.stop_reason
            s["attrs"]["in"] = resp.usage.input_tokens
            s["attrs"]["out"] = resp.usage.output_tokens

        if resp.stop_reason != "tool_use":
            break

        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for b in resp.content:
            if b.type == "tool_use":
                with span("tool.execute", tool=b.name, args=b.input) as s:
                    out = get_order(**b.input)
                    s["attrs"]["ok"] = "error" not in out
                results.append({"type": "tool_result", "tool_use_id": b.id,
                                "content": json.dumps(out)})
        messages.append({"role": "user", "content": results})

for sp in SPANS:
    print(f"{sp['ms']:>6}ms  {sp['name']:<16} {json.dumps(sp['attrs'])}")

# Model-as-judge with a JSON schema.

import json

from agent import ORDERS, client

RUBRIC = """Score the ASSISTANT ANSWER against the rubric. Reply with JSON only:
{"grounded": true|false, "answers_question": true|false, "score": 1-5, "why": "..."}

grounded        : every factual claim appears in TOOL DATA
answers_question: it actually addresses the user's question"""

CANDIDATES = {
    "good": "Order A-1042 was delivered on 2026-08-02.",
    "hallucinated": "Order A-1042 is out for delivery and should arrive by 6pm tomorrow.",
}

for label, candidate in CANDIDATES.items():
    r = client.messages.create(
        model="claude-haiku-4-5", max_tokens=300,
        messages=[{"role": "user", "content":
            f"{RUBRIC}\n\nUSER QUESTION: Where is order A-1042?\n"
            f"TOOL DATA: {json.dumps(ORDERS['A-1042'])}\n"
            f"ASSISTANT ANSWER: {candidate}"}],
        # Structured output, so the verdict is parseable rather than prose.
        output_config={"format": {"type": "json_schema", "schema": {
            "type": "object",
            "properties": {
                "grounded": {"type": "boolean"},
                "answers_question": {"type": "boolean"},
                "score": {"type": "integer"},
                "why": {"type": "string"},
            },
            "required": ["grounded", "answers_question", "score", "why"],
            "additionalProperties": False,
        }}},
    )
    verdict = json.loads(next(b.text for b in r.content if b.type == "text"))
    print(f"{label:<14} -> grounded={verdict['grounded']!s:<5} "
          f"score={verdict['score']}  {verdict['why'][:60]}")

# A golden set with substring assertions. Imports the same agent.py from section 2.

from agent import run_agent

GOLDEN = [
    ("Where is order A-1042?",             "delivered"),
    ("Has A-1042 shipped?",                "delivered"),
    ("What's the status of order A-1042?", "delivered"),
    ("Tell me about order A-9999.",        "not found"),
]

passed = 0
for question, expect in GOLDEN:
    text = run_agent(question).lower()
    ok = expect in text            # substring assertion
    passed += ok
    print(f"  {'PASS' if ok else 'FAIL'}  {question:<34} expected '{expect}'")

rate = passed / len(GOLDEN)
print(f"\npass rate: {passed}/{len(GOLDEN)} = {rate:.0%}   threshold: 90%")
print("CI would", "PASS" if rate >= 0.9 else "FAIL", "this build")

# The same golden set, asserted by judge instead of substring. The A-9999 case now passes on any correct phrasing.

import json

from agent import client, run_agent


def judge(answer, criterion):
    """Ask a cheap model whether the answer meets the criterion, in any wording."""
    r = client.messages.create(
        model="claude-haiku-4-5", max_tokens=300,
        messages=[{"role": "user", "content":
            "Does the ANSWER satisfy the CRITERION? Wording does not have to match.\n"
            f"CRITERION: {criterion}\nANSWER: {answer}"}],
        output_config={"format": {"type": "json_schema", "schema": {
            "type": "object",
            "properties": {"meets": {"type": "boolean"}, "why": {"type": "string"}},
            "required": ["meets", "why"],
            "additionalProperties": False,
        }}},
    )
    return json.loads(next(b.text for b in r.content if b.type == "text"))


# Replace the substring assertion with a semantic one.
GOLDEN = [
    ("Where is order A-1042?",      "States the order was delivered on 2026-08-02."),
    ("Tell me about order A-9999.", "Says the order could not be found. Any phrasing is fine."),
]

for question, criterion in GOLDEN:
    answer = run_agent(question)
    verdict = judge(answer, criterion)
    print(f"  {'PASS' if verdict['meets'] else 'FAIL'}  {question}")