Building a Data Extraction Pipeline with CrewAI: Complete Guide

Building a Data Extraction Pipeline with CrewAI: Complete Guide

Data extraction is one of the most valuable applications of large language models. Whether you're pulling structured information from invoices, resumes, research papers, or customer emails, the ability to convert unstructured text into clean, structured data can save organizations countless hours of manual work. CrewAI, an open-source framework for orchestrating role-playing AI agents, makes it straightforward to build robust, multi-step extraction pipelines that are reliable, modular, and easy to maintain.

In this tutorial, you'll learn what CrewAI is, why it's well-suited for data extraction, how to build a complete extraction pipeline from scratch, and the best practices that separate production-grade pipelines from fragile prototypes.

What Is CrewAI?

CrewAI is a Python framework that lets you coordinate multiple AI agents, each with a defined role, goal, and set of tools, to collaborate on complex tasks. Inspired by how human teams work, CrewAI models workflows as a "crew" of agents that hand off work to one another, share context, and produce a final output. It supports both sequential and hierarchical process flows, integrates with any LLM provider through LiteLLM, and ships with a growing library of built-in tools.

At its core, CrewAI gives you four building blocks:

• Agents — autonomous actors with a role, backstory, goal, and tools.
• Tasks — units of work assigned to agents, with expected outputs and optional output schemas.
• Crews — collections of agents and tasks that define a workflow.
• Tools — callable functions agents can use to interact with the outside world.

Why CrewAI for Data Extraction?

Single-prompt extraction approaches often fail on real-world documents. Documents can be long, ambiguous, multi-section, or require cross-referencing. CrewAI addresses these challenges by letting you decompose extraction into specialized sub-tasks handled by focused agents. For example, one agent can clean and chunk raw text, another can extract entities, and a third can validate and normalize the output against a schema.

This separation of concerns improves accuracy, makes debugging easier, and allows you to swap out individual agents or models without rewriting the whole pipeline. CrewAI also supports Pydantic-based structured outputs, which means your pipeline can return validated Python objects rather than free-form text.

Prerequisites and Installation

Before you start, make sure you have Python 3.10 or higher. Create a virtual environment and install CrewAI along with its dependencies:

python -m venv .venv
source .venv/bin/activate
pip install crewai crewai-tools pydantic

You'll also need an API key for your chosen LLM provider. For OpenAI, export the key as an environment variable:

export OPENAI_API_KEY="your-key-here"

CrewAI works with many providers including Anthropic, Groq, Ollama, and Azure OpenAI. For this tutorial, we'll use OpenAI's gpt-4o-mini for cost efficiency, but you can substitute any supported model.

Defining the Extraction Schema

The first step in any extraction pipeline is defining what you want to extract. Pydantic models are the ideal tool for this because they give you type validation, default values, and documentation in one place. Let's build a pipeline that extracts structured information from real estate property listings.

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


class PropertyType(str, Enum):
    HOUSE = "house"
    APARTMENT = "apartment"
    CONDO = "condo"
    TOWNHOUSE = "townhouse"
    LAND = "land"


class Amenity(BaseModel):
    name: str = Field(..., description="Name of the amenity, e.g. 'pool', 'garage'")
    present: bool = Field(..., description="Whether the amenity is present")


class PropertyListing(BaseModel):
    address: str = Field(..., description="Full property address")
    price_usd: Optional[float] = Field(None, description="Listing price in USD")
    property_type: PropertyType
    bedrooms: Optional[int] = Field(None, description="Number of bedrooms")
    bathrooms: Optional[float] = Field(None, description="Number of bathrooms")
    square_feet: Optional[int] = Field(None, description="Interior square footage")
    year_built: Optional[int] = Field(None, description="Year the property was built")
    description: str = Field(..., description="Short summary of the property")
    amenities: List[Amenity] = Field(default_factory=list)
    listing_agent: Optional[str] = Field(None, description="Name of the listing agent if mentioned")
    contact_phone: Optional[str] = Field(None, description="Contact phone number if mentioned")

Notice how each field includes a descriptive description. CrewAI passes these descriptions to the LLM as part of the structured output schema, which dramatically improves extraction accuracy. The more precise your descriptions, the better the model understands what to look for.

Designing the Agent Crew

For a robust extraction pipeline, we'll use three agents working sequentially:

• Document Preprocessor — cleans raw text, removes noise, and identifies the relevant sections.
• Entity Extractor — extracts structured fields from the cleaned text using the Pydantic schema.
• Validation Agent — reviews the extracted data for consistency, fills gaps, and flags low-confidence fields.

Let's define these agents:

