RAG Document Q&A with LangChain

Build a RAG pipeline that answers questions grounded in your own documents.

Introduction

30 Second Summary

When you ask an AI chatbot a question, it can only answer from what it learned during training. What if you need answers grounded in your own documents instead?

In this project, you will build a Retrieval-Augmented Generation (RAG) system that ingests your documents, stores them as vector embeddings, and answers questions using only the information it retrieves from those documents.

What You'll Build

A command-line document Q&A system that chunks text documents, embeds them into a ChromaDB vector store, retrieves relevant context for your queries, and generates grounded answers using gpt-4o-mini.

By the end of this project, you'll have:

  • A complete RAG pipeline that loads, chunks, embeds, and retrieves your own documents.
  • A ChromaDB vector store populated with your document embeddings and queryable by semantic similarity.
  • A query interface that generates answers grounded in retrieved context, with source citations.
  • Secret Mission: Add retrieval evaluation that measures how well your system retrieves relevant context.

Are there any prerequisites?

You should be comfortable writing Python scripts and running commands in a terminal. You also need an OpenAI API key ready to go.

This project uses LangChain and ChromaDB, but no prior experience with either is required.

Set Up Your Environment

Before you can build a RAG pipeline, you need a workspace with the right tools installed.

In this step, you will create a Python project folder, install LangChain and ChromaDB, and confirm that your OpenAI API key works by calling the embeddings endpoint.

In this step, get ready to:
  • Create a project directory and Python virtual environment.
  • Install LangChain, ChromaDB, and supporting packages.
  • Verify your OpenAI API key by running a test embedding.
Create your project directory and virtual environment
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir rag-project
cd rag-project
  • Create a Python virtual environment by running this command:

macOS

python3 -m venv venv

Windows

python -m venv venv
  • Confirm the venv folder was created by running:
ls

You should see a venv folder listed in the output.

What is a virtual environment?

A virtual environment is an isolated Python workspace. Packages you install inside it stay separate from your system Python, preventing version conflicts between projects.

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

.\venv\Scripts\Activate.ps1

Seeing an execution policy error?

  • Run this command first to allow scripts, then try the activation command again:
Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

You can close this terminal window now. You will use Cursor's integrated terminal for the rest of the project.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the rag-project folder, then click Open.
Install the required packages
  • Open Cursor's integrated terminal by pressing Ctrl+` (backtick key, located above Tab).
  • Activate your virtual environment in this new terminal session:

macOS

source venv/bin/activate

Windows

.\venv\Scripts\Activate.ps1

You should see (venv) appear in your terminal prompt again.

  • Create a new file by clicking the new file icon in Cursor's Explorer sidebar (or press Cmd+N on macOS, Ctrl+N on Windows).
  • Save the file immediately as requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Paste the following package list into requirements.txt:
langchain==1.3.11
langchain-openai==1.3.3
langchain-text-splitters==1.1.2
chromadb==1.5.9

What are these packages?

  • langchain is the orchestration framework that connects your documents, embeddings, and LLM into a pipeline.
  • langchain-openai provides the OpenAI-specific integrations for embeddings and chat models.
  • langchain-text-splitters handles document chunking strategies.
  • chromadb is a vector database that stores and searches your document embeddings locally.
  • Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Install all packages by running this command in your terminal:
pip install -r requirements.txt

This takes a minute or two. You should see lines scrolling as each package downloads and installs, ending with a message like Successfully installed... listing all the packages.

Seeing a build error for chromadb on macOS?

If the install fails mentioning hnswlib, you need Xcode command-line tools. Run this, then retry the install:

xcode-select --install

Alternatively, set this environment variable before retrying:

export HNSWLIB_NO_NATIVE=1

Help me fix a chromadb installation error on my Mac.

✔️ Awesome, I've got everything!

Great. Make sure you have saved requirements.txt before continuing.

ⓧ I'd like to double check the full code

langchain==1.3.11
langchain-openai==1.3.3
langchain-text-splitters==1.1.2
chromadb==1.5.9
Set up your API key and verify connectivity

LangChain reads your OpenAI credentials from an environment variable called OPENAI_API_KEY. You will set this in your terminal so that your scripts can authenticate with the OpenAI API.

  • Set your API key in the terminal by running this command (replace the placeholder with your actual key):

macOS

export OPENAI_API_KEY='your-api-key-here'

Windows

$env:OPENAI_API_KEY = "your-api-key-here"

This variable lasts only for your current session

If you close and reopen your terminal, you will need to set OPENAI_API_KEY again. This is intentional. Keeping keys out of files prevents accidental commits to version control.

  • Create a new file in Cursor and save it as setup_test.py.
  • Paste the following test script into setup_test.py:
