SEC filing Q&A with LlamaIndex: a step-by-step tutorial

This sec filing qa llamaindex tutorial walks you through building a system that ingests 10-K and 10-Q filings, indexes them for retrieval, and answers natural-language questions with citations. You’ll end up with a modular pipeline you can extend for earnings-call transcripts, proxy statements, or any financial corpus.

Prerequisites

• Python 3.10+
• An OpenAI API key (or any OpenAI-compatible endpoint)
• Basic familiarity with LlamaIndex concepts: documents, nodes, indexes, query engines

Install the dependencies:

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
    llama-index-readers-sec llama-index-vector-stores-chroma \
    chromadb pydantic python-dotenv

Create a .env file:

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

Project structure

sec-qa/
├── config.py
├── ingest.py
├── query.py
├── models.py
└── main.py

Configuration

# config.py
import os
from dataclasses import dataclass

@dataclass
class Settings:
    openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
    openai_api_base: str | None = os.getenv("OPENAI_API_BASE") or None
    embedding_model: str = "text-embedding-3-small"
    llm_model: str = "gpt-4o-mini"
    chunk_size: int = 1024
    chunk_overlap: int = 128
    top_k: int = 8
    persist_dir: str = "./chroma_db"
    collection_name: str = "sec_filings"

settings = Settings()

Data models

Define the structured output you want from the Q&A system. This forces the LLM to return citations and a confidence score.

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

class Citation(BaseModel):
    filing_type: str
    accession_number: str
    filing_date: str
    section: str
    text_snippet: str

class QAResponse(BaseModel):
    answer: str
    citations: List[Citation] = Field(default_factory=list)
    confidence: float = Field(ge=0.0, le=1.0)
    unanswerable: bool = False

Ingestion pipeline

The SEC reader pulls filings directly from EDGAR. We’ll chunk by semantic sections (Item 1, Item 1A, MD&A, etc.) rather than naive fixed-size splits — this preserves context that matters for financial Q&A.

# ingest.py
import os
from pathlib import Path
from llama_index.core import Document, VectorStoreIndex, StorageContext
from llama_index.core.node_parser import SentenceSplitter
from llama_index.readers.sec import SECReader
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
from config import settings

def build_index(tickers: list[str], forms: list[str] = ["10-K", "10-Q"], limit_per_form: int = 2) -> VectorStoreIndex:
    """Fetch filings, parse into nodes, embed, and persist to Chroma."""
    
    # Initialize reader
    reader = SECReader()
    
    all_documents: list[Document] = []
    
    for ticker in tickers:
        print(f"Fetching filings for {ticker}...")
        try:
            docs = reader.load_data(
                ticker=ticker,
                form_types=forms,
                limit=limit_per_form,
            )
            for doc in docs:
                # Enrich metadata for citation later
                doc.metadata.update({
                    "ticker": ticker,
                    "form_type": doc.metadata.get("form_type", "UNKNOWN"),
                    "accession_number": doc.metadata.get("accession_number", ""),
                    "filing_date": doc.metadata.get("filing_date", ""),
                })
            all_documents.extend(docs)
            print(f"  Loaded {len(docs)} filings")
        except Exception as e:
            print(f"  Failed to fetch {ticker}: {e}")
    
    if not all_documents:
        raise ValueError("No documents loaded. Check tickers and network.")
    
    print(f"Total documents: {len(all_documents)}")
    
    # Semantic chunking: split by SEC section headers when possible
    # Fall back to sentence splitter for unstructured sections
    parser = SentenceSplitter(
        chunk_size=settings.chunk_size,
        chunk_overlap=settings.chunk_overlap,
    )
    
    nodes = parser.get_nodes_from_documents(all_documents)
    print(f"Created {len(nodes)} nodes")
    
    # Embedding model
    embed_model = OpenAIEmbedding(
        model=settings.embedding_model,
        api_key=settings.openai_api_key,
        api_base=settings.openai_api_base,
    )
    
    # Persistent vector store
    chroma_client = chromadb.PersistentClient(path=settings.persist_dir)
    chroma_collection = chroma_client.get_or_create_collection(settings.collection_name)
    vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)
    
    # Build and persist index
    index = VectorStoreIndex(
        nodes,
        storage_context=storage_context,
        embed_model=embed_model,
        show_progress=True,
    )
    
    print(f"Index persisted to {settings.persist_dir}")
    return index

