Building a RAG Chatbot with LangChain and FastAPI in Python

Retrieval-Augmented Generation (RAG) is the most practical way to make a large language model answer questions about your own data. Instead of fine-tuning, you retrieve the most relevant passages at query time and hand them to the model as context. This guide walks through a production-shaped RAG chatbot in Python using LangChain, a vector store and FastAPI.

Why RAG instead of fine-tuning

Fine-tuning changes how a model behaves; RAG changes what it knows. For most business use cases — support docs, internal wikis, product catalogs — RAG is faster to ship, cheaper to update and far easier to audit because every answer can be traced back to a source document.

Updates are instant: re-index a document instead of retraining a model.
Grounded answers: you can cite the exact chunk the model used.
Lower cost: no GPU training runs and no custom model hosting.
Better privacy: your data stays in your own vector store.

The four-stage architecture

Ingest: load documents and split them into overlapping chunks.
Embed: turn each chunk into a vector with an embedding model.
Retrieve: at query time, find the closest chunks to the question.
Generate: pass the question and retrieved chunks to the LLM and stream the answer.

Step 1 — Load and split your documents

Start by loading your source files and splitting them into chunks. Overlap matters: it prevents an answer from being cut in half at a chunk boundary.

from langchain_community.document_loaders import DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = DirectoryLoader("./docs", glob="**/*.md")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=120,
    separators=["\n\n", "\n", ". ", " "],
)
chunks = splitter.split_documents(documents)
print(f"Loaded {len(documents)} docs into {len(chunks)} chunks")

Step 2 — Embed and store in a vector database

Each chunk becomes a vector that captures its meaning. Store those vectors in a database that supports similarity search — pgvector, Qdrant, Chroma or Pinecone all work well.

from langchain_openai import OpenAIEmbeddings
from langchain_postgres import PGVector

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = PGVector.from_documents(
    documents=chunks,
    embedding=embeddings,
    collection_name="handbook",
    connection="postgresql+psycopg://user:pass@localhost:5432/app",
)

Step 3 — Build the retrieval chain

The retriever fetches the top-k most similar chunks, and a prompt template instructs the model to answer only from that context. The instruction to say "I do not know" when the context is insufficient is what keeps a RAG system honest.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

retriever = vector_store.as_retriever(search_kwargs={"k": 4})
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "Answer using only the context below. "
     "If the answer is not in the context, say you do not know.\n\n{context}"),
    ("human", "{question}"),
])

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
)

Step 4 — Serve it with FastAPI

A thin FastAPI layer turns the chain into an HTTP service. Note the dependency injection and the explicit request model — both make the endpoint easy to test and document via the automatic OpenAPI schema.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="RAG Chatbot API")

class Query(BaseModel):
    question: str = Field(min_length=1, max_length=1000)

class Answer(BaseModel):
    answer: str

@app.post("/chat", response_model=Answer)
async def chat(query: Query) -> Answer:
    try:
        result = await chain.ainvoke(query.question)
    except Exception as exc:
        raise HTTPException(status_code=502, detail="Upstream model error") from exc
    return Answer(answer=result.content)

Production tips that matter

Stream tokens to the client so users see progress immediately.
Log the retrieved chunk IDs with each answer for debugging and citations.
Cache embeddings so re-indexing only reprocesses changed documents.
Evaluate retrieval separately from generation — most bad answers are retrieval failures.
Add rate limiting and timeouts; LLM calls fail in ways a normal API does not.

Most RAG quality problems are retrieval problems. Fix the search before you blame the model.

Once the retrieval layer is reliable, RAG stops feeling like a demo and starts behaving like infrastructure. That is the point where it becomes genuinely useful to a business.