Instructor extracts structured data from LLMs using Pydantic schemas. pip install instructor. import instructor, from anthropic import Anthropic. Patch: client = instructor.from_anthropic(Anthropic()). Response model: from pydantic import BaseModel, Field, class User(BaseModel): name: str; email: str; age: int = Field(ge=0, le=120). Extract: user = client.messages.create(model="claude-sonnet-4-6", max_tokens=1024, messages=[{"role":"user","content":"Extract: John Doe, [email protected], age 30"}], response_model=User). Access: user.name, user.email. OpenAI: import openai; client = instructor.from_openai(openai.OpenAI()). Nested: class Address(BaseModel): street: str; city: str; country: str. class Person(BaseModel): name: str; address: Address. Classification with Literal: class Sentiment(BaseModel): label: Literal["positive","neutral","negative"]; confidence: float. List extraction with Iterable: from typing import Iterable, users = client.messages.create(..., response_model=Iterable[User]) — streams list of validated objects. Partial streaming: response = client.messages.create_partial(model=..., response_model=User, ...) — yields partial User as tokens arrive. Retry: client.messages.create(..., response_model=User, max_retries=3) — auto-retries with validation error feedback. Async: client = instructor.from_anthropic(anthropic.AsyncAnthropic()), user = await client.messages.create(...). instructor.Mode.ANTHROPIC_TOOLS uses tool calling instead of JSON for models that support it. ValidationContext for runtime cross-field validation. Claude Code generates Instructor extraction classes, nested Pydantic schemas, classification models, batch extractors, and async pipelines.
CLAUDE.md for Instructor
## Instructor Stack
- Version: instructor >= 1.6
- Patch: instructor.from_anthropic(Anthropic()) or from_openai(OpenAI())
- Extract: client.messages.create(..., response_model=MyModel) → validated Pydantic instance
- Schema: Pydantic BaseModel with Field(description=...) guides the LLM
- Retry: max_retries=3 — LLM sees validation errors and self-corrects
- List: response_model=Iterable[Model] for extracting arrays
- Async: instructor.from_anthropic(AsyncAnthropic()) → await client.messages.create()
- Mode: instructor.Mode.ANTHROPIC_TOOLS (default for Anthropic) or JSONStructured Extraction Examples
# extraction/instructor_extract.py — structured LLM outputs with Instructor
from __future__ import annotations
import asyncio
import os
from enum import Enum
from typing import Iterable, Literal, Optional
import anthropic
import instructor
from pydantic import BaseModel, Field, field_validator, model_validator
# ── Patch the Anthropic client ────────────────────────────────────────────────
# Sync client
client = instructor.from_anthropic(
anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY", "")),
mode=instructor.Mode.ANTHROPIC_TOOLS, # Uses tool_use for reliable JSON
)
# Async client
async_client = instructor.from_anthropic(
anthropic.AsyncAnthropic(),
mode=instructor.Mode.ANTHROPIC_TOOLS,
)
MODEL = "claude-sonnet-4-6"
# ── 1. Simple entity extraction ───────────────────────────────────────────────
class ContactInfo(BaseModel):
"""Contact information extracted from unstructured text."""
name: str = Field(description="Full name of the person")
email: Optional[str] = Field(None, description="Email address if present")
phone: Optional[str] = Field(None, description="Phone number if present")
company: Optional[str] = Field(None, description="Company or organization")
role: Optional[str] = Field(None, description="Job title or role")
@field_validator("email")
@classmethod
def validate_email(cls, v: str | None) -> str | None:
if v and "@" not in v:
raise ValueError(f"Invalid email: {v}")
return v
def extract_contact(text: str) -> ContactInfo:
return client.messages.create(
model=MODEL,
max_tokens=512,
messages=[{"role": "user", "content": f"Extract contact information:\n\n{text}"}],
response_model=ContactInfo,
max_retries=3, # Auto-retry with validation errors fed back to LLM
)
# ── 2. Nested schema extraction ───────────────────────────────────────────────
class LineItem(BaseModel):
description: str
quantity: int = Field(ge=1)
unit_price: float = Field(ge=0)
total: float = Field(ge=0)
@model_validator(mode="after")
def check_total(self) -> "LineItem":
expected = round(self.quantity * self.unit_price, 2)
if abs(self.total - expected) > 0.02:
raise ValueError(f"Total {self.total} doesn't match qty * price = {expected}")
return self
class Invoice(BaseModel):
invoice_number: str
vendor: str
date: str = Field(description="Date in YYYY-MM-DD format")
line_items: list[LineItem]
subtotal: float
tax_rate: float = Field(ge=0, le=1, description="Tax rate as decimal (0.1 = 10%)")
total: float
@property
def calculated_total(self) -> float:
return round(self.subtotal * (1 + self.tax_rate), 2)
def extract_invoice(ocr_text: str) -> Invoice:
"""Extract structured invoice data from OCR text."""
return client.messages.create(
model=MODEL,
max_tokens=2048,
messages=[{
"role": "user",
"content": f"Extract all invoice details from this text:\n\n{ocr_text}"
}],
response_model=Invoice,
max_retries=4,
)
# ── 3. Classification with confidence ─────────────────────────────────────────
class SentimentClass(str, Enum):
POSITIVE = "positive"
NEUTRAL = "neutral"
NEGATIVE = "negative"
class SentimentResult(BaseModel):
label: SentimentClass
confidence: float = Field(ge=0, le=1, description="Confidence score between 0 and 1")
rationale: str = Field(description="One-sentence explanation of the classification")
class TicketClassification(BaseModel):
"""Support ticket classification for routing."""
priority: Literal["critical", "high", "medium", "low"]
category: Literal["billing", "technical", "account", "feature_request", "other"]
department: Literal["support", "engineering", "billing", "product"]
summary: str = Field(max_length=100, description="One-line summary of the issue")
requires_followup: bool
def classify_support_ticket(ticket_text: str) -> TicketClassification:
return client.messages.create(
model=MODEL,
max_tokens=512,
system="You are a support ticket classifier. Always classify accurately.",
messages=[{"role": "user", "content": f"Classify this support ticket:\n\n{ticket_text}"}],
response_model=TicketClassification,
)
# ── 4. List / batch extraction ────────────────────────────────────────────────
class SkillMention(BaseModel):
skill: str
level: Optional[Literal["beginner", "intermediate", "advanced", "expert"]] = None
years: Optional[float] = Field(None, description="Years of experience")
context: str = Field(description="Short quote or context from resume")
def extract_skills_from_resume(resume_text: str) -> list[SkillMention]:
"""Extract all skills mentioned in a resume as a list."""
skills = client.messages.create(
model=MODEL,
max_tokens=2000,
messages=[{
"role": "user",
"content": f"Extract all technical skills from this resume. Return ALL skills mentioned:\n\n{resume_text}"
}],
response_model=Iterable[SkillMention], # Returns generator of SkillMention
)
return list(skills)
# ── 5. Async extraction ───────────────────────────────────────────────────────
async def extract_batch_async(
texts: list[str],
response_model: type[BaseModel],
concurrency: int = 5,
) -> list[BaseModel]:
"""Extract structured data from multiple texts concurrently."""
semaphore = asyncio.Semaphore(concurrency)
async def extract_one(text: str) -> BaseModel:
async with semaphore:
return await async_client.messages.create(
model=MODEL,
max_tokens=1024,
messages=[{"role": "user", "content": text}],
response_model=response_model,
max_retries=2,
)
results = await asyncio.gather(*[extract_one(t) for t in texts], return_exceptions=True)
return [r for r in results if not isinstance(r, Exception)]
# ── 6. Streaming partial extraction ──────────────────────────────────────────
def stream_extraction_demo():
"""Stream a partial model — yields incomplete but typed objects as tokens arrive."""
print("Streaming extraction (partial objects):")
for partial_contact in client.messages.create_partial(
model=MODEL,
max_tokens=512,
messages=[{
"role": "user",
"content": "Extract: Alice Smith, [email protected], CTO at TechCorp, +1-555-0100"
}],
response_model=ContactInfo,
):
# Partial object — fields may be None until fully generated
print(f" name={partial_contact.name!r:<25} email={partial_contact.email!r}")
# ── Usage examples ────────────────────────────────────────────────────────────
if __name__ == "__main__":
# Extract contact
contact = extract_contact("Please reach out to Sarah Connor at [email protected], she's the VP of Engineering at Skynet Inc, phone 555-1234.")
print(f"Contact: {contact.name}, {contact.email}, {contact.role} @ {contact.company}")
# Classify ticket
ticket = classify_support_ticket(
"URGENT: Our production API is returning 500 errors since the last deployment. "
"Affecting all customers. Team is blocked."
)
print(f"Ticket: priority={ticket.priority}, dept={ticket.department}, summary={ticket.summary}")
# Extract skills
resume = "Senior Python engineer with 8 years experience. Expert in PyTorch, advanced Kubernetes, intermediate Go. Led 3 ML projects using scikit-learn and MLflow."
skills = extract_skills_from_resume(resume)
for s in skills:
print(f" - {s.skill} ({s.level}, {s.years}yr)")
# Streaming
stream_extraction_demo()
# Async batch
contacts_text = [
"Bob Johnson, [email protected], Sales Manager",
"Carol White, [email protected], Founder, +44-20-1234",
]
contacts = asyncio.run(extract_batch_async(contacts_text, ContactInfo))
for c in contacts:
print(f"Async: {c.name}, {c.email}")For the direct JSON mode (OpenAI response_format=json_object or Anthropic prefill technique) alternative when using a model without function calling support or wanting to avoid SDK dependencies — raw JSON prompting works but loses schema validation, retry logic, and the LLM-readable Pydantic field descriptions that guide extraction quality, whereas Instructor’s max_retries feeds validation errors back to the model so it can self-correct in 2-3 additional API calls. For the LangChain with_structured_output alternative when already using LangChain for chain composition — LangChain’s structured output works differently per model while Instructor uses the same Pydantic-first API across all providers (Anthropic, OpenAI, Gemini, Cohere, Ollama) making it easier to switch LLM providers without changing extraction logic. The Claude Skills 360 bundle includes Instructor skill sets covering entity extraction schemas, nested models, classification with Literal, list extraction, async batch processing, and streaming partial models. Start with the free tier to try structured LLM output generation.