Earnings call analysis with LlamaIndex and RAG

Earnings call analysis with LlamaIndex and RAG gives you a repeatable pipeline for extracting structured insights from unstructured transcripts. This tutorial walks through building a system that ingests earnings calls, indexes them for retrieval, and answers quantitative questions like “What was the revenue guidance for Q3?” or “How did management characterize margin pressure?” You’ll end up with a query engine that cites sources and a structured extractor that populates a financial schema.

Prerequisites

• Python 3.10+
• An OpenAI API key (or compatible endpoint) for embeddings and LLM calls
• Basic familiarity with LlamaIndex concepts: documents, indices, query engines

Install dependencies:

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
    llama-index-readers-file pypdf python-dotenv pandas

Create a .env file:

OPENAI_API_KEY=sk-...
# Optional: if you route through a gateway that speaks OpenAI-compatible API
# OPENAI_API_BASE=https://api.n4n.ai/v1

Project structure

earnings_rag/
├── data/
│   └── transcripts/          # PDF or TXT earnings call transcripts
├── src/
│   ├── ingest.py             # Load and chunk transcripts
│   ├── index.py              # Build and persist vector index
│   ├── query.py              # Natural language query engine
│   ├── extract.py            # Structured extraction to Pydantic models
│   └── schema.py             # Financial data models
├── storage/                  # Persisted index (gitignored)
└── main.py                   # CLI entry point

Step 1: Define the financial schema

Before indexing, decide what structured output you want. Pydantic models give you validation and make downstream consumption trivial.

# src/schema.py
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum

class Sentiment(str, Enum):
    POSITIVE = "positive"
    NEUTRAL = "neutral"
    NEGATIVE = "negative"

class GuidanceMetric(BaseModel):
    metric: str = Field(description="Metric name, e.g., 'Revenue', 'EPS', 'Operating Margin'")
    period: str = Field(description="Period referenced, e.g., 'Q3 2024', 'FY 2025'")
    value: Optional[str] = Field(default=None, description="Explicit value if stated")
    direction: Optional[str] = Field(default=None, description="e.g., 'raised', 'lowered', 'maintained', 'reaffirmed'")
    context: str = Field(description="Verbatim quote or close paraphrase from transcript")

class KeyTakeaway(BaseModel):
    topic: str = Field(description="Topic area: Revenue, Margins, Guidance, Product, Competition, Macro")
    summary: str = Field(description="One-sentence summary")
    sentiment: Sentiment
    supporting_quote: str = Field(description="Exact quote from transcript")

class EarningsAnalysis(BaseModel):
    company: str
    quarter: str
    guidance: List[GuidanceMetric] = Field(default_factory=list)
    key_takeaways: List[KeyTakeaway] = Field(default_factory=list)
    risks_mentioned: List[str] = Field(default_factory=list)
    management_tone: Sentiment

Step 2: Ingest and chunk transcripts

Earnings calls have structure: prepared remarks then Q&A. Preserve that boundary — it matters for retrieval.

# src/ingest.py
from pathlib import Path
from llama_index.core import Document
from llama_index.readers.file import PDFReader
from llama_index.core.node_parser import SentenceSplitter

TRANSCRIPT_DIR = Path("data/transcripts")
CHUNK_SIZE = 1024
CHUNK_OVERLAP = 128

def load_transcripts() -> list[Document]:
    reader = PDFReader()
    docs = []
    for pdf_path in TRANSCRIPT_DIR.glob("*.pdf"):
        # PDFReader returns one Document per page; we'll merge per file
        pages = reader.load_data(pdf_path)
        full_text = "\n\n".join(p.text for p in pages)
        
        # Heuristic: split prepared remarks vs Q&A
        qa_marker = "question-and-answer"  # varies by transcript provider
        qa_idx = full_text.lower().find(qa_marker)
        if qa_idx > 0:
            prepared = full_text[:qa_idx]
            qa = full_text[qa_idx:]
            docs.append(Document(
                text=prepared,
                metadata={"source": pdf_path.name, "section": "prepared_remarks"}
            ))
            docs.append(Document(
                text=qa,
                metadata={"source": pdf_path.name, "section": "qa"}
            ))
        else:
            docs.append(Document(
                text=full_text,
                metadata={"source": pdf_path.name, "section": "full"}
            ))
    return docs

