Build a RAG Pipeline with ChromaDB

Build a Python RAG pipeline that answers questions from your own documents.

Introduction

30 Second Summary

Ever wished you could ask questions about your own notes, documents, or company files and get accurate answers instantly? That is exactly what AI-powered document Q&A tools do behind the scenes.

In this project, you will build a RAG (Retrieval-Augmented Generation) pipeline from scratch using Python, ChromaDB, and Claude. Your pipeline will load text documents, store them in a vector database, and answer questions grounded in your own data.

What You'll Build

A command-line Python application that chunks documents, embeds them in a vector database, and uses Claude to generate answers based on retrieved context.

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

  • A working RAG pipeline that loads, chunks, and embeds your text documents into ChromaDB.
  • A semantic search retrieval function that finds the most relevant chunks for any question.
  • An interactive Q&A loop that sends retrieved context to Claude and prints grounded answers.
  • Secret Mission: Add source citations so every answer traces back to the exact document and chunk it came from.

Are there any prerequisites?

You need basic Python knowledge (variables, functions, and loops). No AI or machine learning experience is required.

Set Up Your Environment

Before you can build your RAG pipeline, you need a Python environment with the right packages installed. Your pipeline depends on two libraries: ChromaDB for storing and searching document embeddings, and the Anthropic SDK for sending queries to Claude.

In this step, you will install Python, create an isolated project environment, install your dependencies, and connect to the Claude API.

In this step, get ready to:
  • Install Python and set up a virtual environment.
  • Install the ChromaDB and Anthropic packages.
  • Get an Anthropic API key and set it as an environment variable.
Install Python 3.9+ and create your project folder

The Anthropic Python SDK requires Python 3.9 or higher. Let's confirm you have the right version installed.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows) and typing Terminal.
  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.9 or higher

You are good to go. Move on to creating your project folder below.

ⓧ I see an older version

Your system has a Python version below 3.9. You need to install a newer version.

  • Download and install Python 3.11 from the official site: https://www.python.org/downloads/.
  • On macOS, the system default may still resolve to the old version after installing. Use python3.11 instead of python3 if python3 --version still shows the old version.
  • After installation, close and reopen your terminal, then run the version check again.

ⓧ Command not found

Python is not installed on your machine.

  • Download and install Python 3.11 from the official site: https://www.python.org/downloads/.
  • On Windows, make sure to check the box that says Add Python to PATH during installation.
  • After installation, close and reopen your terminal, then run python3 --version again to confirm it works.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir rag-pipeline
cd rag-pipeline

A virtual environment isolates your project's packages from other Python projects on your machine. This prevents version conflicts between projects.

  • Create a virtual environment by running this command:
python3 -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

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

Don't see (venv) in your prompt?

Make sure you ran the activation command from inside the rag-pipeline folder.

On Windows, if you get an execution policy error, run this command first:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

Then try the activate command again.

Help me activate my Python virtual environment

Install the required packages

Your RAG pipeline needs two packages: ChromaDB for the vector database and the Anthropic SDK for communicating with Claude.

  • Install both packages by running this command:
pip install chromadb anthropic

What does this command do?

pip install chromadb installs ChromaDB, a vector database that stores your document chunks as embeddings and lets you search them by meaning.

pip install anthropic installs the official Anthropic Python SDK so you can send prompts to Claude and receive generated answers.

The installation may take a minute as it downloads dependencies. When it finishes, you should see a success message in your terminal listing the installed packages.

Why the first run takes extra time

ChromaDB bundles a sentence-transformer model (all-MiniLM-L6-v2) for generating embeddings. The first time you run your pipeline later, it may download this model (~90MB). This is normal and only happens once.

Seeing installation errors?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt before running pip install.

If you closed your terminal since the last substep, navigate back to your rag-pipeline folder and activate the virtual environment again.

Help me fix my pip install error

Get your Anthropic API key

To call Claude from your Python code, you need an API key. The Anthropic SDK reads this key from an environment variable so you never have to hard-code it into your script.

  • Open your browser and go to console.anthropic.com.
  • Create an account or sign in if you already have one.
  • Navigate to Settings, then API keys.
  • Click Create Key.
  • Give it a name like rag-pipeline and click Create Key.
  • Copy the key immediately. You will not be able to see it again after closing this dialog.

What is an API key?

An API key is a secret string that identifies your application to a service. When your script sends a request to Claude, the key proves you have permission to use the API.

  • Back in your terminal (with (venv) still showing), set the environment variable:

macOS

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

Windows

$env:ANTHROPIC_API_KEY="your-api-key-here"
  • Replace your-api-key-here with the key you just copied from the console.

Why an environment variable?

Storing secrets in environment variables keeps them out of your code files. If you ever share your script or push it to GitHub, the key stays safe on your machine instead of leaking into version control.

Keep this terminal open

The environment variable only lasts for this terminal session. If you close your terminal, you will need to run the export command again before running your script.

Not sure if the key is set?

You can verify by running echo $ANTHROPIC_API_KEY (macOS) or echo $env:ANTHROPIC_API_KEY (Windows PowerShell). You should see your key printed in the terminal.

Help me set my Anthropic API key

Your environment is ready. Next up, you will create sample documents that your RAG pipeline will search through.

Create Sample Documents

Your environment is ready. Now you need something for the RAG pipeline to actually retrieve from.

A RAG pipeline is only as good as the documents it has access to. In this step, you will create a small collection of text files covering different topics. Having distinct topics makes it easy to verify that retrieval is working correctly later.

