Building a Data Extraction Pipeline with OpenAI Agents SDK: Complete Guide

Building a Data Extraction Pipeline with OpenAI Agents SDK: Complete Guide

Modern applications increasingly rely on extracting structured information from messy, unstructured sources — emails, PDFs, web pages, customer support tickets, and more. Traditional rule-based extraction pipelines are brittle and require constant maintenance. The OpenAI Agents SDK changes this paradigm by letting you orchestrate LLM-powered agents that can reason, call tools, and hand off tasks to one another. In this guide, you'll build a complete, production-ready data extraction pipeline using the Agents SDK.

What Is the OpenAI Agents SDK?

The OpenAI Agents SDK is a lightweight framework for building agentic AI workflows in Python. It provides primitives like Agent, Runner, Tool, and Handoff that let you compose multi-step reasoning pipelines. Unlike raw chat completions, the SDK handles the loop of calling the model, executing tools, feeding results back, and terminating when an agent produces a final structured output.

For data extraction, this means you can build agents that:

• Parse unstructured text and emit typed, validated objects.
• Call external tools to enrich or validate extracted data.
• Hand off specialized sub-tasks (e.g., address normalization, currency conversion) to dedicated agents.
• Run guardrails to reject malformed or unsafe inputs.

Why It Matters

Extraction pipelines built on the Agents SDK offer several advantages over traditional approaches:

• Flexibility: Agents adapt to schema variations without retraining.
• Composability: Specialized agents can be reused across pipelines.
• Structured outputs: Pydantic models guarantee type safety downstream.
• Observability: Built-in tracing shows every tool call and reasoning step.
• Guardrails: Input and output validation run automatically around each agent.

Prerequisites and Setup

Install the SDK and supporting libraries:

pip install openai-agents pydantic python-dotenv

Create a .env file with your API key:

OPENAI_API_KEY=sk-your-key-here

Then set up your project entry point:

import os
from dotenv import load_dotenv
from agents import Agent, Runner, function_tool
from pydantic import BaseModel, Field

load_dotenv()

Defining Your Extraction Schema

The foundation of any extraction pipeline is a clear output schema. Pydantic models serve as both the contract and the validation layer. Let's model an invoice extraction pipeline that pulls structured data from free-text invoice descriptions.

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

class LineItem(BaseModel):
    description: str = Field(description="Description of the product or service")
    quantity: int = Field(description="Number of units", ge=1)
    unit_price: float = Field(description="Price per unit in USD", ge=0)
    total: float = Field(description="Line total in USD", ge=0)

class InvoiceData(BaseModel):
    vendor_name: str = Field(description="Name of the vendor or supplier")
    invoice_number: str = Field(description="Unique invoice identifier")
    invoice_date: str = Field(description="Invoice issue date in YYYY-MM-DD format")
    due_date: Optional[str] = Field(
        default=None, description="Payment due date in YYYY-MM-DD format"
    )
    line_items: List[LineItem] = Field(description="Itemized charges")
    subtotal: float = Field(description="Sum of line items before tax", ge=0)
    tax_amount: float = Field(default=0.0, description="Total tax in USD", ge=0)
    total_amount: float = Field(description="Final amount due in USD", ge=0)
    currency: str = Field(default="USD", description="ISO 4217 currency code")

Using Field(description=...) is critical — the SDK passes these descriptions to the model as part of the structured output schema, dramatically improving extraction accuracy.

Creating the Primary Extraction Agent

With the schema defined, create the main extraction agent. The output_type parameter tells the SDK to enforce the Pydantic model as the final response.

extraction_agent = Agent(
    name="InvoiceExtractor",
    instructions=(
        "You are a precise invoice data extraction specialist. "
        "Given raw invoice text, extract all fields into the InvoiceData schema. "
        "If a field is missing, leave it null or use the default. "
        "Always compute line item totals as quantity * unit_price. "
        "Verify that subtotal equals the sum of line item totals. "
        "If the math does not match, recompute and use the corrected subtotal. "
        "Never invent data that is not present in the source text."
    ),
    output_type=InvoiceData,
    model="gpt-4o-mini",
)

The instructions encode domain rules that the model must follow. Be explicit about calculations, defaults, and edge cases — vague instructions produce inconsistent extractions.

Adding Validation Tools

Real pipelines need tools that validate or enrich extracted data. Let's add a tool that checks whether a vendor exists in an internal registry and another that normalizes currency codes.

import requests

VENDOR_REGISTRY = {
    "acme corp": {"id": "V001", "payment_terms": "net30"},
    "globex inc": {"id": "V002", "payment_terms": "net15"},
    "initech": {"id": "V003", "payment_terms": "net45"},
}

@function_tool
def lookup_vendor(vendor_name: str) -> dict:
    """Look up a vendor in the internal registry by name.
    
    Args:
        vendor_name: The vendor name to search for.
    
    Returns:
        A dictionary with vendor id and payment terms, or an empty dict.
    """
    key = vendor_name.lower().strip()
    return VENDOR_REGISTRY.get(key, {})

