7 LangChain Output Fixer and Retry Strategies (2026)

# The model was asked for JSON but returned this:
broken_output_1 = """
Sure! Here's the JSON you requested:
# [json block start]
{
  "name": "Alice",
  "age": 30,
  "skills": ["Python", "ML",]  // trailing comma
}
# [block end]
"""

broken_output_2 = '{"name": "Bob", "age": 25, "skills": ["Python"'  # truncated

broken_output_3 = """
{
  "name": "Charlie"
  "age": 28,           // missing comma after "Charlie"
  "skills": ["Go"]
}
"""

from langchain.output_parsers import OutputFixingParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

class UserProfile(BaseModel):
    name: str = Field(description="User's full name")
    age: int = Field(description="User's age")
    skills: List[str] = Field(description="List of technical skills")

# Base parser
json_parser = JsonOutputParser(pydantic_object=UserProfile)

# Wrapping parser with auto-fix capability
fixing_parser = OutputFixingParser.from_llm(
    parser=json_parser,
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0)
)

# Test with broken JSON
broken_json = """
Sure! Here's the data:
# [json block start]
{
  "name": "Alice Johnson",
  "age": 30,
  "skills": ["Python", "Machine Learning",]
}
# [block end]
"""

try:
    result = fixing_parser.parse(broken_json)
    print(f"Fixed successfully: {result}")
    # → UserProfile(name='Alice Johnson', age=30, skills=['Python', 'Machine Learning'])
except Exception as e:
    print(f"Still failed after fixing: {e}")

from langchain.output_parsers import RetryOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Define expected output
class ProductReview(BaseModel):
    product_name: str
    rating: int = Field(ge=1, le=5, description="Rating from 1 to 5")
    pros: List[str] = Field(min_items=1, description="List of pros")
    cons: List[str] = Field(description="List of cons")
    summary: str

parser = JsonOutputParser(pydantic_object=ProductReview)

# Build the chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a product reviewer. Extract structured review data."),
    ("human", "{review_text}\n\n{format_instructions}")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Retry parser wraps the original chain
retry_parser = RetryOutputParser.from_llm(
    parser=parser,
    llm=llm,
    max_retries=3
)

# Manual retry usage
chain = prompt | llm

def parse_with_retry(review_text: str) -> ProductReview:
    prompt_value = prompt.invoke({
        "review_text": review_text,
        "format_instructions": parser.get_format_instructions()
    })
    
    output = llm.invoke(prompt_value)
    
    try:
        return parser.parse(output.content)
    except Exception:
        # Retry with full context
        return retry_parser.parse_with_prompt(
            completion=output.content,
            prompt_value=prompt_value
        )

result = parse_with_retry("This laptop is amazing! Great battery life and speed, but a bit heavy.")
print(result)

from langchain_core.runnables import RunnableRetry
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = JsonOutputParser()

# Simple retry on the chain
chain_with_retry = (
    ChatPromptTemplate.from_template(
        "Extract a JSON object with 'title' and 'year' from: {text}\n\nReturn only JSON."
    )
    | llm
    | parser
).with_retry(
    retry_if_exception_type=(ValueError, Exception),
    stop_after_attempt=3,
    wait_exponential_jitter=True  # adds random jitter to exponential backoff
)

# Or configure more precisely
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

chain_with_advanced_retry = (
    ChatPromptTemplate.from_template("Extract JSON: {text}")
    | llm
    | parser
).with_retry(
    retry_if_exception_type=(ValueError,),
    stop_after_attempt=3,
)

try:
    result = chain_with_retry.invoke({"text": "The Matrix was released in 1999"})
    print(result)
except Exception as e:
    print(f"Failed after 3 attempts: {e}")

from langchain_core.runnables import RunnableWithFallbacks
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

# Primary chain: structured JSON output
primary_llm = ChatOpenAI(model="gpt-4o", temperature=0)
primary_parser = JsonOutputParser(pydantic_object=UserProfile)

primary_chain = (
    ChatPromptTemplate.from_template(
        "Extract user profile JSON from: {text}\n\n{format_instructions}"
    ).partial(format_instructions=primary_parser.get_format_instructions())
    | primary_llm
    | primary_parser
)

# Fallback 1: cheaper model with the same parser
fallback_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
fallback_chain_1 = (
    ChatPromptTemplate.from_template(
        "Extract user profile JSON from: {text}\n\n{format_instructions}"
    ).partial(format_instructions=primary_parser.get_format_instructions())
    | fallback_llm
    | OutputFixingParser.from_llm(parser=primary_parser, llm=fallback_llm)
)

