CrewAI real-world example: financial report summarization

This crewai financial report summarization example shows you how to build a multi-agent pipeline that ingests earnings releases, extracts key metrics, compares them against consensus estimates, and produces a structured summary memo. You will wire together three specialized agents — an extractor, an analyst, and a writer — using CrewAI’s task delegation and context passing. The complete runnable code is included so you can adapt it to your own document formats and output requirements.

Step 1: Set up the environment and dependencies

Create a fresh virtual environment and install the minimal set of packages. CrewAI sits on top of LangChain, so you need an LLM provider configured. This example uses OpenAI-compatible endpoints; swap the base URL and model name for your preferred provider.

python -m venv .venv
source .venv/bin/activate
pip install "crewai[tools]" langchain-openai python-dotenv pypdf pandas

Create a .env file with your credentials. If you route through a gateway that normalizes 240+ models behind one OpenAI-compatible endpoint, you only need one base URL and key.

# .env
OPENAI_API_KEY=sk-...
OPENAI_API_BASE=https://api.your-gateway.example/v1
MODEL_NAME=gpt-4o-mini

Verify the environment loads correctly:

# test_env.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1,
)
print(llm.invoke("Reply with OK if you see this.").content)

Run python test_env.py — you should see OK.

Step 2: Define the document ingestion utility

Financial reports arrive as PDFs, HTML, or raw text. Build a small loader that normalizes everything to plain text so the extractor agent receives consistent input. Keep it dependency-light; pypdf handles most earnings PDFs.

# ingestion.py
from pathlib import Path
from pypdf import PdfReader

def load_text(source: str) -> str:
    """
    Accept a file path or raw text. Return plain text.
    """
    path = Path(source)
    if path.exists() and path.suffix.lower() == ".pdf":
        reader = PdfReader(str(path))
        return "\n".join(page.extract_text() or "" for page in reader.pages)
    return source  # assume already text

Test it with a sample earnings PDF (download any 10-Q or press release):

python -c "from ingestion import load_text; print(load_text('sample_earnings.pdf')[:500])"

Step 3: Create the extractor agent and task

The extractor pulls structured fields from the raw text: revenue, EPS, guidance, segment breakdown, and any non-GAAP adjustments. Define the expected schema first so the agent’s output is parseable.

# schema.py
from pydantic import BaseModel, Field
from typing import Optional, List

class SegmentMetric(BaseModel):
    name: str
    revenue: Optional[float] = None
    yoy_change_pct: Optional[float] = None

class ExtractedFinancials(BaseModel):
    period: str = Field(description="Fiscal period, e.g., 'Q3 FY2024'")
    revenue: Optional[float] = Field(description="Total revenue in millions USD")
    eps_gaap: Optional[float] = Field(description="GAAP EPS")
    eps_non_gaap: Optional[float] = Field(description="Non-GAAP EPS")
    guidance_next_quarter_revenue: Optional[str] = None
    guidance_fy_revenue: Optional[str] = None
    segments: List[SegmentMetric] = []
    key_notes: List[str] = []

Now the agent. Use a low temperature and a system prompt that forces JSON output matching the schema.

# agents.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from schema import ExtractedFinancials
import os, json

def make_llm():
    return ChatOpenAI(
        model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_API_BASE"),
        temperature=0.1,
    )

extractor = Agent(
    role="Financial Data Extractor",
    goal="Extract precise financial metrics from earnings releases as structured JSON",
    backstory=(
        "You are a meticulous equity research associate. "
        "You read earnings press releases and 8-K filings and pull out every "
        "quantitative metric: revenue, EPS (GAAP and non-GAAP), segment revenue, "
        "guidance ranges, and notable one-time items. You never hallucinate numbers. "
        "If a value is not explicitly stated, you omit it."
    ),
    llm=make_llm(),
    allow_delegation=False,
    verbose=True,
)

extract_task = Task(
    description=(
        "Read the provided earnings release text and extract all financial metrics "
        "into the JSON schema. Return ONLY valid JSON matching the schema. "
        "Text:\n{report_text}"
    ),
    expected_output="Valid JSON matching the ExtractedFinancials schema",
    agent=extractor,
    output_json=ExtractedFinancials,
)