import os
from langchain_openai import OpenAIEmbeddings

# Verify OpenAI API key is set
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
    print("Error: OPENAI_API_KEY environment variable not set.")
    print("Run: export OPENAI_API_KEY='your-api-key-here'")
    exit(1)

# Test embeddings endpoint
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
result = embeddings.embed_query("Hello, world!")
print(f"Success! Embedding dimension: {len(result)}")
print(f"First 5 values: {result[:5]}")

What does this script do?

  • os.environ.get("OPENAI_API_KEY") checks that your API key is available in the environment. If it is missing, the script exits with a helpful message.
  • OpenAIEmbeddings(model="text-embedding-3-small") creates an embeddings client using the text-embedding-3-small model, which converts text into 1536-dimension vector embeddings.
  • embed_query("Hello, world!") sends a test string to the OpenAI API and returns a list of numbers representing the text's meaning in vector space.
  • Save setup_test.py.
  • Run the test script to confirm everything works:
python setup_test.py

You should see output like this:

Success! Embedding dimension: 1536
First 5 values: [0.0123, -0.0456, 0.0789, ...]

The exact numbers will differ, but the dimension should be 1536. This confirms your API key is valid and the embeddings endpoint is responding.

Seeing an authentication error?

  • Make sure you replaced your-api-key-here with your actual OpenAI API key in the export command.
  • Confirm the key starts with sk-. If you copied extra spaces or quotes, remove them.
  • If you opened a new terminal tab since setting the key, you need to run the export command again in the new tab.

Help me debug an OpenAI API authentication error in my setup_test.py script.

✔️ Awesome, I've got everything!

Great. Make sure you have saved setup_test.py before continuing.

ⓧ I'd like to double check the full code

import os
from langchain_openai import OpenAIEmbeddings

# Verify OpenAI API key is set
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
    print("Error: OPENAI_API_KEY environment variable not set.")
    print("Run: export OPENAI_API_KEY='your-api-key-here'")
    exit(1)

# Test embeddings endpoint
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
result = embeddings.embed_query("Hello, world!")
print(f"Success! Embedding dimension: {len(result)}")
print(f"First 5 values: {result[:5]}")

Your environment is ready. Next up, you will create your document set and learn how chunking strategies affect what your RAG system can retrieve.

Ingest and Chunk Documents

Your environment is ready, and the OpenAI API responded to your test call. Now it's time to give your RAG system something to search through.

Before the LLM can answer questions grounded in your documents, those documents need to be broken into small, searchable pieces called chunks. In this step, you'll create a set of knowledge-base documents and write the Python functions that load and chunk them.

In this step, get ready to:
  • Create a document set of 4 text files covering AI engineering topics.
  • Load documents into LangChain Document objects with metadata.
  • Split documents into chunks using RecursiveCharacterTextSplitter.
Create your document set

A RAG system needs a knowledge base to retrieve from. You'll create 4 text files covering AI engineering topics that your system will later search through.

  • Create the documents/ directory inside your project folder by running this command:
mkdir documents
  • In Cursor, create a new file called ai_engineering.txt inside the documents/ folder with this content:
AI Engineering is a discipline focused on building production applications powered by large language models (LLMs). Unlike machine learning engineering, which involves training models from scratch, AI engineering leverages pre-trained models through APIs and focuses on the application layer.

Key skills for AI engineers include prompt engineering, retrieval-augmented generation (RAG), vector databases, and LLM orchestration frameworks. The role emerged in 2023-2024 as companies shifted from research to production deployment of LLM-powered features.

RAG is the most widely deployed enterprise LLM architecture. It connects LLMs to private data and documents, allowing models to generate answers grounded in specific information rather than relying solely on training data. Production RAG systems require careful attention to chunking strategies, embedding model selection, and retrieval evaluation.
  • Create vector_databases.txt inside documents/ with this content:
Vector databases are specialized storage systems designed to store, index, and query high-dimensional vector embeddings efficiently. They are the backbone of most RAG systems and semantic search applications.

Popular vector databases include ChromaDB, Pinecone, Weaviate, Qdrant, and pgvector. ChromaDB is an open-source option that can run locally without any cloud infrastructure, making it ideal for development and small-scale applications. Pinecone is a managed cloud service that handles scaling automatically.

The choice between vector databases depends on factors like scale requirements, hosting preferences (managed vs self-hosted), query latency needs, and filtering capabilities. For development and prototyping, ChromaDB offers the fastest path to a working system with zero infrastructure overhead.
  • Create chunking_strategies.txt inside documents/ with this content:
Document chunking is the process of splitting large documents into smaller pieces for embedding and retrieval. The chunking strategy significantly impacts retrieval quality and is one of the most important design decisions in a RAG system.

