Mini Project: Build an Intelligent Document Q&A App with LangChain

Your company has a 200-page employee handbook. People ask HR the same forty questions about it every week. The obvious idea: put the handbook in front of a model and let people ask.

The obvious implementation pastes the whole thing into the prompt. Price it: 200 pages at roughly 500 words a page is 100,000 words, about 133,000 tokens. At 3 dollars per million input tokens that is 0.40 dollars per question — 400 dollars a day at a thousand questions, to answer things that occupy two paragraphs of the document.

It is also worse at the job than you expect. A single relevant clause competes with 130,000 tokens of irrelevant text, and answers get vaguer as the haystack grows. Add a second document — the expenses policy, the security handbook — and you are past what many models accept in one request at all.

The fix is to stop sending the document and start sending the part of the document that answers this question. Find the three or four passages that matter, put those in the prompt, and answer from them. That is retrieval-augmented generation, and the same request now costs about 0.011 dollars — a 36-fold reduction — while being more accurate, not less, because the model is reading two relevant pages instead of two hundred mostly-irrelevant ones.

You are going to build that: a document question-answering service where you upload PDFs, ask questions in plain English, and get answers with citations back to the source. It is the most common LLM application in industry, and building one end to end exercises everything that makes these systems hard — chunking, embeddings, vector search, prompt construction, an API, a UI, and a container that runs the lot.

What the finished thing does

Hold yourself to these. They separate a demo from something a team will use.

The architecture, and why it splits where it does

INGESTION (once per document)  PDF ──▶ extract text ──▶ split into chunks ──▶ embed each chunk                                   │                    │                                   ▼                    ▼                            chunk store            vector index                          (id → text, page)     (id → 1536 floats)QUERY (once per question)  question ──▶ embed ──▶ search index ──▶ top-k chunk ids                                              │                                              ▼                                      fetch text from chunk store                                              │                                              ▼                              prompt = question + chunks ──▶ model ──▶ answer + citations

The two paths run at completely different times and rates. Ingestion happens once per document and is slow and expensive. Query happens thousands of times and must be fast and cheap. Everything that can be precomputed belongs in ingestion.

Note the two stores — the piece most first attempts get wrong. A vector index stores vectors and returns identifiers and distances; it does not necessarily hand back the original text. Without a separate chunk store, search returns identifiers you cannot turn into prompt text, and you discover it at the last step, after everything else already works.

document-qa/├── app/│   ├── config.py       # one validated settings object│   ├── documents.py    # extract text from files│   ├── chunking.py     # split text into overlapping chunks│   ├── embeddings.py   # text → vectors, batched│   ├── store.py        # vector index + chunk store together│   ├── qa.py           # retrieve, prompt, answer│   └── api.py          # FastAPI├── ui/app.py           # Streamlit, talks to the API over HTTP└── tests/

Setup

pip install fastapi uvicorn[standard] streamlit langchain langchain-openai \            chromadb pypdf python-docx tiktoken pydantic-settings \            python-multipart pytest

1# app/config.py2from pydantic_settings import BaseSettings, SettingsConfigDict3from pydantic import SecretStr, Field45class Settings(BaseSettings):6    model_config = SettingsConfigDict(env_file=".env", extra="ignore")78    openai_api_key: SecretStr9    chat_model: str = "gpt-6-luna"10    embedding_model: str = "text-embedding-3-small"11    chunk_size: int = Field(default=800, ge=200, le=2000)      # tokens12    chunk_overlap: int = Field(default=150, ge=0, le=500)13    top_k: int = Field(default=4, ge=1, le=20)14    max_upload_mb: int = 5015    persist_dir: str = "./storage"1617settings = Settings()

Commit a .env.example with the key names and no values, put .env in .gitignore, and never let a real key into the repository. A committed key is compromised permanently — deleting it in a later commit does nothing, because the history keeps it and automated scanners find public pushes within minutes.

Extracting text

