🎉 DevOps Interview Prep Bundle is live — 1000+ Q&A across 20 topicsGet it →
All Articles
mlopsaimonitoring

LLM Structured Outputs with Pydantic and Claude API in Production

Stop parsing JSON manually from LLM responses. Use Pydantic models with Claude API to get validated, typed structured outputs — with retry logic, partial failure handling, and real production patterns.

Shubham5 min read
Share:Tweet

The #1 source of LLM application bugs in production is malformed JSON from model responses. Models hallucinate fields, swap types, return markdown instead of JSON, or truncate responses mid-object. Pydantic + Claude API solves this properly.

The Problem

python
# Fragile code that breaks constantly
response = client.messages.create(...)
data = json.loads(response.content[0].text)  # Fails if model adds ```json wrapper
result = data["recommendations"]  # KeyError if model used different field name

The Solution: Claude API + Pydantic

Claude supports structured output natively via tool use — it forces the model to return valid JSON matching your schema.

python
pip install anthropic pydantic

Pattern 1: Tool Use for Strict Schemas

python
import anthropic
import json
from pydantic import BaseModel, Field, validator
from typing import Optional
 
 
class ResourceRecommendation(BaseModel):
    workload: str = Field(description="namespace/deployment/container")
    current_cpu: str = Field(description="Current CPU request e.g. '1000m'")
    recommended_cpu: str = Field(description="Recommended CPU request e.g. '250m'")
    current_memory: str = Field(description="Current memory request e.g. '2Gi'")
    recommended_memory: str = Field(description="Recommended memory request")
    monthly_savings_usd: float = Field(ge=0, description="Estimated monthly savings")
    confidence: str = Field(description="high|medium|low")
    reason: str = Field(description="One sentence explanation")
 
    @validator("confidence")
    def validate_confidence(cls, v):
        if v not in ("high", "medium", "low"):
            raise ValueError(f"confidence must be high/medium/low, got: {v}")
        return v
 
 
class OptimizationReport(BaseModel):
    recommendations: list[ResourceRecommendation]
    total_monthly_savings: float = Field(ge=0)
    critical_issues: list[str] = Field(default_factory=list)
    summary: str
 
 
def get_structured_analysis(workload_data: str) -> OptimizationReport:
    client = anthropic.Anthropic()
 
    # Define the schema as a Claude tool
    tool_schema = {
        "name": "submit_optimization_report",
        "description": "Submit the structured optimization analysis",
        "input_schema": OptimizationReport.model_json_schema()
    }
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=4000,
        tools=[tool_schema],
        tool_choice={"type": "tool", "name": "submit_optimization_report"},
        messages=[{
            "role": "user",
            "content": f"Analyze these Kubernetes workloads and submit an optimization report:\n\n{workload_data}"
        }]
    )
 
    # Extract tool use result
    for block in response.content:
        if block.type == "tool_use":
            # Pydantic validates automatically — raises ValidationError if invalid
            return OptimizationReport(**block.input)
 
    raise ValueError("Model did not call the expected tool")

Pattern 2: Retry with Validation Feedback

When Claude returns invalid data, send the error back and let it self-correct:

python
from pydantic import ValidationError
import time
 
 
def get_with_retry(
    prompt: str,
    model_class: type[BaseModel],
    max_retries: int = 3
) -> BaseModel:
    client = anthropic.Anthropic()
    last_error = None
 
    messages = [{"role": "user", "content": prompt}]
 
    for attempt in range(max_retries):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=2000,
            messages=messages
        )
 
        text = response.content[0].text.strip()
 
        # Extract JSON from markdown code blocks if present
        if "```" in text:
            parts = text.split("```")
            for i, part in enumerate(parts):
                if part.startswith("json\n"):
                    text = part[5:]
                    break
                elif i % 2 == 1:  # Odd indices are inside code blocks
                    text = part
                    break
 
        try:
            data = json.loads(text)
            return model_class(**data)
 
        except (json.JSONDecodeError, ValidationError) as e:
            last_error = e
            error_msg = str(e)
 
            # Add the failed response + error to messages for self-correction
            messages.append({"role": "assistant", "content": response.content[0].text})
            messages.append({
                "role": "user",
                "content": f"Your response had a validation error:\n{error_msg}\n\nPlease fix it and return valid JSON matching the schema."
            })
 
            if attempt < max_retries - 1:
                time.sleep(1)
 
    raise ValueError(f"Failed after {max_retries} retries. Last error: {last_error}")

