Build a Document Question-Answering Tool

By the end of this guide you will have a small command-line tool that answers plain-English questions about your own PDFs and prints the file name and page number behind every answer. It is four short Python files, a single SQLite database, and roughly forty minutes of work. Ask it "what is our refund window for EU orders?" and it finds the paragraph in your policy document rather than guessing from what a model happened to memorise.

This is one guide in AI Document Processing with Python, written for people who can run a Python file but have never built a search system.

What retrieval-augmented generation means here

A language model does not know anything about your handbook. Pasting the whole handbook into every question is slow, expensive, and eventually hits the model's input limit. Retrieval-augmented generation, usually shortened to RAG, solves that by splitting the work in two: a search step that finds the handful of paragraphs most likely to contain the answer, and a writing step where the model reads only those paragraphs and phrases a reply. The search step uses embeddings — lists of numbers that represent meaning, so two passages about refunds land close together even when they share no words. If embeddings are new to you, Group Keywords with Python and Embeddings shows the same numbers doing a simpler job.

That is the whole idea. Everything below is plumbing: cut the documents up, turn each piece into numbers once, keep the numbers somewhere, and compare them against the question at ask-time. If you want the conversational version that remembers previous turns, Connect a Chatbot to Your Docs with RAG builds on exactly this foundation.

Prerequisites

You need Python 3.10 or newer inside a virtual environment — see Create a Python Virtual Environment for AI if you have not made one yet. Install four packages:

pip install "openai>=1.40" "pypdf>=4.2" "numpy>=1.26" "python-dotenv>=1.0"

Put your key in a .env file beside your scripts:

OPENAI_API_KEY=sk-your-key-here

Then keep that file out of version control before you write another line:

echo ".env" >> .gitignore
echo "docs.sqlite" >> .gitignore

The second line matters too — the database will contain the full text of your documents, and those are usually the last thing you want in a shared repository. Finally, make a docs/ folder and drop a few PDFs in it. Two or three real documents beat one toy file, because the interesting failure is retrieving from the wrong document.

Step 1 — Cut each PDF into overlapping chunks

Whole pages are the wrong unit. A page can hold three unrelated topics, which dilutes its embedding until it matches nothing well. Aim for pieces of roughly 180 words with a 40-word overlap so a sentence that straddles a boundary still appears whole in one chunk. Every chunk carries its file name and page number from birth; that metadata is what lets you cite sources later. Save the following as chunker.py.

import re
from pathlib import Path

from pypdf import PdfReader


def read_pdf_pages(path: Path):
    """Yield (page_number, cleaned_text) for every page that has text."""
    reader = PdfReader(str(path))
    for number, page in enumerate(reader.pages, start=1):
        text = (page.extract_text() or "").strip()
        if text:
            yield number, re.sub(r"\s+", " ", text)


def chunk_document(path: Path, words_per_chunk: int = 180, overlap: int = 40):
    chunks = []
    step = words_per_chunk - overlap
    for page_number, text in read_pdf_pages(path):
        words = text.split()
        for start in range(0, max(len(words), 1), step):
            piece = " ".join(words[start:start + words_per_chunk])
            if len(piece.split()) < 20:      # drop headers and page furniture
                continue
            chunks.append({"source": path.name, "page": page_number, "text": piece})
    return chunks


if __name__ == "__main__":
    for chunk in chunk_document(Path("docs/handbook.pdf"))[:3]:
        print(chunk["source"], "p.", chunk["page"], "-", chunk["text"][:80], "...")

Run python chunker.py and read the three sample chunks out loud. If they look like fragments of sentences or like navigation menus, adjust words_per_chunk before you spend money embedding them. The trade-offs of different splitting strategies are covered in Split Long Text into Chunks for AI APIs.

Step 2 — Embed every chunk once and store it in SQLite

Embedding costs money and time, so you do it once per chunk and keep the result. A vector from text-embedding-3-small is a list of floats; numpy can turn that list into compact bytes with .tobytes(), and SQLite stores bytes happily in a BLOB column. No vector database, no server, no extra service to keep running. The diagram below shows both halves of the system — the indexing pass you run when documents change, and the query pass you run for every question.