Common chunking approaches include fixed-size chunking (splitting at a set character or token count), recursive character splitting (splitting at natural boundaries like paragraphs and sentences), and semantic chunking (splitting based on meaning shifts in the text).

Key parameters for chunking are chunk size and chunk overlap. Chunk size determines how much context each chunk contains. Smaller chunks are more precise for retrieval but may lack context. Larger chunks provide more context but may dilute the relevance signal. Chunk overlap ensures that information at chunk boundaries is not lost. A typical starting configuration is chunk_size=1000 characters with chunk_overlap=200 characters.
  • Create prompt_engineering.txt inside documents/ with this content:
Prompt engineering for RAG systems requires careful design to ensure the LLM grounds its answers in the retrieved context rather than hallucinating from training data. Effective RAG prompts typically include explicit instructions to only use provided context, formatting guidelines for citations, and fallback behavior when the context does not contain relevant information.

Key prompt engineering patterns for RAG include: system prompts that define the assistant's role and constraints, context injection where retrieved documents are formatted and inserted into the prompt, and output formatting that requires source attribution. A well-designed RAG prompt will instruct the model to say "I don't know" when the retrieved context does not contain sufficient information to answer the question.

Advanced prompt techniques include chain-of-thought reasoning over retrieved context, multi-step retrieval where initial results inform follow-up queries, and structured output formats like JSON for programmatic consumption of answers.
  • Save all 4 files.
  • Confirm the files exist by running:
ls documents

You should see ai_engineering.txt, chunking_strategies.txt, prompt_engineering.txt, and vector_databases.txt in the output.

Don't see all 4 files?

Make sure you created each file inside the documents/ subdirectory, not in the project root. Check that each filename ends with .txt and matches the exact names above.

help me find my missing document files

Write the document loading function

Now you need Python code that reads these text files and converts them into LangChain Document objects. Each Document pairs the file content with metadata (like the source filename) so your RAG system knows where each piece of information came from.

  • Create a new file called rag_pipeline.py in your project root.
  • Add the following imports at the top of the file:
import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import chromadb

Why import modules we haven't used yet?

You're importing the full set of dependencies upfront because you'll add more functions to this file in later steps. Setting up all imports now means you won't need to revisit the top of the file each time you add a new feature.

  • Below the imports, add the load_documents function:
def load_documents(directory: str) -> list[Document]:
    """Load all .txt files from a directory as LangChain Documents."""
    documents = []
    for filename in os.listdir(directory):
        if filename.endswith(".txt"):
            filepath = os.path.join(directory, filename)
            with open(filepath, "r") as f:
                content = f.read()
            documents.append(
                Document(page_content=content, metadata={"source": filename})
            )
    print(f"Loaded {len(documents)} documents")
    return documents

What does this code do?

  • The function scans a directory for .txt files and reads each one.
  • Each file becomes a LangChain Document object that pairs the file's text (page_content) with metadata tracking which file it came from.
  • This metadata travels with every chunk later, so your RAG system can cite its sources.
  • Save rag_pipeline.py.

Seeing a red underline on an import?

Cursor may show import errors if your virtual environment isn't selected as the Python interpreter. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) and search for Python: Select Interpreter, then choose the interpreter inside your virtual environment.

help me fix import errors in my RAG pipeline file

Configure the text splitter and chunk your documents

Documents in their raw form are too large for a vector search to work well. The RecursiveCharacterTextSplitter breaks them into smaller chunks at natural boundaries like paragraphs and sentences.

  • In rag_pipeline.py, add the chunk_documents function below load_documents:
def chunk_documents(
    documents: list[Document], chunk_size: int = 1000, chunk_overlap: int = 200
) -> list[Document]:
    """Split documents into chunks using RecursiveCharacterTextSplitter."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks (size={chunk_size}, overlap={chunk_overlap})")
    for i, chunk in enumerate(chunks[:3]):
        print(f"  Chunk {i}: {len(chunk.page_content)} chars from {chunk.metadata['source']}")
    return chunks

What does this code do?

  • chunk_size=1000 means each chunk holds up to 1000 characters of text.
  • chunk_overlap=200 means consecutive chunks share 200 characters at their boundaries, so context isn't lost at the seams.
  • The function prints the first 3 chunks so you can verify the splitter is working correctly.

Why these specific values?

A chunk size of 1000 characters is a common starting point for RAG systems. It's large enough to hold a meaningful paragraph but small enough to keep retrieval precise.

The 200-character overlap prevents the splitter from cutting an important sentence in half and losing that sentence from both neighboring chunks.

  • Save rag_pipeline.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import chromadb


def load_documents(directory: str) -> list[Document]:
    """Load all .txt files from a directory as LangChain Documents."""
    documents = []
    for filename in os.listdir(directory):
        if filename.endswith(".txt"):
            filepath = os.path.join(directory, filename)
            with open(filepath, "r") as f:
                content = f.read()
            documents.append(
                Document(page_content=content, metadata={"source": filename})
            )
    print(f"Loaded {len(documents)} documents")
    return documents


def chunk_documents(
    documents: list[Document], chunk_size: int = 1000, chunk_overlap: int = 200
) -> list[Document]:
    """Split documents into chunks using RecursiveCharacterTextSplitter."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks (size={chunk_size}, overlap={chunk_overlap})")
    for i, chunk in enumerate(chunks[:3]):
        print(f"  Chunk {i}: {len(chunk.page_content)} chars from {chunk.metadata['source']}")
    return chunks