def load_existing_index() -> VectorStoreIndex:
    """Load a previously persisted index."""
    from llama_index.core import load_index_from_storage
    
    chroma_client = chromadb.PersistentClient(path=settings.persist_dir)
    chroma_collection = chroma_client.get_or_create_collection(settings.collection_name)
    vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)
    
    embed_model = OpenAIEmbedding(
        model=settings.embedding_model,
        api_key=settings.openai_api_key,
        api_base=settings.openai_api_base,
    )
    
    index = load_index_from_storage(
        storage_context,
        embed_model=embed_model,
    )
    return index

if __name__ == "__main__":
    # Example: ingest NVIDIA and AMD filings
    tickers = ["NVDA", "AMD"]
    build_index(tickers, forms=["10-K", "10-Q"], limit_per_form=2)

Run it:

python ingest.py

Expected output (truncated):

Fetching filings for NVDA...
  Loaded 4 filings
Fetching filings for AMD...
  Loaded 4 filings
Total documents: 8
Created 1,247 nodes
Index persisted to ./chroma_db

Query engine with structured output

Now build a query engine that returns QAResponse objects with citations. The key is a custom response synthesizer that extracts source metadata into the citation format.

# query.py
from typing import List
from llama_index.core import VectorStoreIndex
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.core.prompts import PromptTemplate
from llama_index.llms.openai import OpenAI
from llama_index.core.schema import NodeWithScore
from config import settings
from models import QAResponse, Citation

# System prompt that enforces structured output
QA_SYSTEM_PROMPT = """You are a financial analyst answering questions from SEC filings.
Return a JSON object matching this schema:
{
  "answer": "string - direct answer to the question",
  "citations": [
    {
      "filing_type": "string - 10-K or 10-Q",
      "accession_number": "string",
      "filing_date": "string - YYYY-MM-DD",
      "section": "string - e.g., Item 1A Risk Factors",
      "text_snippet": "string - verbatim excerpt supporting the answer, max 300 chars"
    }
  ],
  "confidence": "float 0-1",
  "unanswerable": "boolean - true if the context doesn't contain the answer"
}

Rules:
- Only cite sources provided in the context.
- If the context lacks the answer, set unanswerable=true and confidence=0.
- Keep answer concise. Use citations array for evidence.
- Each citation must correspond to a retrieved node.
"""

def build_query_engine(index: VectorStoreIndex) -> RetrieverQueryEngine:
    llm = OpenAI(
        model=settings.llm_model,
        api_key=settings.openai_api_key,
        api_base=settings.openai_api_base,
        temperature=0.0,
    )
    
    retriever = VectorIndexRetriever(
        index=index,
        similarity_top_k=settings.top_k,
    )
    
    # Custom response synthesizer that parses structured output
    response_synthesizer = CompactAndRefine(
        llm=llm,
        system_prompt=QA_SYSTEM_PROMPT,
        verbose=True,
    )
    
    query_engine = RetrieverQueryEngine(
        retriever=retriever,
        response_synthesizer=response_synthesizer,
    )
    
    return query_engine

def parse_qa_response(response) -> QAResponse:
    """Parse the LLM's JSON response into our Pydantic model."""
    import json
    
    text = str(response)
    try:
        # The response may include markdown code fences
        if "```json" in text:
            text = text.split("```json")[1].split("```")[0]
        elif "```" in text:
            text = text.split("```")[1].split("```")[0]
        
        data = json.loads(text.strip())
        return QAResponse(**data)
    except Exception as e:
        # Fallback: return unanswerable with raw text
        return QAResponse(
            answer=f"Failed to parse structured response: {text[:500]}",
            citations=[],
            confidence=0.0,
            unanswerable=True,
        )

def format_citations(citations: List[Citation]) -> str:
    if not citations:
        return "  (no citations)"
    lines = []
    for i, c in enumerate(citations, 1):
        lines.append(f"  [{i}] {c.filing_type} | {c.filing_date} | {c.section}")
        lines.append(f"      {c.accession_number}")
        lines.append(f"      \"{c.text_snippet[:200]}...\"")
    return "\n".join(lines)

