Information Extraction Pipelines: Turning Unstructured Text into Agent-Readable Data

Why Agents Need Information Extraction

AI agents operate on structured data — function parameters, database queries, API payloads. But users communicate in unstructured natural language: emails, chat messages, documents, and voice transcripts. Information extraction bridges this gap by converting free-form text into structured records that agents can act upon.

Consider an email: "Please book a conference room for 10 people next Wednesday from 2pm to 4pm. We need a projector and video conferencing setup." An agent needs to extract: capacity (10), date (next Wednesday), time range (2pm-4pm), and equipment requirements (projector, video conferencing). This is information extraction.

Pattern-Based Extraction with Regular Expressions

For well-defined formats, regex extraction is fast, predictable, and requires no model inference.

import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class ContactInfo:
    email: Optional[str] = None
    phone: Optional[str] = None
    name: Optional[str] = None

def extract_contact_info(text: str) -> ContactInfo:
    """Extract contact information using regex patterns."""
    email_pattern = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
    phone_pattern = r"(?:\+1[\s-]?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}"
    name_pattern = r"(?:(?:Mr|Mrs|Ms|Dr)\.?\s+)?([A-Z][a-z]+(?:\s+[A-Z][a-z]+)+)"

    email_match = re.search(email_pattern, text)
    phone_match = re.search(phone_pattern, text)
    name_match = re.search(name_pattern, text)

    return ContactInfo(
        email=email_match.group() if email_match else None,
        phone=phone_match.group() if phone_match else None,
        name=name_match.group(1) if name_match else None,
    )

text = "Contact Dr. Sarah Johnson at sarah.j@hospital.org or (555) 123-4567"
info = extract_contact_info(text)
# ContactInfo(email='sarah.j@hospital.org', phone='(555) 123-4567',
#             name='Sarah Johnson')

Template Filling with LLMs

For complex, variable-format text, LLMs excel at extracting information into predefined templates.

import openai
import json
from pydantic import BaseModel, Field
from typing import Optional

class MeetingRequest(BaseModel):
    date: Optional[str] = Field(None, description="Meeting date")
    start_time: Optional[str] = Field(None, description="Start time")
    end_time: Optional[str] = Field(None, description="End time")
    attendee_count: Optional[int] = Field(None, description="Number of attendees")
    equipment: list[str] = Field(default_factory=list)
    location_preference: Optional[str] = None

def extract_meeting_details(text: str) -> MeetingRequest:
    """Extract structured meeting details from free-form text."""
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """Extract meeting request details from the text.
Return a JSON object with these fields:
date, start_time, end_time, attendee_count, equipment (list), location_preference.
Use null for missing fields.""",
            },
            {"role": "user", "content": text},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )

    data = json.loads(response.choices[0].message.content)
    return MeetingRequest(**data)

text = """Book a room for 10 people next Wednesday 2pm to 4pm.
Need a projector and video conferencing. Prefer building A if available."""

meeting = extract_meeting_details(text)
# MeetingRequest(date='next Wednesday', start_time='2pm',
#   end_time='4pm', attendee_count=10,
#   equipment=['projector', 'video conferencing'],
#   location_preference='building A')

Relation Extraction

Relation extraction identifies how entities in text are connected — "works at," "located in," "reports to." This is essential for agents building knowledge graphs or understanding organizational structures.

import openai
import json

def extract_relations(
    text: str,
    relation_types: list[str],
) -> list[dict]:
    """Extract entity relations from text."""
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"""Extract relations from the text.
Only extract these relation types: {', '.join(relation_types)}

Return a JSON array where each item has:
- subject: the source entity
- relation: the relation type
- object: the target entity
- confidence: your confidence from 0.0 to 1.0""",
            },
            {"role": "user", "content": text},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )

    data = json.loads(response.choices[0].message.content)
    return data.get("relations", [])

text = """Dr. Amara Osei works at Nairobi General Hospital in the
cardiology department. She reports to Dr. James Mwangi, the Chief
of Medicine. The hospital is located in Nairobi, Kenya."""

