Fix malformed JSON in LangChain with OutputFixingParser

LLMs frequently return invalid JSON — trailing commas, missing quotes, truncated output, or markdown code fences wrapped around the payload. The langchain outputfixingparser json tutorial pattern solves this by feeding the malformed output back to a model with a targeted correction prompt. This post walks through the complete setup, customization, and production hardening steps.

Step 1: understand the failure modes

Before reaching for a parser, catalog what actually breaks. In practice, four patterns account for most JSON failures:

1. Trailing commas — {"items": [1, 2, 3],}
2. Unescaped newlines in strings — "description": "line one\nline two"
3. Markdown fences — json\n{...}\n
4. Truncation — output cuts off mid-string due to token limits

LangChain’s OutputFixingParser wraps any base parser (typically JsonOutputParser or PydanticOutputParser) and delegates repair to an LLM when the base parser raises an exception.

from langchain.output_parsers import JsonOutputParser, OutputFixingParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
base_parser = JsonOutputParser()
fixing_parser = OutputFixingParser.from_llm(parser=base_parser, llm=llm)

The from_llm class method constructs a default fixing prompt. That default works for simple cases but often needs tuning — see Step 3.

Step 2: wire the parser into your chain

Attach the fixing parser to any chain that produces structured output. The parser interface is identical to the base parser, so downstream code doesn’t change.

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class Extraction(BaseModel):
    entities: list[str] = Field(description="Named entities found in the text")
    sentiment: str = Field(description="positive, negative, or neutral")

parser = PydanticOutputParser(pydantic_object=Extraction)
fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)

