Building a financial statement Q&A bot with LlamaIndex

If you’ve tried to extract specific metrics from a 10-K or earnings transcript using generic RAG, you know the problem: chunking destroys the table structure that financial statements depend on, and vector similarity alone can’t distinguish “revenue” from “net revenue” or “GAAP revenue.” This tutorial builds a financial statement Q&A bot that preserves tabular structure, routes queries to the right retrieval strategy, and cites sources at the line-item level. We’ll use LlamaIndex’s document parsers, structured query engines, and a small reranking step to get answers you can actually trust.

Prerequisites

You need Python 3.10+ and an OpenAI-compatible API key. The examples use gpt-4o-mini for generation and text-embedding-3-small for embeddings, but any OpenAI-compatible endpoint works — including n4n.ai if you want automatic fallback across providers.

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
    llama-index-readers-file pypdf pdfplumber pandas openpyxl \
    rank-bm25 sentence-transformers

Set your API key and base URL:

export OPENAI_API_KEY="sk-..."
export OPENAI_API_BASE="https://api.openai.com/v1"  # or your gateway endpoint

Project structure

financial-qa/
├── data/                    # Drop PDFs/Excel files here
├── src/
│   ├── ingest.py           # Document parsing and indexing
│   ├── query_engine.py     # Multi-strategy query routing
│   └── cli.py              # Interactive REPL
├── storage/                # Persisted indexes (git-ignored)
└── requirements.txt

Create the directories:

mkdir -p financial-qa/data financial-qa/storage financial-qa/src

Step 1: Parse financial documents with structure preservation

Financial statements live in tables. Standard PDF chunkers split rows across chunks, losing the column headers that give cells meaning. LlamaIndex’s PDFTableExtractor (backed by pdfplumber) extracts tables as structured data, then we convert each table to a markdown representation that preserves row/column relationships.

Create src/ingest.py:

"""Document ingestion: parse PDFs, extract tables, build indexes."""
import os
from pathlib import Path
from typing import List

from llama_index.core import (
    Document,
    SimpleDirectoryReader,
    StorageContext,
    VectorStoreIndex,
    Settings,
)
from llama_index.core.node_parser import MarkdownElementNodeParser
from llama_index.core.extractors import TitleExtractor
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.readers.file import PDFReader

# Configure global settings
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

DATA_DIR = Path(__file__).parent.parent / "data"
STORAGE_DIR = Path(__file__).parent.parent / "storage"


def load_documents() -> List[Document]:
    """Load PDFs from data directory with table extraction."""
    reader = PDFReader()
    docs = reader.load_data(file=DATA_DIR / "sample_10k.pdf")  # single file for demo
    # For multiple files: docs = SimpleDirectoryReader(DATA_DIR).load_data()
    return docs


def parse_tables_to_markdown(docs: List[Document]) -> List[Document]:
    """Convert extracted tables to markdown-preserving Documents."""
    parser = MarkdownElementNodeParser(
        llm=Settings.llm,
        num_workers=4,
    )
    nodes = parser.get_nodes_from_documents(docs)
    # Filter to table-derived nodes (they have 'table' in metadata)
    table_nodes = [n for n in nodes if n.metadata.get("type") == "table"]
    text_nodes = [n for n in nodes if n.metadata.get("type") != "table"]
    
    print(f"Extracted {len(table_nodes)} table nodes, {len(text_nodes)} text nodes")
    return table_nodes + text_nodes


def build_indexes(nodes: List[Document]) -> VectorStoreIndex:
    """Build and persist vector index."""
    STORAGE_DIR.mkdir(parents=True, exist_ok=True)
    
    storage_context = StorageContext.from_defaults()
    index = VectorStoreIndex(nodes, storage_context=storage_context, show_progress=True)
    storage_context.persist(persist_dir=STORAGE_DIR)
    return index


if __name__ == "__main__":
    docs = load_documents()
    nodes = parse_tables_to_markdown(docs)
    index = build_indexes(nodes)
    print(f"Index built and persisted to {STORAGE_DIR}")

