Expert Python

Who this is for optional

You do not need P5 to complete the course or the projects — P1–P4 are enough to read and write every lab. Reach for P5 when you're building something real and want it maintainable, fast, and well-tested: a production support bot, the DevOps agent for a real team, or a project you'll extend for months. The testing section (§8) is the highest-value part for anyone shipping agents — read that even if you skip the rest.

1 · Advanced typing — precise contracts expert

Here's a bug you've almost certainly shipped. Somewhere deep in an agent, a function expected a dict shaped like {"role": "user", "content": "..."}. One day a caller passed {"role": "usr", ...} — a one-letter typo — and nothing complained. The code ran, the API rejected the malformed message twelve stack-frames later, and the traceback pointed at the SDK, not at your typo. You spent an hour finding a mistake a machine could have caught in a millisecond. That millisecond is what precise typing buys you.

The instinct most people have is that Python "doesn't have types," so hints are decoration. That's the mistake. Python doesn't enforce hints at runtime, but a static checker (mypy or Pyright) reads them and fails your build the moment a shape is wrong — before a single line runs, before CI, before a user. Think of type hints as a contract you write down: this function promises to receive X and return Y. The checker is the auditor who reads every call site and tells you where the contract is broken. In a small script you can hold the shapes in your head; in an agent codebase with retrievers, tool results, message lists, and streamed content blocks all flowing through dozens of functions, you can't — and that's exactly where the silent-typo bugs breed.

Beyond the basic hints from P3, four tools do the heavy lifting for agent code, and each answers a real question. How do I describe a dict with a fixed set of keys (a message, a tool_result)? TypedDict. How do I write a container that works for any inner type (an eval result holding a str today, a dataclass tomorrow)? Generics with TypeVar. How do I accept "any object that has a retrieve method" without forcing everyone to inherit a base class? Protocol — structural typing, the Pythonic answer to "duck typing, but checked." How do I type a method that returns its own class for chaining? Self.

The common mistake here is over-typing: plastering Any everywhere (which silently switches the checker off) or, at the other extreme, building baroque generic hierarchies nobody can read. The goal is not maximal types — it's precise contracts at the seams, the boundaries where one component hands data to another. Type the message that crosses from your code into the SDK; type the retriever your agent depends on; type the shape a tool returns. Those are the joints where a wrong shape does the most damage, and where a checker earns its keep. Beyond the hints in P3, precise types catch bugs before runtime and make an agent codebase navigable.

pythonfrom typing import TypeVar, Generic, Protocol, TypedDict, overload, Self, Literal

# TypedDict: a dict with a known shape (great for message/tool_result dicts)
class Message(TypedDict):
    role: Literal["user", "assistant", "system"]
    content: str
msgs: list[Message] = [{"role": "user", "content": "hi"}]   # type-checked keys

# Generics: a typed, reusable container (e.g. an eval result for any output type)
T = TypeVar("T")
class EvalResult(Generic[T]):
    def __init__(self, output: T, passed: bool):
        self.output, self.passed = output, passed
r: EvalResult[str] = EvalResult("answer", True)

# Protocol: structural typing — "anything with a .retrieve(str) method"
class Retriever(Protocol):
    def retrieve(self, query: str) -> list[str]: ...
def answer(q: str, store: Retriever): ...   # accepts ANY object with retrieve()

# Self: a method that returns its own class (fluent builders)
class PromptBuilder:
    def add(self, text: str) -> Self:
        ...; return self

2 · functools — caching, partials, dispatch expert

Picture the RAG index you built in Chapter 3, running in production. Users ask the same handful of questions all day — "how do I reset my password," "what's the refund policy" — and every single time, your code dutifully re-embeds the identical query string, paying the same milliseconds and, if you're calling an embedding API, the same fractions of a cent. Multiply by ten thousand requests and you've burned real money and latency computing an answer you already had. functools is the standard-library toolbox that makes that waste disappear with one decorator — and it does three other things you'd otherwise hand-roll badly.

