Building a Data Extraction Pipeline with AutoGen: Complete Guide
Modern applications increasingly rely on extracting structured information from messy, unstructured sources — PDFs, emails, web pages, and free-form text. While large language models are excellent at understanding such content, orchestrating them into a reliable, multi-step pipeline is non-trivial. Microsoft's AutoGen provides a multi-agent conversation framework that lets you compose specialized agents, each handling a distinct stage of extraction, validation, and transformation. This tutorial walks you through building a production-grade data extraction pipeline using AutoGen, from basic concepts to advanced patterns.
What Is AutoGen?
AutoGen is an open-source framework from Microsoft Research for building multi-agent conversational systems. Instead of treating an LLM as a single function call, AutoGen lets you define multiple agents — each with a role, system prompt, and toolset — that converse with one another to solve complex tasks. Agents can be backed by different models, configured with different temperatures, and given different tools, allowing you to decompose a problem into specialized subtasks.
In a data extraction context, this means you can have one agent focused purely on reading raw documents, another on parsing fields into a schema, a third on validating the output against business rules, and a fourth on formatting the final result. Each agent does one thing well, and the conversation between them enforces a structured workflow.
Why Use AutoGen for Data Extraction?
Single-prompt extraction approaches often fail in real-world scenarios because they conflate several responsibilities: reading, interpreting, validating, and formatting. When something goes wrong, it's hard to know which stage failed. AutoGen addresses this by separating concerns into distinct agents.
- Separation of concerns: Each agent owns one responsibility, making the pipeline easier to debug and extend.
- Built-in retry and reflection: Validation agents can reject outputs and ask extraction agents to try again, creating a self-correcting loop.
- Tool integration: Agents can call Python functions, APIs, or databases to enrich or verify extracted data.
- Model flexibility: You can route cheap tasks to smaller models and complex reasoning to larger ones, optimizing cost.
- Observability: The conversation log provides a full audit trail of how data was extracted and transformed.
Prerequisites and Installation
Before building the pipeline, install AutoGen and a few supporting libraries. This tutorial uses AutoGen's 0.2 API with the ConversableAgent and GroupChat classes.
pip install "pyautogen>=0.2.25" python-dotenv pydantic
You'll also need an OpenAI API key (or any compatible endpoint). Store it in a .env file:
OPENAI_API_KEY=sk-your-key-here
Defining the Extraction Schema
Start by defining the target schema using Pydantic. A clear schema is the backbone of any extraction pipeline — it tells agents exactly what fields to produce and what types to expect. For this tutorial, imagine extracting invoice data from free-form text.
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import date
class LineItem(BaseModel):
description: str = Field(..., description="Description of the product or service")
quantity: float = Field(..., description="Number of units")
unit_price: float = Field(..., description="Price per unit in USD")
total: float = Field(..., description="Line total in USD")
class InvoiceData(BaseModel):
vendor_name: str = Field(..., description="Name of the vendor or supplier")
invoice_number: str = Field(..., description="Unique invoice identifier")
invoice_date: Optional[date] = Field(None, description="Date the invoice was issued")
due_date: Optional[date] = Field(None, description="Payment due date")
line_items: List[LineItem] = Field(default_factory=list)
subtotal: float = Field(..., description="Sum of line item totals")
tax: float = Field(0.0, description="Tax amount in USD")
total: float = Field(..., description="Final amount due in USD")
Pydantic models serve double duty: they document the schema for the LLM and validate the output programmatically. We'll use both capabilities in the pipeline.
Configuring the LLM Backend
AutoGen uses a configuration dictionary to specify which model to call. You can define multiple configurations and assign them to different agents, which is useful when you want a cheaper model for simple tasks and a stronger one for reasoning.
import os
from dotenv import load_dotenv
load_dotenv()
llm_config_strong = {
"config_list": [
{
"model": "gpt-4o",
"api_key": os.getenv("OPENAI_API_KEY"),
}
],
"temperature": 0.2,
}
llm_config_fast = {
"config_list": [
{
"model": "gpt-4o-mini",
"api_key": os.getenv("OPENAI_API_KEY"),
}
],
"temperature": 0.0,
}
Low temperatures are important for extraction tasks because you want deterministic, reproducible outputs rather than creative variation.
Creating the Agents
Our pipeline uses four agents, each with a focused role. The Extractor reads raw text and produces a JSON draft. The Validator checks the draft against the schema and business rules. The Enricher fills in missing or ambiguous fields using tools. The Formatter produces the final clean output.
from autogen import ConversableAgent
extractor = ConversableAgent(
name="Extractor",
system_message=(
"You are a data extraction specialist. Given raw invoice text, "
"extract all fields and return ONLY valid JSON matching this schema: "
"vendor_name, invoice_number, invoice_date (YYYY-MM-DD), due_date, "
"line_items (list of {description, quantity, unit_price, total}), "
"subtotal, tax, total. Do not include explanations."
),
llm_config=llm_config_strong,
human_input_mode="NEVER",
)
validator = ConversableAgent(
name="Validator",
system_message=(
"You are a data validation specialist. You receive JSON extracted "
"from an invoice. Check that: (1) all required fields are present, "
"(2) numeric fields are numbers, (3) subtotal equals the sum of "
"line item totals, (4) total equals subtotal plus tax. "
"If valid, respond with 'VALID' followed by the JSON. "
"If invalid, respond with 'INVALID' and a list of specific errors."
),
llm_config=llm_config_strong,
human_input_mode="NEVER",
)
enricher = ConversableAgent(
name="Enricher",
system_message=(
"You fill in missing or ambiguous invoice fields. If a date is "
"missing, infer a reasonable default and note the assumption. "
"If a vendor name is abbreviated, expand it. Return updated JSON."
),
llm_config=llm_config_fast,
human_input_mode="NEVER",
)
formatter = ConversableAgent(
name="Formatter",
system_message=(
"You produce the final clean JSON output. Remove any commentary, "
"ensure consistent key ordering, and return only the JSON object."
),
llm_config=llm_config_fast,
human_input_mode="NEVER",
)
Notice that the Extractor and Validator use the stronger model because they require careful reading and arithmetic reasoning, while the Enricher and Formatter use the faster, cheaper model since their tasks are simpler transformations.
Adding a Programmatic Validation Tool
LLM-based validation is useful but not bulletproof. For arithmetic checks, a Python function is far more reliable. AutoGen lets you register functions as tools that agents can call. Let's add a strict validation function the Validator can invoke.
import json
def validate_invoice_json(raw_json: str) -> dict:
"""Validate invoice JSON against the Pydantic schema and arithmetic rules."""
try:
data = json.loads(raw_json)
except json.JSONDecodeError as e:
return {"valid": False, "errors": [f"Invalid JSON: {e}"]}
# Schema validation via Pydantic
try:
invoice = InvoiceData(**data)
except Exception as e:
return {"valid": False, "errors": [str(e)]}
# Arithmetic checks
errors = []
line_sum = round(sum(li.total for li in invoice.line_items), 2)
if abs(line_sum - invoice.subtotal) > 0.01:
errors.append(
f"Subtotal mismatch: line items sum to {line_sum} but subtotal is {invoice.subtotal}"
)
expected_total = round(invoice.subtotal + invoice.tax, 2)
if abs(expected_total - invoice.total) > 0.01:
errors.append(
f"Total mismatch: subtotal + tax = {expected_total} but total is {invoice.total}"
)
if errors:
return {"valid": False, "errors": errors}
return {"valid": True, "data": invoice.model_dump(mode="json")}
Register this function with the Validator agent so it can call the tool during conversation:
@validator.register_for_execution()
@validator.register_for_llm(
description="Validate invoice JSON against schema and arithmetic rules. "
"Input is the JSON string. Returns validity and errors."
)
def validate_invoice(raw_json: str) -> dict:
return validate_invoice_json(raw_json)
Orchestrating with GroupChat
With agents defined, orchestrate them using a GroupChat managed by a GroupChatManager. The manager decides which agent speaks next based on the conversation state. You can provide an explicit speaker selection method or let AutoGen choose automatically.
from autogen import GroupChat, GroupChatManager
groupchat = GroupChat(
agents=[extractor, validator, enricher, formatter],
messages=[],
max_round=12,
speaker_selection_method="auto",
)
manager = GroupChatManager(
groupchat=groupchat,
llm_config=llm_config_fast,
)
The max_round parameter caps the conversation length, preventing infinite loops if agents get stuck disagreeing. Twelve rounds is usually enough for extract-validate-fix-format cycles.
Running the Pipeline
Now let's feed a sample invoice through the pipeline. Initiate the chat by sending the raw text to the manager, which will route it to the Extractor first.
raw_invoice_text = """
INVOICE
From: Acme Supplies Co.
Invoice #: INV-2024-0887
Date: March 14, 2024
Due: April 13, 2024
Items:
- Widget A, 10 units @ $12.50 each = $125.00
- Widget B, 4 units @ $30.00 each = $120.00
- Shipping service = $25.00
Subtotal: $270.00
Tax (8%): $21.60
Total Due: $291.60
"""
result = extractor.initiate_chat(
manager,
message=f"Please extract structured data from this invoice:\n\n{raw_invoice_text}",
summary_method="last_msg",
)
print(result.summary)
The conversation proceeds roughly as follows: the Extractor produces a JSON draft, the Validator calls the validation tool, if invalid the Enricher or Extractor corrects issues, and finally the Formatter emits the clean JSON. The summary_method="last_msg" option captures the final message as the pipeline output.
Parsing and Persisting the Result
After the chat completes, parse the final JSON into your Pydantic model and persist it. This step bridges the agent pipeline and your application's storage layer.
def extract_final_json(chat_result) -> InvoiceData:
"""Pull the JSON object out of the final agent message."""
text = chat_result.summary
# Find the first '{' and last '}' to isolate JSON
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1:
raise ValueError("No JSON object found in output")
payload = text[start : end + 1]
return InvoiceData(**json.loads(payload))
invoice = extract_final_json(result)
print(invoice.model_dump_json(indent=2))
You can then write invoice to a database, emit it to a message queue, or return it from an API endpoint. Because it's a validated Pydantic object, downstream code can rely on the types.
Handling Batch Extraction
Real pipelines process many documents. Wrap the single-document flow in a function and iterate, collecting successes and failures separately so one bad document doesn't halt the batch.
from typing import Tuple, List
def process_document(text: str) -> Tuple[InvoiceData, None] | Tuple[None, str]:
try:
chat = extractor.initiate_chat(
manager,
message=f"Extract structured data from this invoice:\n\n{text}",
summary_method="last_msg",
)
invoice = extract_final_json(chat)
return invoice, None
except Exception as e:
return None, str(e)
def process_batch(documents: List[str]) -> dict:
results = {"success": [], "failed": []}
for doc in documents:
invoice, error = process_document(doc)
if invoice:
results["success"].append(invoice)
else:
results["failed"].append({"document": doc[:200], "error": error})
return results
For higher throughput, consider running documents concurrently with asyncio or a thread pool, keeping in mind API rate limits.
Best Practices
Building a robust extraction pipeline requires more than wiring agents together. The following practices will save you significant debugging time.
- Keep system prompts narrow. Each agent should have one clear job. Overloaded prompts cause agents to step on each other in group chats.
- Use programmatic validation for arithmetic. LLMs are unreliable at summing numbers. Delegate math to Python tools and let the LLM handle language understanding.
- Set low temperatures. Extraction is a deterministic task. Temperatures of 0.0 to 0.2 reduce variance and improve reproducibility.
- Cap conversation rounds. Always set
max_roundto prevent runaway loops where agents endlessly correct each other. - Log full conversations. Save the entire message history for each run. When extraction fails, the transcript is your best debugging tool.
- Separate models by complexity. Route simple formatting to cheap models and complex reasoning to capable ones. This can cut costs dramatically on large batches.
- Test with edge cases. Include invoices with missing dates, unusual currencies, handwritten-style text, and multi-page content in your test suite.
- Cache deterministic steps. If the same document is processed twice, cache the Extractor's output to avoid redundant API calls.
Adding Observability
To monitor pipeline health in production, log each agent transition and tool call. AutoGen exposes hooks via the register_reply mechanism, or you can post-process the group chat's message list.
def log_conversation(groupchat: GroupChat) -> None:
for msg in groupchat.messages:
sender = msg.get("name", "unknown")
content = msg.get("content", "")[:300]
print(f"[{sender}] {content}...")
# After a run:
log_conversation(groupchat)
For production systems, replace the print statements with structured logging to a service like Datadog or CloudWatch, including timestamps, token counts, and latency per agent.
Conclusion
AutoGen's multi-agent approach turns the messy problem of data extraction into a clean, modular workflow where each agent owns a single responsibility. By combining an Extractor for reading, a Validator backed by Pydantic and Python tools for correctness, an Enricher for gap-filling, and a Formatter for clean output, you get a pipeline that is easier to debug, cheaper to run, and more accurate than a single monolithic prompt. Start with the four-agent pattern shown here, then adapt the roles and tools to your specific document types. With careful prompt design, programmatic validation, and observability, this architecture scales from prototypes to production systems handling thousands of documents reliably.