@function_tool
def validate_currency_code(code: str) -> bool:
    """Validate that a currency code is a supported ISO 4217 code.
    
    Args:
        code: Three-letter currency code to validate.
    
    Returns:
        True if the code is supported, False otherwise.
    """
    supported = {"USD", "EUR", "GBP", "JPY", "CAD", "AUD"}
    return code.upper() in supported

Attach these tools to a second agent that enriches the extracted data:

enrichment_agent = Agent(
    name="InvoiceEnricher",
    instructions=(
        "You receive extracted invoice data and enrich it. "
        "Use lookup_vendor to find the vendor's internal ID and payment terms. "
        "Use validate_currency_code to confirm the currency is supported. "
        "Return the enriched data along with any validation warnings."
    ),
    tools=[lookup_vendor, validate_currency_code],
    model="gpt-4o-mini",
)

Building the Pipeline with Handoffs

The Agents SDK supports handoffs, allowing one agent to delegate to another. Configure the extraction agent to hand off to the enrichment agent once extraction is complete.

enrichment_output_model = InvoiceData

enrichment_agent = Agent(
    name="InvoiceEnricher",
    instructions=(
        "You receive extracted invoice data and enrich it. "
        "Use lookup_vendor to find the vendor's internal ID and payment terms. "
        "Use validate_currency_code to confirm the currency is supported. "
        "Return the final InvoiceData object with any corrections applied."
    ),
    tools=[lookup_vendor, validate_currency_code],
    output_type=InvoiceData,
    model="gpt-4o-mini",
)

extraction_agent = Agent(
    name="InvoiceExtractor",
    instructions=(
        "You are a precise invoice data extraction specialist. "
        "Extract all fields into the InvoiceData schema from raw invoice text. "
        "Compute line item totals as quantity * unit_price. "
        "Verify subtotal equals the sum of line item totals. "
        "After extraction, hand off to the InvoiceEnricher for validation."
    ),
    output_type=InvoiceData,
    handoffs=[enrichment_agent],
    model="gpt-4o-mini",
)

Running the Pipeline

Execute the pipeline by passing raw invoice text to the Runner. The SDK handles the full loop: extraction, handoff, tool calls, and final structured output.

import asyncio
import json

async def extract_invoice(raw_text: str) -> InvoiceData:
    result = await Runner.run(extraction_agent, raw_text)
    return result.final_output

sample_invoice = """
    ACME CORP
    Invoice #INV-2024-0871
    Date: March 15, 2024
    Due: April 14, 2024

    Bill To: Initech LLC

    1. Consulting Services - 40 hours @ $150.00/hr   $6,000.00
    2. Software License - 5 seats @ $320.00/seat      $1,600.00
    3. On-site Training - 1 day @ $1,200.00/day       $1,200.00

    Subtotal: $8,800.00
    Tax (8.5%): $748.00
    Total Due: $9,548.00
    Currency: USD
    """

async def main():
    invoice = await extract_invoice(sample_invoice)
    print(json.dumps(invoice.model_dump(), indent=2))

if __name__ == "__main__":
    asyncio.run(main())

Expected output:

{
  "vendor_name": "ACME CORP",
  "invoice_number": "INV-2024-0871",
  "invoice_date": "2024-03-15",
  "due_date": "2024-04-14",
  "line_items": [
    {
      "description": "Consulting Services",
      "quantity": 40,
      "unit_price": 150.0,
      "total": 6000.0
    },
    {
      "description": "Software License",
      "quantity": 5,
      "unit_price": 320.0,
      "total": 1600.0
    },
    {
      "description": "On-site Training",
      "quantity": 1,
      "unit_price": 1200.0,
      "total": 1200.0
    }
  ],
  "subtotal": 8800.0,
  "tax_amount": 748.0,
  "total_amount": 9548.0,
  "currency": "USD"
}

Adding Output Guardrails

Guardrails run after an agent produces output and can reject invalid results. Define a guardrail that checks the math on extracted invoices.

from agents import GuardrailFunctionOutput, OutputGuardrailTripwireTriggered

async def invoice_math_guardrail(ctx, agent, output: InvoiceData) -> GuardrailFunctionOutput:
    computed_subtotal = sum(item.total for item in output.line_items)
    computed_total = computed_subtotal + output.tax_amount
    
    issues = []
    if abs(computed_subtotal - output.subtotal) > 0.01:
        issues.append(
            f"Subtotal mismatch: computed {computed_subtotal}, "
            f"extracted {output.subtotal}"
        )
    if abs(computed_total - output.total_amount) > 0.01:
        issues.append(
            f"Total mismatch: computed {computed_total}, "
            f"extracted {output.total_amount}"
        )
    
    if issues:
        raise OutputGuardrailTripwireTriggered(
            "Invoice math validation failed: " + "; ".join(issues)
        )
    
    return GuardrailFunctionOutput(output_info={"status": "valid"})

Attach the guardrail to the enrichment agent:

enrichment_agent = Agent(
    name="InvoiceEnricher",
    instructions="...",
    tools=[lookup_vendor, validate_currency_code],
    output_type=InvoiceData,
    output_guardrails=[invoice_math_guardrail],
    model="gpt-4o-mini",
)

