Expert Python

A type hint is a note you write next to a value saying what kind of thing it should be (a str, an int, a list, …). Python does not enforce hints while running, but a checker tool (mypy/Pyright) reads them and warns you before you run — catching typos and wrong shapes early. This block shows four power-tools for typing agent code.

1. TypedDict describes a dictionary whose keys are known ahead of time. Message says every message dict must have a role (only one of the three listed words) and a content string. The checker flags a misspelled key like "rol".
2. Generic + TypeVar makes a reusable container that works for any inner type. T is a placeholder; EvalResult[str] means "an EvalResult whose output is a string" — same class, filled in with a concrete type.
3. Protocol is "duck typing, checked." It says "I accept any object that has a retrieve(query) method" — the object does not need to inherit from anything. So answer() works with any retriever that has the right method shape.
4. Self is the return type for a method that hands back its own object, so calls can be chained (builder.add(...).add(...)) — a "fluent" style. The ... (ellipsis) is a real placeholder meaning "body filled in elsewhere."

Try this: Install a checker (pip install mypy) and run mypy yourfile.py. Change a message's "role" to "boss" and watch mypy complain — that error is the type system earning its keep before the code ever runs.

functools is a standard-library toolbox for working with functions. These four helpers each replace a chunk of hand-written code. A key word here is pure function: one that always returns the same output for the same input and changes nothing else — safe to remember.

1. @cache is a decorator (the @ line above a function wraps it with extra behavior). It memoizes: remembers past results so calling embed_once("hi") twice computes only once and returns the stored answer the second time.
2. partial makes a new function with some arguments pre-filled. ask_opus is client.messages.create with model and max_tokens already set, so you only pass messages each time — less repetition.
3. reduce boils a whole list down to one value by combining items pairwise. Here it sums [0.2, 0.5, 0.9] starting from 0.0, giving 1.6.
4. singledispatch lets one function name do different things depending on the type of its first argument. The base render handles anything; the @render.register version runs only when a dict is passed.

Try this: Add print(embed_once.cache_info()) after calling embed_once twice with the same text — you'll see hits=1, proof the second call was served from memory.

itertools builds lazy iterators — sequences produced one item at a time, on demand, instead of all at once in memory. That is ideal for streaming model output or huge lists. An iterable is anything you can loop over (a list, a file, a live stream).

1. batched(documents, 20) hands you the items in groups of 20 — perfect for sending documents to a batch API a chunk at a time. (Needs Python 3.12+.)
2. islice takes just the first N items from an iterator, even an endless streaming one. Here islice(stream.text_stream, 5) grabs only the first 5 chunks and stops.
3. chain glues several lists into one sequence without building a big combined list in memory — it just walks the first, then the next.
4. groupby groups neighbouring equal items. It only groups items that are already next to each other, so you sort by the same key first to gather every service together.

What the output means: After sorting and grouping, each printed line is a service name and the count of its log lines, e.g. auth 12.

Try this: Comment out the logs.sort(...) line and re-run: groupby will split the same service into several small groups whenever its rows are not already adjacent — that shows why the sort matters.

Pydantic turns loose data (like JSON from a model) into checked, typed objects — if a field is missing or the wrong type, it raises a clear error instead of failing later. This block shows the advanced pieces you need when an agent must return one of several shapes.

1. A discriminated union means "the value is exactly one of these models, and one field tells us which." Answer and Escalate both have a kind field with a fixed value; Field(discriminator="kind") tells Pydantic to read kind and pick the matching model.
2. model_validate({...}) takes a plain dict and returns a real, typed object. Because kind was "escalate", d.action is an Escalate instance — so isinstance(d.action, Escalate) is True.
3. Annotated constrained types bundle a type with rules. Confidence is a float that must sit between 0 and 1 (ge=0, le=1); reuse it anywhere instead of re-writing the check.
4. model_dump turns the object back into a dict. exclude_none=True drops empty fields; mode="json" converts values (like dates) into JSON-safe text.

Try this: Change the dict to {"kind": "escalate"} (drop reason) and re-run. Pydantic raises a validation error naming the missing field — that guard is exactly why you use it on model output.

Concurrency means doing many slow things (like API calls) at once instead of one after another, so a batch finishes in the time of the slowest call. async/await is Python's way to do this on one thread: await pauses a task while it waits and lets others run. This shows the modern, correct way to fan out agent calls.

1. Semaphore(5) is a permit counter. async with sem: takes one of 5 permits and returns it at the end — so at most 5 calls run at once. That respects the API's rate limit even if you launch hundreds of tasks.
2. async with asyncio.timeout(60): gives each call 60 seconds; if it hangs longer it is cancelled and raises — so one stuck call can't freeze the whole run.
3. TaskGroup (Python 3.11+) starts many tasks with tg.create_task(...) and, at the end of its async with, waits for all of them. Key behavior: if any task fails, it cancels the rest and re-raises — "all or nothing."
4. After the group finishes, [t.result() for t in tasks] collects every task's return value into a list.

What the output means: Nothing prints — these are just the definitions. Calling grade_all(cases) would run up to 5 grading calls at a time and return the list of responses.

