Build RAG with Tables: Extract Data from PDFs and Excel 2026

# Create project with uv
uv init rag-tables && cd rag-tables
uv add langchain langchain-community langchain-openai \
       chromadb camelot-py[cv] pdfplumber openpyxl \
       pandas tiktoken python-dotenv

OPENAI_API_KEY=sk-...

# table_extractor.py
import camelot
import pdfplumber
import pandas as pd
from pathlib import Path


def extract_tables_camelot(pdf_path: str) -> list[pd.DataFrame]:
    """
    Lattice mode: bordered tables with visible grid lines.
    accuracy > 90 filters out partial/broken extractions.
    """
    tables = camelot.read_pdf(
        pdf_path,
        flavor="lattice",   # use "stream" for borderless tables
        pages="all",
    )
    return [t.df for t in tables if t.accuracy > 90]


def extract_tables_pdfplumber(pdf_path: str) -> list[pd.DataFrame]:
    """
    Stream mode: whitespace-aligned tables, scanned-adjacent PDFs.
    Falls back to pdfplumber when camelot accuracy is low.
    """
    dfs = []
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            for table in page.extract_tables():
                df = pd.DataFrame(table[1:], columns=table[0])
                dfs.append(df)
    return dfs


def extract_pdf_tables(pdf_path: str) -> list[pd.DataFrame]:
    """Try camelot first; fall back to pdfplumber."""
    try:
        tables = extract_tables_camelot(pdf_path)
        if tables:
            return tables
    except Exception:
        pass
    return extract_tables_pdfplumber(pdf_path)

# excel_extractor.py
import openpyxl
import pandas as pd


def extract_excel_tables(xlsx_path: str) -> list[dict]:
    """
    Returns list of {sheet_name, table_name, df} dicts.
    Uses openpyxl named tables first; falls back to full sheet data.
    """
    wb = openpyxl.load_workbook(xlsx_path, data_only=True)  # data_only=True reads computed values, not formulas
    results = []

    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]

        if ws.tables:
            # Named Excel tables — best structure preservation
            for table_name, table in ws.tables.items():
                data = ws[table.ref]
                rows = [[cell.value for cell in row] for row in data]
                df = pd.DataFrame(rows[1:], columns=rows[0])
                results.append({
                    "sheet": sheet_name,
                    "table": table_name,
                    "df": df,
                })
        else:
            # Full sheet fallback
            df = pd.read_excel(xlsx_path, sheet_name=sheet_name, engine="openpyxl")
            results.append({
                "sheet": sheet_name,
                "table": sheet_name,
                "df": df,
            })

    return results

# serializer.py
import pandas as pd


def serialize_table(
    df: pd.DataFrame,
    source: str,
    table_label: str,
    max_rows: int = 50,
) -> str:
    """
    Convert a DataFrame into a labeled text block for embedding.
    max_rows=50 keeps chunks under ~512 tokens for most financial tables.
    """
    df = df.fillna("").astype(str)

    # Truncate large tables — embed summary row count for LLM context
    truncated = len(df) > max_rows
    display_df = df.head(max_rows)

    header = f"Source: {source}\nTable: {table_label}\n"
    col_line = "Columns: " + " | ".join(display_df.columns.tolist()) + "\n"
    rows = "\n".join(
        " | ".join(row) for row in display_df.itertuples(index=False, name=None)
    )
    footer = f"\n[Truncated: showing {max_rows} of {len(df)} rows]" if truncated else ""

    return header + col_line + rows + footer

# indexer.py
from langchain_core.documents import Document
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from table_extractor import extract_pdf_tables
from excel_extractor import extract_excel_tables
from serializer import serialize_table
from pathlib import Path
import os
from dotenv import load_dotenv

load_dotenv()


def build_table_index(
    pdf_paths: list[str],
    xlsx_paths: list[str],
    persist_dir: str = "./chroma_tables",
) -> Chroma:
    docs = []

    # PDF tables
    for pdf_path in pdf_paths:
        tables = extract_pdf_tables(pdf_path)
        for i, df in enumerate(tables):
            text = serialize_table(df, source=Path(pdf_path).name, table_label=f"Table {i + 1}")
            docs.append(Document(
                page_content=text,
                metadata={"source": pdf_path, "type": "pdf_table", "table_index": i},
            ))

    # Excel tables
    for xlsx_path in xlsx_paths:
        tables = extract_excel_tables(xlsx_path)
        for t in tables:
            text = serialize_table(t["df"], source=Path(xlsx_path).name, table_label=t["table"])
            docs.append(Document(
                page_content=text,
                metadata={"source": xlsx_path, "type": "excel_table", "sheet": t["sheet"]},
            ))

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")  # $0.02/1M tokens — cheapest option for table text
    vectorstore = Chroma.from_documents(
        documents=docs,
        embedding=embeddings,
        persist_directory=persist_dir,
    )
    return vectorstore

# query.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings


def load_index(persist_dir: str = "./chroma_tables") -> Chroma:
    return Chroma(
        persist_directory=persist_dir,
        embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
    )


TABLE_PROMPT = ChatPromptTemplate.from_messages([
    ("system", (
        "You are a data analyst. Answer questions using ONLY the table data provided. "
        "Quote exact values from the table. If a value is missing, say so explicitly."
    )),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])


def ask(question: str, vectorstore: Chroma) -> str:
    retriever = vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 4},  # 4 chunks covers most multi-table questions without exceeding context
    )
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # temperature=0 — deterministic for numerical answers

    chain = (
        {"context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
         "question": RunnablePassthrough()}
        | TABLE_PROMPT
        | llm
    )
    return chain.invoke(question).content


if __name__ == "__main__":
    vs = load_index()
    print(ask("What was Q3 2025 revenue?", vs))

uv run python -c "
from indexer import build_table_index
vs = build_table_index(['annual_report_2025.pdf'], ['financials.xlsx'])
print(f'Indexed {vs._collection.count()} table chunks')
"

uv run python query.py