The mental model here is a parsing pipeline, not a single function call. Think of raw LLM output as untrusted user input. You would never take a string from a web form and insert it directly into a SQL query. Same logic applies: you take raw text, put it through a series of transformations that each handle a specific class of failure, and only trust the output once it has passed all gates.
The failure modes you need to handle fall into three categories. First, framing failures: the model wraps the JSON in a markdown code fence (```json ...``), adds a preamble like "Here is your JSON:", or appends a trailing explanation after the closing brace. These are fixable with a regex or string slicing pass before you even attempt to parse. Second, syntax failures: trailing commas, unquoted keys, single-quoted strings, missing closing brackets, or control characters inside string values. The json-repair` library (pip installable, MIT licensed) handles most of these by walking the token stream and guessing intent. Third, schema failures: the JSON is syntactically valid but the model put a string where you expected an integer, omitted a required field, or nested objects one level deeper than your schema expects. These require schema validation, and Pydantic v2 is the practical standard for Python.
A real-world scenario: you are building a product catalog ingestion pipeline. You send batches of raw product descriptions to GPT-4o and ask it to return structured records with fields like sku, price_usd, category, and in_stock. Works great in dev with clean descriptions. In production, a vendor sends you a description that is 3800 tokens long, and the model truncates mid-response because you forgot to budget output tokens correctly. Now you have a half-formed JSON object. A naive parser crashes. A robust parser detects the truncation (no closing brace), calls repair_json, gets a partial but usable record, validates that the required fields that did make it through are correct, logs the truncation event with the original prompt and model response, and either queues a retry with a shorter input or accepts the partial record with a flag. The pipeline continues. The on-call engineer sees a structured alert in Datadog, not a 500 error storm.
The tradeoff versus constrained decoding (covered in aidev-structured-output-decoding) is worth naming explicitly. Constrained decoding, as implemented by tools like outlines or guidance, forces the model to only emit tokens that keep the output valid according to a grammar. It eliminates syntax failures entirely. But it requires running the model yourself or using a provider that exposes logit-bias-level control. OpenAI's structured outputs feature is the hosted equivalent and handles most cases. The gap where post-hoc repair still wins: you are calling a third-party API that does not support constrained decoding, you are processing legacy responses stored in a database, or you need to handle model output from multiple providers in a single pipeline. Defense in depth means even if you use structured output mode, you still validate with Pydantic, because the provider's guarantee is about syntax, not your business schema.
What changes at scale matters a lot. At 10 users, a try/except that logs and returns None is fine. At 10k users, you need to distinguish error categories so you can route retriable errors (truncation, temporary model degradation) to a retry queue and permanent errors (model consistently misunderstood the schema) to a dead-letter queue for human review. You want metrics: parse success rate, repair success rate, validation failure rate, broken down by model and prompt version. At 10M users, you are likely running this in a worker pool, parsing is on the hot path, and you need to ensure json-repair is not the bottleneck. Profile it: the library does string manipulation and is fast enough for most workloads, but if you are parsing 50k responses per second, consider a Rust-based JSON parser with repair for the hot path and fall back to Python only on failure. You also want circuit breakers: if the parse failure rate for a given model or prompt version spikes above a threshold, stop sending requests and alert, because something upstream changed.
Cost and latency implications: retrying a failed parse means you are paying for another API call and adding latency. Budget your retry budget carefully. One retry is usually worth it. Three retries on every failure will 3x your tail latency and cost. A smarter approach is to classify the failure first: if repair produced a valid schema-conforming result, accept it without a retry. Only retry if you genuinely lost required fields and the failure looks like truncation rather than a schema misunderstanding. Log the repair diff so you can audit whether the repaired output was actually correct.
Key Takeaways
- Always wrap LLM output parsing in a multi-stage pipeline: extract, parse, repair, validate.
- Use
json-repairor similar to fix common syntax errors before giving up on a response. - Validate structure and types with Pydantic, not just syntactic JSON validity.
- Distinguish between retriable parse failures and permanent schema violations to avoid wasting tokens.
Pro tips
- Track your repair success rate in production. If
json-repairis fixing more than 5% of responses, your prompt is the root cause, not the parser. Fix the prompt first; robust parsing is a safety net, not a crutch. - When you log parse failures, always include the raw model output (truncated to a safe length) and the prompt hash. Without the raw output, you cannot reproduce or diagnose the failure later.
- Pydantic's
model_validatewithstrict=False(the default) will coerce"4"to4for anintfield. That is usually what you want from LLM output, but explicitly decide: strict mode catches more bugs, lenient mode recovers more gracefully. - Dead-letter queues are underused in LLM pipelines. Responses that fail both repair and validation should land in a persistent store with full context so a human or a future retry can process them. Silent drops are worse than noisy failures.
Common pitfalls
- Mistake: Calling
json.loads()directly on the full LLM response including markdown fences and prose. Fix: Always run an extraction step first to isolate the JSON region before parsing. - Mistake: Treating all parse failures as retriable, leading to 3x token spend on every bad response. Fix: Classify failures -- truncation is retriable, schema misunderstanding usually is not without a prompt change.
- Mistake: Using
json-repairblindly and trusting its output without schema validation. Fix: Always validate repaired JSON against your Pydantic model; repair can produce syntactically valid but semantically wrong data. - Mistake: Swallowing
ValidationErrorsilently and returningNoneeverywhere. Fix: Log the full Pydantic error including which fields failed, so you can distinguish systemic schema mismatches from one-off anomalies.
When to use repair vs retry vs fail
| Option | Use when | Avoid when |
|---|---|---|
| Accept repaired output | json-repair produces output that passes Pydantic validation and required fields are present | The repaired data looks structurally correct but semantic values are clearly wrong (hallucinated IDs, impossible numbers) |
| Retry the LLM call | Response was truncated (missing closing braces at the end), or output tokens hit the limit | The model consistently misunderstands the schema -- retrying without changing the prompt will reproduce the same failure |
| Fail with structured error | Required fields are missing after repair, or data types cannot be coerced, and you need downstream systems to know | You have a reasonable fallback or default that is safe to use without surfacing an error to the user |
| Fallback to regex extraction | You only need one or two scalar values from the response and full JSON parsing keeps failing | You need a complete nested object -- regex on partial JSON is brittle and hard to maintain |
Code Example
# Requires: json-repair==0.10.0, pydantic==2.x
import json
from json_repair import repair_json
from pydantic import BaseModel, ValidationError
class ProductReview(BaseModel):
product_id: str
rating: int
summary: str
RAW_LLM_OUTPUT = '''```json
{"product_id": "SKU-99", "rating": "4", "summary": "Great build quality,}
```'''
def parse_review(raw: str) -> ProductReview | None:
# Step 1: Strip markdown fences if present
text = raw.strip().removeprefix("```json").removesuffix("```").strip()
# Step 2: Attempt standard parse, then repair
try:
data = json.loads(text)
except json.JSONDecodeError:
data = json.loads(repair_json(text))
# Step 3: Validate and coerce types with Pydantic
try:
return ProductReview.model_validate(data)
except ValidationError as e:
print(f"Schema validation failed: {e}")
return None
result = parse_review(RAW_LLM_OUTPUT)
print(result) # ProductReview(product_id='SKU-99', rating=4, summary='Great build quality,')How this code works
This code demonstrates how to robustly process potentially messy, JSON-like text output, often from AI models, into a reliably structured data object. The parse_review function first cleans the RAW_LLM_OUTPUT by using strip().removeprefix("").removesuffix("") to remove common markdown fences, leaving just the JSON string. Next, it attempts to parse this cleaned string using standard json.loads. A crucial step for robustness is the try-except json.JSONDecodeError block: if standard parsing fails (for example, due to a missing closing brace or quote), it falls back to repair_json from the json_repair library. This function intelligently fixes common malformed JSON issues, making the data parsable even if it's slightly broken.
After the JSON is successfully parsed into a Python dictionary, the code leverages Pydantic to validate and structure the data. A ProductReview class, defined using BaseModel, establishes the expected schema, specifying product_id as a string, rating as an integer, and summary as a string. The ProductReview.model_validate(data) call then checks if the repaired data conforms to this schema. A subtle yet powerful feature here is Pydantic's type coercion: even if the rating came as a string "4" from the JSON, model_validate automatically converts it to an integer 4 because rating: int is specified. If the data still doesn't match the schema after repair and parsing, a ValidationError is caught, printing an error and returning None, ensuring only valid ProductReview objects are ever produced.
Production-grade example
Adds markdown stripping, json-repair fallback, Pydantic validation, structured logging, token tracking, timeouts, and tenacity retries.
# Requires: openai>=1.30, json-repair==0.10.0, pydantic==2.x, tenacity==8.x
import json
import logging
import os
import time
from typing import TypeVar, Type
from json_repair import repair_json
from openai import OpenAI, APIError, APITimeoutError
from pydantic import BaseModel, ValidationError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
log = logging.getLogger(__name__)
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
T = TypeVar("T", bound=BaseModel)
PARSE_ERRORS = 0
REPAIR_SUCCESSES = 0
def extract_json_region(text: str) -> str:
"""Strip markdown fences and leading/trailing prose."""
import re
# Try to find content between ```json ... ``` or ``` ... ```
match = re.search(r"```(?:json)?\s*([\s\S]+?)```", text)
if match:
return match.group(1).strip()
# Fallback: find first { to last }
start = text.find("{")
end = text.rfind("}")
if start != -1 and end != -1 and end > start:
return text[start:end+1]
return text.strip()
def parse_and_validate(raw: str, schema: Type[T]) -> T:
global PARSE_ERRORS, REPAIR_SUCCESSES
region = extract_json_region(raw)
try:
data = json.loads(region)
except json.JSONDecodeError as exc:
PARSE_ERRORS += 1
log.warning("json.loads failed, attempting repair", extra={"error": str(exc), "snippet": region[:200]})
repaired = repair_json(region)
try:
data = json.loads(repaired)
REPAIR_SUCCESSES += 1
log.info("json_repair succeeded", extra={"original_len": len(region), "repaired_len": len(repaired)})
except json.JSONDecodeError:
log.error("json_repair also failed", extra={"snippet": region[:200]})
raise ValueError(f"Unparseable JSON after repair attempt: {region[:300]}")
try:
return schema.model_validate(data)
except ValidationError as exc:
log.error("Schema validation failed", extra={"errors": exc.errors(), "data": data})
raise
@retry(
retry=retry_if_exception_type((APITimeoutError, APIError, ValueError)),
wait=wait_exponential(multiplier=1, min=2, max=30),
stop=stop_after_attempt(3),
reraise=True,
)
def call_and_parse(prompt: str, schema: Type[T], model: str = "gpt-4o-mini") -> T:
start = time.monotonic()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
timeout=15,
)
latency_ms = (time.monotonic() - start) * 1000
usage = response.usage
log.info(
"llm_call_complete",
extra={
"model": model,
"input_tokens": usage.prompt_tokens,
"output_tokens": usage.completion_tokens,
"latency_ms": round(latency_ms, 1),
},
)
raw = response.choices[0].message.content or ""
return parse_and_validate(raw, schema)
# Usage
class ReviewRecord(BaseModel):
product_id: str
rating: int
summary: str
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
result = call_and_parse(
prompt='Return a JSON object with product_id="SKU-42", rating=5, summary="Excellent."',
schema=ReviewRecord,
)
print(result)How this code works
This code offers a robust method for interacting with large language models (LLMs) to retrieve and validate structured JSON data, even if the LLM's response is malformed. It effectively transforms potentially messy text output into clean, schema-compliant data. The system starts by defining the expected data shape using a pydantic BaseModel, like ReviewRecord, specifying fields such as product_id and rating. The main call_and_parse function sends a prompt to an OpenAI model and wraps this call with tenacity's @retry decorator, which automatically reattempts the LLM query if APITimeoutError or APIError occur, making API calls resilient.
The parsing magic happens within parse_and_validate. First, extract_json_region intelligently strips extraneous text and markdown fences (like ...) from the raw LLM output, isolating the core JSON. A subtle but powerful feature is its handling of malformed JSON: if the initial json.loads fails, the code silently attempts to repair the JSON using json_repair.repair_json. This crucial step fixes common issues like missing commas or quotes, then attempts parsing again. If both parsing and repair fail, it raises an error. Finally, schema.model_validate strictly checks the cleaned data against the ReviewRecord schema, ensuring data integrity before it's used.
Practice & master
Try the exercise, check your understanding, then mark this lesson mastered to track your path to pro.
Exercise
Build a safe_parse function that takes a raw string from an LLM and a Pydantic model class, runs the full extract-parse-repair-validate pipeline, and returns either a validated model instance or a ParseResult dataclass with fields success: bool, data: BaseModel | None, and error: str | None. Test it against at least three malformed inputs.
# Requires: json-repair==0.10.0, pydantic==2.x
from dataclasses import dataclass
from typing import Type, TypeVar
from pydantic import BaseModel
T = TypeVar("T", bound=BaseModel)
@dataclass
class ParseResult:
success: bool
data: BaseModel | None
error: str | None
class Movie(BaseModel):
title: str
year: int
rating: float
def safe_parse(raw: str, schema: Type[T]) -> ParseResult:
# TODO 1: Extract the JSON region (strip markdown fences, find { to })
# TODO 2: Try json.loads; on failure, try json_repair then json.loads again
# TODO 3: Validate with schema.model_validate(); catch ValidationError
# TODO 4: Return ParseResult with appropriate fields in each branch
pass
# Test cases
test_inputs = [
'```json\n{"title": "Dune", "year": 2021, "rating": 8.0}\n```', # markdown fence
'{"title": "Oppenheimer", "year": 2023, "rating": "8.9",}', # trailing comma, string rating
'{"title": "Alien", "year": 1979', # truncated
]
for raw in test_inputs:
result = safe_parse(raw, Movie)
print(result)Quick check
A model returns
{"count": "42"}but your schema expectscount: int. Pydantic v2 with default settings will:Your parse failure rate jumps from 1% to 18% after a prompt change. The right first response is:
When is regex-based extraction a reasonable fallback instead of full JSON parsing?