relations = extract_relations(text, [
    "works_at", "located_in", "reports_to", "department_of"
])
# [{'subject': 'Dr. Amara Osei', 'relation': 'works_at',
#   'object': 'Nairobi General Hospital', 'confidence': 0.95},
#  {'subject': 'Dr. Amara Osei', 'relation': 'reports_to',
#   'object': 'Dr. James Mwangi', 'confidence': 0.92}, ...]

Building a Complete Extraction Pipeline

Production extraction pipelines chain multiple stages — each one refining and validating the output of the previous stage.

from dataclasses import dataclass, field
from typing import Any
from pydantic import BaseModel, ValidationError

@dataclass
class ExtractionResult:
    raw_text: str
    extracted_data: dict
    validation_errors: list[str] = field(default_factory=list)
    confidence: float = 0.0

class ExtractionPipeline:
    def __init__(self):
        self.stages: list[callable] = []

    def add_stage(self, stage_fn):
        self.stages.append(stage_fn)
        return self

    def run(self, text: str) -> ExtractionResult:
        result = ExtractionResult(raw_text=text, extracted_data={})

        for stage in self.stages:
            try:
                stage_output = stage(text, result.extracted_data)
                result.extracted_data.update(stage_output)
            except Exception as e:
                result.validation_errors.append(
                    f"Stage {stage.__name__} failed: {str(e)}"
                )

        result.confidence = self._compute_confidence(result)
        return result

    def _compute_confidence(self, result: ExtractionResult) -> float:
        if result.validation_errors:
            return 0.0
        filled = sum(
            1 for v in result.extracted_data.values()
            if v is not None
        )
        total = max(len(result.extracted_data), 1)
        return round(filled / total, 2)

def validate_extracted_data(
    data: dict,
    schema: type[BaseModel],
) -> tuple[Any, list[str]]:
    """Validate extracted data against a Pydantic schema."""
    try:
        validated = schema(**data)
        return validated, []
    except ValidationError as e:
        errors = [
            f"{err['loc']}: {err['msg']}"
            for err in e.errors()
        ]
        return None, errors

Handling Extraction Failures Gracefully

Extraction from unstructured text is inherently unreliable. Agents must handle partial extractions and ask users for clarification on missing fields.

def identify_missing_fields(
    extracted: dict,
    required_fields: list[str],
) -> list[str]:
    """Identify which required fields are missing or empty."""
    missing = []
    for field_name in required_fields:
        value = extracted.get(field_name)
        if value is None or value == "" or value == []:
            missing.append(field_name)
    return missing

def generate_clarification(missing_fields: list[str]) -> str:
    """Generate a user-friendly clarification request."""
    field_labels = {
        "date": "the date",
        "start_time": "the start time",
        "attendee_count": "how many people will attend",
        "location_preference": "your preferred location",
    }
    items = [field_labels.get(f, f) for f in missing_fields]
    if len(items) == 1:
        return f"Could you also let me know {items[0]}?"
    return f"Could you also provide {', '.join(items[:-1])} and {items[-1]}?"

This pattern ensures the agent never guesses at missing information. Instead, it extracts what it can, validates the result, and asks targeted follow-up questions for anything that is missing or ambiguous.

FAQ

How do I choose between regex-based and LLM-based extraction?

Use regex for structured, predictable formats — email addresses, phone numbers, dates in known formats, product codes. Use LLM-based extraction for variable, natural language content where the same information can be expressed in dozens of different ways. Many production systems combine both: regex for fast extraction of well-defined fields, LLM for everything else.

How do I handle extraction from very long documents?

Split the document into semantically meaningful chunks (by section, paragraph, or topic) rather than arbitrary character limits. Run extraction on each chunk independently, then merge and deduplicate the results. For documents with structured sections (like contracts or reports), use the section headers to target extraction to the most relevant parts.

What is the best way to validate LLM-extracted data?

Layer three validation strategies: (1) Schema validation with Pydantic to ensure correct types and required fields. (2) Business rule validation — check that dates are in the future, quantities are positive, email addresses are properly formatted. (3) Cross-field consistency — if a meeting is 2pm to 4pm, verify that end time is after start time. Reject extractions that fail validation and either retry with a more specific prompt or ask the user for clarification.

---

#InformationExtraction #NLP #StructuredData #RelationExtraction #AIAgents #Python #AgenticAI #LearnAI #AIEngineering