Build a Document Q&A Bot with RAG
Build a RAG pipeline that answers questions from your own documents.
Introduction
30 Second Summary
You have dozens of documents sitting in a folder. Finding a specific answer means opening each one, skimming, and hoping you spot the right paragraph. What if you could just ask a question and get an accurate, sourced answer in seconds?
In this project, you will build a Document Q&A Bot that uses Retrieval-Augmented Generation (RAG) to answer questions grounded in your own text files. You'll create a complete pipeline that loads, chunks, embeds, stores, retrieves, and generates answers using OpenAI and ChromaDB.
What You'll Build
A command-line RAG pipeline that takes natural-language questions about your documents and returns grounded answers using semantic search (vector embeddings + ChromaDB) and OpenAI's chat model.
By the end of this project, you'll have:
- A working RAG pipeline that loads text documents, chunks them, embeds them, and stores them in a vector database.
- A local ChromaDB vector store populated with your embedded document data that persists between sessions.
- A clear understanding of each stage of the RAG pipeline: Load, Chunk, Embed, Store, Retrieve, Generate.
- Secret Mission: Add metadata filtering and source citations to your answers.
Are there any prerequisites?
You should have basic Python knowledge (functions, loops, file I/O) and an OpenAI API key with some credit on your account. The project costs approximately $0.05 in API calls for a small document set.
Before We Start
Before We Start...
Set Up Your Environment
Before your bot can answer questions from documents, it needs a workspace with the right tools installed.
In this step, you will create a project folder, install three Python packages, and prepare sample documents for your bot to read.
In this step, get ready to:
- Create a project directory with a Python virtual environment.
- Install the required packages and configure your API key.
- Add sample documents and verify everything works.
Create your project directory and virtual environment
A virtual environment isolates your project's packages from the rest of your system. This prevents version conflicts between different Python projects on your machine.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir rag-bot
cd rag-bot- Create a virtual environment by running:
macOS
python3 -m venv venvWindows
python -m venv venv- Activate the virtual environment by running:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt.
What does the (venv) prefix mean?
The (venv) prefix tells you that any pip install commands will install packages inside this project's folder only. Your system Python stays untouched.
- Install the required packages by running:
pip install openai==2.43.0 chromadb==1.5.9 python-dotenvWhat are these three packages?
openai is the official Python SDK for calling OpenAI's embedding and chat APIs.
chromadb is a vector database that stores your document chunks alongside their embeddings so you can search by meaning.
python-dotenv loads environment variables from a .env file so you never hard-code secrets into your scripts.
The install may take a minute. ChromaDB pulls in several dependencies.
Seeing an installation error?
Make sure the (venv) prefix is visible in your terminal. If it is not, re-run the activation command above.
If you see a permissions error, do NOT use sudo. Deactivate, delete the venv folder, and recreate it.
Help me troubleshoot my pip install error
Set up your environment file and sample documents
Your bot needs an OpenAI API key to generate embeddings and answers. A .env file keeps this key out of your code.
- In your code editor, create a new file called .env in the rag-bot folder.
- Paste the following into .env:
OPENAI_API_KEY=your-api-key-here- Replace your-api-key-here with your actual OpenAI API key (it starts with sk-).
Where do I find my OpenAI API key?
Go to platform.openai.com/api-keys and copy an existing key or create a new one. You need credit on your account for API calls to work.
- Create a documents folder inside rag-bot by running:
mkdir documents- Add 2-3 sample text files inside the documents folder. Create each file in your code editor and save it with a .txt extension.
What to put in your sample documents
Copy and paste 2-3 paragraphs of text into each file. Good options include a Wikipedia article section, meeting notes, a product FAQ, or any text you would like to ask questions about later.
Name them descriptively, such as python_overview.txt or company_faq.txt. The filenames become metadata your bot uses to track sources.
- Confirm your project structure by running:
lsYou should see .env, documents, and venv listed in the output.
Don't see .env in the list?
Files starting with a dot are hidden by default on macOS and Linux. Run ls -a to show hidden files.
If the file is truly missing, make sure you saved it inside the rag-bot folder and not one level up.
Help me find my .env file
Verify your setup
A quick verification script confirms all three packages import correctly and shows you which versions are installed.
- Create a file called verify_setup.py in the rag-bot folder and paste the following code:
import openai
import chromadb
import dotenv
# Print installed versions to confirm correct packages
print(f"openai: {openai.__version__}")
print(f"chromadb: {chromadb.__version__}")
print(f"python-dotenv: {dotenv.__version__}")
print("\nAll packages installed successfully!")What does this script do?
Each import line tries to load a package. If any package is missing or broken, Python raises an ImportError immediately.
The __version__ attribute prints the exact version number so you can confirm the pinned versions installed correctly.
- Save verify_setup.py and run it:
macOS
python3 verify_setup.pyWindows
python verify_setup.pyYou should see output similar to this:
openai: 2.43.0
chromadb: 1.5.9
python-dotenv: 1.2.2
All packages installed successfully!Seeing a ModuleNotFoundError?
This means the package did not install into the active virtual environment. Check that (venv) appears in your terminal prompt.
If it does not, activate the virtual environment again and re-run the pip install command from Substep 1.
Help me fix my ModuleNotFoundError
Your environment is ready. Next up, you will write the code that loads your text files and splits them into chunks your bot can search through.
Load and Chunk Your Documents
Your environment is ready with all the packages installed and your sample documents waiting in the documents/ folder. Now it's time to tackle the first real challenge of any RAG system: getting your documents into a format the AI can search through.
Raw text files are too large and unstructured for effective semantic search. You need to break them into smaller, meaningful pieces called chunks. In this step, you'll write the loading and chunking logic that forms the foundation of your pipeline.
In this step, get ready to:
- Write a function to load all text files from your documents folder.
- Write a chunking function that splits text into overlapping pieces.
- Test the pipeline to confirm chunks are sized and overlapping correctly.
Write a function to load your documents
The first stage of any RAG pipeline is loading your source data. You need a function that scans your documents/ folder and reads every .txt file into memory along with its metadata.
- Create a new file called rag_bot.py in your project folder by running:
touch rag_bot.py- Confirm the file exists by running:
lsYou should see rag_bot.py listed alongside your documents/ folder and .env file.
- Open rag_bot.py in your code editor and add the following imports and configuration:
import os
import sys
import glob
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
COLLECTION_NAME = "documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50What does this code do?
- load_dotenv() reads your .env file so the OPENAI_API_KEY environment variable is available.
- OpenAI() creates an API client that automatically reads OPENAI_API_KEY from the environment.
- PersistentClient initializes a ChromaDB database that saves to disk at ./chroma_db. This means your embedded documents survive between script runs.
- The constants define which models you'll use and how big your chunks will be.
- Below the configuration, add the load_documents() function:
def load_documents(directory="documents"):
"""Load all .txt files from the specified directory."""
documents = []
txt_files = glob.glob(os.path.join(directory, "*.txt"))
for file_path in txt_files:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"filename": os.path.basename(file_path),
"path": file_path,
})
print(f"Loaded {len(documents)} document(s) from '{directory}/'")
return documentsWhat does this function do?
- glob.glob(os.path.join(directory, "*.txt")) finds every file ending in .txt inside the specified folder.
- For each file, it reads the full text content and stores it in a dictionary along with the filename and path.
- The function returns a list of these document dictionaries, which the rest of your pipeline will process.
- Save rag_bot.py.
Seeing an import error?
Make sure your virtual environment is activated. If you see a module not found error for chromadb or openai, your terminal session may have lost the virtual environment activation.
Help me fix my import errors in rag_bot.py
Write a function to chunk your text
Large Language Models have limited context windows. Sending an entire document as context is inefficient and often impossible. Text chunking solves this by splitting documents into smaller overlapping pieces that can be individually searched and retrieved.
Why use overlapping chunks?
Overlap ensures that important information sitting at the boundary between two chunks is not lost. If a key sentence spans the break between chunk A and chunk B, the overlap keeps that sentence intact in at least one of the chunks.
Your chunking strategy uses a paragraph-first approach. It tries to split on paragraph boundaries (double newlines) first to preserve natural text structure. Only when a single paragraph exceeds the chunk size does it fall back to word-level splitting.
- In rag_bot.py, add the chunk_text() function below load_documents():
def chunk_text(text, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
"""Split text into overlapping chunks."""
chunks = []
# First try splitting on double newlines (paragraphs)
paragraphs = text.split("\n\n")
current_chunk = ""
for paragraph in paragraphs:
if len(current_chunk) + len(paragraph) + 2 <= chunk_size:
current_chunk += paragraph + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
# If a single paragraph is longer than chunk_size, split by character
if len(paragraph) > chunk_size:
words = paragraph.split()
current_chunk = ""
for word in words:
if len(current_chunk) + len(word) + 1 <= chunk_size:
current_chunk += word + " "
else:
chunks.append(current_chunk.strip())
# Keep overlap from end of previous chunk
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + word + " "
else:
# Keep overlap from end of previous chunk
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + paragraph + "\n\n"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunksWhat does this function do?
- It splits text on double newlines first, preserving paragraph structure wherever possible.
- When adding a paragraph would exceed CHUNK_SIZE (500 characters), it saves the current chunk and starts a new one.
- If a single paragraph is longer than the chunk size, it falls back to word-level splitting.
- The overlap_text variable carries the last 50 characters from the previous chunk into the start of the next one, creating the overlap.
- Save rag_bot.py.
Seeing a syntax error?
Check that the indentation is consistent. Python requires either all tabs or all spaces (4 spaces is standard). Also verify that the chunk_text() function is at the same indentation level as load_documents() (not nested inside it).
Help me fix a syntax error in my chunk_text function
✔️ Awesome, I've got everything!
Great. Double check you've saved rag_bot.py with both functions in place.
ⓧ I'd like to double check the full code
Here is the full rag_bot.py so far:
import os
import sys
import glob
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
COLLECTION_NAME = "documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
def load_documents(directory="documents"):
"""Load all .txt files from the specified directory."""
documents = []
txt_files = glob.glob(os.path.join(directory, "*.txt"))
for file_path in txt_files:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"filename": os.path.basename(file_path),
"path": file_path,
})
print(f"Loaded {len(documents)} document(s) from '{directory}/'")
return documents
def chunk_text(text, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
"""Split text into overlapping chunks."""
chunks = []
# First try splitting on double newlines (paragraphs)
paragraphs = text.split("\n\n")
current_chunk = ""
for paragraph in paragraphs:
if len(current_chunk) + len(paragraph) + 2 <= chunk_size:
current_chunk += paragraph + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
# If a single paragraph is longer than chunk_size, split by character
if len(paragraph) > chunk_size:
words = paragraph.split()
current_chunk = ""
for word in words:
if len(current_chunk) + len(word) + 1 <= chunk_size:
current_chunk += word + " "
else:
chunks.append(current_chunk.strip())
# Keep overlap from end of previous chunk
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + word + " "
else:
# Keep overlap from end of previous chunk
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + paragraph + "\n\n"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunksTest the chunking pipeline
Before moving on, you should verify that your loading and chunking logic works correctly with your sample documents.
- Test your functions by running this command in your terminal:
python -c "
from rag_bot import load_documents, chunk_text
docs = load_documents()
for doc in docs:
chunks = chunk_text(doc['content'])
print(f'\n{doc["filename"]} -> {len(chunks)} chunks')
print(f'First chunk ({len(chunks[0])} chars):')
print(chunks[0][:200])
"You should see output showing how many documents were loaded and how many chunks each document produced. The first chunk preview confirms that real text content is being processed.
What should I see?
Your output will look something like this (with your own filenames and content):
- A line saying Loaded 2 document(s) from 'documents/' (or however many files you added).
- For each file, a chunk count (e.g., notes.txt -> 4 chunks) and a preview of the first chunk's content.
- Chunk sizes should mostly be under 500 characters. Some variation is normal because the function respects paragraph boundaries.
Seeing zero documents loaded?
Make sure your documents/ folder contains .txt files (not .pdf or other formats). Also check that you're running the command from your project's root folder where rag_bot.py lives.
Help me debug why load_documents is returning an empty list
Your loading and chunking functions are working. Next up, you'll turn these text chunks into vector embeddings and store them in ChromaDB so they become searchable.
Embed and Store in ChromaDB
Your documents are loaded and chunked into retrieval-friendly pieces. Now you need to turn those text chunks into something a computer can search through.
In this step, you will generate vector embeddings for each chunk using OpenAI's API, then store them in a ChromaDB collection. Once stored, you can search for chunks by meaning rather than exact keywords.
In this step, get ready to:
- Write a helper function that turns text into vector embeddings via the OpenAI API.
- Write an ingestion function that chunks documents, embeds them in batches, and stores everything in ChromaDB.
- Run the full ingestion pipeline and verify your local vector database is populated.
Write the get_embeddings() helper function
A vector embedding is a list of numbers that represents the meaning of a piece of text. Two similar sentences produce embeddings that are close together in vector space, which is how semantic search works.
- In rag_bot.py, add the following get_embeddings() function below your chunk_text() function:
def get_embeddings(texts):
"""Get embeddings for a list of texts using OpenAI API."""
response = client.embeddings.create(
input=texts,
model=EMBEDDING_MODEL
)
return [item.embedding for item in response.data]What does this code do?
This function takes a list of text strings and sends them to OpenAI's text-embedding-3-small model. The API returns a numerical vector (1536 dimensions) for each input text.
The EMBEDDING_MODEL constant you defined earlier points to text-embedding-3-small, and the client object reads your OPENAI_API_KEY automatically from the environment.
- Save rag_bot.py.
Seeing a NameError for client?
Make sure get_embeddings() is placed after the top-level client = OpenAI() line. The function references the global client variable, so it must be defined below it.
help me fix a NameError in my get_embeddings function
Write the embed_and_store() function
Now you need a function that ties the whole ingestion process together. It will chunk each document, generate embeddings in batches, and store everything in ChromaDB with metadata so you can trace results back to their source files.
- In rag_bot.py, add the following embed_and_store() function below get_embeddings():
def embed_and_store(documents):
"""Chunk documents, generate embeddings, and store in ChromaDB."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
all_chunks = []
all_metadatas = []
all_ids = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({
"source": doc["filename"],
"chunk_index": i,
})
all_ids.append(f"{doc['filename']}_chunk_{i}")
print(f"Created {len(all_chunks)} chunks from {len(documents)} document(s)")
print("Generating embeddings...")
# Process in batches of 100 (OpenAI limit)
batch_size = 100
all_embeddings = []
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i + batch_size]
embeddings = get_embeddings(batch)
all_embeddings.extend(embeddings)
# Store in ChromaDB
collection.add(
ids=all_ids,
documents=all_chunks,
embeddings=all_embeddings,
metadatas=all_metadatas,
)
print(f"Stored {len(all_chunks)} chunks in ChromaDB collection '{COLLECTION_NAME}'")
return collectionWhat does this code do?
This function handles the full ingestion pipeline in four stages:
- get_or_create_collection() creates the ChromaDB collection (or reuses it if it already exists).
- The loop chunks each document and builds parallel lists of chunks, metadata, and unique IDs. Each ID combines the filename and chunk index so you can trace any result back to its source.
- Embeddings are generated in batches of 100 to stay within API limits. The OpenAI embeddings API enforces a maximum of 300,000 tokens across all inputs in a single request.
- collection.add() stores the chunks, their embeddings, and metadata together in ChromaDB. ChromaDB persists everything to the ./chroma_db directory automatically.
Why batch the embeddings?
The OpenAI embeddings API has a total token limit of 300,000 tokens per request. Processing in batches of 100 keeps each request well within this limit, even for longer chunks.
For this learning project your documents are small enough to fit in a single batch. The batching pattern becomes essential when you scale to hundreds or thousands of documents.
- Save rag_bot.py.
Seeing an error about duplicate IDs?
If you run ingestion twice on the same documents, ChromaDB will reject duplicate IDs. Delete the chroma_db folder and re-run, or use collection.upsert() instead of collection.add() if you want to overwrite existing entries.
help me fix duplicate ID errors in ChromaDB
Run the ingestion pipeline end-to-end
Your functions are ready. Time to run the full pipeline and watch your documents transform into searchable vectors.
- Run the ingestion pipeline by executing this command in your terminal:
python -c "from rag_bot import load_documents, embed_and_store; docs = load_documents(); embed_and_store(docs)"What does this command do?
This one-liner imports your two functions directly from rag_bot.py, loads all text files from the documents/ folder, then chunks, embeds, and stores them in ChromaDB. You will add a proper command-line interface in a later step.
You should see output similar to this:
Loaded 3 document(s) from 'documents/'
Created 12 chunks from 3 document(s)
Generating embeddings...
Stored 12 chunks in ChromaDB collection 'documents'The exact numbers depend on how many files you placed in your documents/ folder and how long each file is.
- Confirm the chroma_db directory was created by running:
ls chroma_dbYou should see files and subdirectories that ChromaDB uses to persist your vectors.
Seeing an authentication error from OpenAI?
Make sure your .env file contains a valid OPENAI_API_KEY and that you are running the command from your project directory (where .env lives).
Also confirm your API key has credit available on the OpenAI dashboard.
help me fix an OpenAI authentication error in my RAG pipeline
✔️ Awesome, I've got everything!
Great. Double check you have saved rag_bot.py and that the chroma_db directory exists in your project folder.
ⓧ I'd like to double check the full code
import os
import sys
import glob
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
COLLECTION_NAME = "documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
def load_documents(directory="documents"):
"""Load all .txt files from the specified directory."""
documents = []
txt_files = glob.glob(os.path.join(directory, "*.txt"))
for file_path in txt_files:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"filename": os.path.basename(file_path),
"path": file_path,
})
print(f"Loaded {len(documents)} document(s) from '{directory}/'")
return documents
def chunk_text(text, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
"""Split text into overlapping chunks."""
chunks = []
paragraphs = text.split("\n\n")
current_chunk = ""
for paragraph in paragraphs:
if len(current_chunk) + len(paragraph) + 2 <= chunk_size:
current_chunk += paragraph + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
if len(paragraph) > chunk_size:
words = paragraph.split()
current_chunk = ""
for word in words:
if len(current_chunk) + len(word) + 1 <= chunk_size:
current_chunk += word + " "
else:
chunks.append(current_chunk.strip())
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + word + " "
else:
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + paragraph + "\n\n"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def get_embeddings(texts):
"""Get embeddings for a list of texts using OpenAI API."""
response = client.embeddings.create(
input=texts,
model=EMBEDDING_MODEL
)
return [item.embedding for item in response.data]
def embed_and_store(documents):
"""Chunk documents, generate embeddings, and store in ChromaDB."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
all_chunks = []
all_metadatas = []
all_ids = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({
"source": doc["filename"],
"chunk_index": i,
})
all_ids.append(f"{doc['filename']}_chunk_{i}")
print(f"Created {len(all_chunks)} chunks from {len(documents)} document(s)")
print("Generating embeddings...")
# Process in batches of 100 (OpenAI limit)
batch_size = 100
all_embeddings = []
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i + batch_size]
embeddings = get_embeddings(batch)
all_embeddings.extend(embeddings)
# Store in ChromaDB
collection.add(
ids=all_ids,
documents=all_chunks,
embeddings=all_embeddings,
metadatas=all_metadatas,
)
print(f"Stored {len(all_chunks)} chunks in ChromaDB collection '{COLLECTION_NAME}'")
return collectionYour documents are now embedded and stored in a local vector database. Next up, you will write the retrieval and generation functions that turn user questions into grounded answers.
Retrieve and Generate Answers
Your documents are chunked, embedded, and stored in ChromaDB. The hard part of building a knowledge base is done.
Now you need a way to actually use that knowledge base. When a user asks a question, your bot needs to find the most relevant chunks and pass them to an LLM that generates a grounded answer. This is the "Retrieve" and "Generate" stages of the RAG pipeline.
In this step, get ready to:
- Write a retrieval function that finds the most relevant chunks for any question.
- Write a generation function that uses retrieved context to produce grounded answers.
- Create an interactive terminal loop so you can ask your bot questions.
Write the retrieve function
When a user types a question, you need to convert that question into an embedding using the same model you used to embed your document chunks. Then you can ask ChromaDB to find the chunks whose embeddings are closest to the question's embedding.
- In rag_bot.py, scroll to the bottom of your file (below the embed_and_store() function).
- Add the retrieve() function by copying this code:
def retrieve(question, n_results=3):
"""Retrieve the most relevant chunks for a question."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
question_embedding = get_embeddings([question])[0]
results = collection.query(
query_embeddings=[question_embedding],
n_results=n_results,
)
return resultsWhat does this code do?
This function performs semantic search. It converts the user's question into a vector embedding, then asks ChromaDB to find the 3 chunks whose embeddings are most similar.
- get_or_create_collection() reconnects to the existing collection without creating a duplicate.
- get_embeddings([question])[0] embeds the question using the same text-embedding-3-small model so the vectors are comparable.
- collection.query() returns the top matches ranked by cosine similarity.
- Save rag_bot.py.
You won't see a change yet. The next chunk uses this function to build the full answer pipeline.
Write the answer generation function
Retrieval alone just gives you raw text chunks. You need an LLM to read those chunks and synthesize a coherent answer. This is where prompt engineering becomes critical. The system prompt must instruct the model to only use the provided context.
- Below the retrieve() function, add the generate_answer() function:
def generate_answer(question, context_chunks):
"""Generate an answer using retrieved context and OpenAI chat."""
context = "\n\n---\n\n".join(context_chunks)
system_prompt = """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 in my documents to answer that question."
Do not make up information that is not in the context."""
user_prompt = f"""Context:
{context}
Question: {question}
Answer based only on the context above:"""
response = client.chat.completions.create(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
temperature=0,
)
return response.choices[0].message.contentWhat does this code do?
This function constructs a prompt that keeps the LLM grounded in your documents.
- The retrieved chunks are joined with a --- separator so the model can distinguish between them.
- The system prompt explicitly says "answer based only on the provided context." This prevents hallucination.
- temperature=0 makes the output deterministic. The model picks the most likely token at each step, reducing creative improvisation.
- The model is gpt-4o-mini, which is fast and cost-effective for Q&A tasks.
Why is grounding important?
Without grounding, an LLM will confidently answer any question using its training data. With RAG, you want the model to admit when it does not have enough information rather than inventing an answer.
The system prompt's explicit instruction to say "I don't have enough information" is what makes this a reliable Q&A system instead of a general chatbot.
- Save rag_bot.py.
You won't see a change yet. The next chunk wires these functions into a usable CLI.
Create the interactive query loop
You have retrieval and generation functions, but no way to call them from the terminal. You need helper functions that orchestrate the pipeline and a command-line interface to tie everything together.
- Below generate_answer(), add the ingest() and query() helper functions:
def ingest():
"""Run the ingestion pipeline."""
documents = load_documents()
if not documents:
print("No .txt files found in 'documents/' directory.")
print("Add some .txt files and try again.")
return
embed_and_store(documents)
print("\nIngestion complete! You can now query your documents.")
def query(question):
"""Query the RAG pipeline."""
print(f"\nQuestion: {question}")
print("Retrieving relevant context...")
results = retrieve(question, n_results=3)
if not results["documents"][0]:
print("No documents found. Run 'python rag_bot.py ingest' first.")
return
context_chunks = results["documents"][0]
sources = [meta["source"] for meta in results["metadatas"][0]]
print(f"Found {len(context_chunks)} relevant chunks from: {', '.join(set(sources))}")
print("\nGenerating answer...\n")
answer = generate_answer(question, context_chunks)
print(f"Answer: {answer}")What do these functions do?
- ingest() wraps your load-chunk-embed-store pipeline into a single command. It checks that documents exist before proceeding.
- query() orchestrates the full retrieval-to-generation flow. It retrieves relevant chunks, extracts source filenames from the metadata, and prints the generated answer.
- Below the query() function, add the interactive() loop and the __main__ block:
def interactive():
"""Run an interactive query loop."""
print("\nRAG Document Q&A Bot")
print("Type your questions below. Type 'quit' to exit.\n")
while True:
question = input("You: ").strip()
if question.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not question:
continue
query(question)
print()
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage:")
print(" python rag_bot.py ingest - Load and embed documents")
print(" python rag_bot.py query \"question\" - Ask a question")
print(" python rag_bot.py interactive - Start interactive mode")
sys.exit(1)
command = sys.argv[1]
if command == "ingest":
ingest()
elif command == "query":
if len(sys.argv) < 3:
print("Please provide a question: python rag_bot.py query \"your question\"")
sys.exit(1)
query(sys.argv[2])
elif command == "interactive":
interactive()
else:
print(f"Unknown command: {command}")
print("Use 'ingest', 'query', or 'interactive'")What does this code do?
- interactive() runs a loop that reads user input, passes it through the full RAG pipeline, and prints the answer. Typing quit, exit, or q breaks the loop.
- The __main__ block gives your script three CLI commands: ingest to embed documents, query for a single question, and interactive for a continuous session.
- Save rag_bot.py.
- Test the interactive mode by running:
python rag_bot.py interactiveYou should see RAG Document Q&A Bot printed, followed by a You: prompt waiting for input. Type a question about your documents and press Enter. The bot retrieves relevant chunks and prints a generated answer.
- Type quit to exit the interactive loop.
Not seeing any answers?
- If you see "No documents found", run python rag_bot.py ingest first to populate the ChromaDB collection.
- If you see an authentication error, check that your .env file contains a valid OPENAI_API_KEY and that you have credit on your OpenAI account.
- If you see a module import error, make sure your virtual environment is active.
Why isn't my RAG bot returning answers when I run the interactive command?
✔️ Awesome, I've got everything!
Great. Double check you have saved rag_bot.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete rag_bot.py as it should look at the end of this step:
import os
import sys
import glob
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
COLLECTION_NAME = "documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
def load_documents(directory="documents"):
"""Load all .txt files from the specified directory."""
documents = []
txt_files = glob.glob(os.path.join(directory, "*.txt"))
for file_path in txt_files:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"filename": os.path.basename(file_path),
"path": file_path,
})
print(f"Loaded {len(documents)} document(s) from '{directory}/'")
return documents
def chunk_text(text, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
"""Split text into overlapping chunks."""
chunks = []
paragraphs = text.split("\n\n")
current_chunk = ""
for paragraph in paragraphs:
if len(current_chunk) + len(paragraph) + 2 <= chunk_size:
current_chunk += paragraph + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
if len(paragraph) > chunk_size:
words = paragraph.split()
current_chunk = ""
for word in words:
if len(current_chunk) + len(word) + 1 <= chunk_size:
current_chunk += word + " "
else:
chunks.append(current_chunk.strip())
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + word + " "
else:
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + paragraph + "\n\n"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def get_embeddings(texts):
"""Get embeddings for a list of texts using OpenAI API."""
response = client.embeddings.create(
input=texts,
model=EMBEDDING_MODEL
)
return [item.embedding for item in response.data]
def embed_and_store(documents):
"""Chunk documents, generate embeddings, and store in ChromaDB."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
all_chunks = []
all_metadatas = []
all_ids = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({
"source": doc["filename"],
"chunk_index": i,
})
all_ids.append(f"{doc['filename']}_chunk_{i}")
print(f"Created {len(all_chunks)} chunks from {len(documents)} document(s)")
print("Generating embeddings...")
batch_size = 100
all_embeddings = []
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i + batch_size]
embeddings = get_embeddings(batch)
all_embeddings.extend(embeddings)
collection.add(
ids=all_ids,
documents=all_chunks,
embeddings=all_embeddings,
metadatas=all_metadatas,
)
print(f"Stored {len(all_chunks)} chunks in ChromaDB collection '{COLLECTION_NAME}'")
return collection
def retrieve(question, n_results=3):
"""Retrieve the most relevant chunks for a question."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
question_embedding = get_embeddings([question])[0]
results = collection.query(
query_embeddings=[question_embedding],
n_results=n_results,
)
return results
def generate_answer(question, context_chunks):
"""Generate an answer using retrieved context and OpenAI chat."""
context = "\n\n---\n\n".join(context_chunks)
system_prompt = """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 in my documents to answer that question."
Do not make up information that is not in the context."""
user_prompt = f"""Context:
{context}
Question: {question}
Answer based only on the context above:"""
response = client.chat.completions.create(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
temperature=0,
)
return response.choices[0].message.content
def ingest():
"""Run the ingestion pipeline."""
documents = load_documents()
if not documents:
print("No .txt files found in 'documents/' directory.")
print("Add some .txt files and try again.")
return
embed_and_store(documents)
print("\nIngestion complete! You can now query your documents.")
def query(question):
"""Query the RAG pipeline."""
print(f"\nQuestion: {question}")
print("Retrieving relevant context...")
results = retrieve(question, n_results=3)
if not results["documents"][0]:
print("No documents found. Run 'python rag_bot.py ingest' first.")
return
context_chunks = results["documents"][0]
sources = [meta["source"] for meta in results["metadatas"][0]]
print(f"Found {len(context_chunks)} relevant chunks from: {', '.join(set(sources))}")
print("\nGenerating answer...\n")
answer = generate_answer(question, context_chunks)
print(f"Answer: {answer}")
def interactive():
"""Run an interactive query loop."""
print("\nRAG Document Q&A Bot")
print("Type your questions below. Type 'quit' to exit.\n")
while True:
question = input("You: ").strip()
if question.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not question:
continue
query(question)
print()
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage:")
print(" python rag_bot.py ingest - Load and embed documents")
print(" python rag_bot.py query \"question\" - Ask a question")
print(" python rag_bot.py interactive - Start interactive mode")
sys.exit(1)
command = sys.argv[1]
if command == "ingest":
ingest()
elif command == "query":
if len(sys.argv) < 3:
print("Please provide a question: python rag_bot.py query \"your question\"")
sys.exit(1)
query(sys.argv[2])
elif command == "interactive":
interactive()
else:
print(f"Unknown command: {command}")
print("Use 'ingest', 'query', or 'interactive'")Your RAG bot is fully functional. Next up, you will test it with different types of questions and tune the retrieval parameters to improve answer quality.
Test and Improve Your RAG Pipeline
Your RAG pipeline is fully built. Every stage is wired together: loading, chunking, embedding, storing, retrieving, and generating.
A pipeline only becomes trustworthy once you stress-test it. In this step, you will probe its strengths and weaknesses by asking different types of questions, tuning retrieval parameters, and confirming the final CLI structure works cleanly.
In this step, get ready to:
- Test the bot with answerable, partially covered, and out-of-scope questions.
- Experiment with retrieval chunk counts and observe answer quality changes.
- Verify the final CLI structure supports all three commands.
Test with different question types
A well-grounded RAG bot should answer confidently when the information exists in its documents. It should also refuse to hallucinate when the answer is not there.
- Start by asking a question that is directly answerable from your documents. Run this command, replacing the question with something specific to your text files:
python rag_bot.py query "[[DIRECT_QUESTION="a question directly answered by your documents"]]"You should see the bot retrieve relevant chunks and produce an answer grounded in your document content.
- Now ask a question that is only partially covered by your documents:
python rag_bot.py query "[[PARTIAL_QUESTION="a question only partially covered by your documents"]]"The bot should answer with what it can find but not fabricate details beyond the retrieved context.
- Finally, ask a question about a topic that is completely absent from your documents:
python rag_bot.py query "What is the recipe for chocolate lava cake?"The bot should respond with something like "I don't have enough information in my documents to answer that question." This is the grounding behavior your system prompt enforces.
Why test with out-of-scope questions?
The biggest risk with AI systems is hallucination. An LLM will confidently generate plausible-sounding answers even when the context contains nothing relevant.
Your system prompt includes an explicit instruction to say "I don't have enough information" rather than guess. Testing with off-topic questions confirms this guardrail is working.
Bot answering off-topic questions confidently?
If the bot is generating answers for questions not in your documents, check the system_prompt variable inside generate_answer(). Make sure it includes the instruction to only answer based on the provided context.
Also verify that temperature=0 is set in your chat.completions.create() call. Higher temperatures encourage creative (and potentially hallucinated) responses.
Why is my RAG bot hallucinating instead of saying it doesn't know?
Experiment with retrieval chunk count
The number of chunks you retrieve directly affects answer quality. Too few chunks might miss relevant context. Too many might dilute the signal with noise.
- In rag_bot.py, find this line inside the query() function:
results = retrieve(question, n_results=3)- Change n_results=3 to n_results=1, so the line looks like this:
results = retrieve(question, n_results=1)- Save rag_bot.py and test the same directly-answerable question you used earlier:
python rag_bot.py query "[[DIRECT_QUESTION="the same directly-answerable question from earlier"]]"Notice how retrieving only 1 chunk limits the context available to the model. The answer may be less complete or miss relevant details.
- Now change n_results=1 to n_results=5:
results = retrieve(question, n_results=5)- Save rag_bot.py and run the same query again.
python rag_bot.py query "[[DIRECT_QUESTION="the same directly-answerable question from earlier"]]"With 5 chunks, the model has more context to draw from. The answer may be more comprehensive, but could also include less relevant information that dilutes precision.
What is the ideal chunk count?
There is no universal answer. It depends on your document length, chunk size, and how spread out the relevant information is. For most use cases, 3 chunks is a strong default that balances coverage with precision.
In production RAG systems, teams often run experiments with different chunk counts and measure answer quality using metrics like faithfulness and relevance. Chunking strategy alone can shift retrieval accuracy by 20-30%.
- Set n_results back to 3 for the final version:
results = retrieve(question, n_results=3)- Save rag_bot.py.
Verify the final CLI structure
Your rag_bot.py is already structured as a clean CLI tool with three commands. Let's confirm each one works end-to-end.
- Verify the ingest command works by running:
python rag_bot.py ingestYou should see output showing how many documents were loaded, how many chunks were created, and a confirmation that they were stored in ChromaDB.
- Verify the query command works with a question:
python rag_bot.py query "[[FINAL_QUESTION="a question about your documents"]]"You should see the retrieval status, source filenames, and a generated answer grounded in your document content.
- Verify the interactive command launches the loop:
python rag_bot.py interactive- Ask a few questions in the interactive session, then type quit to exit.
What does the CLI structure give you?
The __main__ block at the bottom of rag_bot.py parses sys.argv to determine which function to call. This separates the ingestion step (expensive, done once) from the query step (cheap, done many times).
You ingest once to populate your vector database, then query as many times as you want without re-embedding.
Seeing 'No documents found' on query?
This means the chroma_db directory is empty or missing. Run python rag_bot.py ingest first to populate the vector database, then try your query again.
My RAG bot says no documents found even though I have text files in my documents folder. Help me debug.
✔️ Awesome, I've got everything!
Great. Double check you have saved rag_bot.py with n_results=3 as the final value.
ⓧ I'd like to double check the full code
import os
import sys
import glob
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")
EMBEDDING_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
COLLECTION_NAME = "documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
def load_documents(directory="documents"):
"""Load all .txt files from the specified directory."""
documents = []
txt_files = glob.glob(os.path.join(directory, "*.txt"))
for file_path in txt_files:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"filename": os.path.basename(file_path),
"path": file_path,
})
print(f"Loaded {len(documents)} document(s) from '{directory}/'")
return documents
def chunk_text(text, chunk_size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
"""Split text into overlapping chunks."""
chunks = []
paragraphs = text.split("\n\n")
current_chunk = ""
for paragraph in paragraphs:
if len(current_chunk) + len(paragraph) + 2 <= chunk_size:
current_chunk += paragraph + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
if len(paragraph) > chunk_size:
words = paragraph.split()
current_chunk = ""
for word in words:
if len(current_chunk) + len(word) + 1 <= chunk_size:
current_chunk += word + " "
else:
chunks.append(current_chunk.strip())
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + word + " "
else:
overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else ""
current_chunk = overlap_text + paragraph + "\n\n"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def get_embeddings(texts):
"""Get embeddings for a list of texts using OpenAI API."""
response = client.embeddings.create(
input=texts,
model=EMBEDDING_MODEL
)
return [item.embedding for item in response.data]
def embed_and_store(documents):
"""Chunk documents, generate embeddings, and store in ChromaDB."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
all_chunks = []
all_metadatas = []
all_ids = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append(chunk)
all_metadatas.append({
"source": doc["filename"],
"chunk_index": i,
})
all_ids.append(f"{doc['filename']}_chunk_{i}")
print(f"Created {len(all_chunks)} chunks from {len(documents)} document(s)")
print("Generating embeddings...")
batch_size = 100
all_embeddings = []
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i + batch_size]
embeddings = get_embeddings(batch)
all_embeddings.extend(embeddings)
collection.add(
ids=all_ids,
documents=all_chunks,
embeddings=all_embeddings,
metadatas=all_metadatas,
)
print(f"Stored {len(all_chunks)} chunks in ChromaDB collection '{COLLECTION_NAME}'")
return collection
def retrieve(question, n_results=3):
"""Retrieve the most relevant chunks for a question."""
collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)
question_embedding = get_embeddings([question])[0]
results = collection.query(
query_embeddings=[question_embedding],
n_results=n_results,
)
return results
def generate_answer(question, context_chunks):
"""Generate an answer using retrieved context and OpenAI chat."""
context = "\n\n---\n\n".join(context_chunks)
system_prompt = """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 in my documents to answer that question."
Do not make up information that is not in the context."""
user_prompt = f"""Context:
{context}
Question: {question}
Answer based only on the context above:"""
response = client.chat.completions.create(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
temperature=0,
)
return response.choices[0].message.content
def ingest():
"""Run the ingestion pipeline."""
documents = load_documents()
if not documents:
print("No .txt files found in 'documents/' directory.")
print("Add some .txt files and try again.")
return
embed_and_store(documents)
print("\nIngestion complete! You can now query your documents.")
def query(question):
"""Query the RAG pipeline."""
print(f"\nQuestion: {question}")
print("Retrieving relevant context...")
results = retrieve(question, n_results=3)
if not results["documents"][0]:
print("No documents found. Run 'python rag_bot.py ingest' first.")
return
context_chunks = results["documents"][0]
sources = [meta["source"] for meta in results["metadatas"][0]]
print(f"Found {len(context_chunks)} relevant chunks from: {', '.join(set(sources))}")
print("\nGenerating answer...\n")
answer = generate_answer(question, context_chunks)
print(f"Answer: {answer}")
def interactive():
"""Run an interactive query loop."""
print("\nRAG Document Q&A Bot")
print("Type your questions below. Type 'quit' to exit.\n")
while True:
question = input("You: ").strip()
if question.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not question:
continue
query(question)
print()
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage:")
print(" python rag_bot.py ingest - Load and embed documents")
print(" python rag_bot.py query \"question\" - Ask a question")
print(" python rag_bot.py interactive - Start interactive mode")
sys.exit(1)
command = sys.argv[1]
if command == "ingest":
ingest()
elif command == "query":
if len(sys.argv) < 3:
print("Please provide a question: python rag_bot.py query \"your question\"")
sys.exit(1)
query(sys.argv[2])
elif command == "interactive":
interactive()
else:
print(f"Unknown command: {command}")
print("Use 'ingest', 'query', or 'interactive'")Your RAG pipeline is tested, tuned, and production-ready. You have a complete system that can ingest any set of text documents and answer questions about them with grounded, context-aware responses. Next up, the Secret Mission pushes your bot further with metadata filtering and source citations.
Secret mission
Add Metadata Filtering and Source Citations
Your RAG bot answers questions, but it doesn't tell you WHERE the answer came from. In a real-world system, users need to verify claims against their source documents. Can you add source-scoped retrieval and inline citations?
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, so there are no ongoing costs. Your OpenAI API key is only charged when you actively make API calls.
Resources you used:
- Project folder containing rag_bot.py, .env, documents/, and chroma_db/.
- Python virtual environment (inside your project folder).
Keep everything running
No action needed. Choose this if you're still actively experimenting with your RAG bot or plan to add more documents.
- Your chroma_db/ directory persists your embeddings, so you can query without re-ingesting.
- Add more .txt files to documents/ and run python rag_bot.py ingest again to expand your knowledge base.
- Your virtual environment and installed packages will remain available.
Pause - I'll come back to this later
Since everything is local and there are no running processes, there is nothing to pause. Your project files and ChromaDB data will stay on disk until you return.
- If your virtual environment is currently active, deactivate it by running:
deactivate- When you come back, navigate into your project folder and reactivate the virtual environment.
Delete - I don't want to use this again
Remove all project files and start fresh. This deletes your embedded documents, the ChromaDB database, and all project code.
- If your virtual environment is currently active, deactivate it by running:
deactivate- Navigate to your Desktop (or wherever you created your project folder) by running:
cd ~/Desktop- Delete your project folder (macOS/Linux) by running:
rm -rf [[PROJECT_FOLDER="your project folder name"]]- On Windows, open File Explorer, navigate to your Desktop, right-click the project folder, and select Delete.
- Your .env file containing your API key will be deleted with the project folder. Your OpenAI API key itself remains active in your OpenAI account.
Nice Work!
Nice Work!
You just built a complete Retrieval-Augmented Generation pipeline from scratch. Your bot loads documents, chunks them, embeds them into a vector database, and retrieves relevant context to generate grounded answers.
You've learned how to:
- Build a RAG pipeline end-to-end: loading documents, chunking text with overlap, generating vector embeddings, and storing them in a persistent ChromaDB collection.
- Use OpenAI's text-embedding-3-small for semantic search and gpt-4o-mini for generating answers grounded in retrieved context.
- Test and tune retrieval quality by adjusting chunk count and refining the system prompt to reduce hallucinations.
- Secret Mission: Added metadata filtering for scoped retrieval and source citations for answer transparency.
Ready to quiz yourself?