Step 4: Create the analyst agent and task

The analyst takes the extracted metrics and adds context: year-over-year growth rates, beat/miss versus consensus, margin trends, and flag items for the writer. This agent needs access to consensus estimates — either passed in as context or fetched from a data vendor. For this example, we pass consensus as a simple dict in the task context.

# agents.py (continued)
from typing import Dict, Any

class AnalystOutput(BaseModel):
    revenue_yoy_pct: Optional[float] = None
    eps_gaap_yoy_pct: Optional[float] = None
    revenue_vs_consensus_pct: Optional[float] = None
    eps_vs_consensus_pct: Optional[float] = None
    gross_margin_trend: Optional[str] = None
    operating_margin_trend: Optional[str] = None
    flags: List[str] = []
    narrative_bullets: List[str] = []

analyst = Agent(
    role="Equity Research Analyst",
    goal="Analyze extracted financials against consensus and historical trends, produce structured insights",
    backstory=(
        "You are a senior analyst covering this sector. You take raw extracted metrics, "
        "compare them to consensus estimates (provided in context), compute growth rates, "
        "assess margin trajectory, and flag anything unusual — one-time charges, "
        "guidance changes, segment inflections. Your output is structured for a writer."
    ),
    llm=make_llm(),
    allow_delegation=False,
    verbose=True,
)

analyze_task = Task(
    description=(
        "You receive extracted financials (JSON) and consensus estimates (JSON) in context. "
        "Compute year-over-year growth for revenue and EPS. Calculate beat/miss percentages "
        "versus consensus. Assess margin trends if enough data exists. List 3-5 flags and "
        "3-5 narrative bullets for the summary memo. Return ONLY valid JSON matching AnalystOutput."
    ),
    expected_output="Valid JSON matching the AnalystOutput schema",
    agent=analyst,
    output_json=AnalystOutput,
    context={"consensus": {}},  # populated at runtime
)

Step 5: Create the writer agent and task

The writer consumes the analyst’s structured insights and produces the final deliverable: a one-page memo with headline, key metrics table, commentary, and flags. Give it a style guide so output is consistent across runs.

# agents.py (continued)
writer = Agent(
    role="Research Memo Writer",
    goal="Produce a concise, professional equity research summary memo",
    backstory=(
        "You write the morning note that portfolio managers read before the market opens. "
        "Style: crisp, scannable, no fluff. Structure: Headline, Key Metrics table, "
        "Commentary (3-4 short paragraphs), Flags (bulleted). Use plain text with markdown tables."
    ),
    llm=make_llm(),
    allow_delegation=False,
    verbose=True,
)

write_task = Task(
    description=(
        "Using the analyst's structured output (JSON) and the original extracted financials (JSON), "
        "write a one-page research memo. Include:\n"
        "1. Headline: Company — Period — Beat/Miss/Inline\n"
        "2. Key Metrics table: Period, Revenue, YoY%, EPS GAAP, EPS Non-GAAP, Consensus Rev, Consensus EPS\n"
        "3. Commentary: 3-4 paragraphs covering top-line, profitability, guidance, segments\n"
        "4. Flags: bulleted list of items requiring attention\n"
        "Return plain text with markdown formatting."
    ),
    expected_output="Formatted research memo in markdown",
    agent=writer,
    context={"extracted": {}, "analyst": {}},  # populated at runtime
)

Step 6: Wire the crew and run the pipeline

CrewAI’s Process.sequential passes each task’s output to the next via context. You need a small orchestration script that loads the document, injects consensus estimates, kicks off the crew, and saves the memo.

# run_crew.py
import json
import os
from pathlib import Path
from crewai import Crew, Process
from ingestion import load_text
from agents import (
    extractor, extract_task,
    analyst, analyze_task,
    writer, write_task,
)

# --- Configuration ---
REPORT_PATH = "sample_earnings.pdf"  # or path to your PDF/text
CONSENSUS = {
    "revenue": 89200,      # in millions
    "eps_gaap": 1.42,
    "eps_non_gaap": 1.55,
}
OUTPUT_PATH = "memo.md"
# ---------------------