1# app/documents.py2from pypdf import PdfReader3from docx import Document as Docx4from pathlib import Path56def extract(path: str) -> list[tuple[int, str]]:7    """Return [(page_number, text)] — page numbers make citations possible."""8    suffix = Path(path).suffix.lower()910    if suffix == ".pdf":11        reader = PdfReader(path)12        pages = [(i + 1, (p.extract_text() or "").strip())13                 for i, p in enumerate(reader.pages)]14        pages = [(n, t) for n, t in pages if t]15        if not pages:16            raise ValueError(17                "No text found. This PDF is probably a scan — it needs OCR first.")18        return pages1920    if suffix == ".docx":21        return [(1, "\n".join(p.text for p in Docx(path).paragraphs if p.text.strip()))]22    if suffix == ".txt":23        return [(1, Path(path).read_text(encoding="utf-8", errors="replace"))]24    raise ValueError(f"Unsupported file type: {suffix}")

Returning page numbers rather than one blob of text is what makes "handbook.pdf, page 47" possible in a citation — and retrofitting citations once the page structure is gone means reprocessing everything.

The scanned-PDF check is the most common ingestion failure in practice. A scanned document is images of text; extract_text() returns empty strings, chunking produces nothing, the index stays empty, and the app answers every question with "I don't know" while reporting a successful upload. Detect it and say so.

Chunking: the decision that determines answer quality

Chunking is where most of the quality of a document Q&A system is won or lost, and it gets far less attention than prompt wording.

1# app/chunking.py2from langchain_text_splitters import RecursiveCharacterTextSplitter3from app.config import settings45splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(6    encoding_name="cl100k_base",7    chunk_size=settings.chunk_size,8    chunk_overlap=settings.chunk_overlap,9    separators=["\n\n", "\n", ". ", " ", ""],10)1112def chunk_pages(pages: list[tuple[int, str]], doc_name: str) -> list[dict]:13    out = []14    for page_no, text in pages:15        for i, piece in enumerate(splitter.split_text(text)):16            out.append({17                "id": f"{doc_name}:{page_no}:{i}",18                "text": piece,19                "metadata": {"document": doc_name, "page": page_no, "index": i},20            })21    return out

The separator list is doing real work. RecursiveCharacterTextSplitter tries each separator in turn and only falls through to a cruder one when a chunk is still too big — so it prefers to break at paragraph boundaries, then line breaks, then sentence ends, and only splits mid-word as a last resort. Splitting on a fixed character count instead cuts sentences in half, and half a sentence embeds to a vector that means something different from either the sentence or its neighbours.

Now the size trade-off, which is a genuine engineering choice rather than a default to accept:

The overlap exists for a specific failure. Suppose a chunk boundary lands mid-way through the notice-period clause: the first chunk ends with "employees must give notice of" and the second begins "three months, except during probation". Neither chunk answers "how much notice must I give". With a 150-token overlap, the second chunk begins 150 tokens earlier and contains the whole clause. Roughly 15–20% of chunk size is a good starting overlap; the cost is storing and embedding about that much duplicate text.

If your Q&A system gives vague answers, look at the chunks it retrieved before you touch the prompt. Nine times out of ten the model answered the best it could from chunks that did not contain the answer.

Embeddings

An embedding turns text into a list of numbers positioned so that texts about similar things sit close together. "How much holiday do I get?" and "Annual leave entitlement is 25 days" share almost no words, yet their vectors are near neighbours — which is why this beats keyword search for questions asked in a user's own words.

1# app/embeddings.py2from langchain_openai import OpenAIEmbeddings3from app.config import settings45embedder = OpenAIEmbeddings(model=settings.embedding_model,6                            api_key=settings.openai_api_key.get_secret_value())78def embed_documents(texts: list[str], batch_size: int = 100) -> list[list[float]]:9    vectors = []10    for i in range(0, len(texts), batch_size):11        vectors.extend(embedder.embed_documents(texts[i:i + batch_size]))12    return vectors1314embed_query = embedder.embed_query

Batching matters at ingestion scale. Embedding 4,000 chunks one at a time is 4,000 round trips; at 120 ms each that is eight minutes of pure network latency. In batches of 100 it is 40 requests and under a minute.

The cost is worth internalising, because people over-estimate it badly. The handbook is about 133,000 tokens; at 0.02 dollars per million for a small embedding model, indexing all of it costs 0.0027 dollars — under a third of a penny, and 500 such documents cost about 1.33 dollars. Embeddings are effectively free; generation is where your money goes.

