Building RAG Applications with LangChain: Step-by-Step

Retrieval-Augmented Generation (RAG) is the most practical pattern for building LLM applications that work with your own data. Instead of fine-tuning a model, you retrieve relevant context at query time and feed it to the LLM alongside the user's question.

The concept is simple. Building a RAG pipeline that actually works in production — with good retrieval quality, reasonable latency, and manageable cost — requires careful decisions at every step.

This guide walks through building a complete RAG application with LangChain, from document ingestion to evaluation.

Architecture Overview

A RAG pipeline has two phases:

Indexing (offline):

1. Load documents
2. Split into chunks
3. Generate embeddings
4. Store in vector database

Retrieval + Generation (runtime):

1. User asks a question
2. Embed the question
3. Search vector store for similar chunks
4. Feed chunks + question to LLM
5. Return generated answer

┌─────────┐    ┌──────────┐    ┌────────────┐    ┌──────────┐
│Documents│───>│ Chunking │───>│ Embeddings │───>│VectorDB  │
└─────────┘    └──────────┘    └────────────┘    └──────────┘
                                                       │
                                                       ▼
┌─────────┐    ┌──────────┐    ┌────────────┐    ┌──────────┐
│ Answer  │<───│   LLM    │<───│  Prompt +   │<───│Retriever │
└─────────┘    └──────────┘    │  Context   │    └──────────┘
                               └────────────┘

Setup and Dependencies

pip install langchain langchain-openai langchain-community \
    chromadb tiktoken unstructured pypdf

import os
os.environ["OPENAI_API_KEY"] = "sk-..."  # Use env vars in production

Step 1: Document Loading

LangChain supports dozens of document loaders. Here are the most common patterns:

from langchain_community.document_loaders import (
    PyPDFLoader,
    DirectoryLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    WebBaseLoader,
)


def load_documents(source_dir: str) -> list:
    """Load documents from a directory of mixed file types."""

    loaders = {
        "*.pdf": PyPDFLoader,
        "*.txt": TextLoader,
        "*.md": UnstructuredMarkdownLoader,
    }

    all_docs = []
    for glob_pattern, loader_cls in loaders.items():
        dir_loader = DirectoryLoader(
            source_dir,
            glob=glob_pattern,
            loader_cls=loader_cls,
            show_progress=True,
            use_multithreading=True,
        )
        docs = dir_loader.load()
        all_docs.extend(docs)
        print(f"Loaded {len(docs)} documents matching {glob_pattern}")

    return all_docs


# Load from web
def load_web_documents(urls: list[str]) -> list:
    """Load documents from web URLs."""
    loader = WebBaseLoader(urls)
    return loader.load()


# Usage
docs = load_documents("./data/knowledge_base/")
print(f"Total documents loaded: {len(docs)}")

Step 2: Text Chunking

Chunking is the most critical step for retrieval quality. The wrong chunk size means the LLM either gets too little context or too much noise.

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
)


def create_chunks(
    documents: list,
    chunk_size: int = 1000,
    chunk_overlap: int = 200,
) -> list:
    """Split documents into chunks with metadata preservation."""

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
        is_separator_regex=False,
    )

    chunks = splitter.split_documents(documents)

    # Add chunk metadata for debugging and filtering
    for i, chunk in enumerate(chunks):
        chunk.metadata["chunk_id"] = i
        chunk.metadata["chunk_size"] = len(chunk.page_content)

    print(f"Created {len(chunks)} chunks from {len(documents)} documents")
    print(f"Avg chunk size: {sum(len(c.page_content) for c in chunks) / len(chunks):.0f} chars")

    return chunks


def create_markdown_chunks(documents: list) -> list:
    """Chunk markdown by headers for better semantic boundaries."""

    headers_to_split_on = [
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ]

    md_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on,
        strip_headers=False,
    )

    # Further split large sections
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
    )

    all_chunks = []
    for doc in documents:
        md_chunks = md_splitter.split_text(doc.page_content)
        sub_chunks = text_splitter.split_documents(md_chunks)
        all_chunks.extend(sub_chunks)

    return all_chunks

Chunking Strategy Guidelines

Content Type	Chunk Size	Overlap	Strategy
Technical docs	800-1200	200	Recursive by paragraphs
Legal documents	1000-1500	300	Recursive with high overlap
Code documentation	500-800	100	Markdown header splitting
FAQ / Q&A	300-500	50	Per question-answer pair

Step 3: Vector Store with ChromaDB

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma


def create_vector_store(
    chunks: list,
    persist_directory: str = "./chroma_db",
    collection_name: str = "knowledge_base",
) -> Chroma:
    """Create and persist a ChromaDB vector store."""

    embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small",
        # Cost: ~$0.02 per 1M tokens
    )

    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_directory,
        collection_name=collection_name,
    )

    print(f"Vector store created with {vector_store._collection.count()} vectors")
    return vector_store


def load_vector_store(
    persist_directory: str = "./chroma_db",
    collection_name: str = "knowledge_base",
) -> Chroma:
    """Load an existing vector store from disk."""

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

    return Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings,
        collection_name=collection_name,
    )

Step 4: Building the Retriever

The retriever is where you control retrieval quality. Basic similarity search is a starting point, but production systems need more.

from langchain.retrievers import (
    ContextualCompressionRetriever,
    MultiQueryRetriever,
)
from langchain.retrievers.document_compressors import (
    LLMChainExtractor,
)
from langchain_openai import ChatOpenAI


def create_basic_retriever(vector_store: Chroma, k: int = 4):
    """Simple similarity search retriever."""
    return vector_store.as_retriever(
        search_type="similarity",
        search_kwargs={"k": k},
    )


def create_mmr_retriever(
    vector_store: Chroma, k: int = 4, fetch_k: int = 20
):
    """MMR retriever for diverse results (reduces redundancy)."""
    return vector_store.as_retriever(
        search_type="mmr",
        search_kwargs={
            "k": k,
            "fetch_k": fetch_k,
            "lambda_mult": 0.7,  # 0 = max diversity, 1 = max relevance
        },
    )


def create_multi_query_retriever(vector_store: Chroma):
    """Generate multiple query variations for better recall."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

    return MultiQueryRetriever.from_llm(
        retriever=vector_store.as_retriever(search_kwargs={"k": 4}),
        llm=llm,
    )


def create_compression_retriever(vector_store: Chroma):
    """Retrieve then compress — extract only relevant parts."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    compressor = LLMChainExtractor.from_llm(llm)

    base_retriever = vector_store.as_retriever(
        search_kwargs={"k": 6}
    )

    return ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=base_retriever,
    )

Step 5: The RAG Chain

Now we connect everything into a chain that takes a question and returns an answer with sources.

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