# Fallback 2: unstructured text extraction (always succeeds)
fallback_chain_2 = (
    ChatPromptTemplate.from_template(
        "Extract the person's name, age, and skills from this text as plain text: {text}"
    )
    | fallback_llm
    | StrOutputParser()
)

# Chain them together
resilient_chain = primary_chain.with_fallbacks(
    fallbacks=[fallback_chain_1, fallback_chain_2],
    exception_to_ignore_types=[ValueError]  # only fall back on parsing errors
)

# The chain tries primary, then fallback_1, then fallback_2
result = resilient_chain.invoke({
    "text": "Sarah, 28, works as a backend engineer specializing in Rust and distributed systems"
})
print(result)

from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
import logging
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def create_resilient_parser(schema_class, llm=None, max_attempts=3):
    """Create a parser with logging, fixing, and fallback logic."""
    
    base_parser = JsonOutputParser(pydantic_object=schema_class)
    fix_llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)
    fixing_parser = OutputFixingParser.from_llm(parser=base_parser, llm=fix_llm)
    
    def parse_with_recovery(text: str):
        attempt = 0
        last_error = None
        
        while attempt < max_attempts:
            try:
                if attempt == 0:
                    # First attempt: direct parse
                    result = base_parser.parse(text)
                    logger.info(f"Parsed successfully on attempt {attempt + 1}")
                    return result
                elif attempt == 1:
                    # Second attempt: auto-fix
                    result = fixing_parser.parse(text)
                    logger.info(f"Fixed and parsed on attempt {attempt + 1}")
                    return result
                else:
                    # Third attempt: aggressive JSON extraction
                    extracted = extract_json_aggressive(text)
                    result = base_parser.parse(extracted)
                    logger.info(f"Aggressively extracted on attempt {attempt + 1}")
                    return result
            except Exception as e:
                last_error = e
                logger.warning(f"Attempt {attempt + 1} failed: {e}")
                attempt += 1
        
        logger.error(f"All {max_attempts} attempts failed. Last error: {last_error}")
        raise last_error
    
    return RunnableLambda(parse_with_recovery)

def extract_json_aggressive(text: str) -> str:
    """Try multiple strategies to extract valid JSON from messy text."""
    import re
    
    # Strategy 1: Find JSON between code blocks
    code_block_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
    if code_block_match:
        return code_block_match.group(1).strip()
    
    # Strategy 2: Find first complete JSON object
    brace_match = re.search(r'\{[\s\S]*\}', text)
    if brace_match:
        candidate = brace_match.group(0)
        try:
            json.loads(candidate)
            return candidate
        except json.JSONDecodeError:
            pass
    
    # Strategy 3: Fix common issues
    cleaned = text.strip()
    cleaned = re.sub(r',\s*([}\]])', r'\1', cleaned)  # Remove trailing commas
    cleaned = re.sub(r'//.*$', '', cleaned, flags=re.MULTILINE)  # Remove JS comments
    cleaned = re.sub(r'/\*.*?\*/', '', cleaned, flags=re.DOTALL)  # Remove block comments
    
    return cleaned

# Usage
resilient = create_resilient_parser(UserProfile, max_attempts=3)
result = resilient.invoke('{"name": "Dave", "age": 35, "skills": ["Python", "Go",]}')

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import List, Optional, Union
import re

class RobustUserProfile(BaseModel):
    name: str
    age: Union[int, str]  # accept string age and coerce
    skills: Union[List[str], str]  # accept comma-separated string
    email: Optional[str] = None

    @field_validator("age", mode="before")
    @classmethod
    def coerce_age(cls, v):
        """Accept '30 years old', '30', or 30."""
        if isinstance(v, int):
            return v
        # Extract digits
        digits = re.findall(r'\d+', str(v))
        if digits:
            return int(digits[0])
        raise ValueError(f"Cannot parse age from: {v}")

    @field_validator("skills", mode="before")
    @classmethod
    def coerce_skills(cls, v):
        """Accept list, comma-separated string, or JSON string."""
        if isinstance(v, list):
            return v
        if isinstance(v, str):
            # Try JSON parsing first
            try:
                parsed = json.loads(v)
                if isinstance(parsed, list):
                    return parsed
            except json.JSONDecodeError:
                pass
            # Fall back to comma splitting
            return [s.strip() for s in v.split(",") if s.strip()]
        return v

    @field_validator("email", mode="before")
    @classmethod
    def validate_email(cls, v):
        """Accept None, empty string, or valid email."""
        if not v:
            return None
        # Basic email validation — don't reject, just normalize
        v = str(v).strip().lower()
        return v if "@" in v else None

    @model_validator(mode="after")
    def check_age_range(self):
        """Post-parse validation with auto-correction."""
        if self.age < 0:
            self.age = 0
        if self.age > 150:
            self.age = 150
        return self

