Build a RAG Document Q&A System

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

Introduction

30 Second Summary

Every time you upload a document to ChatGPT or ask Notion AI a question about your notes, a pattern called Retrieval-Augmented Generation is working behind the scenes. It finds the right pieces of your documents, then feeds them to an AI so it answers with facts instead of guesses.

In this project, you will build your own RAG document Q&A system from scratch using Python. You'll load text files, split them into chunks, store them as vector embeddings, and query them with natural language questions that get grounded, cited answers.

What You'll Build

A command-line RAG system that ingests text files into a local ChromaDB vector database and answers natural language questions using OpenAI's embedding and generation models.

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

  • A working RAG pipeline that loads documents, chunks them, and stores them as vector embeddings in a local database.
  • A retrieval system that finds the most relevant document chunks for any natural language question using similarity search.
  • A generation pipeline that produces cited answers grounded in your documents instead of hallucinated facts.
  • Secret Mission: Add a multi-document source tracker that shows which files each answer draws from, with relevance scores and confidence indicators.

Are there any prerequisites?

You'll need basic Python knowledge (variables, functions, and loops). You'll also need an OpenAI API account with billing enabled. OpenAI requires a minimum $5 credit purchase to use the API.

The total cost for completing this project is under $0.05 in API usage.

Set Up Your Python Environment

Your RAG system depends on several Python packages working together. Installing them globally can cause version conflicts with other projects on your machine.

In this step, you will create an isolated virtual environment for your project, install the required packages, and connect to the OpenAI API to verify everything works.

In this step, get ready to:
  • Create and activate a Python virtual environment.
  • Install the RAG pipeline packages (LangChain, ChromaDB).
  • Connect to OpenAI and verify your API key works.
Create a virtual environment

A virtual environment is a self-contained folder that holds its own copy of Python and any packages you install. This keeps your project's dependencies separate from everything else on your system.

  • Open the integrated terminal in VS Code by pressing Ctrl+` (Windows/Linux) or Control+` (macOS).
  • Create your virtual environment by running this command:

macOS

python3 -m venv venv
  • Activate the virtual environment by running:
source venv/bin/activate

Windows

python -m venv venv
  • Activate the virtual environment by running:
venv\Scripts\activate
  • Confirm the virtual environment is active by checking your terminal prompt. You should see (venv) at the beginning of your command line.

What is a virtual environment?

Think of it as a clean room for your project. Packages you install here stay here and do not interfere with other Python projects on your machine.

The (venv) prefix in your terminal confirms that any pip install commands will target this environment, not your system Python.

Don't see (venv) in your prompt?

Make sure you ran the activation command (not just the creation command). On macOS, the command starts with source. On Windows, run the activate script directly from the venv\Scripts\ folder.

If you see a permissions error on Windows PowerShell, you may need to allow script execution first by running Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser and then retrying the activate command.

Help me activate my Python virtual environment.

Install the project dependencies

Your RAG system needs four packages. You will list them in a requirements.txt file so that anyone can recreate your environment with a single command.

  • In VS Code, create a new file called requirements.txt in your project folder by clicking File then New File.
  • Paste the following contents into requirements.txt:
langchain==1.3.2
langchain-openai==1.2.2
langchain-text-splitters==1.1.0
chromadb==1.5.9

What are these packages?

  • langchain provides the core abstractions for building LLM-powered applications (prompts, chains, output parsers).
  • langchain-openai connects LangChain to OpenAI's embedding and chat models.
  • langchain-text-splitters provides tools to break long documents into smaller chunks for embedding.
  • chromadb is a lightweight vector database that stores and retrieves embeddings locally with zero configuration.
  • Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Install all four packages by running this command in your terminal:
pip install -r requirements.txt

This will download and install the packages along with their dependencies. It may take a minute or two.

  • Once the install finishes, confirm the packages are available by running:
pip list | grep -i langchain

You should see langchain, langchain-openai, and langchain-text-splitters listed with their version numbers.

Install failing or timing out?

Make sure (venv) is showing in your prompt. If it is not, re-run the activation command from the previous substep before retrying the install.

If you see a network timeout, check your internet connection and retry. Large packages like chromadb download compiled binaries that can take a moment on slower connections.

Help me fix my pip install error.

✔️ Awesome, I've got everything!

Great. Make sure you have saved requirements.txt and that all four packages installed without errors.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt file for comparison:

langchain==1.3.2
langchain-openai==1.2.2
langchain-text-splitters==1.1.0
chromadb==1.5.9
Configure your OpenAI API key

The RAG system uses OpenAI to generate vector embeddings from your documents and to produce answers from retrieved context. You need a valid API key to connect.

Do I need to pay for the OpenAI API?

Yes. OpenAI requires prepaid credits to use the API. A minimum $5 credit purchase is needed to get started. The total cost for this entire project is well under $0.05, so your credits will last a very long time.

  • If you do not already have an OpenAI API key, create one at platform.openai.com/api-keys.
  • Copy your API key, then set it as an environment variable in your terminal:

macOS

export OPENAI_API_KEY=your-api-key-here

Windows

set OPENAI_API_KEY=your-api-key-here
  • Replace your-api-key-here with the actual key you copied from OpenAI.
  • Verify connectivity by running a quick test embedding:
