Advanced Python for Agents

This shows Pydantic, a library that turns a class into a strict data shape. You describe what fields you expect and their types; Pydantic checks the data matches the moment you build the object, and raises a clear error if it doesn't. This is how the course gets trustworthy JSON out of an LLM.

1. class Diagnosis(BaseModel): — inheriting from BaseModel is what gives the class its validation powers. The lines under it are the fields and their types: likely_cause: str means "this must be text", evidence: list[str] means "a list of text items".
2. Literal["read_only", "reversible", "irreversible"] restricts fix_risk to exactly those three words — anything else is rejected. Think of it as a fixed menu.
3. confidence: float = Field(ge=0.0, le=1.0) — a decimal number that must be greater-or-equal to 0 and less-or-equal to 1. runbook_ref: Optional[str] = None means that field may be text or left out entirely (its default is None, Python's "no value").
4. The line d = Diagnosis(likely_cause=..., confidence=0.9) is where validation actually runs. Because the data is valid, you get an object d and can read fields with a dot: d.confidence gives back 0.9.

What the output means: On good data you get a typed object (not a string to parse). The commented-out lines show what bad data does: confidence=5.0 or fix_risk="nuke" would each raise a ValidationError that names the exact problem.

Try this: Mentally set confidence=1.5 — Pydantic would refuse it because of le=1.0. That refusal is the whole point: bad data is stopped before your code ever sees it.

When something goes wrong at runtime (network drops, rate limit hit, bad input), Python raises an exception — an error object that stops normal flow. try/except lets you catch it and react calmly instead of crashing. The rule: catch the specific errors you can handle, and list the most specific ones first.

1. Everything under try: is the risky code. Here it's the API call client.messages.create(...). If it works, the except blocks are skipped.
2. Each except names one error type. anthropic.RateLimitError (you sent requests too fast) comes first because it's the most specific. Order matters: Python uses the first matching block, so a broad type listed early would swallow the narrow ones.
3. except anthropic.APIStatusError as e: catches a broader family of server responses; as e gives you the error object so you can inspect it — here e.status_code >= 500 checks whether it was a server-side failure.
4. At the bottom, raise RefusalError(resp) shows you can also throw your own exception when the model declines — signalling a problem upward for a caller to handle.

What the output means: Nothing prints unless a matching error occurs; then exactly one print runs (e.g. "slow down, retry later" for a rate limit). The program keeps going instead of dying.

Try this: Ask yourself what happens if you put except Exception at the top — it would catch everything and none of the specific blocks below would ever run. That's the mistake the "Don't catch everything" warning is about.

A with block is a context manager: it sets something up, lets you use it, and guarantees cleanup at the end — even if an error happens inside. You already saw this with files; the same shape powers streaming from the model.

1. with open("runbook.md") as f: opens the file and names it f. The indented text = f.read() reads its contents. When the block ends, the file is closed automatically — you never have to remember to close it.
2. The second block does the same for a live network stream: with client.messages.stream(...) as stream: opens the connection and cleans it up when done, so a crash mid-stream can't leak an open connection.
3. Inside, for text in stream.text_stream: loops over pieces of the reply as they arrive and prints each immediately (end="" avoids line breaks, flush=True shows it right away).
4. final = stream.get_final_message() grabs the complete assembled message after the stream finishes.

What the output means: You'd see the model's answer appear gradually, chunk by chunk, rather than all at once when it's fully done.

Try this: Compare to closing a file by hand with f.close() — if an error struck before that line, the file would stay open. with removes that whole class of bug.

A generator is a function that produces values one at a time, on demand, instead of building a whole list up front. The keyword yield is what makes it a generator: it hands back a single value and pauses, resuming where it left off next time you ask.

1. def count_to(n): looks like a normal function, but because its body uses yield i, calling it doesn't run the code — it returns a generator you can loop over.
2. Each time the loop asks for the next value, execution runs to yield i, hands back i, then freezes until the next request. So it never holds all the numbers in memory at once.
3. for x in count_to(3): drives the generator, pulling 0, then 1, then 2 — printed as they're produced ("lazily").
4. The last two lines show the real payoff: stream.text_stream is itself a generator, so looping over it gives you each token of the model's reply the instant it arrives.

What the output means: Prints 0, 1, 2 on separate lines. The key idea isn't the output — it's that values are made just-in-time, which is exactly how token streaming works.

Try this: Imagine count_to(1000000). A generator handles it with almost no memory because it only ever holds one number at a time; building a full list of a million items would not.

A decorator is the @name line written directly above a function or class. It wraps that function/class to add behavior, without you editing the function itself. You mainly need to recognize them here — the SDK uses them heavily.

1. @dataclass above class Tool: ... tells Python to auto-write boilerplate (like the __init__ that stores your fields) so you don't have to type it by hand.
2. @beta_tool above get_weather is the interesting one: it takes an ordinary function and turns it into a tool the model can call.
3. The magic is in the last comment — the decorator reads the function's signature and its docstring ("""Get weather for a location.""") to automatically build the description ("schema") the model needs. You write a normal function; the decorator does the wiring.

Try this: Read @beta_tool as "take the function below and make it a tool." The @ is not decoration in the visual sense — it's applying a wrapper. Recognizing that is the whole goal of this section.

Async lets one program juggle many slow I/O tasks (like API calls) at the same time, instead of standing idle waiting for each to finish. This course uses the simpler synchronous client, but you'll meet async when scaling a service — so learn to recognize its two keywords.

1. client = AsyncAnthropic() is the async version of the client. Its calls must be "awaited" rather than called normally.
2. async def ask(q): defines a coroutine — a special function that can pause and let other work run while it waits. You can only await inside an async def.
3. resp = await client.messages.create(...) — await means "start this, and while it's waiting on the network, let other coroutines run." That's how many calls overlap.
4. asyncio.gather(ask("a"), ask("b"), ask("c")) launches three calls together and waits for all of them — running concurrently instead of one after another.

What the output means: Functionally the answers are the same as the sync client; the win is time: three calls that each take a second finish in about one second total, not three.

Try this: Spot the two keywords — async def and await. Wherever you see them, the code is overlapping waits. For the labs you don't need to write this; just recognize it.

Never paste an API key directly into your code — it can leak into git or logs. Instead you keep it in a hidden .env file (which git ignores) and load it into the program's environment at startup. This block is the standard, safe setup.

1. load_dotenv() reads the key/value lines from your .env file and puts them into the environment, as if you'd set them in your terminal.
2. os.environ.get("ANTHROPIC_API_KEY") fetches that value by name. Using .get(...) returns None if it's missing rather than crashing.
3. client = Anthropic() — notice there's no key written in the code. The SDK looks up ANTHROPIC_API_KEY from the environment on its own.
4. if not key: is a fail-fast check: if the key wasn't found, raise SystemExit(...) stops the program with a friendly message telling you how to fix it, instead of a confusing error deeper in.

What the output means: If your key is set, the client is ready to use and nothing prints. If it's missing, the program exits immediately with the helpful message.

Try this: Look at how the key never appears as text in the file. That single habit — keys in .env, never in .py or prompts — is the most important security rule in the lesson.

This extends §1's Pydantic basics with the two things real schemas need: nesting (a model that contains other models) and custom validators (rules the type system alone can't express).