The script below creates the tables, batches chunks sixty-four at a time so you spend one API call per batch instead of one per chunk, and records a hash of every file it indexes. Save it as index.py, next to chunker.py.

import hashlib
import os
import sqlite3
from pathlib import Path

import numpy as np
from dotenv import load_dotenv
from openai import OpenAI

from chunker import chunk_document

load_dotenv()                                   # .env is in .gitignore
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
EMBED_MODEL = "text-embedding-3-small"
DB_PATH = "docs.sqlite"


def connect(db_path: str = DB_PATH) -> sqlite3.Connection:
    db = sqlite3.connect(db_path)
    db.execute("""CREATE TABLE IF NOT EXISTS chunks (
                      id INTEGER PRIMARY KEY,
                      source TEXT NOT NULL,
                      page INTEGER NOT NULL,
                      text TEXT NOT NULL,
                      vector BLOB NOT NULL)""")
    db.execute("""CREATE TABLE IF NOT EXISTS files (
                      source TEXT PRIMARY KEY,
                      digest TEXT NOT NULL)""")
    return db


def embed(texts: list[str]) -> list[np.ndarray]:
    response = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return [np.array(item.embedding, dtype=np.float32) for item in response.data]


def index_file(db: sqlite3.Connection, path: Path) -> None:
    digest = hashlib.sha256(path.read_bytes()).hexdigest()
    row = db.execute("SELECT digest FROM files WHERE source = ?", (path.name,)).fetchone()
    if row and row[0] == digest:
        print(f"skip     {path.name} (unchanged)")
        return

    db.execute("DELETE FROM chunks WHERE source = ?", (path.name,))
    chunks = chunk_document(path)
    for start in range(0, len(chunks), 64):
        batch = chunks[start:start + 64]
        vectors = embed([c["text"] for c in batch])
        db.executemany(
            "INSERT INTO chunks (source, page, text, vector) VALUES (?, ?, ?, ?)",
            [(c["source"], c["page"], c["text"], v.tobytes())
             for c, v in zip(batch, vectors)],
        )
    db.execute("INSERT OR REPLACE INTO files (source, digest) VALUES (?, ?)",
               (path.name, digest))
    db.commit()
    print(f"indexed  {path.name}: {len(chunks)} chunks")


if __name__ == "__main__":
    db = connect()
    for pdf in sorted(Path("docs").glob("*.pdf")):
        index_file(db, pdf)
    db.close()

Run python index.py. A 60-page handbook yields a few hundred chunks and costs a fraction of a cent on the small embedding model — embeddings are far cheaper than chat calls. If you want the exact figure before you press enter, Estimate OpenAI API Costs with Python shows how to price a run up front.

The batch size of 64 is a deliberate compromise rather than a limit of the API. Larger batches mean fewer round trips, but a single failure loses more work and a very large request is more likely to be throttled. Sixty-four chunks of 180 words is a few thousand words per call, which sails well under any input ceiling and re-runs cheaply if it fails. Note also that the rows are committed once per file, not once per batch: if the process dies halfway through a big PDF, that file simply gets re-indexed from scratch next time instead of leaving half its chunks in the table.

Step 3 — Rank the chunks with numpy

Similarity between two embeddings is measured with cosine similarity: normalise both vectors to length one, multiply them together element by element, and add up the result. The score runs from -1 to 1, and higher means closer in meaning. Because every stored vector sits in one numpy matrix, you compute all the scores in a single multiplication rather than looping in Python. Save this one as retrieve.py.

import os
import sqlite3

import numpy as np
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
EMBED_MODEL = "text-embedding-3-small"


