Structured output on Claude 3.5 Sonnet via LangChain

Structured output is the difference between a demo that works once and a production feature that doesn’t break when the model gets creative. Claude 3.5 Sonnet supports tool calling natively, and LangChain exposes it through a clean Pydantic-based API. This tutorial walks through the practical patterns you’ll actually use — validation, retries, streaming, and fallback — with code you can drop into a service.

Prerequisites

You need Python 3.10+, an Anthropic API key, and the following packages:

pip install langchain-anthropic langchain-core pydantic python-dotenv

Create a .env file with your key:

ANTHROPIC_API_KEY=sk-ant-...

All examples assume this loads via python-dotenv at module import:

from dotenv import load_dotenv
load_dotenv()

Minimal working example

Start with the simplest thing that validates: a Pydantic model and with_structured_output.

from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage

class TicketClassification(BaseModel):
    """Schema for routing incoming support tickets."""
    category: str = Field(description="One of: billing, technical, account, other")
    priority: str = Field(description="One of: low, medium, high, critical")
    confidence: float = Field(ge=0.0, le=1.0, description="Model's confidence in classification")

llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
    max_tokens=1024,
)

structured_llm = llm.with_structured_output(TicketClassification)

result = structured_llm.invoke([
    HumanMessage(content="I've been charged twice for my subscription this month. Need a refund ASAP.")
])

print(result)
# category='billing' priority='high' confidence=0.95

Checkpoint output:

category='billing' priority='high' confidence=0.95

The model returns a TicketClassification instance, not a string. Validation failures raise ValidationError before your code sees the result.

Understanding what happens under the hood

with_structured_output converts your Pydantic model to a tool schema and binds it to the model. The call looks like this internally:

# Equivalent manual construction
from langchain_core.utils.function_calling import convert_to_openai_tool

tool_schema = convert_to_openai_tool(TicketClassification)
llm_with_tools = llm.bind_tools([tool_schema], tool_choice="auto")

raw_response = llm_with_tools.invoke([...])
# raw_response.tool_calls[0]['args'] contains the parsed dict

You rarely need the manual form, but it’s useful when you want multiple tool options or custom tool_choice logic.

Handling validation failures with retries

Models occasionally miss required fields or violate constraints. LangChain’s create_structured_output_runnable (the lower-level builder behind with_structured_output) accepts a retry config.

from langchain_core.runnables import RunnableConfig
from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import PydanticToolsParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Classify the support ticket. Return only the structured output."),
    ("human", "{ticket_text}"),
])

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
chain = prompt | llm.with_structured_output(TicketClassification, include_raw=False)

# Retry up to 2 times on validation error
config = RunnableConfig(max_retries=2)

result = chain.invoke({"ticket_text": "Login broken"}, config=config)

If validation fails after retries, you get a ValidationError with the raw model response attached for debugging.

Streaming structured output

For long-running classifications or UIs that show progress, stream partial parses:

from typing import AsyncIterator

async def stream_classification(ticket_text: str) -> AsyncIterator[TicketClassification]:
    async for chunk in structured_llm.astream([HumanMessage(content=ticket_text)]):
        # Each chunk is a partial TicketClassification with fields filled so far
        yield chunk

# Usage
import asyncio

async def main():
    async for partial in stream_classification("Cannot reset password, link expires instantly"):
        print(f"Partial: {partial.model_dump(exclude_unset=True)}")

asyncio.run(main())

Checkpoint output:

Partial: {'category': 'account'}
Partial: {'category': 'account', 'priority': 'medium'}
Partial: {'category': 'account', 'priority': 'medium', 'confidence': 0.87}

Note: streaming yields partial models. Only the final chunk passes full validation.

Complex nested schemas

Real schemas nest. Here’s a triage output that includes extracted entities and suggested actions.

from typing import List, Optional
from pydantic import BaseModel, Field
from enum import Enum

class ActionType(str, Enum):
    REFUND = "refund"
    ESCALATE = "escalate"
    AUTO_REPLY = "auto_reply"
    NO_ACTION = "no_action"

class Entity(BaseModel):
    type: str = Field(description="Entity type: order_id, account_id, email, amount, date")
    value: str
    confidence: float = Field(ge=0.0, le=1.0)

class TriageResult(BaseModel):
    classification: TicketClassification
    entities: List[Entity] = Field(default_factory=list)
    suggested_actions: List[ActionType] = Field(default_factory=list)
    summary: str = Field(max_length=200)
    requires_human: bool

structured_llm = llm.with_structured_output(TriageResult)

result = structured_llm.invoke([HumanMessage(content="""
Customer John Doe (john@example.com) reports order ORD-8842 arrived damaged.
Photos attached. Requesting full refund of $149.99. This is their third order.
""")])

print(result.model_dump_json(indent=2))

Checkpoint output:

{
  "classification": {
    "category": "billing",
    "priority": "high",
    "confidence": 0.96
  },
  "entities": [
    {"type": "email", "value": "john@example.com", "confidence": 0.99},
    {"type": "order_id", "value": "ORD-8842", "confidence": 0.98},
    {"type": "amount", "value": "149.99", "confidence": 0.95}
  ],
  "suggested_actions": ["refund", "escalate"],
  "summary": "Customer reports damaged order ORD-8842, requests $149.99 refund. Third order, high priority.",
  "requires_human": true
}