python -c "from langchain_openai import OpenAIEmbeddings; e = OpenAIEmbeddings(model='text-embedding-3-small'); print(len(e.embed_query('hello world')), 'dimensions')"

You should see 1536 dimensions printed in your terminal. This confirms your API key is valid and the embedding model is reachable.

What just happened?

You sent the text "hello world" to OpenAI's text-embedding-3-small model. It returned a vector with 1536 numbers that represent the meaning of that text. This is the same process your RAG system will use to convert document chunks into searchable vectors.

Seeing an authentication error?

Double check that you copied the full API key without extra spaces. The key should start with sk-.

If you see a message about insufficient credits, log in to platform.openai.com and add billing credits under Settings then Billing.

Help me fix my OpenAI API connection error.

Your environment is fully wired up. Next, you will create sample documents and learn how to break them into chunks that your RAG system can search.

Load and Chunk Your Documents

Your Python environment is ready with all the packages installed. Now it's time to give your RAG system something to work with.

Before an LLM can answer questions about your documents, it needs to break them into small, searchable pieces called chunks. In this step, you'll create sample documents and write the Python code that loads and splits them into chunks.

In this step, get ready to:
  • Create sample documents in a documents/ folder.
  • Write functions to load and chunk text files.
  • Verify the chunking output in your terminal.
Create sample documents

Your RAG system needs a knowledge base to search through. You'll create three text files about a fictional company called TechFlow Solutions. These cover different topics so you can test retrieval across multiple documents later.

  • Create a documents/ folder inside your project directory by running this command:
mkdir documents
  • Confirm the folder exists by running:
ls

You should see documents and requirements.txt in the output.

  • Create company_overview.txt inside the documents/ folder with the following content:
TechFlow Solutions Company Overview

TechFlow Solutions is a software consulting firm founded in 2019, headquartered in Austin, Texas. The company specializes in cloud migration, data engineering, and AI integration services for mid-market enterprises.

Our team of 85 engineers serves clients across healthcare, fintech, and e-commerce sectors. We have completed over 200 projects since founding, with a 96% client satisfaction rate.

Key Services:
- Cloud Migration: AWS, Azure, and GCP migration strategies with zero-downtime deployment
- Data Engineering: Building scalable data pipelines using Apache Spark, Airflow, and dbt
- AI Integration: Implementing RAG systems, fine-tuned models, and ML pipelines for production use

Revenue in 2024 was $12.4 million, representing 34% year-over-year growth. The company is privately held with Series A funding of $8 million raised in 2022.
  • Create engineering_practices.txt inside the documents/ folder with the following content:
TechFlow Engineering Practices

Code Review Policy:
All code changes require at least two approvals before merging. Senior engineers must approve changes to production infrastructure. Code reviews should be completed within 24 hours of submission.

Testing Standards:
- Unit test coverage must exceed 80% for all new code
- Integration tests are required for any API endpoint
- End-to-end tests run nightly in staging environments
- Performance benchmarks are tracked for critical paths

Deployment Process:
We use a trunk-based development model with feature flags. Deployments happen automatically when code merges to main, going through staging first. Production deployments require a manual approval gate. Rollbacks are automated and trigger within 5 minutes if error rates exceed 2%.

On-Call Rotation:
Engineers rotate on-call weekly. Primary responder has a 15-minute SLA for P1 incidents. Secondary responder backs up after 30 minutes. All incidents require a blameless postmortem within 48 hours.
  • Create benefits_policy.txt inside the documents/ folder with the following content:
TechFlow Employee Benefits

Health Insurance:
Full medical, dental, and vision coverage for employees and dependents. The company covers 90% of premiums. HSA contributions of $1,500 annually for individual plans, $3,000 for family plans.

Time Off:
- 20 days PTO per year (increases to 25 after 3 years)
- 10 paid holidays
- 5 sick days
- 12 weeks paid parental leave

Professional Development:
Each employee receives a $3,000 annual learning budget for conferences, courses, and certifications. Engineers get one dedicated learning day per month (12 days per year).

Remote Work:
TechFlow operates on a hybrid model. Employees are expected in-office Tuesday through Thursday. Fully remote arrangements are available for employees outside the Austin metro area with manager approval.

Compensation:
Annual performance reviews in March. Equity refresh grants vest over 4 years. 401(k) matching up to 4% of salary.

Why three different documents?

Each file covers a different topic: company overview, engineering practices, and employee benefits. When you ask a question later, your RAG system will need to figure out which document (and which section) contains the answer.

This simulates a real-world scenario where knowledge is spread across many files. The retrieval step is what makes RAG powerful.

  • Confirm all three files exist by running:
ls documents/

You should see benefits_policy.txt, company_overview.txt, and engineering_practices.txt listed.

Missing files?

Make sure you saved each file inside the documents/ folder, not in the project root. The file names must be lowercase with underscores.

help me figure out why my document files aren't showing up

Write functions to load and chunk text files

Now you'll write the Python code that reads your documents and splits them into chunks. Chunking is essential because LLMs have limited context windows. Sending an entire document wastes tokens and dilutes relevance.

  • Create a new file called rag_system.py in your project root and add the following imports:
import os
import sys
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import chromadb

What are these imports?