1. class Evidence(BaseModel): is a small model. Then in Diagnosis, the field evidence: list[Evidence] says "a list of Evidence objects" — a model nested inside another.
2. The @field_validator("evidence") block is a custom rule. After the basic type checks pass, this function runs on the evidence value; if not v: means "if the list is empty", and it raises an error demanding at least one item. return v passes the (approved) value through.
3. Building d = Diagnosis(...) from plain dictionaries shows Pydantic converting nested dicts like {"source":"logs", "line":"auth failed"} into real Evidence objects for you.
4. d.evidence[0].source reads a field of the first nested object — typed access all the way down. The last three lines convert the object back out: model_dump() → a dict, model_dump_json() → a JSON string, model_json_schema() → the schema the SDK sends the model.

What the output means: You get a fully-typed Diagnosis whose evidence is a list of real Evidence objects; the dump methods turn it into plain data for storage or sending.

Try this: A validator lets you enforce domain rules like "a plan must not target prod." It's a guardrail at the data layer — bad output is rejected before your code acts on it.

You can define your own exception types so callers can catch your specific failures distinctly — telling "the model refused" apart from "a policy blocked it" apart from "the network failed." Each can then be handled differently.

1. class RefusalError(Exception): creates a new error type by inheriting from the built-in Exception. That inheritance is all it takes to make a usable custom error.
2. Its __init__ stores the model's response on the error object (self.response = response) so a handler can inspect it later, and calls super().__init__(...) to set the human-readable message.
3. class PolicyBlocked(Exception): shows you don't even need a body — a docstring alone is enough to define a distinct type.
4. You raise RefusalError(resp) where the condition happens, and callers use except RefusalError as e: to catch that exact case — reading e.response to log details. A separate except PolicyBlocked: handles the other case its own way.

