Build a Multi-Agent RAG System
Build a local multi-agent RAG system with ChromaDB and Ollama.
Introduction
30 Second Summary
When you ask an AI assistant a question about your own files, one model alone rarely gets you the best answer. Production systems use multiple specialized agents that search, reason, and write in a coordinated pipeline.
In this project, you will build a multi-agent RAG system from scratch using pure Python. A Supervisor agent routes your questions to a Knowledge Agent (which searches your documents) and a Writer Agent (which polishes the findings into clear responses).
What You'll Build
A fully local multi-agent orchestration system that loads text documents into a vector store, then uses specialized AI agents to search, analyze, and respond to your questions.
By the end of this project, you'll have:
- A reusable agent loop engine with structured JSON prompting and tool dispatch that powers every agent in the system.
- A document store backed by ChromaDB with local embeddings for semantic search.
- A Supervisor agent that routes queries to specialized Knowledge and Writer agents, chaining their outputs into polished answers.
- Secret Mission: Add conversation memory with a sliding window so the system remembers previous exchanges without overwhelming the model's context.
Are there any prerequisites?
You should be comfortable with Python fundamentals like classes, dicts, JSON, and file I/O.
You also need Ollama installed on your machine with the qwen2.5:3b model already pulled. The project will walk you through pulling the embedding model and installing the Python packages.
Before We Start
Before We Start...
Set Up Your Project
Before your agents can search documents and produce answers, they need the right tools installed. This step gets your environment ready with two Python packages, a dedicated embedding model, and a project folder with sample documents.
Ollama connects your Python code to models running on your machine. ChromaDB handles document storage and semantic search using vector embeddings.
In this step, get ready to:
- Install the ollama and chromadb Python packages.
- Pull the nomic-embed-text embedding model.
- Create the project directory and sample documents for testing.
Install Python packages
- Install the ollama and chromadb packages by running this command:
pip install ollama==0.6.2 chromadb==1.5.8What do these packages do?
The ollama package lets your Python code send prompts to models running on your local Ollama server and receive responses back.
The chromadb package is a vector database. It stores text as numerical vectors and retrieves the most relevant chunks when you run a search query.
- Verify both packages installed correctly by running these commands:
python3 -c "import ollama; print('ollama OK')"
python3 -c "import chromadb; print('chromadb OK')"You should see ollama OK and chromadb OK printed in your terminal.
Seeing an import error?
- Make sure you ran pip install from the same Python environment you are running python3 from.
- If you have multiple Python versions, try pip3 install ollama==0.6.2 chromadb==1.5.8 instead.
- Help me fix pip install errors for ollama or chromadb.
Pull the embedding model
Your agents will search documents by meaning, not by matching keywords. This requires an embedding model that converts text into numerical vectors.
- Pull nomic-embed-text, a small embedding model optimized for vector search (274 MB), by running this command:
ollama pull nomic-embed-textWhy a separate embedding model?
Specialized embedding models like nomic-embed-text produce smaller, more efficient vectors (1024 dimensions) compared to general-purpose LLMs. They are faster and produce better similarity results for retrieval tasks.
- Confirm both models are available by running:
ollama listYou should see both qwen2.5:3b and nomic-embed-text in the output.
Don't see both models?
- If nomic-embed-text is missing, run ollama pull nomic-embed-text again and wait for the download to complete.
- Make sure Ollama is running. Open a separate terminal window and run ollama serve.
- Help me get nomic-embed-text working in Ollama.
Create the project structure
Now you need a project folder and some sample documents. The documents give your agents something to search against when you test the system.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder with a documents/ subdirectory, then move into it by running these commands:
mkdir -p multi-agent-rag/documents
cd multi-agent-rag- Confirm the directory structure by running:
lsYou should see a documents folder in the output.
Next, you'll create two sample text files inside documents/. Each file needs 3-5 paragraphs of content so the chunking system has enough text to work with.
- Create a sample document about Python programming by running this command:
cat > documents/python.txt << 'EOF'
Python is a high-level, general-purpose programming language known for its clean syntax and readability. Created by Guido van Rossum and first released in 1991, Python has grown into one of the most popular languages in the world. Its design philosophy emphasizes code readability and simplicity, making it an excellent choice for beginners and experienced developers alike.
Python supports multiple programming paradigms, including procedural, object-oriented, and functional programming. The language comes with a comprehensive standard library that provides tools for file handling, networking, web development, and more. This batteries-included approach means developers can accomplish many tasks without installing additional packages.
One of Python's greatest strengths is its ecosystem. Package managers like pip make it easy to install thousands of third-party libraries. Popular frameworks like Django and Flask power web applications, while NumPy and pandas are staples in data science. TensorFlow and PyTorch have made Python the dominant language in machine learning and artificial intelligence.
Python is widely used in automation and scripting. System administrators use it to automate repetitive tasks, while data engineers build ETL pipelines with it. The language's simplicity makes it ideal for writing quick scripts that process files, interact with APIs, or manage infrastructure.
EOFWhat does the cat << 'EOF' command do?
This is a heredoc command. It writes everything between the two EOF markers into the specified file. It is a quick way to create a multi-line file directly from the terminal.
- Create a second sample document about coffee by running this command:
cat > documents/coffee.txt << 'EOF'
Coffee is one of the most widely consumed beverages in the world, enjoyed by millions of people every day. The drink is made from roasted coffee beans, which are the seeds of berries from the Coffea plant. Originally discovered in Ethiopia, coffee has a rich history that spans centuries and continents.
The two most common species of coffee are Arabica and Robusta. Arabica beans are known for their smooth, complex flavor and account for about 60 percent of global production. Robusta beans have a stronger, more bitter taste and contain nearly twice the caffeine of Arabica. The flavor of coffee depends on many factors including the growing region, altitude, soil conditions, and processing method.
Brewing coffee is both an art and a science. Popular methods include drip brewing, espresso, French press, pour-over, and cold brew. Each method extracts different compounds from the beans, resulting in distinct flavor profiles. Water temperature, grind size, and brew time all affect the final cup.
The coffee industry is a massive global enterprise. Coffee is the second most traded commodity in the world after crude oil. Major producing countries include Brazil, Vietnam, Colombia, and Ethiopia. The industry supports the livelihoods of an estimated 125 million people worldwide, from farmers to baristas.
EOF- Confirm both files were created by running:
ls documents/You should see coffee.txt and python.txt in the output.
Don't see both files?
- Make sure you are inside the multi-agent-rag directory. Run pwd to check your current location.
- If a file is missing, run the cat > documents/filename.txt << 'EOF' command again for that file.
- Help me create the sample text files in my documents folder.
Use your own documents
The sample content works fine for testing, but feel free to drop any .txt files you want into the documents/ folder. The more text, the more interesting the search results.
Your environment is ready. Next up, you'll build the agent loop engine that powers every agent in the system.
Build the Agent Loop
Your project directory is set up and your sample documents are ready to go. Now it's time to build the engine that powers every agent in this system.
Every AI agent follows the same core pattern: send a prompt, read the response, and decide what to do next. If the model asks to use a tool, the loop executes that tool and feeds the result back. If the model gives a final answer, the loop stops. In this step, you'll build that pattern from scratch as a reusable Python function.
In this step, get ready to:
- Create prompt templates that force the model to respond in structured JSON.
- Build a JSON parser that extracts structured data from raw model output.
- Implement the agent loop with tool dispatch and turn limits.
Create the prompt templates and JSON parser
Most LLM APIs support native tool calling, where the model returns structured function calls as part of its response. Qwen 2.5 3B is too small to use that feature reliably. Instead, you'll use structured JSON prompting: the prompt tells the model to respond only in JSON, and your code parses the response.
Why structured JSON prompting instead of native tool calling?
Qwen 2.5 3B frequently misses the structured tool_calls format, hallucinates arguments, or puts JSON in the wrong field. The 7B variant handles it, but 3B does not.
Structured JSON prompting is also more educational. It teaches what tool calling really is under the hood: text parsing plus a dispatch loop.
- Create a new file called agent_loop.py in your multi-agent-rag folder.
- Add the following imports and prompt templates:
import json
import re
from ollama import chat
TOOL_PROMPT_TEMPLATE = """You are {agent_name}. {agent_description}
You have access to the following tools:
{tool_descriptions}
You MUST respond with valid JSON only. No other text before or after the JSON.
If you need to use a tool, respond with:
{{"action": "tool_call", "tool": "<tool_name>", "args": {{"<arg_name>": "<value>"}}}}
If you have enough information to give a final answer, respond with:
{{"action": "final_answer", "content": "<your answer>"}}
Remember: respond with JSON only. No markdown, no explanation outside the JSON."""
NO_TOOL_PROMPT_TEMPLATE = """You are {agent_name}. {agent_description}
You MUST respond with valid JSON only. No other text before or after the JSON.
Respond with:
{{"action": "final_answer", "content": "<your answer>"}}
Remember: respond with JSON only. No markdown, no explanation outside the JSON."""What do these templates do?
Both templates use Python's .format() method. The placeholders like {agent_name} get filled in later when you create each agent. The double braces {{ produce literal { characters in the output so the JSON examples survive formatting.
- TOOL_PROMPT_TEMPLATE is for agents that have tools. It tells the model to respond with either a tool_call or a final_answer JSON object.
- NO_TOOL_PROMPT_TEMPLATE is for agents with no tools. It only allows final_answer responses.
The model sometimes wraps its JSON in extra text or markdown. You need a parser that can handle messy output.
- Add the parse_json_response() function below the prompt templates:
def parse_json_response(text):
"""Extract and parse JSON from the model's response."""
text = text.strip()
json_match = re.search(r"\{.*\}", text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
return NoneWhat does this function do?
- re.search(r"\{.*\}", text, re.DOTALL) uses a regex to find anything between the first { and last } in the text. The re.DOTALL flag makes .* match across line breaks.
- json.loads() tries to parse the matched text as JSON. If it fails, the function returns None so the agent loop can fall back gracefully.
- Save agent_loop.py.
You won't see a change yet. The next section adds the agent loop function that ties everything together.
Implement the agent loop function
The agent loop is the decision engine. It sends a conversation to Ollama, parses the JSON response, and branches on the action field. Tool calls get executed and fed back into the conversation. Final answers end the loop.
- In agent_loop.py, add the agent_loop() function below parse_json_response():
def agent_loop(
system_prompt, user_message, tools=None, tool_dispatch=None, max_turns=5
):
"""Run an agent loop until the model returns a final answer or hits the turn limit."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
]
for turn in range(max_turns):
response = chat(
model="qwen2.5:3b",
messages=messages,
options={"temperature": 0.1},
)
raw_text = response.message.content
print(f" [Turn {turn + 1}] Raw: {raw_text[:200]}")
parsed = parse_json_response(raw_text)
if parsed is None:
return raw_text
action = parsed.get("action", "")
if action == "final_answer":
return parsed.get("content", raw_text)
if action == "tool_call" and tool_dispatch:
tool_name = parsed.get("tool", "")
tool_args = parsed.get("args", {})
if tool_name in tool_dispatch:
print(f" [Tool Call] {tool_name}({tool_args})")
result = tool_dispatch[tool_name](**tool_args)
result_str = json.dumps(result, default=str)
print(f" [Tool Result] {result_str[:200]}")
messages.append({"role": "assistant", "content": raw_text})
messages.append(
{
"role": "user",
"content": (
f"Tool '{tool_name}' returned: {result_str}\n\n"
"Use this information to continue. Respond with JSON only."
),
},
)
else:
return f"Error: Unknown tool '{tool_name}'"
else:
return parsed.get("content", raw_text)
return "Agent reached the maximum number of turns without a final answer."What does this function do?
This is the core loop that every agent in your system will run through.
- ollama.chat() sends the conversation to Qwen 2.5 3B. The temperature: 0.1 setting keeps the model deterministic so it produces valid JSON and picks the right tools consistently.
- parse_json_response() extracts JSON from the raw text. If parsing fails, the function returns the raw text as a fallback.
- If the action is tool_call, the loop looks up the tool name in tool_dispatch, executes it, and appends the result back into the conversation as a new user message. Then it loops again.
- If the action is final_answer, the loop stops and returns the content.
- max_turns caps the loop at 5 iterations by default. This prevents infinite loops if the model keeps calling tools without ever reaching a final answer.
- Save agent_loop.py.
- Verify the file loads without errors by running:
python3 -c "from agent_loop import agent_loop; print('agent_loop OK')"You should see agent_loop OK printed in your terminal.
Seeing an import error?
- Make sure your terminal is inside the multi-agent-rag folder where agent_loop.py lives.
- If you see ModuleNotFoundError: No module named 'ollama', the ollama pip package is not installed. Run pip install ollama==0.6.2 to install it.
- Check for typos in the function and variable names. Python is case-sensitive.
Still stuck? Ask for help.
✔️ Awesome, I've got everything!
Great. Double check you've saved agent_loop.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete agent_loop.py file. Compare it against yours and fix any differences.
import json
import re
from ollama import chat
TOOL_PROMPT_TEMPLATE = """You are {agent_name}. {agent_description}
You have access to the following tools:
{tool_descriptions}
You MUST respond with valid JSON only. No other text before or after the JSON.
If you need to use a tool, respond with:
{{"action": "tool_call", "tool": "<tool_name>", "args": {{"<arg_name>": "<value>"}}}}
If you have enough information to give a final answer, respond with:
{{"action": "final_answer", "content": "<your answer>"}}
Remember: respond with JSON only. No markdown, no explanation outside the JSON."""
NO_TOOL_PROMPT_TEMPLATE = """You are {agent_name}. {agent_description}
You MUST respond with valid JSON only. No other text before or after the JSON.
Respond with:
{{"action": "final_answer", "content": "<your answer>"}}
Remember: respond with JSON only. No markdown, no explanation outside the JSON."""
def parse_json_response(text):
"""Extract and parse JSON from the model's response."""
text = text.strip()
json_match = re.search(r"\{.*\}", text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
return None
def agent_loop(
system_prompt, user_message, tools=None, tool_dispatch=None, max_turns=5
):
"""Run an agent loop until the model returns a final answer or hits the turn limit."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
]
for turn in range(max_turns):
response = chat(
model="qwen2.5:3b",
messages=messages,
options={"temperature": 0.1},
)
raw_text = response.message.content
print(f" [Turn {turn + 1}] Raw: {raw_text[:200]}")
parsed = parse_json_response(raw_text)
if parsed is None:
return raw_text
action = parsed.get("action", "")
if action == "final_answer":
return parsed.get("content", raw_text)
if action == "tool_call" and tool_dispatch:
tool_name = parsed.get("tool", "")
tool_args = parsed.get("args", {})
if tool_name in tool_dispatch:
print(f" [Tool Call] {tool_name}({tool_args})")
result = tool_dispatch[tool_name](**tool_args)
result_str = json.dumps(result, default=str)
print(f" [Tool Result] {result_str[:200]}")
messages.append({"role": "assistant", "content": raw_text})
messages.append(
{
"role": "user",
"content": (
f"Tool '{tool_name}' returned: {result_str}\n\n"
"Use this information to continue. Respond with JSON only."
),
},
)
else:
return f"Error: Unknown tool '{tool_name}'"
else:
return parsed.get("content", raw_text)
return "Agent reached the maximum number of turns without a final answer."Your agent loop is built and ready to power any agent you throw at it. Next, you'll build the document store that gives your agents something to search.
Build the Document Store
Your agent loop can now talk to a model, parse JSON responses, and dispatch tools. The missing piece is data.
In this step, you will build the RAG infrastructure that gives your agents something to search. You will load text files, split them into overlapping chunks, and store them in ChromaDB using Ollama's nomic-embed-text model for vector embeddings.
In this step, get ready to:
- Create the document store with chunking, loading, and ChromaDB storage.
- Build search and listing functions for semantic retrieval.
Create document_store.py with chunking and storage
The document store handles the full journey of a text file: reading it from disk, splitting it into small overlapping pieces, and storing those pieces as embeddings in a vector database.
Why nomic-embed-text instead of Qwen 2.5 3B for embeddings?
Specialized embedding models like nomic-embed-text produce smaller, more efficient vectors (1024 dimensions vs 4096 for general LLMs). They are faster and produce better similarity results for retrieval tasks.
- Create a new file called document_store.py inside your multi-agent-rag folder (the same folder where agent_loop.py lives).
- Add the imports and the create_document_store() function by copying this code into the file:
import os
import chromadb
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
def create_document_store(collection_name="knowledge_base"):
"""Create a ChromaDB collection with Ollama embeddings."""
# Start an in-memory ChromaDB instance (no server needed)
client = chromadb.Client()
# Connect embeddings to Ollama's nomic-embed-text model
embedding_function = OllamaEmbeddingFunction(
url="http://localhost:11434",
model_name="nomic-embed-text",
)
# Create (or reuse) a collection that auto-embeds documents
collection = client.get_or_create_collection(
name=collection_name,
embedding_function=embedding_function,
)
return collectionWhat does this code do?
- chromadb.Client() creates an in-memory ChromaDB instance. No server to start and no disk paths to manage. Documents are re-embedded each time you run the script, which is fine for small tutorial files.
- OllamaEmbeddingFunction tells ChromaDB to send text to your local Ollama server for embedding instead of using its default model.
- get_or_create_collection() creates a named collection that automatically embeds any document you add to it.
You won't see a change yet. The next chunk adds the functions that fill this store with document data.
- Below the create_document_store() function in document_store.py, add the chunking and loading functions by copying this code:
def chunk_text(text, chunk_size=500, chunk_overlap=50):
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
# Only keep chunks that aren't blank
if chunk.strip():
chunks.append(chunk.strip())
# Slide forward, overlapping by chunk_overlap characters
start += chunk_size - chunk_overlap
return chunks
def load_and_store_documents(collection, directory_path):
"""Load .txt files from a directory, chunk them, and store in ChromaDB."""
doc_count = 0
for filename in os.listdir(directory_path):
if not filename.endswith(".txt"):
continue
filepath = os.path.join(directory_path, filename)
with open(filepath, "r", encoding="utf-8") as f:
text = f.read()
# Split the file's text into overlapping chunks
chunks = chunk_text(text)
for i, chunk in enumerate(chunks):
doc_id = f"{filename}_chunk_{i}"
# Upsert so re-running doesn't create duplicates
collection.upsert(
ids=[doc_id],
documents=[chunk],
metadatas=[{"source": filename, "chunk_index": i}],
)
doc_count += 1
return doc_countWhat does this code do?
- chunk_text() slides a 500-character window across the text, stepping forward by 450 characters each time. The 50-character overlap means sentences at chunk boundaries appear in both the previous and next chunk, so searches don't miss information that falls on a seam.
- load_and_store_documents() reads every .txt file in a directory, chunks each one, and upserts the chunks into ChromaDB. Each chunk gets a unique ID and metadata tracking its source file and position.
- upsert() creates or updates records. This means re-running the script won't create duplicate chunks.
Why overlapping chunks?
Imagine a sentence that falls right at the 500-character boundary. Without overlap, that sentence would be split between two chunks, and neither chunk would contain the full thought. The 50-character overlap ensures boundary sentences appear in both neighboring chunks, so semantic search can still find them.
- Save document_store.py.
You won't see output yet. The next chunk adds the search functions that let agents query this store.
Create the search and listing functions
Your documents are now chunked and stored, but agents need a way to retrieve relevant pieces. These two functions give agents the tools to perform semantic search and discover what sources are available.
- Below the load_and_store_documents() function in document_store.py, add the search and listing functions by copying this code:
def search_documents(collection, query, n_results=3):
"""Search the document store for relevant chunks."""
# Run a semantic similarity search against stored embeddings
results = collection.query(
query_texts=[query],
n_results=n_results,
)
# Results come as lists of lists (one per query), grab the first query's results
documents = results["documents"][0] if results["documents"] else []
metadatas = results["metadatas"][0] if results["metadatas"] else []
# Pair each chunk with its source filename
output = []
for doc, meta in zip(documents, metadatas):
output.append({"text": doc, "source": meta.get("source", "unknown")})
return output
def list_sources(collection):
"""List all unique document sources in the collection."""
# Fetch all metadata without pulling embeddings or documents
all_data = collection.get(include=["metadatas"])
sources = set()
for meta in all_data["metadatas"]:
if "source" in meta:
sources.add(meta["source"])
return sorted(sources)What does this code do?
- search_documents() takes a query string and calls collection.query(). ChromaDB embeds the query using nomic-embed-text, compares it against all stored chunk embeddings, and returns the top 3 most similar chunks.
- ChromaDB returns results as lists of lists (one inner list per input query). Since you always pass one query, you access index [0] to get that query's results.
- list_sources() retrieves metadata for all stored chunks and extracts the unique source filenames. This lets agents discover which documents are available without reading them.
- Save document_store.py.
- Test the full document store by running this command in your terminal:
python3 -c "from document_store import *; c = create_document_store(); n = load_and_store_documents(c, 'documents'); print(f'Loaded {n} chunks'); print(f'Sources: {list_sources(c)}')"You should see output showing the number of chunks loaded and both source files listed. The exact chunk count depends on the length of your sample files.
Seeing a connection error?
- Make sure Ollama is running. Open a separate terminal and run ollama list to check. If Ollama isn't running, start it with ollama serve.
- Check that nomic-embed-text is pulled. Run ollama list and confirm you see nomic-embed-text in the output.
- If you see an ImportError, double-check the OllamaEmbeddingFunction import path matches exactly what's in the code above. The path changed in newer versions of ChromaDB.
Why is my document store test failing?
✔️ Awesome, I've got everything!
Double check that you have saved document_store.py.
ⓧ I'd like to double check the full code
import os
import chromadb
from chromadb.utils.embedding_functions.ollama_embedding_function import (
OllamaEmbeddingFunction,
)
def create_document_store(collection_name="knowledge_base"):
"""Create a ChromaDB collection with Ollama embeddings."""
client = chromadb.Client()
embedding_function = OllamaEmbeddingFunction(
url="http://localhost:11434",
model_name="nomic-embed-text",
)
collection = client.get_or_create_collection(
name=collection_name,
embedding_function=embedding_function,
)
return collection
def chunk_text(text, chunk_size=500, chunk_overlap=50):
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if chunk.strip():
chunks.append(chunk.strip())
start += chunk_size - chunk_overlap
return chunks
def load_and_store_documents(collection, directory_path):
"""Load .txt files from a directory, chunk them, and store in ChromaDB."""
doc_count = 0
for filename in os.listdir(directory_path):
if not filename.endswith(".txt"):
continue
filepath = os.path.join(directory_path, filename)
with open(filepath, "r", encoding="utf-8") as f:
text = f.read()
chunks = chunk_text(text)
for i, chunk in enumerate(chunks):
doc_id = f"{filename}_chunk_{i}"
collection.upsert(
ids=[doc_id],
documents=[chunk],
metadatas=[{"source": filename, "chunk_index": i}],
)
doc_count += 1
return doc_count
def search_documents(collection, query, n_results=3):
"""Search the document store for relevant chunks."""
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 []
output = []
for doc, meta in zip(documents, metadatas):
output.append({"text": doc, "source": meta.get("source", "unknown")})
return output
def list_sources(collection):
"""List all unique document sources in the collection."""
all_data = collection.get(include=["metadatas"])
sources = set()
for meta in all_data["metadatas"]:
if "source" in meta:
sources.add(meta["source"])
return sorted(sources)Your document store is ready to go. It can load files, chunk them, embed them, and search them by meaning. Next up, you will build the agents that actually use these tools to answer questions.
Build the Agent Pipeline
You now have two independent pieces: an agent loop that can talk to a model and dispatch tools, and a document store that can chunk, embed, and search text. On their own, they don't do much.
In this step, you'll wire them together into specialized agents and a Supervisor that coordinates them. The Supervisor receives a user question, routes it to a Knowledge Agent for research, then passes the findings to a Writer Agent for polish.
In this step, get ready to:
- Define tool descriptions that agents can reference.
- Build a Knowledge Agent and a Writer Agent with distinct roles.
- Create a Supervisor that routes queries through the agent pipeline.
Create orchestrator.py with tool definitions
The orchestrator needs to tell each agent what tools are available. You'll define the tools as plain Python dicts and write a helper that formats them into readable text for the system prompt.
- Create a new file called orchestrator.py inside your multi-agent-rag folder, then add the following code:
from agent_loop import agent_loop, TOOL_PROMPT_TEMPLATE, NO_TOOL_PROMPT_TEMPLATE
from document_store import search_documents, list_sources
def build_tool_descriptions(tools):
"""Format tool descriptions for the system prompt."""
lines = []
for tool in tools:
# Build a readable string for each tool's parameters
params = ", ".join(
f'{p["name"]} ({p["type"]}): {p["description"]}'
for p in tool.get("parameters", [])
)
lines.append(f'- {tool["name"]}: {tool["description"]}. Parameters: {params}')
return "\n".join(lines)
# Tool definitions for the Knowledge Agent
KNOWLEDGE_TOOLS = [
{
"name": "search_documents",
"description": "Search the document store for chunks relevant to a query",
"parameters": [
{"name": "query", "type": "string", "description": "The search query"},
],
},
{
"name": "list_sources",
"description": "List all available document sources in the store",
"parameters": [],
},
]What does this code do?
The imports pull in the agent loop engine and both prompt templates you built earlier. They also pull in the document store search and listing functions your agents will call as tools.
build_tool_descriptions() takes a list of tool dicts and formats them into human-readable text. This text gets injected into the system prompt so the model knows what tools it can call and what arguments each tool expects.
KNOWLEDGE_TOOLS defines two tools: search_documents (which takes a query string) and list_sources (which takes no arguments). These are descriptions only. The actual function calls are wired up separately in the agent functions you'll write next.
- Save orchestrator.py.
You won't see a change yet. The next chunk uses these definitions to build the actual agents.
Create the Knowledge and Writer agents
Each agent in the system has one job. The Knowledge Agent searches your documents using tools. The Writer Agent takes raw research findings and produces a polished response with no tools at all.
Why two agents instead of one?
A single agent could search documents and write a polished answer. But splitting the work lets each agent focus on a narrower task, which makes their prompts simpler and their outputs more reliable.
This is the same pattern production frameworks like LangGraph and CrewAI use. Small, focused agents chained together outperform a single monolithic agent.
- In orchestrator.py, add the following two functions below the KNOWLEDGE_TOOLS list:
def run_knowledge_agent(user_message, collection):
"""Run the Knowledge Agent to search documents and extract information."""
# Format tool descriptions into readable text for the prompt
tool_descriptions = build_tool_descriptions(KNOWLEDGE_TOOLS)
system_prompt = TOOL_PROMPT_TEMPLATE.format(
agent_name="Knowledge Agent",
agent_description=(
"Your job is to search the document store and extract relevant information. "
"Use the search_documents tool to find relevant passages, then compile "
"the findings into a clear summary. Always cite which source file the "
"information came from."
),
tool_descriptions=tool_descriptions,
)
# Map tool names to actual functions, binding the collection via lambda
tool_dispatch = {
"search_documents": lambda query: search_documents(collection, query),
"list_sources": lambda: list_sources(collection),
}
return agent_loop(
system_prompt=system_prompt,
user_message=user_message,
tools=KNOWLEDGE_TOOLS,
tool_dispatch=tool_dispatch,
max_turns=5,
)
def run_writer_agent(user_message, context):
"""Run the Writer Agent to produce a polished response from raw findings."""
system_prompt = NO_TOOL_PROMPT_TEMPLATE.format(
agent_name="Writer Agent",
agent_description=(
"Your job is to take raw research findings and produce a clear, "
"well-structured response for the user. Organize the information "
"logically. If sources are mentioned, include them in your response."
),
)
# Combine the raw research with the original question
full_message = (
f"Here are the research findings:\n\n{context}\n\n"
f"Based on these findings, write a clear response to: {user_message}"
)
return agent_loop(
system_prompt=system_prompt,
user_message=full_message,
max_turns=3,
)What does this code do?
run_knowledge_agent() builds a system prompt using TOOL_PROMPT_TEMPLATE and fills in the agent's name, description, and tool list. The tool_dispatch dict maps tool name strings to actual Python functions. Each lambda binds the collection argument so the agent loop only needs to pass the query.
run_writer_agent() uses NO_TOOL_PROMPT_TEMPLATE because the Writer has no tools. It receives the raw research as context and combines it with the original question into a single message. The max_turns=3 cap is lower because the Writer only needs one turn to produce a final answer.
- Save orchestrator.py.
You won't see a change yet. The Supervisor you'll write next is what actually calls these agents.
Create the Supervisor
The Supervisor is the entry point for every user query. It checks whether documents are loaded, routes the question to the Knowledge Agent, then passes the research to the Writer Agent.
- In orchestrator.py, add the following function below run_writer_agent():
def run_supervisor(user_message, collection):
"""The Supervisor routes queries to the appropriate agent pipeline."""
print(f"\n{'=' * 60}")
print(f"Supervisor received: {user_message}")
print(f"{'=' * 60}")
# Check if any documents are loaded
sources = list_sources(collection)
if not sources:
print("\n[Supervisor] No documents loaded. Using Writer Agent directly.")
return run_writer_agent(user_message, "No documents are available.")
print(f"\n[Supervisor] Documents available: {sources}")
print("[Supervisor] Routing to Knowledge Agent...")
# Step 1: Route to Knowledge Agent for research
knowledge_result = run_knowledge_agent(user_message, collection)
print(f"\n[Knowledge Agent Result] {knowledge_result[:300]}")
# Step 2: Route research findings to Writer Agent for polish
print("\n[Supervisor] Routing to Writer Agent...")
final_result = run_writer_agent(user_message, knowledge_result)
print(f"\n[Writer Agent Result] {final_result[:300]}")
return final_resultWhat does this code do?
The Supervisor first calls list_sources() to check whether there are any documents in the store. If there are none, it skips the Knowledge Agent and sends the query directly to the Writer Agent.
When documents exist, the Supervisor runs a two-step pipeline. First it sends the question to the Knowledge Agent, which searches the document store and produces raw findings. Then it takes those findings and passes them to the Writer Agent as context.
The print() statements at each routing step let you see the orchestration flow in the terminal. You'll see exactly which agent is running and what it returns.
- Save orchestrator.py.
- Verify the file has no syntax errors by running this command:
python3 -c "from orchestrator import run_supervisor; print('orchestrator OK')"You should see orchestrator OK printed in the terminal.
Seeing an ImportError or SyntaxError?
- If you see ImportError: cannot import name 'agent_loop', make sure agent_loop.py is in the same folder as orchestrator.py. Both files must live directly inside multi-agent-rag/.
- If you see a SyntaxError, double check that every parenthesis and quote is closed. The lambda expressions in tool_dispatch are easy to mistype.
- Help me debug my orchestrator import error.
✔️ Awesome, I've got everything!
Double check you've saved orchestrator.py and saw orchestrator OK in the terminal.
ⓧ I'd like to double check the full code
Here is the complete orchestrator.py file. Make sure yours matches exactly.
from agent_loop import agent_loop, TOOL_PROMPT_TEMPLATE, NO_TOOL_PROMPT_TEMPLATE
from document_store import search_documents, list_sources
def build_tool_descriptions(tools):
"""Format tool descriptions for the system prompt."""
lines = []
for tool in tools:
params = ", ".join(
f'{p["name"]} ({p["type"]}): {p["description"]}'
for p in tool.get("parameters", [])
)
lines.append(f'- {tool["name"]}: {tool["description"]}. Parameters: {params}')
return "\n".join(lines)
KNOWLEDGE_TOOLS = [
{
"name": "search_documents",
"description": "Search the document store for chunks relevant to a query",
"parameters": [
{"name": "query", "type": "string", "description": "The search query"},
],
},
{
"name": "list_sources",
"description": "List all available document sources in the store",
"parameters": [],
},
]
def run_knowledge_agent(user_message, collection):
"""Run the Knowledge Agent to search documents and extract information."""
tool_descriptions = build_tool_descriptions(KNOWLEDGE_TOOLS)
system_prompt = TOOL_PROMPT_TEMPLATE.format(
agent_name="Knowledge Agent",
agent_description=(
"Your job is to search the document store and extract relevant information. "
"Use the search_documents tool to find relevant passages, then compile "
"the findings into a clear summary. Always cite which source file the "
"information came from."
),
tool_descriptions=tool_descriptions,
)
tool_dispatch = {
"search_documents": lambda query: search_documents(collection, query),
"list_sources": lambda: list_sources(collection),
}
return agent_loop(
system_prompt=system_prompt,
user_message=user_message,
tools=KNOWLEDGE_TOOLS,
tool_dispatch=tool_dispatch,
max_turns=5,
)
def run_writer_agent(user_message, context):
"""Run the Writer Agent to produce a polished response from raw findings."""
system_prompt = NO_TOOL_PROMPT_TEMPLATE.format(
agent_name="Writer Agent",
agent_description=(
"Your job is to take raw research findings and produce a clear, "
"well-structured response for the user. Organize the information "
"logically. If sources are mentioned, include them in your response."
),
)
full_message = (
f"Here are the research findings:\n\n{context}\n\n"
f"Based on these findings, write a clear response to: {user_message}"
)
return agent_loop(
system_prompt=system_prompt,
user_message=full_message,
max_turns=3,
)
def run_supervisor(user_message, collection):
"""The Supervisor routes queries to the appropriate agent pipeline."""
print(f"\n{'=' * 60}")
print(f"Supervisor received: {user_message}")
print(f"{'=' * 60}")
sources = list_sources(collection)
if not sources:
print("\n[Supervisor] No documents loaded. Using Writer Agent directly.")
return run_writer_agent(user_message, "No documents are available.")
print(f"\n[Supervisor] Documents available: {sources}")
print("[Supervisor] Routing to Knowledge Agent...")
knowledge_result = run_knowledge_agent(user_message, collection)
print(f"\n[Knowledge Agent Result] {knowledge_result[:300]}")
print("\n[Supervisor] Routing to Writer Agent...")
final_result = run_writer_agent(user_message, knowledge_result)
print(f"\n[Writer Agent Result] {final_result[:300]}")
return final_resultYour agent pipeline is complete. Next up, you'll wire everything into an interactive terminal and test the full system with real questions.
Wire Everything Together
Your agent pipeline is fully built. The Supervisor knows how to route queries to the Knowledge Agent, collect its findings, and pass them to the Writer Agent for a polished answer.
All that is missing is an entry point that loads your documents into ChromaDB and lets you talk to the system from the terminal. In this step, you will create main.py and run the full multi-agent RAG system end to end.
In this step, get ready to:
- Create the main.py entry point that initializes the document store and starts an interactive loop.
- Run the system and test it with single-document, cross-document, and out-of-scope questions.
Create the entry point in main.py
The entry point ties your three modules together. It initializes the document store, loads your text files, then drops you into an interactive prompt where every question flows through the Supervisor pipeline.
- Create a new file called main.py inside your multi-agent-rag folder, then paste the following code:
import os
from document_store import create_document_store, load_and_store_documents
from orchestrator import run_supervisor
def main():
# Initialize ChromaDB and load documents
print("Initializing document store...")
collection = create_document_store()
# Load text files if the documents/ folder exists
docs_dir = "documents"
if os.path.exists(docs_dir):
print(f"Loading documents from '{docs_dir}/'...")
count = load_and_store_documents(collection, docs_dir)
print(f"Loaded {count} chunks from documents.\n")
else:
print(f"No '{docs_dir}/' directory found. Create it and add .txt files.")
print("Continuing without documents.\n")
print("Multi-Agent RAG System Ready!")
print("Type 'quit' to exit.\n")
# Interactive loop: read input, route through Supervisor, print result
while True:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not user_input:
continue
result = run_supervisor(user_input, collection)
print(f"\nAssistant: {result}\n")
if __name__ == "__main__":
main()What does this code do?
- create_document_store() spins up an in-memory ChromaDB collection with the nomic-embed-text embedding model.
- load_and_store_documents() reads every .txt file in documents/, chunks them, and embeds each chunk into the collection.
- The while True loop reads user input, passes it to run_supervisor(), and prints the final answer. Typing quit, exit, or q breaks the loop.
- The if __name__ == "__main__" guard ensures main() only runs when you execute the file directly, not when it is imported.
- Save main.py.
You will not see a change yet. The next substep runs the file and shows the full orchestration in action.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py before moving on.
ⓧ I'd like to double check the full code
Here is the full main.py for comparison.
import os
from document_store import create_document_store, load_and_store_documents
from orchestrator import run_supervisor
def main():
print("Initializing document store...")
collection = create_document_store()
docs_dir = "documents"
if os.path.exists(docs_dir):
print(f"Loading documents from '{docs_dir}/'...")
count = load_and_store_documents(collection, docs_dir)
print(f"Loaded {count} chunks from documents.\n")
else:
print(f"No '{docs_dir}/' directory found. Create it and add .txt files.")
print("Continuing without documents.\n")
print("Multi-Agent RAG System Ready!")
print("Type 'quit' to exit.\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not user_input:
continue
result = run_supervisor(user_input, collection)
print(f"\nAssistant: {result}\n")
if __name__ == "__main__":
main()Run and test the system
Time to see the full pipeline in action. You will ask questions and watch the Supervisor route them through the Knowledge Agent and Writer Agent in real time.
- Make sure Ollama is running in the background. If it is not, open a separate terminal and start it by running:
ollama serve- In your main terminal (inside the multi-agent-rag folder), start the system by running:
python3 main.pyYou should see initialization output showing the number of chunks loaded from your documents. Then a You: prompt appears, waiting for your input.
What should I see?
The exact chunk count depends on how long your sample files are. With 3-5 paragraphs each, you might see something like Loaded 8 chunks from documents. followed by Multi-Agent RAG System Ready!
Seeing a ModuleNotFoundError?
- Make sure you are running python3 main.py from inside the multi-agent-rag folder where all four .py files live.
- Confirm that agent_loop.py, document_store.py, and orchestrator.py all exist in the same directory as main.py.
- Check that the ollama and chromadb packages are installed in the same Python environment you are running.
Why is my main.py failing to import the other modules?
- At the You: prompt, type a question about one of your sample documents. For example:
What is Python used for?Watch the terminal output. You will see the Supervisor receive the message, route to the Knowledge Agent, a tool call to search_documents, the raw search results, and then the Writer Agent producing the final polished response.
What is happening in the terminal output?
Each [Supervisor] line shows the routing decisions. The [Tool Call] and [Tool Result] lines show the Knowledge Agent calling search_documents and receiving matching chunks from ChromaDB.
The [Turn N] Raw: lines show the raw JSON responses from Qwen 2.5 3B. This is the structured JSON prompting pattern in action. The agent loop parses these responses and decides whether to call another tool or return a final answer.
- Try a cross-document question that touches both sample files. For example:
What topics do my documents cover?The Knowledge Agent searches across both documents and the Writer Agent combines the findings into one response.
- Try a question that is not covered by any of your documents. For example:
What is the capital of France?The Knowledge Agent still searches, but the retrieved chunks will not be relevant. The Writer Agent should produce a response noting the limited information available.
Responses vary with a 3B model
Qwen 2.5 3B is a small model. Responses may sometimes be incomplete, oddly formatted, or take a few seconds per turn. This is normal for a 3B parameter model running locally.
If the agent loop hits the maximum number of turns, try rephrasing your question with simpler wording. Shorter, more specific questions tend to produce better results.
Getting connection errors or timeouts?
- Make sure Ollama is running. In a separate terminal, run ollama serve if it is not already started.
- Verify both models are pulled by running ollama list. You should see both qwen2.5:3b and nomic-embed-text.
- If the first query is slow, that is Ollama loading the model into memory. Subsequent queries should be faster.
Why is my multi-agent system timing out or failing to connect to Ollama?
- When you are done testing, type quit to exit the system.
Secret mission
Add Conversation Memory
Right now, your multi-agent system treats every question as a brand new conversation. Ask about Python, then say "tell me more about that," and the system has no idea what "that" refers to. Your mission: add a sliding window memory so the Supervisor remembers recent exchanges without overloading the 3B model's context.
Before you go
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.
Resources you used:
- Project files: agent_loop.py, document_store.py, orchestrator.py, main.py.
- Sample documents in the documents/ folder.
- Python packages: ollama and chromadb.
- Ollama models: qwen2.5:3b and nomic-embed-text.
- ChromaDB in-memory store (ephemeral, cleared automatically when the script exits).
Keep everything running
No action needed. Choose this if you are still actively building or experimenting with your multi-agent system.
- Your project files stay in the multi-agent-rag/ folder on your Desktop.
- Ollama continues running in the background and serving models when requested.
- The Python packages remain installed in your environment.
Pause - I'll come back to this later
Shut down Ollama to free up memory. Your project files and installed packages stay in place for when you return.
- Stop the Ollama server by running:
ollama stop- When you are ready to come back, start Ollama again and run python3 main.py from the multi-agent-rag/ folder.
Delete - I don't want to use this again
Remove all project resources and start fresh.
- Delete the project folder by running:
cd ~/Desktop
rm -rf multi-agent-rag- Uninstall the Python packages by running:
pip uninstall ollama chromadb -y- Remove the embedding model from Ollama by running:
ollama rm nomic-embed-text- If you also want to remove the qwen2.5:3b model and are not using it for other projects, run:
ollama rm qwen2.5:3bMission accomplished
That's a Wrap!
Nice work! You just built a multi-agent RAG orchestration system from scratch. No LangGraph, no CrewAI, no frameworks.
You've learned how to:
- Build a reusable agent loop engine with structured JSON prompting and tool dispatch to power AI agents without relying on native tool calling.
- Create a RAG pipeline that loads documents into ChromaDB with Ollama embeddings for semantic search.
- Wire together a multi-agent pipeline with a Supervisor that routes queries to specialized Knowledge and Writer agents and chains their outputs together.
- Add conversation memory with a sliding window so the system remembers previous exchanges without overwhelming the model's context.
Ready to quiz yourself?