Two rules prevent silent corruption. Use the same model for documents and queries — vectors from different models are not comparable, and mixing them makes retrieval essentially random. And if you change embedding model, re-embed everything; the old vectors are meaningless in the new space and nothing will warn you.

The store: vectors and text together

1# app/store.py2import chromadb3from chromadb.config import Settings as ChromaSettings4from app.config import settings5from app.embeddings import embed_documents, embed_query67client = chromadb.PersistentClient(path=settings.persist_dir,8    settings=ChromaSettings(anonymized_telemetry=False))9collection = client.get_or_create_collection("documents",10    metadata={"hnsw:space": "cosine"})1112def add_chunks(chunks: list[dict]) -> int:13    vectors = embed_documents([c["text"] for c in chunks])14    collection.add(15        ids=[c["id"] for c in chunks],16        embeddings=vectors,17        documents=[c["text"] for c in chunks],       # the chunk store18        metadatas=[c["metadata"] for c in chunks],19    )20    return len(chunks)2122def search(question: str, k: int | None = None, document: str | None = None):23    res = collection.query(24        query_embeddings=[embed_query(question)],25        n_results=k or settings.top_k,26        where={"document": document} if document else None,27    )28    return [29        {"text": t, "metadata": m, "distance": d}30        for t, m, d in zip(res["documents"][0], res["metadatas"][0], res["distances"][0])31    ]

Passing documents= alongside embeddings= is what makes Chroma serve as both stores at once, which is why it is a good choice for a project this size. With a bare vector index you would keep a separate table mapping identifier to text — the "chunk store" from the architecture diagram — and the retrieval step would do two lookups instead of one.

hnsw:space: cosine sets the distance metric. Cosine distance compares direction rather than magnitude, which is what text embeddings want; a euclidean default gives subtly worse neighbours in a way that is very hard to notice from outside.

Answering, with citations and an honest "I don't know"

1# app/qa.py2from langchain_core.prompts import ChatPromptTemplate3from langchain_core.output_parsers import StrOutputParser4from langchain_openai import ChatOpenAI5from app.store import search6from app.config import settings78PROMPT = ChatPromptTemplate.from_messages([9    ("system",10     "You answer questions using ONLY the provided context.\n"11     "Rules:\n"12     "- If the context does not contain the answer, reply exactly: "13     "  'I could not find that in the uploaded documents.'\n"14     "- Never use outside knowledge, even if you are confident.\n"15     "- Cite the source of each claim as [document, page N].\n"16     "- Be concise. Quote the document's wording where it matters."),17    ("human", "Context:\n{context}\n\nQuestion: {question}"),18])1920model = ChatOpenAI(model=settings.chat_model,21                   api_key=settings.openai_api_key.get_secret_value())22chain = PROMPT | model | StrOutputParser()2324MAX_DISTANCE = 0.55        # tune against your own documents2526def answer(question: str, document: str | None = None) -> dict:27    hits = search(question, document=document)28    relevant = [h for h in hits if h["distance"] <= MAX_DISTANCE]2930    if not relevant:31        return {"answer": "I could not find that in the uploaded documents.",32                "sources": [], "confidence": "none"}3334    context = "\n\n---\n\n".join(35        f"[{h['metadata']['document']}, page {h['metadata']['page']}]\n{h['text']}"36        for h in relevant)3738    return {39        "answer": chain.invoke({"context": context, "question": question}),40        "sources": [{"document": h["metadata"]["document"],41                     "page": h["metadata"]["page"],42                     "distance": round(h["distance"], 3)} for h in relevant],43        "confidence": "high" if relevant[0]["distance"] < 0.35 else "medium",44    }

Three things in there separate a trustworthy system from a plausible one.

The distance threshold. Vector search always returns k results — it returns the four nearest chunks whether or not any of them is relevant. Ask an employee handbook about the offside rule and you get four chunks about something else, and a model handed irrelevant context will still try to answer from it. Filtering by distance is what turns "always answers" into "answers when it knows". Tune the threshold against your own documents by logging distances for questions you know are answerable and questions you know are not; the gap between the two distributions is where the threshold goes.