You're importing more than you need right now. That's intentional. These imports set up everything for the full RAG pipeline you'll build across the next few steps.

  • langchain_openai provides the OpenAI embedding and chat model wrappers.
  • langchain_text_splitters provides the chunking tool.
  • langchain_core provides prompt templates and output parsing.
  • chromadb is the vector database that will store your embeddings.
  • Below the imports, add the load_documents function:

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

What does this function do?

This function scans a directory for .txt files and loads each one into a dictionary with its content and source filename.

  • sorted(os.listdir(directory)) lists files in alphabetical order so results are consistent across runs.
  • Each document is stored as a dictionary with content (the full text) and source (the filename for citations later).
  • Below the load_documents function, add the chunk_documents function:

def chunk_documents(documents):
    """Split documents into chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({"text": split, "source": doc["source"], "chunk_index": i})

    print(f"\nCreated {len(chunks)} chunks from {len(documents)} documents")
    return chunks

What does this function do?

RecursiveCharacterTextSplitter is LangChain's recommended general-purpose text splitter. It tries to split at natural boundaries first.

  • chunk_size=500 means each chunk is at most 500 characters. This keeps chunks focused on one idea.
  • chunk_overlap=50 means adjacent chunks share 50 characters. This prevents information from being cut off at boundaries.
  • The separators list tells the splitter to try paragraph breaks first, then line breaks, then sentences, then words. It falls back through the list until pieces fit under 500 characters.
  • Save rag_system.py.

Seeing a red underline on imports?

If your editor shows import errors, make sure your virtual environment is activated and selected as the Python interpreter in VS Code. Click the Python version in the bottom status bar and select the interpreter from your venv folder.

help me fix import errors in my RAG system file

Verify the chunking output

Both functions print status messages when they run. You can test them directly from the terminal to confirm your documents load and chunk correctly.

  • Test your functions by running this command:
python -c "from rag_system import load_documents, chunk_documents; docs = load_documents('documents'); chunks = chunk_documents(docs); print(f'First chunk: {chunks[0][chr(34)+chr(116)+chr(101)+chr(120)+chr(116)+chr(34)][:80]}...')"

Simpler alternative

If the one-liner above gives you trouble with quote escaping, open a Python shell instead:

python
from rag_system import load_documents, chunk_documents
docs = load_documents('documents')
chunks = chunk_documents(docs)
print(f"First chunk preview: {chunks[0]['text'][:80]}...")
exit()

You should see output showing each file loaded with its character count, followed by the total number of chunks created.

What should I see?

Your terminal should print something similar to:

Loaded: benefits_policy.txt (742 characters)
Loaded: company_overview.txt (818 characters)
Loaded: engineering_practices.txt (849 characters)

Created 9 chunks from 3 documents

The exact character counts and chunk numbers may vary slightly depending on line endings. The key thing is that all three files loaded successfully and were split into multiple chunks.

Seeing ModuleNotFoundError?

Make sure your virtual environment is activated before running the command. You should see (venv) at the start of your terminal prompt. If not, activate it again with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

help me fix my ModuleNotFoundError when importing rag_system

✔️ Awesome, I've got everything!

Great. Double check you've saved rag_system.py and confirmed all three documents loaded correctly.

ⓧ I'd like to double check the full code

Here is the full rag_system.py file at this point:

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


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


def chunk_documents(documents):
    """Split documents into chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({"text": split, "source": doc["source"], "chunk_index": i})

    print(f"\nCreated {len(chunks)} chunks from {len(documents)} documents")
    return chunks

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

Generate Embeddings and Store in ChromaDB

Your documents are chunked and ready. But text chunks alone can't be searched by meaning. To find the chunks most relevant to a question, you need to convert each chunk into a vector embedding that captures its semantic meaning as a list of numbers.

In this step, you'll send your chunks through OpenAI's embedding model and store the resulting vectors in ChromaDB, a local vector database. Once stored, you can query the database with a question and get back the most semantically similar chunks in milliseconds.

In this step, get ready to:
  • Initialize the OpenAI embedding model and ChromaDB vector store.
  • Embed all document chunks and store them in the database.
  • Test retrieval with a similarity search.
Initialize the embedding model and vector store

A vector database needs two things: an embedding model to convert text into vectors, and a storage layer to persist those vectors on disk. You'll set up both now.

  • In rag_system.py, add the following create_vector_store function below your chunk_documents function:
def create_vector_store(chunks):
    """Create embeddings and store in ChromaDB."""
    # Initialize OpenAI's small embedding model
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    # Create a persistent ChromaDB client that stores data on disk
    client = chromadb.PersistentClient(path="./chroma_db")

    # Create a collection using cosine similarity for distance measurement
    collection = client.get_or_create_collection(
        name="documents", metadata={"hnsw:space": "cosine"}
    )

    # Skip embedding if data already exists (avoids duplicate entries on re-runs)
    if collection.count() > 0:
        print(f"Vector store already has {collection.count()} entries. Skipping ingestion.")
        return client, collection, embeddings

What does this code do?

OpenAIEmbeddings(model="text-embedding-3-small") creates an embedding client that converts text into 1536-dimensional vectors using OpenAI's API. Each vector captures the semantic meaning of a chunk.