In this step, get ready to:
  • Create a documents/ folder in your project directory.
  • Add three sample text files with distinct topics.
  • Verify your folder structure is correct.
Create the documents folder

Your RAG pipeline will read all .txt files from a dedicated folder. Creating a separate folder keeps your documents organized and makes it easy for the script to find them.

  • Make sure you are inside your project folder in the terminal. Create the documents folder by running this command:
mkdir documents
  • Confirm the folder exists by running:
ls

You should see documents listed in the output.

Create sample text files with distinct topics

You will create three text files covering different subjects. Each file represents a different type of document your pipeline might encounter in the real world. Having distinct topics helps you verify later that the retrieval step pulls the right document for each question.

  • In your code editor, create a new file called company_faq.txt inside the documents/ folder.
  • Paste the following content into the file:
NextWork is an online learning platform that helps people build real-world tech projects. The platform offers guided projects in cloud computing, AI, web development, and more. Each project takes between 30 and 90 minutes to complete.

NextWork was founded in 2023 and is based in San Francisco. The platform is free to use for basic projects, with premium projects available through a subscription. Students earn certificates upon completing projects.

The most popular projects on NextWork cover AWS, Python, and machine learning. Projects are designed by industry professionals and updated regularly to reflect current best practices.

Why this file?

This file describes a company and its offerings. It contains facts like founding year, location, and product details. These are the kinds of specific facts that RAG retrieval is great at surfacing.

  • Save company_faq.txt.
  • Create a second file called product_guide.txt inside the documents/ folder.
  • Paste the following content into the file:
The CloudSync Pro is a file synchronization tool designed for small teams. It automatically syncs files across devices in real-time, with end-to-end encryption for security.

Key features of CloudSync Pro include selective sync, version history for 90 days, and integration with Slack and Microsoft Teams. The tool supports files up to 10GB each and offers 1TB of storage per team member.

Pricing starts at $8 per user per month for the Starter plan (up to 5 users). The Business plan is $15 per user per month with unlimited users and priority support. Both plans include a 14-day free trial.

CloudSync Pro is available on macOS, Windows, Linux, iOS, and Android. The desktop app requires at least 500MB of free disk space and 4GB of RAM.

Why this file?

This file contains product specifications, pricing, and system requirements. It is the kind of document where you would ask questions like "How much does the Business plan cost?" and expect a precise answer from the retrieval system.

  • Save product_guide.txt.
  • Create a third file called remote_work_policy.txt inside the documents/ folder.
  • Paste the following content into the file:
All full-time employees are eligible for remote work up to 3 days per week. Remote work days must be scheduled in advance using the HR portal by the previous Friday.

Employees working remotely must be available during core hours (10 AM to 3 PM local time) and maintain a reliable internet connection of at least 25 Mbps. A dedicated workspace free from distractions is required.

Equipment provided for remote work includes a laptop, external monitor, keyboard, and mouse. Employees are responsible for their own internet service. A monthly stipend of $75 is provided to offset internet and utility costs.

Managers may require in-office presence for team meetings, client presentations, or collaborative projects with at least 48 hours notice. Exceptions to the 3-day limit require VP-level approval.

Why this file?

Policy documents contain rules, requirements, and specific numbers (like stipend amounts or notice periods). This is a common use case for RAG in enterprise settings where employees need quick answers about company policies.

  • Save remote_work_policy.txt.

Use your own documents

You can replace these sample files with your own .txt files if you prefer. Any plain text content works. Just make sure you have at least 2-3 files with different topics so you can test retrieval later.

Not sure where to create the files?

The three .txt files go inside the documents/ folder you just created. In your code editor, right-click the documents folder in the sidebar and select **New File** to create each one.

help me find where to create my text files in my code editor

Verify your folder structure

Before moving on, confirm that all three files are in the right place. Your Python script will look for .txt files inside the documents/ folder. If they are somewhere else, retrieval will not work.

  • List the contents of your documents/ folder by running this command:
ls documents/

You should see three files listed: company_faq.txt, product_guide.txt, and remote_work_policy.txt.

Don't see all three files?

Make sure you saved each file after pasting the content. Check that the files are inside documents/ and not in the project root folder. The filenames must end in .txt for the pipeline to find them.

help me check my folder structure

✔️ Awesome, I've got everything!

Great. Make sure all three files are saved before moving on.

ⓧ I'd like to double check the full code

Here is the complete content for each file. Compare yours against these:

NextWork is an online learning platform that helps people build real-world tech projects. The platform offers guided projects in cloud computing, AI, web development, and more. Each project takes between 30 and 90 minutes to complete.

NextWork was founded in 2023 and is based in San Francisco. The platform is free to use for basic projects, with premium projects available through a subscription. Students earn certificates upon completing projects.

The most popular projects on NextWork cover AWS, Python, and machine learning. Projects are designed by industry professionals and updated regularly to reflect current best practices.
The CloudSync Pro is a file synchronization tool designed for small teams. It automatically syncs files across devices in real-time, with end-to-end encryption for security.

Key features of CloudSync Pro include selective sync, version history for 90 days, and integration with Slack and Microsoft Teams. The tool supports files up to 10GB each and offers 1TB of storage per team member.

Pricing starts at $8 per user per month for the Starter plan (up to 5 users). The Business plan is $15 per user per month with unlimited users and priority support. Both plans include a 14-day free trial.

CloudSync Pro is available on macOS, Windows, Linux, iOS, and Android. The desktop app requires at least 500MB of free disk space and 4GB of RAM.
All full-time employees are eligible for remote work up to 3 days per week. Remote work days must be scheduled in advance using the HR portal by the previous Friday.

