Phase 2: Prompt Engineering & LLM Patterns

Robust parsers for edge cases & malformed responses

Intermediate ~16 min read
Think of it this way A friendly analogy. Read this if the technical version feels dense. Show Hide

Imagine you're trying to bake your favorite cake, but instead of getting a recipe from a perfect cookbook, you have a super clever, but sometimes a bit messy, robot helper writing it out for you. Most of the time, the robot is amazing: "Add two cups of flour. Mix three eggs." Perfect! But sometimes, it makes tiny mistakes. It might write "Add two cups flour" (missing the period!), or "Mix three eggs and a sparkle cloud" (something extra and weird!), or even "Add two cups of flower" (a typo!). If you just follow exactly what it says, your cake might not turn out right, or you might get so confused you just stop baking altogether!

In the world of computers, when special AI programs give us information, it's kind of like that. We expect the AI to give us neat, organized instructions, like a perfect recipe. But just like our messy robot chef, the AI sometimes makes small blips – an extra comma, a missing bracket, or adding something totally unexpected. A "robust parser" is like having a super-smart kitchen assistant. When the messy robot gives you a recipe, this assistant quickly looks it over. If it sees "Add two cups flower," it thinks, "Hmm, 'flower' should probably be 'flour'!" It quietly fixes the little mistakes or gently removes the weird extra bits, like the sparkle cloud. It tries its best to get a working recipe for you.

So, instead of just throwing up your hands and saying, "I can't bake this cake because the recipe is weird!" this smart assistant lets you keep going. It means you can still bake your awesome cake, even with slightly messy instructions. Programmers use this idea when they build apps or games that talk to AIs. For example, if an AI is supposed to tell your game what items a player found, and it accidentally sends "sword,sheild," instead of "sword, shield," a robust parser fixes it so your game knows you found both.

This means you can build programs that are much more forgiving and don't stop working just because an AI made a tiny typo. Your programs become like a chef who can still make a delicious meal even if the recipe has a few smudges or scribbles on it, ensuring your creations keep working smoothly and reliably, no matter what.

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-repair or 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-repair is 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_validate with strict=False (the default) will coerce "4" to 4 for an int field. 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-repair blindly 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 ValidationError silently and returning None everywhere. 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

python
# 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.

python
# 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.

python
# 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

  1. A model returns {"count": "42"} but your schema expects count: int. Pydantic v2 with default settings will:

  2. Your parse failure rate jumps from 1% to 18% after a prompt change. The right first response is:

  3. When is regex-based extraction a reasonable fallback instead of full JSON parsing?

Self-check: Without looking at your notes, walk through the four stages of a robust parsing pipeline for an LLM response. For each stage, name one specific failure class it handles and one tool or technique you would use to handle it.