Retry failed LangChain parses with RetryOutputParser

LangChain’s RetryOutputParser wraps any output parser and automatically re-prompts the model when parsing fails, feeding the error back so the model can correct itself. This tutorial walks through wiring it up with Pydantic models, custom retry prompts, and observable fallback behavior you can test in a notebook or CI pipeline.

Step 1: Install the required packages

Use a clean virtual environment. The examples target LangChain 0.2+ and Pydantic v2.

python -m venv .venv && source .venv/bin/activate
pip install "langchain>=0.2.0" "langchain-openai>=0.1.0" pydantic

If you prefer a different provider, swap langchain-openai for langchain-anthropic, langchain-google-genai, or the n4n.ai OpenAI-compatible endpoint — the parser logic stays identical.

Step 2: Define a strict Pydantic schema

RetryOutputParser works best when the parser validates structure and semantics. A bare JsonOutputParser only checks JSON syntax; a PydanticOutputParser enforces field types, constraints, and custom validators.

# models.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal

class TicketClassification(BaseModel):
    category: Literal["billing", "technical", "account", "other"]
    priority: Literal["low", "medium", "high", "critical"]
    confidence: float = Field(ge=0.0, le=1.0)
    summary: str = Field(min_length=10, max_length=200)

    @field_validator("summary")
    @classmethod
    def no_all_caps(cls, v: str) -> str:
        if v.isupper():
            raise ValueError("summary must not be all caps")
        return v

The Literal fields and Field constraints give the model concrete guardrails. The custom validator demonstrates that any Pydantic error — not just type mismatches — triggers a retry.

Step 3: Build the base parser and retry wrapper

# parser_setup.py
from langchain.output_parsers import PydanticOutputParser, RetryOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from models import TicketClassification

# 1. Base parser that knows the schema
base_parser = PydanticOutputParser(pydantic_object=TicketClassification)

# 2. LLM that will do the *correction* (can be same or cheaper model)
retry_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 3. Retry parser with a prompt that includes the schema + error
retry_parser = RetryOutputParser.from_llm(
    parser=base_parser,
    llm=retry_llm,
    max_retries=3,                    # hard cap on correction rounds
    prompt=PromptTemplate.from_template(
        "The previous response failed validation:\n{error}\n\n"
        "Schema requirements:\n{format_instructions}\n\n"
        "Original input:\n{completion}\n\n"
        "Return ONLY corrected JSON that conforms to the schema."
    ),
)

Key parameters:

• max_retries: Prevents infinite loops on unrecoverable errors. Three is a sensible default; increase for flaky models.
• prompt: Receives {error} (the Pydantic ValidationError message), {format_instructions} (auto-generated from the schema), and {completion} (the raw model output). Keep it terse — the correction model doesn’t need the full original prompt.

Step 4: Create the primary extraction chain

# chain.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from parser_setup import base_parser, retry_parser

primary_llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "Classify the support ticket. Output ONLY valid JSON matching the schema.\n"
     "{format_instructions}"),
    ("human", "{ticket_text}"),
]).partial(format_instructions=base_parser.get_format_instructions())

chain = prompt | primary_llm | retry_parser

The chain order matters: prompt → primary_llm → retry_parser. The retry parser sits at the end so it sees the raw model output, not a pre-parsed object.

Step 5: Run a happy-path example

# test_happy.py
from chain import chain

ticket = """
Customer reports being charged twice for the same Pro subscription
invoice #INV-8842 on 2024-03-15. Amount: $49.00 each. Requests refund
for duplicate charge.
"""

result = chain.invoke({"ticket_text": ticket})
print(result)
# category='billing' priority='high' confidence=0.95 summary='Duplicate charge for Pro subscription invoice INV-8842'

Verify success: the returned object is a TicketClassification instance, not a dict. Access fields with dot notation (result.category).

Step 6: Force a validation failure to watch the retry loop

# test_retry.py
from chain import chain

# Intentionally vague ticket — model may omit required fields or
# pick invalid enum values on first try.
ticket = "My thing is broken pls help"

result = chain.invoke({"ticket_text": ticket})
print(result)

Typical first-attempt failure: the model returns {"category": "bug", "priority": "urgent", ...} — both enums invalid. You’ll see the retry parser:

1. Catch the ValidationError
2. Format the correction prompt with the error message
3. Call retry_llm (gpt-4o-mini)
4. Parse the corrected output
5. Repeat up to max_retries

Add verbose=True to the ChatOpenAI constructors to see each correction call in the console.

Step 7: Customize the correction prompt for your domain