def load_index(db_path: str = "docs.sqlite"):
    db = sqlite3.connect(db_path)
    rows = db.execute("SELECT source, page, text, vector FROM chunks").fetchall()
    db.close()
    if not rows:
        raise SystemExit("Nothing indexed yet - run python index.py first.")
    matrix = np.vstack([np.frombuffer(r[3], dtype=np.float32) for r in rows])
    matrix /= np.linalg.norm(matrix, axis=1, keepdims=True)   # unit length
    return rows, matrix


def search(question: str, rows, matrix, top_k: int = 4):
    reply = client.embeddings.create(model=EMBED_MODEL, input=[question])
    q = np.array(reply.data[0].embedding, dtype=np.float32)
    q /= np.linalg.norm(q)
    scores = matrix @ q                                       # one score per chunk
    best = np.argsort(scores)[::-1][:top_k]
    return [(float(scores[i]), rows[i][0], rows[i][1], rows[i][2]) for i in best]


if __name__ == "__main__":
    rows, matrix = load_index()
    for score, source, page, text in search("what is the refund window?", rows, matrix):
        print(f"{score:.3f}  {source} p.{page}  {text[:70]}...")

Run this on its own before you wire up the model. The printed scores tell you whether retrieval works at all, and that is the part that usually breaks. Scores above roughly 0.4 on the small embedding model normally indicate a genuine topical match; a top score near 0.15 means nothing in your documents is close and the model is about to be handed junk.

Step 4 — Build a grounded prompt and print the sources

Now the prompt. A grounded prompt has three parts in a fixed order, and the order matters: rules first, evidence second, question last. The model reads instructions best when they arrive before the material they apply to, and it treats the final line as the thing to act on. The diagram breaks down exactly what your script sends.

Two details do the heavy lifting. Numbering the passages [1], [2] gives the model a token it can copy into its answer, which is how you get citations without any parsing work. And naming the exact refusal phrase — here, not in the documents — gives you a string you can test for in code instead of trying to detect a polite hedge. This last piece goes in ask.py.

import os
import sys
import textwrap

from dotenv import load_dotenv
from openai import OpenAI

from retrieve import load_index, search

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

SYSTEM = (
    "You answer questions about the user's own documents. "
    "Use only the numbered context passages given to you. "
    "Never use outside knowledge and never guess. "
    "If the passages do not contain the answer, reply exactly: not in the documents. "
    "Cite the passages you used with their numbers, like [1]."
)


def build_prompt(question: str, hits) -> str:
    blocks = [
        f"[{n}] {source} (page {page})\n{text}"
        for n, (_score, source, page, text) in enumerate(hits, start=1)
    ]
    return "Context passages:\n\n" + "\n\n".join(blocks) + f"\n\nQuestion: {question}"


def answer(question: str, rows, matrix, top_k: int = 4):
    hits = search(question, rows, matrix, top_k=top_k)
    completion = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": build_prompt(question, hits)},
        ],
    )
    return completion.choices[0].message.content, hits


if __name__ == "__main__":
    question = " ".join(sys.argv[1:]) or input("Question: ")
    rows, matrix = load_index()
    reply, hits = answer(question, rows, matrix)
    print("\n" + textwrap.fill(reply, 88) + "\n")
    print("Sources:")
    for n, (score, source, page, _text) in enumerate(hits, start=1):
        print(f"  [{n}] {source} page {page}   similarity {score:.2f}")

Ask it something with python ask.py "what is our refund window for EU orders?". You get a short answer, then a list of the four passages it was allowed to read with their similarity scores. Now ask something your documents definitely do not cover — the capital of Peru, say — and confirm you get the refusal phrase back. A tool that refuses correctly is worth far more than one that always says something.

Once that works, add one guard of your own. The similarity scores come back before the model is called, so you can check the best score first and skip the chat request entirely when it is hopeless — something like if hits[0][0] < 0.2: print("not in the documents"). That saves a call on every out-of-scope question and, more importantly, removes the temptation for the model to stitch an answer out of four unrelated paragraphs. Retrieval failures are the root cause of almost every wrong answer these tools give.

Setting temperature=0 keeps replies as literal as the source text allows. If you want the reasoning behind that number, What Is Temperature in an LLM API? covers it properly.