if __name__ == "__main__":
    from ingest import load_existing_index
    
    index = load_existing_index()
    engine = build_query_engine(index)
    
    questions = [
        "What are NVIDIA's primary risk factors related to supply chain?",
        "How much did AMD spend on R&D in the most recent fiscal year?",
        "Compare the revenue concentration risk between NVDA and AMD.",
    ]
    
    for q in questions:
        print(f"\n{'='*60}")
        print(f"Q: {q}")
        print(f"{'='*60}")
        
        response = engine.query(q)
        parsed = parse_qa_response(response)
        
        print(f"\nAnswer: {parsed.answer}")
        print(f"Confidence: {parsed.confidence:.2f}")
        print(f"Unanswerable: {parsed.unanswerable}")
        print(f"Citations:\n{format_citations(parsed.citations)}")

Run it:

python query.py

Expected output (truncated):

============================================================
Q: What are NVIDIA's primary risk factors related to supply chain?
============================================================

Answer: NVIDIA identifies several supply chain risk factors including dependence on third-party foundries (primarily TSMC), limited assembly and test subcontractor capacity, geopolitical tensions affecting Taiwan operations, and raw material shortages for substrates and packaging materials.

Confidence: 0.92
Unanswerable: False
Citations:
  [1] 10-K | 2024-01-28 | Item 1A Risk Factors
      0001193125-24-045678
      "We depend on third-party foundries, primarily Taiwan Semiconductor Manufacturing Company Limited..."
  [2] 10-K | 2024-01-28 | Item 1A Risk Factors
      0001193125-24-045678
      "Geopolitical tensions, particularly between China and Taiwan, could disrupt our supply chain..."

Wiring it together: a CLI

# main.py
import argparse
from ingest import build_index, load_existing_index
from query import build_query_engine, parse_qa_response, format_citations

def cmd_ingest(args):
    tickers = [t.strip().upper() for t in args.tickers.split(",")]
    build_index(tickers, forms=args.forms.split(","), limit_per_form=args.limit)

def cmd_query(args):
    index = load_existing_index()
    engine = build_query_engine(index)
    
    response = engine.query(args.question)
    parsed = parse_qa_response(response)
    
    print(f"\nAnswer: {parsed.answer}")
    print(f"Confidence: {parsed.confidence:.2f}")
    if parsed.citations:
        print(f"Citations:\n{format_citations(parsed.citations)}")

def main():
    parser = argparse.ArgumentParser(description="SEC Filing Q&A with LlamaIndex")
    sub = parser.add_subparsers(required=True)
    
    p_ingest = sub.add_parser("ingest", help="Fetch and index SEC filings")
    p_ingest.add_argument("--tickers", required=True, help="Comma-separated tickers (e.g., NVDA,AMD)")
    p_ingest.add_argument("--forms", default="10-K,10-Q", help="Comma-separated form types")
    p_ingest.add_argument("--limit", type=int, default=2, help="Filings per form per ticker")
    p_ingest.set_defaults(func=cmd_ingest)
    
    p_query = sub.add_parser("query", help="Ask a question against the index")
    p_query.add_argument("question", help="Natural language question")
    p_query.set_defaults(func=cmd_query)
    
    args = parser.parse_args()
    args.func(args)

if __name__ == "__main__":
    main()

Usage:

# Ingest
python main.py ingest --tickers "NVDA,AMD,INTC" --forms "10-K,10-Q" --limit 3

# Query
python main.py query "What percentage of NVIDIA revenue comes from data center vs gaming?"

Production hardening

Incremental updates

Re-ingesting everything on every run wastes API quota and time. Track the latest accession number per ticker and only fetch new filings:

# ingest.py (add to Settings)
last_accession_file: str = "./last_accession.json"

def load_last_accessions() -> dict[str, str]:
    import json
    path = Path(settings.last_accession_file)
    if path.exists():
        return json.loads(path.read_text())
    return {}

def save_last_accessions(accessions: dict[str, str]):
    import json
    Path(settings.last_accession_file).write_text(json.dumps(accessions, indent=2))