Try this: Lower the semaphore to Semaphore(1) and you've serialized everything (one at a time); raise it and calls overlap. That one number is your concurrency dial.

Dunder methods ("double-underscore," e.g. __len__) are special hooks Python calls for you when you use built-in syntax on your object. Define them and your own class starts acting like a list, a function, or a with-block.

1. In Transcript, defining __len__ makes len(transcript) work, __iter__ makes for m in transcript work, __getitem__ enables transcript[-1], and __bool__ decides what if transcript: means. Your object now feels built-in.
2. __call__ makes an instance callable like a function. After t = Tool(fn, risk), writing t(**args) runs fn — a "function object" that also carries data (its risk level).
3. __enter__/__exit__ let an object be used in a with block. Timer records the start time on enter and prints the elapsed seconds on exit — so with Timer(): times whatever runs inside.

Try this: Wrap a slow line in with Timer(): and run it — the elapsed time prints automatically when the block ends, because Python calls __exit__ for you.

Threading a value like a trace_id or tenant through every function call is tedious. contextvars gives you a variable that is set once and readable anywhere in the same request — and, crucially, kept separate per concurrent task, so one request never sees another's value.

1. ContextVar("trace_id", default="-") creates the request-scoped variable with a fallback used before anything is set.
2. handle_request calls .set(...) once at the top to stamp this request with a fresh unique id (uuid4()) and its tenant. Deeper code doesn't need these passed in.
3. log() — however deep in the call stack — reads the current values with .get(), so every log line is automatically tagged with the right request's id and tenant.

Try this: Picture two requests running at once: because each task has its own context, request A's log() prints A's id and request B's prints B's — no mix-up. Try replacing the ContextVar with a plain global and you'd get exactly that bug.

A mock is a fake stand-in object you drop in during a test. Real LLM calls are slow, cost money, and change their wording — awful for automated tests. So you replace the client with a mock that instantly returns a canned response, and test your code (the loop, parsing, error handling) with no network at all.

1. fake_response(text) builds a lightweight object shaped like the SDK's real reply. SimpleNamespace just makes an object with the attributes you name (.content, .stop_reason, …) so your code can read them the same way.
2. eng.client = MagicMock() swaps the real client for a fake. ...create.return_value = fake_response("Paris") says "whenever the code calls create, hand back this fake message."
3. eng.run(...) then exercises your real logic against the fake — no API key, no network, instant and identical every time.
4. assert_called_once() checks your code called the model exactly once — a test of behavior, not just output. The second test feeds a "refusal" reply to prove your refusal branch is handled.

Try this: To test a full tool-using loop, set create.side_effect = [tool_use_resp, end_turn_resp] so the mock returns a tool call first and a final answer second — now you can assert the agent ran the tool and then stopped, all without the model.

The worked example combines the page's patterns in one tiny file: a Protocol (structural interface), a plain class that satisfies it, @cache, and a dunder — a component you could swap and test easily.

1. @runtime_checkable class Retriever(Protocol) defines the contract: "anything with a retrieve(query, k) method is a Retriever." runtime_checkable lets you verify it at run time with isinstance.
2. KeywordRetriever holds some docs. __len__ makes len(retriever) report the doc count. @cache on retrieve remembers answers so an identical query is instant the second time.
3. The ranking line scores each doc by how many query words it contains and sorts best-first (the leading - sorts high-to-low). It returns a tuple, not a list, because @cache can only store hashable (unchangeable) results — a tuple is; a list isn't.
4. answer_with(store, q) takes any Retriever, so you could pass the Ch 3 VectorStore instead with no change here. The __main__ block builds one, asserts it structurally counts as a Retriever, and queries it twice.

What the output means: It prints 3 docs, then the top hit reset your password twice — the second identical query is served from the @cache, not recomputed.

Try this: Add a 4th doc and a different query, and add print(KeywordRetriever.retrieve.cache_info()) at the end to see cache hits vs misses.

A classic interview question: "how do you unit-test code that calls an LLM?" The expected answer is dependency injection + mocking — pass the client in as an argument so a test can hand you a fake one.

1. summarize(client, text) takes the client as a parameter instead of creating its own. That single design choice is what makes it testable — you can supply any client.
2. In the test, fake = MagicMock() is a stand-in that auto-creates any attribute you touch. The next line pre-programs its reply: create returns an object whose content[0].text is " hi ".
3. assert summarize(fake, "x") == "hi" checks your .strip() trimmed the spaces — testing your logic, with no network and no API key.

Try this: In an interview, say this out loud: "I inject the client so I can pass a mock; the test asserts on my parsing, not the model's wording." That framing is what they're listening for.

Another common question: "run many async tasks, but never more than K at the same time." The answer is a semaphore — a counter of permits that caps how many workers run concurrently.

1. asyncio.Semaphore(limit) creates limit permits (default 10).
2. Each worker does async with sem: — it waits for a free permit, runs await fn(x), then releases the permit. So no more than limit workers are ever inside at once.
3. asyncio.gather(*[worker(i) for i in items]) launches a worker per item and waits for all results, returned in the original order.

Try this: Set limit=2 and give fn a print plus await asyncio.sleep(1). You'll see items processed two at a time — visible proof the semaphore is bounding the parallelism.