def format_docs(docs: list) -> str:
    """Format retrieved documents for the prompt."""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "Unknown")
        formatted.append(
            f"[Source {i}: {source}]\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)


def create_rag_chain(retriever, model_name: str = "gpt-4o"):
    """Create a complete RAG chain with source attribution."""

    llm = ChatOpenAI(model=model_name, temperature=0.1)

    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a helpful assistant that answers questions 
based on the provided context. Follow these rules:

1. Only answer based on the provided context
2. If the context doesn't contain enough information, say so
3. Cite your sources using [Source N] notation
4. Be concise and direct
5. If you're unsure, express your uncertainty

Context:
{context}"""),
        ("human", "{question}"),
    ])

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

    return chain


# Build and use the chain
vector_store = load_vector_store()
retriever = create_mmr_retriever(vector_store)
rag_chain = create_rag_chain(retriever)

# Ask a question
answer = rag_chain.invoke(
    "How do I configure auto-scaling for the data pipeline?"
)
print(answer)

Step 6: Adding Chat History

For conversational RAG, you need to rephrase follow-up questions using chat history.

from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chains.history_aware_retriever import (
    create_history_aware_retriever,
)
from langchain.chains.retrieval import create_retrieval_chain
from langchain.chains.combine_documents import (
    create_stuff_documents_chain,
)


def create_conversational_rag(retriever):
    """RAG chain with conversation history support."""

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

    # Step 1: Rephrase question using history
    contextualize_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "Given the chat history and latest question, "
         "rephrase the question to be standalone. "
         "Do NOT answer the question."),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ])

    history_aware_retriever = create_history_aware_retriever(
        llm, retriever, contextualize_prompt
    )

    # Step 2: Answer with context
    answer_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "Answer based on the context below. "
         "If unsure, say you don't know.\n\n{context}"),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ])

    question_answer_chain = create_stuff_documents_chain(
        llm, answer_prompt
    )

    return create_retrieval_chain(
        history_aware_retriever, question_answer_chain
    )


# Usage with history
conversational_chain = create_conversational_rag(retriever)
chat_history = []

# First question
result = conversational_chain.invoke({
    "input": "What databases does the platform support?",
    "chat_history": chat_history,
})
print(result["answer"])

# Track history
chat_history.extend([
    HumanMessage(content="What databases does the platform support?"),
    AIMessage(content=result["answer"]),
])

# Follow-up question (uses history for context)
result = conversational_chain.invoke({
    "input": "Which one has the best performance?",
    "chat_history": chat_history,
})
print(result["answer"])

Step 7: Evaluation

You can't improve what you don't measure. Here's a practical evaluation framework:

from dataclasses import dataclass


@dataclass
class EvalCase:
    question: str
    expected_answer: str
    expected_sources: list[str] | None = None


def evaluate_rag(
    chain,
    retriever,
    eval_cases: list[EvalCase],
) -> dict:
    """Evaluate RAG pipeline on a test set."""

    results = {
        "total": len(eval_cases),
        "retrieval_hits": 0,
        "answer_quality": [],
        "latencies": [],
    }

    llm_judge = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    for case in eval_cases:
        import time
        start = time.time()

        # Get retrieved docs
        retrieved_docs = retriever.invoke(case.question)
        answer = chain.invoke(case.question)

        latency = time.time() - start
        results["latencies"].append(latency)

        # Check if expected sources were retrieved
        if case.expected_sources:
            retrieved_sources = [
                d.metadata.get("source", "") for d in retrieved_docs
            ]
            hit = any(
                exp in src
                for exp in case.expected_sources
                for src in retrieved_sources
            )
            if hit:
                results["retrieval_hits"] += 1

        # LLM-as-judge for answer quality
        judge_prompt = f"""Rate this answer from 1-5:
Question: {case.question}
Expected: {case.expected_answer}
Actual: {answer}

Score (1-5):"""

        score_response = llm_judge.invoke(judge_prompt)
        try:
            score = int(score_response.content.strip()[0])
        except (ValueError, IndexError):
            score = 3
        results["answer_quality"].append(score)

    # Calculate metrics
    results["avg_quality"] = (
        sum(results["answer_quality"]) / len(results["answer_quality"])
    )
    results["avg_latency"] = (
        sum(results["latencies"]) / len(results["latencies"])
    )
    results["retrieval_accuracy"] = (
        results["retrieval_hits"] / results["total"]
    )

    return results


# Define test cases
eval_cases = [
    EvalCase(
        question="How do I set up auto-scaling?",
        expected_answer="Configure min/max instances in the scaling policy...",
        expected_sources=["auto-scaling-guide.md"],
    ),
    EvalCase(
        question="What authentication methods are supported?",
        expected_answer="OAuth2, API keys, and SAML SSO...",
        expected_sources=["auth-docs.md"],
    ),
]

results = evaluate_rag(rag_chain, retriever, eval_cases)
print(f"Retrieval Accuracy: {results['retrieval_accuracy']:.1%}")
print(f"Answer Quality: {results['avg_quality']:.1f}/5")
print(f"Avg Latency: {results['avg_latency']:.2f}s")

Production Tips

1. Chunk size matters more than you think. Start with 800-1000 characters and experiment. Too small = missing context. Too large = noise.

2. Use hybrid search. Combine vector similarity with keyword (BM25) search for better results on exact term matches.

3. Cache embeddings. Don't re-embed unchanged documents. Track file hashes and only re-index what changed.

4. Monitor retrieval quality. Log every query, the retrieved chunks, and user feedback. This is your training data for improvement.

5. Set token budgets. Calculate: context_tokens + prompt_tokens + max_output_tokens < model_limit. Budget accordingly.

Summary

Building a RAG pipeline is iterative. Start simple, measure quality, and improve one component at a time:

1. Get documents loaded and chunked
2. Build basic retrieval with similarity search
3. Add a simple prompt and chain
4. Evaluate with test cases
5. Improve chunking, retrieval, and prompts based on results

The code in this article gives you a solid foundation. Every component is modular and swappable.

---

Get Production RAG Templates

Building RAG from scratch means solving the same problems everyone else has solved. The RAG Pipeline Framework from AI Builder Pro gives you a complete, production-tested RAG implementation with document ingestion, multiple chunking strategies, vector store integration, evaluation harnesses, and deployment configs.

The full AI Builder Pro collection includes 11 AI/LLM tools: RAG pipelines, prompt engineering kits, agent frameworks, fine-tuning pipelines, guardrails, and more.

Use code LAUNCH40 for 40% off any product, or STUDENT for 50% off.

Browse the AI Builder Pro store