{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "*Lab 8 Answer Key*\n", "\n", "# Claude Code, Evals & Debugging\n", "\n", "**Objective:** practise the isolate-before-you-fix discipline, classify errors correctly, and stand up a minimal eval with a CI gate.\n", "\n", "When output is wrong, the first decision is not how to fix it but where the failure lives: the integration layer (your code) or the model output. A failure that reproduces on replay is your code; a failure that varies is the model. This lab instruments a client so you can replay at all, walks one bug of each kind, classifies errors as fatal or retryable, and finishes with an eval harness plus a pinned, least-privilege Claude Code step in CI.\n", "\n", "**Setup:** `pip install anthropic jsonschema` and set `ANTHROPIC_API_KEY` in your environment. Never hardcode a key.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 1: Instrument first\n", "\n", "You cannot analyze a trace you did not record. The floor for every call is the model ID, `stop_reason`, all four `usage` fields, `response._request_id`, and latency. Miss `stop_reason` and you cannot tell after the fact whether an answer was truncated; miss the `usage` breakdown and you cannot tell whether a prompt-cache hit silently became a miss and doubled your bill.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import time, json, logging\n", "import anthropic\n", "\n", "logging.basicConfig(level=logging.INFO, format=\"%(message)s\")\n", "log = logging.getLogger(\"llm\")\n", "\n", "client = anthropic.Anthropic() # key resolved from the environment\n", "\n", "\n", "def extract_text(response) -> str:\n", " \"\"\"Block-safe: never index content[0]. Narrow on .type.\"\"\"\n", " return \"\".join(b.text for b in response.content if b.type == \"text\")\n", "\n", "\n", "def logged_create(client, **kwargs):\n", " t0 = time.perf_counter()\n", " resp = client.messages.create(**kwargs)\n", " u = resp.usage\n", " log.info(json.dumps({\n", " \"model\": resp.model,\n", " \"request_id\": resp._request_id, # public despite the underscore\n", " \"stop_reason\": resp.stop_reason, # truncation shows up here\n", " \"latency_ms\": round((time.perf_counter() - t0) * 1000),\n", " \"usage\": {\n", " \"input\": u.input_tokens, # uncached remainder only\n", " \"cache_creation\": u.cache_creation_input_tokens,\n", " \"cache_read\": u.cache_read_input_tokens, # cache hit shows up here\n", " \"output\": u.output_tokens,\n", " },\n", " }))\n", " # Total prompt tokens = input + cache_creation + cache_read.\n", " return resp\n", "\n", "\n", "resp = logged_create(\n", " client,\n", " model=\"claude-haiku-4-5\",\n", " max_tokens=128,\n", " messages=[{\"role\": \"user\", \"content\": \"Name three uses for a paperclip.\"}],\n", ")\n", "print(extract_text(resp))\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 2: Isolate a deterministic bug\n", "\n", "Replay the exact request. If the same request reproduces the failure every time, it is deterministic, which is almost always the integration layer: your code does the same wrong thing on every run. The fix is the model ID, not the prompt.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# A bogus model ID fails the same way on every replay. That reproducibility\n", "# is the signature of an integration-layer bug.\n", "failures = []\n", "for attempt in range(3):\n", " try:\n", " logged_create(\n", " client,\n", " model=\"claude-opus-9-9\", # bogus on purpose\n", " max_tokens=64,\n", " messages=[{\"role\": \"user\", \"content\": \"hi\"}],\n", " )\n", " except anthropic.NotFoundError as e:\n", " failures.append((type(e).__name__, e.status_code))\n", "\n", "print(\"three replays:\", failures)\n", "assert len(set(failures)) == 1, \"expected the identical failure every time\"\n", "print(\"reproducible on every replay -> integration layer. Fix the model ID, not the prompt.\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 3: Isolate a probabilistic bug\n", "\n", "If the output varies run to run on an identical request, the failure lives in the model output, and the lever is prompt, context, or effort, not a code patch. Prove it by tightening the instruction and watching the variance drop; do not touch parsing.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "VAGUE = \"Summarize this: revenue up, costs flat, churn worse.\"\n", "\n", "vague_outputs = set()\n", "for run in range(5):\n", " resp = logged_create(\n", " client,\n", " model=\"claude-haiku-4-5\",\n", " max_tokens=128,\n", " messages=[{\"role\": \"user\", \"content\": VAGUE}],\n", " )\n", " vague_outputs.add(extract_text(resp))\n", "\n", "print(f\"{len(vague_outputs)} distinct outputs across 5 identical requests\")\n", "# Varies run to run -> the model, not your code. The lever is the prompt.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Tighten the instruction: same request shape, far less room to wander.\n", "TIGHT = (\n", " \"Summarize in exactly one sentence of at most 15 words, naming revenue, \"\n", " \"costs, and churn in that order: revenue up, costs flat, churn worse.\"\n", ")\n", "\n", "tight_outputs = set()\n", "for run in range(5):\n", " resp = logged_create(\n", " client,\n", " model=\"claude-haiku-4-5\",\n", " max_tokens=128,\n", " messages=[{\"role\": \"user\", \"content\": TIGHT}],\n", " )\n", " tight_outputs.add(extract_text(resp))\n", "\n", "print(f\"vague: {len(vague_outputs)} distinct | tight: {len(tight_outputs)} distinct\")\n", "print(\"the spread collapsed via the prompt; parsing code was never the problem\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 4: Classify errors\n", "\n", "Recovery starts with one binary: retryable or fatal. Retrying a fatal error hammers the API with a request that can never succeed; failing a retryable error hard drops work that would have gone through on the next attempt. The SDK already retries connection errors, 408, 409, 429, and 5xx twice with exponential backoff.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "RETRYABLE = (anthropic.RateLimitError,\n", " anthropic.InternalServerError,\n", " anthropic.APIConnectionError)\n", "\n", "\n", "def classify(err: Exception) -> str:\n", " \"\"\"Return 'retry' or 'fatal': the only decision that matters first.\"\"\"\n", " if isinstance(err, RETRYABLE):\n", " return \"retry\"\n", " if isinstance(err, (anthropic.BadRequestError, # 400\n", " anthropic.AuthenticationError, # 401\n", " anthropic.PermissionDeniedError, # 403\n", " anthropic.NotFoundError)): # 404\n", " return \"fatal\"\n", " if isinstance(err, anthropic.APIStatusError): # catch-all by code\n", " return \"retry\" if err.status_code >= 500 else \"fatal\"\n", " return \"fatal\"\n", "\n", "\n", "# Trigger a real 404 with a bad model ID.\n", "try:\n", " client.messages.create(model=\"claude-opus-9-9\", max_tokens=16,\n", " messages=[{\"role\": \"user\", \"content\": \"hi\"}])\n", "except anthropic.NotFoundError as e:\n", " print(\"404 classified as:\", classify(e))\n", " assert classify(e) == \"fatal\"\n", "\n", "# Mock a 429 rather than hammering the endpoint; the class is what matters.\n", "import httpx\n", "req = httpx.Request(\"POST\", \"https://api.anthropic.com/v1/messages\")\n", "mock_429 = anthropic.RateLimitError(\n", " \"rate limited\",\n", " response=httpx.Response(429, request=req, headers={\"retry-after\": \"30\"}),\n", " body=None,\n", ")\n", "print(\"429 classified as:\", classify(mock_429))\n", "assert classify(mock_429) == \"retry\"\n", "\n", "# The SDK absorbs the retryable set twice by default. With max_retries=0 a 429\n", "# surfaces to YOUR code on the first hit instead of being handled for you.\n", "no_retry = anthropic.Anthropic(max_retries=0)\n", "print(\"default retries:\", client.max_retries, \"| disabled:\", no_retry.max_retries)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 5: Build the eval\n", "\n", "Editing a prompt is editing code, and code changes need tests. An eval harness is a fixed set of inputs, a grader, and a score you record per version. Schema validity is a free, deterministic grader: make it the first gate before the exact-match check.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import jsonschema\n", "\n", "SCHEMA = {\n", " \"type\": \"object\",\n", " \"properties\": {\"intent\": {\"type\": \"string\"}, \"order_id\": {\"type\": \"string\"}},\n", " \"required\": [\"intent\"], \"additionalProperties\": False,\n", "}\n", "\n", "# 1. Fixed inputs with expected properties.\n", "CASES = [\n", " {\"text\": \"Refund my order #A17, it arrived broken.\",\n", " \"expect\": {\"intent\": \"refund\", \"order_id\": \"A17\"}},\n", " {\"text\": \"What are your hours?\",\n", " \"expect\": {\"intent\": \"hours\"}},\n", " {\"text\": \"Cancel order B22 before it ships.\",\n", " \"expect\": {\"intent\": \"cancel\", \"order_id\": \"B22\"}},\n", "]\n", "\n", "PROMPTS = {\n", " \"v1\": \"Classify the customer message.\",\n", " \"v2\": (\"Classify the customer message. intent is one of: refund, cancel, \"\n", " \"hours, other. If an order id like A17 appears, copy it into \"\n", " \"order_id exactly as written; otherwise omit the field.\"),\n", "}\n", "\n", "\n", "def grade(output: dict, case: dict) -> float:\n", " # (a) schema validity: a free, deterministic grader.\n", " try:\n", " jsonschema.validate(output, SCHEMA)\n", " except jsonschema.ValidationError:\n", " return 0.0\n", " # (b) exact-match the fields we care about.\n", " return 1.0 if all(output.get(k) == v for k, v in case[\"expect\"].items()) else 0.5\n", "\n", "\n", "def run_eval(prompt_version: str) -> float:\n", " scores = []\n", " for case in CASES:\n", " resp = logged_create(\n", " client,\n", " model=\"claude-haiku-4-5\",\n", " max_tokens=512,\n", " output_config={\"format\": {\"type\": \"json_schema\", \"schema\": SCHEMA}},\n", " system=PROMPTS[prompt_version],\n", " messages=[{\"role\": \"user\", \"content\": case[\"text\"]}],\n", " )\n", " out = json.loads(extract_text(resp))\n", " scores.append(grade(out, case))\n", " return sum(scores) / len(scores)\n", "\n", "\n", "score_v1 = run_eval(\"v1\")\n", "print(\"v1 score:\", score_v1)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 6: Regression-gate it\n", "\n", "\"The new prompt seems better\" is a vibe, not evidence. Compare the recorded scores across versions and let an assert block the change when the number goes down. The same gate protects a model upgrade: run the eval against the new pinned ID before switching.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Compare versions the way you compare test runs.\n", "score_v2 = run_eval(\"v2\")\n", "print(\"v1:\", score_v1, \" v2:\", score_v2)\n", "assert score_v2 >= score_v1, \"prompt regression: block the merge\"\n", "print(\"gate passed: v2 is at least as good as v1\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "In CI the assert becomes the exit code: put `run_eval` in a script, run it as a\n", "build step, and a lower score fails the build before the prompt ships. Report a\n", "pass rate over the case set, not a single trial; the score is comparable across\n", "prompt versions and across model versions, which is exactly what you need before\n", "either kind of change reaches production.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part 7: Claude Code in CI\n", "\n", "A Claude Code invocation in CI is a build step and deserves the same discipline as any other: pin the version (never `@latest`), run headless with `-p`, and grant least privilege in `settings.json`. The review job needs to read and run tests, not to push or delete.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import json, pathlib\n", "\n", "# Least privilege: pre-approve safe reads and test commands, gate pushes,\n", "# deny destructive commands and secret files outright.\n", "settings = {\n", " \"permissions\": {\n", " \"allow\": [\"Read\", \"Grep\", \"Glob\", \"Bash(pytest:*)\", \"Bash(ruff:*)\"],\n", " \"ask\": [\"Bash(git push:*)\"],\n", " \"deny\": [\"Bash(rm -rf:*)\", \"Read(./.env)\"],\n", " }\n", "}\n", "pathlib.Path(\".claude\").mkdir(exist_ok=True)\n", "pathlib.Path(\".claude/settings.json\").write_text(json.dumps(settings, indent=2))\n", "print(pathlib.Path(\".claude/settings.json\").read_text())\n", "\n", "# The CI step itself: pinned version, headless -p, key from the secret store.\n", "workflow = r\"\"\"\n", "- name: Claude review gate\n", " run: |\n", " npm install -g @anthropic-ai/claude-code@2.4.1 # PINNED version, not @latest\n", " claude -p \"Review the staged diff for security regressions. \\\n", " Exit non-zero if any HIGH-severity issue is found.\" \\\n", " --output-format stream-json\n", " env:\n", " ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}\n", "\"\"\"\n", "print(workflow)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Prove the gate before trusting it: plant a HIGH-severity issue on a branch (a\n", "hardcoded API key in a test file works), open a PR, and confirm the job exits\n", "non-zero. Remove the plant and confirm the job goes green. A gate you never\n", "watched fail is a gate you are trusting on faith, which is the same lesson the\n", "eval taught: evidence, not vibes.\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.0" } }, "nbformat": 4, "nbformat_minor": 5 }