chromadb.PersistentClient(path="./chroma_db") creates a local database that saves vectors to disk. Data persists between script runs so you don't re-embed every time.

The get_or_create_collection call creates a named collection with cosine similarity as the distance metric. The early-return check prevents duplicate embeddings if you run the script again.

Embed and store all chunks

The initialization is set up. Now you need to add the code that actually generates the embeddings and stores everything in ChromaDB.

  • In rag_system.py, add the following code directly below the if collection.count() > 0: block you just wrote (still inside the create_vector_store function):
    texts = [chunk["text"] for chunk in chunks]
    metadatas = [{"source": chunk["source"], "chunk_index": chunk["chunk_index"]} for chunk in chunks]
    ids = [f"chunk_{i}" for i in range(len(chunks))]

    print("\nGenerating embeddings...")
    embedded_texts = embeddings.embed_documents(texts)

    collection.add(
        documents=texts,
        embeddings=embedded_texts,
        metadatas=metadatas,
        ids=ids,
    )

    print(f"Stored {collection.count()} chunks in ChromaDB")
    return client, collection, embeddings

What does this code do?

This code prepares three parallel lists: the raw text of each chunk, metadata tracking where each chunk came from, and a unique ID for each entry.

embed_documents(texts) sends all chunk texts to the OpenAI API in one batch call. Each chunk comes back as a 1536-number vector.

collection.add() stores the texts, their embeddings, and metadata together. ChromaDB indexes everything so that similarity searches are fast.

  • Save rag_system.py.

Seeing an indentation error?

The embedding and storage code must be indented one level inside the create_vector_store function (4 spaces). If you see an IndentationError, make sure the new code starts at the same indent level as the if collection.count() > 0: line above it.

help me fix an indentation error in my create_vector_store function

Test retrieval with a similarity search

You need a function that takes a question, converts it to a vector, and finds the closest matching chunks. This is the retrieval half of RAG.

  • In rag_system.py, add the following query_documents function below create_vector_store:
def query_documents(collection, embeddings, question, k=3):
    """Retrieve relevant chunks for a question."""
    # Convert the question into a vector
    query_embedding = embeddings.embed_query(question)

    # Find the k most similar chunks in the collection
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=k,
        include=["documents", "metadatas", "distances"],
    )

    return results

What does this code do?

embed_query(question) converts the user's question into the same 1536-dimensional vector space as the stored chunks.

collection.query() searches the database for the k most similar vectors. It returns the matching documents, their metadata (source file info), and cosine distances. Lower distance means higher similarity.

  • Save rag_system.py.
  • Test the full pipeline by running this command in your terminal:
python3 << 'EOF'
from rag_system import load_documents, chunk_documents, create_vector_store, query_documents

docs = load_documents("documents")
chunks = chunk_documents(docs)
client, collection, embeddings = create_vector_store(chunks)

results = query_documents(collection, embeddings, "What services does TechFlow offer?")

for i, (doc, meta, dist) in enumerate(zip(results["documents"][0], results["metadatas"][0], results["distances"][0])):
    similarity = 1 - dist
    print(f"\nChunk {i+1} (similarity: {similarity:.4f}, source: {meta['source']}):")
    print(f"  {doc[:100]}...")
EOF

You should see 3 chunks returned, with the top result coming from company_overview.txt and a high similarity score (above 0.7). The chunks should contain information about TechFlow's services.

Why is similarity calculated as 1 - distance?

ChromaDB returns cosine distances where lower values mean more similar vectors. Subtracting from 1 converts distance to similarity, giving you a 0-to-1 scale where 1 means identical meaning.

Seeing an authentication error?

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

If you see a rate limit error, wait a few seconds and try again. The text-embedding-3-small model has generous rate limits, but new accounts may have lower thresholds.

help me debug an OpenAI API error when generating embeddings in my RAG system

✔️ Awesome, I've got everything!

Great. Double check you've saved rag_system.py with both the create_vector_store and query_documents functions.

ⓧ I'd like to double check the full code

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


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


def chunk_documents(documents):
    """Split documents into chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({"text": split, "source": doc["source"], "chunk_index": i})

    print(f"\nCreated {len(chunks)} chunks from {len(documents)} documents")
    return chunks


def create_vector_store(chunks):
    """Create embeddings and store in ChromaDB."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

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

    collection = client.get_or_create_collection(
        name="documents", metadata={"hnsw:space": "cosine"}
    )

    if collection.count() > 0:
        print(f"Vector store already has {collection.count()} entries. Skipping ingestion.")
        return client, collection, embeddings

    texts = [chunk["text"] for chunk in chunks]
    metadatas = [{"source": chunk["source"], "chunk_index": chunk["chunk_index"]} for chunk in chunks]
    ids = [f"chunk_{i}" for i in range(len(chunks))]

    print("\nGenerating embeddings...")
    embedded_texts = embeddings.embed_documents(texts)

    collection.add(
        documents=texts,
        embeddings=embedded_texts,
        metadatas=metadatas,
        ids=ids,
    )

    print(f"Stored {collection.count()} chunks in ChromaDB")
    return client, collection, embeddings


def query_documents(collection, embeddings, question, k=3):
    """Retrieve relevant chunks for a question."""
    query_embedding = embeddings.embed_query(question)

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

    return results