The unifying idea is that functools works on functions themselves — it treats a function as a value you can wrap, specialize, or dispatch on. That's a genuinely different mental model from "a function is just a thing I call," and it's worth pausing on, because once it clicks you stop writing so much glue code. The clearest analogy is a coffee machine with presets: partial is you programming the "always a large oat-milk latte" button so you press one key instead of dialing in three settings every morning; @cache is the machine remembering it already ground beans for this exact order and handing you the stored cup; singledispatch is one "make drink" button that behaves differently depending on whether you feed it beans, pods, or instant.

The four tools map to four everyday agent needs. @cache / @lru_cache memoize a pure function — one that always returns the same output for the same input and touches nothing else — so a repeated call is a dictionary lookup instead of a recomputation (embedding a fixed string is the textbook case). partial pre-fills arguments to make a specialized function, so client.messages.create(model=..., max_tokens=...) becomes a tidy ask() you call with just the messages. reduce folds a whole sequence down to one value. singledispatch lets one function name behave differently per argument type — perfect for rendering the different content-block shapes (text, tool_use, thinking) you meet in every response.

The mistake that bites people — and it bites hard — is caching something that isn't pure. @cache keys on the arguments and never evicts, so caching a per-user or time-sensitive result means user B silently receives user A's answer, or a stale value lives forever. The rule: cache deterministic, input-only functions; bound the size with @lru_cache(maxsize=N) when the key space is large; and never, ever cache across users when the result is user-specific. Get that right and @cache is the single cheapest speed-up on this page.

pythonfrom functools import lru_cache, cache, partial, reduce, singledispatch

# @cache: memoize a pure function — e.g. don't re-embed the same text
@cache
def embed_once(text: str) -> tuple:
    return expensive_embed(text)     # computed once per unique text

# partial: pre-fill arguments to make a specialized function
from anthropic import Anthropic
client = Anthropic()
ask_opus = partial(client.messages.create, model="claude-opus-4-8", max_tokens=1024)
ask_opus(messages=[...])          # model/max_tokens already set

# reduce: fold a sequence (e.g. combine per-doc scores)
total = reduce(lambda acc, x: acc + x, [0.2, 0.5, 0.9], 0.0)

# singledispatch: one function, different behavior per argument type
@singledispatch
def render(block): return str(block)
@render.register
def _(block: dict): return block.get("text", "")   # handle dict blocks specially

3 · itertools — streaming-friendly iteration expert

A story that ends in an out-of-memory crash. You need to embed a million document chunks. The obvious code — chunks = list(load_all_documents()) then loop — works perfectly on your test folder of fifty files and dies on the real corpus, because it tried to hold a million chunks in RAM at once before doing any work. The fix isn't a bigger machine; it's to stop materializing the whole thing. itertools is the standard library's kit for exactly that: building lazy iterators that produce one item at a time, on demand, so memory stays flat no matter how large — or how infinite — the source is.

The mental shift is from "a list is a box holding all my items" to "an iterator is a tap I open when I want the next item." A list eagerly computes and stores everything; an iterator is a recipe that yields the next value only when asked. This is precisely the right shape for two things you do constantly with models: consuming a streamed response token-by-token as it arrives (you don't have the whole thing yet, and you don't want to wait for it), and walking a huge collection you can't fit in memory. Think of a conveyor belt versus a warehouse: itertools gives you the belt.

Four of its tools cover most agent pipelines. batched(iterable, n) (Python 3.12+) hands you the items in fixed-size groups — the natural way to feed documents to a Batch API twenty at a time without loading them all. islice takes just the first N items from an iterator, even an endless streaming one, so you can peek at or cap a live response. chain glues several sequences into one walk-through without building a combined list. groupby collapses runs of equal items into groups — invaluable for tallying logs by service.