Employees working remotely must be available during core hours (10 AM to 3 PM local time) and maintain a reliable internet connection of at least 25 Mbps. A dedicated workspace free from distractions is required.

Equipment provided for remote work includes a laptop, external monitor, keyboard, and mouse. Employees are responsible for their own internet service. A monthly stipend of $75 is provided to offset internet and utility costs.

Managers may require in-office presence for team meetings, client presentations, or collaborative projects with at least 48 hours notice. Exceptions to the 3-day limit require VP-level approval.

Your documents are ready. Next up, you will write Python code to load these files and split them into chunks that the vector database can store.

Load and Chunk Documents

Your documents folder is ready with real content to query against. Now you need to turn those raw text files into something a vector database can work with.

An LLM can only process a limited amount of text at once. If you sent entire documents as context, you would waste tokens on irrelevant paragraphs and risk hitting the model's context limit. The solution is to split documents into small, overlapping chunks so only the most relevant pieces get sent to the model later.

In this step, get ready to:
  • Write a function that reads all text files from your documents folder.
  • Write a chunking function that splits text into overlapping pieces.
  • Run the script to verify your documents load and chunk correctly.
Write the document loading function

The loading function scans your documents/ folder, reads every .txt file, and returns the content along with filenames so you can track where each piece of text originated.

  • Create a new file called rag.py in your project folder (the same folder that contains your documents/ directory).
  • Add the imports and configuration constants by pasting this code into rag.py:
import os
import chromadb
import anthropic


# --- Configuration ---
DOCS_DIR = "./documents"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "my_documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
MODEL = "claude-haiku-4-5"
TOP_K = 3

What do these constants do?

  • DOCS_DIR points to the folder where your text files live.
  • CHUNK_SIZE and CHUNK_OVERLAP control how documents get split. Each chunk is 500 characters with 50 characters of overlap between consecutive chunks.
  • MODEL is the Claude model you will use for generating answers (claude-haiku-4-5 is fast and inexpensive).
  • TOP_K sets how many chunks to retrieve when answering a question.
  • Now add the load_documents() function below the configuration constants:
# --- Document Loading ---
def load_documents(docs_dir):
    """Load all .txt files from the documents directory."""
    documents = []
    for filename in sorted(os.listdir(docs_dir)):
        if filename.endswith(".txt"):
            filepath = os.path.join(docs_dir, filename)
            with open(filepath, "r") as f:
                text = f.read().strip()
            documents.append({"filename": filename, "text": text})
            print(f"Loaded: {filename} ({len(text)} characters)")
    return documents

What does this function do?

  • sorted(os.listdir(docs_dir)) lists all files in the documents folder in alphabetical order so results are consistent across runs.
  • The .endswith(".txt") filter ensures only text files get loaded. Any other files in the folder are ignored.
  • Each document is stored as a dictionary with filename and text keys so you always know which file the content came from.
  • Save rag.py.

Seeing a syntax error?

  • Make sure the function is indented consistently. Python requires all lines inside a function to use the same indentation (4 spaces is standard).
  • Check that the f-string uses matching quotes. The outer quotes around the f-string must be different from any quotes used inside it.

Help me fix a syntax error in my load_documents function in rag.py.

Write the chunking function

A single document might be thousands of characters long. When you search for an answer later, you want to retrieve only the most relevant paragraph, not the entire document. Chunking solves this by splitting each document into smaller pieces.

The overlap between chunks ensures that sentences spanning a chunk boundary are not lost. Without overlap, a key sentence sitting right at the split point could be cut in half.

  • Add the chunk_text() function below load_documents() in rag.py:
# --- Chunking ---
def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """Split text into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start += size - overlap
    return chunks

What does this function do?

  • The while loop moves a window across the text. Each iteration grabs 500 characters starting from start.
  • After each chunk, start advances by size - overlap (450 characters). This means each new chunk re-includes the last 50 characters of the previous chunk.
  • The final chunk may be shorter than 500 characters. That is fine because Python slicing does not raise an error when end exceeds the string length.
  • Save rag.py.

Not sure if your function is correct?

Double check that the line start += size - overlap uses += (not =). Using plain = would reset start to the same value each loop, causing an infinite loop.

Help me verify my chunk_text function is working correctly.

Run the script and verify your output

Your functions are defined but nothing calls them yet. You will add temporary test code to the bottom of the file, run the script, and then remove it.

  • Add this temporary test code at the very bottom of rag.py:
# --- Test (remove after verifying) ---
if __name__ == "__main__":
    docs = load_documents(DOCS_DIR)
    print(f"\nTotal documents: {len(docs)}")
    for doc in docs:
        chunks = chunk_text(doc["text"])
        print(f"  {doc['filename']}: {len(chunks)} chunks")

What does this test code do?

The if __name__ == "__main__" block runs only when you execute the file directly (not when importing from it). It calls both functions and prints the results so you can confirm everything works.

  • Save rag.py.
  • Run the script by entering this command in your terminal:
python rag.py

You should see output showing each file loaded with its character count, followed by the chunk count for each document. Each of your three documents should produce 2 chunks.

What should I see?

Your terminal should display something like:

Loaded: company_faq.txt (559 characters)
Loaded: product_guide.txt (656 characters)
Loaded: remote_work_policy.txt (700 characters)

Total documents: 3
  company_faq.txt: 2 chunks
  product_guide.txt: 2 chunks
  remote_work_policy.txt: 2 chunks

Your character counts may differ slightly depending on how you typed your sample documents. The important thing is that all three files load and each produces at least 2 chunks.

Seeing 'No module named chromadb'?

This means your virtual environment is not active. Activate it before running the script. On macOS/Linux, run source venv/bin/activate. On Windows, run venv\Scripts\activate.

Help me fix a module import error when running rag.py.

Seeing 'No .txt files found' or zero documents?

  • Make sure your documents/ folder is in the same directory as rag.py.
  • Confirm your files end in .txt (not .txt.txt or .text).
  • Run your script from the project folder, not from inside the documents/ folder.

Help me debug why my RAG script isn't finding my text files.

  • Once you have confirmed the output looks correct, delete the test code block (the # --- Test (remove after verifying) --- section and everything below it) from rag.py.
  • Save rag.py.

✔️ Awesome, I've got everything!

Great. Double check that you have removed the test code and saved your file.

ⓧ I'd like to double check the full code

Here is the complete rag.py as it should look at the end of this step:

import os
import chromadb
import anthropic


# --- Configuration ---
DOCS_DIR = "./documents"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "my_documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
MODEL = "claude-haiku-4-5"
TOP_K = 3


# --- Document Loading ---
def load_documents(docs_dir):
    """Load all .txt files from the documents directory."""
    documents = []
    for filename in sorted(os.listdir(docs_dir)):
        if filename.endswith(".txt"):
            filepath = os.path.join(docs_dir, filename)
            with open(filepath, "r") as f:
                text = f.read().strip()
            documents.append({"filename": filename, "text": text})
            print(f"Loaded: {filename} ({len(text)} characters)")
    return documents


# --- Chunking ---
def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """Split text into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start += size - overlap
    return chunks