Run the pipeline and inspect your chunks

Time to see the splitter in action. You'll call both functions and examine the output to understand how your documents get broken apart.

  • Run the following command in your terminal to load and chunk the documents:
python -c "from rag_pipeline import load_documents, chunk_documents; docs = load_documents('./documents'); chunks = chunk_documents(docs)"

You should see output similar to this:

Loaded 4 documents
Split into 8 chunks (size=1000, overlap=200)
  Chunk 0: 970 chars from ai_engineering.txt
  Chunk 1: 938 chars from vector_databases.txt
  Chunk 2: 856 chars from chunking_strategies.txt

What should I see?

The exact chunk count and character counts may differ slightly depending on how the splitter finds natural boundaries in your text. The important thing is that you see all 4 documents loaded and multiple chunks created from them.

Notice how each chunk preserves its source metadata. This is what lets the RAG system cite where an answer came from.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated before running the command. You should see the environment name in parentheses at the start of your terminal prompt.

If you see an error about rag_pipeline, confirm your terminal is in the project root directory (the folder containing rag_pipeline.py).

help me fix the ModuleNotFoundError when running my RAG pipeline

Your documents are loaded and chunked. Next up, you'll turn these chunks into vector embeddings and store them in ChromaDB so they can be searched by meaning.

Embed and Store in ChromaDB

You have a set of document chunks ready to go. The next challenge is making them searchable by meaning, not just by keywords.

To do this, you will convert each chunk into a vector embedding. This is a list of numbers that captures the semantic meaning of the text. You will store these embeddings in ChromaDB, a vector database that lets you find the most relevant chunks for any question by comparing how close their embeddings are in vector space.

In this step, get ready to:
  • Initialize OpenAI embeddings with the text-embedding-3-small model.
  • Create a persistent ChromaDB vector store populated with document chunk embeddings.
  • Verify retrieval by running a similarity search query.
Create the vector store function

Each document chunk needs to be converted into a numerical vector and stored somewhere you can search it later. The create_vector_store function handles this entire pipeline: initializing the embedding model, creating a ChromaDB collection, and adding each chunk with its embedding.

  • In rag_pipeline.py, find the end of the chunk_documents function (after the return chunks line).
  • Add the following create_vector_store function below it:
def create_vector_store(
    chunks: list[Document], collection_name: str = "rag_documents"
) -> chromadb.Collection:
    """Create a ChromaDB collection and add document chunks with embeddings."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    client = chromadb.PersistentClient(path="./chroma_db")

    # Delete existing collection if it exists
    try:
        client.delete_collection(name=collection_name)
    except ValueError:
        pass

    collection = client.create_collection(name=collection_name)

    # Embed and add documents
    for i, chunk in enumerate(chunks):
        embedding = embeddings.embed_query(chunk.page_content)
        collection.add(
            ids=[f"chunk_{i}"],
            embeddings=[embedding],
            documents=[chunk.page_content],
            metadatas=[chunk.metadata],
        )

    print(f"Added {len(chunks)} chunks to ChromaDB collection '{collection_name}'")
    return collection

What does this code do?

OpenAIEmbeddings(model="text-embedding-3-small") initializes the embedding model. This model converts text into 1536-dimensional vectors at a cost of $0.02 per million tokens.

PersistentClient(path="./chroma_db") creates a ChromaDB instance that saves data to disk. This means your embeddings survive between script runs.

The try/except block deletes any existing collection with the same name so you start fresh each time. The loop then embeds each chunk individually and adds it to the collection with its text content and metadata (source filename).

  • Save rag_pipeline.py.

Seeing a build error when importing chromadb?

On macOS, ChromaDB may fail to build the hnswlib dependency. Fix this by running:

xcode-select --install

If that does not resolve it, try setting the environment variable export HNSWLIB_NO_NATIVE=1 before reinstalling chromadb.

help me fix a ChromaDB hnswlib build error on macOS

Add the similarity search function

Now that chunks are stored as vectors, you need a way to query them. The similarity_search function takes a question, embeds it into the same vector space, and finds the closest document chunks by distance.

  • In rag_pipeline.py, add the following function below create_vector_store:
def similarity_search(
    collection: chromadb.Collection, query: str, k: int = 3
) -> list[dict]:
    """Search the vector store for similar documents."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    query_embedding = embeddings.embed_query(query)

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=k,
        include=["documents", "metadatas", "distances"],
    )

    search_results = []
    for i in range(len(results["documents"][0])):
        search_results.append(
            {
                "content": results["documents"][0][i],
                "source": results["metadatas"][0][i]["source"],
                "distance": results["distances"][0][i],
            }
        )
    return search_results

