Structured output with LangChain and Pydantic tools

Structured output is the difference between a demo that works once and a pipeline that runs in production. LangChain’s tool calling abstraction, combined with Pydantic for validation, gives you a path to deterministic JSON from non-deterministic models. This guide walks through building a typed extraction pipeline you can actually ship — schema definition, prompt construction, retry logic, and observability hooks included.

Step 1: Define your contract with Pydantic

Start with the schema. Pydantic v2 is the de facto standard for this — it handles validation, serialization, and gives you clear error messages when the model hallucinates fields.

# schemas.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional
from datetime import datetime


class LineItem(BaseModel):
    quantity: int = Field(..., ge=1)
    unit_price_cents: int = Field(..., ge=0)
    sku: Optional[str] = Field(None, pattern=r"^[A-Z]{2,4}-\d{4,6}$")

    @field_validator("description")
    @classmethod
    def no_all_caps(cls, v: str) -> str:
        if v.isupper() and len(v) > 3:
            return v.title()
        return v


class InvoiceExtraction(BaseModel):
    invoice_number: str = Field(..., pattern=r"^INV-\d{8}-\d{4}$")
    date: datetime
    vendor_name: str = Field(..., min_length=1, max_length=100)
    line_items: list[LineItem] = Field(..., min_length=1)
    subtotal_cents: int = Field(..., ge=0)
    tax_cents: int = Field(..., ge=0)
    total_cents: int = Field(..., ge=0)
    currency: Literal["USD", "EUR", "GBP"] = "USD"
    payment_terms: Optional[str] = None

    @field_validator("total_cents")
    @classmethod
    def totals_match(cls, v: int, info) -> int:
        if "subtotal_cents" in info.data and "tax_cents" in info.data:
            expected = info.data["subtotal_cents"] + info.data["tax_cents"]
            if v != expected:
                raise ValueError(f"total_cents {v} != subtotal + tax {expected}")
        return v

The validators catch the class of errors LLMs make most: arithmetic mismatches, format violations, and enum drift. Keep schemas in a separate module — you’ll import them into both the extraction code and your tests.

Step 2: Wrap the schema as a LangChain tool

LangChain’s StructuredTool (or the newer tool decorator) converts a Pydantic model into an OpenAI-compatible function definition. The model sees the JSON schema; your code gets a validated object back.

# tools.py
from langchain_core.tools import StructuredTool
from schemas import InvoiceExtraction


def extract_invoice(text: str) -> InvoiceExtraction:
    """
    Extract structured invoice data from raw text.
    Returns a validated InvoiceExtraction or raises ValidationError.
    """
    # This function body never executes — the LLM calls the tool.
    # The docstring becomes the function description in the schema.
    raise NotImplementedError("Called by LLM via tool calling")


invoice_tool = StructuredTool.from_function(
    func=extract_invoice,
    name="extract_invoice",
    description=extract_invoice.__doc__,
    args_schema=InvoiceExtraction,
    return_direct=False,  # return the tool result to the model for further reasoning
)

return_direct=False matters: it lets the model see the validated result and decide whether to retry or proceed. If you set True, the tool output becomes the final chain output — useful for single-shot extraction, less so for multi-step agents.

Step 3: Build the prompt with few-shot examples

Tool calling quality lives or dies on the system prompt. Include 2-3 diverse examples covering edge cases your schema handles (missing optional fields, validation failures, nested arrays).

# prompts.py
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate

EXAMPLES = [
    {
        "input": """Invoice INV-20240115-0042
Date: 2024-01-15
Vendor: Acme Corporation
Items:
  1. Widget Pro (SKU: WD-12345) x 10 @ $29.99
  2. Gadget Plus (SKU: GD-987654) x 5 @ $49.50
Subtotal: $547.40
Tax (8.5%): $46.53
Total: $593.93
Terms: Net 30""",
        "output": {
            "invoice_number": "INV-20240115-0042",
            "date": "2024-01-15T00:00:00",
            "vendor_name": "Acme Corporation",
            "line_items": [
                {"description": "Widget Pro", "quantity": 10, "unit_price_cents": 2999, "sku": "WD-12345"},
                {"description": "Gadget Plus", "quantity": 5, "unit_price_cents": 4950, "sku": "GD-987654"},
            ],
            "subtotal_cents": 54740,
            "tax_cents": 4653,
            "total_cents": 59393,
            "currency": "USD",
            "payment_terms": "Net 30",
        },
    },
    {
        "input": """INV-20240220-0101
2024-02-20
Beta LLC
- Service: Consulting hours x 40 @ $150/hr
Subtotal: $6000.00
Tax: $0.00
Total: $6000.00""",
        "output": {
            "invoice_number": "INV-20240220-0101",
            "date": "2024-02-20T00:00:00",
            "vendor_name": "Beta LLC",
            "line_items": [
                {"description": "Consulting hours", "quantity": 40, "unit_price_cents": 15000, "sku": None},
            ],
            "subtotal_cents": 600000,
            "tax_cents": 0,
            "total_cents": 600000,
            "currency": "USD",
            "payment_terms": None,
        },
    },
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    examples=EXAMPLES,
    example_prompt=example_prompt,
)