The classic trap is groupby, and nearly everyone hits it once. It groups only consecutive equal items — it is not SQL's GROUP BY. If your rows aren't already adjacent, you get several tiny fragmented groups for the same key instead of one. The fix is a one-liner most people forget: sort by the same key first, so all the matching rows sit together, then group. A second, subtler trap: an iterator is exhausted after one pass. Once you've walked it, it's empty — if you need the data twice, keep a list or use itertools.tee. Respect those two rules and lazy iteration turns memory-bound code into streaming code that scales.

pythonfrom itertools import islice, chain, groupby, batched

# batched (3.12+): chunk an iterable into fixed-size groups
# e.g. send documents to a batch API in groups of 20
for group in batched(documents, 20):
    submit_batch(group)

# islice: take the first N from a (possibly infinite/streaming) iterator
first5 = list(islice(stream.text_stream, 5))   # first 5 chunks only

# chain: flatten several lists lazily (no intermediate list)
all_findings = list(chain(review1.findings, review2.findings))

# groupby: group CONSECUTIVE items (sort first if you want all groups)
logs.sort(key=lambda l: l["service"])
for service, group in groupby(logs, key=lambda l: l["service"]):
    print(service, len(list(group)))

4 · Pydantic mastery — unions, custom types, settings expert

The single most fragile line in most agent codebases is the one right after the model responds. The model hands you back text, you json.loads it, and then you write data["action"]["reason"] — trusting that a probabilistic system produced exactly the shape you imagined. Sometimes it returns {"action": {"kind": "escalate"}} with no reason. Sometimes kind is "Escalate" with a capital E. Sometimes it's a whole different action you didn't handle. Your code doesn't crash there — it crashes three functions downstream with a KeyError that tells you nothing about the real cause. Pydantic exists to move that failure to the front door and make it loud, specific, and typed.

The core idea is a boundary guard. Loose, untrusted data — JSON from a model, a request body, a config file — comes in; Pydantic validates it against a schema and either hands you a real, typed object or raises a precise error naming the field that's wrong. Think of a bouncer with a guest list: nobody gets past the door in the wrong outfit, so everyone inside the party is guaranteed to be dressed correctly. Every function past that boundary can then trust its inputs completely, which is a huge simplification — the checks live in one place, not scattered as defensive ifs through the whole codebase. P4 covered the basics — models and field validators; this section covers the pieces that specifically make agents reliable.

The star of the show for agent code is the discriminated union. An agent's output is often "exactly one of several things": it either answered, or it escalated, or it called one of N tools. A discriminated union expresses precisely that — a set of models that share a "tag" field (here, kind) — and Field(discriminator="kind") tells Pydantic to read that tag and build the matching subtype. The payoff is that d.action comes back as a real Escalate instance you can branch on with isinstance and that your type checker understands — not an anonymous dict you index into and hope. Alongside it: annotated constrained types (a Confidence that must lie in [0, 1], defined once and reused) and serialization control (model_dump(mode="json") to get audit-log-safe output).

The mistake to avoid is treating Pydantic as mere documentation and then bypassing it — parsing the JSON yourself, or catching the ValidationError and pressing on with half-built data. The whole value is that a validation failure is a signal: the model gave you something you didn't expect, which is exactly the moment to retry, ask again, or escalate — not to limp forward on a malformed object. Let it raise, handle it at the boundary, and everything inside stays clean. Here's the rest of what makes Pydantic the backbone of reliable agents.

pythonfrom pydantic import BaseModel, Field
from typing import Literal, Annotated, Union

# Discriminated unions: the agent returns ONE OF several action types.
# Pydantic picks the right model by the 'kind' field — perfect for tool calls.
class Answer(BaseModel):
    kind: Literal["answer"]
    text: str
class Escalate(BaseModel):
    kind: Literal["escalate"]
    reason: str
Action = Annotated[Union[Answer, Escalate], Field(discriminator="kind")]

class Decision(BaseModel):
    action: Action                     # validated into the right subtype

