8. Project: Multi-Stage Analysis Pipeline

Project Overview

In this project you'll build a complete, production-ready document analysis pipeline. The system receives a document of any length, processes it in four specialized stages (Extract → Analyze → Synthesize → Format), with smart context management for long documents, routing by complexity, robust error handling with retry, and structured output.

This project integrates every technique from Module 06:

• Prompt Chaining (sequential stages)
• Task Decomposition (each stage has a single responsibility)
• Multi-Stage Pipelines (state, retry, fallbacks)
• Context Window Management (chunking and summarization)
• Routing by Complexity (model selection based on size)

System Specifications

Required capabilities

Architecture

Input Document
      │
      ▼
┌─────────────────────────────────────────────┐
│  PRE-PROCESSING                              │
│  - Count tokens with tiktoken                │
│  - If > 8K tokens: Summarization Chain       │
│  - Prepare the document metadata             │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  STAGE 1: EXTRACT                            │
│  - Extract the key sections                  │
│  - Identify data, figures, dates             │
│  - Detect the important entities             │
│  Model: gpt-4o-mini                          │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  STAGE 2: ANALYZE                            │
│  - Analysis of patterns and trends           │
│  - Identify causes and effects               │
│  - Assess risks and opportunities            │
│  Model: gpt-4o (greater capability)          │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  STAGE 3: SYNTHESIZE                         │
│  - Generate actionable insights              │
│  - Prioritize findings by impact             │
│  - Create a coherent narrative               │
│  Model: gpt-4o                               │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  STAGE 4: FORMAT                             │
│  - Structure into JSON with Pydantic         │
│  - Validate coherence and completeness       │
│  - Generate the executive summary            │
│  Model: gpt-4o-mini                          │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
         Structured output (JSON)

Complete Implementation

Data models

from openai import OpenAI
from pydantic import BaseModel, Field
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Any
import time
import json
import tiktoken
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

client = OpenAI()


# ============================================================
# DATA MODELS
# ============================================================

class Finding(BaseModel):
    category: str = Field(..., description="Category of the finding: key_data, risk, opportunity, etc.")
    description: str = Field(..., description="Detailed description of the finding")
    impact: str = Field(..., description="low, medium, high")
    evidence: str = Field(..., description="Quote or data point from the document that supports the finding")

class Recommendation(BaseModel):
    action: str = Field(..., description="Concrete action to take")
    priority: str = Field(..., description="urgent, important, nice-to-have")
    owner: Optional[str] = Field(None, description="Responsible area or role")
    deadline: Optional[str] = Field(None, description="Suggested deadline: immediate, 30 days, 90 days, etc.")

class AnalysisResult(BaseModel):
    title: str = Field(..., description="Descriptive title of the analyzed document")
    executive_summary: str = Field(..., description="A 3-5 sentence summary for an executive")
    document_type: str = Field(..., description="report, contract, article, data, other")
    findings: list[Finding] = Field(..., description="List of the main findings (10 max)")
    recommendations: list[Recommendation] = Field(..., description="Recommended actions (5 max)")
    confidence_score: float = Field(..., description="Confidence in the analysis (0-1)")
    metadata: dict = Field(default_factory=dict, description="Pipeline metrics")

@dataclass
class PipelineState:
    """Shared state during the pipeline execution."""
    original_document: str
    processed_document: str  # May be summarized if it's very long
    
    # Outputs of each stage
    output_extract: str = ""
    output_analyze: str = ""
    output_synthesize: str = ""
    output_final: Optional[AnalysisResult] = None
    
    # Metrics
    tokens_input: int = 0
    total_tokens_used: int = 0
    start_time: float = field(default_factory=time.time)
    stage_times: dict = field(default_factory=dict)
    models_used: dict = field(default_factory=dict)
    stage_retries: dict = field(default_factory=dict)
    errors: list[str] = field(default_factory=list)
    was_summarized: bool = False

Context utilities

# ============================================================
# CONTEXT UTILITIES
# ============================================================

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    """Counts the tokens of a text using tiktoken."""
    try:
        enc = tiktoken.encoding_for_model(model)
    except KeyError:
        enc = tiktoken.get_encoding("cl100k_base")
    return len(enc.encode(text))