Batch Processing Multiple Documents

Production pipelines process many documents. Use asyncio.gather with concurrency limits to handle batches efficiently.

import asyncio
from typing import List

async def process_batch(
    documents: List[str], 
    concurrency: int = 5
) -> List[dict]:
    semaphore = asyncio.Semaphore(concurrency)
    
    async def process_one(doc: str) -> dict:
        async with semaphore:
            try:
                result = await Runner.run(extraction_agent, doc)
                return {
                    "status": "success",
                    "data": result.final_output.model_dump(),
                }
            except OutputGuardrailTripwireTriggered as e:
                return {"status": "guardrail_failed", "error": str(e)}
            except Exception as e:
                return {"status": "error", "error": str(e)}
    
    return await asyncio.gather(*[process_one(doc) for doc in documents])

async def main():
    documents = [sample_invoice, another_invoice, malformed_invoice]
    results = await process_batch(documents, concurrency=3)
    for i, r in enumerate(results):
        print(f"Document {i}: {r['status']}")

Integrating with External Sources

Real pipelines pull documents from databases, object stores, or queues. Here's an example reading from a directory of text files and writing structured results to JSON.

import asyncio
import json
from pathlib import Path

async def process_directory(input_dir: str, output_path: str):
    input_path = Path(input_dir)
    files = list(input_path.glob("*.txt"))
    
    documents = []
    for f in files:
        text = f.read_text(encoding="utf-8")
        documents.append({"filename": f.name, "text": text})
    
    semaphore = asyncio.Semaphore(5)
    
    async def process_one(doc: dict) -> dict:
        async with semaphore:
            result = await Runner.run(extraction_agent, doc["text"])
            return {
                "filename": doc["filename"],
                "extracted": result.final_output.model_dump(),
            }
    
    results = await asyncio.gather(*[process_one(d) for d in documents])
    
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, default=str)
    
    print(f"Processed {len(results)} documents -> {output_path}")

if __name__ == "__main__":
    asyncio.run(process_directory("./invoices", "./output.json"))

Tracing and Observability

The Agents SDK includes built-in tracing. Enable it to inspect every step of your pipeline during development.

from agents import enable_verbose_stdout_logging

enable_verbose_stdout_logging()

# Or use the trace context manager for custom spans
from agents import trace

async def extract_with_tracing(text: str):
    with trace("InvoiceExtractionPipeline"):
        result = await Runner.run(extraction_agent, text)
        return result.final_output

Traces are also visible in the OpenAI dashboard under the Tracing section, where you can inspect token usage, latency, and tool call sequences for each run.

Best Practices

• Be explicit in instructions: Spell out calculations, defaults, and edge cases. The model follows what you write literally.
• Use descriptive field names and descriptions: Pydantic field descriptions become part of the schema sent to the model. Clear descriptions improve accuracy.
• Validate with guardrails: Never trust LLM output blindly. Guardrails catch math errors, missing fields, and schema violations before data reaches downstream systems.
• Keep agents focused: One agent per responsibility. Use handoffs to chain specialized agents rather than overloading a single agent with many tools.
• Limit concurrency: Use semaphores to avoid rate limits when processing batches. Start with 3–5 concurrent requests and tune based on your tier.
• Cache vendor lookups: External tool calls add latency. Cache registry lookups in memory or Redis to reduce repeated calls.
• Log raw inputs: Always persist the original document alongside the extracted output. You'll need it for debugging and reprocessing.
• Test with edge cases: Build a test suite of malformed, ambiguous, and multi-currency invoices. Run it against every change to your agents or schema.
• Choose the right model: Use gpt-4o-mini for high-volume, straightforward extraction. Switch to gpt-4o for complex documents with ambiguous formatting.
• Version your schemas: When your Pydantic models change, version them. Downstream consumers should pin to a schema version to avoid breaking changes.

Error Handling and Retries

Network failures and transient API errors are inevitable. Wrap your pipeline calls with retry logic:

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True,
)
async def extract_with_retry(text: str) -> InvoiceData:
    try:
        result = await Runner.run(extraction_agent, text)
        return result.final_output
    except OutputGuardrailTripwireTriggered:
        # Don't retry guardrail failures — they indicate data issues
        raise
    except Exception as e:
        print(f"Retrying due to error: {e}")
        raise

Install tenacity with pip install tenacity. Guardrail failures should not be retried because they indicate a problem with the data, not a transient error.

Conclusion

The OpenAI Agents SDK provides a clean, composable way to build robust data extraction pipelines. By combining typed Pydantic schemas, focused agents, validation tools, handoffs, and guardrails, you can transform messy unstructured documents into reliable structured data with minimal boilerplate. Start with a single extraction agent, add enrichment tools as your needs grow, and always guard your outputs with validation logic. With proper concurrency control, tracing, and retry handling, this pattern scales from prototype to production without architectural rewrites. The result is a pipeline that adapts to new document formats simply by updating instructions and schemas — a dramatic improvement over brittle rule-based systems.