d = Decision.model_validate({"action": {"kind": "escalate", "reason": "low conf"}})
isinstance(d.action, Escalate)         # True — typed, not a raw dict

# Constrained/annotated types — reusable field constraints
Confidence = Annotated[float, Field(ge=0, le=1)]
Severity = Literal["low", "medium", "high"]

# Serialization control
d.model_dump(exclude_none=True)         # drop null fields
d.model_dump(mode="json")              # JSON-safe types (datetime -> str)

5 · Structured concurrency — TaskGroup, timeouts, cancellation expert

You have fifty evaluation cases to grade, each a round-trip to the model that takes two seconds. Run them one after another and your test suite takes a hundred seconds — long enough that people stop running it. Run them all at once and you've fired fifty simultaneous requests, tripped the API's rate limit, and half of them fail. The right answer is neither serial nor a free-for-all: it's controlled concurrency — many calls in flight, but a firm cap, per-call timeouts so nothing hangs forever, and clean behavior when one fails. That's what this section is about, and modern Python (3.11+) finally makes it correct-by-default.

First, the intuition for why concurrency helps here at all. These calls are I/O-bound — your program spends almost all its time waiting for the network, not computing. async/await lets a single thread start a request, and while it's waiting on the reply, await hands control to another task to start its request. Fifty tasks all wait at the same time, so the batch finishes in roughly the time of the slowest single call rather than the sum of all of them. The analogy is a chef with fifty pots: you don't stand watching one pot boil (serial) — you get every pot going and tend whichever is ready, because the bottleneck is boiling time, not your hands.

Two tools tame the chaos. A Semaphore(K) is a bowl of K permits: async with sem: takes one and returns it at the end, so no matter how many tasks you launch, at most K are ever running — that's your rate-limit dial. asyncio.timeout(s) wraps a call so a hung request is cancelled after s seconds instead of freezing the whole run. And TaskGroup (3.11+) is the structured way to launch a batch and wait for all of it: if any task raises, it cancels the siblings and re-raises, so you never leak a half-finished task into the background.

The decision that trips people is TaskGroup versus gather, and they encode opposite intentions. TaskGroup is fail-fast: one error aborts the whole group — right when "all must succeed or the result is meaningless." gather(..., return_exceptions=True) is resilient: it runs everything to completion and hands you the failures alongside the successes — right for an eval run where one bad case shouldn't kill the other forty-nine. The classic bug is reaching for whichever you learned first without asking which semantics you actually want. And the mistake beginners make regardless is launching an unbounded fan-out with no semaphore and no timeout — it works in a five-item test and melts the moment real volume arrives. P4 showed gather; this is the disciplined, production-grade way to fan out agent calls.

pythonimport asyncio
from anthropic import AsyncAnthropic
client = AsyncAnthropic()
sem = asyncio.Semaphore(5)                # cap concurrency (rate limits)

async def grade(case):
    async with sem:
        # per-call timeout — don't let one hang forever
        async with asyncio.timeout(60):
            return await client.messages.create(model="claude-opus-4-8",
                max_tokens=200, messages=[{"role":"user","content":case}])

async def grade_all(cases):
    results = []
    async with asyncio.TaskGroup() as tg:     # 3.11+
        tasks = [tg.create_task(grade(c)) for c in cases]
    # TaskGroup waits for all; if ANY raises, it cancels the rest & re-raises
    return [t.result() for t in tasks]

6 · Protocols, dunder methods & making things swappable expert

Ever wondered why len(my_list) works but you write len(), not my_list.length()? Python's built-in functions and syntax — len(x), for m in x, x[-1], if x:, with x: — don't do the work themselves. They quietly call special methods on your object: __len__, __iter__, __getitem__, __bool__, __enter__/__exit__. These are the dunder methods ("double-underscore"), and they're the hooks that let your own classes plug into the language as if they were built-in types. Define them, and a conversation-history object stops being a clumsy wrapper you call history.get_messages() on and becomes something you can len(), loop over, and index like a native list.