def summarize_chunk(chunk: str, n_words: int = 200) -> str:
    """Summarizes a fragment of text in n_words."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Summarize the following text in roughly {n_words} words.
Keep the most important data, figures, dates and key conclusions.
Do not omit critical information.

Text:
{chunk}"""
        }],
        temperature=0,
        max_tokens=400
    )
    return response.choices[0].message.content


def preprocess_document(
    document: str,
    max_tokens: int = 8000,
    verbose: bool = True
) -> tuple[str, bool, int]:
    """
    Preprocesses the document so it fits into the pipeline's context.
    
    Returns:
        Tuple (processed_document, was_summarized, original_tokens)
    """
    original_tokens = count_tokens(document)
    
    if verbose:
        logger.info(f"Original document: {original_tokens:,} tokens ({len(document):,} chars)")
    
    if original_tokens <= max_tokens:
        return document, False, original_tokens
    
    # The document is too long: chunking + summarization
    if verbose:
        logger.info(f"Document exceeds {max_tokens:,} tokens. Applying summarization...")
    
    # Split into ~3000-token chunks with overlap
    encoder = tiktoken.encoding_for_model("gpt-4o-mini")
    tokens = encoder.encode(document)
    
    chunk_size = 3000
    overlap = 200
    token_chunks = []
    
    i = 0
    while i < len(tokens):
        chunk_tokens = tokens[i:i + chunk_size]
        token_chunks.append(encoder.decode(chunk_tokens))
        i += chunk_size - overlap
    
    if verbose:
        logger.info(f"Split into {len(token_chunks)} chunks")
    
    # Summarize each chunk
    summaries = []
    for idx, chunk in enumerate(token_chunks):
        if verbose:
            logger.info(f"  Summarizing chunk {idx + 1}/{len(token_chunks)}...")
        summary = summarize_chunk(chunk, n_words=150)
        summaries.append(f"[Fragment {idx + 1}]\n{summary}")
    
    summarized_document = "\n\n".join(summaries)
    final_tokens = count_tokens(summarized_document)
    
    if verbose:
        logger.info(f"Summarized document: {final_tokens:,} tokens ({original_tokens/final_tokens:.1f}x reduction)")
    
    return summarized_document, True, original_tokens

The four pipeline stages

# ============================================================
# PIPELINE STAGES
# ============================================================

PROMPT_EXTRACT = """You are an expert in document analysis. Your task is to extract structured information.

Document to analyze:
{document}

Extract and organize:
1. **Main topic**: What is this document about?
2. **Document type**: report, contract, article, data, presentation, other
3. **Key data**: figures, percentages, dates, names, important places
4. **Main sections**: List the sections or topics covered
5. **Arguments or conclusions**: The document's main claims
6. **Entities mentioned**: People, organizations, products

Be thorough but concise. Prioritize the quantitative data."""


PROMPT_ANALYZE = """You are a senior analyst. Based on the following extraction of a document, carry out a deep analysis.

Document extraction:
{extracted}

Analyze:
1. **Patterns and trends**: What patterns emerge from the data and arguments?
2. **Strengths**: What positive aspects or strengths does the document identify?
3. **Risks or problems**: What risks, weaknesses or problems are mentioned?
4. **Opportunities**: What opportunities for improvement or growth are there?
5. **Causes and effects**: What cause-effect relationships are evident?
6. **Context**: What external context is relevant to interpreting this?
7. **Information gaps**: What important information is missing or ambiguous?

Be critical and objective. Distinguish between facts and opinions."""


PROMPT_SYNTHESIZE = """You are a strategy consultant. Based on the analysis below, synthesize the key findings and generate actionable recommendations.

Analysis:
{analyzed}

Synthesize:
1. **The 5 most important findings** (ordered by impact): Each with a category (key_data, risk, opportunity, trend), a clear description, an impact level (high/medium/low), and evidence from the document.
2. **The 3-5 most important recommendations**: Each with a specific action, a priority (urgent/important/nice-to-have), a suggested owner, and a deadline.
3. **Executive summary**: 3-5 sentences an executive can read in 30 seconds.
4. **Confidence score** (0.0-1.0): How reliable is this analysis given the original document?

Be specific and actionable. Avoid generalities."""