Step 5 — Re-index only what changed

Documents get edited, replaced, and deleted. Re-embedding the whole folder each time is wasteful, and leaving stale chunks in the table is worse — your tool will confidently quote a paragraph that no longer exists. The hash column added in Step 2 already gives you the answer for edits; you only need to handle deletions and wire it into one command, which is what reindex.py below does.

from pathlib import Path

from index import connect, index_file

DOCS = Path("docs")


def prune_missing(db) -> None:
    """Forget files that have been deleted from the docs folder."""
    on_disk = {p.name for p in DOCS.glob("*.pdf")}
    known = {row[0] for row in db.execute("SELECT source FROM files")}
    for gone in known - on_disk:
        db.execute("DELETE FROM chunks WHERE source = ?", (gone,))
        db.execute("DELETE FROM files WHERE source = ?", (gone,))
        print(f"removed  {gone}")
    db.commit()


if __name__ == "__main__":
    db = connect()
    for pdf in sorted(DOCS.glob("*.pdf")):
        index_file(db, pdf)
    prune_missing(db)
    db.close()

Run python reindex.py whenever the folder changes, or put it on a schedule. One caveat worth knowing before you scale: if you ever switch embedding models, delete docs.sqlite and rebuild from scratch. Vectors from different models have different lengths and are not comparable, and mixing them produces silently terrible retrieval.

Settings you will actually tune

Troubleshooting

• sqlite3.OperationalError: no such table: chunks — you ran ask.py before building the index, or from a different folder. Run python index.py in the directory that holds docs/.
• ValueError: buffer size must be a multiple of element size — the table contains vectors from two different embedding models. Delete docs.sqlite and re-index everything with one model.
• openai.RateLimitError on a large first index — you are sending batches faster than your account tier allows. Add a short time.sleep between batches, or follow Fix the 429 Rate-Limit Error in Python to retry properly.
• Every chunk list comes back empty — the PDF is a scan, so extract_text() finds no text layer. Confirm by opening the file and trying to select a word. Scanned pages need optical character recognition before any of this works.
• The model answers, but the sources look unrelated — retrieval failed and the model improvised anyway. Tighten the system message, and check the similarity scores: if the best is under 0.2, print the refusal yourself rather than calling the model at all.

When to use this vs. alternatives

• Versus pasting a whole PDF into a prompt. Pasting is fine for one 10-page document you will ask about once. Beyond that you pay for the entire text on every question and eventually trip the input limit — see Fix the Context-Length-Exceeded Error in Python. Retrieval keeps cost flat as your library grows.
• Versus summarising instead. If you want the gist of one long report rather than answers to specific questions, Summarize Long PDFs with Python and Chunking is the cheaper and more direct route. Use question-answering when you know what you are looking for.
• Versus a hosted vector database. Managed services add filtering, replication and horizontal scale. None of that helps at a few thousand chunks, where SQLite plus numpy is faster to set up and easier to debug. Move when your index outgrows memory, not before.
• Versus structured extraction. If you want the same fields out of every document, questions are the wrong tool. Extract Invoice Data from PDFs with Python and AI gives you a schema instead of prose.

You now have a working private search-and-answer tool: chunk, embed once, rank with numpy, answer only from the evidence, and cite it. The obvious next moves are wrapping answer() behind an HTTP endpoint with Turn a Python AI Script into an API with FastAPI, or sorting your incoming files first with Classify Incoming Documents with Python and AI so each question searches only the relevant shelf. Back to AI Document Processing with Python.

Related guides

• Connect a Chatbot to Your Docs with RAG — turn this one-shot tool into a conversation that remembers.
• Split Long Text into Chunks for AI APIs — chunking strategies that change retrieval quality more than any prompt.
• Count Tokens in Python Before You Send — measure what each grounded prompt really costs.
• Fix the 401 Unauthorized Error in OpenAI Python — the first wall people hit when .env is not loading.
• AI Document Processing with Python — the main guide for this section, from raw PDFs to structured output.