Pattern 3: Partial Validation for Large Responses

When you expect a list and the model might return partial data:

python
from pydantic import BaseModel
from typing import Any
 
 
class PartialList(BaseModel):
    items: list[Any]
    is_complete: bool = True
 
 
def parse_list_response(text: str, item_class: type[BaseModel]) -> list[BaseModel]:
    """Parse a list response, validating each item individually."""
    text = text.strip()
 
    if "```" in text:
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
 
    try:
        raw_list = json.loads(text)
        if not isinstance(raw_list, list):
            raw_list = raw_list.get("items", raw_list.get("recommendations", []))
    except json.JSONDecodeError:
        return []
 
    valid_items = []
    for i, item in enumerate(raw_list):
        try:
            valid_items.append(item_class(**item))
        except ValidationError as e:
            print(f"Item {i} failed validation (skipping): {e}")
 
    return valid_items

Pattern 4: Enum Fields That Always Validate

python
from enum import Enum
 
 
class Severity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
 
 
class Alert(BaseModel):
    title: str
    severity: Severity   # Pydantic validates this is a valid enum value
    affected_service: str
    recommended_action: str
 
 
# Claude must return "critical", "high", "medium", or "low" for severity
# If it returns "CRITICAL" or "Critical", Pydantic handles case-insensitive matching
# for str Enums automatically

Production Checklist

python
# 1. Always set max_tokens high enough — truncated JSON fails validation
response = client.messages.create(max_tokens=4000, ...)
 
# 2. Use tool_choice to force structured output
tool_choice={"type": "tool", "name": "your_tool_name"}
 
# 3. Log validation failures for monitoring
import logging
try:
    result = MyModel(**data)
except ValidationError as e:
    logging.error("LLM validation failure", extra={"errors": e.errors(), "raw": text})
    raise
 
# 4. Set reasonable field constraints
class Report(BaseModel):
    score: float = Field(ge=0, le=100)       # 0-100 range
    items: list[str] = Field(max_length=20)   # No unbounded lists
    summary: str = Field(max_length=500)      # No runaway text
 
# 5. Use Optional for fields the model might omit
class Result(BaseModel):
    required_field: str
    optional_detail: Optional[str] = None   # Won't fail if model skips it

Testing Your Schema

Before shipping to production, test your schema with adversarial prompts:

python
import pytest
 
def test_validation_rejects_invalid_confidence():
    with pytest.raises(ValidationError):
        ResourceRecommendation(
            workload="prod/api/app",
            current_cpu="1000m",
            recommended_cpu="250m",
            current_memory="2Gi",
            recommended_memory="512Mi",
            monthly_savings_usd=150.0,
            confidence="very_high",  # Invalid!
            reason="Over-provisioned"
        )
 
def test_real_claude_response_validates():
    result = get_structured_analysis("Sample workload data...")
    assert isinstance(result, OptimizationReport)
    assert result.total_monthly_savings >= 0
    for rec in result.recommendations:
        assert rec.confidence in ("high", "medium", "low")

Structured outputs with Pydantic is the difference between an LLM application that works in demos and one that runs reliably at 3 AM in production.

More LLMOps patterns? Read our LLM output validation guide and LLM error handling and retry patterns.

Free Tools
Free Tool

CI/CD Pipeline Generator

Generate GitHub Actions, GitLab CI, or Jenkins pipelines for your stack instantly.

Free Tool

K8s Resource Calculator

Calculate CPU & memory requests/limits for your pods. Never get OOMKilled again.

🔧

Today I Fixed

Short real fixes from production — posted daily

Browse fixes
Back to all articles
Share:Tweet
Newsletter

Stay ahead of the curve

Get the latest DevOps, Kubernetes, AWS, and AI/ML guides delivered straight to your inbox. No spam — just practical engineering content.

Related Articles

aimlops

Build an AI SRE Incident Commander with Claude API

Step-by-step tutorial to build an AI incident commander that takes an alert, gathers context from Kubernetes and AWS, generates a structured runbook, and coordinates the incident response — using Claude API with tool use.

S
6 min readRead

Comments