The reason this matters for agents is swappability, and it pairs directly with the Protocol idea from §1. A well-built agent doesn't hard-code "the keyword retriever" — it depends on anything shaped like a retriever, so you can drop in the Chapter 3 vector store later without touching the caller. Dunders are what make those swappable components feel seamless: the retriever supports len(), the tool is callable like a function via __call__, the timer works in a with block. The interface is the contract; the dunders are the polish that make honoring the contract invisible to callers. The mental model is an electrical socket: as long as your device has the standard plug shape (the dunders), it works in the wall (Python's syntax) — the wall doesn't care what's inside.

Three dunders earn their place immediately. The collection dunders (__len__, __iter__, __getitem__, __bool__) make a custom object behave like a sequence — a Transcript you can measure, iterate, and slice. __call__ makes an instance callable, so a Tool object that also carries data (its risk level, its schema) can still be invoked like a plain function — a "function object." And the context-manager dunders __enter__/__exit__ let you build your own with-blocks: a Timer that starts on entry and prints elapsed time on exit, guaranteed to run even if the body raises.

The mistake to sidestep is dunder enthusiasm — implementing a dozen special methods because you can, turning a simple class into a puzzle where + and [] do surprising things. Dunders should make an object behave the way its real-world role suggests: a transcript is list-like, so give it list dunders; a tool is action-like, so give it __call__. Add the ones that match the mental model your callers already have, and stop. Used with that restraint, dunders plus Protocols are the whole toolkit for making the retriever, the store, and the tool set cleanly interchangeable across the projects.

pythonimport time

# dunder methods make your objects behave like built-ins
class Transcript:
    def __init__(self): self._msgs = []
    def __len__(self): return len(self._msgs)          # len(transcript)
    def __iter__(self): return iter(self._msgs)         # for m in transcript
    def __getitem__(self, i): return self._msgs[i]      # transcript[-1]
    def __bool__(self): return bool(self._msgs)        # if transcript:

# __call__ makes an instance callable — a configurable "function object"
class Tool:
    def __init__(self, fn, risk): self.fn, self.risk = fn, risk
    def __call__(self, **kw): return self.fn(**kw)         # tool(**args)

# context-manager dunders — build your own 'with' object
class Timer:
    def __enter__(self): self.t = time.monotonic(); return self
    def __exit__(self, *exc): print(f"{time.monotonic()-self.t:.2f}s")

7 · contextvars — request-scoped state expert

Here's the parameter that poisons agent codebases. You add tracing, so every log line needs a trace_id. The only way to get it into log() — buried six calls deep inside a tool — is to pass it down through every function in between: run_agent(msg, trace_id) → call_tool(args, trace_id) → fetch(url, trace_id). None of those functions use the id; they just ferry it along. Then you add a tenant, and every signature grows again. This "tramp data" — arguments threaded through functions that don't care about them — is a real smell, and contextvars is the clean cure: a value you set once at the top of a request and read anywhere below, without it appearing in a single signature between.

The obvious shortcut is a module-level global — and it's a landmine the moment you have concurrency. Picture two requests handled at once: request A sets the global tenant to "acme," then awaits a slow model call; while it waits, request B runs and overwrites the global to "globex"; A resumes and now logs — and, worse, filters tenant-scoped data — as "globex." That's a cross-tenant data leak born from one shared variable. The fix is that a ContextVar is not shared: each concurrent task (and each thread) gets its own isolated copy, automatically. The analogy is a hospital wristband — the patient's ID travels with them into every room and every test, and nobody confuses two patients because each wears their own band, not a shared clipboard on the wall.

Mechanically it's tiny: ContextVar("trace_id", default="-") declares the variable with a fallback; .set(value) stamps the current request near the entry point; .get() reads it from however deep in the call stack you are. The async runtime is what makes the isolation work — when a task awaits and another runs, the runtime swaps in that task's context, so each .get() sees its own request's value. You get the ergonomics of a global with none of the sharing.