What does this code do?

The function embeds the user's query using the same text-embedding-3-small model as the documents. This puts the query into the same vector space so distances are meaningful.

collection.query() finds the k nearest chunks by cosine distance. Lower distance means higher relevance. The function returns a clean list of dictionaries with the chunk content, source filename, and distance score.

  • Save rag_pipeline.py.
Verify your vector store

Time to see the full pipeline in action. You will load your documents, chunk them, embed them into ChromaDB, and run a similarity search to confirm everything works.

  • Run this test in your terminal to verify the vector store:
python3 << 'EOF'
from rag_pipeline import load_documents, chunk_documents, create_vector_store, similarity_search

docs = load_documents("./documents")
chunks = chunk_documents(docs)
collection = create_vector_store(chunks)

print("\nSimilarity search: 'What is RAG?'")
results = similarity_search(collection, "What is RAG?")
for i, r in enumerate(results):
    print(f"  Result {i+1} ({r['source']}): distance={r['distance']:.4f}")
    print(f"    Preview: {r['content'][:100]}...")
EOF

You should see output showing your documents loaded, chunks created, chunks added to ChromaDB, and three search results ranked by distance. The results closest to zero are the most relevant to the query "What is RAG?"

What should I see?

Your output will show the chunks sorted by distance from lowest (most relevant) to highest. The ai_engineering.txt source should appear near the top because it directly discusses RAG architecture.

The distance values represent how semantically close each chunk is to your query. A distance near 0 means very similar. A distance near 1 or above means less related.

Not seeing any results or getting an error?

Make sure your OPENAI_API_KEY environment variable is still set in your current terminal session. If you opened a new terminal, you need to export it again.

Check that you are running the command from inside your project directory where rag_pipeline.py and the documents/ folder exist.

help me debug my ChromaDB similarity search returning an error

✔️ Awesome, I've got everything!

Great. Double check you have saved rag_pipeline.py with both new functions.

ⓧ I'd like to double check the full code

Here is the full rag_pipeline.py as it should look at this point:

import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import chromadb


def load_documents(directory: str) -> list[Document]:
    """Load all .txt files from a directory as LangChain Documents."""
    documents = []
    for filename in os.listdir(directory):
        if filename.endswith(".txt"):
            filepath = os.path.join(directory, filename)
            with open(filepath, "r") as f:
                content = f.read()
            documents.append(
                Document(page_content=content, metadata={"source": filename})
            )
    print(f"Loaded {len(documents)} documents")
    return documents


def chunk_documents(
    documents: list[Document], chunk_size: int = 1000, chunk_overlap: int = 200
) -> list[Document]:
    """Split documents into chunks using RecursiveCharacterTextSplitter."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks (size={chunk_size}, overlap={chunk_overlap})")
    for i, chunk in enumerate(chunks[:3]):
        print(f"  Chunk {i}: {len(chunk.page_content)} chars from {chunk.metadata['source']}")
    return chunks


def create_vector_store(
    chunks: list[Document], collection_name: str = "rag_documents"
) -> chromadb.Collection:
    """Create a ChromaDB collection and add document chunks with embeddings."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    client = chromadb.PersistentClient(path="./chroma_db")

    # Delete existing collection if it exists
    try:
        client.delete_collection(name=collection_name)
    except ValueError:
        pass

    collection = client.create_collection(name=collection_name)

    # Embed and add documents
    for i, chunk in enumerate(chunks):
        embedding = embeddings.embed_query(chunk.page_content)
        collection.add(
            ids=[f"chunk_{i}"],
            embeddings=[embedding],
            documents=[chunk.page_content],
            metadatas=[chunk.metadata],
        )

    print(f"Added {len(chunks)} chunks to ChromaDB collection '{collection_name}'")
    return collection


