Structured Output — JSON, Validation, Parsing

LLM: "Person is Anna, lives in Berlin, age 28, developer"

{
  "name": "Anna",
  "city": "Berlin",
  "age": 28,
  "job": "developer"
}

from anthropic import Anthropic

client = Anthropic()

prompt = """Extract info as JSON:
Name, city, age, job

Text: "I'm Anna. Live in Berlin. Age 28. Developer."

Response: JSON ONLY, no extra text."""

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}]
)

import json
data = json.loads(response.content[0].text)
print(f"Name: {data['name']}")

pip install instructor

import instructor
from anthropic import Anthropic
from pydantic import BaseModel

client = instructor.from_anthropic(Anthropic())

class Person(BaseModel):
    name: str
    city: str
    age: int
    job: str

# Automatic parsing + validation!
person = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    response_model=Person,
    messages=[
        {
            "role": "user",
            "content": "Extract: Name Anna, City Berlin, Age 28, Job Developer"
        }
    ]
)

print(f"{person.name} is {person.age} years old")
# Type checking works!
age_plus_5 = person.age + 5  # IDE knows this is int

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

class Address(BaseModel):
    street: str
    city: str

class Company(BaseModel):
    name: str
    employees: List[str]
    address: Address

class ExtractedData(BaseModel):
    companies: List[Company]
    total: int = Field(..., ge=0)  # Must be >= 0

# Instructor auto-parses nested structures
result = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    response_model=ExtractedData,
    messages=[...]
)

from pydantic import BaseModel, validator

class Person(BaseModel):
    age: int

    @validator("age")
    def age_valid(cls, v):
        if v < 0 or v > 150:
            raise ValueError("Age 0-150")
        return v

# Instructor retries automatically!
person = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    response_model=Person,
    messages=[{"role": "user", "content": "Age: 28"}]
)

from typing import Union, Literal

class Article(BaseModel):
    type: Literal["article"]
    title: str
    content: str

class Video(BaseModel):
    type: Literal["video"]
    title: str
    duration: int
    url: str

class Content(BaseModel):
    items: list[Union[Article, Video]]

# Either Article OR Video
result = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    response_model=Content,
    messages=[...]
)

for item in result.items:
    if isinstance(item, Article):
        print(f"Article: {item.title}")
    else:
        print(f"Video: {item.title} ({item.duration}s)")

# Better prompt
"Respond EXCLUSIVELY with JSON. No other text."

@validator("age")
def age_to_int(cls, v):
    if isinstance(v, str):
        return int(v)
    return v

# Instead of MegaSchema with 100 fields
# Use SectionA + SectionB + SectionC

import instructor
from anthropic import Anthropic
from pydantic import BaseModel

client = instructor.from_anthropic(Anthropic())

class Article(BaseModel):
    title: str
    summary: str
    keywords: list[str]

# Streaming: Real-time output
with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    response_model=Article,
    messages=[{
        "role": "user",
        "content": "Extract info from article: ..."
    }]
) as stream:
    for partial in stream:
        # Partial result (incomplete but structured)
        print(f"Title so far: {partial.title}")
        print(f"Keywords accumulated: {partial.keywords}")

    final_article = stream.get_final_message()

class ProductData(BaseModel):
    id: int = Field(..., gt=0)
    price: float = Field(..., ge=0)
    in_stock: bool

# Automatic retry on validation error
product = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    response_model=ProductData,
    max_retries=3,  # Retry up to 3 times
    messages=[{
        "role": "user",
        "content": "Extract: Product ID 123, Price $99.99, In stock: yes"
    }]
)

# If all retries fail, raises ValueError
# Process error in batch handler:

class BatchProcessor:
    def process_items(self, items: list[str]) -> tuple[list, list]:
        successes = []
        failures = []

        for item in items:
            try:
                result = client.messages.create(
                    model="claude-3-5-sonnet-20241022",
                    response_model=ProductData,
                    max_retries=2,
                    messages=[{"role": "user", "content": f"Extract: {item}"}]
                )
                successes.append(result)
            except ValueError as e:
                failures.append({"item": item, "error": str(e)})

        return successes, failures

from pydantic import BaseModel
from typing import Optional, Literal

class BlogPost(BaseModel):
    title: str
    category: Literal["tech", "business", "lifestyle"]
    published: bool
    estimated_reading_time: Optional[int] = None

# IDE knows BlogPost has these fields with these types
post = client.messages.create(...)
post.title  # IDE autocomplete works!
post.category  # Type is "tech" | "business" | "lifestyle"
post.reading_time  # Can be int or None

# This catches errors at write-time:
print(post.category.upper())  # OK
print(post.reading_time.upper())  # ERROR: int | None has no upper()

from pydantic import BaseModel, Field

class Concept(BaseModel):
    name: str
    definition: str
    related_concepts: list['Concept'] = Field(default_factory=list)

# Enable recursive models
Concept.model_rebuild()

result = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    response_model=Concept,
    messages=[{
        "role": "user",
        "content": "Extract knowledge graph: Machine Learning encompasses supervised learning, which includes classification and regression..."
    }]
)

# Now you have a tree:
print(result.name)  # "Machine Learning"
print(result.related_concepts[0].name)  # "Supervised Learning"
print(result.related_concepts[0].related_concepts[0].name)  # "Classification"

import asyncio
import instructor
from anthropic import AsyncAnthropic
from pydantic import BaseModel

client = instructor.from_anthropic(AsyncAnthropic())

class DataItem(BaseModel):
    extracted_value: str
    confidence: float

async def process_batch(items: list[str], batch_size: int = 5) -> list[DataItem]:
    """Process items concurrently, max 5 at a time."""
    results = []

    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]

        # Process batch concurrently
        tasks = [
            client.messages.create(
                model="claude-3-5-sonnet-20241022",
                response_model=DataItem,
                messages=[{"role": "user", "content": item}]
            )
            for item in batch
        ]

        batch_results = await asyncio.gather(*tasks)
        results.extend(batch_results)

    return results

# Usage
items = [f"Item {i}" for i in range(100)]
results = asyncio.run(process_batch(items))

1. Define Pydantic model
2. Use Instructor with streaming
3. response_model=YourSchema
4. Auto parsing + validation + retries
5. Type-safe data out
6. Batch process with concurrency
7. Handle failures gracefully