Your script can now load and chunk documents. Next up, you will store these chunks in a ChromaDB vector database so you can search them by meaning instead of exact keywords.

Store Chunks in ChromaDB

In the last step, you split your documents into overlapping chunks. Those chunks are just strings in memory right now.

To make them searchable, you need to store them in a vector database. A vector database converts text into numerical representations called embeddings and lets you find chunks that are semantically similar to a question. ChromaDB handles this automatically using a built-in embedding model, so you don't need any extra setup.

In this step, get ready to:
  • Initialize a persistent ChromaDB database that saves to disk.
  • Store all document chunks with metadata tracking their source.
  • Verify the database contains the correct number of chunks.
Initialize a ChromaDB persistent client and create a collection

ChromaDB can run in-memory (data disappears when the script ends) or persist to disk. You want your embeddings saved between runs so you don't re-process documents every time you ask a question.

  • In rag.py, find the chunk_text function. Below it, add this new ingest_documents function:
# --- Ingestion ---
def ingest_documents():
    """Load documents, chunk them, and store in ChromaDB."""
    documents = load_documents(DOCS_DIR)
    if not documents:
        print("No .txt files found in the documents/ folder.")
        return

    # Connect to a persistent database on disk
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)

    # Delete existing collection if it exists, then create fresh
    try:
        client.delete_collection(COLLECTION_NAME)
    except ValueError:
        pass

    collection = client.create_collection(name=COLLECTION_NAME)

What does this code do?

This is the first half of the ingest_documents() function. It handles setting up the database before any chunks are added.

  • load_documents(DOCS_DIR) calls the function you wrote in the previous step to read all .txt files.
  • PersistentClient(path=CHROMA_DB_PATH) creates a database that saves to the chroma_db/ folder on disk.
  • The try/except block deletes any existing collection with the same name so you start fresh each time. If the collection doesn't exist yet, ChromaDB raises a ValueError which the code safely ignores.
  • create_collection() makes a new empty collection ready to receive your chunks.

What is a collection?

A collection in ChromaDB is like a table in a traditional database. It groups related documents together and indexes their embeddings for fast similarity search.

ChromaDB uses the all-MiniLM-L6-v2 embedding model by default. This model converts each chunk into a 384-dimensional vector that captures its meaning. No extra setup or API key is needed for this.

Add all chunks to the collection with metadata

Now you need to loop through each document, chunk it, and add every chunk to the collection. Each chunk gets a unique ID and metadata so you can trace it back to its source file later.

  • Still inside the ingest_documents() function, add this code directly below the collection = client.create_collection(name=COLLECTION_NAME) line:
    all_ids = []
    all_documents = []
    all_metadatas = []

    for doc in documents:
        chunks = chunk_text(doc["text"])
        print(f"  {doc['filename']}: {len(chunks)} chunks")
        for i, chunk in enumerate(chunks):
            # Create a unique ID combining filename and chunk position
            all_ids.append(f"{doc['filename']}::chunk_{i}")
            all_documents.append(chunk)
            all_metadatas.append({
                "source": doc["filename"],
                "chunk_index": i
            })

    collection.add(
        ids=all_ids,
        documents=all_documents,
        metadatas=all_metadatas
    )

    print(f"\nIngested {len(all_ids)} chunks into collection '{COLLECTION_NAME}'")
    return collection

What does this code do?

This is the second half of ingest_documents(). It builds three parallel lists and adds everything to ChromaDB in one batch.

  • all_ids stores unique identifiers in the format filename::chunk_0, filename::chunk_1, etc.
  • all_documents stores the actual chunk text. ChromaDB automatically generates embeddings for each one when you call collection.add().
  • all_metadatas stores the source filename and chunk index for each chunk. This is how you trace an answer back to its original document.
  • Save rag.py.

Seeing indentation errors?