def similarity_search(
    collection: chromadb.Collection, query: str, k: int = 3
) -> list[dict]:
    """Search the vector store for similar documents."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    query_embedding = embeddings.embed_query(query)

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=k,
        include=["documents", "metadatas", "distances"],
    )

    search_results = []
    for i in range(len(results["documents"][0])):
        search_results.append(
            {
                "content": results["documents"][0][i],
                "source": results["metadatas"][0][i]["source"],
                "distance": results["distances"][0][i],
            }
        )
    return search_results

Your documents are now embedded and searchable. Next up, you will wire a retriever and an LLM together into a complete RAG chain that generates grounded answers from these chunks.

Build the RAG Chain

You have a vector store full of document embeddings and a working similarity search. The missing piece is connecting retrieval to generation.

In this step, you will build the complete RAG chain. It retrieves relevant context from your vector store and passes it to GPT-4o-mini with instructions to only answer from that context. The result is grounded answers with source citations.

In this step, get ready to:
  • Create a retriever that fetches the top-3 most relevant chunks for any question.
  • Write a prompt template that instructs the LLM to answer only from provided context and cite sources.
  • Wire the full pipeline together and test it with sample questions.
Create a retriever and prompt template

A retriever is the component that takes a user question, converts it to an embedding, and fetches the most relevant chunks from your vector store. The prompt template then tells the LLM exactly how to use that retrieved context.

  • In rag_pipeline.py, scroll to the bottom of the file (below the similarity_search function).
  • Add the build_rag_chain function by pasting the following code:
def build_rag_chain(collection: chromadb.Collection, k: int = 3):
    """Build a RAG chain that retrieves context and generates grounded answers."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = ChatPromptTemplate.from_template(
        """You are a helpful assistant that answers questions based only on the provided context.
If the context does not contain enough information to answer the question, say "I don't have enough information to answer that question."
Always cite which source document(s) your answer comes from.

Context:
{context}

Question: {question}

Answer:"""
    )

    def retrieve_context(question: str) -> str:
        query_embedding = embeddings.embed_query(question)
        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=k,
            include=["documents", "metadatas"],
        )
        context_parts = []
        for i in range(len(results["documents"][0])):
            source = results["metadatas"][0][i]["source"]
            content = results["documents"][0][i]
            context_parts.append(f"[Source: {source}]\n{content}")
        return "\n\n---\n\n".join(context_parts)

    chain = (
        {"context": lambda x: retrieve_context(x["question"]), "question": lambda x: x["question"]}
        | prompt
        | llm
        | StrOutputParser()
    )

    return chain

What does this function do?

This function has three key parts working together:

  • The retriever (retrieve_context) embeds the user's question, queries ChromaDB for the top k results, and formats each chunk with its source filename.
  • The prompt template tells the LLM to only use provided context, cite sources, and say "I don't know" when the context is insufficient. This prevents hallucination.
  • The chain uses LangChain's LCEL pipe syntax to connect retrieval, prompting, the LLM, and output parsing into one callable sequence.

Why temperature=0?

Setting temperature=0 makes the LLM's output deterministic. For a Q&A system grounded in documents, you want consistent, factual answers rather than creative variation.

Wire the full pipeline together and test it

Now you need a main() function that runs the entire pipeline from document loading through to answering questions. This orchestrates every function you have built so far.

  • Below the build_rag_chain function, add the main() function and the entry-point guard by pasting this code:
def main():
    # Step 2: Load and chunk documents
    print("=" * 60)
    print("STEP 1: Loading and chunking documents")
    print("=" * 60)
    documents = load_documents("./documents")
    chunks = chunk_documents(documents)

    # Step 3: Embed and store in ChromaDB
    print("\n" + "=" * 60)
    print("STEP 2: Embedding and storing in ChromaDB")
    print("=" * 60)
    collection = create_vector_store(chunks)

    # Verify with a test search
    print("\nTest similarity search: 'What is RAG?'")
    results = similarity_search(collection, "What is RAG?")
    for i, result in enumerate(results):
        print(f"  Result {i+1} (distance: {result['distance']:.4f}): {result['source']}")
        print(f"    Preview: {result['content'][:100]}...")

    # Step 4: Build and test the RAG chain
    print("\n" + "=" * 60)
    print("STEP 3: Building and testing RAG chain")
    print("=" * 60)
    chain = build_rag_chain(collection)

    test_questions = [
        "What is the most common production LLM pattern?",
        "What are the advantages of ChromaDB over other vector databases?",
        "What is a good starting configuration for chunk size and overlap?",
        "How should RAG prompts handle cases where the context lacks relevant information?",
    ]

    for question in test_questions:
        print(f"\nQ: {question}")
        answer = chain.invoke({"question": question})
        print(f"A: {answer}")
        print("-" * 40)


