RAG (Retrieval-Augmented Generation) is the most practical way to give an LLM access to your private data without fine-tuning. You store your documents in a vector database, retrieve the most relevant chunks at query time, and inject them into the prompt as context. This guide builds the pipeline with LangChain and FastAPI, then covers the parts that separate a demo from something you can run in production.

The three stages: indexing, retrieval, generation

• ✓Indexing: load your documents, split them into chunks, embed each chunk, store the vectors. Done once, then incrementally as documents change.
• ✓Retrieval: embed the user question, run a similarity search, return the top-k chunks, optionally filtered by metadata.
• ✓Generation: build a prompt from the retrieved chunks, call the LLM, return the answer with its sources.

Indexing: loaders, splitting, embeddings

Use a loader per source type, split with RecursiveCharacterTextSplitter so chunks break on paragraph and sentence boundaries, then embed and persist. Chunk size is a trade-off: smaller chunks give sharper retrieval but lose surrounding context, larger chunks keep context but dilute the match. Start around 800 tokens with 15 percent overlap and tune later against your evaluation set.

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

loader = PyPDFLoader("docs/technical-spec.pdf")
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
chunks = splitter.split_documents(loader.load())

for chunk in chunks:
    chunk.metadata["source"] = "technical-spec"
    chunk.metadata["visibility"] = "internal"

vectorstore = Chroma.from_documents(
    chunks,
    OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db",
)

Retrieval: where answer quality is won or lost

Most quality gains come from retrieval, not the model. Tune k to the smallest number of chunks that reliably contains the answer. Use MMR when your corpus has near-duplicate passages, so the context is diverse rather than five copies of the same paragraph. Apply metadata filters for access control and freshness. For corpora with a lot of exact terms (product codes, error messages), add a keyword retriever alongside the vector one and merge the results.

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,
        "filter": {"visibility": "internal"},
    },
)

Generation: a grounded prompt with LCEL

Build the chain with LCEL: create_stuff_documents_chain formats the retrieved chunks into the prompt, create_retrieval_chain wires the retriever in front of it. The prompt must tell the model to answer only from the context and to say it does not know otherwise, which is the single most effective guard against hallucination.

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "Answer the question using only the context below. "
     "If the context does not contain the answer, say you do not know. "
     "Cite the source of each fact.\n\nContext:\n{context}"),
    ("human", "{input}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
combine = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, combine)

Exposing it via FastAPI, with streaming

A naive endpoint waits for the full generation before responding, which feels slow on a multi-second answer. Stream the tokens as they arrive with a StreamingResponse over chain.astream, and return the source documents so the client can show citations. Keep a non-streaming variant for server-to-server callers that just want the JSON.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class Query(BaseModel):
    question: str

@app.post("/ask")
async def ask(body: Query):
    async def tokens():
        async for part in rag_chain.astream({"input": body.question}):
            if answer := part.get("answer"):
                yield answer
    return StreamingResponse(tokens(), media_type="text/plain")

Production concerns

• ✓Rate-limit LLM calls per user and per API key, with a clear 429 response.
• ✓Cache answers to frequent or identical questions, ideally with a semantic cache keyed on the embedded question.
• ✓Track tokens and cost per request, attributed to a user or tenant, so scale does not surprise you on the invoice.
• ✓Set a timeout on the LLM call and a fallback response (a static help page, a contact form) when it fails.
• ✓Filter at retrieval for access control and personal data, never rely on the prompt to keep a document hidden.
• ✓Trace every request (retrieved chunks, prompt, latency, cost) so you can debug a bad answer after the fact.

Evaluating a RAG system

Build a reference set of questions with their expected answer and expected source. Measure retrieval and generation separately: whether the right chunk lands in the top-k (retrieval), and whether the answer stays faithful to the context (generation). Re-run the whole set on every change to chunking, retrieval or the prompt, and treat a regression as a blocker.

A production RAG pipeline is not much more code than a demo, but it is a different mindset: tuned retrieval, a grounded prompt, streaming, caching, cost tracking, and an evaluation set that runs on every change. Get those in place and the system stays trustworthy as the corpus and the traffic grow.