Module 3: Function Calling Patterns

8. Project: Extraction + Routing System

Project Overview

Throughout this module you learned the function calling patterns that separate a demo agent from a production one: parallel calls, forced tool calls, routing, structured extraction, tool composition, and retry patterns. Now you'll integrate them into a project that combines at least 3 of those patterns into a real system.

The project is an extraction + routing system: it takes free-text documents (emails, invoices, news articles), extracts entities (people, companies, dates, amounts) using with_structured_output with Pydantic schemas, classifies each entity type, and routes them to specialized processors that enrich each entity by type. The result is a structured report with every processed entity.

Why this project? Because extraction + routing is the most common pattern in production systems that process documents. A CRM that receives emails and extracts contacts. An accounting system that parses invoices and extracts amounts. A news aggregator that extracts mentioned people and companies. They all follow the same flow: document comes in → entities get extracted → each type gets processed differently → structured result comes out.

The patterns you'll use: structured extraction (capsule 04), routing (capsule 03), retry with backoff (capsule 07), and tool composition (capsule 06) to chain extraction → classification → processing. Three patterns minimum, four if you do it right.

Estimated time: 60-90 minutes.

Project Goal

Build a complete extraction + routing system that takes free-text documents, extracts typed entities with Pydantic schemas, routes each entity to a specialized processor, and produces a structured report — with automatic retry when extraction fails.

By the end you'll be able to:

Design Pydantic schemas for 4 entity types with validation
Implement an extractor with with_structured_output that handles varied documents
Build a router that directs each entity type to a specialized processor
Add retry with exponential backoff for resilient extraction
Combine 3+ function calling patterns into an integrated system

Technical Specifications

Stack

Technology	Version	Use
Python	3.11+	Runtime
langchain	v1.2+	`init_chat_model`, `with_structured_output`
langchain-openai	latest	OpenAI provider
tenacity	latest	Retry with exponential backoff
pydantic	v2	Entity schemas and validation
python-dotenv	any	Environment variables

Setup

pip install langchain langchain-openai tenacity python-dotenv

Create a .env file:

OPENAI_API_KEY=sk-proj-your-api-key-here

Project files

extraction_routing/
├── .env                        # OpenAI API key
├── extraction_system.py        # The complete system
└── sample_documents/
    ├── commercial_email.txt    # Test email
    ├── service_invoice.txt     # Test invoice
    └── news_article.txt        # Test article

Create the test documents. The complete files are available in the Recommended Tests section.

sample_documents/commercial_email.txt — An email from Maria Garcia (CTO, TechCorp) proposing a server migration to CloudServ Solutions. Mentions 4 people, 3 companies, 2 dates, 2 amounts ($45K USD budget, $15K deposit).

sample_documents/service_invoice.txt — Invoice #INV-2026-0847 from CloudServ to TechCorp. 4 service lines, 16% tax, total $4,408 USD. 2 companies, 1 contact, 2 dates (issue/due).

sample_documents/news_article.txt — An article about TechCorp's $12M USD Series A round. Multiple companies (Sequoia, Andreessen, ALLVP, DataAI, NeuralSoft), 3+ people, 6+ investment amounts.

The documents need enough variety to exercise extraction of all 4 entity types. Create your own or use these as a reference:

# commercial_email.txt (abridged example)
From: Maria Garcia <maria.garcia@techcorp.com>
Subject: Q2 2026 migration proposal

I'm Maria Garcia, CTO of TechCorp. I spoke with Juan Ramirez on March 3.
We need to migrate servers before April 15, 2026.
Budget: $45,000 USD, with a $15,000 USD deposit.
Roberto Sanchez (VP of Infrastructure) will be the technical contact.
Recommended by Ana Lopez of DataAI.

System Architecture

┌──────────┐    ┌──────────────┐    ┌────────────┐    ┌──────────────┐    ┌───────────┐
│ Document │───▶│   Extract    │───▶│  Classify  │───▶│    Route     │───▶│ Aggregate │
│  Intake  │    │  Entities    │    │   Types    │    │  & Process   │    │  Results  │
└──────────┘    └──────────────┘    └────────────┘    └──────────────┘    └───────────┘
                 with_structured     by entity          each entity         structured
                 _output + retry     type               → processor         report