if __name__ == "__main__":
    main()

What does this code do?

  • main() calls every function in sequence: load documents, chunk them, embed into ChromaDB, run a test similarity search, then build the RAG chain.
  • The 4 test questions cover different documents and difficulty levels. Each answer should cite the specific source file it drew from.
  • if __name__ == "__main__": ensures main() only runs when you execute the file directly, not when importing from it.
  • Save rag_pipeline.py.
  • Run the full pipeline by executing this command in your terminal:
python rag_pipeline.py
  • You should see three sections of output: document loading, ChromaDB embedding, and the RAG chain answering all 4 test questions with source citations.

What should I see?

Each answer should reference a specific source document (like ai_engineering.txt or chunking_strategies.txt). The answers are grounded in your documents rather than the LLM's general knowledge.

If a question has no relevant context in the documents, the model should respond with "I don't have enough information to answer that question."

Seeing an error when running the script?

  • If you see AuthenticationError, check that your OPENAI_API_KEY environment variable is still set in your current terminal session.
  • If you see IndentationError, make sure the retrieve_context function is indented inside build_rag_chain. It is a nested inner function.
  • If you see NameError for build_rag_chain, confirm it is defined above the main() function in the file.

Help me debug my RAG chain errors.

✔️ Awesome, I've got everything!

Great. Double check you have saved rag_pipeline.py and that all 4 questions returned grounded answers with source citations.

ⓧ I'd like to double check the full code

Here is the complete rag_pipeline.py file. Compare it with yours to spot any differences.

import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import chromadb


def load_documents(directory: str) -> list[Document]:
    """Load all .txt files from a directory as LangChain Documents."""
    documents = []
    for filename in os.listdir(directory):
        if filename.endswith(".txt"):
            filepath = os.path.join(directory, filename)
            with open(filepath, "r") as f:
                content = f.read()
            documents.append(
                Document(page_content=content, metadata={"source": filename})
            )
    print(f"Loaded {len(documents)} documents")
    return documents


def chunk_documents(
    documents: list[Document], chunk_size: int = 1000, chunk_overlap: int = 200
) -> list[Document]:
    """Split documents into chunks using RecursiveCharacterTextSplitter."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split into {len(chunks)} chunks (size={chunk_size}, overlap={chunk_overlap})")
    for i, chunk in enumerate(chunks[:3]):
        print(f"  Chunk {i}: {len(chunk.page_content)} chars from {chunk.metadata['source']}")
    return chunks


def create_vector_store(
    chunks: list[Document], collection_name: str = "rag_documents"
) -> chromadb.Collection:
    """Create a ChromaDB collection and add document chunks with embeddings."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    client = chromadb.PersistentClient(path="./chroma_db")

    # Delete existing collection if it exists
    try:
        client.delete_collection(name=collection_name)
    except ValueError:
        pass

    collection = client.create_collection(name=collection_name)

    # Embed and add documents
    for i, chunk in enumerate(chunks):
        embedding = embeddings.embed_query(chunk.page_content)
        collection.add(
            ids=[f"chunk_{i}"],
            embeddings=[embedding],
            documents=[chunk.page_content],
            metadatas=[chunk.metadata],
        )

    print(f"Added {len(chunks)} chunks to ChromaDB collection '{collection_name}'")
    return collection


def similarity_search(
    collection: chromadb.Collection, query: str, k: int = 3
) -> list[dict]:
    """Search the vector store for similar documents."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    query_embedding = embeddings.embed_query(query)

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=k,
        include=["documents", "metadatas", "distances"],
    )

    search_results = []
    for i in range(len(results["documents"][0])):
        search_results.append(
            {
                "content": results["documents"][0][i],
                "source": results["metadatas"][0][i]["source"],
                "distance": results["distances"][0][i],
            }
        )
    return search_results


def build_rag_chain(collection: chromadb.Collection, k: int = 3):
    """Build a RAG chain that retrieves context and generates grounded answers."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = ChatPromptTemplate.from_template(
        """You are a helpful assistant that answers questions based only on the provided context.
If the context does not contain enough information to answer the question, say "I don't have enough information to answer that question."
Always cite which source document(s) your answer comes from.

Context:
{context}

Question: {question}

Answer:"""
    )

    def retrieve_context(question: str) -> str:
        query_embedding = embeddings.embed_query(question)
        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=k,
            include=["documents", "metadatas"],
        )
        context_parts = []
        for i in range(len(results["documents"][0])):
            source = results["metadatas"][0][i]["source"]
            content = results["documents"][0][i]
            context_parts.append(f"[Source: {source}]\n{content}")
        return "\n\n---\n\n".join(context_parts)

    chain = (
        {"context": lambda x: retrieve_context(x["question"]), "question": lambda x: x["question"]}
        | prompt
        | llm
        | StrOutputParser()
    )

    return chain


