Building a Production AI Chatbot with LangChain, RAG, and FastAPI

Why Most AI Chatbots Fail in Production

I have built and deployed over a dozen AI chatbot systems for clients across e-commerce, SaaS, and professional services. In almost every case, the prototype worked beautifully in a Jupyter notebook and fell apart the moment it hit real users. The failure modes are always the same: hallucinated answers, no grounding in actual business data, context windows that collapse under long conversations, and no way to observe what the model is doing when something goes wrong.

The fix is not a better prompt. The fix is a proper architecture — one built around Retrieval-Augmented Generation, a disciplined LangChain chain design, and a FastAPI backend that gives you real observability. This guide walks through exactly that system, the one I now use as my default starting point for every production chatbot project.

Before we write a single line of code, understand the core problem: a language model is a probability engine trained on static data. Your users want answers about your product, your documents, your policies — none of which exist in the model's training data. RAG bridges that gap by dynamically injecting relevant context from your own knowledge base into every prompt, so the model answers based on real information rather than educated guesses.

Architecture Overview

The system has three distinct layers. The indexing layer runs offline: it loads your documents, splits them into chunks, generates embeddings, and stores them in a vector database. The retrieval layer runs at inference time: it takes a user query, embeds it, and fetches the most semantically similar chunks. The generation layer feeds those chunks plus the conversation history into the LLM to produce a grounded response.

For this guide, the stack is: Python 3.11, LangChain 0.3, OpenAI gpt-4o-mini for generation, text-embedding-3-small for embeddings, ChromaDB for the local vector store, and FastAPI 0.115 for the API layer. The same architecture works with Pinecone, Weaviate, or pgvector — swap the vector store adapter and the rest is identical.

Dependencies

pip install langchain langchain-openai langchain-community chromadb   fastapi uvicorn python-dotenv pypdf sentence-transformers

Setting Up the RAG Pipeline

The indexing step is the foundation. Get this wrong and nothing else matters. The two most common mistakes are chunking too large (the LLM gets drowned in irrelevant context) and chunking too small (individual chunks lose their semantic meaning). A chunk size of 512 tokens with a 64-token overlap works well for most document types.

import os
from langchain_community.document_loaders import PyPDFDirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from dotenv import load_dotenv

load_dotenv()

def build_vector_store(docs_path: str, persist_directory: str) -> Chroma:
    """Load documents, chunk them, embed and persist to ChromaDB."""
    loader = PyPDFDirectoryLoader(docs_path)
    raw_docs = loader.load()

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=["

", "
", ". ", " ", ""],
    )
    chunks = splitter.split_documents(raw_docs)
    print(f"Indexed {len(raw_docs)} documents into {len(chunks)} chunks")

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_directory,
    )
    return vector_store


if __name__ == "__main__":
    build_vector_store(docs_path="./docs", persist_directory="./chroma_db")

Run this script once to build the index. For production, wrap it in a background task that re-indexes whenever you update your documents. A simple file watcher or a webhook from your CMS works well here.

Building the LangChain Chain

LangChain's LCEL (LangChain Expression Language) makes chain composition readable and debuggable. The conversational retrieval chain below handles multi-turn dialogue by first condensing the chat history and new question into a standalone question, then running retrieval, then generating the final answer. This two-step approach is critical — naive RAG that just embeds the raw user message fails on follow-up questions like "What about the pricing?" because it has no idea what "it" refers to.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import PromptTemplate

SYSTEM_PROMPT = """You are a helpful assistant for {company_name}.
Answer the user's question using ONLY the context provided below.
If the context does not contain enough information, say:
"I don't have that information. Please contact our support team."

Context:
{context}

Chat History:
{chat_history}

Question: {question}
Answer:"""

def create_chain(persist_directory: str, company_name: str) -> ConversationalRetrievalChain:
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vector_store = Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings,
    )

    retriever = vector_store.as_retriever(
        search_type="mmr",
        search_kwargs={"k": 6, "fetch_k": 20, "lambda_mult": 0.7},
    )

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1, max_tokens=1024)

    memory = ConversationBufferWindowMemory(
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",
        k=10,
    )

    qa_prompt = PromptTemplate(
        template=SYSTEM_PROMPT,
        input_variables=["context", "chat_history", "question", "company_name"],
    )

    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        combine_docs_chain_kwargs={"prompt": qa_prompt.partial(company_name=company_name)},
        return_source_documents=True,
        verbose=False,
    )
    return chain

Note the use of MMR (Maximal Marginal Relevance) as the search type. Standard similarity search often returns five nearly identical chunks from the same document section. MMR balances relevance with diversity, giving the LLM broader context coverage from different parts of your knowledge base.

FastAPI Deployment

The API layer is where most tutorials skip the important details. You need session management (each conversation needs its own memory instance), rate limiting, streaming responses for a better UX, and proper error handling that does not leak API keys or internal stack traces to the client.

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import AsyncGenerator
import uuid
import json

app = FastAPI(title="AI Chatbot API", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_methods=["POST"],
    allow_headers=["*"],
)

# Session store — use Redis in production
sessions: dict[str, ConversationalRetrievalChain] = {}

class ChatRequest(BaseModel):
    session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    message: str = Field(..., min_length=1, max_length=2000)

class ChatResponse(BaseModel):
    session_id: str
    answer: str
    sources: list[str]

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    if request.session_id not in sessions:
        sessions[request.session_id] = create_chain(
            persist_directory="./chroma_db",
            company_name="Acme Corp",
        )

    chain = sessions[request.session_id]

    try:
        result = await asyncio.to_thread(
            chain,
            {"question": request.message}
        )
        sources = list({
            doc.metadata.get("source", "unknown")
            for doc in result.get("source_documents", [])
        })
        return ChatResponse(
            session_id=request.session_id,
            answer=result["answer"],
            sources=sources,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail="Failed to process your message. Please try again.")

Production Hardening & Cost Control

A chatbot that works in development but bankrupts you in production is not useful. Three levers control cost: the choice of embedding model, the LLM model tier, and input token volume. text-embedding-3-small costs $0.02 per million tokens — use it. For generation, gpt-4o-mini delivers 90% of gpt-4o quality at 15x lower cost for most customer-facing chatbot use cases.

For token control, cap the retriever at 6 chunks and set max_tokens on the LLM. Implement a conversation window — do not let the chat history grow unbounded. ConversationBufferWindowMemory with k=10 keeps the last 10 exchanges, which is more than sufficient for most sessions while keeping context windows predictable.

For observability, integrate LangSmith or log every chain invocation with the question, retrieved chunk IDs, token counts, and latency. Without this, debugging hallucinations is guesswork. Set up an alert if retrieval returns zero results — that signals a gap in your knowledge base, not a model failure.

Deploy on a single Fly.io or Railway instance with 512MB RAM for most use cases. ChromaDB on disk handles millions of embeddings within this budget. Only move to managed Pinecone or Weaviate when you exceed 100k documents or need multi-region replication.