The one caveat worth internalizing: contextvars is for genuinely request-scoped, cross-cutting state — observability ids, the authenticated tenant, a request-start clock — not a back door to smuggle ordinary business data around because passing it feels tedious. If a function's logic depends on a value, that value belongs in its signature where it's visible and testable; if the value is ambient plumbing that every layer merely tags along with, that's the contextvar sweet spot. And note the security angle for tenancy: because you set the tenant from the authenticated request — never from anything the model produced — the tenant-scoped tools filter correctly no matter what the model tries to claim. Passing a trace_id / tenant / user through every function is tedious; contextvars gives you request-scoped globals that are safe under async concurrency — the clean way to thread observability + tenancy through an agent.

class _Any:
    '''stands in for any undefined demo value; supports call/attr/index/
    iteration and basic arithmetic (as 0.7) so demo snippets run.'''
    def __call__(self, *a, **k): return _Any()
    def __getattr__(self, k): return _Any()
    def __getitem__(self, k): return _Any()
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __contains__(self, o): return True
    def __enter__(self, *a): return _Any()
    def __exit__(self, *a): return False
    def __float__(self): return 0.7
    def __int__(self): return 1
    def __lt__(self, o): return True
    def __gt__(self, o): return False
    def __le__(self, o): return True
    def __ge__(self, o): return False
    def __add__(self, o): return o
    def __radd__(self, o): return o
    def __bool__(self): return True
    def __repr__(self): return 'demo'
    def __str__(self): return 'demo'
def run_agent(*a, **k):  # demo stub
    return _Any()

pythonimport contextvars, uuid

trace_id = contextvars.ContextVar("trace_id", default="-")
tenant   = contextvars.ContextVar("tenant")

def handle_request(user_msg, tenant_id):
    trace_id.set(str(uuid.uuid4()))       # set once per request
    tenant.set(tenant_id)
    run_agent(user_msg)                    # deep calls can read it without passing it

def log(msg):
    # any function, however deep, sees the current request's values
    print(f"[{trace_id.get()}][{tenant.get()}] {msg}")

8 · Testing LLM code — mock the model essential for shipping

If you read one section on this page, read this one. Here's the wall every engineer hits the first time they try to test an agent: you write def test_agent(): assert agent.run("hi") == ... and immediately everything is wrong. The test costs money (a real API call). It's slow (a network round-trip in a suite that should run in milliseconds). And it's flaky — the model phrases things differently each run, so your assert passes today and fails tomorrow on wording alone. A test that costs money, crawls, and fails randomly is a test people delete. So agents go untested, and untested agents break in production.

The fix flips the problem on its head. You are not trying to test the model — the model isn't yours, and its quality is a separate question. You're trying to test your code around it: the loop that keeps calling until the model stops, the safety gate that blocks a dangerous tool, the parser that pulls text out of the response, the branch that handles a refusal. All of that is deterministic logic that has nothing to do with which words the model chose. So you mock the client: swap the real API object for a fake stand-in that instantly returns a canned response you control. Now the model-shaped hole in your code is filled by a puppet, and every test runs in a millisecond, for free, identically every time — safe for CI on every commit.

A mock is just a fake object you drop in during a test; Python's unittest.mock.MagicMock gives you one that auto-creates any attribute you touch and lets you pre-program return values. The technique that makes it work cleanly is dependency injection — designing the function to receive its client as an argument (summarize(client, text)) rather than reaching for a global one, so a test can hand it a mock. The real power move is side_effect = [resp1, resp2]: make the fake return a tool_use response first and an end_turn response second, and now you can drive a full tool-using loop through its paces — assert it called the right tool with the right arguments, then stopped — all without a single network call.