Adding few-shot examples for consistency

Claude follows examples better than instructions alone. Bind examples into the prompt:

from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {
        "input": "My card was charged twice for the same subscription",
        "output": TicketClassification(category="billing", priority="high", confidence=0.95)
    },
    {
        "input": "How do I change my email address?",
        "output": TicketClassification(category="account", priority="low", confidence=0.9)
    },
    {
        "input": "The API returns 500 errors on /v1/users endpoint",
        "output": TicketClassification(category="technical", priority="critical", confidence=0.98)
    },
]

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

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "Classify support tickets. Follow the examples exactly."),
    few_shot_prompt,
    ("human", "{ticket_text}"),
])

chain = prompt | llm.with_structured_output(TicketClassification)

result = chain.invoke({"ticket_text": "API timeout on payment webhook handler"})
# category='technical' priority='critical' confidence=0.97

Production pattern: fallback and observability

In production you need fallback when the primary model is degraded, and you need to log inputs/outputs for debugging. Here’s a runnable pattern:

import logging
import time
from typing import Optional
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.outputs import Generation

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

class StructuredOutputWithFallback:
    def __init__(self, primary_model: str, fallback_model: str):
        self.primary = ChatAnthropic(model=primary_model, temperature=0)
        self.fallback = ChatAnthropic(model=fallback_model, temperature=0)
        self.primary_structured = self.primary.with_structured_output(TriageResult)
        self.fallback_structured = self.fallback.with_structured_output(TriageResult)

    def invoke(self, ticket_text: str, config: Optional[RunnableConfig] = None) -> TriageResult:
        start = time.time()
        try:
            result = self.primary_structured.invoke(
                [HumanMessage(content=ticket_text)],
                config=config
            )
            logger.info(f"Primary succeeded in {time.time() - start:.2f}s")
            return result
        except Exception as e:
            logger.warning(f"Primary failed: {e}. Trying fallback.")
            try:
                result = self.fallback_structured.invoke(
                    [HumanMessage(content=ticket_text)],
                    config=config
                )
                logger.info(f"Fallback succeeded in {time.time() - start:.2f}s")
                return result
            except Exception as e2:
                logger.error(f"Both models failed: {e2}")
                raise

# Usage
triage = StructuredOutputWithFallback(
    primary_model="claude-3-5-sonnet-20241022",
    fallback_model="claude-3-haiku-20240307"
)

result = triage.invoke("Urgent: production database down, losing orders")

If you’re routing through a gateway like n4n.ai, the fallback happens at the infrastructure layer — you keep the same code and the gateway handles provider failover automatically.

Testing with deterministic outputs

For unit tests, use a fake model that returns canned structured responses:

from langchain_core.language_models import BaseChatModel
from langchain_core.messages import BaseMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from typing import List, Optional, Any

class FakeStructuredModel(BaseChatModel):
    def __init__(self, canned_response: TriageResult):
        self.canned = canned_response

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[Any] = None,
        **kwargs: Any,
    ) -> ChatResult:
        # Simulate tool call output
        from langchain_core.messages import AIMessage
        tool_call = {
            "name": "TriageResult",
            "args": self.canned.model_dump(),
            "id": "test-call-1",
            "type": "tool_call",
        }
        msg = AIMessage(content="", tool_calls=[tool_call])
        return ChatResult(generations=[ChatGeneration(message=msg)])

    @property
    def _llm_type(self) -> str:
        return "fake-structured"

# In tests
fake = FakeStructuredModel(canned_response=TriageResult(
    classification=TicketClassification(category="technical", priority="critical", confidence=1.0),
    entities=[],
    suggested_actions=[ActionType.ESCALATE],
    summary="Test summary",
    requires_human=True,
))

structured_fake = fake.with_structured_output(TriageResult)
result = structured_fake.invoke([HumanMessage(content="anything")])
assert result.classification.priority == "critical"

Common pitfalls

Temperature > 0 — Structured output needs determinism. Keep temperature=0 unless you have a specific reason not to.

Missing tool_choice — If the model returns text instead of a tool call, add tool_choice={"type": "function", "function": {"name": "YourModelName"}} to the bind call, or use with_structured_output(..., method="function_calling") (the default for Anthropic).

Schema drift — Changing Pydantic models without updating prompts causes silent failures. Version your schemas and include the version in the system prompt.

Large schemas — Claude 3.5 Sonnet handles ~200 fields comfortably. Beyond that, split into multiple calls or use a smaller focused schema.

When to use each approach

Next steps

• Add langsmith tracing to capture every structured call with inputs, outputs, and latency
• Build a schema registry so prompt versions and Pydantic versions stay in sync
• Consider instructor or jsonformer if you need strict JSON mode without tool calling overhead

The patterns above cover 90% of production structured-output needs with Claude 3.5 Sonnet and LangChain. Start minimal, add retries and fallback when you have real traffic, and keep your schemas versioned.