SYSTEM_PROMPT = """You are an invoice extraction specialist. Extract all fields exactly as they appear.
- Convert dollar amounts to integer cents (multiply by 100, round).
- Dates must be ISO 8601 (YYYY-MM-DDTHH:MM:SS).
- SKU format: 2-4 uppercase letters, hyphen, 4-6 digits. Omit if not present.
- If payment terms are not mentioned, omit the field entirely.
- The total_cents must equal subtotal_cents + tax_cents. Verify before calling the tool."""

extraction_prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    few_shot_prompt,
    ("human", "{input}"),
])

Note the explicit instructions on cents conversion and date format. Models still miss these without repetition. The few-shot examples show the exact JSON shape the tool expects — including null for optional fields that are absent.

Step 4: Wire the chain with a model that supports tool calling

You need a model with native function calling. GPT-4o, GPT-4-turbo, Claude 3.5 Sonnet, and most recent open models via providers like Together or Fireworks work. Configure tool_choice to force the extraction tool.

# chain.py
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI
from tools import invoice_tool
from prompts import extraction_prompt
from schemas import InvoiceExtraction
from pydantic import ValidationError
import json


def build_extraction_chain(model_name: str = "gpt-4o", temperature: float = 0):
    llm = ChatOpenAI(
        model=model_name,
        temperature=temperature,
        # n4n.ai users: replace with base_url="https://api.n4n.ai/v1" and your API key
    )

    # Bind the tool and force its use
    llm_with_tools = llm.bind_tools(
        [invoice_tool],
        tool_choice="extract_invoice",  # forces the model to call this tool
        parallel_tool_calls=False,
    )

    chain = extraction_prompt | llm_with_tools

    return chain


def parse_tool_call(ai_message) -> InvoiceExtraction:
    """Extract and validate the tool call arguments from the AI message."""
    tool_calls = ai_message.tool_calls
    if not tool_calls:
        raise ValueError("Model did not call the extraction tool")

    # We forced tool_choice, so there should be exactly one
    call = tool_calls[0]
    if call["name"] != "extract_invoice":
        raise ValueError(f"Unexpected tool call: {call['name']}")

    # Validate against the Pydantic model — this catches schema violations
    try:
        return InvoiceExtraction.model_validate(call["args"])
    except ValidationError as e:
        # Re-raise with context for retry logic
        raise ValidationError(f"Tool arguments failed validation: {e}") from e


# Composable chain: prompt -> model -> parse -> validated object
extraction_chain = build_extraction_chain() | RunnableLambda(parse_tool_call)

The RunnableLambda(parse_tool_call) step is where validation happens. If the model returns malformed arguments, you get a ValidationError with field-level detail — not a generic JSON parse error.

Step 5: Add retry with exponential backoff and correction

Models occasionally miss a validator (especially arithmetic). Feed the validation error back to the model and retry. Three attempts with increasing temperature usually resolves it.

# retry.py
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from pydantic import ValidationError
from chain import extraction_chain
from schemas import InvoiceExtraction
import logging

logger = logging.getLogger(__name__)


class ExtractionError(Exception):
    """Raised when all retries exhausted."""
    pass


@retry(
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(ValidationError),
    reraise=True,
)
def extract_with_retry(text: str) -> InvoiceExtraction:
    """Extract invoice with automatic retry on validation failure."""
    try:
        result = extraction_chain.invoke({"input": text})
        logger.info(f"Extracted invoice {result.invoice_number} on attempt")
        return result
    except ValidationError as e:
        # Log the specific validation failures for debugging
        logger.warning(f"Validation failed: {e.errors()}")
        # The retry decorator will re-invoke; the chain sees the same prompt
        # For smarter correction, see the correction_chain below
        raise


def extract_with_correction(text: str, max_attempts: int = 3) -> InvoiceExtraction:
    """
    Retry with error feedback injected into the prompt.
    More effective than blind retry for systematic errors.
    """
    last_error = None
    messages = extraction_prompt.format_messages(input=text)

    for attempt in range(max_attempts):
        try:
            ai_message = build_extraction_chain().invoke(messages)
            return parse_tool_call(ai_message)
        except ValidationError as e:
            last_error = e
            logger.warning(f"Attempt {attempt + 1} failed: {e.errors()}")

            # Inject error feedback as a correction message
            error_summary = "\n".join(
                f"- {err['loc']}: {err['msg']}" for err in e.errors()
            )
            correction = (
                f"Your previous extraction failed validation:\n{error_summary}\n"
                "Fix the errors and call the tool again."
            )
            messages.append(ai_message)
            messages.append(("tool", correction, "extract_invoice"))

    raise ExtractionError(f"Failed after {max_attempts} attempts: {last_error}")