Pattern	Where it's used	Capsule
Structured Extraction	Extract Entities: `with_structured_output` with Pydantic schemas	04
Tool Routing	Route & Process: each entity type goes to a different processor	03
Retry + Backoff	Extract Entities: retry with backoff if extraction fails	07
Composition	The whole pipeline: extraction → routing → processing → aggregation	06

Step 1: Define the Extraction Schemas

The Pydantic schemas define the entities the system extracts. Each schema has typed fields with descriptions that guide the model and validators that guarantee quality.

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


class Person(BaseModel):
    name: str = Field(
        description="The person's full name."
    )
    role: Optional[str] = Field(
        default=None,
        description="Professional title or role. ONLY if explicitly mentioned."
    )
    company: Optional[str] = Field(
        default=None,
        description="Company they belong to. ONLY if mentioned."
    )
    email: Optional[str] = Field(
        default=None,
        description="Email if mentioned. NEVER invent an email."
    )


class Company(BaseModel):
    name: str = Field(description="The company's full name.")
    industry: Optional[str] = Field(
        default=None,
        description="Industry or sector if it can be inferred from the context."
    )
    country: Optional[str] = Field(
        default=None,
        description="Country if mentioned or inferred."
    )

    @field_validator("name")
    @classmethod
    def name_not_empty(cls, v: str) -> str:
        if len(v.strip()) < 2:
            raise ValueError("Company name must have 2+ characters")
        return v.strip()


class DateEntity(BaseModel):
    original_text: str = Field(
        description="The original date text exactly as it appears in the document."
    )
    normalized: Optional[str] = Field(
        default=None,
        description="Date in YYYY-MM-DD format if possible. Null if it's vague."
    )
    context: str = Field(
        description="What this date is for: deadline, event, founding, etc."
    )

    @field_validator("normalized")
    @classmethod
    def validate_date_format(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        datetime.strptime(v, "%Y-%m-%d")
        return v


class Amount(BaseModel):
    value: float = Field(
        description="Expanded numeric value. '12 million' = 12000000.0",
        ge=0
    )
    currency: str = Field(
        description="ISO 4217 code: USD, MXN, EUR."
    )
    context: str = Field(
        description="What it's for: budget, investment, price, etc."
    )

    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        valid = {"USD", "MXN", "EUR", "GBP", "BRL", "COP", "ARS", "CLP", "PEN"}
        v = v.upper().strip()
        if v not in valid:
            raise ValueError(f"Currency '{v}' not recognized")
        return v

Every Optional field has "ONLY if mentioned" or "NEVER invent" in its description. Without this, the model tends to fill empty fields with plausible but invented data — especially emails.

The validator on Company.name prevents empty strings. The one on DateEntity.normalized guarantees YYYY-MM-DD format. The one on Amount.currency normalizes to uppercase and validates against known currencies. The validators act as a second line of defense: if the model returns badly formatted data, Pydantic rejects it and the retry corrects it.

Wrapper schema: DocumentEntities

class DocumentEntities(BaseModel):
    people: List[Person] = Field(
        default_factory=list,
        description="EVERY person mentioned. Don't skip any."
    )
    companies: List[Company] = Field(
        default_factory=list,
        description="EVERY company and organization."
    )
    dates: List[DateEntity] = Field(
        default_factory=list,
        description="EVERY date and time reference."
    )
    amounts: List[Amount] = Field(
        default_factory=list,
        description="EVERY monetary amount."
    )
    document_type: str = Field(
        description="Type: 'email', 'invoice', 'news_article', 'contract', 'other'."
    )
    summary: str = Field(
        description="One-sentence summary of the document."
    )

document_type and summary get extracted along with the entities — the model is already reading the whole document, so classifying at the same time costs no extra tokens.

Step 2: Implement the Extractor

The extractor uses with_structured_output with include_raw=True to capture parsing errors, and includes retry with exponential backoff.

from langchain.chat_models import init_chat_model
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log,
)
import logging
from dotenv import load_dotenv

load_dotenv()
logger = logging.getLogger(__name__)