The code from both chunks above must be indented inside the ingest_documents() function body. Make sure every line after def ingest_documents(): is indented by exactly 4 spaces.

Help me fix an indentation error in my ingest_documents function.

Verify the database was created

Time to run the ingestion and confirm ChromaDB stored your chunks correctly.

  • Run the ingestion function by entering this command in your terminal:
python -c "import rag; rag.ingest_documents()"

You should see output listing each loaded file, its chunk counts, and a final line confirming the total number of chunks ingested.

What should I see?

Your terminal output will look something like this:

Loaded: company_faq.txt (568 characters)
Loaded: product_guide.txt (658 characters)
Loaded: remote_work_policy.txt (724 characters)
  company_faq.txt: 2 chunks
  product_guide.txt: 2 chunks
  remote_work_policy.txt: 2 chunks

Ingested 6 chunks into collection 'my_documents'

Your exact chunk counts may differ slightly depending on how your text files are formatted. The important thing is that the total matches the sum of all individual file chunks.

Why does the first run take longer?

On the first run, ChromaDB downloads the all-MiniLM-L6-v2 embedding model (about 90MB). This is a one-time download. Future runs will start immediately.

  • Confirm the chroma_db/ folder was created in your project directory by running:
ls

You should see chroma_db, documents, and rag.py in the output.

Not seeing any output?

Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt.

If you see a ModuleNotFoundError for chromadb, reactivate your virtual environment and confirm the package is installed.

Help me troubleshoot my ChromaDB ingestion not producing output.

✔️ Awesome, I've got everything!

Great. Make sure you've saved rag.py before moving on.

ⓧ I'd like to double check the full code

Here is the full rag.py file as it should look at the end of this step:

import os
import chromadb
import anthropic


# --- Configuration ---
DOCS_DIR = "./documents"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "my_documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
MODEL = "claude-haiku-4-5"
TOP_K = 3


# --- Document Loading ---
def load_documents(docs_dir):
    """Load all .txt files from the documents directory."""
    documents = []
    for filename in sorted(os.listdir(docs_dir)):
        if filename.endswith(".txt"):
            filepath = os.path.join(docs_dir, filename)
            with open(filepath, "r") as f:
                text = f.read().strip()
            documents.append({"filename": filename, "text": text})
            print(f"Loaded: {filename} ({len(text)} characters)")
    return documents


# --- Chunking ---
def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """Split text into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start += size - overlap
    return chunks


# --- Ingestion ---
def ingest_documents():
    """Load documents, chunk them, and store in ChromaDB."""
    documents = load_documents(DOCS_DIR)
    if not documents:
        print("No .txt files found in the documents/ folder.")
        return

    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)

    # Delete existing collection if it exists, then create fresh
    try:
        client.delete_collection(COLLECTION_NAME)
    except ValueError:
        pass

    collection = client.create_collection(name=COLLECTION_NAME)

    all_ids = []
    all_documents = []
    all_metadatas = []

    for doc in documents:
        chunks = chunk_text(doc["text"])
        print(f"  {doc['filename']}: {len(chunks)} chunks")
        for i, chunk in enumerate(chunks):
            all_ids.append(f"{doc['filename']}::chunk_{i}")
            all_documents.append(chunk)
            all_metadatas.append({
                "source": doc["filename"],
                "chunk_index": i
            })

    collection.add(
        ids=all_ids,
        documents=all_documents,
        metadatas=all_metadatas
    )

    print(f"\nIngested {len(all_ids)} chunks into collection '{COLLECTION_NAME}'")
    return collection

Your documents are now embedded and searchable. Next up, you'll write the query pipeline that retrieves relevant chunks and sends them to Claude for answer generation.

Build the Query Pipeline

Your documents are chunked and stored as vector embeddings in ChromaDB. Now you need a way to actually use them.

In this step, you'll build the two core functions that make RAG work: a retrieval function that finds the most relevant chunks for a question, and a generation function that sends those chunks to Claude as context for answering.

In this step, get ready to:
  • Write a retrieval function that searches ChromaDB for relevant chunks.
  • Write a generation function that sends context to Claude.
  • Test the full pipeline with a real question about your documents.
Write the retrieval function

When a user asks a question, the pipeline needs to find the chunks most likely to contain the answer. ChromaDB handles this by comparing the question's embedding to every stored chunk embedding and returning the closest matches.

  • In rag.py, scroll to the bottom of the file (below the ingest_documents() function) and add the following retrieval function:
# --- Retrieval ---
def retrieve(question, top_k=TOP_K):
    """Retrieve the most relevant chunks for a question."""
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
    collection = client.get_collection(name=COLLECTION_NAME)

    results = collection.query(
        query_texts=[question],
        n_results=top_k
    )

    return results["documents"][0], results["metadatas"][0]

What does this code do?

This function connects to your existing ChromaDB database and retrieves the collection you created in the previous step.

The collection.query() call does the heavy lifting. It converts your question into an embedding (using the same all-MiniLM-L6-v2 model), compares it against every stored chunk embedding, and returns the top 3 closest matches. The [0] index is needed because ChromaDB returns results nested in a list (one set per query text).

  • Save rag.py.

Why semantic search instead of keyword search?

Traditional keyword search requires an exact word match. Semantic search compares meaning instead. A question like "What does NextWork teach?" will match chunks containing "guided projects in cloud computing, AI, web development" even though none of those words appear in the question.

Write the generation function

Retrieval alone just gives you raw text chunks. The generation function takes those chunks, formats them as context, and asks Claude to answer the question using only that context.

  • In rag.py, add the following function below your retrieve() function:
# --- Generation ---
def generate_answer(question, context_chunks, metadatas):
    """Send the question and retrieved context to Claude."""
    client = anthropic.Anthropic()

    context = "\n\n---\n\n".join([
        f"[Chunk from {meta['source']}]\n{chunk}"
        for chunk, meta in zip(context_chunks, metadatas)
    ])

    message = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"""Use the following context to answer the question. If the context does not contain enough information to answer, say so clearly.