from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

llm = LLM(model="openai/gpt-4o-mini", temperature=0.1)

preprocessor = Agent(
    role="Document Preprocessor",
    goal="Clean and normalize raw property listing text so that key information is easy to locate.",
    backstory=(
        "You are a meticulous data preparation specialist. You remove boilerplate, "
        "fix formatting issues, and highlight sections that contain pricing, "
        "specifications, and contact details."
    ),
    llm=llm,
    verbose=True,
)

extractor = Agent(
    role="Property Data Extractor",
    goal="Extract structured property information from cleaned listing text.",
    backstory=(
        "You are an expert real estate data analyst. You carefully read listing "
        "descriptions and pull out every relevant detail, never guessing when "
        "information is missing."
    ),
    llm=llm,
    verbose=True,
)

validator = Agent(
    role="Data Validation Specialist",
    goal="Review extracted property data for accuracy and completeness.",
    backstory=(
        "You are a quality assurance reviewer for real estate databases. You "
        "cross-check extracted fields against the source text, correct obvious "
        "errors, and flag fields that could not be confidently determined."
    ),
    llm=llm,
    verbose=True,
)

Setting temperature=0.1 keeps the model's output deterministic, which is critical for extraction tasks where consistency matters more than creativity.

Creating the Tasks

Each agent needs a task. Tasks define the expected output and, importantly, can specify a Pydantic model as output_pydantic to enforce structured output. Let's wire up the three tasks:

preprocess_task = Task(
    description=(
        "Take the following raw property listing text and produce a cleaned, "
        "well-organized version. Remove duplicate lines, fix obvious typos, "
        "and clearly separate sections such as description, specifications, "
        "and contact information.\n\n"
        "RAW TEXT:\n{raw_text}"
    ),
    expected_output="A cleaned and organized version of the listing text.",
    agent=preprocessor,
)

extraction_task = Task(
    description=(
        "Using the cleaned listing text from the previous step, extract all "
        "available property information into the PropertyListing schema. "
        "Only populate fields that are explicitly stated or can be directly "
        "inferred. Leave optional fields as null if the information is not "
        "present. Do not fabricate values."
    ),
    expected_output="A PropertyListing object populated with extracted data.",
    agent=extractor,
    output_pydantic=PropertyListing,
    context=[preprocess_task],
)

validation_task = Task(
    description=(
        "Review the extracted PropertyListing object. Verify that each "
        "populated field is supported by the source text. Correct any "
        "formatting issues (for example, normalize phone numbers and "
        "addresses). If a field appears incorrect, fix it. Return the "
        "final validated PropertyListing object."
    ),
    expected_output="A validated PropertyListing object ready for database insertion.",
    agent=validator,
    output_pydantic=PropertyListing,
    context=[preprocess_task, extraction_task],
)

The context parameter is what enables agents to see each other's work. The extraction task receives the preprocessed text, and the validation task receives both the preprocessed text and the extracted object, so it can cross-reference.

Assembling and Running the Crew

Now we bring everything together into a Crew and execute it:

crew = Crew(
    agents=[preprocessor, extractor, validator],
    tasks=[preprocess_task, extraction_task, validation_task],
    process=Process.sequential,
    verbose=True,
)

raw_listing = """
Stunning 3BR/2BA Home in Sunnyvale - $1,250,000

Beautifully renovated single family home at 742 Evergreen Terrace, Sunnyvale, CA 94086.
Built in 1962, this 1,850 sq ft house features hardwood floors, central AC, and a
two-car garage. The kitchen was fully updated in 2023 with stainless steel appliances.
Backyard includes a covered patio and mature landscaping. Community pool and clubhouse
available to residents.

Contact: Sarah Chen, Peninsula Realty
Phone: (408) 555-0142
Email: s.chen@peninsularealty.com

Don't miss this move-in ready gem!
"""

result = crew.kickoff(inputs={"raw_text": raw_listing})

# The final task's output is automatically parsed into the Pydantic model
property_data: PropertyListing = result.pydantic

print(property_data.model_dump_json(indent=2))

When you run this, CrewAI executes the three tasks in order. The final result.pydantic attribute gives you a fully validated PropertyListing instance. Here's an example of what the output might look like:

{
  "address": "742 Evergreen Terrace, Sunnyvale, CA 94086",
  "price_usd": 1250000.0,
  "property_type": "house",
  "bedrooms": 3,
  "bathrooms": 2.0,
  "square_feet": 1850,
  "year_built": 1962,
  "description": "Beautifully renovated single family home with hardwood floors, central AC, updated kitchen with stainless steel appliances, and a backyard with covered patio and mature landscaping.",
  "amenities": [
    {"name": "garage", "present": true},
    {"name": "central_ac", "present": true},
    {"name": "community_pool", "present": true},
    {"name": "clubhouse", "present": true}
  ],
  "listing_agent": "Sarah Chen",
  "contact_phone": "(408) 555-0142"
}