def chunk_documents(docs: list[Document]) -> list[Document]:
    splitter = SentenceSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        paragraph_separator="\n\n"
    )
    nodes = splitter.get_nodes_from_documents(docs)
    # Convert back to Documents for simpler indexing
    return [Document(text=n.get_content(), metadata=n.metadata) for n in nodes]

if __name__ == "__main__":
    docs = load_transcripts()
    chunks = chunk_documents(docs)
    print(f"Loaded {len(docs)} documents, split into {len(chunks)} chunks")
    # Quick sanity check
    for c in chunks[:3]:
        print(f"  {c.metadata['source']} [{c.metadata['section']}] {len(c.text)} chars")

Expected output:

Loaded 6 documents, split into 142 chunks
  NVDA_Q2_2024.pdf [prepared_remarks] 1024 chars
  NVDA_Q2_2024.pdf [prepared_remarks] 1024 chars
  NVDA_Q2_2024.pdf [qa] 1024 chars

Step 3: Build and persist the vector index

Use a persistent vector store so you don’t re-embed on every run. LlamaIndex’s default SimpleVectorStore writes to disk; swap in Pinecone, Weaviate, or Postgres+pgvector for production scale.

# src/index.py
from pathlib import Path
from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings
from src.ingest import load_transcripts, chunk_documents

PERSIST_DIR = Path("storage")
EMBED_MODEL = "text-embedding-3-small"

def build_index(force_rebuild: bool = False) -> VectorStoreIndex:
    Settings.embed_model = OpenAIEmbedding(model=EMBED_MODEL)
    
    if PERSIST_DIR.exists() and not force_rebuild:
        print("Loading existing index...")
        storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
        return load_index_from_storage(storage_context)
    
    print("Building new index...")
    docs = load_transcripts()
    chunks = chunk_documents(docs)
    
    index = VectorStoreIndex.from_documents(chunks, show_progress=True)
    index.storage_context.persist(persist_dir=PERSIST_DIR)
    return index

def get_retriever(index: VectorStoreIndex, top_k: int = 8):
    return index.as_retriever(similarity_top_k=top_k)

if __name__ == "__main__":
    idx = build_index()
    print(f"Index built with {len(idx.docstore.docs)} nodes")

Run it once to populate storage/:

python -m src.index

Expected output:

Building new index...
[========================================] 142/142
Index built with 142 nodes

Step 4: Natural language query engine

The query engine handles “What did Jensen say about Data Center revenue?” with citations. Configure a response synthesizer that includes source nodes.

# src/query.py
from llama_index.core import VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.core.prompts import PromptTemplate
from src.index import build_index, get_retriever

LLM_MODEL = "gpt-4o-mini"
EMBED_MODEL = "text-embedding-3-small"

QA_PROMPT = PromptTemplate(
    "You are a financial analyst reviewing earnings call transcripts.\n"
    "Answer the question using ONLY the provided context.\n"
    "Cite sources inline like [source: NVDA_Q2_2024.pdf, section: prepared_remarks].\n"
    "If the answer isn't in the context, say you don't know.\n\n"
    "Context:\n{context_str}\n\n"
    "Question: {query_str}\n\n"
    "Answer:"
)

def build_query_engine(top_k: int = 8) -> RetrieverQueryEngine:
    Settings.llm = OpenAI(model=LLM_MODEL, temperature=0)
    Settings.embed_model = OpenAIEmbedding(model=EMBED_MODEL)
    
    index = build_index()
    retriever = get_retriever(index, top_k=top_k)
    
    synthesizer = CompactAndRefine(
        text_qa_template=QA_PROMPT,
        streaming=False
    )
    
    return RetrieverQueryEngine(retriever=retriever, response_synthesizer=synthesizer)

def ask(question: str) -> str:
    engine = build_query_engine()
    response = engine.query(question)
    return str(response)

if __name__ == "__main__":
    import sys
    q = " ".join(sys.argv[1:]) or "What was the revenue guidance for next quarter?"
    print(f"Q: {q}\n")
    print(ask(q))

Test it:

python -m src.query "What was the Data Center revenue growth year-over-year?"

Expected output:

Q: What was the Data Center revenue growth year-over-year?

Data Center revenue grew 154% year-over-year to $26.3 billion [source: NVDA_Q2_2024.pdf, section: prepared_remarks]. 
The company attributed this to strong demand for Hopper and Blackwell architectures [source: NVDA_Q2_2024.pdf, section: qa].

Step 5: Structured extraction with Pydantic