EXTRACTION_PROMPT = """Analyze the following document and extract EVERY entity.

Rules:
- Extract EVERY person, company, date and amount. Don't skip any.
- For Optional fields, use null if the information isn't in the text.
- NEVER invent data that isn't in the document.
- Normalize dates to YYYY-MM-DD whenever possible.
- For amounts, convert text like "12 million" to 12000000.0.

Document:
{document}"""

model = init_chat_model("openai:gpt-4.1-mini")
extractor = model.with_structured_output(DocumentEntities, include_raw=True)


class ExtractionError(Exception):
    pass


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((ExtractionError, ValueError)),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def extract_with_retry(document: str) -> DocumentEntities:
    """Extract entities with automatic retry and exponential backoff."""
    prompt = EXTRACTION_PROMPT.format(document=document)
    result = extractor.invoke(prompt)

    if result["parsing_error"] is not None:
        raise ExtractionError(
            f"Validation failed: {str(result['parsing_error'])[:200]}"
        )

    parsed = result["parsed"]
    if parsed is None:
        raise ExtractionError("Extraction returned None")

    total = (
        len(parsed.people) + len(parsed.companies)
        + len(parsed.dates) + len(parsed.amounts)
    )
    if total == 0:
        raise ExtractionError("0 entities extracted")

    return parsed

include_raw=True returns a dict with parsed (a Pydantic object or None), raw (the raw response), and parsing_error (the error, if any). That gives you three levels of defense: (1) if Pydantic rejects the response → retry, (2) if parsed is None → retry, (3) if there are 0 entities → retry. A real document almost always has at least one entity; if it returns 0, the model probably failed and a retry with a different seed usually fixes it.

wait_exponential(multiplier=1, min=2, max=10) produces: first retry at ~2s, second at ~4s, third at ~8s. Three attempts with backoff cover 99% of transient failures.

Step 3: Implement the Router

The router examines the extracted entities and directs them to specialized processors based on their type — static, deterministic routing, with no extra LLM cost.

from typing import Callable


class EntityRouter:
    def __init__(self):
        self.processors: dict[str, Callable] = {}

    def register(self, entity_type: str, processor: Callable):
        self.processors[entity_type] = processor

    def route(self, entities: DocumentEntities) -> dict:
        results = {
            "document_type": entities.document_type,
            "summary": entities.summary,
            "processed_entities": {},
            "stats": {},
        }

        entity_map = {
            "people": entities.people,
            "companies": entities.companies,
            "dates": entities.dates,
            "amounts": entities.amounts,
        }

        for entity_type, entity_list in entity_map.items():
            if not entity_list:
                results["stats"][entity_type] = 0
                continue

            processor = self.processors.get(entity_type)
            if processor is None:
                results["processed_entities"][entity_type] = [
                    {"raw": e.model_dump(), "status": "no_processor"}
                    for e in entity_list
                ]
                results["stats"][entity_type] = len(entity_list)
                continue

            processed = []
            for entity in entity_list:
                result = processor(entity)
                status = "error" if "error" in result else "ok"
                processed.append({"data": result, "status": status})

            results["processed_entities"][entity_type] = processed
            results["stats"][entity_type] = len(entity_list)

        return results

If a processor fails for a specific entity, it marks that entity as "error" and moves on to the rest. One failed entity doesn't block processing of the others. If a type has no registered processor, its entities pass through as "no_processor".

Step 4: Implement the Processors

Each processor takes a typed entity and returns an enriched dictionary. The resilient_processor decorator catches exceptions so the router never crashes.

from functools import wraps


def resilient_processor(func):
    @wraps(func)
    def wrapper(entity):
        try:
            return func(entity)
        except Exception as e:
            return {
                "error": str(e), "entity_raw": entity.model_dump(),
                "processor": func.__name__, "status": "failed",
            }
    return wrapper