def main():
    # 1. Load document
    report_text = load_text(REPORT_PATH)
    if not report_text.strip():
        raise ValueError("Empty report text")

    # 2. Build crew with dynamic context injection
    # We'll mutate task contexts at runtime
    extract_task.context = {"report_text": report_text}
    analyze_task.context = {"consensus": CONSENSUS}
    # writer context will be filled by crew automatically from previous outputs

    crew = Crew(
        agents=[extractor, analyst, writer],
        tasks=[extract_task, analyze_task, write_task],
        process=Process.sequential,
        verbose=True,
    )

    # 3. Execute
    result = crew.kickoff()

    # 4. Save memo
    Path(OUTPUT_PATH).write_text(str(result))
    print(f"\nMemo written to {OUTPUT_PATH}")
    print("--- MEMO PREVIEW ---")
    print(str(result)[:1500])

if __name__ == "__main__":
    main()

Run it:

python run_crew.py

You should see the three agents execute in sequence, with verbose logs showing each agent’s reasoning. The final memo appears in memo.md.

Step 7: Verify success and iterate

Open memo.md and check for:

Headline contains company, period, and beat/miss call
Key Metrics table aligns with extracted numbers and consensus
Commentary references specific segments, margins, guidance
Flags surface actionable items (e.g., “Guidance lowered”, “One-time tax benefit inflated GAAP EPS”)

If any section is weak, adjust the corresponding agent’s backstory or task description. Common fixes:

Extractor misses a segment → add “Pay special attention to segment reporting tables” to extractor backstory
Analyst computes wrong YoY → provide prior-period numbers in context
Writer uses flowery language → tighten the style guide in writer backstory

Adding prior-period context for YoY calculations

The analyst needs prior-period data to compute growth rates. Extend the ingestion step to also load the year-ago release, or pass a small dict of historicals in analyze_task.context:

HISTORICALS = {
    "revenue_prior_yoy": 84500,
    "eps_gaap_prior_yoy": 1.31,
}
analyze_task.context = {"consensus": CONSENSUS, "historicals": HISTORICALS}

Update the analyst task description to reference historicals.

Step 8: Production hardening notes

This pipeline works for ad-hoc analysis. To run it repeatedly in production, consider:

Idempotency and caching — Hash the input PDF and skip re-extraction if the hash matches a stored result. Store extracted JSON alongside the source.

Structured logging — Replace verbose=True with a custom callback that emits JSON logs per agent step. This lets you trace latency and token usage per agent.

Error handling — Wrap crew.kickoff() in a retry loop with exponential backoff. Validate each agent’s JSON output against its Pydantic model before passing downstream; if validation fails, re-prompt the same agent once with the validation error.

Model routing — If you use a gateway that honors client routing directives, you can send the extractor to a cheaper model (e.g., gpt-4o-mini) and the writer to a higher-quality model (gpt-4o) by constructing separate ChatOpenAI instances per agent.

Evaluation — Build a small golden set of 10-20 earnings releases with human-written memos. Score each run on metric extraction accuracy (F1 on field-level), beat/miss classification accuracy, and memo quality (LLM-as-judge or human review). Track regression over model upgrades.

Step 9: Extending the pattern

The three-agent pattern — extract, analyze, write — generalizes to other document types:

Domain	Extractor schema	Analyst context	Writer output
M&A announcements	Deal terms, valuation multiples, synergies	Precedent comps, accretion/dilution	Deal memo
Credit agreements	Covenants, baskets, maturity wall	Peer leverage, rating agency criteria	Credit review
ESG reports	Scope 1/2/3 emissions, targets, frameworks	Sector benchmarks, regulatory deadlines	ESG summary

Swap the Pydantic models, backstories, and task descriptions. The orchestration skeleton stays identical.

You now have a working crewai financial report summarization example that you can drop into a scheduled job, wrap in an API, or extend with additional agents (e.g., a “risk flagger” that runs in parallel with the analyst). The key architectural decision is keeping each agent single-purpose and passing structured JSON between them — this makes the system debuggable, testable, and model-agnostic.