PROMPT_FORMAT = """Convert the following synthesis into structured JSON.

Synthesis:
{synthesized}

Required JSON:
{{
  "title": "descriptive title of the document",
  "executive_summary": "3-5 sentences for an executive",
  "document_type": "report|contract|article|data|other",
  "findings": [
    {{
      "category": "key_data|risk|opportunity|trend",
      "description": "description of the finding",
      "impact": "high|medium|low",
      "evidence": "quote or data point that supports it"
    }}
  ],
  "recommendations": [
    {{
      "action": "specific action to take",
      "priority": "urgent|important|nice-to-have",
      "owner": "area or role (can be null)",
      "deadline": "immediate|30 days|90 days (can be null)"
    }}
  ],
  "confidence_score": 0.0
}}

Maximum 10 findings and 5 recommendations. Keep the descriptions concise but complete."""


def run_stage(
    name: str,
    prompt: str,
    model: str,
    max_tokens: int,
    max_retries: int = 2,
    json_mode: bool = False,
    verbose: bool = True
) -> tuple[str, int, int]:
    """
    Runs one pipeline stage with automatic retry.
    
    Returns:
        Tuple (answer, retries_done, tokens_used)
    """
    last_error = None
    
    for attempt in range(max_retries + 1):
        try:
            kwargs = {
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0,
                "max_tokens": max_tokens
            }
            if json_mode:
                kwargs["response_format"] = {"type": "json_object"}
            
            response = client.chat.completions.create(**kwargs)
            answer = response.choices[0].message.content
            tokens = response.usage.total_tokens
            
            if verbose and attempt > 0:
                logger.info(f"  {name}: Success on attempt {attempt + 1}")
            
            return answer, attempt, tokens
            
        except Exception as e:
            last_error = str(e)
            if verbose:
                logger.warning(f"  {name}: Error on attempt {attempt + 1}: {e}")
            
            if attempt < max_retries:
                time.sleep(2 ** attempt)  # Exponential backoff
    
    raise RuntimeError(f"Stage '{name}' failed after {max_retries + 1} attempts. Last error: {last_error}")


def stage_extract(state: PipelineState, verbose: bool = True) -> None:
    """Stage 1: Extraction of structured information."""
    if verbose:
        logger.info("▶ Stage 1/4: EXTRACT")
    
    t0 = time.time()
    prompt = PROMPT_EXTRACT.format(document=state.processed_document)
    
    answer, retries, tokens = run_stage(
        name="extract",
        prompt=prompt,
        model="gpt-4o-mini",
        max_tokens=1500,
        verbose=verbose
    )
    
    state.output_extract = answer
    state.stage_times["extract"] = time.time() - t0
    state.models_used["extract"] = "gpt-4o-mini"
    state.stage_retries["extract"] = retries
    state.total_tokens_used += tokens
    
    if verbose:
        logger.info(f"  ✓ Extract: {len(answer)} chars, {time.time()-t0:.1f}s")


def stage_analyze(state: PipelineState, verbose: bool = True) -> None:
    """Stage 2: Deep analysis with gpt-4o."""
    if verbose:
        logger.info("▶ Stage 2/4: ANALYZE")
    
    t0 = time.time()
    
    # If the document was very long, use gpt-4o for greater synthesis capability
    model = "gpt-4o" if state.was_summarized or state.tokens_input > 3000 else "gpt-4o-mini"
    prompt = PROMPT_ANALYZE.format(extracted=state.output_extract)
    
    answer, retries, tokens = run_stage(
        name="analyze",
        prompt=prompt,
        model=model,
        max_tokens=2000,
        verbose=verbose
    )
    
    state.output_analyze = answer
    state.stage_times["analyze"] = time.time() - t0
    state.models_used["analyze"] = model
    state.stage_retries["analyze"] = retries
    state.total_tokens_used += tokens
    
    if verbose:
        logger.info(f"  ✓ Analyze ({model}): {len(answer)} chars, {time.time()-t0:.1f}s")


def stage_synthesize(state: PipelineState, verbose: bool = True) -> None:
    """Stage 3: Synthesis and generation of recommendations."""
    if verbose:
        logger.info("▶ Stage 3/4: SYNTHESIZE")
    
    t0 = time.time()
    prompt = PROMPT_SYNTHESIZE.format(analyzed=state.output_analyze)
    
    answer, retries, tokens = run_stage(
        name="synthesize",
        prompt=prompt,
        model="gpt-4o",
        max_tokens=2000,
        verbose=verbose
    )
    
    state.output_synthesize = answer
    state.stage_times["synthesize"] = time.time() - t0
    state.models_used["synthesize"] = "gpt-4o"
    state.stage_retries["synthesize"] = retries
    state.total_tokens_used += tokens
    
    if verbose:
        logger.info(f"  ✓ Synthesize: {len(answer)} chars, {time.time()-t0:.1f}s")