Adding Custom Tools for External Data

Real extraction pipelines often need to fetch documents from external sources. CrewAI lets you build custom tools that agents can call. Here's a tool that reads a file from disk:

from crewai.tools import tool
import pathlib


@tool("Read File")
def read_file(file_path: str) -> str:
    """Read the contents of a text file from the local filesystem.
    Args:
        file_path: Absolute or relative path to the file.
    Returns:
        The file contents as a string.
    """
    return pathlib.Path(file_path).read_text(encoding="utf-8")

You can attach this tool to the preprocessor agent so it can load documents directly:

preprocessor = Agent(
    role="Document Preprocessor",
    goal="Clean and normalize raw property listing text.",
    backstory="You are a meticulous data preparation specialist.",
    llm=llm,
    tools=[read_file],
    verbose=True,
)

For web-based extraction, you could add a tool that fetches URLs using requests or BeautifulSoup, or use CrewAI's built-in ScrapeWebsiteTool from crewai-tools.

Batch Processing Multiple Documents

In production, you'll rarely extract from a single document. Here's a pattern for processing a directory of listing files:

import json
from pathlib import Path

input_dir = Path("./listings")
output_dir = Path("./extracted")
output_dir.mkdir(exist_ok=True)

for file_path in input_dir.glob("*.txt"):
    raw_text = file_path.read_text(encoding="utf-8")
    result = crew.kickoff(inputs={"raw_text": raw_text})
    property_data = result.pydantic

    output_file = output_dir / f"{file_path.stem}.json"
    output_file.write_text(
        property_data.model_dump_json(indent=2),
        encoding="utf-8"
    )
    print(f"Extracted: {file_path.name} -> {output_file.name}")

For higher throughput, consider running crews concurrently with concurrent.futures.ThreadPoolExecutor, keeping in mind your LLM provider's rate limits.

Best Practices

Building a reliable extraction pipeline requires more than just wiring up agents. Here are the practices that make the biggest difference:

• Use low temperature settings. Extraction is a precision task. Keep temperature at 0.0 to 0.2 to minimize hallucination and improve reproducibility.
• Write detailed field descriptions. The LLM relies on schema descriptions to understand what to extract. Include units, formats, and examples where helpful.
• Separate extraction from validation. Having a dedicated validation agent that cross-references the source text catches errors that a single-pass extractor would miss.
• Never let agents guess. Explicitly instruct agents to leave fields null when information is absent. Fabricated data is worse than missing data.
• Use Pydantic enums for categorical fields. Enums constrain the model to valid values and prevent free-form variations like "Single Family" vs "house".
• Log intermediate outputs. Set verbose=True and save each task's output during development. This makes it easy to identify which agent introduced an error.
• Test with edge cases. Include documents with missing fields, unusual formats, and ambiguous language in your test suite. Extraction quality is only as good as your worst-case document.
• Cache LLM calls during development. Use LiteLLM's caching or a local cache to avoid repeated API costs while iterating on prompts.
• Choose the right model for each agent. Preprocessing can use a cheaper, faster model, while extraction and validation benefit from a more capable model. CrewAI lets you assign different LLMs per agent.

Handling Errors and Retries

LLM outputs can occasionally fail Pydantic validation, especially with complex schemas. Wrap your crew execution in error handling and implement retries:

import time
from pydantic import ValidationError

def extract_with_retry(crew, inputs, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = crew.kickoff(inputs=inputs)
            return result.pydantic
        except (ValidationError, Exception) as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
    return None

For production systems, consider persisting failed inputs to a review queue so a human can inspect documents that the pipeline could not process.

Conclusion

CrewAI provides a powerful, structured approach to building data extraction pipelines that go beyond what single-prompt solutions can achieve. By decomposing extraction into preprocessing, extraction, and validation stages — each handled by a specialized agent — you get a pipeline that is more accurate, easier to debug, and simpler to extend. Combined with Pydantic schemas for type-safe outputs and custom tools for integrating external data sources, CrewAI gives you everything you need to turn messy unstructured documents into clean, structured records ready for your database or downstream applications. Start with the three-agent pattern shown here, iterate on your schema descriptions, test against real-world edge cases, and you'll have a production-ready extraction pipeline that scales with your data needs.