Structured output and output parsers

with_structured_output, Pydantic and JSON schemas, plain text parsers, and handling parse failures with retries instead of hope.

Structured output with a schema

from typing import Literal, Optional
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

class Ticket(BaseModel):
    """A support ticket extracted from a customer message."""

    category: Literal["billing", "technical", "account", "other"] = Field(
        description="The single best matching category.")
    urgency: Literal["low", "normal", "high"] = Field(
        description="high only if the customer is blocked from working.")
    order_id: Optional[str] = Field(default=None, description="Order reference, if present.")
    summary: str = Field(description="One sentence, no more than twenty words.")

model = init_chat_model("gpt-4o-mini", model_provider="openai", temperature=0)
extractor = model.with_structured_output(Ticket, include_raw=True)

result = extractor.invoke(
    "I was charged twice for order A-1042 and now I cannot access my account at all."
)

ticket, raw, error = result["parsed"], result["raw"], result["parsing_error"]
if error:
    print("parse failed:", error)
else:
    print(ticket.model_dump())

• Give every field a description. The model sees the schema and the descriptions; vague field names and missing descriptions are the main cause of wrong values.
• Use Literal for closed sets. A free-text category: str will eventually return billing issue and break your downstream switch.
• include_raw=True is the difference between a debuggable failure and a mystery: you get the parsed object, the raw message, and the parsing error together.
• Mark optional fields with a default of None. A required field the text does not mention is an instruction to invent a value.

Parsers and validation

from langchain_core.output_parsers import (
    StrOutputParser, JsonOutputParser, PydanticOutputParser
)
from langchain_core.prompts import ChatPromptTemplate

str_parser = StrOutputParser()
json_parser = JsonOutputParser(pydantic_object=Ticket)
pydantic_parser = PydanticOutputParser(pydantic_object=Ticket)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract the ticket.\n{format_instructions}"),
    ("human", "{message}"),
]).partial(format_instructions=pydantic_parser.get_format_instructions())

chain = prompt | model | pydantic_parser

try:
    ticket = chain.invoke({"message": "Double charge on order A-1042."})
    print(ticket.category, ticket.urgency)
except Exception as exc:
    print(type(exc).__name__, exc)

# a retry wrapper that feeds the error back to the model once
from langchain.output_parsers import OutputFixingParser
fixing = OutputFixingParser.from_llm(parser=pydantic_parser, llm=model)
repaired = fixing.parse("{'category': 'billing issue', 'urgency': 'HIGH'}")

• PydanticOutputParser injects format instructions into the prompt and validates the result. It is portable but relies on the model following the instructions.
• OutputFixingParser sends the malformed output and the validation error back to the model for a second attempt. Bound it to one retry: repeated fixing rarely converges and doubles your cost.
• Validator errors are the useful signal. Log exc.errors() rather than just the string, so you can see which field failed and why.
• Order a chain as prompt, model, parser. Putting the parser before the model is a common mistake in examples copied without reading.

Structured output in production

from langchain_core.runnables import RunnableLambda
from pydantic import ValidationError

def safe_extract(message: str):
    """Return a validated object, or an explicit failure record. Never guess."""
    try:
        ticket = extractor.invoke(message)["parsed"]
        if ticket is None:
            raise ValueError("model returned nothing parseable")
        return {"ok": True, "ticket": ticket}
    except (ValidationError, ValueError) as exc:
        return {"ok": False, "reason": str(exc)[:200], "input": message[:200]}

robust = RunnableLambda(safe_extract)

results = [robust.invoke(text) for text in incoming_messages]
failures = [r for r in results if not r["ok"]]
print(f"{len(failures)} of {len(results)} needed review")

# metrics worth tracking from day one
def metrics(results):
    total = len(results)
    ok = sum(1 for r in results if r["ok"])
    return {"success_rate": round(ok / total, 3), "review_queue": total - ok}

• A parse failure should produce a reviewable record, not an exception that kills the batch and not a silently defaulted value.
• Track the success rate per field, not just overall. A schema where one optional field fails 30% of the time is a schema problem.
• Schema changes are breaking changes. Version the schema and store the version with every extracted record, or you cannot compare results across releases.
• Downstream code should never index into a dict of model output. Validate into a typed model at the boundary and pass objects internally.

FAQ

Related

Prompt templates and chains Models, messages and providers

Last refreshed 2026-09-18.