Introduction to Pydantic AI for Data Extraction
Pydantic AI is a relatively new framework that brings type safety, structured outputs, and developer-friendly abstractions to working with Large Language Models (LLMs). Built by the same team behind the widely adopted Pydantic library, it leverages Python's type hints to define agent behaviors, validate model responses, and orchestrate complex workflows. For data extraction tasks—where precision and schema conformance are critical—Pydantic AI offers a compelling alternative to ad-hoc prompt engineering.
In this tutorial, you'll learn how to build a robust data extraction pipeline using Pydantic AI. We'll cover everything from defining schemas and agents to handling validation errors, streaming structured outputs, and deploying in production. By the end, you'll have a complete, working pipeline capable of extracting structured information from unstructured text sources.
What Is Pydantic AI?
Pydantic AI is a Python agent framework designed to make LLM-based application development feel like writing ordinary Python code. It introduces the concept of Agent objects that are parameterized with typed result models. Instead of parsing free-form text responses, you declare the shape of the data you want, and the framework ensures the LLM returns data that conforms to that schema.
Key features include:
- Type-safe outputs: Results are validated against Pydantic models, guaranteeing structure.
- Multi-provider support: Works with OpenAI, Anthropic, Gemini, Groq, Ollama, and others.
- Dependency injection: Pass databases, APIs, or other resources into agents cleanly.
- Structured retries: Automatically re-prompts the model when validation fails.
- Streaming support: Stream partial validated results as they arrive.
- Observability: Integrates with Logfire for tracing and debugging.
Why It Matters for Data Extraction
Data extraction is one of the most common LLM use cases, but it's also one of the most error-prone. Traditional approaches involve crafting prompts that ask the model to return JSON, then wrapping that JSON in try/except blocks and hoping for the best. This leads to brittle pipelines that fail silently or produce inconsistent data.
Pydantic AI addresses these issues by:
- Generating JSON schemas from your Python type hints and passing them to the model automatically.
- Validating responses with Pydantic's powerful validation engine.
- Re-prompting the model with validation errors when outputs don't conform, dramatically improving reliability.
- Providing a clean separation between the extraction logic, the schema, and the prompt.
This means you spend less time debugging malformed JSON and more time refining your extraction logic and schemas.
Setting Up Your Environment
Before we start building, let's set up a clean environment. Create a new directory for your project and install the required packages:
mkdir pydantic-ai-extraction
cd pydantic-ai-extraction
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
pip install pydantic-ai pydantic python-dotenv
You'll also need an API key for your chosen LLM provider. For this tutorial, we'll use OpenAI, but the patterns apply to any supported provider. Create a .env file:
OPENAI_API_KEY=sk-your-api-key-here
Pydantic AI will automatically read environment variables for provider keys, so no extra configuration is needed beyond setting them.
Defining Your Extraction Schema
The foundation of any extraction pipeline is the schema. In Pydantic AI, schemas are just Pydantic models. Let's build a pipeline that extracts structured information from job postings. We'll start by defining the models that represent the data we want to extract.
from typing import Optional, List
from enum import Enum
from pydantic import BaseModel, Field, HttpUrl
class EmploymentType(str, Enum):
FULL_TIME = "full_time"
PART_TIME = "part_time"
CONTRACT = "contract"
INTERNSHIP = "internship"
class SalaryRange(BaseModel):
"""Represents a salary range for a job posting."""
minimum: Optional[float] = Field(
None, description="Minimum annual salary in USD"
)
maximum: Optional[float] = Field(
None, description="Maximum annual salary in USD"
)
currency: str = Field(
default="USD", description="ISO currency code"
)
class JobRequirement(BaseModel):
"""A single requirement listed in a job posting."""
skill: str = Field(description="The skill or qualification name")
is_required: bool = Field(
default=True, description="Whether this is a must-have or nice-to-have"
)
years_of_experience: Optional[float] = Field(
None, description="Years of experience required, if specified"
)
class JobPosting(BaseModel):
"""Structured representation of a job posting."""
title: str = Field(description="The job title as stated in the posting")
company: str = Field(description="The hiring company name")
location: Optional[str] = Field(
None, description="Job location, including remote if applicable"
)
is_remote: bool = Field(
default=False, description="Whether the role is remote"
)
employment_type: EmploymentType = Field(
default=EmploymentType.FULL_TIME,
description="Type of employment"
)
salary: Optional[SalaryRange] = Field(
None, description="Salary information if provided"
)
requirements: List[JobRequirement] = Field(
default_factory=list,
description="List of skills and qualifications required"
)
description: str = Field(
description="A brief summary of the role"
)
application_url: Optional[HttpUrl] = Field(
None, description="URL to apply, if available"
)
Notice how every field includes a description. These descriptions are not just documentation—they're passed to the LLM as part of the JSON schema, guiding the model to fill in each field correctly. This is one of the most important practices in Pydantic AI: descriptive field documentation directly improves extraction quality.
Creating Your First Extraction Agent
Now that we have a schema, let's create an agent that uses it. An agent in Pydantic AI encapsulates the model, the system prompt, and the expected output type.
from pydantic_ai import Agent
job_extraction_agent = Agent(
model="openai:gpt-4o",
result_type=JobPosting,
system_prompt=(
"You are a precise data extraction assistant specialized in parsing "
"job postings. Extract all available information into the structured "
"schema. If a field is not mentioned in the text, leave it as null "
"or its default value. Do not invent or guess information that is "
"not present in the source text. For requirements, distinguish "
"between 'required' and 'preferred' or 'nice-to-have' qualifications."
),
)
The result_type parameter tells the agent what structure to expect. Pydantic AI handles the JSON schema generation and validation automatically. Let's use this agent to extract data from a sample job posting:
import asyncio
from pydantic_ai import Agent
sample_posting = """
Senior Backend Engineer at TechFlow Inc.
Location: San Francisco, CA (Hybrid, 3 days in office)
Employment: Full-time
About the Role:
We're looking for a senior backend engineer to join our platform team.
You'll be building scalable APIs and microservices that power our
real-time analytics product.
Requirements:
- 5+ years of Python experience (required)
- Strong knowledge of PostgreSQL and Redis (required)
- Experience with Kubernetes (preferred)
- Familiarity with event-driven architecture (nice-to-have)
Compensation: $160,000 - $200,000 USD annually
Apply at https://techflow.example.com/careers/senior-backend-engineer
"""
async def extract_job_posting(text: str) -> JobPosting:
result = await job_extraction_agent.run(text)
return result.data
async def main():
posting = await extract_job_posting(sample_posting)
print(f"Title: {posting.title}")
print(f"Company: {posting.company}")
print(f"Location: {posting.location}")
print(f"Remote: {posting.is_remote}")
print(f"Type: {posting.employment_type}")
if posting.salary:
print(f"Salary: ${posting.salary.minimum} - ${posting.salary.maximum}")
print(f"\nRequirements ({len(posting.requirements)}):")
for req in posting.requirements:
status = "Required" if req.is_required else "Preferred"
exp = f" ({req.years_of_experience} years)" if req.years_of_experience else ""
print(f" - {req.skill} [{status}]{exp}")
print(f"\nApply: {posting.application_url}")
if __name__ == "__main__":
asyncio.run(main())
When you run this script, the agent sends the text to the LLM along with the JSON schema derived from JobPosting. The model returns structured JSON, which Pydantic validates against the model. If validation fails, Pydantic AI automatically retries with the error message included in the next prompt.
Adding Validation with Field Validators
Schema-level validation catches structural issues, but you often need business-logic validation too. Pydantic's @field_validator and @model_validator decorators let you enforce custom rules. These validators run after the LLM returns data, and if they fail, the error is fed back to the model for correction.
from pydantic import field_validator, model_validator
class JobPosting(BaseModel):
title: str = Field(description="The job title")
company: str = Field(description="The hiring company name")
location: Optional[str] = Field(None, description="Job location")
is_remote: bool = Field(default=False)
employment_type: EmploymentType = Field(default=EmploymentType.FULL_TIME)
salary: Optional[SalaryRange] = Field(None)
requirements: List[JobRequirement] = Field(default_factory=list)
description: str = Field(description="Brief summary of the role")
application_url: Optional[HttpUrl] = Field(None)
@field_validator("title")
@classmethod
def title_must_not_be_empty(cls, v: str) -> str:
if not v or not v.strip():
raise ValueError("Job title must not be empty")
return v.strip()
@field_validator("company")
@classmethod
def normalize_company_name(cls, v: str) -> str:
# Remove common suffixes and normalize
cleaned = v.strip()
for suffix in [" Inc.", " Inc", " LLC", " Ltd.", " Ltd"]:
if cleaned.endswith(suffix):
cleaned = cleaned[: -len(suffix)]
return cleaned.strip()
@model_validator(mode="after")
def check_salary_consistency(self):
if self.salary and self.salary.minimum and self.salary.maximum:
if self.salary.minimum > self.salary.maximum:
raise ValueError(
f"Salary minimum ({self.salary.minimum}) cannot be "
f"greater than maximum ({self.salary.maximum})"
)
return self
When a validator raises a ValueError, Pydantic AI catches it and sends the error back to the LLM with instructions to fix the issue. This retry loop is configurable via the retries parameter on the agent:
job_extraction_agent = Agent(
model="openai:gpt-4o",
result_type=JobPosting,
retries=3,
system_prompt="You are a precise data extraction assistant...",
)
Building a Batch Processing Pipeline
In real-world scenarios, you rarely extract data from a single document. Let's build a pipeline that processes multiple job postings concurrently, handles errors gracefully, and collects results.
import asyncio
from dataclasses import dataclass, field
from typing import List, Tuple
from pydantic_ai import Agent
from pydantic_ai.usage import UsageLimits
@dataclass
class ExtractionResult:
success: bool
data: Optional[JobPosting] = None
error: Optional[str] = None
source_text: str = ""
async def extract_single(
agent: Agent,
text: str,
semaphore: asyncio.Semaphore
) -> ExtractionResult:
"""Extract a single job posting with concurrency control."""
async with semaphore:
try:
result = await agent.run(
text,
usage_limits=UsageLimits(request_limit=1)
)
return ExtractionResult(
success=True,
data=result.data,
source_text=text
)
except Exception as e:
return ExtractionResult(
success=False,
error=str(e),
source_text=text
)
async def extract_batch(
texts: List[str],
agent: Agent,
max_concurrent: int = 5
) -> List[ExtractionResult]:
"""Process a batch of texts with controlled concurrency."""
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [extract_single(agent, text, semaphore) for text in texts]
return await asyncio.gather(*tasks)
async def run_pipeline():
postings = [
sample_posting,
"""
Junior Data Analyst at DataCorp
Remote, Part-time
Looking for someone with Excel and SQL skills.
$25/hour
""",
"""
DevOps Contractor at CloudScale
6-month contract, remote
Need AWS and Terraform experience, 3+ years.
Rate: $80-100/hr
""",
]
results = await extract_batch(postings, job_extraction_agent, max_concurrent=3)
successes = [r for r in results if r.success]
failures = [r for r in results if not r.success]
print(f"Processed {len(results)} postings")
print(f" Successful: {len(successes)}")
print(f" Failed: {len(failures)}")
for result in successes:
posting = result.data
print(f"\n--- {posting.title} at {posting.company} ---")
print(f" Type: {posting.employment_type.value}")
print(f" Remote: {posting.is_remote}")
if posting.salary:
print(f" Salary: {posting.salary.minimum}-{posting.salary.maximum} {posting.salary.currency}")
for result in failures:
print(f"\nExtraction failed: {result.error}")
if __name__ == "__main__":
asyncio.run(run_pipeline())
The asyncio.Semaphore limits concurrent API calls, preventing rate limit issues. The UsageLimits object caps the number of requests per extraction, protecting against runaway retry loops.
Using Dependency Injection for Enrichment
Often, extraction isn't just about parsing text—you need to enrich the data by looking things up in databases or external APIs. Pydantic AI supports dependency injection, allowing you to pass resources into agent tools.
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
@dataclass
class CompanyDatabase:
"""Simulated database for company lookups."""
companies: dict
def lookup(self, name: str) -> dict:
return self.companies.get(name.lower(), {})
# Define the agent with typed dependencies
enrichment_agent = Agent(
model="openai:gpt-4o",
result_type=JobPosting,
deps_type=CompanyDatabase,
system_prompt=(
"You are a data extraction assistant. Extract job posting "
"information and use the lookup_company tool to enrich company "
"data when available."
),
)
@enrichment_agent.tool
async def lookup_company(ctx: RunContext[CompanyDatabase], name: str) -> dict:
"""Look up company information by name.
Args:
name: The company name to search for.
"""
return ctx.deps.lookup(name)
async def run_with_enrichment():
db = CompanyDatabase(companies={
"techflow": {
"industry": "SaaS",
"size": "50-200",
"founded": 2019,
},
"datacorp": {
"industry": "Consulting",
"size": "10-50",
"founded": 2021,
},
})
result = await enrichment_agent.run(sample_posting, deps=db)
posting = result.data
print(f"Extracted: {posting.title} at {posting.company}")
print(f"Requirements: {len(posting.requirements)} items")
The RunContext provides typed access to your dependencies inside tool functions. The LLM can decide when to call these tools, making the extraction process more intelligent and context-aware.
Streaming Partial Results
For long documents or interactive applications, streaming partial results improves user experience. Pydantic AI supports streaming validated partial outputs:
async def stream_extraction(text: str):
"""Stream partial extraction results as they arrive."""
async with job_extraction_agent.run_stream(text) as result:
print("Streaming partial results...\n")
async for posting in result.stream():
# Each iteration gives a partially-filled JobPosting
print(f" Title: {posting.title or '...'}")
print(f" Company: {posting.company or '...'}")
print(f" Requirements so far: {len(posting.requirements)}")
print(" ---")
# Final validated result
final = await result.get_data()
print(f"\nFinal result: {final.title} at {final.company}")
print(f"Total requirements: {len(final.requirements)}")
async def main():
await stream_extraction(sample_posting)
if __name__ == "__main__":
asyncio.run(main())
Streaming is particularly useful when building user interfaces that show extraction progress, or when processing very large documents where you want to display results incrementally.
Handling Edge Cases and Error Recovery
Production pipelines must handle edge cases gracefully. Here are common scenarios and how to address them:
Empty or Irrelevant Input
async def safe_extract(agent: Agent, text: str) -> Optional[JobPosting]:
"""Extract with comprehensive error handling."""
if not text or len(text.strip()) < 20:
print("Input too short to be a valid job posting")
return None
try:
result = await agent.run(text)
return result.data
except Exception as e:
print(f"Extraction failed after retries: {e}")
# Optionally log to an error tracking service
return None
Model Fallback Strategy
async def extract_with_fallback(text: str) -> Optional[JobPosting]:
"""Try a fast model first, fall back to a more capable one."""
# Fast, cheaper model
fast_agent = Agent(
model="openai:gpt-4o-mini",
result_type=JobPosting,
retries=2,
system_prompt="You are a precise data extraction assistant...",
)
try:
result = await fast_agent.run(text)
return result.data
except Exception:
print("Fast model failed, falling back to GPT-4o...")
# More capable model
powerful_agent = Agent(
model="openai:gpt-4o",
result_type=JobPosting,
retries=3,
system_prompt="You are a precise data extraction assistant...",
)
try:
result = await powerful_agent.run(text)
return result.data
except Exception as e:
print(f"Both models failed: {e}")
return None
Logging and Observability
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("extraction_pipeline")
async def extract_with_logging(text: str, agent: Agent) -> Optional[JobPosting]:
logger.info(f"Starting extraction for text of length {len(text)}")
try:
result = await agent.run(text)
posting = result.data
logger.info(
f"Extraction successful: {posting.title} at {posting.company}"
)
logger.info(f"Usage: {result.usage()}")
return posting
except Exception as e:
logger.error(f"Extraction failed: {e}", exc_info=True)
return None
The result.usage() method returns token counts and cost information, which is essential for monitoring and budgeting in production.
Best Practices
1. Write Detailed Field Descriptions
Every field in your schema should have a clear, unambiguous description. This is the single most impactful thing you can do to improve extraction quality. Compare these two approaches:
# Bad - vague, no guidance
class BadSchema(BaseModel):
name: str
value: float
# Good - descriptive, guides the model
class GoodSchema(BaseModel):
name: str = Field(
description="The full legal name of the product as it appears on packaging"
)
value: float = Field(
description="The retail price in USD, excluding tax and shipping"
)
2. Use Enums for Categorical Fields
Enums constrain the model's output to valid values and reduce hallucination:
class SeniorityLevel(str, Enum):
ENTRY = "entry"
JUNIOR = "junior"
MID = "mid"
SENIOR = "senior"
LEAD = "lead"
EXECUTIVE = "executive"
3. Keep Schemas Focused
Avoid creating massive schemas with dozens of fields. If you need to extract a lot of data, consider breaking the extraction into multiple agents, each focused on a specific aspect:
# Agent 1: Extract basic job info
basic_info_agent = Agent(
model="openai:gpt-4o-mini",
result_type=BasicJobInfo,
system_prompt="Extract title, company, and location only.",
)
# Agent 2: Extract requirements
requirements_agent = Agent(
model="openai:gpt-4o",
result_type=List[JobRequirement],
system_prompt="Extract all skills and qualifications listed.",
)
# Agent 3: Extract compensation
compensation_agent = Agent(
model="openai:gpt-4o-mini",
result_type=Optional[SalaryRange],
system_prompt="Extract salary information if available.",
)
4. Set Reasonable Retry Limits
While retries improve reliability, unlimited retries can lead to excessive costs. A value of 2-3 retries is usually sufficient. Monitor failure rates and adjust your prompts or schemas if failures are frequent.
5. Test with Diverse Inputs
Create a test suite with varied inputs—well-formatted postings, messy text, incomplete data, and edge cases. This helps you identify schema weaknesses before production:
import pytest
@pytest.mark.asyncio
async def test_extract_complete_posting():
result = await extract_job_posting(sample_posting)
assert result.title == "Senior Backend Engineer"
assert result.company == "TechFlow"
assert result.is_remote is False
assert len(result.requirements) == 4
assert result.salary.minimum == 160000
assert result.salary.maximum == 200000
@pytest.mark.asyncio
async def test_extract_minimal_posting():
minimal = "Developer at Startup. Full-time."
result = await extract_job_posting(minimal)
assert result.title == "Developer"
assert result.company == "Startup"
assert result.employment_type == EmploymentType.FULL_TIME
assert result.requirements == []
@pytest.mark.asyncio
async def test_extract_remote_posting():
remote = "Engineer at RemoteCo. Fully remote position."
result = await extract_job_posting(remote)
assert result.is_remote is True
6. Cache Results to Avoid Redundant API Calls
import hashlib
import json
from functools import lru_cache
def make_cache_key(text: str, model: str) -> str:
content_hash = hashlib.sha256(text.encode()).hexdigest()
return f"{model}:{content_hash}"
class ExtractionCache:
def __init__(self):
self._cache = {}
async def get_or_extract(
self, agent: Agent, text: str, model_name: str = "gpt-4o"
) -> JobPosting:
key = make_cache_key(text, model_name)
if key in self._cache:
return self._cache[key]
result = await agent.run(text)
self._cache[key] = result.data
return result.data
Putting It All Together: Complete Pipeline
Let's assemble everything into a complete, production-ready pipeline module:
"""
Job Posting Extraction Pipeline
A complete data extraction pipeline using Pydantic AI.
"""
import asyncio
import logging
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator
from pydantic_ai import Agent, RunContext
from pydantic_ai.usage import UsageLimits
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("extraction_pipeline")
# --- Schemas ---
class EmploymentType(str, Enum):
FULL_TIME = "full_time"
PART_TIME = "part_time"
CONTRACT = "contract"
INTERNSHIP = "internship"
class SalaryRange(BaseModel):
minimum: Optional[float] = Field(None, description="Minimum annual salary in USD")
maximum: Optional[float] = Field(None, description="Maximum annual salary in USD")
currency: str = Field(default="USD", description="ISO currency code")
@model_validator(mode="after")
def check_range(self):
if self.minimum and self.maximum and self.minimum > self.maximum:
raise ValueError("Salary minimum cannot exceed maximum")
return self
class JobRequirement(BaseModel):
skill: str = Field(description="The skill or qualification name")
is_required: bool = Field(default=True, description="Required vs. preferred")
years_of_experience: Optional[float] = Field(
None, description="Years of experience required, if specified"
)
class JobPosting(BaseModel):
title: str = Field(description="The job title as stated in the posting")
company: str = Field(description="The hiring company name")
location: Optional[str] = Field(None, description="Job location")
is_remote: bool = Field(default=False, description="Whether the role is remote")
employment_type: EmploymentType = Field(
default=EmploymentType.FULL_TIME, description="Type of employment"
)
salary: Optional[SalaryRange] = Field(None, description="Salary information")
requirements: List[JobRequirement] = Field(
default_factory=list, description="List of required skills and qualifications"
)
description: str = Field(description="A brief summary of the role")
application_url: Optional[HttpUrl] = Field(None, description="URL to apply")
@field_validator("title")
@classmethod
def validate_title(cls, v: str) -> str:
if not v or not v.strip():
raise ValueError("Job title must not be empty")
return v.strip()
# --- Agent Definition ---
extraction_agent = Agent(
model="openai:gpt-4o",
result_type=JobPosting,
retries=3,
system_prompt=(
"You are a precise data extraction assistant specialized in parsing "
"job postings. Extract all available information into the structured "
"schema. If a field is not mentioned, leave it as null or default. "
"Do not invent information not present in the source text. "
"Distinguish between 'required' and 'preferred' qualifications."
),
)
# --- Pipeline Components ---
@dataclass
class ExtractionResult:
success: bool
data: Optional[JobPosting] = None
error: Optional[str] = None
tokens_used: int = 0
async def extract_single(text: str, semaphore: asyncio.Semaphore) -> ExtractionResult:
if not text or len(text.strip()) < 20:
return ExtractionResult(success=False, error="Input too short")
async with semaphore:
try:
result = await extraction_agent.run(
text,
usage_limits=UsageLimits(request_limit=5)
)
usage = result.usage()
return ExtractionResult(
success=True,
data=result.data,
tokens_used=usage.total_tokens if usage else 0,
)
except Exception as e:
logger.error(f"Extraction failed: {e}")
return ExtractionResult(success=False, error=str(e))
async def run_pipeline(texts: List[str], max_concurrent: int = 5) -> List[ExtractionResult]:
logger.info(f"Starting pipeline with {len(texts)} documents")
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [extract_single(text, semaphore) for text in texts]
results = await asyncio.gather(*tasks)
successes = sum(1 for r in results if r.success)
total_tokens = sum(r.tokens_used for r in results)
logger.info(
f"Pipeline complete: {successes}/{len(results)} successful, "
f"{total_tokens} tokens used"
)
return results
# --- Main Entry Point ---
async def main():
sample_texts = [
"""
Senior Backend Engineer at TechFlow Inc.
Location: San Francisco, CA (Hybrid)
Full-time
We need a Python expert with 5+ years experience.
PostgreSQL and Redis required. Kubernetes preferred.
Salary: $160,000 - $200,000 USD
Apply at https://techflow.example.com/careers/senior-backend-engineer
""",
"""
Junior Data Analyst at DataCorp
Remote, Part-time
Excel and SQL skills needed.
$25/hour
""",
"""
DevOps Contractor at CloudScale
6-month contract, fully remote
AWS and Terraform, 3+ years.
Rate: $80-100/hr
""",
]
results = await run_pipeline(sample_texts, max_concurrent=3)
for i, result in enumerate(results, 1):
if result.success and result.data:
p = result.data
print(f"\n[{i}] {p.title} at {p.company}")
print(f" Type: {p.employment_type.value} | Remote: {p.is_remote}")
if p.salary and p.salary.minimum:
print(f" Salary: ${p.salary.minimum:,.0f} - ${p.salary.maximum:,.0f}")
print(f" Requirements: {len(p.requirements)}")
print(f" Tokens: {result.tokens_used}")
else:
print(f"\n[{i}] FAILED: {result.error}")
if __name__ == "__main__":
asyncio.run(main())
Conclusion
Building a data extraction pipeline with Pydantic AI transforms what is traditionally a fragile, prompt-engineering-heavy process into a structured, type-safe, and maintainable workflow. By defining clear Pydantic schemas with descriptive field annotations, leveraging automatic validation and retry mechanisms, and incorporating best practices like concurrency control, dependency injection, and comprehensive error handling, you can build extraction systems that are reliable enough for production use. The framework's philosophy—treating LLM interactions as typed Python functions rather than string manipulation exercises—means your code stays readable, testable, and resilient. As you scale your pipeline, remember that the quality of your schemas and field descriptions will have the largest impact on extraction accuracy, so invest time in iterating on those alongside your prompts. With the patterns covered in this guide, you're well-equipped to build extraction pipelines for job postings, invoices, medical records, legal documents, or any other domain where turning unstructured text into structured data is the goal.