Context:
{context}

Question: {question}"""
            }
        ]
    )

    return message.content[0].text

What does this code do?

The function first builds a formatted context string by joining all retrieved chunks together, separated by --- dividers. Each chunk is labelled with its source filename so Claude knows where each piece of information came from.

The prompt instructs Claude to answer based only on the provided context. This is what makes RAG answers grounded. If the context doesn't contain the answer, Claude will say so instead of making something up.

The anthropic.Anthropic() client automatically reads your ANTHROPIC_API_KEY environment variable. No need to pass it explicitly.

  • Save rag.py.

Why answer only from context?

Without the "answer based only on the provided context" instruction, Claude would blend its general knowledge with the retrieved chunks. This defeats the purpose of RAG. You want answers that come exclusively from your documents so you can trust and verify them.

✔️ Awesome, I've got everything!

Great. Double check you've saved rag.py before moving on to testing.

ⓧ I'd like to double check the full code

Here is what your full rag.py should look like at this point:

import os
import chromadb
import anthropic


# --- Configuration ---
DOCS_DIR = "./documents"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "my_documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
MODEL = "claude-haiku-4-5"
TOP_K = 3


# --- Document Loading ---
def load_documents(docs_dir):
    """Load all .txt files from the documents directory."""
    documents = []
    for filename in sorted(os.listdir(docs_dir)):
        if filename.endswith(".txt"):
            filepath = os.path.join(docs_dir, filename)
            with open(filepath, "r") as f:
                text = f.read().strip()
            documents.append({"filename": filename, "text": text})
            print(f"Loaded: {filename} ({len(text)} characters)")
    return documents


# --- Chunking ---
def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """Split text into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start += size - overlap
    return chunks


# --- Ingestion ---
def ingest_documents():
    """Load documents, chunk them, and store in ChromaDB."""
    documents = load_documents(DOCS_DIR)
    if not documents:
        print("No .txt files found in the documents/ folder.")
        return

    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)

    try:
        client.delete_collection(COLLECTION_NAME)
    except ValueError:
        pass

    collection = client.create_collection(name=COLLECTION_NAME)

    all_ids = []
    all_documents = []
    all_metadatas = []

    for doc in documents:
        chunks = chunk_text(doc["text"])
        print(f"  {doc['filename']}: {len(chunks)} chunks")
        for i, chunk in enumerate(chunks):
            all_ids.append(f"{doc['filename']}::chunk_{i}")
            all_documents.append(chunk)
            all_metadatas.append({
                "source": doc["filename"],
                "chunk_index": i
            })

    collection.add(
        ids=all_ids,
        documents=all_documents,
        metadatas=all_metadatas
    )

    print(f"\nIngested {len(all_ids)} chunks into collection '{COLLECTION_NAME}'")
    return collection


# --- Retrieval ---
def retrieve(question, top_k=TOP_K):
    """Retrieve the most relevant chunks for a question."""
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
    collection = client.get_collection(name=COLLECTION_NAME)

    results = collection.query(
        query_texts=[question],
        n_results=top_k
    )

    return results["documents"][0], results["metadatas"][0]


# --- Generation ---
def generate_answer(question, context_chunks, metadatas):
    """Send the question and retrieved context to Claude."""
    client = anthropic.Anthropic()

    context = "\n\n---\n\n".join([
        f"[Chunk from {meta['source']}]\n{chunk}"
        for chunk, meta in zip(context_chunks, metadatas)
    ])

    message = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"""Use the following context to answer the question. If the context does not contain enough information to answer, say so clearly.

Context:
{context}

Question: {question}"""
            }
        ]
    )

    return message.content[0].text
Test the query pipeline

Both functions are in place. Time to run them together and see if Claude can answer a question using your documents as context.

  • Test your pipeline by running this command in your terminal:
python -c "from rag import retrieve, generate_answer; chunks, metas = retrieve('What is NextWork?'); print('Sources:', [m['source'] for m in metas]); print(); print(generate_answer('What is NextWork?', chunks, metas))"

What should I see?

You should see a list of retrieved sources (showing which document chunks were found relevant), followed by Claude's answer. The answer should reference details from your company_faq.txt file, like NextWork being an online learning platform, offering guided projects, and being founded in 2023.

If the answer mentions specific facts from your documents (project duration, location, topics covered), your pipeline is working correctly. Claude is answering from your data, not from its general training.

  • Verify that the answer references specific details from your documents rather than generic information.

Seeing an authentication error?

If you see a message about an invalid API key or missing authentication, your ANTHROPIC_API_KEY environment variable may not be set in this terminal session. Re-run the export command from Step 1.

If you see a "collection not found" error, run your ingestion script again from the previous step to repopulate the database.

Help me debug my RAG pipeline

Your retrieval and generation functions are working together. Next up, you'll wrap them in an interactive loop so you can ask multiple questions without rerunning the script each time.

Add an Interactive Query Loop

You have all the building blocks of a RAG pipeline: loading, chunking, storing, retrieving, and generating.