@resilient_processor
def process_person(person: Person) -> dict:
    name_parts = person.name.strip().split()
    first_name = name_parts[0] if name_parts else ""
    last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else ""

    seniority = "unknown"
    if person.role:
        role_lower = person.role.lower()
        if any(t in role_lower for t in ["ceo", "cto", "cfo", "coo", "vp", "president", "founder"]):
            seniority = "executive"
        elif any(t in role_lower for t in ["director", "head", "lead", "manager"]):
            seniority = "senior"
        elif any(t in role_lower for t in ["engineer", "developer", "analyst"]):
            seniority = "mid"

    return {
        "original_name": person.name,
        "first_name": first_name,
        "last_name": last_name,
        "role": person.role,
        "company": person.company,
        "email": person.email,
        "seniority": seniority,
        "is_decision_maker": seniority in ("executive", "senior"),
    }

is_decision_maker is a derived field a CRM would use to prioritize leads. An email that mentions a CTO and a VP is automatically flagged as a high-value opportunity.

The other three processors follow the same pattern (complete code in The Complete System):

process_company: normalizes the name (strips suffixes like "S.A. de C.V."), enriches the industry against a KNOWN_INDUSTRIES dictionary that simulates a database (Clearbit/Crunchbase in production), flags whether it's an investor.
process_date: computes days_until, classifies urgency as overdue/urgent/upcoming/future. It turns a date into actionable information.
process_amount: converts to USD with EXCHANGE_RATES, classifies scale (small/medium/large/enterprise), flags is_significant.

Step 5: Add Retry and Error Handling

The retry is already implemented in the extractor (Step 2) with tenacity. Here we add a variant that injects the previous attempt's error into the prompt:

def extract_with_error_feedback(document: str, max_retries: int = 3) -> DocumentEntities:
    """Extract with a retry that includes feedback from the previous error."""
    raw_extractor = model.with_structured_output(DocumentEntities, include_raw=True)
    last_error = None

    for attempt in range(max_retries):
        prompt = EXTRACTION_PROMPT.format(document=document)
        if last_error:
            prompt += (
                f"\n\nPREVIOUS ATTEMPT FAILED: {last_error}\n"
                "Fix the format so it complies with the schema."
            )

        result = raw_extractor.invoke(prompt)
        if result["parsing_error"] is None and result["parsed"] is not None:
            return result["parsed"]

        last_error = str(result["parsing_error"])[:300] if result["parsing_error"] else "parsed=None"
        logger.warning(f"Attempt {attempt + 1} failed: {last_error[:100]}")

    raise ExtractionError(f"Failed after {max_retries} attempts: {last_error}")

If the model formatted a date wrong, as "March 15" instead of "2026-03-15", the next attempt receives that error as context and corrects it. Smarter than a blind retry.

The Complete System

Here's everything consolidated into one runnable file. Copy it into extraction_system.py and run it.

# extraction_system.py
import os
import logging
from datetime import datetime, date
from typing import Optional, List, Callable
from functools import wraps

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field, field_validator
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ============================================================
# Schemas
# ============================================================

class Person(BaseModel):
    name: str = Field(description="The person's full name.")
    role: Optional[str] = Field(default=None, description="Title. ONLY if mentioned.")
    company: Optional[str] = Field(default=None, description="Company. ONLY if mentioned.")
    email: Optional[str] = Field(default=None, description="Email. NEVER invent one.")

class Company(BaseModel):
    name: str = Field(description="The company's full name.")
    industry: Optional[str] = Field(default=None, description="Industry if inferable.")
    country: Optional[str] = Field(default=None, description="Country if mentioned.")

    @field_validator("name")
    @classmethod
    def name_not_empty(cls, v: str) -> str:
        if len(v.strip()) < 2:
            raise ValueError("Company name must have 2+ characters")
        return v.strip()

