Build a Multi-Context RAG Agent
Build a local RAG agent that answers questions from your own documents.
Introduction
30 Second Summary
You have documents scattered everywhere. Research notes, design references, analysis reports. Wouldn't it be great if you could just ask an AI questions about your own files, without uploading anything to the cloud?
In this project, you will build a local RAG agent that answers questions grounded in your personal documents. It runs entirely on your machine using Ollama and ChromaDB, so there's zero cost and full privacy.
What You'll Build
A Python command-line RAG agent that ingests PDFs and text files into separate vector collections, retrieves relevant chunks based on your questions, and generates grounded answers using a local LLM.
By the end of this project, you'll have:
- A document ingestion pipeline that chunks PDFs and text files into named ChromaDB vector embeddings collections.
- A query engine that retrieves relevant context from the right collection and generates cited answers via Ollama.
- A CLI interface for ingesting documents, switching between knowledge domains, and chatting with your data.
- Secret Mission: Add cross-collection search that queries all your knowledge domains at once and ranks results by relevance.
Are there any prerequisites?
You'll need Python 3.14, VS Code, and Ollama already installed on your machine. If you don't have these yet, get them set up before starting this project.
Set Up Your Environment
Your Python packages and Ollama models are ready to go. Now it's time to organize your project so the RAG agent knows where to find documents and how to connect to your local infrastructure.
In this step, you'll create a folder structure for categorizing your documents and write the configuration file that ties ChromaDB and Ollama together.
In this step, get ready to:
- Create a folder structure for organizing documents by category.
- Write the configuration constants that control your RAG agent's behavior.
- Initialize the embedding function and database client that power retrieval.
Create the document folder structure
Your RAG agent organizes knowledge into separate collections. Each collection maps to a folder of documents on disk.
- Create the documents directory and two category subdirectories by running these commands in your terminal:
mkdir documents
mkdir documents/ux-design
mkdir documents/game-designWhat does this do?
Each subdirectory inside documents/ becomes its own knowledge collection. When you ingest documents later, the agent stores each folder's content separately so you can query one domain at a time.
- Confirm the folders exist by running:
ls documentsYou should see game-design and ux-design listed in the output.
Why these categories?
These are just examples. You can rename or add more categories later to match your own knowledge domains. The agent doesn't care what the folders are called. It uses the folder name as the collection name.
Write the imports and configuration constants
The RAG agent needs a central configuration file that defines which models to use, where to store data, and how to chunk documents.
- Create a new file called rag_agent.py in your project folder.
- Add the following imports and configuration constants:
import os
import sys
from pathlib import Path
from chromadb import PersistentClient
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
from ollama import chat
from pypdf import PdfReader
# --- Configuration ---
OLLAMA_BASE_URL = "http://localhost:11434"
EMBEDDING_MODEL = "nomic-embed-text"
CHAT_MODEL = "qwen3:4b"
CHROMA_DB_PATH = "./chroma_data"
DOCUMENTS_PATH = "./documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50What does this code do?
The imports bring in the three libraries you installed: chromadb for vector storage, ollama for chat generation, and pypdf for reading PDF files.
The configuration constants control how the agent behaves:
- CHUNK_SIZE = 500 means each document gets split into 500-character passages for storage.
- CHUNK_OVERLAP = 50 means each chunk shares 50 characters with its neighbor, so ideas that span a boundary aren't lost.
- CHROMA_DB_PATH and DOCUMENTS_PATH point to the folders the agent reads from and writes to.
Initialize the embedding function and database client
The embedding function converts text into vector embeddings. The database client stores those vectors so you can search them later.
- In rag_agent.py, add the following code below your configuration constants:
# --- Embedding Function ---
ollama_ef = OllamaEmbeddingFunction(
url=OLLAMA_BASE_URL,
model_name=EMBEDDING_MODEL,
)
# --- ChromaDB Client ---
chroma_client = PersistentClient(path=CHROMA_DB_PATH)What does this code do?
ollama_ef wraps the nomic-embed-text model running in Ollama. ChromaDB calls this function automatically whenever it needs to embed text for storage or search.
chroma_client creates a persistent database in the ./chroma_data folder. Persistent means your data survives between script runs. You won't have to re-ingest documents every time.
- Save rag_agent.py.
- Verify the configuration connects to Ollama by running:
python rag_agent.pyThe script should exit with no output and no errors. This confirms that all imports resolve, Ollama is reachable, and the ChromaDB client initialized successfully.
Seeing an error?
- ModuleNotFoundError: Make sure your virtual environment is activated. You should see the venv name in your terminal prompt.
- ConnectionError or connection refused: Confirm Ollama is running. Open a separate terminal and run ollama list to check.
- If you see a warning about the embedding model not being found, make sure you pulled it earlier with ollama pull nomic-embed-text.
Help me debug my rag_agent.py connection error
✔️ Awesome, I've got everything!
Great. Double check you've saved rag_agent.py and confirmed it runs without errors.
ⓧ I'd like to double check the full code
Here's the complete rag_agent.py so far:
import os
import sys
from pathlib import Path
from chromadb import PersistentClient
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
from ollama import chat
from pypdf import PdfReader
# --- Configuration ---
OLLAMA_BASE_URL = "http://localhost:11434"
EMBEDDING_MODEL = "nomic-embed-text"
CHAT_MODEL = "qwen3:4b"
CHROMA_DB_PATH = "./chroma_data"
DOCUMENTS_PATH = "./documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
# --- Embedding Function ---
ollama_ef = OllamaEmbeddingFunction(
url=OLLAMA_BASE_URL,
model_name=EMBEDDING_MODEL,
)
# --- ChromaDB Client ---
chroma_client = PersistentClient(path=CHROMA_DB_PATH)Your project skeleton is in place. Next up, you'll build the ingestion pipeline that reads documents, splits them into chunks, and stores them as vectors in ChromaDB.
Build the Document Ingestion Pipeline
Your environment is ready, with all dependencies installed and Ollama models pulled. Now it's time to teach your agent how to read and store documents.
Before your RAG agent can answer questions, it needs to break documents into small passages and store them as vector embeddings. In this step, you'll build the full ingestion pipeline that reads files, splits them into chunks, and saves them into ChromaDB collections.
In this step, get ready to:
- Add sample documents to your collection folders.
- Write functions that load, chunk, and ingest documents into ChromaDB.
- Run the ingestion pipeline and verify data is persisted.
Add sample documents to your collection folders
Your RAG agent needs actual content to learn from. You'll add simple text files to the category folders you created earlier so the ingestion pipeline has something to process.
- In VS Code's Explorer sidebar, right-click the documents/ux-design folder and select New File.
- Name the file design-principles.txt.
- Paste the following sample content into the file:
User experience design focuses on creating products that provide meaningful experiences to users.
Key principles include usability, accessibility, and desirability.
Usability means the product is easy to use and learn.
Accessibility ensures the product works for people with diverse abilities.
Desirability covers the emotional and aesthetic aspects of the experience.
Good UX design requires understanding user needs through research, prototyping, and testing.
Iterative design allows teams to refine products based on real user feedback.- Save design-principles.txt.
- Now right-click the documents/game-design folder and select New File.
- Name this file level-design.txt.
- Paste the following sample content:
Level design is the process of creating environments and challenges for players to navigate.
Good level design teaches mechanics through play rather than explicit tutorials.
Pacing is critical. Levels should alternate between high-intensity and low-intensity moments.
The difficulty curve should gradually increase to keep players in a state of flow.
Environmental storytelling uses visual cues to communicate narrative without text.
Playtesting reveals problems that designers cannot anticipate on their own.
Blockouts are rough versions of levels used to test layout and flow before adding art.- Save level-design.txt.
Why plain text files?
Text files are the simplest way to verify your pipeline works. Once ingestion is running, you can drop PDFs and markdown files into the same folders and the agent will handle them automatically.
- Confirm both files exist by checking the Explorer sidebar.
You should see design-principles.txt inside documents/ux-design/ and level-design.txt inside documents/game-design/.
Write the document loading and chunking functions
The ingestion pipeline needs to handle different file formats and split their content into manageable passages. You'll write functions that load text and PDF files, then chunk the text using a sliding window approach.
- In rag_agent.py, scroll to the bottom of the file (below the chroma_client line).
- Add the document loading and chunking functions by pasting the following code:
# --- Document Loading ---
def load_text_file(file_path: str) -> str:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
def load_pdf_file(file_path: str) -> str:
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\n"
return text
def load_document(file_path: str) -> str:
path = Path(file_path)
if path.suffix.lower() == ".pdf":
return load_pdf_file(file_path)
elif path.suffix.lower() in (".txt", ".md"):
return load_text_file(file_path)
else:
print(f"Unsupported file type: {path.suffix}")
return ""
# --- Chunking ---
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if chunk.strip():
chunks.append(chunk.strip())
start = end - overlap
return chunksWhat does this code do?
The loading functions handle two file formats. load_text_file reads .txt and .md files directly. load_pdf_file uses pypdf to extract text from each page of a PDF.
The load_document function acts as a router, picking the right loader based on the file extension.
The chunk_text function uses a sliding window to split text into 500-character passages with 50 characters of overlap. The overlap ensures important context at chunk boundaries isn't lost.
- Save rag_agent.py.
You won't see any output yet. The next chunk ties everything together by writing the ingestion function that calls these helpers.
Seeing a red underline in VS Code?
If VS Code highlights PdfReader or Path as unresolved, make sure your virtual environment is selected as the Python interpreter. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), type Python: Select Interpreter, and choose the one from your venv folder.
help me fix import errors in my RAG agent script
Write the ingestion function and run it
The ingestion function orchestrates the full pipeline. It reads every file in a collection folder, chunks the content, attaches metadata, and stores everything in ChromaDB.
- In rag_agent.py, add the ingestion function below the chunk_text function you just wrote:
# --- Ingestion ---
def ingest_collection(collection_name: str) -> None:
collection_dir = Path(DOCUMENTS_PATH) / collection_name
if not collection_dir.exists():
print(f"Directory not found: {collection_dir}")
return
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
files = list(collection_dir.iterdir())
if not files:
print(f"No files found in {collection_dir}")
return
all_chunks = []
all_ids = []
all_metadatas = []
for file_path in files:
if file_path.is_file():
print(f" Processing: {file_path.name}")
text = load_document(str(file_path))
if not text:
continue
chunks = chunk_text(text)
for i, chunk in enumerate(chunks):
doc_id = f"{file_path.stem}_{i}"
all_chunks.append(chunk)
all_ids.append(doc_id)
all_metadatas.append({
"source": file_path.name,
"chunk_index": i,
"collection": collection_name,
})
if all_chunks:
collection.add(
documents=all_chunks,
ids=all_ids,
metadatas=all_metadatas,
)
print(f" Ingested {len(all_chunks)} chunks into '{collection_name}'")
else:
print(f" No content found to ingest.")What does this code do?
The ingest_collection function takes a collection name (like ux-design), finds the matching folder in ./documents/, and processes every file inside it.
Each chunk gets a unique ID built from the filename and chunk index. Metadata tracks which file and collection the chunk came from, so you can trace answers back to their source later.
The collection.add() call stores the chunks in ChromaDB. ChromaDB automatically generates vector embeddings using the ollama_ef embedding function you configured earlier.
- Save rag_agent.py.
- Test the ingestion pipeline by running this command in your terminal:
python -c "from rag_agent import ingest_collection; ingest_collection('ux-design')"You should see output like this:
Processing: design-principles.txt
Ingested 2 chunks into 'ux-design'Why 2 chunks?
Your sample file is about 550 characters. With a chunk size of 500 and overlap of 50, the text splits into 2 passages. Longer documents produce more chunks.
- Now ingest the second collection by running:
python -c "from rag_agent import ingest_collection; ingest_collection('game-design')"- Verify the data is persisted by checking that a chroma_data folder appeared in your project directory:
ls chroma_dataYou should see files and folders inside chroma_data/ confirming that ChromaDB has written your embeddings to disk.
Seeing 'Connection refused' or a timeout?
This means Ollama isn't running. ChromaDB needs Ollama to generate embeddings during ingestion. Make sure Ollama is active by running ollama list in a separate terminal. If it's not running, start it with ollama serve.
help me fix the Ollama connection error during ChromaDB ingestion
✔️ Awesome, I've got everything!
Great. Double check you've saved rag_agent.py and that both collections ingested successfully.
ⓧ I'd like to double check the full code
Here is the complete rag_agent.py file up to this point:
import os
import sys
from pathlib import Path
from chromadb import PersistentClient
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
from ollama import chat
from pypdf import PdfReader
# --- Configuration ---
OLLAMA_BASE_URL = "http://localhost:11434"
EMBEDDING_MODEL = "nomic-embed-text"
CHAT_MODEL = "qwen3:4b"
CHROMA_DB_PATH = "./chroma_data"
DOCUMENTS_PATH = "./documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
# --- Embedding Function ---
ollama_ef = OllamaEmbeddingFunction(
url=OLLAMA_BASE_URL,
model_name=EMBEDDING_MODEL,
)
# --- ChromaDB Client ---
chroma_client = PersistentClient(path=CHROMA_DB_PATH)
# --- Document Loading ---
def load_text_file(file_path: str) -> str:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
def load_pdf_file(file_path: str) -> str:
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\n"
return text
def load_document(file_path: str) -> str:
path = Path(file_path)
if path.suffix.lower() == ".pdf":
return load_pdf_file(file_path)
elif path.suffix.lower() in (".txt", ".md"):
return load_text_file(file_path)
else:
print(f"Unsupported file type: {path.suffix}")
return ""
# --- Chunking ---
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if chunk.strip():
chunks.append(chunk.strip())
start = end - overlap
return chunks
# --- Ingestion ---
def ingest_collection(collection_name: str) -> None:
collection_dir = Path(DOCUMENTS_PATH) / collection_name
if not collection_dir.exists():
print(f"Directory not found: {collection_dir}")
return
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
files = list(collection_dir.iterdir())
if not files:
print(f"No files found in {collection_dir}")
return
all_chunks = []
all_ids = []
all_metadatas = []
for file_path in files:
if file_path.is_file():
print(f" Processing: {file_path.name}")
text = load_document(str(file_path))
if not text:
continue
chunks = chunk_text(text)
for i, chunk in enumerate(chunks):
doc_id = f"{file_path.stem}_{i}"
all_chunks.append(chunk)
all_ids.append(doc_id)
all_metadatas.append({
"source": file_path.name,
"chunk_index": i,
"collection": collection_name,
})
if all_chunks:
collection.add(
documents=all_chunks,
ids=all_ids,
metadatas=all_metadatas,
)
print(f" Ingested {len(all_chunks)} chunks into '{collection_name}'")
else:
print(f" No content found to ingest.")Your ingestion pipeline is working. Documents are chunked, embedded, and stored in named collections. Next up, you'll build the query engine that retrieves relevant chunks and generates grounded answers.
Build the Query Engine
Your ingestion pipeline is working. Documents are chunked and stored as vectors inside ChromaDB collections.
Now you need the other half of the RAG pattern: retrieval and generation. When you ask a question, the agent needs to find the most relevant chunks from your collection, then pass them to the LLM so it can produce a grounded answer instead of hallucinating.
In this step, get ready to:
- Write a retrieval function that queries ChromaDB for the most relevant chunks.
- Write a generation function that sends retrieved context to Ollama for a grounded answer.
- Wire both functions into a complete RAG pipeline that prints answers with source citations.
Write the retrieval function
When a user asks a question, you need to convert that question into a vector and find the stored chunks closest to it. This is called semantic search. ChromaDB handles the embedding and comparison for you.
- In rag_agent.py, scroll to the bottom of your file (below the ingest_collection function).
- Add the retrieval function by copying this code:
# --- Retrieval ---
def retrieve(collection_name: str, query: str, n_results: int = 3) -> list[dict]:
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
results = collection.query(
query_texts=[query],
n_results=n_results,
)
documents = results["documents"][0] if results["documents"] else []
metadatas = results["metadatas"][0] if results["metadatas"] else []
distances = results["distances"][0] if results["distances"] else []
retrieved = []
for doc, meta, dist in zip(documents, metadatas, distances):
retrieved.append({
"text": doc,
"source": meta.get("source", "unknown"),
"distance": dist,
})
return retrievedWhat does this code do?
The retrieve function takes a collection name and a query string, then asks ChromaDB to find the 3 most similar chunks.
- collection.query(query_texts=[query]) embeds your query using the same nomic-embed-text model and searches for nearby vectors.
- ChromaDB returns results in a columnar format (nested arrays). The [0] indexing pulls the results for the first (and only) query.
- The function packages each result into a dictionary with text, source, and distance (lower distance means higher relevance).
- Save rag_agent.py.
Seeing a NameError?
Make sure you placed the retrieve function below the ingest_collection function. It relies on chroma_client and ollama_ef being defined earlier in the file.
help me debug a NameError in my retrieve function
Write the generation function
Retrieval alone gives you raw text chunks. You need a function that formats those chunks into a prompt and sends it to your local LLM so it can synthesize a human-readable answer.
- Below the retrieve function you just added, copy in the generation function:
# --- Generation ---
def generate_answer(query: str, context_chunks: list[dict]) -> str:
context_text = "\n\n".join(
f"[Source: {chunk['source']}]\n{chunk['text']}"
for chunk in context_chunks
)
system_prompt = (
"You are a helpful assistant. Answer the user's question based ONLY on the "
"provided context. If the context does not contain enough information to answer, "
"say so. Cite the source file when referencing information."
)
user_message = f"Context:\n{context_text}\n\nQuestion: {query}"
response = chat(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
)
return response.message.contentWhat does this code do?
The generate_answer function builds a structured prompt from your retrieved chunks and sends it to Ollama.
- Each chunk is labelled with its source filename so the LLM can cite where information came from.
- The system prompt constrains the model to answer ONLY from the provided context. This prevents hallucination.
- chat() calls the local qwen3:4b model via Ollama's API at http://localhost:11434.
- Save rag_agent.py.
Getting a connection error from Ollama?
Make sure Ollama is still running in the background. If you closed it, reopen it. You can verify by running:
ollama listYou should see qwen3:4b and nomic-embed-text in the output.
help me fix my Ollama connection error
Wire them into a RAG pipeline and test
Now you have retrieval and generation as separate functions. The final piece connects them into a single pipeline that a user can call with one question.
- Below the generate_answer function, add the ask function:
# --- RAG Pipeline ---
def ask(collection_name: str, query: str) -> None:
print(f"\nSearching '{collection_name}' for: {query}")
print("-" * 50)
chunks = retrieve(collection_name, query)
if not chunks:
print("No relevant documents found.")
return
answer = generate_answer(query, chunks)
print(f"\nAnswer:\n{answer}")
print(f"\nSources:")
for chunk in chunks:
print(f" - {chunk['source']} (distance: {chunk['distance']:.4f})")What does this code do?
The ask function is your complete RAG pipeline in one call.
- It retrieves the top 3 chunks from the named collection.
- It passes those chunks to generate_answer which sends them to the LLM.
- It prints the answer followed by the source files and their distance scores so you can see which documents contributed.
- Save rag_agent.py.
- Test the full pipeline by running this in your terminal:
python -c "from rag_agent import ask; ask('ux-design', 'What are the key principles of good UX?')"What should I see?
You should see output with three sections: a "Searching" header, a generated answer grounded in your documents, and a "Sources" list showing which files contributed (with distance scores like 0.4521).
The first run may take 10-20 seconds because Ollama needs to load the model into memory. Subsequent queries will be faster.
Adjust your test query
Replace ux-design with the name of one of your actual collections, and change the question to match your documents. The collection name must match the folder name you used during ingestion.
Seeing 'No relevant documents found'?
This means the collection is empty or the name does not match. Double check that you ran ingest_collection for the collection name you are querying. Collection names are case-sensitive and must match the folder name exactly.
help me figure out why my RAG query returns no results
✔️ Awesome, I've got everything!
Great. Double check you have saved rag_agent.py with all three new functions.
ⓧ I'd like to double check the full code
Here is the complete rag_agent.py file as it should look at the end of this step:
import os
import sys
from pathlib import Path
from chromadb import PersistentClient
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
from ollama import chat
from pypdf import PdfReader
# --- Configuration ---
OLLAMA_BASE_URL = "http://localhost:11434"
EMBEDDING_MODEL = "nomic-embed-text"
CHAT_MODEL = "qwen3:4b"
CHROMA_DB_PATH = "./chroma_data"
DOCUMENTS_PATH = "./documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
# --- Embedding Function ---
ollama_ef = OllamaEmbeddingFunction(
url=OLLAMA_BASE_URL,
model_name=EMBEDDING_MODEL,
)
# --- ChromaDB Client ---
chroma_client = PersistentClient(path=CHROMA_DB_PATH)
# --- Document Loading ---
def load_text_file(file_path: str) -> str:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
def load_pdf_file(file_path: str) -> str:
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\n"
return text
def load_document(file_path: str) -> str:
path = Path(file_path)
if path.suffix.lower() == ".pdf":
return load_pdf_file(file_path)
elif path.suffix.lower() in (".txt", ".md"):
return load_text_file(file_path)
else:
print(f"Unsupported file type: {path.suffix}")
return ""
# --- Chunking ---
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if chunk.strip():
chunks.append(chunk.strip())
start = end - overlap
return chunks
# --- Ingestion ---
def ingest_collection(collection_name: str) -> None:
collection_dir = Path(DOCUMENTS_PATH) / collection_name
if not collection_dir.exists():
print(f"Directory not found: {collection_dir}")
return
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
files = list(collection_dir.iterdir())
if not files:
print(f"No files found in {collection_dir}")
return
all_chunks = []
all_ids = []
all_metadatas = []
for file_path in files:
if file_path.is_file():
print(f" Processing: {file_path.name}")
text = load_document(str(file_path))
if not text:
continue
chunks = chunk_text(text)
for i, chunk in enumerate(chunks):
doc_id = f"{file_path.stem}_{i}"
all_chunks.append(chunk)
all_ids.append(doc_id)
all_metadatas.append({
"source": file_path.name,
"chunk_index": i,
"collection": collection_name,
})
if all_chunks:
collection.add(
documents=all_chunks,
ids=all_ids,
metadatas=all_metadatas,
)
print(f" Ingested {len(all_chunks)} chunks into '{collection_name}'")
else:
print(f" No content found to ingest.")
# --- Retrieval ---
def retrieve(collection_name: str, query: str, n_results: int = 3) -> list[dict]:
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
results = collection.query(
query_texts=[query],
n_results=n_results,
)
documents = results["documents"][0] if results["documents"] else []
metadatas = results["metadatas"][0] if results["metadatas"] else []
distances = results["distances"][0] if results["distances"] else []
retrieved = []
for doc, meta, dist in zip(documents, metadatas, distances):
retrieved.append({
"text": doc,
"source": meta.get("source", "unknown"),
"distance": dist,
})
return retrieved
# --- Generation ---
def generate_answer(query: str, context_chunks: list[dict]) -> str:
context_text = "\n\n".join(
f"[Source: {chunk['source']}]\n{chunk['text']}"
for chunk in context_chunks
)
system_prompt = (
"You are a helpful assistant. Answer the user's question based ONLY on the "
"provided context. If the context does not contain enough information to answer, "
"say so. Cite the source file when referencing information."
)
user_message = f"Context:\n{context_text}\n\nQuestion: {query}"
response = chat(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
)
return response.message.content
# --- RAG Pipeline ---
def ask(collection_name: str, query: str) -> None:
print(f"\nSearching '{collection_name}' for: {query}")
print("-" * 50)
chunks = retrieve(collection_name, query)
if not chunks:
print("No relevant documents found.")
return
answer = generate_answer(query, chunks)
print(f"\nAnswer:\n{answer}")
print(f"\nSources:")
for chunk in chunks:
print(f" - {chunk['source']} (distance: {chunk['distance']:.4f})")Your query engine is live. You can ask questions about any collection and get grounded, cited answers from your local model. Next up, you will wrap this into a proper CLI interface so you can interact with your agent without writing Python one-liners.
Build the CLI Interface
Your RAG pipeline can ingest documents, retrieve relevant chunks, and generate grounded answers. All the core intelligence is built.
Right now, running any of these functions requires editing Python code directly. In this step, you'll wrap everything in a command-line interface so you can ingest, query, and manage collections interactively from your terminal.
In this step, get ready to:
- Create a helper function to list all available collections.
- Build an interactive CLI loop with commands for ingesting, querying, and listing.
- Test the full end-to-end workflow with your sample data.
Create the list_collections helper
Before building the interactive loop, you need a way to check which collections exist in your ChromaDB database. This small helper wraps the client method and returns just the names.
- In rag_agent.py, scroll to the bottom of the file (below the ask function) and add the following code:
# --- CLI ---
def list_collections() -> list[str]:
collections = chroma_client.list_collections()
return [c.name for c in collections]What does this code do?
The list_collections() function calls ChromaDB's built-in method to get all stored collections. It extracts just the .name from each collection object so you get a clean list of strings like ["ux-design", "game-design"].
- Save rag_agent.py.
Build the main command loop
The CLI needs to run in a loop, prompting for commands until the user exits. Each command maps directly to a function you already built.
- Below the list_collections function, add the main() function and entry point:
def main() -> None:
print("=== Personal RAG Agent ===")
print(f"Chat model: {CHAT_MODEL} | Embedding model: {EMBEDDING_MODEL}")
print()
while True:
print("\nCommands: [i]ngest, [q]uery, [l]ist collections, [e]xit")
cmd = input("> ").strip().lower()
if cmd == "i":
name = input("Collection name to ingest (folder name): ").strip()
if name:
print(f"Ingesting documents from ./documents/{name}/")
ingest_collection(name)
elif cmd == "q":
collections = list_collections()
if not collections:
print("No collections found. Ingest documents first.")
continue
print(f"Available collections: {', '.join(collections)}")
name = input("Which collection? ").strip()
if name not in collections:
print(f"Collection '{name}' not found.")
continue
query = input("Your question: ").strip()
if query:
ask(name, query)
elif cmd == "l":
collections = list_collections()
if collections:
print(f"Collections: {', '.join(collections)}")
else:
print("No collections yet.")
elif cmd == "e":
print("Goodbye!")
break
else:
print("Unknown command.")
if __name__ == "__main__":
main()What does this code do?
The main() function prints a welcome header and enters an infinite loop. Each iteration shows the available commands and waits for input.
The i command asks for a folder name and calls ingest_collection(). The q command lists available collections, prompts for a choice and a question, then calls ask(). The l command displays all collection names. The e command breaks the loop and exits.
The if __name__ == "__main__": guard ensures main() only runs when you execute the file directly, not when importing it.
- Save rag_agent.py.
Seeing a syntax error when you save?
Check that the main() function is at the same indentation level as your other functions (no extra indent). Also verify the if __name__ line is at the top level of the file, not inside main().
help me fix a syntax error in my RAG agent CLI
✔️ 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 is the complete rag_agent.py file at this point:
import os
import sys
from pathlib import Path
from chromadb import PersistentClient
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
from ollama import chat
from pypdf import PdfReader
# --- Configuration ---
OLLAMA_BASE_URL = "http://localhost:11434"
EMBEDDING_MODEL = "nomic-embed-text"
CHAT_MODEL = "qwen3:4b"
CHROMA_DB_PATH = "./chroma_data"
DOCUMENTS_PATH = "./documents"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
# --- Embedding Function ---
ollama_ef = OllamaEmbeddingFunction(
url=OLLAMA_BASE_URL,
model_name=EMBEDDING_MODEL,
)
# --- ChromaDB Client ---
chroma_client = PersistentClient(path=CHROMA_DB_PATH)
# --- Document Loading ---
def load_text_file(file_path: str) -> str:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
def load_pdf_file(file_path: str) -> str:
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
page_text = page.extract_text()
if page_text:
text += page_text + "\n"
return text
def load_document(file_path: str) -> str:
path = Path(file_path)
if path.suffix.lower() == ".pdf":
return load_pdf_file(file_path)
elif path.suffix.lower() in (".txt", ".md"):
return load_text_file(file_path)
else:
print(f"Unsupported file type: {path.suffix}")
return ""
# --- Chunking ---
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if chunk.strip():
chunks.append(chunk.strip())
start = end - overlap
return chunks
# --- Ingestion ---
def ingest_collection(collection_name: str) -> None:
collection_dir = Path(DOCUMENTS_PATH) / collection_name
if not collection_dir.exists():
print(f"Directory not found: {collection_dir}")
return
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
files = list(collection_dir.iterdir())
if not files:
print(f"No files found in {collection_dir}")
return
all_chunks = []
all_ids = []
all_metadatas = []
for file_path in files:
if file_path.is_file():
print(f" Processing: {file_path.name}")
text = load_document(str(file_path))
if not text:
continue
chunks = chunk_text(text)
for i, chunk in enumerate(chunks):
doc_id = f"{file_path.stem}_{i}"
all_chunks.append(chunk)
all_ids.append(doc_id)
all_metadatas.append({
"source": file_path.name,
"chunk_index": i,
"collection": collection_name,
})
if all_chunks:
collection.add(
documents=all_chunks,
ids=all_ids,
metadatas=all_metadatas,
)
print(f" Ingested {len(all_chunks)} chunks into '{collection_name}'")
else:
print(f" No content found to ingest.")
# --- Retrieval ---
def retrieve(collection_name: str, query: str, n_results: int = 3) -> list[dict]:
collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=ollama_ef,
)
results = collection.query(
query_texts=[query],
n_results=n_results,
)
documents = results["documents"][0] if results["documents"] else []
metadatas = results["metadatas"][0] if results["metadatas"] else []
distances = results["distances"][0] if results["distances"] else []
retrieved = []
for doc, meta, dist in zip(documents, metadatas, distances):
retrieved.append({
"text": doc,
"source": meta.get("source", "unknown"),
"distance": dist,
})
return retrieved
# --- Generation ---
def generate_answer(query: str, context_chunks: list[dict]) -> str:
context_text = "\n\n".join(
f"[Source: {chunk['source']}]\n{chunk['text']}"
for chunk in context_chunks
)
system_prompt = (
"You are a helpful assistant. Answer the user's question based ONLY on the "
"provided context. If the context does not contain enough information to answer, "
"say so. Cite the source file when referencing information."
)
user_message = f"Context:\n{context_text}\n\nQuestion: {query}"
response = chat(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
)
return response.message.content
# --- RAG Pipeline ---
def ask(collection_name: str, query: str) -> None:
print(f"\nSearching '{collection_name}' for: {query}")
print("-" * 50)
chunks = retrieve(collection_name, query)
if not chunks:
print("No relevant documents found.")
return
answer = generate_answer(query, chunks)
print(f"\nAnswer:\n{answer}")
print(f"\nSources:")
for chunk in chunks:
print(f" - {chunk['source']} (distance: {chunk['distance']:.4f})")
# --- CLI ---
def list_collections() -> list[str]:
collections = chroma_client.list_collections()
return [c.name for c in collections]
def main() -> None:
print("=== Personal RAG Agent ===")
print(f"Chat model: {CHAT_MODEL} | Embedding model: {EMBEDDING_MODEL}")
print()
while True:
print("\nCommands: [i]ngest, [q]uery, [l]ist collections, [e]xit")
cmd = input("> ").strip().lower()
if cmd == "i":
name = input("Collection name to ingest (folder name): ").strip()
if name:
print(f"Ingesting documents from ./documents/{name}/")
ingest_collection(name)
elif cmd == "q":
collections = list_collections()
if not collections:
print("No collections found. Ingest documents first.")
continue
print(f"Available collections: {', '.join(collections)}")
name = input("Which collection? ").strip()
if name not in collections:
print(f"Collection '{name}' not found.")
continue
query = input("Your question: ").strip()
if query:
ask(name, query)
elif cmd == "l":
collections = list_collections()
if collections:
print(f"Collections: {', '.join(collections)}")
else:
print("No collections yet.")
elif cmd == "e":
print("Goodbye!")
break
else:
print("Unknown command.")
if __name__ == "__main__":
main()Test the full workflow end-to-end
Time to run the complete agent and test every command. You will ingest documents, list your collections, and ask a question to see the full pipeline in action.
- Run the agent by executing this command in your terminal:
python rag_agent.pyYou should see the welcome banner and the command prompt:
=== Personal RAG Agent ===
Chat model: qwen3:4b | Embedding model: nomic-embed-text
Commands: [i]ngest, [q]uery, [l]ist collections, [e]xit
>- Type i and press Enter to start the ingest command.
- When prompted for a collection name, type the name of one of your document folders (for example, ux-design) and press Enter.
You should see each file being processed and a summary of how many chunks were ingested.
- Type l and press Enter to list your collections.
You should see the collection you just ingested listed by name.
- Type q and press Enter to start a query.
- Select the collection you ingested, then ask a question about the content of your documents.
The agent retrieves relevant chunks, sends them to Ollama, and prints a grounded answer with source citations.
- Type e and press Enter to exit the agent.
Why does the first query take a few seconds?
Ollama needs to load the qwen3:4b model into memory on the first query. Subsequent questions in the same session respond much faster because the model stays loaded.
Getting "No collections found" after ingesting?
Make sure your documents folder matches the name you typed. The ingestion looks for a subfolder inside ./documents/ that matches the collection name exactly. Check that the folder contains .pdf, .txt, or .md files.
help me debug why my RAG agent can't find collections
Your personal RAG agent is fully operational. You can now ingest any document, organize your knowledge into collections, and ask questions grounded in your own data. Next up, the Secret Mission challenges you to search across all collections at once.
Secret mission
Cross-Collection Search
Your RAG agent queries one collection at a time. But what if your question spans multiple domains? Add a cross-collection search mode that queries every collection simultaneously, merges results by relevance score, and returns the best chunks regardless of where they came from.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Project folder on your Desktop (contains rag_agent.py, documents/, and chroma_data/).
- Python virtual environment (inside your project folder).
- Ollama models (qwen3:4b and nomic-embed-text).
Keep everything running
No action needed. Choose this if you're still actively building or plan to add more documents to your RAG agent soon.
- Your project folder, ChromaDB data, and Ollama models stay on your machine ready to use.
- Ollama continues running in the background. It uses minimal resources when idle.
Pause - I'll come back to this later
Shut down running processes to free up memory while keeping your files and models intact for later.
- Deactivate your virtual environment by running:
- Stop any running Ollama models. They unload from memory automatically after 5 minutes of inactivity, so no manual action is needed if you simply close your terminal.
deactivateYour project folder, ChromaDB data, and Ollama models remain on disk. When you return, reactivate the virtual environment and start querying again.
Delete - I don't want to use this again
Remove all project resources and start fresh. This frees up disk space used by your documents, vector data, and downloaded models.
- Deactivate your virtual environment first by running:
deactivate- Delete the project folder by running:
macOS
cd ~/Desktop
rm -rf rag-agentWindows
cd ~/Desktop
rmdir /s /q rag-agent- Remove the Ollama models you pulled for this project by running:
ollama rm qwen3:4b
ollama rm nomic-embed-textWhat about Ollama itself?
These commands only remove the models you downloaded. Ollama itself stays installed on your machine. If you use Ollama for other projects, this is what you want.
Nice Work!
Nice Work!
You just built a fully local AI assistant that answers questions grounded in your own documents. No cloud costs, no data leaving your machine, and no API keys to manage.
You've learned how to:
- Build a document ingestion pipeline that chunks PDFs and text files into ChromaDB vector collections with metadata.
- Create a RAG query engine that retrieves relevant context via semantic search and generates grounded answers through Ollama.
- Build a CLI interface for ingesting documents, switching between named collections, and querying your personal knowledge base.
- Secret Mission: Added cross-collection search that queries all collections simultaneously and ranks merged results by relevance.
Ready to quiz yourself?