def stage_format(state: PipelineState, verbose: bool = True) -> None:
    """Stage 4: Formatting into structured JSON with Pydantic validation."""
    if verbose:
        logger.info("▶ Stage 4/4: FORMAT")
    
    t0 = time.time()
    prompt = PROMPT_FORMAT.format(synthesized=state.output_synthesize)
    
    answer, retries, tokens = run_stage(
        name="format",
        prompt=prompt,
        model="gpt-4o-mini",
        max_tokens=2000,
        json_mode=True,
        verbose=verbose
    )
    
    # Parse and validate with Pydantic
    try:
        data = json.loads(answer)
        
        # Add the pipeline metadata
        total_time = time.time() - state.start_time
        data["metadata"] = {
            "original_tokens": state.tokens_input,
            "processed_tokens": count_tokens(state.processed_document),
            "total_api_tokens": state.total_tokens_used + tokens,
            "was_summarized": state.was_summarized,
            "total_time_seconds": round(total_time, 2),
            "stage_times": {k: round(v, 2) for k, v in state.stage_times.items()},
            "models_used": state.models_used,
            "total_retries": sum(state.stage_retries.values())
        }
        
        state.output_final = AnalysisResult(**data)
        
    except Exception as e:
        logger.error(f"Error validating the JSON: {e}")
        logger.error(f"JSON received: {answer[:200]}")
        # Fallback: build a minimal AnalysisResult
        state.output_final = AnalysisResult(
            title="Document analysis",
            executive_summary=state.output_synthesize[:300] if state.output_synthesize else "Not available",
            document_type="other",
            findings=[Finding(
                category="key_data",
                description="See the analysis in plain text",
                impact="medium",
                evidence=state.output_analyze[:200] if state.output_analyze else ""
            )],
            recommendations=[Recommendation(
                action="Review the analysis manually",
                priority="important"
            )],
            confidence_score=0.5,
            metadata={"error": str(e), "fallback": True}
        )
    
    state.stage_times["format"] = time.time() - t0
    state.models_used["format"] = "gpt-4o-mini"
    state.stage_retries["format"] = retries
    state.total_tokens_used += tokens
    
    if verbose:
        logger.info(f"  ✓ Format: validated with Pydantic, {time.time()-t0:.1f}s")

Main pipeline

# ============================================================
# MAIN PIPELINE
# ============================================================

def analyze_document(
    document: str,
    max_context_tokens: int = 8000,
    verbose: bool = True
) -> AnalysisResult:
    """
    Complete document analysis pipeline.
    
    Args:
        document: Text of the document to analyze
        max_context_tokens: Threshold that triggers summarization
        verbose: Show detailed logs
    
    Returns:
        AnalysisResult with structured findings and recommendations
    """
    if verbose:
        logger.info("=" * 60)
        logger.info("STARTING THE MULTI-STAGE ANALYSIS PIPELINE")
        logger.info("=" * 60)
    
    # Pre-processing: context management
    processed_document, was_summarized, original_tokens = preprocess_document(
        document,
        max_tokens=max_context_tokens,
        verbose=verbose
    )
    
    # Initialize the pipeline state
    state = PipelineState(
        original_document=document,
        processed_document=processed_document,
        tokens_input=original_tokens,
        was_summarized=was_summarized
    )
    
    # Run the four stages
    try:
        stage_extract(state, verbose=verbose)
        stage_analyze(state, verbose=verbose)
        stage_synthesize(state, verbose=verbose)
        stage_format(state, verbose=verbose)
        
        if verbose:
            total_time = time.time() - state.start_time
            logger.info("=" * 60)
            logger.info(f"✅ PIPELINE COMPLETE in {total_time:.1f}s")
            logger.info(f"   API tokens used: {state.total_tokens_used:,}")
            logger.info(f"   Findings: {len(state.output_final.findings)}")
            logger.info(f"   Recommendations: {len(state.output_final.recommendations)}")
            logger.info("=" * 60)
        
        return state.output_final
        
    except Exception as e:
        logger.error(f"Error in the pipeline: {e}")
        state.errors.append(str(e))
        raise