For programmatic consumption — feeding a dashboard, triggering alerts, populating a database — you need typed output. LlamaIndex’s StructuredLLM + PydanticOutputParser handles this.

# src/extract.py
from llama_index.core import VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import TreeSummarize
from llama_index.core.output_parsers import PydanticOutputParser
from llama_index.core.program import LLMTextCompletionProgram
from llama_index.core.prompts import PromptTemplate
from src.index import build_index, get_retriever
from src.schema import EarningsAnalysis, Sentiment

LLM_MODEL = "gpt-4o"
EMBED_MODEL = "text-embedding-3-small"

EXTRACTION_PROMPT = PromptTemplate(
    "You are a financial analyst extracting structured data from an earnings call transcript.\n"
    "Context from the transcript:\n{context_str}\n\n"
    "Extract the following for {company} {quarter}:\n"
    "1. Guidance metrics (revenue, EPS, margins, etc.) with period, value, direction, and verbatim context\n"
    "2. Key takeaways by topic with sentiment and supporting quote\n"
    "3. Risks explicitly mentioned by management\n"
    "4. Overall management tone\n\n"
    "Output ONLY valid JSON matching the schema. No commentary.\n"
    "If a field is not mentioned, use empty array or neutral sentiment."
)

def build_extractor(company: str, quarter: str) -> LLMTextCompletionProgram:
    Settings.llm = OpenAI(model=LLM_MODEL, temperature=0)
    Settings.embed_model = OpenAIEmbedding(model=EMBED_MODEL)
    
    index = build_index()
    retriever = get_retriever(index, top_k=15)  # wider context for extraction
    
    # Retrieve relevant chunks first
    nodes = retriever.retrieve(f"{company} {quarter} earnings call guidance revenue margins risks")
    context = "\n\n---\n\n".join(n.get_content() for n in nodes)
    
    parser = PydanticOutputParser(output_cls=EarningsAnalysis)
    
    program = LLMTextCompletionProgram.from_defaults(
        output_parser=parser,
        prompt=EXTRACTION_PROMPT.partial_format(
            company=company,
            quarter=quarter,
            context_str=context
        ),
        llm=Settings.llm,
        verbose=True
    )
    return program

def extract(company: str, quarter: str) -> EarningsAnalysis:
    program = build_extractor(company, quarter)
    result = program()
    return result

if __name__ == "__main__":
    import sys
    company = sys.argv[1] if len(sys.argv) > 1 else "NVIDIA"
    quarter = sys.argv[2] if len(sys.argv) > 2 else "Q2 2024"
    
    analysis = extract(company, quarter)
    print(analysis.model_dump_json(indent=2))

Run extraction:

python -m src.extract NVIDIA "Q2 2024"

Expected output (truncated):

{
  "company": "NVIDIA",
  "quarter": "Q2 2024",
  "guidance": [
    {
      "metric": "Revenue",
      "period": "Q3 2024",
      "value": "$32.5 billion",
      "direction": "raised",
      "context": "We expect Q3 revenue of $32.5 billion, plus or minus 2%, up from prior guidance of $28 billion."
    },
    {
      "metric": "Gross Margin",
      "period": "Q3 2024",
      "value": "74.4%",
      "direction": "maintained",
      "context": "GAAP gross margin expected to be 74.4%, non-GAAP 75.0%."
    }
  ],
  "key_takeaways": [
    {
      "topic": "Revenue",
      "summary": "Data Center drove 154% YoY growth, exceeding expectations.",
      "sentiment": "positive",
      "supporting_quote": "Data Center revenue was a record $26.3 billion, up 154% year-over-year."
    },
    {
      "topic": "Guidance",
      "summary": "Q3 revenue guidance raised significantly above street estimates.",
      "sentiment": "positive",
      "supporting_quote": "We expect Q3 revenue of $32.5 billion, plus or minus 2%."
    }
  ],
  "risks_mentioned": [
    "Supply chain constraints for Blackwell ramp",
    "China export control restrictions",
    "Customer inventory digestion in Gaming"
  ],
  "management_tone": "positive"
}

Step 6: CLI entry point

Wire it together for interactive use or batch processing.

# main.py
import argparse
import json
from src.query import ask
from src.extract import extract
from src.schema import EarningsAnalysis