Your embedding pipeline is working. You can store documents once and query them instantly. Next up, you'll connect this retrieval system to an LLM so it can generate natural language answers grounded in your documents.

Build the RAG Chain

Your documents are chunked and stored as vectors in ChromaDB. The retrieval pipeline can find the most relevant chunks for any question.

Retrieval alone doesn't produce a human-readable answer. In this step, you'll connect those retrieval results to an LLM that generates answers grounded in your documents, not from its general training data.

In this step, get ready to:
  • Create a prompt template that constrains the LLM to answer only from provided context.
  • Wire a main function that orchestrates the full RAG pipeline end-to-end.
  • Test the system with a real question to verify grounded answers.
Create a retrieval-augmented prompt

The key to preventing hallucination is the prompt. A regular LLM will happily make up answers from its training data. Your prompt template needs to pass retrieved document chunks as context and explicitly instruct the model to answer only from that context.

  • In rag_system.py, add the following generate_answer() function below your existing query_documents() function:
def generate_answer(question, context_chunks, sources):
    """Generate an answer using retrieved context."""
    llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0)

    prompt = ChatPromptTemplate.from_template(
        """You are a helpful assistant that answers questions based only on the provided context.
If the context does not contain enough information to answer the question, say "I don't have enough information to answer that question."

Context:
{context}

Question: {question}

Answer the question based only on the context above. If you reference specific information, mention which source it comes from."""
    )

    chain = prompt | llm | StrOutputParser()

    context = "\n\n".join(
        [f"[Source: {src}]\n{chunk}" for chunk, src in zip(context_chunks, sources)]
    )

    answer = chain.invoke({"context": context, "question": question})
    return answer

What does this code do?

This function takes the retrieved chunks and generates a grounded answer using three connected components.

  • ChatOpenAI(model="gpt-4.1-nano", temperature=0) initializes the cheapest OpenAI chat model with zero randomness for consistent, deterministic answers.
  • The ChatPromptTemplate tells the LLM to answer only from the provided context and to say "I don't know" if the context is insufficient. This is what prevents hallucination.
  • chain = prompt | llm | StrOutputParser() uses LangChain's pipe operator to compose a pipeline. The prompt formats the input, the LLM generates a response, and the StrOutputParser extracts the text string.
  • The context string is built by joining each chunk with its source file name in brackets, so the LLM knows where each piece of information came from.

Why gpt-4.1-nano?

At $0.10 per 1M input tokens, gpt-4.1-nano is OpenAI's cheapest chat model. For a RAG system where most of the "intelligence" comes from retrieval (not reasoning), a smaller model is the right choice. It follows instructions well and costs almost nothing for this project.

  • Save rag_system.py.

Seeing a red underline on ChatOpenAI?

Your imports from the previous step already include ChatOpenAI, ChatPromptTemplate, and StrOutputParser. If you see an error, scroll to the top of your file and confirm all imports are present.

help me fix my LangChain imports for the RAG chain

Wire up the RAG pipeline

You have all the individual pieces: document loading, chunking, embedding, retrieval, and now generation. The main() function orchestrates them into a single end-to-end flow.

  • Below your generate_answer() function, add the main() function and entry point:
def main():
    """Main function to run the RAG system."""
    print("=" * 60)
    print("RAG Document Q&A System")
    print("=" * 60)

    doc_dir = "documents"
    if not os.path.exists(doc_dir):
        print(f"Error: '{doc_dir}' directory not found. Create it with .txt files.")
        sys.exit(1)

    print("\n\U0001f4c4 Loading documents...")
    documents = load_documents(doc_dir)
    if not documents:
        print("No .txt files found in documents/ directory.")
        sys.exit(1)

    print("\n\u2702\ufe0f  Chunking documents...")
    chunks = chunk_documents(documents)

    print("\n\U0001f522 Creating vector store...")
    client, collection, embeddings = create_vector_store(chunks)

    question = "What services does TechFlow offer?"
    results = query_documents(collection, embeddings, question)
    context_chunks = results["documents"][0]
    sources = [m["source"] for m in results["metadatas"][0]]
    print("\n\U0001f4a1 Test Answer:")
    answer = generate_answer(question, context_chunks, sources)
    print(answer)


if __name__ == "__main__":
    main()

What does this code do?

  • The function first checks that the documents/ directory exists and contains .txt files.
  • It calls your existing functions in order: load_documents(), chunk_documents(), then create_vector_store().
  • It runs a test question through query_documents() to retrieve relevant chunks, then passes those chunks to generate_answer() for a grounded response.
  • The if __name__ == "__main__" guard ensures the script only runs when executed directly, not when imported.
  • Save rag_system.py.

Seeing an indentation error?

Make sure main() is defined at the top level (no indentation) and is not accidentally nested inside generate_answer(). The if __name__ block should also be at the top level with no indentation.

help me fix indentation in my rag_system.py

Test the complete pipeline

Time to see all the pieces working together. The test question asks about TechFlow's services, which is covered in company_overview.txt.

  • Run the RAG system by executing this command in your terminal:
python rag_system.py

You should see the loading, chunking, and vector store messages, followed by a grounded answer about TechFlow's services (cloud migration, data engineering, and AI integration).