Result presentation function

def print_result(result: AnalysisResult) -> None:
    """Prints the result of the analysis in a readable way."""
    print("\n" + "=" * 70)
    print(f"📄 {result.title}")
    print(f"Type: {result.document_type} | Confidence: {result.confidence_score:.0%}")
    print("=" * 70)
    
    print(f"\n📋 EXECUTIVE SUMMARY\n{result.executive_summary}")
    
    print(f"\n🔍 KEY FINDINGS ({len(result.findings)})")
    for i, h in enumerate(result.findings, 1):
        impact_emoji = {"high": "🔴", "medium": "🟡", "low": "🟢"}.get(h.impact, "⚪")
        print(f"\n  {i}. {impact_emoji} [{h.category.upper()}] {h.description}")
        print(f"     Evidence: {h.evidence[:100]}...")
    
    print(f"\n💡 RECOMMENDATIONS ({len(result.recommendations)})")
    for i, r in enumerate(result.recommendations, 1):
        prio_emoji = {"urgent": "🚨", "important": "⚡", "nice-to-have": "💭"}.get(r.priority, "•")
        print(f"\n  {i}. {prio_emoji} {r.action}")
        if r.owner:
            print(f"     Owner: {r.owner}", end="")
        if r.deadline:
            print(f" | Deadline: {r.deadline}", end="")
        print()
    
    if result.metadata:
        meta = result.metadata
        print(f"\n⚙️ PIPELINE METADATA")
        print(f"  Total time: {meta.get('total_time_seconds', 'N/A')}s")
        print(f"  Tokens used: {meta.get('total_api_tokens', 'N/A'):,}")
        print(f"  Was summarized: {'Yes' if meta.get('was_summarized') else 'No'}")
        if meta.get("total_retries", 0) > 0:
            print(f"  Retries: {meta['total_retries']}")

Testing the Pipeline

# ============================================================
# TEST DOCUMENTS
# ============================================================

SHORT_DOCUMENT = """
QUARTERLY REPORT Q3 2024 - TECH STARTUP XYZ

Executive summary:
In the third quarter of 2024, XYZ achieved 45% revenue growth over the previous quarter, 
with total revenue of $2.3M. However, operating expenses rose 62%, mainly due to 
the expansion of the engineering team from 12 to 28 people and the opening of our office in Mexico.

Key metrics:
- Monthly active users (MAU): 45,000 (+38% vs Q2)
- 30-day retention rate: 72% (target: 75%)
- Customer acquisition cost (CAC): $85 (vs $120 in Q2, a 29% improvement)
- Average Lifetime Value (LTV): $340 (LTV/CAC ratio: 4.0x)
- Net Promoter Score: 67 (industry benchmark: 45)

Wins of the quarter:
1. Launch of the "Smart Analytics" feature, which increased engagement by 23%
2. Signed an enterprise contract with RetailMega Corp ($400K ARR)
3. A TechCrunch feature that generated 1,200 organic leads
4. Reduced onboarding time from 14 days to 5 days

Challenges:
- The team expansion is squeezing operating margins (gross margin fell from 68% to 61%)
- Three enterprise customers paused their contracts citing "Q4 budget"
- The enterprise sales pipeline has an average delay of 45 days

Next steps for Q4:
- Optimize the enterprise sales process to bring the cycle down to 30 days
- Launch a partner program for expansion across LATAM
- Roll out an SMB pricing tier ($49/month) to diversify the customer base
- Q4 revenue target: $3.2M (+39% vs Q3)
"""