# Test with messy inputs
parser = JsonOutputParser(pydantic_object=RobustUserProfile)

messy_json = '{"name": "Eve", "age": "28 years old", "skills": "Python, ML, LangChain", "email": ""}'
result = RobustUserProfile.model_validate_json(messy_json)
print(result)
# → RobustUserProfile(name='Eve', age=28, skills=['Python', 'ML', 'LangChain'], email=None)

import time
import random
from functools import wraps
from typing import Callable, TypeVar, Any
import openai

T = TypeVar("T")

def exponential_backoff(
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True,
    retryable_exceptions=(openai.RateLimitError, openai.APIConnectionError, openai.APITimeoutError)
):
    """Decorator for exponential backoff with jitter."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            delay = base_delay
            last_exception = None
            
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except retryable_exceptions as e:
                    last_exception = e
                    if attempt == max_retries:
                        break
                    
                    actual_delay = min(delay, max_delay)
                    if jitter:
                        actual_delay *= (0.5 + random.random())
                    
                    logger.warning(
                        f"Attempt {attempt + 1}/{max_retries} failed: {e}. "
                        f"Retrying in {actual_delay:.1f}s..."
                    )
                    time.sleep(actual_delay)
                    delay *= 2
                except Exception as e:
                    # Non-retryable exceptions fail immediately
                    raise
            
            raise last_exception
        return wrapper
    return decorator

# Apply to LangChain chain invocation
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

@exponential_backoff(max_retries=5, base_delay=1.0)
def invoke_with_backoff(chain, input_data):
    return chain.invoke(input_data)

# Usage
llm = ChatOpenAI(model="gpt-4o")
parser = JsonOutputParser(pydantic_object=UserProfile)
chain = ChatPromptTemplate.from_template("Extract JSON: {text}") | llm | parser

result = invoke_with_backoff(chain, {"text": "Alice is 30 and knows Python"})

import asyncio
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

async def robust_chain_invoke(chain, inputs: list[dict], max_concurrent: int = 5) -> list:
    """Run multiple chain invocations with concurrency limiting and retry."""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def invoke_one(input_data: dict):
        async with semaphore:
            for attempt in range(3):
                try:
                    return await chain.ainvoke(input_data)
                except Exception as e:
                    if attempt == 2:
                        logger.error(f"Failed after 3 attempts: {e}")
                        return {"error": str(e), "input": input_data}
                    wait = 2 ** attempt + random.uniform(0, 1)
                    await asyncio.sleep(wait)
    
    tasks = [invoke_one(inp) for inp in inputs]
    return await asyncio.gather(*tasks)

# Process 50 documents concurrently (max 5 at a time)
chain_with_retry = (
    ChatPromptTemplate.from_template("Extract JSON from: {text}")
    | ChatOpenAI(model="gpt-4o-mini")
    | OutputFixingParser.from_llm(
        parser=JsonOutputParser(pydantic_object=UserProfile),
        llm=ChatOpenAI(model="gpt-4o-mini")
    )
)

documents = [{"text": f"User {i}: age {20+i}, skills Python"} for i in range(50)]
results = asyncio.run(robust_chain_invoke(chain_with_retry, documents, max_concurrent=5))

from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict

@dataclass
class ParseMetrics:
    total: int = 0
    successes: int = 0
    fixed: int = 0
    failed: int = 0
    errors: list = field(default_factory=list)

    @property
    def success_rate(self):
        return self.successes / self.total if self.total > 0 else 0

    @property
    def fix_rate(self):
        return self.fixed / self.total if self.total > 0 else 0

metrics = ParseMetrics()

def tracked_parse(text: str, parser, fixing_parser):
    metrics.total += 1
    try:
        result = parser.parse(text)
        metrics.successes += 1
        return result
    except Exception as e1:
        try:
            result = fixing_parser.parse(text)
            metrics.fixed += 1
            return result
        except Exception as e2:
            metrics.failed += 1
            metrics.errors.append({"text": text[:100], "error": str(e2), "time": datetime.now().isoformat()})
            raise e2

# After processing 1000 documents:
print(f"Success rate: {metrics.success_rate:.1%}")
print(f"Fix rate: {metrics.fix_rate:.1%}")
print(f"Failure rate: {metrics.failed/metrics.total:.1%}")