What should I see?

The output starts with your familiar loading and chunking messages. Since you already ran the embedding step previously, you'll see "Vector store already has X entries. Skipping ingestion." instead of regenerating embeddings.

The answer should mention cloud migration, data engineering, and AI integration. It should reference company_overview.txt as its source. If the answer mentions facts not in your documents, your prompt constraints aren't working correctly.

Getting an API error?

If you see an authentication error, confirm your OPENAI_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you need to re-export it.

If you see a model not found error, double-check that your code uses gpt-4.1-nano exactly as written.

help me debug my RAG system API error

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

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


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


def chunk_documents(documents):
    """Split documents into chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({"text": split, "source": doc["source"], "chunk_index": i})

    print(f"\nCreated {len(chunks)} chunks from {len(documents)} documents")
    return chunks


def create_vector_store(chunks):
    """Create embeddings and store in ChromaDB."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

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

    collection = client.get_or_create_collection(
        name="documents", metadata={"hnsw:space": "cosine"}
    )

    if collection.count() > 0:
        print(f"Vector store already has {collection.count()} entries. Skipping ingestion.")
        return client, collection, embeddings

    texts = [chunk["text"] for chunk in chunks]
    metadatas = [{"source": chunk["source"], "chunk_index": chunk["chunk_index"]} for chunk in chunks]
    ids = [f"chunk_{i}" for i in range(len(chunks))]

    print("\nGenerating embeddings...")
    embedded_texts = embeddings.embed_documents(texts)

    collection.add(
        documents=texts,
        embeddings=embedded_texts,
        metadatas=metadatas,
        ids=ids,
    )

    print(f"Stored {collection.count()} chunks in ChromaDB")
    return client, collection, embeddings


def query_documents(collection, embeddings, question, k=3):
    """Retrieve relevant chunks for a question."""
    query_embedding = embeddings.embed_query(question)

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

    return results


def generate_answer(question, context_chunks, sources):
    """Generate an answer using retrieved context."""
    llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0)

    prompt = ChatPromptTemplate.from_template(
        """You are a helpful assistant that answers questions based only on the provided context.
If the context does not contain enough information to answer the question, say "I don't have enough information to answer that question."

Context:
{context}

Question: {question}

Answer the question based only on the context above. If you reference specific information, mention which source it comes from."""
    )

    chain = prompt | llm | StrOutputParser()

    context = "\n\n".join(
        [f"[Source: {src}]\n{chunk}" for chunk, src in zip(context_chunks, sources)]
    )

    answer = chain.invoke({"context": context, "question": question})
    return answer


def main():
    """Main function to run the RAG system."""
    print("=" * 60)
    print("RAG Document Q&A System")
    print("=" * 60)

    doc_dir = "documents"
    if not os.path.exists(doc_dir):
        print(f"Error: '{doc_dir}' directory not found. Create it with .txt files.")
        sys.exit(1)

    print("\n\U0001f4c4 Loading documents...")
    documents = load_documents(doc_dir)
    if not documents:
        print("No .txt files found in documents/ directory.")
        sys.exit(1)

    print("\n\u2702\ufe0f  Chunking documents...")
    chunks = chunk_documents(documents)

    print("\n\U0001f522 Creating vector store...")
    client, collection, embeddings = create_vector_store(chunks)

    question = "What services does TechFlow offer?"
    results = query_documents(collection, embeddings, question)
    context_chunks = results["documents"][0]
    sources = [m["source"] for m in results["metadatas"][0]]
    print("\n\U0001f4a1 Test Answer:")
    answer = generate_answer(question, context_chunks, sources)
    print(answer)


if __name__ == "__main__":
    main()

Your RAG chain is working. The LLM generates answers grounded in your documents instead of making things up. Next up, you'll build an interactive conversational interface so you can ask any question without restarting the script.

Add a Conversational Interface

Your RAG system can retrieve relevant chunks and generate grounded answers. Right now, though, it runs a single hardcoded test question and exits.

In this step, you'll turn the script into a proper interactive tool. You'll build a terminal-based Q&A loop where you can ask unlimited questions, see which source files the answer came from, and verify that the system refuses to hallucinate when asked about topics outside your documents.

In this step, get ready to:
  • Build a while loop that accepts questions from the terminal.
  • Display source file names and relevance scores alongside each answer.
  • Test out-of-scope questions to confirm the system refuses to hallucinate.
Build an interactive question loop

A single-shot test is useful during development, but a real Q&A tool needs to accept repeated questions without restarting. A while True loop keeps the program running until the user explicitly types a quit command.

  • In rag_system.py, scroll down to the main() function.
  • Find this block of test code at the bottom of the function:
    question = "What services does TechFlow offer?"
    results = query_documents(collection, embeddings, question)
    context_chunks = results["documents"][0]
    sources = [m["source"] for m in results["metadatas"][0]]
    print("\n💡 Test Answer:")
    answer = generate_answer(question, context_chunks, sources)
    print(answer)
  • Delete that entire block and replace it with the following interactive loop:
    # Step 4: Interactive Q&A loop
    print("\n" + "=" * 60)
    print("Ready! Ask questions about your documents.")
    print("Type 'quit' to exit.")
    print("=" * 60)

    while True:
        print()
        question = input("❓ Your question: ").strip()
        if question.lower() in ("quit", "exit", "q"):
            print("Goodbye!")
            break
        if not question:
            continue

        # Retrieve relevant chunks
        results = query_documents(collection, embeddings, question)

        context_chunks = results["documents"][0]
        sources = [m["source"] for m in results["metadatas"][0]]

        # Generate answer
        print("\n💡 Answer:")
        answer = generate_answer(question, context_chunks, sources)
        print(answer)