LONG_DOCUMENT = """
MARKET ANALYSIS: ARTIFICIAL INTELLIGENCE IN LATIN AMERICA 2024-2026

SECTION 1: GENERAL LANDSCAPE

The adoption of artificial intelligence in Latin America has seen unprecedented growth 
during 2023-2024. According to the IDC Latin America report, spending on AI technologies in the region 
reached $4.2B USD in 2023, with a projection of $9.8B by 2026, representing a CAGR of 32.5%.

Brazil leads with 48% of the regional market, followed by Mexico (22%), Colombia (10%), Argentina (8%) 
and Chile (6%). The sectors with the highest adoption are: Financial Services (35%), Retail (22%), 
Telecommunications (18%) and Healthcare (12%).

The main growth catalysts are: a 40% reduction in cloud computing costs 
over the last 3 years, the availability of LLM APIs in Spanish, clearer regulatory frameworks 
(especially in Brazil and Chile), and a growing technical talent pool (the region graduated 180,000 engineers in 2023).

SECTION 2: BARRIERS TO ADOPTION

The main barrier identified is the talent gap: there is an estimated shortfall of 85,000 
professionals specialized in AI across the region. Companies report that 67% of their AI projects 
fail or get delayed for lack of qualified talent.

The second barrier is data infrastructure. 72% of Latin American companies still keep 
their data in legacy systems, and only 23% have the structured data strategy needed for 
AI projects. This contrasts with the 64% of North American companies that already have cloud data lakes.

Implementation costs also represent a barrier. The average AI project in LATAM 
requires $500K-$2M of initial investment, out of reach for the SMBs that make up 99% of the region's 
business fabric.

SECTION 3: SUCCESS STORIES

Nubank (Brazil): Implemented ML credit scoring models that reduced the default rate by 31% 
while expanding access to credit to 8M new customers with no banking history. 
Reported ROI: 340% in 18 months.

Rappi (Colombia/Mexico): Its AI-powered recommendation system increased the average order 
value by 22% and reduced predicted delivery time by 15%, improving NPS by 12 points.

MercadoLibre: Its ML-based anti-fraud system blocks 98.7% of fraudulent transactions 
in real time, saving an estimated $800M a year in prevented fraud.

SECTION 4: OUTLOOK 2025-2026

The trends that will dominate the LATAM market over the next two years:

1. Generative AI for Spanish-Language Content: The market for Spanish content generation tools 
will grow 180% in 2025, driven by demand for personalized marketing.

2. AI in Healthcare: 50 new HealthTech AI startups are expected in Brazil and Mexico, with particular 
emphasis on image-based diagnostics and rural telemedicine.

3. Regulation: Chile will enact its AI Law in Q1 2025. Brazil is finalizing its regulatory framework.
Mexico is still in the early stages of regulatory definition.

4. Talent: ML Engineer salaries in LATAM grew 45% in 2024 and are expected to grow 
another 25-30% in 2025-2026. The region will see 45,000 new AI jobs created.

CONCLUSIONS AND RECOMMENDATIONS

For companies looking to implement AI in LATAM, the main recommendations are:
- Prioritize building internal data capabilities before implementing AI
- Consider partnerships with local universities for talent development
- Evaluate AI-as-a-service (AIaaS) solutions to reduce the initial investment
- Monitor regulatory developments, especially in Chile and Brazil
- Explore high-impact, low-complexity use cases as a starting point
""" * 2  # Multiply it to simulate a longer document


if __name__ == "__main__":
    print("=" * 70)
    print("TEST 1: Short document (quarterly report)")
    print("=" * 70)
    
    result = analyze_document(SHORT_DOCUMENT, verbose=True)
    print_result(result)
    
    # Export to JSON
    with open("analysis_result.json", "w", encoding="utf-8") as f:
        json.dump(result.model_dump(), f, ensure_ascii=False, indent=2)
    print("\n✅ Result exported to analysis_result.json")
    
    print("\n" + "=" * 70)
    print("TEST 2: Long document (market analysis)")
    print("=" * 70)
    
    long_result = analyze_document(LONG_DOCUMENT, max_context_tokens=6000, verbose=True)
    print_result(long_result)

Optional Extensions

Extension 1: Support for multiple languages

SUPPORTED_LANGUAGES = {
    "es": "Spanish",
    "en": "English",
    "pt": "Portuguese"
}

def analyze_document_multilingual(
    document: str,
    output_language: str = "en"
) -> AnalysisResult:
    """
    Pipeline with multilingual support.
    Detects the language of the input and generates the output in the specified language.
    """
    # Detect the language
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Detect the language of this text (respond with the 2-letter ISO 639-1 code only): {document[:200]}"
        }],
        temperature=0, max_tokens=5
    )
    input_language = response.choices[0].message.content.strip().lower()[:2]
    
    language_instruction = f"\n\nIMPORTANT: Generate your entire response in {SUPPORTED_LANGUAGES.get(output_language, 'English')}."
    
    # Inject the language instruction into the prompts
    modified_prompts = {
        "extract": PROMPT_EXTRACT + language_instruction,
        "analyze": PROMPT_ANALYZE + language_instruction,
        "synthesize": PROMPT_SYNTHESIZE + language_instruction,
    }
    
    # Run the pipeline with the modified prompts...
    # (full implementation similar to analyze_document)
    return analyze_document(document)