The mistake here is a mirror-image pair, and honest engineers name both. Under-mocking is letting real calls sneak into your unit tests — back to slow, costly, flaky. Over-trusting the mock is the subtler one: a green suite of mocked tests proves your plumbing is correct, but it says nothing about whether the model actually diagnoses the bug or answers the question well. Mocks can't test output quality — only real evals (Chapter 5), run less often and with a key, can. The professional stance is to hold both truths: mock the model to test your logic deterministically on every commit, and run evals to test quality on a cadence. This is how you test agents in CI.

test_agent_logic.pyfrom unittest.mock import MagicMock, patch
import types

def fake_response(text, stop="end_turn"):
    """Build an object shaped like the SDK's Message."""
    block = types.SimpleNamespace(type="text", text=text)
    return types.SimpleNamespace(content=[block], stop_reason=stop,
                                usage=types.SimpleNamespace(input_tokens=10,
                                    output_tokens=5, cache_read_input_tokens=0))

def test_agent_returns_text_without_calling_the_api():
    import agent.engine as eng
    # replace the real client with a mock that returns our fake response
    eng.client = MagicMock()
    eng.client.messages.create.return_value = fake_response("Paris")

    out = eng.run("capital of France?")          # runs YOUR loop, no network
    assert "Paris" in out
    eng.client.messages.create.assert_called_once()   # it made exactly one call

def test_agent_stops_on_refusal():
    import agent.engine as eng
    eng.client = MagicMock()
    eng.client.messages.create.return_value = fake_response("", stop="refusal")
    # assert your code handles the refusal branch correctly
    ...

Worked example · a typed, cached, testable retriever putting it together

Isolated tricks are easy to admire and hard to use — the real skill is making them cohere in one small, shippable component. Everything on this page has been a single tool held up to the light: a Protocol here, a cache there, a dunder in section six. That's how you learn them, but it's not how you meet them in real code, where a good component quietly wears several at once. This worked example is that convergence point: a retriever that is swappable (a Protocol, so the Chapter 3 vector store drops in with zero caller changes), fast (a @cache, so a repeated query is a lookup not a recompute), native-feeling (a __len__ dunder, so len(retriever) just works), and testable (a clean seam a mock can slot into).

Watch for the one detail that ties two ideas together and would silently bite you: @cache can only store hashable results, and a list is not hashable — a tuple is. So retrieve returns tuple(ranked[:k]), not the list you'd instinctively write. That's the kind of interaction you only feel when the patterns share a body: caching forces immutability, immutability changes your return type, and the type annotation follows. Read it as a template — the shape of a retriever you'd actually ship inside an agent — and notice how each expert pattern is pulling a specific, nameable weight.

typed_retriever.pyfrom typing import Protocol, runtime_checkable
from functools import cache

@runtime_checkable
class Retriever(Protocol):
    def retrieve(self, query: str, k: int = 3) -> list[str]: ...

class KeywordRetriever:
    def __init__(self, docs: list[str]):
        self._docs = docs
    def __len__(self) -> int:
        return len(self._docs)                # len(retriever)
    @cache                                     # memoize identical queries
    def retrieve(self, query: str, k: int = 3) -> tuple[str, ...]:
        q = set(query.lower().split())
        ranked = sorted(self._docs,
            key=lambda d: -sum(w in d.lower() for w in q))
        return tuple(ranked[:k])              # tuple so it's hashable/cacheable

def answer_with(store: Retriever, q: str) -> str:
    # works with ANY Retriever — swap in the Ch 3 VectorStore, no change here
    hits = store.retrieve(q)
    return f"top hit: {hits[0] if hits else '(none)'}"

if __name__ == "__main__":
    r = KeywordRetriever(["reset your password", "refund policy", "enable 2fa"])
    assert isinstance(r, Retriever)          # structural check passes
    print(len(r), "docs")
    print(answer_with(r, "how to reset password"))
    print(answer_with(r, "how to reset password"))   # 2nd call hits the cache