What the output means: No output on its own — this is machinery. The value is that different failures become separately catchable, so your program can respond appropriately to each.

Try this: Notice RefusalError and PolicyBlocked are just classes that extend Exception. Custom exceptions are one of the simplest, highest-value patterns for readable error handling.

APIs sometimes fail temporarily — a rate limit (429) or a server hiccup (5xx). The professional fix is to retry, but wait longer between each attempt ("exponential backoff") plus a little randomness ("jitter") so many clients don't all retry in lockstep. This one function ties together most of the course's Python.

1. call_with_retry(fn, *args, max_retries=5, ...) takes the function to call plus its arguments. *args/**kwargs mean "accept whatever arguments and pass them straight through" to fn.
2. The for attempt in range(max_retries): loop tries up to 5 times. Inside, return fn(*args, **kwargs) attempts the call — and if it succeeds, return exits immediately with the result.
3. A RateLimitError is caught and remembered (last = e) so we can retry. But in the APIStatusError block, if e.status_code < 500: raise — a 4xx like "bad request" is your fault and won't fix itself, so don't waste retries; re-raise now.
4. delay = min(base * (2 ** attempt) + random.uniform(0, 1), cap) doubles the wait each round (2 ** attempt → 1, 2, 4, 8…), adds up to 1 second of jitter, and min(..., cap) keeps it from growing past 60s. time.sleep(delay) waits, then the loop tries again. If all attempts fail, raise last gives up with the last error.

What the output means: On a transient failure you'd see lines like retry 1/5 in 1.3s, with the delay roughly doubling each time, until it either succeeds or exhausts all retries and re-raises.

Try this: Trace the delays for attempts 0–3 with base=1.0: about 1s, 2s, 4s, 8s (plus jitter). The SDK does this for you automatically — now you know exactly what its max_retries setting controls.

§3 and §5 showed how to use with and @decorators. Here you write your own — a timer context manager and a logging decorator. Both are tiny and extremely useful.

1. @contextmanager above def timer(label): lets you build a with-block from a function. Everything before yield is setup (record the start time); the yield is where your with-block's body runs; everything after runs as cleanup.
2. The try/finally matters: finally runs no matter what — so the timing print fires even if the code inside the with raises an error. That's the guarantee context managers give.
3. Using it, with timer("llm call"): automatically prints how many milliseconds the wrapped API call took.
4. The second half writes a decorator: def logged(fn): takes a function and returns a new wrapper that prints the call, runs the real function (result = fn(*args, **kwargs)), prints the result, then returns it. @wraps(fn) keeps the original function's name and docstring intact. Writing @logged above scale swaps in that wrapper.

What the output means: with timer("llm call"): prints something like llm call: 812ms. Calling scale("web", 3) prints a -> line before and a <- line after, showing the arguments and the return value.

Try this: Notice a context manager uses yield (like a generator) to split setup from teardown, while a decorator returns a replacement function. Both are ways to wrap behavior around your code.

§6 introduced async; this is the part that saves real time: running many API calls at once but capping how many run simultaneously so you don't trip the rate limit. This is how you'd grade a 200-case eval suite in a fraction of the wall-clock time.

1. sem = asyncio.Semaphore(5) creates a counter that allows at most 5 tasks through at a time. Think of it as 5 tickets: a task must hold a ticket to run, and returns it when done.
2. async with sem: is that ticket in action — a task waits here until a slot is free, then holds it for the duration of its call. (It's a context manager, so the slot is always released, even on error.)
3. grade_one(case) makes one awaited API call while holding a slot. grade_all(cases) builds one grade_one(c) per case and hands them all to asyncio.gather(...), which runs them concurrently and returns results in order.
4. asyncio.run(grade_all(golden_set)) is the entry point that actually starts the async machinery and runs everything to completion.

What the output means: You get a list of results for every case, but they complete far faster than one-at-a-time — with never more than 5 requests hitting the API simultaneously.

Try this: Remove the Semaphore in your head: all 200 calls would fire at once and you'd hit a 429 rate-limit error. The async with sem line is what keeps concurrency safe.

In real programs you use the logging module instead of print. Logging gives you severity levels, timestamps, and structured output you can filter and ship to a dashboard — the foundation of observability.

1. logging.basicConfig(level=logging.INFO, format=...) configures logging once: show everything at INFO level and above, and prefix each line with a timestamp and level.
2. log = logging.getLogger("agent") gets a named logger so you can tell this component's messages apart from others'.
3. The three calls show the levels: log.info(...) for normal events, log.warning(...) for something notable, log.error(..., exc_info=True) which also attaches the full error traceback. Note log.warning("pod %s restarting", pod_name) — the %s placeholder is filled in only if the message is actually logged ("lazy" formatting, slightly faster than building the string yourself).
4. The final call logs a structured event: json.dumps({...}) writes one JSON object per event, so later you can query logs like data ("show every tool_call that wasn't allowed").

What the output means: Each line comes out prefixed with a timestamp and level, e.g. 2026-09-06 10:00:00 INFO starting diagnosis. The JSON line is a single machine-readable record.

Try this: Compare to print: logging lets you dial the level up or down without deleting lines, and the JSON form is what makes production logs searchable.

A classic senior-Python interview question: write a @retry decorator you can configure (how many times, how long to wait). A configurable decorator needs three nested layers — that structure is exactly what interviewers are checking for.

1. The outer retry(times=3, delay=1.0) takes the settings and returns the real decorator. This layer exists only so you can write @retry(times=5) with arguments.
2. The middle deco(fn) is the actual decorator: it receives the function being decorated and returns a replacement.
3. The inner wrapper(*a, **kw) is what runs when you call the decorated function. Its loop tries fn(*a, **kw) up to times; on any Exception, if attempt == times - 1: raise re-raises on the last attempt (give up), otherwise it sleeps and retries.
4. time.sleep(delay * 2 ** attempt) is the backoff — the wait doubles each attempt (1s, 2s, 4s…). @functools.wraps(fn) keeps the wrapped function's real name and docstring so it isn't disguised as wrapper.

What the output means: Used as @retry(times=5) above a flaky function, calls that fail get retried automatically with growing pauses, and only the final failure propagates.

Try this: Count the defs: three, one inside the next. The rule of thumb — "a decorator that takes arguments has one extra layer than one that doesn't" — is the insight the question tests.

"Process a 10GB file" is a favorite interview trap: if you read the whole file into memory it won't fit. The answer is a generator that yields one line at a time, so memory stays tiny no matter how big the file is.

1. with open(path) as f: opens the file safely (auto-closed at the end — the context manager from §3).
2. for line in f: is the key trick: iterating a file object reads it one line at a time instead of loading it all. That's the "O(1) memory" comment — memory used stays constant regardless of file size.
3. if "ERROR" in line: keeps only the lines you care about, and yield line.rstrip() hands each matching line back to the caller (rstrip() trims the trailing newline).
4. Because it's a generator, the caller can loop over error_lines(path) and process results as they stream out — the 10GB file is never fully in memory at once.

What the output means: You get an iterator that produces just the ERROR lines, one by one — you could count them or write them elsewhere while barely using any memory.

Try this: Contrast with lines = open(path).read().splitlines(), which loads everything first. The generator version is the same idea as token streaming: produce values lazily, hold almost nothing.