Extension 2: Async pipeline to process multiple documents

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def analyze_document_async(
    document: str,
    name: str = "document"
) -> tuple[str, AnalysisResult]:
    """Async version of the pipeline for parallel processing."""
    # Async implementation of the pipeline
    # Each stage uses async_client.chat.completions.create(...)
    result = analyze_document(document, verbose=False)
    return name, result


async def analyze_multiple_documents(
    documents: dict[str, str]  # name -> content
) -> dict[str, AnalysisResult]:
    """Analyzes multiple documents in parallel."""
    tasks = [
        analyze_document_async(content, name)
        for name, content in documents.items()
    ]
    results_list = await asyncio.gather(*tasks, return_exceptions=True)
    
    results = {}
    for item in results_list:
        if isinstance(item, Exception):
            print(f"Error: {item}")
        else:
            name, result = item
            results[name] = result
    
    return results


# Usage:
batch_documents = {
    "q3_report": SHORT_DOCUMENT,
    "market_analysis": LONG_DOCUMENT[:3000]
}
# results = asyncio.run(analyze_multiple_documents(batch_documents))

Project Success Criteria

Evaluate your implementation against these criteria:

Total: 100 points. Minimum to pass: 70 pts.

Validation checklist

def validate_full_pipeline() -> bool:
    """Validation suite for the pipeline."""
    print("=== PIPELINE VALIDATION ===\n")
    checks = []
    
    # Check 1: Short document
    print("1. Processing a short document...")
    try:
        res = analyze_document(SHORT_DOCUMENT[:500], verbose=False)
        assert res.title, "Missing title"
        assert len(res.findings) >= 1, "No findings"
        assert len(res.recommendations) >= 1, "No recommendations"
        assert 0 <= res.confidence_score <= 1, "Invalid score"
        checks.append(("Short document", True))
        print("   ✅ OK")
    except Exception as e:
        checks.append(("Short document", False))
        print(f"   ❌ FAILED: {e}")
    
    # Check 2: Context management
    print("\n2. Testing context management...")
    long_text = "This is a document. " * 1000  # ~4000 words
    doc_proc, was_summarized, tokens = preprocess_document(long_text, max_tokens=500, verbose=False)
    if was_summarized:
        checks.append(("Context management", True))
        print("   ✅ Summarization triggered correctly")
    else:
        checks.append(("Context management", False))
        print("   ❌ Summarization did not trigger")
    
    # Check 3: JSON format
    print("\n3. Validating the Pydantic output...")
    try:
        fake_data = {
            "title": "Test",
            "executive_summary": "Test summary.",
            "document_type": "report",
            "findings": [{"category": "key_data", "description": "Test", "impact": "high", "evidence": "Test"}],
            "recommendations": [{"action": "Test", "priority": "urgent"}],
            "confidence_score": 0.8
        }
        result = AnalysisResult(**fake_data)
        checks.append(("Pydantic validation", True))
        print("   ✅ Pydantic works correctly")
    except Exception as e:
        checks.append(("Pydantic validation", False))
        print(f"   ❌ Pydantic error: {e}")
    
    # Summary
    total = len(checks)
    successful = sum(1 for _, ok in checks if ok)
    print(f"\n=== RESULT: {successful}/{total} checks passed ===")
    
    return successful == total


if __name__ == "__main__":
    validate_full_pipeline()

Exercises

Exercise 1: Add a "Fact-Checking" stage

Add a fifth stage between Analyze and Synthesize that verifies whether the findings of the analysis are consistent with the original document.

PROMPT_FACTCHECK = """You are a rigorous fact-checker. 
Review whether the following findings from the analysis are supported by the original document.

Original document (summarized):
{document}

Analysis to verify:
{analyzed}

For each important claim:
1. Is it directly supported by the document? (yes/no/partially)
2. Is there exaggeration or over-interpretation?
3. Are there claims that contradict the document?

Generate a corrected version of the analysis that:
- Removes unsupported claims
- Tones down exaggerated claims
- Keeps only what the document supports
- States the level of certainty (high/medium/low) for each point"""

