Validation, Structured Outputs & Resiliency

This defines the shape of a piece of data once, and Pydantic then checks any incoming values against it for free. A BaseModel subclass is like a form with typed, rule-checked fields — perfect for the messy JSON an LLM hands back.

1. Each line inside the class is a field: a name, its type, and optional rules. summary: str = Field(min_length=1, max_length=500) says "a string, 1–500 characters".
2. severity: Literal["low", "medium", "high", "critical"] restricts the value to exactly those four words — anything else is rejected. confidence must be a number between 0 and 1 (ge=0.0, le=1.0 = greater-or-equal / less-or-equal).
3. affected: list[str] = [] gives a default (an empty list), so the field is optional.
4. Creating Diagnosis(...) runs all the checks at once. d.model_dump() turns the validated object back into a plain dictionary you can send over the wire.

What the output means: A clean dictionary of the fields. The commented-out bad example would instead raise a ValidationError that names every bad field at once (empty summary, invalid severity, confidence out of range) — not just the first.

Try this: Change severity="high" to "urgent" and run it. Pydantic rejects it and tells you the allowed values — that rejection is the whole point: bad data never gets past the boundary.

Types alone can't express every rule. Validators are small methods you attach to a model to clean values or enforce custom logic. A field validator checks one field; a model validator checks how fields relate to each other.

1. The @field_validator("service") method runs on the service value and return v.strip().lower() normalizes it — trimming spaces and lowercasing — so " Web " becomes "web". Whatever you return replaces the stored value.
2. The non_negative validator raises a ValueError when replicas is below zero. Raising inside a validator is how you reject bad input.
3. @model_validator(mode="after") runs once all fields exist, so it can compare them: here it checks replicas against max_replicas — a cross-field rule a single field couldn't do alone.

What the output means: The print shows 'web' — the raw " Web " after the field validator cleaned it. Passing replicas=99 would trip the model validator and raise "replicas exceeds max_replicas".

Try this: Call ScaleAction(service="api", replicas=-1) and read the error. Then try replicas=50 to see the cross-field check fire instead.

An agent's tool call could be any of several shapes (scale, restart, rollback), each with different fields. A discriminated union uses one tag field to tell Pydantic which model to validate against — so "figure out what this is" becomes a clean, typed dispatch.

1. Each class has an action field pinned to a single literal value (e.g. Literal["scale"]). That field is the discriminator — the tag Pydantic reads first to pick the right model.
2. Action = Annotated[Union[Scale, Restart, Rollback], Field(discriminator="action")] bundles the three shapes and tells Pydantic: look at action to decide which one.
3. TypeAdapter(Action) wraps that union so you can validate raw data against it. adapter.validate_python({...}) reads "action": "rollback" and validates the dict as a Rollback, requiring to_version.

What the output means: Rollback v1.2 — the dict was parsed into the correct typed object. A dict whose fields don't match its tag (e.g. a "rollback" missing to_version) raises a ValidationError naming the missing field.

Try this: Feed it {"action": "scale", "service": "api", "replicas": 3} and check type(a).__name__ is now Scale. This one pattern is how you turn LLM JSON into safe, typed actions.

The most reliable way to get structured data from a model is to bind it to a schema and validate the result. If validation fails, you feed the error back and ask again. This wrapper does that retry loop by hand for cases where the SDK can't do it for you.

1. The commented-out top shows the preferred path: client.messages.parse(..., output_format=Diagnosis) lets the SDK parse straight into your model. Use that when you can.
2. parse_with_retry loops up to tries times. On the first pass it sends the plain prompt; on later passes it appends the previous error so the model can correct itself.
3. schema.model_validate_json(raw) tries to validate the model's raw text. If it works, we return immediately — success.
4. If it raises ValidationError, we save the message in err and loop. After all tries fail, we raise ValueError so the caller knows the model never complied.

What the output means: On success you get a fully validated schema object. Feeding the error back is the key trick — the model usually fixes its own mistake on the second attempt.

Try this: Write a fake call_model that returns broken JSON once, then valid JSON, and confirm parse_with_retry succeeds on the second try.

Resiliency starts by sorting errors into buckets. Some are transient (retrying helps); others are permanent (retrying just wastes time and money). This tiny classifier turns an HTTP status code into one of two custom exceptions so the rest of your code knows what to do.

1. Retryable and Fatal are custom exception classes (the ... body just means "empty"). Their type carries the decision.
2. If status is one of the transient codes (429 rate-limit, 500/502/503, 529 overloaded), it raises Retryable — a signal that backing off and trying again may work.
3. The check if 400 <= status < 500 catches the other 4xx client errors (bad request, auth, not-found) and raises Fatal — retrying can't fix a broken request, so fail fast.

What the output means: Nothing prints; the point is which exception type comes out. Downstream code catches Retryable to back off and re-try, and lets Fatal bubble up unchanged.