Right now, each piece works individually but nothing ties them together into a usable tool. In this step, you will wrap everything in an interactive loop so you can ask questions and see answers in real time.

In this step, get ready to:
  • Write a main function that initializes the database and accepts user input.
  • Display retrieved context chunks alongside each answer.
  • Test the full pipeline with multiple questions.
Write the interactive loop

The main() function is the entry point for your RAG pipeline. It checks whether documents have already been ingested, then enters a loop that waits for your questions.

  • In rag.py, scroll to the bottom of the file (below the generate_answer function) and add the following code:
# --- Interactive Loop ---
def main():
    # Check if collection exists, if not ingest
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
    collections = [c.name for c in client.list_collections()]

    if COLLECTION_NAME not in collections:
        print("No existing database found. Ingesting documents...\n")
        ingest_documents()
    else:
        collection = client.get_collection(name=COLLECTION_NAME)
        print(f"Found existing database with {collection.count()} chunks.\n")

    print("RAG Pipeline Ready! Ask questions about your documents.")
    print("Type 'quit' to exit, 'reingest' to reload documents.\n")

    while True:
        question = input("Question: ").strip()

        if not question:
            continue
        if question.lower() == "quit":
            print("Goodbye!")
            break
        if question.lower() == "reingest":
            ingest_documents()
            print()
            continue

What does this code do?

The function first connects to your existing ChromaDB database and checks if your collection already exists. If not, it automatically ingests your documents so you never run into a "collection not found" error.

The while True loop keeps the program running until you type quit. Typing reingest reloads your documents if you edit them.

Add the context display and answer generation

When a valid question comes in, the pipeline needs to retrieve relevant chunks, show you what context it found, and then generate an answer. Displaying the retrieved chunks lets you verify that the system is pulling the right information.

  • Still in rag.py, add the following code directly below the continue line (still inside the while True loop):
        # Retrieve
        chunks, metadatas = retrieve(question)

        print("\n--- Retrieved Context ---")
        for i, (chunk, meta) in enumerate(zip(chunks, metadatas)):
            print(f"[{i+1}] {meta['source']} (chunk {meta['chunk_index']})")
            print(f"    {chunk[:100]}...")
            print()

        # Generate
        answer = generate_answer(question, chunks, metadatas)

        print("--- Answer ---")
        print(answer)
        print()


if __name__ == "__main__":
    main()

What does this code do?

The retrieve(question) call queries ChromaDB for the top 3 most similar chunks to your question. The display loop prints each chunk's source file, chunk index, and a 100-character preview so you can see exactly what context Claude will receive.

The generate_answer() call sends those chunks plus your question to Claude, and the answer prints below. The if __name__ == "__main__" block ensures main() only runs when you execute the script directly.

  • Save rag.py.

Seeing an indentation error?

The retrieve, print, and generate lines must be indented inside the while True block (two levels of indentation: one for the function, one for the loop). Make sure the if __name__ block is at the top level with no indentation.

help me fix an indentation error in my Python RAG script

Test the pipeline with questions

Now you will run the complete pipeline and ask it questions about your documents. You will also test what happens when you ask something the documents cannot answer.

  • Run your RAG pipeline by executing this command in your terminal:
python rag.py
  • You should see a message like Found existing database with X chunks. followed by RAG Pipeline Ready!.
  • At the Question: prompt, type a question your documents can answer:
What is CloudSync Pro?
  • You should see the --- Retrieved Context --- section showing chunks from product_guide.txt, followed by a --- Answer --- grounded in that content.
  • Now ask a question that spans multiple documents:
What equipment do remote workers get?
  • You should see chunks retrieved from remote_work_policy.txt with details about the laptop, monitor, and stipend.
  • Finally, test with a question the documents cannot answer:
What is the capital of France?
  • Claude should indicate that the context does not contain enough information to answer this question. This proves your pipeline is grounding answers in your documents instead of relying on general knowledge.
  • Type quit to exit the pipeline.

Why does grounding matter?

An LLM without retrieval will happily answer "What is the capital of France?" from its training data. Your RAG pipeline restricts Claude to only use the context you provide. This is exactly how production document Q&A tools prevent hallucination.

Getting an API error?

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

help me debug an API authentication error in my RAG pipeline

✔️ Awesome, I've got everything!

Great. Double check you have saved rag.py with both the main() function and the if __name__ entry point.

ⓧ I'd like to double check the full code

Here is the complete rag.py file:

import os
import chromadb
import anthropic


# --- Configuration ---
DOCS_DIR = "./documents"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "my_documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
MODEL = "claude-haiku-4-5"
TOP_K = 3


# --- Document Loading ---
def load_documents(docs_dir):
    """Load all .txt files from the documents directory."""
    documents = []
    for filename in sorted(os.listdir(docs_dir)):
        if filename.endswith(".txt"):
            filepath = os.path.join(docs_dir, filename)
            with open(filepath, "r") as f:
                text = f.read().strip()
            documents.append({"filename": filename, "text": text})
            print(f"Loaded: {filename} ({len(text)} characters)")
    return documents