What does this code do?

  • The banner (the print statements with = separators) signals that the system is ready for questions.
  • input("❓ Your question: ").strip() waits for the user to type something, then removes any leading/trailing whitespace.
  • The if question.lower() in ("quit", "exit", "q") check lets users exit gracefully with any of three commands.
  • if not question: continue skips empty inputs so the user doesn't trigger an API call by pressing Enter accidentally.
  • Save rag_system.py.

Seeing an IndentationError?

The entire while loop must be indented one level inside main() (4 spaces). The code inside the loop needs 8 spaces of indentation. Make sure you deleted the old test code completely before pasting.

Help me fix my indentation in rag_system.py

Display retrieved sources alongside answers

Showing where each answer came from builds trust. ChromaDB returns cosine distances alongside each retrieved chunk. You can convert these to a relevance score (1 minus the distance) so higher numbers mean more relevant.

  • In rag_system.py, find this line inside the while loop:
        sources = [m["source"] for m in results["metadatas"][0]]
  • Add this line directly below it to extract the distance scores:
        distances = results["distances"][0]
  • Now find print(answer) at the bottom of the while loop.
  • Add the following source display code directly below print(answer):
        print("\n📚 Sources used:")
        for i, (source, distance) in enumerate(zip(sources, distances), 1):
            similarity = 1 - distance
            print(f"   {i}. {source} (relevance: {similarity:.2f})")

What does this code do?

  • results["distances"][0] gets the cosine distances ChromaDB calculated between your question and each retrieved chunk.
  • 1 - distance converts distance to similarity. A score of 0.85 means the chunk is highly relevant to your question.
  • The loop prints each source file name with its relevance score, numbered for easy reference.
  • Save rag_system.py.

Seeing a KeyError for distances?

Make sure the query_documents() function includes "distances" in the include list. Check that your function has include=["documents", "metadatas", "distances"] in the collection.query() call.

Help me debug the distances KeyError in my RAG system

Test edge cases

The true test of a RAG system is whether it refuses to answer when the documents don't contain relevant information. Your prompt instructs the LLM to say it doesn't have enough information rather than making something up.

  • Run the RAG system by executing this command in your terminal:
python rag_system.py
  • When you see the ❓ Your question: prompt, try an in-scope question first:
What is TechFlow's remote work policy?

You should see a detailed answer mentioning the hybrid model, Tuesday through Thursday in-office, and the option for fully remote with manager approval. Below the answer, the sources section shows which files contributed, with relevance scores.

  • Now try an out-of-scope question to test the guardrails:
What is the capital of France?

The system should respond that it doesn't have enough information to answer that question. The retrieved chunks will still appear in the sources section, but with lower relevance scores.

  • Type quit to exit the loop when you're done testing.

Why does the system still show sources for irrelevant questions?

ChromaDB always returns the k most similar chunks, even if none are truly relevant. The similarity scores tell you how confident to be. A score above 0.8 means high relevance. A score below 0.5 means the system is grasping at straws.

The prompt template handles the other half of this problem. It instructs the LLM to admit when context isn't sufficient, rather than inventing an answer from its training data.

System is hallucinating answers for out-of-scope questions?

Check your prompt template in generate_answer(). Make sure it includes the instruction about saying "I don't have enough information" when context is insufficient. Also verify temperature=0 is set on your ChatOpenAI instance to minimize creative responses.

Help me fix hallucination in my RAG system

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete rag_system.py as it should look at the end of this step:

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


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


def chunk_documents(documents):
    """Split documents into chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({"text": split, "source": doc["source"], "chunk_index": i})

    print(f"\nCreated {len(chunks)} chunks from {len(documents)} documents")
    return chunks


def create_vector_store(chunks):
    """Create embeddings and store in ChromaDB."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

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

    collection = client.get_or_create_collection(
        name="documents", metadata={"hnsw:space": "cosine"}
    )

    if collection.count() > 0:
        print(f"Vector store already has {collection.count()} entries. Skipping ingestion.")
        return client, collection, embeddings

    texts = [chunk["text"] for chunk in chunks]
    metadatas = [{"source": chunk["source"], "chunk_index": chunk["chunk_index"]} for chunk in chunks]
    ids = [f"chunk_{i}" for i in range(len(chunks))]

    print("\nGenerating embeddings...")
    embedded_texts = embeddings.embed_documents(texts)

    collection.add(
        documents=texts,
        embeddings=embedded_texts,
        metadatas=metadatas,
        ids=ids,
    )

    print(f"Stored {collection.count()} chunks in ChromaDB")
    return client, collection, embeddings


def query_documents(collection, embeddings, question, k=3):
    """Retrieve relevant chunks for a question."""
    query_embedding = embeddings.embed_query(question)

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

    return results


