← Back to DevBytes

Building a Data Extraction Pipeline with Pydantic AI: Complete Guide

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:

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:

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.

— Ad —

Google AdSense will appear here after approval

← Back to all articles