Run it with a sample 10-K PDF in data/:

python -m src.ingest

Expected output:

Extracted 47 table nodes, 23 text nodes
Index built and persisted to /path/to/financial-qa/storage

The MarkdownElementNodeParser uses an LLM to convert each extracted table into a clean markdown table, preserving headers, units, and footnotes. Each table becomes a separate node with metadata including the original page number and table index.

Step 2: Build a hybrid query engine with structured routing

Financial questions fall into distinct categories:

• Point lookups: “What was Q3 2024 revenue?” → needs exact table cell retrieval
• Trend questions: “How has gross margin changed over 3 years?” → needs multi-period table assembly
• Narrative questions: “What risks did management highlight?” → needs text search

A single vector index handles none of these well. We’ll build a router that classifies the query and dispatches to the right tool.

Create src/query_engine.py:

"""Multi-strategy query engine for financial statements."""
from typing import Optional
from enum import Enum

from llama_index.core import (
    VectorStoreIndex,
    StorageContext,
    load_index_from_storage,
    Settings,
    QueryBundle,
)
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.tools import QueryEngineTool
from llama_index.core.response_synthesizers import TreeSummarize
from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

from .ingest import STORAGE_DIR

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")


class QueryType(Enum):
    POINT_LOOKUP = "point_lookup"
    TREND_ANALYSIS = "trend_analysis"
    NARRATIVE = "narrative"


def load_index() -> VectorStoreIndex:
    storage_context = StorageContext.from_defaults(persist_dir=STORAGE_DIR)
    return load_index_from_storage(storage_context)


def build_point_lookup_engine(index: VectorStoreIndex):
    """Exact-match retrieval for specific line items."""
    retriever = index.as_retriever(similarity_top_k=8)
    # Rerank with cross-encoder for precise cell matching
    from llama_index.postprocessor.cohere_rerank import CohereRerank
    # If no Cohere key, fall back to similarity threshold
    try:
        reranker = CohereRerank(top_n=3, model="rerank-english-v3.0")
        postprocessors = [reranker]
    except Exception:
        postprocessors = [SimilarityPostprocessor(similarity_cutoff=0.72)]
    
    return index.as_query_engine(
        retriever=retriever,
        node_postprocessors=postprocessors,
        response_mode="compact",
        system_prompt=(
            "You are a financial analyst. Answer ONLY from the provided context. "
            "Cite the exact table, row, column, and value. "
            "If the value is not found, say 'Not found in provided documents.'"
        ),
    )


def build_trend_engine(index: VectorStoreIndex):
    """Retrieve multiple periods for trend analysis."""
    retriever = index.as_retriever(similarity_top_k=12)
    return index.as_query_engine(
        retriever=retriever,
        response_mode="tree_summarize",
        system_prompt=(
            "You are a financial analyst. Synthesize data across multiple time periods. "
            "Present trends in a markdown table with columns: Period, Metric, Value, YoY Change. "
            "Only use data explicitly present in the context."
        ),
    )


def build_narrative_engine(index: VectorStoreIndex):
    """Text-heavy retrieval for qualitative sections."""
    retriever = index.as_retriever(similarity_top_k=6)
    return index.as_query_engine(
        retriever=retriever,
        response_mode="compact",
        system_prompt=(
            "You are a financial analyst. Summarize qualitative disclosures. "
            "Quote relevant passages with section references. "
            "Distinguish between management statements and auditor opinions."
        ),
    )