Try this: Call classify(429) and classify(404) and note how the exception type changes. That type is what the retry loop in the next section keys off of.

This is the production retry loop. It re-tries only errors marked Retryable, and it spaces attempts out exponentially (1s, 2s, 4s…) with jitter (randomness) so a crowd of clients doesn't all retry at the same instant and re-crush a recovering server.

1. The for attempt in range(tries) loop tries fn() and returns its result the moment it succeeds.
2. except Fatal: raise re-throws permanent errors immediately — never waste retries on something that can't succeed.
3. except Retryable handles the transient case. If this was the last attempt (attempt == tries - 1) it gives up and re-raises; otherwise it waits and loops.
4. delay = min(cap, base * (2 ** attempt)) doubles the wait each round but caps it. time.sleep(random.uniform(0, delay)) picks a random point up to that delay — that randomness is the jitter that prevents the "thundering herd".

What the output means: No visible output on its own; it either returns fn()'s result or, after enough transient failures, re-raises the last error. The pauses grow and are randomized between runs.

Try this: Set base=0.1 and print attempt each loop while raising Retryable, to watch the backoff grow. Real code should also honor a Retry-After header when the server sends one.

When a downstream service is truly down, even smart retries pile on load. A circuit breaker watches for consecutive failures and, after too many, "opens" — failing instantly for a cooldown instead of calling the dead service. Then it cautiously tests recovery.

1. threshold is how many failures in a row trip the breaker; cooldown is how long it stays open. opened_at records when it tripped (None means closed/healthy).
2. In call(), if the breaker is open and still within the cooldown, it raises immediately — "circuit OPEN — failing fast" — without touching the service.
3. Once the cooldown passes it clears opened_at to go half-open: it allows one trial call to see if the service is back.
4. On any exception it bumps self.failures and trips the breaker once the threshold is hit. On success it resets failures to 0 — the service is healthy again.

What the output means: No output by itself. In use, the first few failures pass through and count up; after threshold failures, further calls fail instantly for cooldown seconds — protecting both you and the struggling dependency.

Try this: Create CircuitBreaker(threshold=2) and call it with a function that always raises. After 2 failures the next call should raise "circuit OPEN" instead of running your function.

Retrying an action that changes state risks doing it twice (double-charge, double-deploy). An idempotency key makes a repeat safe — the operation runs once, and later calls with the same key return the stored result. A fallback keeps things useful when the primary path fails.

1. _done is a dictionary standing in for a real dedup store (a DB or Redis in production). It remembers which keys already ran.
2. execute_once(key, action) checks the store first: if the key is present it returns the cached result without re-running the action. Otherwise it runs the action once, saves the result under the key, and returns it.
3. with_fallback(primary, fallback) tries primary() and, on any exception, quietly runs fallback() instead — e.g. a cheaper model, a cached answer, or an "escalate to a human" path.
4. The commented agent example shows the pattern: try the frontier model, and on refusal or overload fall back to something safe.

What the output means: No print here. The behavior to picture: calling execute_once twice with the same key runs the real action only once; with_fallback never lets a single failure surface as a crash.

Try this: Call execute_once("deploy-42", do_deploy) twice with a do_deploy that prints — you'll see it print only once. That's idempotency protecting a state-changing tool.

The interview version of section 4, tightened up. The question is: how do you get reliable structured data out of an unreliable model? Answer: define a schema, validate the output, and retry feeding the error back — the pattern interviewers want to hear.

1. Answer is a minimal Pydantic model — a summary string and a confidence number — the target shape you insist the model produce.
2. parse_with_retry loops tries times. On retries it appends \nFix: {err} to the prompt so the model sees exactly what it got wrong.
3. Answer.model_validate_json(raw) validates the raw text; success returns the typed object, failure is caught as ValidationError and stored in err for the next pass.
4. After all attempts fail, it raises so the caller never receives unvalidated data.

What the output means: A validated Answer on success, or a ValueError if the model never complies. The key idea to state out loud in an interview: feed the validation error back as part of the retry prompt.

Try this: Explain aloud why this beats string-parsing the reply: the schema is the contract, and the model self-corrects from the error message you return to it.

The compact interview form of the retry loop. Same idea as section 6 — retry transient failures with exponentially growing, randomized delays — but short enough to write on a whiteboard.

1. for attempt in range(tries) tries fn() and returns its result on the first success.
2. if attempt == tries - 1: raise gives up after the final attempt so you don't loop forever.
3. time.sleep(random.uniform(0, base * 2 ** attempt)) is the whole trick: 2 ** attempt doubles the window each round (exponential backoff) and random.uniform(0, ...) picks a random point inside it (jitter) so clients don't retry in lockstep.

What the output means: No direct output — it returns fn()'s value or re-raises after the last try, with growing, randomized pauses between attempts.

Try this: Be ready to name the two ingredients and why each matters: exponential (don't hammer a recovering server) and jitter (avoid the thundering herd where everyone retries at once).