class DateEntity(BaseModel):
    original_text: str = Field(description="The original date text.")
    normalized: Optional[str] = Field(default=None, description="YYYY-MM-DD if possible.")
    context: str = Field(description="What it's for: deadline, event, founding, etc.")

    @field_validator("normalized")
    @classmethod
    def validate_date_format(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        datetime.strptime(v, "%Y-%m-%d")
        return v

class Amount(BaseModel):
    value: float = Field(description="Numeric value. '12 million' = 12000000.0", ge=0)
    currency: str = Field(description="ISO 4217 code: USD, MXN, EUR.")
    context: str = Field(description="What it's for: budget, investment, price, etc.")

    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        valid = {"USD", "MXN", "EUR", "GBP", "BRL", "COP", "ARS", "CLP", "PEN"}
        v = v.upper().strip()
        if v not in valid:
            raise ValueError(f"Currency '{v}' not recognized")
        return v

class DocumentEntities(BaseModel):
    people: List[Person] = Field(default_factory=list, description="EVERY person.")
    companies: List[Company] = Field(default_factory=list, description="EVERY company.")
    dates: List[DateEntity] = Field(default_factory=list, description="EVERY date.")
    amounts: List[Amount] = Field(default_factory=list, description="EVERY amount.")
    document_type: str = Field(description="'email', 'invoice', 'news_article', 'other'.")
    summary: str = Field(description="One-sentence summary.")


# ============================================================
# Extractor with Retry
# ============================================================

EXTRACTION_PROMPT = """Analyze the following document and extract EVERY entity.

Rules:
- Extract EVERY person, company, date and amount. Don't skip any.
- For Optional fields, use null if the information isn't in the text.
- NEVER invent data that isn't in the document.
- Normalize dates to YYYY-MM-DD whenever possible.
- Convert "12 million" to 12000000.0, "45K" to 45000.0.

Document:
{document}"""

model = init_chat_model("openai:gpt-4.1-mini")
extractor = model.with_structured_output(DocumentEntities, include_raw=True)

class ExtractionError(Exception):
    pass

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((ExtractionError, ValueError)),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def extract_with_retry(document: str) -> DocumentEntities:
    prompt = EXTRACTION_PROMPT.format(document=document)
    result = extractor.invoke(prompt)

    if result["parsing_error"] is not None:
        raise ExtractionError(f"Validation: {str(result['parsing_error'])[:200]}")

    parsed = result["parsed"]
    if parsed is None:
        raise ExtractionError("Extraction returned None")

    total = len(parsed.people) + len(parsed.companies) + len(parsed.dates) + len(parsed.amounts)
    if total == 0:
        raise ExtractionError("0 entities extracted")

    return parsed


# ============================================================
# Processors
# ============================================================

def resilient_processor(func):
    @wraps(func)
    def wrapper(entity):
        try:
            return func(entity)
        except Exception as e:
            return {"error": str(e), "entity_raw": entity.model_dump(), "status": "failed"}
    return wrapper

@resilient_processor
def process_person(person: Person) -> dict:
    name_parts = person.name.strip().split()
    first_name = name_parts[0] if name_parts else ""
    last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else ""

    seniority = "unknown"
    if person.role:
        rl = person.role.lower()
        if any(t in rl for t in ["ceo", "cto", "cfo", "coo", "vp", "president", "founder"]):
            seniority = "executive"
        elif any(t in rl for t in ["director", "head", "lead", "manager"]):
            seniority = "senior"
        elif any(t in rl for t in ["engineer", "developer", "analyst"]):
            seniority = "mid"

    return {
        "original_name": person.name, "first_name": first_name, "last_name": last_name,
        "role": person.role, "company": person.company, "email": person.email,
        "seniority": seniority, "is_decision_maker": seniority in ("executive", "senior"),
    }

KNOWN_INDUSTRIES = {
    "techcorp": "AI/ML", "cloudserv": "cloud infrastructure",
    "dataai": "AI/ML", "neuralsoft": "AI/ML",
    "sequoia": "venture capital", "andreessen": "venture capital", "allvp": "venture capital",
}

@resilient_processor
def process_company(company: Company) -> dict:
    name_clean = company.name.strip()
    name_key = name_clean.lower().split()[0] if name_clean else ""
    industry = company.industry or KNOWN_INDUSTRIES.get(name_key, "unknown")

    name_short = name_clean
    for suffix in ["S.A.", "S.A. de C.V.", "Inc.", "LLC", "Corp.", "Ltd."]:
        name_short = name_short.replace(suffix, "").strip()

    return {
        "full_name": name_clean, "short_name": name_short,
        "industry": industry, "country": company.country,
        "is_investor": industry == "venture capital",
    }

EXCHANGE_RATES = {
    "USD": 1.0, "MXN": 0.058, "EUR": 1.08, "GBP": 1.26,
    "BRL": 0.19, "COP": 0.00024, "ARS": 0.0011, "CLP": 0.0011, "PEN": 0.27,
}

@resilient_processor
def process_date(date_entity: DateEntity) -> dict:
    result = {
        "original": date_entity.original_text, "normalized": date_entity.normalized,
        "context": date_entity.context, "is_past": None, "days_until": None, "urgency": "unknown",
    }
    if date_entity.normalized:
        try:
            target = datetime.strptime(date_entity.normalized, "%Y-%m-%d").date()
            delta = (target - date.today()).days
            result["is_past"] = delta < 0
            result["days_until"] = delta
            result["urgency"] = (
                "overdue" if delta < 0 else
                "urgent" if delta <= 7 else
                "upcoming" if delta <= 30 else "future"
            )
        except ValueError:
            pass
    return result

@resilient_processor
def process_amount(amount: Amount) -> dict:
    rate = EXCHANGE_RATES.get(amount.currency)
    usd_value = amount.value * rate if rate else None
    scale = "unknown"
    if usd_value is not None:
        scale = (
            "small" if usd_value < 1000 else
            "medium" if usd_value < 50000 else
            "large" if usd_value < 1000000 else "enterprise"
        )
    return {
        "original_value": amount.value, "currency": amount.currency,
        "usd_equivalent": round(usd_value, 2) if usd_value else None,
        "context": amount.context, "scale": scale,
        "is_significant": scale in ("large", "enterprise"),
        "formatted": f"${amount.value:,.2f} {amount.currency}",
    }


# ============================================================
# Router
# ============================================================

class EntityRouter:
    def __init__(self):
        self.processors: dict[str, Callable] = {}

    def register(self, entity_type: str, processor: Callable):
        self.processors[entity_type] = processor

    def route(self, entities: DocumentEntities) -> dict:
        results = {
            "document_type": entities.document_type, "summary": entities.summary,
            "processed_entities": {}, "stats": {},
        }
        entity_map = {
            "people": entities.people, "companies": entities.companies,
            "dates": entities.dates, "amounts": entities.amounts,
        }
        for entity_type, entity_list in entity_map.items():
            if not entity_list:
                results["stats"][entity_type] = 0
                continue

            processor = self.processors.get(entity_type)
            if processor is None:
                results["processed_entities"][entity_type] = [
                    {"raw": e.model_dump(), "status": "no_processor"} for e in entity_list
                ]
            else:
                results["processed_entities"][entity_type] = [
                    {"data": processor(e), "status": "ok" if "error" not in processor(e) else "error"}
                    for e in entity_list
                ]
            results["stats"][entity_type] = len(entity_list)

        return results


# ============================================================
# Pipeline
# ============================================================

def create_pipeline() -> EntityRouter:
    router = EntityRouter()
    router.register("people", process_person)
    router.register("companies", process_company)
    router.register("dates", process_date)
    router.register("amounts", process_amount)
    return router

def process_document(document: str, router: EntityRouter) -> dict:
    logger.info("Extracting entities...")
    entities = extract_with_retry(document)
    counts = f"{len(entities.people)}P {len(entities.companies)}C {len(entities.dates)}D {len(entities.amounts)}A"
    logger.info(f"Extracted: {counts}")

    logger.info("Routing to processors...")
    results = router.route(entities)
    logger.info("Processing complete.")
    return results

def print_report(results: dict) -> None:
    print(f"\n{'='*70}")
    print(f"  EXTRACTION + ROUTING REPORT")
    print(f"{'='*70}")
    print(f"  Type: {results['document_type']}")
    print(f"  Summary: {results['summary']}")
    print(f"  Stats: {results['stats']}")
    print(f"{'='*70}")

    for etype, entities in results["processed_entities"].items():
        print(f"\n  [{etype.upper()}] ({len(entities)})")
        for i, ent in enumerate(entities, 1):
            d = ent["data"]
            if etype == "people" and "error" not in d:
                dm = " [DECISION MAKER]" if d.get("is_decision_maker") else ""
                print(f"    {i}. {d['original_name']} | {d.get('role') or '—'} | {d['seniority']}{dm}")
            elif etype == "companies" and "error" not in d:
                inv = " [INVESTOR]" if d.get("is_investor") else ""
                print(f"    {i}. {d['short_name']} | {d['industry']}{inv}")
            elif etype == "dates" and "error" not in d:
                urg = f" [{d['urgency'].upper()}]" if d["urgency"] != "unknown" else ""
                days = f" ({d['days_until']}d)" if d.get("days_until") is not None else ""
                print(f"    {i}. {d['original']} → {d['normalized'] or '?'} | {d['context']}{urg}{days}")
            elif etype == "amounts" and "error" not in d:
                usd = f" ≈ ${d['usd_equivalent']:,.2f} USD" if d.get("usd_equivalent") else ""
                sig = " [SIGNIFICANT]" if d.get("is_significant") else ""
                print(f"    {i}. {d['formatted']} | {d['context']} | {d['scale']}{sig}{usd}")
            else:
                print(f"    {i}. [ERROR] {d}")

    print(f"\n{'='*70}\n")


# ============================================================
# Main
# ============================================================

if __name__ == "__main__":
    router = create_pipeline()

    documents = {
        "Commercial email": open("sample_documents/commercial_email.txt").read(),
        "Invoice": open("sample_documents/service_invoice.txt").read(),
        "News article": open("sample_documents/news_article.txt").read(),
    }

    for name, text in documents.items():
        print(f"\n{'#'*70}")
        print(f"  Processing: {name}")
        print(f"{'#'*70}")
        try:
            results = process_document(text, router)
            print_report(results)
        except ExtractionError as e:
            print(f"  ERROR: {e}")
        except Exception as e:
            print(f"  UNEXPECTED ERROR: {e}")

Recommended Tests

Run the system with all three documents and check the output.

Test 1: Commercial email

Expected:

document_type: "email"
4 people: Maria Garcia (CTO, executive), Juan Ramirez, Roberto Sanchez (VP, executive), Ana Lopez
3 companies: TechCorp, CloudServ Solutions, DataAI
2 dates: March 3 (conversation), April 15 (deadline)
2 amounts: $45,000 USD (budget), $15,000 USD (deposit)
2 decision makers flagged, both amounts classified as "medium"

Test 2: Invoice

Expected:

document_type: "invoice"
1-2 people (Roberto Sanchez as the contact)
2 companies (CloudServ as the issuer, TechCorp as the client)
2 dates (issued 2026-03-05, due 2026-04-05)
5+ amounts (per-item subtotals, tax, total of $4,408 USD)

Test 3: News article

Expected:

document_type: "news_article"
3+ people: Maria Garcia (CEO/CTO), Juan Ramirez (CFO), Carlos Mendoza
5+ companies: TechCorp, Sequoia Capital, Andreessen Horowitz, ALLVP, DataAI, NeuralSoft
3+ dates: 2023 (founding), March 7, 2026 (the article), December 2026 (goal)
5+ amounts: $12M, $4M, $5M, $3M, $2.8M, $8.5M — several classified as "enterprise"

Test 4: Edge cases

# Minimal text
results = process_document("Meeting on Tuesday with Pedro.", router)
# Expected: 1 person, 1 date, 0 companies, 0 amounts

# Text with no entities — it should retry and eventually fail or return a minimum
try:
    results = process_document(
        "Software development has evolved over the last decade.", router
    )
except ExtractionError:
    print("Handled correctly")

Success Criteria

1. It extracts entities from varied documents

Emails, invoices and news articles all produce correctly typed entities. A single schema and a single prompt handle all three formats.

2. Correct routing by entity type

Each person goes to process_person, each company to process_company, each date to process_date, each amount to process_amount. No cross-contamination.

3. It handles errors without stopping

If a processor fails for one entity, the rest still get processed. If extraction fails, the retry tries again. The system reports errors but doesn't crash.

4. It combines 3+ of the module's patterns

Pattern	Implementation
Structured Extraction	`with_structured_output` + `DocumentEntities`
Tool Routing	`EntityRouter` dispatches by type
Retry + Backoff	`@retry` with tenacity
Composition	Pipeline: extract → route → process → aggregate

Completion Checklist

Setup:

.env with OPENAI_API_KEY
Dependencies: langchain, langchain-openai, tenacity, python-dotenv
sample_documents/ with 3 test files

Schemas:

Person: name, role, company, email with "NEVER invent" descriptions
Company: validator that rejects names < 2 characters
DateEntity: YYYY-MM-DD format validator, context field
Amount: valid-currency validator, ge=0 on value
DocumentEntities: groups the 4 types + document_type + summary

Extractor:

with_structured_output(include_raw=True) to capture errors
Prompt with explicit extraction rules
Retry: 3 attempts, exponential backoff (2s, 4s, 8s)
Verifies entities > 0

Router + Processors:

EntityRouter with dynamic registration
process_person: infers seniority, flags decision makers
process_company: normalizes the name, enriches the industry
process_date: computes days_until, classifies urgency
process_amount: converts to USD, classifies scale
All of them with the resilient_processor decorator

Pipeline:

process_document orchestrates extract → route → aggregate
print_report shows a formatted result
Handles ExtractionError without crashing

Common Errors

Error 1: The model invents emails that aren't in the text

Symptom: Person(email='juan@techcorp.com') but that email never appears in the document.

Solution: An explicit "NEVER invent an email" description on the field + reinforcement in the prompt: "For Optional fields, use null if the information isn't explicitly there."

Error 2: Amounts converted incorrectly ("12 million" → 12.0)

Symptom: Amount(value=12.0) instead of Amount(value=12000000.0).

Solution: An explicit instruction in the prompt and in the field description: "Convert '12 million' to 12000000.0, '45K' to 45000.0."

Error 3: Dates normalized without a year or without zero-padding

Symptom: normalized='March 15' or normalized='2026-3-15'.

Solution: The validator rejects these formats and the retry corrects them. Reinforce it: "STRICT YYYY-MM-DD format with zero-padding."

Error 4: Infinite retry with documents that have no real entities

Symptom: 3 failed attempts with ExtractionError("0 entities") for abstract texts.

Solution: Condition the check on long documents:

if total == 0 and len(document) > 100:
    raise ExtractionError("0 entities in a long document")

Error 5: with_structured_output fails with models that lack function calling

Symptom: NotImplementedError with local models.

Solution: Use GPT-4.1, Claude 3.5/4, or Gemini 2. Fallback: method="json_mode".

Error 6: Tenacity doesn't catch Pydantic's ValidationError

Symptom: The retry doesn't fire when Pydantic rejects the response.

Solution: include_raw=True captures the error in parsing_error without raising an exception. Our code checks that field and raises ExtractionError, which does trigger the retry.

Error 7: Duplicate entities between Person.company and the companies list

Symptom: "TechCorp" appears both as a company and as a person's field.

Solution: It isn't an error — it's complementary information. Person.company states membership; the companies list indicates relevance in the document. In production, deduplicate in the aggregation phase.

Resources

LangChain — Structured Output — with_structured_output with include_raw and error handling
LangChain — Extraction Tutorial — Extraction with function calling
Pydantic V2 — Validators — field_validator, constraints
Tenacity Documentation — Retry patterns with exponential backoff
Martin Fowler — Circuit Breaker — The circuit breaker pattern
OpenAI Structured Outputs — Native structured outputs

Connection to the Next Module

In this project you built a system that combines 3+ function calling patterns: structured extraction to pull typed entities out of free text, routing to direct each type to a specialized processor, retry for resilience, and composition to integrate it all into a pipeline. It processes real documents and produces structured reports with enriched entities.

But the system has a limitation: the flow is linear. A document comes in, the pipeline processes it, a result comes out. There are no conditional decisions ("if it's an invoice, do X; if it's an email, do Y"), there are no loops ("if it didn't extract enough entities, rephrase and retry with a different strategy"), there's no state persisted between documents.

In Module 4 (State Machines for Agents) you'll model flows as graphs with LangGraph. This project's extraction + routing would become a StateGraph with a node per phase, conditional edges, and the ability to cycle when a phase needs to retry. The evolving project kicks off: you'll build a Research Agent with a state machine that controls how it researches, which tools it uses, and when to stop.

This module's patterns — extraction, routing, retry, composition — are your repertoire. Module 4 gives you the control framework to orchestrate them.

← Previous7. Retry Patterns and Circuit Breakers Next →1. Introduction: Agents as graphs