# --- Chunking ---
def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """Split text into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start += size - overlap
    return chunks


# --- Ingestion ---
def ingest_documents():
    """Load documents, chunk them, and store in ChromaDB."""
    documents = load_documents(DOCS_DIR)
    if not documents:
        print("No .txt files found in the documents/ folder.")
        return

    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)

    # Delete existing collection if it exists, then create fresh
    try:
        client.delete_collection(COLLECTION_NAME)
    except ValueError:
        pass

    collection = client.create_collection(name=COLLECTION_NAME)

    all_ids = []
    all_documents = []
    all_metadatas = []

    for doc in documents:
        chunks = chunk_text(doc["text"])
        print(f"  {doc['filename']}: {len(chunks)} chunks")
        for i, chunk in enumerate(chunks):
            all_ids.append(f"{doc['filename']}::chunk_{i}")
            all_documents.append(chunk)
            all_metadatas.append({
                "source": doc["filename"],
                "chunk_index": i
            })

    collection.add(
        ids=all_ids,
        documents=all_documents,
        metadatas=all_metadatas
    )

    print(f"\nIngested {len(all_ids)} chunks into collection '{COLLECTION_NAME}'")
    return collection


# --- Retrieval ---
def retrieve(question, top_k=TOP_K):
    """Retrieve the most relevant chunks for a question."""
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
    collection = client.get_collection(name=COLLECTION_NAME)

    results = collection.query(
        query_texts=[question],
        n_results=top_k
    )

    return results["documents"][0], results["metadatas"][0]


# --- Generation ---
def generate_answer(question, context_chunks, metadatas):
    """Send the question and retrieved context to Claude."""
    client = anthropic.Anthropic()

    context = "\n\n---\n\n".join([
        f"[Chunk from {meta['source']}]\n{chunk}"
        for chunk, meta in zip(context_chunks, metadatas)
    ])

    message = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"""Use the following context to answer the question. If the context does not contain enough information to answer, say so clearly.

Context:
{context}

Question: {question}"""
            }
        ]
    )

    return message.content[0].text


# --- Interactive Loop ---
def main():
    # Check if collection exists, if not ingest
    client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
    collections = [c.name for c in client.list_collections()]

    if COLLECTION_NAME not in collections:
        print("No existing database found. Ingesting documents...\n")
        ingest_documents()
    else:
        collection = client.get_collection(name=COLLECTION_NAME)
        print(f"Found existing database with {collection.count()} chunks.\n")

    print("RAG Pipeline Ready! Ask questions about your documents.")
    print("Type 'quit' to exit, 'reingest' to reload documents.\n")

    while True:
        question = input("Question: ").strip()

        if not question:
            continue
        if question.lower() == "quit":
            print("Goodbye!")
            break
        if question.lower() == "reingest":
            ingest_documents()
            print()
            continue

        # Retrieve
        chunks, metadatas = retrieve(question)

        print("\n--- Retrieved Context ---")
        for i, (chunk, meta) in enumerate(zip(chunks, metadatas)):
            print(f"[{i+1}] {meta['source']} (chunk {meta['chunk_index']})")
            print(f"    {chunk[:100]}...")
            print()

        # Generate
        answer = generate_answer(question, chunks, metadatas)

        print("--- Answer ---")
        print(answer)
        print()


if __name__ == "__main__":
    main()

Your RAG pipeline is fully functional. You can ask questions, see which chunks get retrieved, and get answers grounded in your own documents. Next up, the Secret Mission challenges you to add source citations so every answer points back to its origin.

Secret mission

Add Source Citations

Your RAG pipeline retrieves context and generates answers, but it does not tell you where each piece of information came from. In this secret mission, you will add numbered source citations so every claim in Claude's answer traces back to a specific document and chunk.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally (aside from the Claude API calls), so there are no ongoing costs unless you actively run the script.

Resources you used:

  • Project folder on your Desktop (contains rag.py, documents/, and chroma_db/).
  • Python virtual environment (inside your project folder).
  • ANTHROPIC_API_KEY environment variable (session-based).
  • Anthropic API key (stored on the Claude Console).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your RAG pipeline or want to add more documents later.

  • Your project files and ChromaDB database will remain on your Desktop.
  • Your Anthropic API key stays active on the Claude Console. You will only be charged when you actively run the script and make API calls.
  • The ANTHROPIC_API_KEY environment variable will disappear when you close your terminal session (unless you added it to your shell profile).

Pause - I'll come back to this later

Shut down running processes and remove the environment variable to prevent accidental API calls. Your files stay on disk so you can pick up where you left off.

  • If the interactive loop is running in your terminal, type quit to exit.
  • Unset the environment variable by running:
unset ANTHROPIC_API_KEY
  • If you added the export line to your shell profile (~/.zshrc or ~/.bashrc), remove that line as well.
  • Your project folder and ChromaDB database remain intact for next time.

Delete - I don't want to use this again

Remove all project resources and revoke your API key so nothing remains.

  • Unset the environment variable by running:
unset ANTHROPIC_API_KEY
  • If you added the export line to your shell profile (~/.zshrc or ~/.bashrc), remove that line as well.
  • Delete your project folder. On macOS/Linux, run:
rm -rf ~/Desktop/rag-pipeline
  • On Windows, run:
rmdir /s /q %USERPROFILE%\Desktop\rag-pipeline
  • Revoke your Anthropic API key so it can no longer be used. Open the Claude Console at platform.claude.com, navigate to your API keys settings, and disable or delete the key you created for this project.

Nice Work!

Nice Work!

You just built a RAG pipeline from scratch. You now have a working system that loads your own documents, stores them as vector embeddings, and generates accurate answers grounded in your data.

You've learned how to:

  • Chunk documents and store vector embeddings in a ChromaDB database for semantic search.
  • Built a retrieval function that finds the most relevant chunks for any question.
  • Connected retrieved context to Claude for grounded answer generation using Retrieval-Augmented Generation.
  • Secret Mission: Added source citations so every claim traces back to its origin document.

Ready to quiz yourself?