prompt = PromptTemplate(
    template="Extract entities and sentiment from: {text}\n{format_instructions}",
    input_variables=["text"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | llm | fixing_parser

result = chain.invoke({"text": "Apple reported record revenue. The stock jumped 5%."})
print(result)
# Extraction(entities=['Apple'], sentiment='positive')

Verify success: run the chain with inputs known to produce malformed JSON (e.g., long texts that trigger truncation). The call should return a validated Pydantic object without raising.

Step 3: customize the fixing prompt

The default fixing prompt is generic. For production, supply a prompt that includes your schema, common failure examples, and explicit instructions.

from langchain_core.prompts import PromptTemplate

FIXING_PROMPT = PromptTemplate.from_template("""
The following output was intended to be valid JSON matching this schema:
{schema}

But it failed to parse with this error:
{error}

The malformed output:
{completion}

Return ONLY corrected JSON. No markdown, no commentary.
""")

fixing_parser = OutputFixingParser(
    parser=parser,
    llm=llm,
    prompt=FIXING_PROMPT,
    max_retries=2,
)

Key prompt elements:

• Schema injection — {schema} passes parser.get_format_instructions() automatically
• Error context — {error} includes the original json.JSONDecodeError or Pydantic validation error
• Strict output constraint — “Return ONLY corrected JSON” prevents the fixer from explaining itself

Verify success: induce a known failure (e.g., feed the parser a string with a trailing comma). Confirm the fixer returns valid JSON on the first retry.

Step 4: handle truncation and schema violations

OutputFixingParser cannot recover content that never arrived. If the model truncates mid-string, the fixer hallucinates the rest. Two defenses:

Increase output tokens

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

Add a length-aware fallback

from langchain.output_parsers import OutputFixingParser
from langchain_core.exceptions import OutputParserException

class TruncationAwareFixingParser(OutputFixingParser):
    def parse(self, text: str) -> any:
        try:
            return super().parse(text)
        except OutputParserException as e:
            if "truncated" in str(e).lower() or "unterminated" in str(e).lower():
                raise ValueError("Output truncated — increase max_tokens or shorten prompt") from e
            raise

Verify success: set max_tokens=50 on the LLM, run a chain that needs 200 tokens, and confirm the custom exception surfaces instead of silent hallucination.

Step 5: build a fallback chain for production

Single-model fixing fails when the fixer itself is rate-limited or degraded. Chain multiple parsers with different models:

from langchain.output_parsers import OutputFixingParser
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
fallback_llm = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)

primary_parser = OutputFixingParser.from_llm(parser=base_parser, llm=primary_llm)
fallback_parser = OutputFixingParser.from_llm(parser=base_parser, llm=fallback_llm)

def parse_with_fallback(text: str):
    try:
        return primary_parser.parse(text)
    except Exception as e:
        # Log the primary failure for observability
        print(f"Primary fixer failed: {e}")
        return fallback_parser.parse(text)

If you route through a gateway that exposes multiple providers behind one endpoint (n4n.ai does this with automatic fallback on rate limits or degradation), you can simplify to a single OutputFixingParser pointed at the gateway and let the infrastructure handle model-level failover.

Verify success: simulate a primary model outage (e.g., invalid API key) and confirm the fallback parser returns valid output.

Step 6: add metrics and observability

You cannot improve what you don’t measure. Wrap the parser to emit structured logs:

import time
import json
from typing import Any

class InstrumentedFixingParser(OutputFixingParser):
    def parse(self, text: str) -> Any:
        start = time.perf_counter()
        retries = 0
        while True:
            try:
                result = super().parse(text)
                duration_ms = (time.perf_counter() - start) * 1000
                print(json.dumps({
                    "event": "parse_success",
                    "retries": retries,
                    "duration_ms": round(duration_ms, 2),
                    "input_chars": len(text),
                }))
                return result
            except OutputParserException as e:
                retries += 1
                if retries >= self.max_retries:
                    duration_ms = (time.perf_counter() - start) * 1000
                    print(json.dumps({
                        "event": "parse_failure",
                        "retries": retries,
                        "duration_ms": round(duration_ms, 2),
                        "error": str(e),
                    }))
                    raise
                # The base class handles the retry loop internally;
                # this wrapper only adds observability.

Verify success: run a batch of 100 varied inputs through the instrumented parser. Check logs for parse_failure events — any non-zero count indicates a prompt or schema issue to fix.

Step 7: test with a curated fixture suite

Automated regression testing prevents prompt changes from breaking the fixer. Store real malformed outputs as fixtures:

# tests/fixtures/malformed_json.py
FIXTURES = [
    {
        "name": "trailing_comma",
        "input": '{"entities": ["Apple", "Microsoft"], "sentiment": "positive",}',
        "expected": {"entities": ["Apple", "Microsoft"], "sentiment": "positive"},
    },
    {
        "name": "markdown_fence",
        "input": '```json\n{"entities": ["Google"], "sentiment": "neutral"}\n```',
        "expected": {"entities": ["Google"], "sentiment": "neutral"},
    },
    {
        "name": "unescaped_newline",
        "input": '{"entities": ["Tesla"], "sentiment": "positive\nextra"}',
        "expected": {"entities": ["Tesla"], "sentiment": "positive\nextra"},
    },
]

# tests/test_fixing_parser.py
import pytest
from your_module import fixing_parser

@pytest.mark.parametrize("fixture", FIXTURES, ids=lambda f: f["name"])
def test_fixing_parser(fixture):
    result = fixing_parser.parse(fixture["input"])
    assert result.model_dump() == fixture["expected"]

Run in CI on every merge. Add new fixtures whenever a production failure reveals a novel malformation.

Common pitfalls

When not to use OutputFixingParser

• High-throughput, low-latency paths — the extra LLM call adds 200–800 ms. Prefer constrained decoding (e.g., json_schema mode on OpenAI, tool_choice on Anthropic) or a lightweight regex pre-filter.
• Security-sensitive contexts — the fixer sees raw model output. If that output could contain injected instructions, the fixer may execute them. Sanitize first.
• Streaming responses — OutputFixingParser requires the full completion. For streaming, use a parser that validates incrementally (e.g., JsonOutputParser with streaming=True on supported models).

Summary checklist

• Base parser (JsonOutputParser or PydanticOutputParser) matches your schema exactly
• Fixing prompt includes schema, error, and strict “JSON only” instruction
• max_retries set (2–3 is typical)
• Truncation detection surfaces actionable errors
• Fallback parser or gateway-level failover configured
• Instrumentation emits parse success/failure with retry count
• Fixture suite covers all observed malformation types
• CI runs fixture suite on every change

The langchain outputfixingparser json tutorial pattern turns brittle JSON extraction into a resilient pipeline. Start with the default from_llm constructor, then harden each layer — prompt, fallback, observability, tests — as production traffic reveals the real failure distribution.