def generate_answer(question, context_chunks, sources):
    """Generate an answer using retrieved context."""
    llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0)

    prompt = ChatPromptTemplate.from_template(
        """You are a helpful assistant that answers questions based only on the provided context.
If the context does not contain enough information to answer the question, say "I don't have enough information to answer that question."

Context:
{context}

Question: {question}

Answer the question based only on the context above. If you reference specific information, mention which source it comes from."""
    )

    chain = prompt | llm | StrOutputParser()

    context = "\n\n".join(
        [f"[Source: {src}]\n{chunk}" for chunk, src in zip(context_chunks, sources)]
    )

    answer = chain.invoke({"context": context, "question": question})
    return answer


def main():
    """Main function to run the RAG system."""
    print("=" * 60)
    print("RAG Document Q&A System")
    print("=" * 60)

    doc_dir = "documents"
    if not os.path.exists(doc_dir):
        print(f"Error: '{doc_dir}' directory not found. Create it with .txt files.")
        sys.exit(1)

    print("\n\U0001f4c4 Loading documents...")
    documents = load_documents(doc_dir)
    if not documents:
        print("No .txt files found in documents/ directory.")
        sys.exit(1)

    print("\n\u2702\ufe0f  Chunking documents...")
    chunks = chunk_documents(documents)

    print("\n\U0001f522 Creating vector store...")
    client, collection, embeddings = create_vector_store(chunks)

    # Step 4: Interactive Q&A loop
    print("\n" + "=" * 60)
    print("Ready! Ask questions about your documents.")
    print("Type 'quit' to exit.")
    print("=" * 60)

    while True:
        print()
        question = input("\u2753 Your question: ").strip()
        if question.lower() in ("quit", "exit", "q"):
            print("Goodbye!")
            break
        if not question:
            continue

        # Retrieve relevant chunks
        results = query_documents(collection, embeddings, question)

        context_chunks = results["documents"][0]
        sources = [m["source"] for m in results["metadatas"][0]]
        distances = results["distances"][0]

        # Generate answer
        print("\n\U0001f4a1 Answer:")
        answer = generate_answer(question, context_chunks, sources)
        print(answer)

        print("\n\U0001f4da Sources used:")
        for i, (source, distance) in enumerate(zip(sources, distances), 1):
            similarity = 1 - distance
            print(f"   {i}. {source} (relevance: {similarity:.2f})")


if __name__ == "__main__":
    main()

Your RAG system is fully functional. You can load any text documents, ask natural language questions, and get grounded answers with source citations. The system knows its own limits and won't hallucinate. Next up, the Secret Mission challenges you to add confidence scoring and detailed source attribution.

Secret mission

Add a Multi-Document Source Tracker With Relevance Scores

Your RAG system answers questions, but how confident is it? Right now, every answer looks equally trustworthy whether it came from a highly relevant chunk or a barely-related one. Upgrade your system to show detailed source attribution with confidence scoring, so users know when to trust the answer and when to dig deeper.

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. The OpenAI API only charges when you actively run the script.

Resources you used:

  • Your project folder (containing rag_system.py, requirements.txt, documents/, and chroma_db/).
  • Your Python virtual environment (inside the project folder).
  • The OPENAI_API_KEY environment variable set in your terminal session.

Keep everything running

No action needed. Choose this if you plan to add more documents or keep experimenting with questions. Your files stay on disk and the ChromaDB database persists between runs.

  • Your project folder and virtual environment remain on your Desktop.
  • The OPENAI_API_KEY environment variable only persists in the current terminal session unless you added it to your shell profile.
  • The OpenAI API will not charge you unless you run the script again.

Pause - I'll come back to this later

Shut down the virtual environment to free up your terminal session. All files remain on disk so you can pick up where you left off.

  • Deactivate the virtual environment by running:
deactivate
  • Close VS Code when you are done.
  • To resume later, open your project folder in VS Code and reactivate the virtual environment with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your documents, the ChromaDB vector database, and all Python code.

  • Deactivate the virtual environment by running:
deactivate
  • Navigate to the parent directory of your project folder by running:
cd ..
  • Delete the entire project folder by running:
rm -rf [[PROJECT_FOLDER="your project folder name"]]
  • On Windows, use this command instead:
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]
  • Unset the OpenAI API key environment variable for your current terminal session by running:
unset OPENAI_API_KEY
  • On Windows, use this command instead:
set OPENAI_API_KEY=
  • If you added OPENAI_API_KEY to your shell profile (.bashrc, .zshrc, or similar), remove that line manually.

Nice Work!

Nice Work!

You just built a working RAG system from scratch. This is the exact pattern behind every modern AI knowledge assistant, from ChatGPT with file uploads to Notion AI.

You've learned how to:

  • Build a document ingestion pipeline that loads text files and chunks them intelligently using RecursiveCharacterTextSplitter.
  • Generate vector embeddings with OpenAI's text-embedding-3-small model and store them in a local ChromaDB database for persistent similarity search.
  • Wire a complete RAG chain that retrieves relevant context and generates grounded answers with source citations, refusing to hallucinate when the documents don't contain the answer.
  • Add confidence scoring and multi-document source attribution that ranks files by relevance and warns users when retrieval quality is low.

Ready to quiz yourself?