The extract_with_correction variant is stronger — it feeds the exact validation errors back into the conversation context. The model sees its own tool call, the validation failure, and corrects. This handles systematic issues like “always forgets to convert to cents” better than blind retry.

Step 6: Observability — log inputs, outputs, and token usage

You need to know what the model saw, what it produced, and what it cost. Wrap the chain with a callback handler.

# observability.py
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List
import time
import json
import logging

logger = logging.getLogger(__name__)


class ExtractionCallbackHandler(BaseCallbackHandler):
    def __init__(self, run_id: str):
        self.run_id = run_id
        self.start_time = time.time()
        self.token_usage = {"prompt": 0, "completion": 0, "total": 0}

    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs):
        logger.info(f"[{self.run_id}] LLM start: model={serialized.get('kwargs', {}).get('model_name')}")
        # Truncate for logs
        for i, p in enumerate(prompts):
            logger.debug(f"[{self.run_id}] Prompt {i}: {p[:500]}...")

    def on_llm_end(self, response: LLMResult, **kwargs):
        usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        self.token_usage = {
            "prompt": usage.get("prompt_tokens", 0),
            "completion": usage.get("completion_tokens", 0),
            "total": usage.get("total_tokens", 0),
        }
        logger.info(f"[{self.run_id}] LLM end: tokens={self.token_usage}")

    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs):
        logger.info(f"[{self.run_id}] Tool call: {serialized.get('name')}")

    def on_tool_end(self, output: str, **kwargs):
        logger.info(f"[{self.run_id}] Tool result: {output[:200]}...")

    def on_chain_error(self, error: Exception, **kwargs):
        logger.error(f"[{self.run_id}] Chain error: {error}", exc_info=True)


def extract_with_observability(text: str, run_id: str) -> InvoiceExtraction:
    handler = ExtractionCallbackHandler(run_id)
    config = {"callbacks": [handler], "run_name": "invoice_extraction"}

    result = extract_with_correction(text)  # or extract_with_retry
    logger.info(f"[{run_id}] Success: {result.invoice_number}, tokens={handler.token_usage}")
    return result

Attach this to your API endpoint or batch job. The token usage lets you track cost per extraction; the tool call logs let you debug when the model picks the wrong tool (not an issue here with forced choice, but critical in multi-tool agents).

Step 7: Verify with a test suite that catches regressions

Test the contract, not the model. Your tests should verify that valid inputs produce valid outputs, invalid inputs raise the right errors, and the schema doesn’t drift.

# test_extraction.py
import pytest
from schemas import InvoiceExtraction, LineItem
from chain import parse_tool_call
from retry import extract_with_correction
from pydantic import ValidationError
from unittest.mock import Mock, patch


class TestSchemaValidation:
    def test_valid_invoice_passes(self):
        data = {
            "invoice_number": "INV-20240115-0042",
            "date": "2024-01-15T00:00:00",
            "vendor_name": "Acme Corp",
            "line_items": [
                {"description": "Widget", "quantity": 1, "unit_price_cents": 1000}
            ],
            "subtotal_cents": 1000,
            "tax_cents": 100,
            "total_cents": 1100,
        }
        invoice = InvoiceExtraction.model_validate(data)
        assert invoice.invoice_number == "INV-20240115-0042"

    def test_total_mismatch_raises(self):
        data = {
            "invoice_number": "INV-20240115-0042",
            "date": "2024-01-15T00:00:00",
            "vendor_name": "Acme Corp",
            "line_items": [{"description": "Widget", "quantity": 1, "unit_price_cents": 1000}],
            "subtotal_cents": 1000,
            "tax_cents": 100,
            "total_cents": 999,  # wrong
        }
        with pytest.raises(ValidationError) as exc:
            InvoiceExtraction.model_validate(data)
        assert "total_cents" in str(exc.value)

    def test_sku_pattern_enforced(self):
        item = LineItem(description="X", quantity=1, unit_price_cents=100, sku="bad-format")
        with pytest.raises(ValidationError):
            LineItem.model_validate(item.model_dump())