Citations embedded in the context. Because each passage is prefixed with its source, the model cites accurately instead of inventing a page number, and anyone can check the answer in ten seconds.

The exact refusal string. Specifying the precise sentence makes refusals detectable in code and countable in metrics. A rising refusal rate means retrieval has degraded — often a scanned PDF that indexed to nothing.

The hard part of document Q&A is not answering — it is declining. Any system will produce an answer for every question; only one with a relevance threshold will tell you when the answer is not in your documents.

The API

1# app/api.py2from fastapi import FastAPI, UploadFile, File, HTTPException3from pydantic import BaseModel, Field4import tempfile, os, asyncio, logging, time5from app import documents, chunking, store, qa6from app.config import settings78app = FastAPI(title="Document Q&A", version="1.0.0")910class AskRequest(BaseModel):11    question: str = Field(min_length=3, max_length=1000)12    document: str | None = None1314@app.post("/documents")15async def upload(file: UploadFile = File(...)):16    data = await file.read()17    if len(data) > settings.max_upload_mb * 1024 * 1024:18        raise HTTPException(413, f"File exceeds {settings.max_upload_mb} MB")19    with tempfile.NamedTemporaryFile(20            suffix=os.path.splitext(file.filename)[1], delete=False) as tmp:21        tmp.write(data); tmp_path = tmp.name22    try:23        pages = await asyncio.to_thread(documents.extract, tmp_path)24        chunks = chunking.chunk_pages(pages, file.filename)25        added = await asyncio.to_thread(store.add_chunks, chunks)26    except ValueError as exc:27        raise HTTPException(422, str(exc))28    finally:29        os.unlink(tmp_path)3031    return {"document": file.filename, "pages": len(pages), "chunks": added}3233@app.post("/ask")34async def ask(req: AskRequest):35    started = time.perf_counter()36    result = await asyncio.to_thread(qa.answer, req.question, req.document)37    result["latency_ms"] = int((time.perf_counter() - started) * 1000)38    logging.info("ask latency=%dms sources=%d confidence=%s",39                 result["latency_ms"], len(result["sources"]), result["confidence"])40    return result

asyncio.to_thread is the important detail. PDF parsing and the synchronous Chroma and model calls all block; running them directly inside an async def handler freezes the event loop and stalls every other in-flight request on that worker.

Note also the finally: os.unlink. Without it, every upload leaves a temporary file behind and the container's disk fills up over weeks — a slow failure that is baffling when it finally arrives.

The interface

1# ui/app.py2import streamlit as st, requests, os34API = os.getenv("API_URL", "http://localhost:8000")5st.set_page_config(page_title="Document Q&A", page_icon="📄")6st.title("Document Q&A")78with st.sidebar:9    up = st.file_uploader("Upload a document", type=["pdf", "docx", "txt"])10    if up and st.button("Index it"):11        with st.spinner("Indexing..."):12            r = requests.post(f"{API}/documents",13                              files={"file": (up.name, up.getvalue())}, timeout=300)14        if r.ok:15            d = r.json()16            st.success(f"{d['document']}: {d['pages']} pages, {d['chunks']} chunks")17        else:18            st.error(r.json().get("detail", "Upload failed"))1920if q := st.chat_input("Ask about your documents"):21    st.chat_message("user").markdown(q)22    with st.chat_message("assistant"):23        with st.spinner("Searching..."):24            r = requests.post(f"{API}/ask", json={"question": q}, timeout=60)25        if not r.ok:26            st.error("The service is unavailable.")27        else:28            d = r.json()29            st.markdown(d["answer"])30            if d["sources"]:31                with st.expander(f"{len(d['sources'])} sources, {d['latency_ms']} ms"):32                    for s in d["sources"]:33                        st.caption(f"{s['document']} page {s['page']} "34                                   f"(distance {s['distance']})")

The UI holds no model logic and no key — it is an HTTP client, so the same backend can serve a Slack bot or a mobile app later. Exposing the distances in the expander is not decoration either: it is the fastest debugging tool you have when someone reports a bad answer.