def build_router_engine() -> RouterQueryEngine:
    index = load_index()
    
    point_engine = build_point_lookup_engine(index)
    trend_engine = build_trend_engine(index)
    narrative_engine = build_narrative_engine(index)
    
    tools = [
        QueryEngineTool.from_defaults(
            query_engine=point_engine,
            description=(
                "Use for exact value lookups: specific line items, dates, amounts, "
                "ratios, or single-period metrics. Examples: 'What was Q3 2024 revenue?', "
                "'Total assets at Dec 31 2023', 'Current ratio for FY2024'."
            ),
            name="point_lookup",
        ),
        QueryEngineTool.from_defaults(
            query_engine=trend_engine,
            description=(
                "Use for multi-period comparisons, trends, growth rates, or changes over time. "
                "Examples: 'Revenue trend over 3 years', 'How has operating margin changed?', "
                "'Year-over-year growth for each quarter 2023'."
            ),
            name="trend_analysis",
        ),
        QueryEngineTool.from_defaults(
            query_engine=narrative_engine,
            description=(
                "Use for qualitative questions: risk factors, management discussion, "
                "accounting policies, auditor opinions, legal proceedings. "
                "Examples: 'What are the key risk factors?', 'Management outlook for 2025', "
                "'Accounting policy for revenue recognition'."
            ),
            name="narrative",
        ),
    ]
    
    selector = LLMSingleSelector.from_defaults(llm=Settings.llm)
    return RouterQueryEngine(selector=selector, query_engine_tools=tools, verbose=True)


def query(question: str) -> str:
    """Convenience function for single queries."""
    engine = build_router_engine()
    response = engine.query(question)
    return str(response)


if __name__ == "__main__":
    # Quick smoke test
    engine = build_router_engine()
    test_queries = [
        "What was total revenue for the year ended December 31, 2023?",
        "Show me the gross margin trend for the last 3 years",
        "What risk factors did management disclose?",
    ]
    for q in test_queries:
        print(f"\n{'='*60}\nQ: {q}\n{'='*60}")
        print(engine.query(q))

Run a quick test:

python -c "from src.query_engine import query; print(query('What was total revenue for the year ended December 31, 2023?'))"

Expected output (router selection visible with verbose=True):

Selected query engine: point_lookup
Total revenue for the year ended December 31, 2023 was $383.29 billion 
(Source: Consolidated Statements of Operations, row "Total revenue", column "2023")

The router uses an LLM to classify the query and pick the right tool. The point_lookup engine uses a reranker (Cohere if available, otherwise similarity threshold) to pull the exact table row. The trend_engine retrieves more candidates and uses tree_summarize to assemble a multi-period view. The narrative_engine keeps it simple for text-heavy sections.

Step 3: Add structured output for programmatic use

LLMs hallucinate numbers. For any downstream system — dashboards, alerts, model inputs — you need structured, validated output. LlamaIndex’s StructuredOutput (via Pydantic) forces the model to return typed fields or fail.

Add to src/query_engine.py:

from pydantic import BaseModel, Field
from typing import List, Optional
from llama_index.core.output_parsers import PydanticOutputParser
from llama_index.core.query_engine import StructuredQueryEngine


class FinancialMetric(BaseModel):
    metric_name: str = Field(description="Standardized metric name, e.g., 'Revenue', 'Gross Margin'")
    period: str = Field(description="Reporting period, e.g., 'Q3 2024', 'FY2023'")
    value: float = Field(description="Numeric value in reporting currency")
    unit: str = Field(description="Unit: 'USD millions', 'USD billions', '%', 'ratio'")
    source_table: str = Field(description="Table title or section name")
    source_row: str = Field(description="Row label in the source table")
    source_column: str = Field(description="Column header in the source table")


class FinancialMetricsResponse(BaseModel):
    metrics: List[FinancialMetric] = Field(description="Extracted financial metrics")
    confidence: float = Field(description="Confidence 0-1 that all metrics are accurate")
    notes: Optional[str] = Field(default=None, description="Caveats or missing data notes")


def build_structured_engine(index: VectorStoreIndex) -> StructuredQueryEngine:
    output_parser = PydanticOutputParser(FinancialMetricsResponse)
    
    base_engine = index.as_query_engine(
        similarity_top_k=10,
        response_mode="compact",
        system_prompt=(
            "Extract financial metrics as structured data. "
            "For each metric, identify the exact table, row, column, value, unit, and period. "
            "Only extract metrics explicitly stated in the document. "
            "If a metric appears in multiple tables, prefer the consolidated statement."
        ),
    )
    
    return StructuredQueryEngine(
        query_engine=base_engine,
        output_parser=output_parser,
    )