def main():
    parser = argparse.ArgumentParser(description="Earnings call RAG pipeline")
    sub = parser.add_subparsers(dest="cmd", required=True)
    
    q = sub.add_parser("ask", help="Ask a natural language question")
    q.add_argument("question", nargs="+")
    
    e = sub.add_parser("extract", help="Structured extraction for a company/quarter")
    e.add_argument("company")
    e.add_argument("quarter")
    e.add_argument("--json", action="store_true", help="Output raw JSON")
    
    args = parser.parse_args()
    
    if args.cmd == "ask":
        question = " ".join(args.question)
        print(ask(question))
    
    elif args.cmd == "extract":
        analysis: EarningsAnalysis = extract(args.company, args.quarter)
        if args.json:
            print(analysis.model_dump_json(indent=2))
        else:
            print(f"\n=== {analysis.company} {analysis.quarter} ===")
            print(f"Management tone: {analysis.management_tone.value}\n")
            
            print("Guidance:")
            for g in analysis.guidance:
                print(f"  {g.metric} ({g.period}): {g.direction or 'N/A'} — {g.context[:120]}...")
            
            print("\nKey takeaways:")
            for k in analysis.key_takeaways:
                print(f"  [{k.topic}] {k.sentiment.value}: {k.summary}")
            
            if analysis.risks_mentioned:
                print("\nRisks:")
                for r in analysis.risks_mentioned:
                    print(f"  - {r}")

if __name__ == "__main__":
    main()

Usage:

# Natural language Q&A
python main.py ask "How did management describe the Blackwell ramp timeline?"

# Structured extraction
python main.py extract NVIDIA "Q2 2024" --json > nvda_q2_2024.json

Production considerations

Retrieval quality

• Hybrid search: Combine dense vectors with BM25 for ticker symbols, metric names, and exact phrases. LlamaIndex supports VectorStoreIndex + KeywordTableIndex via QueryFusionRetriever.
• Reranking: Add a cross-encoder reranker (e.g., sentence-transformers/cross-encoder/ms-marco-MiniLM-L-6-v2) to boost precision at top-k.
• Metadata filtering: Filter by section: prepared_remarks when you only want management’s prepared narrative; include qa for analyst pushback.

Cost control

• Embed once, persist forever. The storage/ directory is portable — copy it across environments.
• Use gpt-4o-mini for Q&A, reserve gpt-4o for structured extraction where schema adherence matters.
• Set top_k=8 for Q&A, top_k=15-20 for extraction (wider context = fewer hallucinations).

Multi-company, multi-quarter

Partition by company/quarter in metadata, then filter at query time:

from llama_index.core.vector_stores import MetadataFilters, ExactMatchFilter

filters = MetadataFilters(filters=[
    ExactMatchFilter(key="company", value="NVDA"),
    ExactMatchFilter(key="quarter", value="2024-Q2")
])
retriever = index.as_retriever(similarity_top_k=8, filters=filters)

Store company and quarter as metadata during ingestion (extract from filename or first page).

Evaluation

Build a small eval set: 20-30 questions with ground-truth answers from the transcripts. Measure:

• Citation accuracy: Does every claim have a supporting source node?
• Hallucination rate: Manual spot-check or LLM-as-judge against ground truth.
• Extraction completeness: Schema coverage vs. known guidance items in the transcript.

# Quick eval harness
eval_questions = [
    ("What was Q3 revenue guidance?", "$32.5B"),
    ("Data Center YoY growth?", "154%"),
    ("Gross margin outlook?", "74.4% GAAP"),
]
for q, expected in eval_questions:
    ans = ask(q)
    # Check expected substring in answer
    print(f"{'✓' if expected in ans else '✗'} {q}")

Observability

Log every query with: question, retrieved node IDs, latency, token counts, model used. This lets you debug retrieval failures and track cost per analysis. If you route through a gateway that forwards provider cache-control hints, you can also capture cache hit rates for repeated extractions.

What’s next

• Incremental ingestion: Watch a transcript drop folder, embed new files, upsert into a managed vector DB (Pinecone, Weaviate, pgvector).
• Time-series comparison: Extend the schema with prior_quarter_comparison fields; run extraction across quarters and diff programmatically.
• Alerting: Schedule daily extraction for watchlist companies; push guidance changes to Slack/Teams via webhook.
• Fine-tuned embeddings: If domain vocabulary (e.g., “Hopper”, “Blackwell”, “NVLink”) hurts retrieval, fine-tune an embedding model on your transcript corpus.

The pipeline above runs locally with zero infrastructure beyond an API key. Swap the vector store, add reranking, and you have a system that scales to thousands of transcripts across hundreds of companies.