class TestParseToolCall:
    def test_parses_valid_tool_call(self):
        ai_msg = Mock()
        ai_msg.tool_calls = [{
            "name": "extract_invoice",
            "args": {
                "invoice_number": "INV-20240115-0042",
                "date": "2024-01-15T00:00:00",
                "vendor_name": "Test",
                "line_items": [{"description": "Item", "quantity": 1, "unit_price_cents": 100}],
                "subtotal_cents": 100,
                "tax_cents": 10,
                "total_cents": 110,
            },
        }]
        result = parse_tool_call(ai_msg)
        assert isinstance(result, InvoiceExtraction)

    def test_raises_on_missing_tool_call(self):
        ai_msg = Mock()
        ai_msg.tool_calls = []
        with pytest.raises(ValueError, match="did not call the extraction tool"):
            parse_tool_call(ai_msg)


class TestEndToEnd:
    @patch("retry.build_extraction_chain")
    def test_extract_with_correction_retries_on_validation_error(self, mock_chain):
        # First call returns invalid total, second returns valid
        invalid_msg = Mock()
        invalid_msg.tool_calls = [{
            "name": "extract_invoice",
            "args": {
                "invoice_number": "INV-20240115-0042",
                "date": "2024-01-15T00:00:00",
                "vendor_name": "Test",
                "line_items": [{"description": "Item", "quantity": 1, "unit_price_cents": 100}],
                "subtotal_cents": 100,
                "tax_cents": 10,
                "total_cents": 999,  # invalid
            },
        }]

        valid_msg = Mock()
        valid_msg.tool_calls = [{
            "name": "extract_invoice",
            "args": {
                "invoice_number": "INV-20240115-0042",
                "date": "2024-01-15T00:00:00",
                "vendor_name": "Test",
                "line_items": [{"description": "Item", "quantity": 1, "unit_price_cents": 100}],
                "subtotal_cents": 100,
                "tax_cents": 10,
                "total_cents": 110,  # valid
            },
        }]

        mock_chain.return_value.invoke.side_effect = [invalid_msg, valid_msg]

        result = extract_with_correction("sample invoice text", max_attempts=2)
        assert result.total_cents == 110
        assert mock_chain.return_value.invoke.call_count == 2

Run these in CI. The schema tests catch Pydantic definition drift. The parsing tests catch changes in how LangChain surfaces tool calls. The end-to-end test verifies the correction loop works — mock the chain to return a validation error first, then a valid result.

Step 8: Deploy as a typed API endpoint

FastAPI + Pydantic gives you request/response validation for free. The endpoint returns the same model you use internally.

# api.py
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from schemas import InvoiceExtraction
from observability import extract_with_observability
import uuid

app = FastAPI(title="Invoice Extraction API")


class ExtractRequest(BaseModel):
    text: str = Field(..., min_length=10, max_length=50000)


class ExtractResponse(BaseModel):
    invoice: InvoiceExtraction
    run_id: str


@app.post("/extract", response_model=ExtractResponse)
async def extract_invoice(request: ExtractRequest, http_request: Request):
    run_id = str(uuid.uuid4())[:8]
    try:
        invoice = extract_with_observability(request.text, run_id)
        return ExtractResponse(invoice=invoice, run_id=run_id)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())
    except Exception as e:
        # Log full traceback internally, return generic error
        logger.error(f"[{run_id}] Extraction failed: {e}")
        raise HTTPException(status_code=500, detail="Extraction failed")

The response model guarantees your API contract matches your internal schema. Clients get typed SDKs automatically via OpenAPI.

How to verify it works

1. Unit tests pass: pytest test_extraction.py -v — all schema, parsing, and retry logic tests green.
2. Smoke test the endpoint: curl -X POST localhost:8000/extract -H "Content-Type: application/json" -d '{"text": "INV-20240115-0042\nDate: 2024-01-15\nVendor: Test Corp\nItems:\n 1. Widget x 1 @ $10.00\nSubtotal: $10.00\nTax: $1.00\nTotal: $11.00"}' — returns 200 with validated JSON.
3. Check logs: You should see the run_id, token usage, and no validation errors on valid inputs.
4. Load test: Send 50 concurrent requests with varied invoice formats. P99 latency should be under 10s (model dependent); error rate near zero on clean inputs.

Common failure modes and fixes

What to extend next

• Streaming: For large documents, split into chunks, extract per-chunk, then merge with a reduction step.
• Confidence scoring: Add a confidence: float field to the schema; have the model self-assess.
• Human-in-the-loop: Route low-confidence extractions to a review queue; feed corrections back as few-shot examples.
• Schema evolution: Version your Pydantic models (InvoiceExtractionV2) and migrate with a background job.

The pattern — Pydantic schema → StructuredTool → forced tool choice → validation → correction retry — generalizes to any extraction task: medical coding, contract clause detection, financial statement parsing. The schema is the contract. Everything else is plumbing to honor it.