Usage:

from src.query_engine import build_structured_engine, load_index

index = load_index()
structured_engine = build_structured_engine(index)
response = structured_engine.query("Extract revenue, gross profit, and operating income for FY2023 and FY2024")

print(response.metrics[0].model_dump_json(indent=2))

Output:

{
  "metric_name": "Revenue",
  "period": "FY2024",
  "value: 383290,
  "unit": "USD millions",
  "source_table": "Consolidated Statements of Operations",
  "source_row": "Total revenue",
  "source_column": "2024"
}

This gives you typed, validated data you can write to a database or feed to a calculation engine without regex parsing.

Step 4: Interactive CLI with session memory

For ad-hoc analysis, a REPL with conversation history beats one-off scripts. LlamaIndex’s ChatMemoryBuffer maintains context across turns.

Create src/cli.py:

"""Interactive CLI for financial statement Q&A."""
import sys
from pathlib import Path

from llama_index.core import Settings
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

from .query_engine import build_router_engine

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")


def main():
    print("Financial Statement Q&A Bot")
    print("Type 'exit' to quit, 'clear' to reset memory\n")
    
    engine = build_router_engine()
    memory = ChatMemoryBuffer.from_defaults(token_limit=3000)
    
    while True:
        try:
            question = input("💰 ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nGoodbye.")
            break
            
        if not question:
            continue
        if question.lower() in ("exit", "quit"):
            break
        if question.lower() == "clear":
            memory.reset()
            print("Memory cleared.\n")
            continue
        
        # Add user message to memory
        memory.put(question)
        
        # Build context from memory
        context = "\n".join([msg.content for msg in memory.get()])
        full_query = f"Conversation context:\n{context}\n\nCurrent question: {question}"
        
        print("\nThinking...")
        response = engine.query(full_query)
        
        print(f"\n{response}\n")
        
        # Store assistant response
        memory.put(str(response))


if __name__ == "__main__":
    main()

Run it:

python -m src.cli

Session example:

Financial Statement Q&A Bot
Type 'exit' to quit, 'clear' to reset memory

💰 What was Apple's revenue in FY2023?

Thinking...
Selected query engine: point_lookup
Total revenue for FY2023 was $383.29 billion (Consolidated Statements of Operations, row "Total revenue", column "2023")

💰 And gross margin?

Thinking...
Selected query engine: point_lookup
Gross margin for FY2023 was 44.1% (Consolidated Statements of Operations, row "Gross margin", column "2023")

💰 Show me the 3-year trend for both

Thinking...
Selected query engine: trend_analysis
| Period | Metric | Value | YoY Change |
|--------|--------|-------|------------|
| FY2021 | Revenue | $365.82B | — |
| FY2022 | Revenue | $394.33B | +7.8% |
| FY2023 | Revenue | $383.29B | -2.8% |
| FY2021 | Gross Margin | 41.8% | — |
| FY2022 | Gross Margin | 43.3% | +1.5pp |
| FY2023 | Gross Margin | 44.1% | +0.8pp |

💰 clear
Memory cleared.

💰 exit
Goodbye.

The router correctly switches strategies: point lookups for the first two, trend analysis for the third. Memory lets you refer to “both” in the third question.

Step 5: Handle multi-document collections (10-K + 10-Q + earnings)

Real workloads span multiple filings. LlamaIndex’s DocumentSummaryIndex builds per-document summaries, then routes queries to the right filing before drilling down.

Add to src/ingest.py:

def build_multi_doc_index(docs: List[Document]) -> DocumentSummaryIndex:
    """Build a document-summary index for multi-filing collections."""
    from llama_index.core import DocumentSummaryIndex
    from llama_index.core.response_synthesizers import TreeSummarize
    
    # First, parse each document into nodes
    parser = MarkdownElementNodeParser(llm=Settings.llm, num_workers=4)
    all_nodes = []
    for doc in docs:
        nodes = parser.get_nodes_from_documents([doc])
        # Tag nodes with filing metadata
        for n in nodes:
            n.metadata["filing_type"] = doc.metadata.get("filing_type", "unknown")
            n.metadata["filing_date"] = doc.metadata.get("filing_date", "unknown")
        all_nodes.extend(nodes)
    
    # Build summary index
    summary_index = DocumentSummaryIndex.from_documents(
        docs,
        llm=Settings.llm,
        response_synthesizer=TreeSummarize(),
        show_progress=True,
    )
    return summary_index

Then in query_engine.py, add a filing router:

def build_filing_router_engine(index: DocumentSummaryIndex):
    """Route to specific filing, then use the appropriate query engine."""
    from llama_index.core.tools import QueryEngineTool
    from llama_index.core.query_engine import RouterQueryEngine
    from llama_index.core.selectors import LLMMultiSelector
    
    # Get unique filings
    filings = set()
    for doc_id, doc in index.docstore.docs.items():
        filing_type = doc.metadata.get("filing_type")
        filing_date = doc.metadata.get("filing_date")
        if filing_type and filing_date:
            filings.add((filing_type, filing_date))
    
    tools = []
    for filing_type, filing_date in sorted(filings):
        # Create a filtered retriever for this filing
        filing_engine = index.as_query_engine(
            filters={"filing_type": filing_type, "filing_date": filing_date},
            response_mode="compact",
        )
        tools.append(QueryEngineTool.from_defaults(
            query_engine=filing_engine,
            description=f"{filing_type} filed {filing_date}",
            name=f"{filing_type}_{filing_date}",
        ))
    
    selector = LLMMultiSelector.from_defaults(llm=Settings.llm, max_outputs=3)
    return RouterQueryEngine(selector=selector, query_engine_tools=tools)

This lets you ask “Compare revenue recognition policy between the 2023 10-K and Q3 2024 10-Q” and get answers from both documents.

Production considerations

Citation granularity: The markdown table nodes include page_number and table_index in metadata. Extend the response synthesizer to emit these as citations:

from llama_index.core.response_synthesizers import ResponseMode

def cite_sources(response):
    for node in response.source_nodes:
        meta = node.metadata
        print(f"  📄 Page {meta.get('page_number')}, Table {meta.get('table_index')}: {node.text[:100]}...")

Cost control: Table extraction uses an LLM call per table. For 100+ page filings, cache parsed markdown to disk and only re-parse on document change. The MarkdownElementNodeParser output is deterministic for a given PDF — store the hash.

Evaluation: Build a small golden set (20-30 questions with verified answers from the source tables). Run nightly against your index to catch regressions when you swap embedding models or chunking strategies.

# eval/golden_set.jsonl
{"question": "What was FY2023 revenue?", "answer": "$383.29 billion", "source": "Consolidated Statements of Operations, row Total revenue, col 2023"}
{"question": "Gross margin FY2022?", "answer": "43.3%", "source": "Consolidated Statements of Operations, row Gross margin, col 2022"}

Scaling: For multi-company collections, add a company-level router above the filing router. Use metadata filtering at query time rather than separate indexes — it keeps storage simpler and enables cross-company queries (“Which companies disclosed cyber risk in 2024?”).

What’s next

• Add a calculation engine that computes derived metrics (FCF, ROIC, leverage ratios) from extracted line items rather than trusting reported non-GAAP figures
• Integrate XBRL parsing for structured tag access — the SEC’s inline XBRL gives you standardized tags (us-gaap:RevenueFromContractWithCustomer) that survive formatting changes
• Build a change detection pipeline that diffs consecutive filings and flags modified line items, new risk factors, or restated numbers
• Wire the structured output to a time-series database (TimescaleDB, InfluxDB) for charting and alerting

The core pattern — preserve table structure, route by query type, emit structured output — applies to any domain with dense tabular data: clinical trial results, supply chain manifests, regulatory filings. The financial statement bot is just the first application.