Running it

A slim Python base image, a non-root user, PYTHONUNBUFFERED=1 so logs are not swallowed, and uvicorn app.api:app --host 0.0.0.0 as the command. Then compose the two services:

services:  api:    build: .    ports: ["8000:8000"]    environment:      - OPENAI_API_KEY=${OPENAI_API_KEY}    volumes:      - ./storage:/app/storage    restart: unless-stopped  ui:    build: { context: ., dockerfile: Dockerfile.ui }    ports: ["8501:8501"]    environment:      - API_URL=http://api:8000    depends_on: [api]

The volume on ./storage stops every redeploy wiping your index and forcing a full re-ingestion. The key arrives from the environment at run time — never baked into the image, where docker history exposes it to anyone who can pull it.

Testing it honestly

Write down twenty questions with known correct answers before you tune anything. Without that set you will "improve" the prompt by feel and never know whether you helped.

1CASES = [2    # (question, must appear in answer, expected page)3    ("How many days of annual leave do I get?", "25", 12),4    ("What is the notice period during probation?", "one week", 8),5    ("Can I carry over unused leave?", "five days", 12),6    ("What is the offside rule?", "could not find", None),   # must refuse7]89def test_answers():10    for question, expected, page in CASES:11        result = qa.answer(question)12        assert expected.lower() in result["answer"].lower(), question13        if page:14            assert page in [s["page"] for s in result["sources"]], question

The refusal case is the most valuable test in the file. A system that answers everything is easy to build and worthless; one that declines when it should is the whole point, and it breaks silently whenever you loosen the distance threshold.

Three failure modes and how to tell them apart, since they look identical from the outside:

Making it faster and cheaper once it works

Do these in order, and measure between each one.

Cache question embeddings. The same forty questions recur constantly. Caching removes a 50–100 ms call from all repeat traffic, in about ten lines of code.

Cache whole answers. Generation is not fully deterministic, and many current models do not accept temperature=0 at all, but for a factual question over unchanged documents one good answer is all you need — and a cache makes every repeat identical by construction. Key on the question plus the document set, and give it a TTL so re-uploads invalidate. On FAQ traffic a 30% hit rate is realistic, which at 0.011 dollars a call and 1,000 calls a day saves about 3.30 dollars a day and, more visibly, returns those answers in milliseconds.

Tune k deliberately. Each extra chunk adds roughly 800 tokens to every prompt, so k = 4 to k = 8 doubles your context cost. Check it against your twenty test cases before paying for it; often it changes nothing.

Stream the answer. Generation dominates your latency. Streaming does not reduce total time, but it puts the first words on screen in a few hundred milliseconds instead of three seconds — the difference users actually perceive.

Log four numbers on every question: retrieval latency, generation latency, top distance, and whether it refused. Those turn every future complaint into a diagnosis instead of a guess.

What to build next, and what you now know how to judge

Finish the core version first, then treat these as the real extensions — each one addresses a limitation you will meet the moment other people start using it.

• Hybrid search. Vector search struggles with exact identifiers — product codes, clause numbers, names. Running a keyword search alongside it and merging the results fixes the single most common "why did it not find that" complaint.
• Multi-document filtering. Once ten documents are indexed, answers start mixing sources. The where filter is already in the search function; wire it to a selector in the UI.
• Conversational follow-ups. "What about during probation?" only makes sense given the previous question, so rewrite it into a standalone question before embedding — otherwise you embed a fragment and retrieve nothing.
• Re-ranking. Retrieve twenty chunks by similarity, then use a cheap model to score which four genuinely answer the question. One extra small call, and it usually improves precision more than any amount of prompt tuning.
• Access control. Once two teams share the service, chunk metadata must carry who may see it and the search filter must enforce it — far easier before the index has documents in it.

The larger point is what this build teaches you to evaluate. When a vendor demonstrates a document Q&A product, you now know the questions that separate a real one from a demo: what is the chunk size and overlap, does it cite pages, what happens when the answer is not in the corpus, what is the distance threshold, and how does it handle a scanned PDF. Those five questions expose more than any feature list, and you know them because you have had to answer each one yourself.