3 docs
top hit: reset your password
top hit: reset your password

Where to go beyond this expert

• Static analysis — run mypy/pyright in CI to enforce the types from §1.
• Property-based testing — hypothesis to fuzz your parsers/validators.
• Profiling — cProfile/py-spy when an agent pipeline is slow.
• Packaging — pyproject.toml, uv/poetry to ship your agent as a library.
• Observability libraries — OpenTelemetry for the tracing sketched in Ch 6.

🎯 Interview practice interview

The interview questions this topic gets asked — worked, with code. For the full pattern catalog see A9 · Big Tech AI-engineering patterns.

pythonfrom unittest.mock import MagicMock
def summarize(client, text):
    resp = client.messages.create(model="claude-opus-4-8",
        max_tokens=100, messages=[{"role":"user","content":text}])
    return resp.content[0].text.strip()

def test_summarize():
    fake = MagicMock()
    fake.messages.create.return_value.content = [MagicMock(text=" hi ")]
    assert summarize(fake, "x") == "hi"   # no real API call

pythonimport asyncio
async def bounded_map(fn, items, limit=10):
    sem = asyncio.Semaphore(limit)
    async def worker(x):
        async with sem:
            return await fn(x)
    return await asyncio.gather(*[worker(i) for i in items])

from typing import Protocol, runtime_checkable

@runtime_checkable
class Retriever(Protocol):
    def retrieve(self, query: str, k: int = 3) -> list[str]: ...

class Keyword:
    def retrieve(self, query, k=3):
        return [query][:k]

r = Keyword()
print(isinstance(r, Retriever))   # True -- matches structurally

from functools import cache

hits = {"n": 0}

@cache
def score(query: str) -> int:
    hits["n"] += 1
    return len(query.split())

print(score("reset my password"))   # computes
print(score("reset my password"))   # cached
print("real calls:", hits["n"])     # 1

from itertools import islice

def batched(iterable, n):
    it = iter(iterable)
    while (batch := tuple(islice(it, n))):
        yield batch

docs = range(1, 8)
for b in batched(docs, 3):
    print(b)   # (1,2,3) (4,5,6) (7,)

class KeywordRetriever:
    def __init__(self, docs):
        self._docs = docs
    def __len__(self):
        return len(self._docs)
    def __call__(self, query, k=3):
        q = set(query.lower().split())
        ranked = sorted(self._docs,
                        key=lambda d: -sum(w in d.lower() for w in q))
        return ranked[:k]

r = KeywordRetriever(["reset your password", "refund policy", "enable 2fa"])
print(len(r), "docs")
print(r("how to reset password"))   # called like a function

from unittest.mock import MagicMock

def answer(client, question):
    reply = client.respond(question)
    return reply.strip().upper()

fake = MagicMock()
fake.respond.return_value = "  reset it in settings  "

out = answer(fake, "how do I reset?")
assert out == "RESET IT IN SETTINGS"
fake.respond.assert_called_once_with("how do I reset?")
print("passed:", out)

from typing import Protocol, runtime_checkable
from functools import cache

@runtime_checkable
class Retriever(Protocol):
    def retrieve(self, query: str, k: int = 3) -> tuple: ...

class KeywordRetriever:
    def __init__(self, docs): self._docs = docs
    def __len__(self): return len(self._docs)
    @cache
    def retrieve(self, query, k=3):
        q = set(query.lower().split())
        ranked = sorted(self._docs, key=lambda d: -sum(w in d.lower() for w in q))
        return tuple(ranked[:k])

def answer_with(store: Retriever, q: str) -> str:
    hits = store.retrieve(q)
    return f"top hit: {hits[0] if hits else '(none)'}"

r = KeywordRetriever(["reset your password", "refund policy", "enable 2fa"])
assert isinstance(r, Retriever)
print(answer_with(r, "how to reset password"))
print(answer_with(r, "how to reset password"))   # 2nd call cached

Knowledge check check yourself