Building RAG Pipelines

This topic covers building production-quality RAG pipelines using popular frameworks. We'll go from raw documents to a fully functional question-answering system.

RAG Frameworks

Two major frameworks simplify RAG development:

Framework	Best For	Philosophy
LangChain	General-purpose, flexible chains	Composable components
LlamaIndex	Document-focused RAG	Data framework for LLMs

RAG Pipeline with LangChain

Installation

bash

pip install langchain langchain-openai langchain-chroma

Complete Pipeline

python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    DirectoryLoader,
    TextLoader,
    PyPDFLoader
)
from langchain.chains import RetrievalQA

# 1. Load documents
loader = DirectoryLoader(
    "./docs",
    glob="**/*.txt",
    loader_cls=TextLoader
)
documents = loader.load()

# 2. Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# 3. Create embeddings and store in vector DB
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. Create retrieval chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # Stuff all docs into prompt
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

# 5. Query
result = qa_chain.invoke({"query": "What is the refund policy?"})
print(result["result"])
print("Sources:", [doc.metadata for doc in result["source_documents"]])

Loading Different Document Types

python

from langchain_community.document_loaders import (
    PyPDFLoader,
    CSVLoader,
    UnstructuredMarkdownLoader,
    WebBaseLoader
)

# PDFs
pdf_loader = PyPDFLoader("report.pdf")
pdf_docs = pdf_loader.load()

# CSV files
csv_loader = CSVLoader("data.csv")
csv_docs = csv_loader.load()

# Web pages
web_loader = WebBaseLoader("https://docs.example.com/guide")
web_docs = web_loader.load()

# Markdown
md_loader = UnstructuredMarkdownLoader("README.md")
md_docs = md_loader.load()

RAG Pipeline with LlamaIndex

Installation

bash

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai

Complete Pipeline

python

from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# Configure
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0.2)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 1. Load documents
documents = SimpleDirectoryReader("./docs").load_data()

# 2. Build index (handles chunking and embedding automatically)
index = VectorStoreIndex.from_documents(documents)

# 3. Create query engine
query_engine = index.as_query_engine(similarity_top_k=4)

# 4. Query
response = query_engine.query("What is the refund policy?")
print(response)
print("Sources:", response.source_nodes)

LlamaIndex is more concise because it handles chunking, embedding, and retrieval internally.

Chunking Strategies in Practice

Recursive Character Splitting (Most Common)

python

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # Target chunk size in characters
    chunk_overlap=200,     # Overlap between chunks
    separators=[
        "\n\n",            # Try paragraph breaks first
        "\n",              # Then line breaks
        ". ",              # Then sentences
        " ",               # Then words
        ""                 # Last resort: characters
    ]
)

Markdown-Aware Splitting

python

from langchain.text_splitter import MarkdownHeaderTextSplitter

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

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)
chunks = splitter.split_text(markdown_content)
# Each chunk knows which section it belongs to

Custom RAG Prompt

The default prompt might not be optimal. Customize it:

python

from langchain.prompts import PromptTemplate

custom_prompt = PromptTemplate(
    template="""You are a helpful assistant for our company.
    Use the following context to answer the question.

    Rules:
    - Only answer based on the provided context
    - If you don't know, say "I don't have that information"
    - Cite which document your answer comes from
    - Be concise and direct

    Context:
    {context}

    Question: {question}

    Answer:""",
    input_variables=["context", "question"]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    chain_type_kwargs={"prompt": custom_prompt}
)

Conversational RAG (Chat with Memory)

python

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    return_messages=True,
    k=5  # Keep last 5 exchanges
)

chat_chain = ConversationalRetrievalChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    retriever=vectorstore.as_retriever(),
    memory=memory,
    return_source_documents=True
)

# Multi-turn conversation
response1 = chat_chain.invoke({"question": "What products do you offer?"})
response2 = chat_chain.invoke({"question": "How much does the premium one cost?"})
# The chain remembers the previous context

Evaluating RAG Quality

Key Metrics

Metric	What It Measures	How
Retrieval precision	Are retrieved docs relevant?	Manual review
Retrieval recall	Are all relevant docs found?	Test with known answers
Answer correctness	Is the final answer right?	Compare to ground truth
Faithfulness	Does the answer stick to the context?	Check for hallucinations

Simple Evaluation

python

test_cases = [
    {
        "question": "What is the refund policy?",
        "expected": "14 days",
        "source": "policies.txt"
    },
    {
        "question": "What are the business hours?",
        "expected": "9 AM to 6 PM",
        "source": "support.txt"
    }
]

for test in test_cases:
    result = qa_chain.invoke({"query": test["question"]})
    answer = result["result"]
    sources = [d.metadata.get("source") for d in result["source_documents"]]

    correct = test["expected"].lower() in answer.lower()
    right_source = test["source"] in str(sources)

    print(f"Q: {test['question']}")
    print(f"  Answer correct: {correct}")
    print(f"  Right source: {right_source}")

Production Checklist

Documents are chunked with appropriate size and overlap
Metadata is preserved (source, page, date)
Embedding model matches between indexing and querying
System prompt prevents hallucination
Error handling for empty retrievals
Response includes source citations
Index is updated when documents change
Query latency is acceptable (< 3 seconds)
Cost per query is within budget

Key Takeaways

Use frameworks (LangChain or LlamaIndex) — don't build from scratch
Chunking strategy matters — use recursive splitting with appropriate overlap
Customize your prompt — default prompts are generic
Evaluate systematically — build test cases with expected answers
Add conversation memory for multi-turn interactions

Next, we'll cover advanced RAG techniques that significantly improve retrieval quality.