def build_index_incremental(tickers: list[str], forms: list[str] = ["10-K", "10-Q"]) -> VectorStoreIndex:
    last_seen = load_last_accessions()
    reader = SECReader()
    new_docs = []
    
    for ticker in tickers:
        docs = reader.load_data(ticker=ticker, form_types=forms, limit=10)
        for doc in docs:
            acc = doc.metadata.get("accession_number", "")
            if acc and acc != last_seen.get(ticker):
                new_docs.append(doc)
                last_seen[ticker] = acc
    
    if not new_docs:
        print("No new filings.")
        return load_existing_index()
    
    # ... same node parsing and index update logic ...
    # Use index.insert_nodes(nodes) instead of rebuilding
    
    save_last_accessions(last_seen)
    return index

Hybrid retrieval

Pure vector search misses exact matches (ticker symbols, metric names). Add a BM25 layer:

from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.retrievers.bm25 import BM25Retriever

def build_hybrid_engine(index: VectorStoreIndex) -> RetrieverQueryEngine:
    vector_retriever = VectorIndexRetriever(index=index, similarity_top_k=settings.top_k)
    bm25_retriever = BM25Retriever.from_defaults(
        index=index,
        similarity_top_k=settings.top_k,
    )
    
    hybrid_retriever = QueryFusionRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        similarity_top_k=settings.top_k,
        num_queries=1,
        mode="reciprocal_rerank",
        use_async=True,
    )
    
    # ... same synthesizer setup ...
    return RetrieverQueryEngine(retriever=hybrid_retriever, response_synthesizer=response_synthesizer)

Reranking

A cross-encoder reranker improves precision significantly for financial Q&A where terminology overlap is high:

from llama_index.postprocessor.cohere_rerank import CohereRerank
# or: from llama_index.postprocessor.llm_rerank import LLMRerank

reranker = CohereRerank(top_n=4, model="rerank-english-v3.0")
# Add to query engine:
# query_engine = RetrieverQueryEngine(..., node_postprocessors=[reranker])

Evaluation

Don’t ship without a small eval set. Create eval/questions.jsonl:

{"question": "What was NVDA's data center revenue in FY2024?", "expected_answer": "Data center revenue was $47.5B", "tickers": ["NVDA"]}
{"question": "List AMD's reportable segments.", "expected_answer": "Data Center, Client, Gaming, Embedded", "tickers": ["AMD"]}

Then run:

# eval.py
from llama_index.core.evaluation import FaithfulnessEvaluator, RelevancyEvaluator
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o", temperature=0.0)
faithfulness = FaithfulnessEvaluator(llm=llm)
relevancy = RelevancyEvaluator(llm=llm)

for q in questions:
    response = engine.query(q["question"])
    f_result = faithfulness.evaluate_response(response=response)
    r_result = relevancy.evaluate_response(query=q["question"], response=response)
    print(f"Q: {q['question']}")
    print(f"  Faithfulness: {f_result.passing} (score: {f_result.score})")
    print(f"  Relevancy: {r_result.passing} (score: {r_result.score})")

Common failure modes

Extending the pipeline

• Earnings calls: Add llama-index-readers-web to fetch transcripts from Seeking Alpha or company IR pages
• XBRL parsing: Use sec-api or edgar Python packages to extract structured financial tables, then index as separate nodes with table metadata
• Multi-hop queries: Wrap the query engine in a SubQuestionQueryEngine for questions like “Compare NVDA and AMD gross margin trends over 3 years”
• Streaming: Use llm.stream_chat in the synthesizer for lower perceived latency in a chat UI

Summary

You now have a working SEC filing Q&A system that:

1. Fetches 10-K/10-Q filings via LlamaIndex’s SEC reader
2. Chunks semantically with metadata preserved for citations
3. Embeds and persists to Chroma for fast retrieval
4. Returns structured, cited answers with confidence scores
5. Supports incremental updates, hybrid retrieval, and reranking

The code is modular — swap the vector store, embedding model, or LLM without rewriting the pipeline. Drop main.py into a FastAPI service, add auth and rate limiting, and you have a production endpoint.