The default correction prompt works, but domain-specific hints reduce retries. Example: if the model consistently confuses “technical” vs “account” categories, bake that into the prompt.

# parser_setup.py (updated)
from langchain_core.prompts import PromptTemplate

CORRECTION_PROMPT = PromptTemplate.from_template(
    "You are fixing a JSON validation error for a support-ticket classifier.\n"
    "Error: {error}\n\n"
    "Rules:\n"
    "- category MUST be one of: billing, technical, account, other\n"
    "- priority MUST be one of: low, medium, high, critical\n"
    "- confidence is a float 0.0-1.0\n"
    "- summary: 10-200 chars, NOT all caps\n\n"
    "Schema: {format_instructions}\n\n"
    "Bad output: {completion}\n\n"
    "Return ONLY corrected JSON."
)

retry_parser = RetryOutputParser.from_llm(
    parser=base_parser,
    llm=retry_llm,
    max_retries=3,
    prompt=CORRECTION_PROMPT,
)

After this change, the same vague ticket usually succeeds on the first retry.

Step 8: Handle unrecoverable failures gracefully

RetryOutputParser raises the last ValidationError if all retries exhaust. Wrap the chain to return a structured fallback instead of crashing.

# resilient_chain.py
from chain import chain
from models import TicketClassification
from pydantic import ValidationError

FALLBACK = TicketClassification(
    category="other",
    priority="low",
    confidence=0.0,
    summary="Classification failed after retries",
)

def safe_classify(ticket_text: str) -> TicketClassification:
    try:
        return chain.invoke({"ticket_text": ticket_text})
    except ValidationError as e:
        # Log e.errors() for observability
        return FALLBACK

This pattern keeps your upstream code simple — it always receives a valid TicketClassification.

Step 9: Add observability with callbacks

LangChain callbacks let you record every retry attempt without cluttering business logic.

# callbacks.py
from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict

class RetryLogger(BaseCallbackHandler):
    def on_retry(self, *, run_id, error: BaseException, **kwargs: Any) -> None:
        print(f"[RETRY] run={run_id} error={error}")

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        if "retry_count" in outputs:
            print(f"[DONE] retries used: {outputs['retry_count']}")

# Usage
from chain import chain
result = chain.invoke(
    {"ticket_text": "vague ticket"},
    config={"callbacks": [RetryLogger()]}
)

The on_retry hook fires on each correction attempt. on_chain_end receives a retry_count key automatically added by RetryOutputParser.

Step 10: Test edge cases in CI

Add a pytest suite that exercises the retry path deterministically. Mock the primary LLM to return known-bad output, then assert the parser recovers.

# test_parser.py
import pytest
from unittest.mock import AsyncMock, patch
from chain import chain
from models import TicketClassification

@patch("chain.primary_llm")
def test_retry_recovers_from_invalid_enum(mock_llm):
    # First call returns bad enum, second returns valid
    mock_llm.ainvoke = AsyncMock(side_effect=[
        type("Msg", (), {"content": '{"category": "bug", "priority": "urgent", "confidence": 0.5, "summary": "test"}'})(),
        type("Msg", (), {"content": '{"category": "technical", "priority": "high", "confidence": 0.9, "summary": "Login fails on mobile"}'})(),
    ])

    result = chain.invoke({"ticket_text": "login broken"})
    assert isinstance(result, TicketClassification)
    assert result.category == "technical"
    assert result.priority == "high"

Run with pytest -q. The test validates that the retry loop executes and produces a schema-compliant object.

Common pitfalls

When not to use RetryOutputParser

• Latency-sensitive paths: Each retry adds a round-trip. For sub-200 ms SLAs, prefer stricter prompting or constrained decoding (e.g., response_format={type: "json_object"} on OpenAI).
• High-stakes decisions: If a misclassification triggers irreversible actions, fail fast and escalate to human review instead of auto-correcting.
• Streaming responses: RetryOutputParser buffers the full completion before parsing. It does not work with token-by-token streaming parsers.

Summary checklist

• Define a strict Pydantic model with enums, constraints, and validators.
• Create PydanticOutputParser → RetryOutputParser with a dedicated correction LLM.
• Build the chain: prompt | primary_llm | retry_parser.
• Test happy path and forced-failure path.
• Customize the correction prompt with domain rules.
• Wrap invocation in a try/except that returns a typed fallback.
• Add callbacks for retry observability.
• Cover the retry path in CI with mocked bad outputs.

You now have a self-healing structured-output pipeline that degrades gracefully instead of throwing 500s when the model hallucinates a field name.