def main():
    # Step 2: Load and chunk documents
    print("=" * 60)
    print("STEP 1: Loading and chunking documents")
    print("=" * 60)
    documents = load_documents("./documents")
    chunks = chunk_documents(documents)

    # Step 3: Embed and store in ChromaDB
    print("\n" + "=" * 60)
    print("STEP 2: Embedding and storing in ChromaDB")
    print("=" * 60)
    collection = create_vector_store(chunks)

    # Verify with a test search
    print("\nTest similarity search: 'What is RAG?'")
    results = similarity_search(collection, "What is RAG?")
    for i, result in enumerate(results):
        print(f"  Result {i+1} (distance: {result['distance']:.4f}): {result['source']}")
        print(f"    Preview: {result['content'][:100]}...")

    # Step 4: Build and test the RAG chain
    print("\n" + "=" * 60)
    print("STEP 3: Building and testing RAG chain")
    print("=" * 60)
    chain = build_rag_chain(collection)

    test_questions = [
        "What is the most common production LLM pattern?",
        "What are the advantages of ChromaDB over other vector databases?",
        "What is a good starting configuration for chunk size and overlap?",
        "How should RAG prompts handle cases where the context lacks relevant information?",
    ]

    for question in test_questions:
        print(f"\nQ: {question}")
        answer = chain.invoke({"question": question})
        print(f"A: {answer}")
        print("-" * 40)


if __name__ == "__main__":
    main()

Your RAG system is fully functional. It loads documents, chunks them, embeds them into a vector store, retrieves relevant context, and generates grounded answers with source citations. Next up, the Secret Mission challenges you to measure how well your retrieval actually works.

Secret mission

Add Retrieval Evaluation

Your RAG system answers questions, but how do you know it's retrieving the right context? In this secret mission, you'll build a retrieval evaluation pipeline that measures how well your system finds relevant chunks. You'll create a golden test set, calculate retrieval recall, and compare different configurations to find the optimal setup.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. OpenAI API charges only apply when you actively run the scripts. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Project folder containing rag_pipeline.py, evaluate_retrieval.py, setup_test.py, requirements.txt, and the documents/ folder.
  • Python virtual environment (inside your project folder).
  • ChromaDB persistent stores at ./chroma_db and ./chroma_db_eval (inside your project folder).
  • The OPENAI_API_KEY environment variable set in your terminal session.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your RAG system, testing new documents, or tweaking chunking parameters.

  • Your project folder stays on your Desktop with all files intact.
  • Your ChromaDB stores remain ready for instant queries.
  • Your virtual environment stays activated whenever you return to the project folder.
  • No ongoing costs are incurred unless you run the scripts again.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files and ChromaDB stores remain on disk for when you return.

  • Deactivate your virtual environment by running:
deactivate
  • When you come back, navigate to your project folder and reactivate the virtual environment with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).
  • You will need to re-export your OPENAI_API_KEY environment variable in your new terminal session.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and the ChromaDB stores from your machine.

  • Deactivate your virtual environment first (if it is still active):
deactivate
  • Navigate up to your Desktop by running:
cd ~/Desktop
  • Delete the entire project folder (macOS/Linux) by running:
rm -rf rag-document-qa
  • On Windows, run this instead:
rmdir /s /q rag-document-qa
  • Unset the OPENAI_API_KEY environment variable from your current terminal session (macOS/Linux):
unset OPENAI_API_KEY
  • On Windows, run:
set OPENAI_API_KEY=
  • If you added OPENAI_API_KEY to your shell profile (e.g. .zshrc or .bashrc), remove that line manually to fully clean up.
  • This removes all Python files, documents, ChromaDB vector stores, and the virtual environment. Nothing else on your machine is affected.

Nice Work!

Nice Work!

Nice work! You just built a complete Retrieval-Augmented Generation system from scratch. Your pipeline ingests documents, chunks and embeds them, stores them in a vector database, retrieves relevant context, and generates grounded answers with source citations.

You've learned how to:

  • Built a complete RAG pipeline using LangChain that loads documents, splits them into chunks, and generates grounded answers from retrieved context.
  • Used ChromaDB as a persistent vector store to embed document chunks with OpenAI's text-embedding-3-small model and retrieve them using semantic similarity search.
  • Engineered a grounded generation prompt that instructs the LLM to answer only from retrieved context and cite its sources, preventing hallucination.
  • Secret Mission: Measured retrieval recall using a golden test set and compared how different chunking configurations affect retrieval quality.

Ready to quiz yourself?