Parse LLM output into Pydantic models

Introduction

When you build an agent that picks a tool or updates state based on the model's reply, raw text isn't enough — you need structured data the rest of your code can branch on. LLMs happily wrap JSON in markdown fences, drop trailing commas, or stream half-finished objects, and a single unhandled parse error cascades into a crashed turn, a stuck workflow, or silent garbage flowing into downstream systems. By the end of this lesson you'll be able to extract validated structured output from messy LLM responses using layered parsing strategies that degrade gracefully instead of failing hard.

Key Terminology

• Pydantic model — a typed schema class that validates parsed JSON against expected fields and types; the validation layer between raw parsing and your business logic.
• Markdown fence extraction — pulling JSON out of ```json ... ``` blocks the model emits even when told not to; the single most common cleanup step in production.
• Recovery prompt — re-sending failed output plus the schema back to the LLM with a "fix this" instruction; the last line of defence before raising.
• Partial parse — completing an incomplete JSON string by tracking open braces and brackets so streaming UIs can render progressively before the full response arrives.

Concepts

Why structured parsing is layered

A single json.loads works only when the model behaves perfectly. Real responses arrive with markdown wrappers, trailing commas, unquoted keys, or mid-stream truncation. Robust parsing chains cheap-first strategies — direct parse, fence extraction, regex cleanup, LLM-assisted retry — so the common case stays fast and the edge cases still resolve (see Code Walkthrough).

Parsing versus validation

Parsing turns bytes into Python objects; validation turns objects into trusted, typed data. Keep them distinct: json.JSONDecodeError means the syntax was bad, while pydantic.ValidationError means the shape was wrong. Different errors deserve different recovery — re-prompting fixes the second far more reliably than the first.

Recovery via the model itself

When deterministic cleanup fails, the model that produced the broken JSON is usually best positioned to fix it. A recovery prompt includes the malformed text, the error message, and the schema; the model returns corrected JSON. Cap retries (1–2 is typical) so transient bugs don't turn into runaway loops.

Partial parsing for streaming output

Streaming agents want to render structured fragments before the full response lands. Tracking the stack of unclosed {, [, and " lets you synthesise a temporarily-valid suffix, parse it, and return a "complete=false" partial. The flow below shows how the layers compose.

Code Walkthrough

The snippet below combines three of the concepts above — fence extraction, deterministic cleanup, and LLM-assisted recovery — into a single layered parser keyed to a Pydantic model. The strategies are tried cheapest-first so happy-path calls cost one json.loads.

1import json
2import re
3from typing import Type, TypeVar
4from pydantic import BaseModel, ValidationError
5
6T = TypeVar("T", bound=BaseModel)
7
8def _extract_from_markdown(text: str) -> str:
9    for pattern in (r"```json\s*(.*?)\s*```", r"```\s*(.*?)\s*```"):
10        m = re.search(pattern, text, re.DOTALL)
11        if m:
12            return m.group(1).strip()
13    return text.strip()
14
15def _clean(text: str) -> str:
16    text = re.sub(r",\s*([}\]])", r"\1", text)        # trailing commas
17    text = re.sub(r"//.*?\n", "\n", text)              # // comments
18    text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
19    return text.strip()
20
21def _try(text: str, model: Type[T]) -> T | None:
22    try:
23        return model.model_validate(json.loads(text))
24    except (json.JSONDecodeError, ValidationError):
25        return None
26
27async def parse_structured(
28    text: str,
29    model: Type[T],
30    llm_client,
31    max_retries: int = 1,
32) -> T:
33    """Layered parse: direct -> fence -> cleanup -> LLM recovery."""
34    for candidate in (text, _extract_from_markdown(text), _clean(_extract_from_markdown(text))):
35        result = _try(candidate, model)
36        if result is not None:
37            return result
38
39    schema = json.dumps(model.model_json_schema(), indent=2)
40    current = text
41    for _ in range(max_retries):
42        fix_prompt = (
43            "The following output is not valid JSON for the given schema.\n\n"
44            f"Output:\n{current}\n\nSchema:\n{schema}\n\n"
45            "Return ONLY corrected JSON, no prose, no markdown."
46        )
47        current = (await llm_client.generate(fix_prompt)).text
48        result = _try(_clean(_extract_from_markdown(current)), model)
49        if result is not None:
50            return result
51
52    raise ValueError(f"Could not parse output into {model.__name__} after recovery")

You'll know it works when a call returns a typed Pydantic instance for well-formed input, recovers a markdown-wrapped response without invoking the LLM, and raises ValueError only after at least one recovery attempt on truly broken output.

Do's and Don'ts

Having walked through parsing LLM output above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do parse and validate in separate stages — distinguish JSONDecodeError from ValidationError so recovery prompts can target the actual failure.
2. ✓Do cap recovery retries at 1-2 — more attempts rarely help and can mask deterministic bugs while burning tokens.
3. ✓Do try the cheapest strategy first — direct json.loads for well-behaved responses keeps the hot path fast.

Don'ts

1. ✗Don't trust raw LLM text as structured data — always parse through a typed model before branching on field values.
2. ✗Don't strip markdown fences with ad-hoc string slicing — anchored regex on ```json / ``` blocks survives whitespace and language hints that manual trims miss.
3. ✗Don't recurse on parse failures without a budget — bounded retries plus a final raise keep agents from hanging on unrecoverable output.