def stage_factcheck(state: PipelineState, verbose: bool = True) -> None:
    """Additional fact-checking stage."""
    if verbose:
        logger.info("▶ FACTCHECK stage")
    
    t0 = time.time()
    prompt = PROMPT_FACTCHECK.format(
        document=state.processed_document[:2000],
        analyzed=state.output_analyze
    )
    
    answer, retries, tokens = run_stage(
        name="factcheck",
        prompt=prompt,
        model="gpt-4o-mini",
        max_tokens=1500,
        verbose=verbose
    )
    
    # Replace the analyze output with the verified version
    state.output_analyze = answer
    state.stage_times["factcheck"] = time.time() - t0
    state.total_tokens_used += tokens
    
    if verbose:
        logger.info(f"  ✓ Factcheck: {time.time()-t0:.1f}s")

Exercise 2: Export to multiple formats

Implement functions to export the AnalysisResult to Markdown and HTML in addition to JSON.

def export_markdown(result: AnalysisResult) -> str:
    """Generates a report in Markdown format."""
    md = f"# {result.title}\n\n"
    md += f"**Type:** {result.document_type} | **Confidence:** {result.confidence_score:.0%}\n\n"
    md += f"## Executive Summary\n{result.executive_summary}\n\n"
    md += "## Key Findings\n\n"
    
    for i, h in enumerate(result.findings, 1):
        impact_badge = f"![{h.impact}]()"
        md += f"### {i}. {h.description}\n"
        md += f"- **Category:** {h.category}\n"
        md += f"- **Impact:** {h.impact}\n"
        md += f"- **Evidence:** _{h.evidence}_\n\n"
    
    md += "## Recommendations\n\n"
    for i, r in enumerate(result.recommendations, 1):
        md += f"### {i}. {r.action}\n"
        md += f"- **Priority:** {r.priority}\n"
        if r.owner:
            md += f"- **Owner:** {r.owner}\n"
        if r.deadline:
            md += f"- **Deadline:** {r.deadline}\n"
        md += "\n"
    
    return md

def export_html(result: AnalysisResult) -> str:
    """Generates a report in basic HTML format."""
    html = f"<html><body><h1>{result.title}</h1>"
    html += f"<p><b>Type:</b> {result.document_type}</p>"
    html += f"<h2>Summary</h2><p>{result.executive_summary}</p>"
    html += "<h2>Findings</h2><ul>"
    for h in result.findings:
        html += f"<li><b>{h.description}</b> ({h.impact}): {h.evidence[:80]}...</li>"
    html += "</ul>"
    html += "<h2>Recommendations</h2><ol>"
    for r in result.recommendations:
        html += f"<li>{r.action} [{r.priority}]</li>"
    html += "</ol></body></html>"
    return html

# Usage:
# result = analyze_document(SHORT_DOCUMENT)
# with open("report.md", "w") as f:
#     f.write(export_markdown(result))
# with open("report.html", "w") as f:
#     f.write(export_html(result))

Module Summary

You've completed Module 06: Prompt Composition. This project has shown how to integrate every technique from the module into a real production system.

What you built:

• Prompt Chaining (Capsule 02): The 4 sequential stages, each receiving the output of the previous one
• Task Decomposition (Capsule 03): Extract, Analyze, Synthesize, Format as specialized sub-tasks
• Multi-Stage Pipelines (Capsule 04): PipelineState, retry, fallbacks, metrics logging
• Context Window Management (Capsule 05): Chunking with tiktoken, automatic summarization
• Routing by Complexity (Capsule 07): Mini for extraction/formatting, gpt-4o for analysis/synthesis

Suggested next steps:

1. Add result persistence in a database
2. Create an API with FastAPI that receives documents and returns the analysis
3. Implement async processing with a job queue
4. Add a cache for documents that have already been analyzed
5. Build a monitoring dashboard with usage metrics

Additional resources

1. tiktoken GitHub
2. Pydantic v2 documentation
3. OpenAI Structured Outputs
4. LangChain Document Loaders
5. Building LLM-Powered Apps - Patterns and Architectures
6. Python asyncio documentation