Agentic RAG with Claude and ChromaDB
Build an AI agent that autonomously searches a vector knowledge base.
Introduction
30 Second Summary
Every time you ask an AI assistant a question about your company's internal docs and it gives you an accurate, cited answer instead of a hallucination, there's a system called RAG working behind the scenes. RAG (Retrieval-Augmented Generation) connects AI models to your actual documents so they answer from facts, not imagination.
In this project, you will build an Agentic RAG Knowledge Base from scratch using Python, ChromaDB, and Claude's tool use. Your AI agent won't just retrieve documents. It will autonomously decide when to search, evaluate whether results are good enough, and re-search with better queries when they're not.
What You'll Build
A Python-based knowledge base agent that combines a local vector database with Claude's native tool use to autonomously search, evaluate, and synthesize answers from your documents with citations.
By the end of this project, you'll have:
- A complete document ingestion pipeline that loads, chunks, and embeds your files into a ChromaDB vector database.
- An agentic tool-use loop where Claude autonomously decides when to search, evaluates relevance, and re-queries when results are poor.
- A working RAG system that answers questions with source citations and handles edge cases honestly.
- Secret Mission: Add a multi-turn conversation mode where the agent remembers previous questions and can reference earlier answers.
Are there any prerequisites?
This project assumes you're comfortable with Python basics (functions, dictionaries, loops) and have made API calls before. If you've completed Parts 1 and 2 of this series, you're ready to go.
Set Up Your Environment and Install Dependencies
Your project folder, virtual environment, and packages are ready from the earlier parts of this series. Now you need to set up the project structure for the RAG system itself.
This means creating the folder where your knowledge base documents will live and the main Python script that ties all your installed packages together.
In this step, get ready to:
- Create the project file structure for your knowledge base.
- Set up your main script with the imports that connect to your packages and API key.
Create the documents folder for your knowledge base
Your RAG system needs a place to store the text files it will search through. You'll create a dedicated documents/ folder inside your project directory.
- In your terminal, make sure you're inside your project folder (the one with your virtual environment active).
- Create the documents folder by running this command:
mkdir documents- Confirm the folder exists by running:
lsYou should see documents listed in the output alongside your .env file and venv folder.
Don't see the documents folder?
Make sure your terminal is inside your project folder. Run pwd (macOS/Linux) or cd (Windows) to print your current directory and confirm you're in the right place.
Help me figure out why my folder didn't get created.
Create the main script with your imports
Every module your RAG system needs is already installed. Now you'll create a single entry-point script that imports them all and loads your API key from the .env file.
- Create a new file called main.py in your project folder.
- Add the following imports and environment loading code:
import os
import json
from dotenv import load_dotenv
import anthropic
import chromadb
load_dotenv()What does this code do?
- os and json are Python standard library modules for file handling and data formatting.
- load_dotenv() reads your .env file and loads ANTHROPIC_API_KEY into the environment so the Anthropic SDK can find it automatically.
- anthropic is the official SDK for calling Claude's API, including tool use.
- chromadb is your local vector database that will store and search document embeddings for free.
- Save main.py.
- Verify the imports work by running:
python main.pyThe script should exit without any errors. If you see no output, that means every import resolved correctly and your API key loaded.
First run might take a moment
ChromaDB may download its default embedding model (all-MiniLM-L6-v2, about 80MB) the first time it's imported. This is a one-time download that runs locally for free. Let it finish before continuing.
Seeing an ImportError?
Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. If it's missing, activate it with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).
Help me fix my import errors when running main.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py and that python main.py exits cleanly with no output.
ⓧ I'd like to double check the full code
import os
import json
from dotenv import load_dotenv
import anthropic
import chromadb
load_dotenv()Your project structure is locked in and every dependency is wired up. Next, you'll fill the documents/ folder with knowledge base content and build the ingestion pipeline that turns raw text into searchable chunks.
Build the Document Ingestion Pipeline
Your environment is ready. Now it's time to give your AI agent something to search through.
A RAG system needs documents broken into small, searchable pieces called chunks. In this step, you'll create sample knowledge base documents, then build functions to load, split, and prepare them for embedding.
In this step, get ready to:
- Create knowledge base documents with distinct topics.
- Build functions to load and chunk documents into searchable pieces.
- Verify the chunking output produces well-formed results.
Create sample knowledge base documents
Your agent needs a realistic knowledge base to search. You'll create four documents covering different topics so you can verify that retrieval returns the right content for each question.
- In the Explorer sidebar of VS Code, right-click the documents folder.
- Select New File and name it company_policies.txt.
- Paste the following content into company_policies.txt:
Remote Work Policy (Effective January 2025)
All full-time employees are eligible for hybrid remote work after completing their 90-day probation period. Employees may work remotely up to 3 days per week, with mandatory in-office days on Tuesday and Thursday for team collaboration.
Remote work requests must be submitted through the HR portal at least 24 hours in advance. Managers retain the right to require in-office presence for specific meetings or projects with 48 hours notice.
Equipment provided for remote work includes a company laptop, external monitor, and ergonomic keyboard. Employees are responsible for maintaining a stable internet connection of at least 50 Mbps download speed.
The company provides a $500 annual stipend for home office improvements, reimbursable through the expense system with receipts. This stipend does not roll over between years.- Create a second file in the documents folder named product_guide.txt.
- Paste this content:
DataSync Pro - Product Documentation (v3.2)
DataSync Pro is our flagship data synchronization platform designed for enterprise customers. It supports real-time bidirectional sync between over 200 data sources including PostgreSQL, MongoDB, Salesforce, and AWS S3.
Key Features:
- Real-time Change Data Capture (CDC) with sub-second latency
- Conflict resolution using last-write-wins, custom merge functions, or manual review
- End-to-end encryption with AES-256 and TLS 1.3
- Automatic schema migration detection and handling
- Built-in data transformation pipeline with 50+ pre-built transforms
Pricing Tiers:
- Starter: Up to 5 connections, 100GB monthly transfer, $299/month
- Professional: Up to 25 connections, 1TB monthly transfer, $899/month
- Enterprise: Unlimited connections and transfer, custom pricing
System Requirements:
- Minimum 4 CPU cores and 16GB RAM for the sync engine
- PostgreSQL 14+ or MongoDB 6+ for the metadata store
- Network bandwidth of at least 100 Mbps between source and target- Create a third file named engineering_standards.txt.
- Paste this content:
Engineering Team Standards and Practices (2025)
Code Review Requirements:
All code changes must receive at least one approval from a senior engineer before merging. Critical path changes (authentication, payments, data migrations) require two approvals. Reviews should be completed within 24 business hours of submission.
Testing Standards:
- Unit test coverage must maintain a minimum of 80% for all new code
- Integration tests are required for any API endpoint changes
- End-to-end tests must pass in the staging environment before production deployment
- Performance tests are required for any database query changes affecting more than 10,000 rows
Deployment Process:
Deployments follow a blue-green strategy with automatic rollback on error rate exceeding 1%. Production deployments are restricted to Monday through Thursday, 10 AM to 2 PM ET. Emergency hotfixes may be deployed outside this window with VP Engineering approval.
Incident Response:
Severity 1 (service down): 15-minute response time, all-hands-on-deck
Severity 2 (degraded service): 1-hour response time, on-call team
Severity 3 (minor issue): Next business day response- Create a fourth file named onboarding_guide.txt.
- Paste this content:
New Employee Onboarding Checklist
Week 1 - Getting Started:
- Complete HR paperwork and benefits enrollment by end of day 1
- Set up company email, Slack, and development environment
- Meet with your manager for 30-minute welcome session
- Attend the Monday all-hands meeting (10 AM ET)
- Complete security awareness training module (takes approximately 2 hours)
Week 2 - Learning the Codebase:
- Clone the main repositories and run the local development setup
- Complete the "Architecture Overview" self-paced course on the learning platform
- Shadow a senior engineer during their daily standup and code review
- Submit your first pull request (can be documentation or a small bug fix)
Week 3 - Contributing:
- Pick up your first ticket from the sprint backlog (labeled "good-first-issue")
- Attend your first sprint planning and retrospective
- Set up your 1:1 cadence with your manager (recommended: weekly, 30 minutes)
Week 4 - Independence:
- Complete your probation checkpoint meeting with your manager
- Begin participating in code reviews for your team
- Identify one area where you'd like mentorship and discuss with your manager
Benefits Overview:
- Health insurance: Coverage begins on day 1, company pays 90% of premiums
- 401(k): 4% company match, eligible after 30 days
- PTO: 20 days annually, plus 10 company holidays
- Parental leave: 16 weeks paid for all new parentsWhy these documents?
Each file covers a distinct topic (HR policies, product details, engineering processes, onboarding). This makes it easy to test whether your retrieval system returns the right document for a given question.
In a real-world system, these would be hundreds or thousands of internal documents. The pattern is the same regardless of scale.
- Save all four files.
- Confirm your files exist by running this in the terminal:
ls documents/You should see company_policies.txt, engineering_standards.txt, onboarding_guide.txt, and product_guide.txt listed in the output.
Don't see all four files?
Make sure you created each file inside the documents folder, not in the project root. Check for typos in the filenames. All four must end with .txt.
help me figure out why my documents folder is missing files
Implement the document loading function
The first step in any RAG pipeline is reading raw documents into memory with their metadata. You need to know where each piece of text came from so you can cite sources later.
- Switch back to main.py in your editor.
- Below the load_dotenv() line, add the section header and the load_documents function by pasting this code:
# ============================================================
# STEP 1: Document Loading and Chunking
# ============================================================
def load_documents(directory: str) -> list[dict]:
"""Load all .txt files from a directory with metadata."""
documents = []
for filename in os.listdir(directory):
if filename.endswith(".txt"):
filepath = os.path.join(directory, filename)
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"source": filename,
})
return documentsWhat does this code do?
- os.listdir(directory) scans the folder for all files, and the .endswith(".txt") filter grabs only text files.
- Each document gets stored as a dictionary with its raw content and a source field tracking the filename. This metadata is how you'll cite sources later.
You won't see a change yet. The next function uses these loaded documents.
Implement the text chunking function
Vector databases work best with small, focused pieces of text. A full document is too large to embed meaningfully, so you split it into overlapping chunks.
What is document chunking?
Chunking splits a long document into smaller pieces (typically 200-1000 characters). The overlap between chunks ensures that sentences split across chunk boundaries still appear in at least one chunk.
Think of it like cutting a book into index cards. Each card holds one idea, and the edges overlap slightly so you never lose a sentence that spans two cards.
- Add the chunk_text function below load_documents by pasting this code:
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
# Try to break at a sentence boundary
if end < len(text):
last_period = chunk.rfind(".")
last_newline = chunk.rfind("\n")
break_point = max(last_period, last_newline)
if break_point > chunk_size * 0.5:
chunk = chunk[: break_point + 1]
end = start + break_point + 1
chunks.append(chunk.strip())
start = end - overlap
return [c for c in chunks if len(c) > 20]What does this code do?
- chunk_size: int = 500 sets each chunk to roughly 500 characters. overlap: int = 50 means adjacent chunks share 50 characters, so no sentence gets lost at the boundary.
- The sentence-boundary logic (rfind(".") and rfind("\n")) tries to cut at a natural stopping point rather than mid-word.
- The final filter (len(c) > 20) removes any tiny leftover fragments that wouldn't contain useful information.
You won't see a change yet. The next function ties loading and chunking together.
Implement the full ingestion pipeline
The ingestion function combines loading and chunking into a single pipeline. It also attaches metadata (source file and chunk index) to each chunk so the agent can cite where it found information.
- Add the ingest_documents function below chunk_text by pasting this code:
def ingest_documents(directory: str) -> list[dict]:
"""Load, chunk, and prepare documents for embedding."""
documents = load_documents(directory)
all_chunks = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append({
"id": f"{doc['source']}_chunk_{i}",
"text": chunk,
"metadata": {
"source": doc["source"],
"chunk_index": i,
},
})
return all_chunksWhat does this code do?
- ingest_documents is the single entry point for your pipeline. It loads all documents, chunks each one, and returns a flat list of chunk dictionaries.
- Each chunk gets a unique id (like company_policies.txt_chunk_0) that ChromaDB will use as a primary key.
- The metadata dictionary tracks which file and which chunk index each piece came from. This is how the agent will cite sources in its answers.
- Save main.py.
✔️ Awesome, I've got everything!
Great. Make sure you've saved main.py before moving on to the verification step.
ⓧ I'd like to double check the full code
Here's the full main.py so far:
import os
import json
from dotenv import load_dotenv
import anthropic
import chromadb
load_dotenv()
# ============================================================
# STEP 1: Document Loading and Chunking
# ============================================================
def load_documents(directory: str) -> list[dict]:
"""Load all .txt files from a directory with metadata."""
documents = []
for filename in os.listdir(directory):
if filename.endswith(".txt"):
filepath = os.path.join(directory, filename)
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"source": filename,
})
return documents
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
# Try to break at a sentence boundary
if end < len(text):
last_period = chunk.rfind(".")
last_newline = chunk.rfind("\n")
break_point = max(last_period, last_newline)
if break_point > chunk_size * 0.5:
chunk = chunk[: break_point + 1]
end = start + break_point + 1
chunks.append(chunk.strip())
start = end - overlap
return [c for c in chunks if len(c) > 20]
def ingest_documents(directory: str) -> list[dict]:
"""Load, chunk, and prepare documents for embedding."""
documents = load_documents(directory)
all_chunks = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append({
"id": f"{doc['source']}_chunk_{i}",
"text": chunk,
"metadata": {
"source": doc["source"],
"chunk_index": i,
},
})
return all_chunksVerify the chunking output
Before moving to the vector database, you should confirm that your pipeline produces well-formed chunks with proper metadata.
- Test the ingestion pipeline by running this command in your terminal:
python -c "from main import ingest_documents; chunks = ingest_documents('documents'); print(f'Total chunks: {len(chunks)}'); print(f'Average size: {sum(len(c[chr(116)+chr(101)+chr(120)+chr(116)]) for c in chunks) // len(chunks)} chars'); print(f'Sample: {chunks[0]}')"What should I see?
You should see output similar to this (exact numbers may vary slightly depending on how the sentence-boundary logic splits your documents):
- Total chunks: somewhere between 10-20 chunks across your 4 documents.
- Average size: roughly 300-500 characters per chunk.
- Sample: a dictionary with id, text, and metadata keys, with the metadata containing source and chunk_index.
Seeing a ModuleNotFoundError?
If you see an error about importing main, make sure your terminal is in the project root folder (the folder containing main.py). Also confirm your virtual environment is still active.
If you see a FileNotFoundError, check that the documents folder is in the same directory as main.py and contains all four .txt files.
help me debug my document ingestion pipeline
Your documents are loaded, split into searchable pieces, and tagged with metadata. Next up, you'll store these chunks in a vector database so your agent can search them by meaning, not just keywords.
Store Embeddings in ChromaDB
Your documents are chunked and ready. Now you need a place to store them where they can be searched by meaning, not just keywords.
A vector database converts text into numerical representations called embeddings, then finds similar text by comparing those numbers. In this step, you will set up ChromaDB as your local vector database and verify that semantic search returns relevant results from your knowledge base documents.
In this step, get ready to:
- Initialize a persistent ChromaDB database that survives between runs.
- Ingest all document chunks into the vector database with automatic embeddings.
- Test semantic search by querying the collection and verifying relevant results.
Initialize ChromaDB with persistent storage
ChromaDB can run in-memory (data lost on exit) or with a persistent client (data saved to disk). You want persistence so you do not have to re-embed documents every time you run the script.
- In main.py, scroll to the bottom of the file (below the ingest_documents function) and add the following section comment and function:
# ============================================================
# STEP 2: ChromaDB Vector Database
# ============================================================
def setup_chromadb(chunks: list[dict]) -> chromadb.Collection:
"""Create/load ChromaDB collection and ingest chunks."""
client = chromadb.PersistentClient(path="./chroma_db")
try:
client.delete_collection("knowledge_base")
except ValueError:
pass
collection = client.create_collection(name="knowledge_base")
collection.add(
ids=[chunk["id"] for chunk in chunks],
documents=[chunk["text"] for chunk in chunks],
metadatas=[chunk["metadata"] for chunk in chunks],
)
print(f"Ingested {len(chunks)} chunks into ChromaDB")
return collectionWhat does this code do?
This function handles the entire database setup and data ingestion in one call:
- PersistentClient(path="./chroma_db") creates a database that saves to disk in a chroma_db folder. This means your embeddings persist between script runs.
- The try/except block deletes any existing collection so you start fresh each run. This avoids duplicate ID errors during development.
- create_collection(name="knowledge_base") creates a new collection. Since no embedding function is specified, ChromaDB uses its default all-MiniLM-L6-v2 model from Sentence Transformers.
- collection.add() inserts all chunks with their text, metadata, and unique IDs. ChromaDB automatically converts each chunk's text into a vector embedding behind the scenes.
Why is the embedding free?
ChromaDB's default embedding function runs the all-MiniLM-L6-v2 model locally on your machine. There are no API calls and no costs. The model files download automatically the first time you run the code.
- Save main.py.
You will not see a change yet. The next substep runs this function to verify it works.
Ingest chunks into the vector database
Now you will run the ingestion to confirm ChromaDB stores your document chunks correctly. ChromaDB handles the embedding automatically using Sentence Transformers. You just pass in the text and it does the rest.
- Add these temporary test lines at the very bottom of main.py (below everything you have written so far):
# Temporary test - delete after verifying
if __name__ == "__main__":
chunks = ingest_documents("documents")
print(f"Loaded {len(chunks)} chunks")
collection = setup_chromadb(chunks)
print("Done! ChromaDB is ready.")- Save main.py, then run it:
python main.pyYou should see output showing the chunk count and a confirmation message. The first run may take 30-60 seconds as ChromaDB downloads the embedding model automatically.
Seeing a long delay or download messages?
The first time ChromaDB runs, it downloads the all-MiniLM-L6-v2 model files (about 80 MB). This is normal and only happens once. Let it finish completely before moving on.
If you see an error about a missing module, make sure your virtual environment is activated and you ran pip install chromadb successfully in Step 1.
Help me debug my ChromaDB setup
Test retrieval with a sample query
Storing embeddings is only useful if you can retrieve relevant results. Semantic search finds chunks by meaning, not exact keywords. A query about "working from home" should match chunks about "remote work" even though the words are different.
- In main.py, add the search_collection function directly below setup_chromadb (and above your temporary test block):
def search_collection(collection: chromadb.Collection, query: str, n_results: int = 3) -> list[dict]:
"""Search the vector database and return results with metadata."""
results = collection.query(
query_texts=[query],
n_results=n_results,
)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"text": results["documents"][0][i],
"source": results["metadatas"][0][i]["source"],
"chunk_index": results["metadatas"][0][i]["chunk_index"],
"distance": results["distances"][0][i],
})
return formatted_resultsWhat does this code do?
This function wraps ChromaDB's query API and returns a clean list of results:
- collection.query(query_texts=[query]) converts your search text into an embedding and finds the most similar chunks in the database.
- n_results controls how many results to return (default 3).
- The distance value indicates how far apart the query and result embeddings are. Lower distance means higher relevance.
- Each result includes the original text, source filename, and chunk index so you can trace answers back to their documents.
- Now update your temporary test block at the bottom of main.py to also test the search function. Replace the existing test block with:
# Temporary test - delete after verifying
if __name__ == "__main__":
chunks = ingest_documents("documents")
print(f"Loaded {len(chunks)} chunks")
collection = setup_chromadb(chunks)
# Test semantic search
print("\nTesting search for 'remote work policy':")
results = search_collection(collection, "remote work policy")
for r in results:
print(f" Source: {r['source']}, Distance: {r['distance']:.4f}")
print(f" Text: {r['text'][:100]}...\n")- Save main.py and run it again:
python main.pyYou should see the top result coming from company_policies.txt with a low distance score. This confirms semantic search is working correctly.
Not seeing relevant results?
If the top result is not from company_policies.txt, make sure your documents folder contains all four .txt files from the previous step.
If you see an error about search_collection not being defined, check that the function is placed above your test block (Python reads top to bottom).
Help me fix my vector search results
- Now that you have confirmed everything works, delete the entire temporary test block (the if __name__ == "__main__": section and everything below it) from main.py.
- Save main.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py with the test block removed.
ⓧ I'd like to double check the full code
Here is the complete main.py at this point (no test block, just the functions):
import os
import json
from dotenv import load_dotenv
import anthropic
import chromadb
load_dotenv()
# ============================================================
# STEP 1: Document Loading and Chunking
# ============================================================
def load_documents(directory: str) -> list[dict]:
"""Load all .txt files from a directory with metadata."""
documents = []
for filename in os.listdir(directory):
if filename.endswith(".txt"):
filepath = os.path.join(directory, filename)
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"source": filename,
})
return documents
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if end < len(text):
last_period = chunk.rfind(".")
last_newline = chunk.rfind("\n")
break_point = max(last_period, last_newline)
if break_point > chunk_size * 0.5:
chunk = chunk[: break_point + 1]
end = start + break_point + 1
chunks.append(chunk.strip())
start = end - overlap
return [c for c in chunks if len(c) > 20]
def ingest_documents(directory: str) -> list[dict]:
"""Load, chunk, and prepare documents for embedding."""
documents = load_documents(directory)
all_chunks = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append({
"id": f"{doc['source']}_chunk_{i}",
"text": chunk,
"metadata": {
"source": doc["source"],
"chunk_index": i,
},
})
return all_chunks
# ============================================================
# STEP 2: ChromaDB Vector Database
# ============================================================
def setup_chromadb(chunks: list[dict]) -> chromadb.Collection:
"""Create/load ChromaDB collection and ingest chunks."""
client = chromadb.PersistentClient(path="./chroma_db")
try:
client.delete_collection("knowledge_base")
except ValueError:
pass
collection = client.create_collection(name="knowledge_base")
collection.add(
ids=[chunk["id"] for chunk in chunks],
documents=[chunk["text"] for chunk in chunks],
metadatas=[chunk["metadata"] for chunk in chunks],
)
print(f"Ingested {len(chunks)} chunks into ChromaDB")
return collection
def search_collection(collection: chromadb.Collection, query: str, n_results: int = 3) -> list[dict]:
"""Search the vector database and return results with metadata."""
results = collection.query(
query_texts=[query],
n_results=n_results,
)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"text": results["documents"][0][i],
"source": results["metadatas"][0][i]["source"],
"chunk_index": results["metadatas"][0][i]["chunk_index"],
"distance": results["distances"][0][i],
})
return formatted_resultsYour vector database is working. You can store documents and retrieve them by meaning. Next up, you will build the agentic tool-use loop that lets Claude decide when and how to search this knowledge base autonomously.
Build the Agentic Tool-Use Loop
You have a working vector database that can store and retrieve document chunks by semantic similarity. Now it's time to build the brain that uses it.
In this step, you'll create the agentic tool-use loop that gives Claude the ability to autonomously search your knowledge base, evaluate results, and re-search when needed. This is the exact pattern used in production AI systems where the model decides what actions to take instead of following a fixed script.
In this step, get ready to:
- Define tool schemas that tell Claude about your knowledge base search functions.
- Implement the agent loop that handles Claude's tool calls and iterates until a final answer.
- Configure a system prompt that drives agentic search-evaluate-refine behavior.
Define the knowledge base search tool
Claude's tool use feature lets you describe functions to Claude using JSON schemas. Claude reads these descriptions and decides when to call each tool based on the user's question.
- In main.py, add the following section header and TOOLS list at the bottom of the file, below your search_collection() function:
# ============================================================
# STEP 3: Agentic Tool-Use Loop
# ============================================================
# Tool definitions for Claude
TOOLS = [
{
"name": "search_knowledge_base",
"description": "Search the company knowledge base for information. Use this tool when you need to find specific facts, policies, procedures, or product details. Returns the most relevant text chunks with their source documents.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. Be specific and use relevant keywords.",
},
"num_results": {
"type": "integer",
"description": "Number of results to return (1-5). Use more for broad questions.",
"default": 3,
},
},
"required": ["query"],
},
},
{
"name": "evaluate_results",
"description": "Evaluate whether search results are relevant enough to answer the user's question. Call this after searching to decide if you need to search again with different terms.",
"input_schema": {
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "The original user question.",
},
"results_summary": {
"type": "string",
"description": "Brief summary of what the search results contain.",
},
"is_relevant": {
"type": "boolean",
"description": "Whether the results contain information that helps answer the question.",
},
},
"required": ["question", "results_summary", "is_relevant"],
},
},
]What does this code do?
You're defining two tools as Python dictionaries that follow the JSON Schema format Claude expects.
- search_knowledge_base tells Claude it can search your vector database. The description helps Claude decide when this tool is appropriate.
- evaluate_results gives Claude a structured way to judge whether search results actually answer the question. This is what enables the self-correction behavior.
- The input_schema tells Claude what parameters each tool accepts. Claude will generate valid inputs matching these schemas.
Now you need to tell Claude how to behave as an agent. The system prompt is what transforms a basic RAG retrieval into an autonomous search-evaluate-refine loop.
- Add the SYSTEM_PROMPT string directly below the TOOLS list:
SYSTEM_PROMPT = """You are a knowledgeable company assistant with access to an internal knowledge base. Your job is to answer questions accurately using information from the knowledge base.
Your workflow for each question:
1. First, determine if the question requires searching the knowledge base. Questions about company policies, products, procedures, or specific internal details require a search. General knowledge questions do not.
2. When searching, formulate a specific search query with relevant keywords.
3. After receiving results, evaluate whether they actually answer the question. If the results are not relevant, try searching with different terms (up to 2 search attempts total).
4. Synthesize your answer from the retrieved information. Always cite your sources by mentioning the document name.
5. If the knowledge base does not contain relevant information after searching, say so honestly rather than making up an answer.
Important: Always cite the source document when providing information from the knowledge base. Format citations as [Source: filename]."""What makes this prompt agentic?
This prompt gives Claude a decision-making framework instead of a fixed instruction. Claude decides IF it needs to search, formulates its OWN queries, and evaluates whether results are good enough.
The key instruction is step 3: "If the results are not relevant, try searching with different terms." This creates the self-correction loop that separates agentic RAG from basic retrieve-and-stuff approaches.
Next, you need a function that actually executes the tools when Claude calls them.
- Add the execute_tool() function below SYSTEM_PROMPT:
def execute_tool(tool_name: str, tool_input: dict, collection: chromadb.Collection) -> str:
"""Execute a tool call and return the result as a string."""
if tool_name == "search_knowledge_base":
results = search_collection(
collection,
tool_input["query"],
tool_input.get("num_results", 3),
)
if not results:
return "No results found for this query."
output = f"Found {len(results)} results:\n\n"
for i, r in enumerate(results, 1):
output += f"Result {i} [Source: {r['source']}]:\n"
output += f"{r['text']}\n"
output += f"(Relevance distance: {r['distance']:.4f})\n\n"
return output
elif tool_name == "evaluate_results":
# This tool is for the agent's own reasoning - just acknowledge it
relevance = "relevant" if tool_input["is_relevant"] else "not relevant"
return f"Evaluation recorded: Results are {relevance} to the question."
return f"Unknown tool: {tool_name}"What does this code do?
This is the bridge between Claude's tool call requests and your actual Python code.
- When Claude calls search_knowledge_base, this function runs your search_collection() function from the previous step and formats the results as a readable string.
- When Claude calls evaluate_results, it simply acknowledges the evaluation. The real value of this tool is in Claude's reasoning process, not the return value.
- Results include the source filename and relevance distance so Claude can judge quality and cite sources.
- Save main.py.
Seeing a syntax error?
Check that the TOOLS list has matching brackets for every opening [ and {. Each tool definition is a nested dictionary, so mismatched braces are the most common issue.
help me find the syntax error in my TOOLS list
Implement the agent loop
The agent loop is the core pattern that makes your system autonomous. It sends a message to Claude, checks if Claude wants to use a tool, executes the tool, sends results back, and repeats until Claude produces a final answer.
- Add the run_agent() function below execute_tool():
def run_agent(question: str, collection: chromadb.Collection) -> str:
"""Run the agentic RAG loop for a single question."""
client = anthropic.Anthropic()
messages = [{"role": "user", "content": question}]
max_iterations = 5
iteration = 0
tool_calls_made = []
print(f"\n{'='*60}")
print(f"Question: {question}")
print(f"{'='*60}")
while iteration < max_iterations:
iteration += 1
response = client.messages.create(
model="claude-sonnet-4-5-20250514",
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=TOOLS,
messages=messages,
)
# Check if Claude wants to use a tool
if response.stop_reason == "tool_use":
# Process all tool use blocks in the response
assistant_content = response.content
tool_results = []
for block in assistant_content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_calls_made.append(tool_name)
print(f" [Agent] Calling tool: {tool_name}")
print(f" Input: {json.dumps(tool_input, indent=2)[:200]}")
result = execute_tool(tool_name, tool_input, collection)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
# Add assistant message and tool results to conversation
messages.append({"role": "assistant", "content": assistant_content})
messages.append({"role": "user", "content": tool_results})
elif response.stop_reason == "end_turn":
# Claude is done - extract the final text response
final_response = ""
for block in response.content:
if hasattr(block, "text"):
final_response += block.text
print(f"\n [Agent] Tools used: {tool_calls_made if tool_calls_made else 'None (answered from general knowledge)'}")
print(f"\n Answer: {final_response}")
return final_response
return "Agent reached maximum iterations without producing a final answer."What does this code do?
This is the agentic loop. Here's how each iteration works:
- Send the conversation (user question + any prior tool results) to Claude with the tools parameter.
- Check response.stop_reason. If it's "tool_use", Claude wants to call a tool. If it's "end_turn", Claude has a final answer.
- When Claude uses a tool, execute it, then append both Claude's request and the tool result to the messages list. This gives Claude the results on the next loop iteration.
- The max_iterations = 5 cap prevents runaway loops if something goes wrong.
Why this pattern over a simpler approach
Basic RAG just embeds the query, grabs the top-3 chunks, and stuffs them into a prompt. The model has no choice in what it retrieves or whether the results are good.
With tool use, Claude decides IF it needs to search, writes its own search query, and can try again with different terms. This produces better answers because the model adapts its search strategy to the question.
- Save main.py.
Seeing an indentation error?
Make sure the entire while loop is indented inside run_agent(). The if response.stop_reason checks should be one level deeper inside the while block.
help me fix the indentation in my run_agent function
Configure the system prompt for agentic behavior
The final piece is a main() function that wires everything together: ingesting documents, setting up ChromaDB, and running test questions through the agent.
- Add the following main() function and entry point at the bottom of main.py:
# ============================================================
# STEP 4: Main Execution
# ============================================================
def main():
print("Agentic RAG Knowledge Base")
print("=" * 60)
# Ingest documents
print("\n[1] Ingesting documents...")
chunks = ingest_documents("documents")
print(f" Loaded {len(chunks)} chunks from documents/")
# Set up ChromaDB
print("\n[2] Setting up ChromaDB vector database...")
collection = setup_chromadb(chunks)
# Test basic retrieval
print("\n[3] Testing basic vector search...")
test_results = search_collection(collection, "remote work policy")
print(f" Query: 'remote work policy'")
print(f" Top result from: {test_results[0]['source']}")
print(f" Distance: {test_results[0]['distance']:.4f}")
# Run the agentic system
print("\n[4] Running Agentic RAG queries...")
questions = [
"How many days per week can employees work remotely?",
"What are the pricing tiers for DataSync Pro?",
"What is the meaning of life?",
"What happens during week 2 of onboarding?",
]
for question in questions:
run_agent(question, collection)
print()
if __name__ == "__main__":
main()What does this code do?
The main() function orchestrates the full pipeline in four stages:
- Stage 1 loads and chunks all documents from the documents/ folder.
- Stage 2 embeds those chunks into ChromaDB.
- Stage 3 runs a quick sanity check with basic vector search.
- Stage 4 runs four test questions through the agent. These questions test different behaviors: answerable questions, unanswerable questions ("meaning of life"), and cross-document retrieval.
- Save main.py.
You won't run the agent yet. In the next step, you'll execute the full system and observe how the agent autonomously searches, evaluates, and refines its queries.
Not sure if everything is wired correctly?
Double check that your functions appear in this order in the file: TOOLS, then SYSTEM_PROMPT, then execute_tool(), then run_agent(), then main(). Python resolves names at call time, but keeping them in logical order makes debugging easier.
help me verify my main.py structure is correct
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py and move on to the next step.
ⓧ I'd like to double check the full code
Here's the complete main.py file at this point:
import os
import json
from dotenv import load_dotenv
import anthropic
import chromadb
load_dotenv()
# ============================================================
# STEP 1: Document Loading and Chunking
# ============================================================
def load_documents(directory: str) -> list[dict]:
"""Load all .txt files from a directory with metadata."""
documents = []
for filename in os.listdir(directory):
if filename.endswith(".txt"):
filepath = os.path.join(directory, filename)
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"source": filename,
})
return documents
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if end < len(text):
last_period = chunk.rfind(".")
last_newline = chunk.rfind("\n")
break_point = max(last_period, last_newline)
if break_point > chunk_size * 0.5:
chunk = chunk[: break_point + 1]
end = start + break_point + 1
chunks.append(chunk.strip())
start = end - overlap
return [c for c in chunks if len(c) > 20]
def ingest_documents(directory: str) -> list[dict]:
"""Load, chunk, and prepare documents for embedding."""
documents = load_documents(directory)
all_chunks = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append({
"id": f"{doc['source']}_chunk_{i}",
"text": chunk,
"metadata": {
"source": doc["source"],
"chunk_index": i,
},
})
return all_chunks
# ============================================================
# STEP 2: ChromaDB Vector Database
# ============================================================
def setup_chromadb(chunks: list[dict]) -> chromadb.Collection:
"""Create/load ChromaDB collection and ingest chunks."""
client = chromadb.PersistentClient(path="./chroma_db")
try:
client.delete_collection("knowledge_base")
except ValueError:
pass
collection = client.create_collection(name="knowledge_base")
collection.add(
ids=[chunk["id"] for chunk in chunks],
documents=[chunk["text"] for chunk in chunks],
metadatas=[chunk["metadata"] for chunk in chunks],
)
print(f"Ingested {len(chunks)} chunks into ChromaDB")
return collection
def search_collection(collection: chromadb.Collection, query: str, n_results: int = 3) -> list[dict]:
"""Search the vector database and return results with metadata."""
results = collection.query(
query_texts=[query],
n_results=n_results,
)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"text": results["documents"][0][i],
"source": results["metadatas"][0][i]["source"],
"chunk_index": results["metadatas"][0][i]["chunk_index"],
"distance": results["distances"][0][i],
})
return formatted_results
# ============================================================
# STEP 3: Agentic Tool-Use Loop
# ============================================================
# Tool definitions for Claude
TOOLS = [
{
"name": "search_knowledge_base",
"description": "Search the company knowledge base for information. Use this tool when you need to find specific facts, policies, procedures, or product details. Returns the most relevant text chunks with their source documents.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. Be specific and use relevant keywords.",
},
"num_results": {
"type": "integer",
"description": "Number of results to return (1-5). Use more for broad questions.",
"default": 3,
},
},
"required": ["query"],
},
},
{
"name": "evaluate_results",
"description": "Evaluate whether search results are relevant enough to answer the user's question. Call this after searching to decide if you need to search again with different terms.",
"input_schema": {
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "The original user question.",
},
"results_summary": {
"type": "string",
"description": "Brief summary of what the search results contain.",
},
"is_relevant": {
"type": "boolean",
"description": "Whether the results contain information that helps answer the question.",
},
},
"required": ["question", "results_summary", "is_relevant"],
},
},
]
SYSTEM_PROMPT = """You are a knowledgeable company assistant with access to an internal knowledge base. Your job is to answer questions accurately using information from the knowledge base.
Your workflow for each question:
1. First, determine if the question requires searching the knowledge base. Questions about company policies, products, procedures, or specific internal details require a search. General knowledge questions do not.
2. When searching, formulate a specific search query with relevant keywords.
3. After receiving results, evaluate whether they actually answer the question. If the results are not relevant, try searching with different terms (up to 2 search attempts total).
4. Synthesize your answer from the retrieved information. Always cite your sources by mentioning the document name.
5. If the knowledge base does not contain relevant information after searching, say so honestly rather than making up an answer.
Important: Always cite the source document when providing information from the knowledge base. Format citations as [Source: filename]."""
def execute_tool(tool_name: str, tool_input: dict, collection: chromadb.Collection) -> str:
"""Execute a tool call and return the result as a string."""
if tool_name == "search_knowledge_base":
results = search_collection(
collection,
tool_input["query"],
tool_input.get("num_results", 3),
)
if not results:
return "No results found for this query."
output = f"Found {len(results)} results:\n\n"
for i, r in enumerate(results, 1):
output += f"Result {i} [Source: {r['source']}]:\n"
output += f"{r['text']}\n"
output += f"(Relevance distance: {r['distance']:.4f})\n\n"
return output
elif tool_name == "evaluate_results":
# This tool is for the agent's own reasoning - just acknowledge it
relevance = "relevant" if tool_input["is_relevant"] else "not relevant"
return f"Evaluation recorded: Results are {relevance} to the question."
return f"Unknown tool: {tool_name}"
def run_agent(question: str, collection: chromadb.Collection) -> str:
"""Run the agentic RAG loop for a single question."""
client = anthropic.Anthropic()
messages = [{"role": "user", "content": question}]
max_iterations = 5
iteration = 0
tool_calls_made = []
print(f"\n{'='*60}")
print(f"Question: {question}")
print(f"{'='*60}")
while iteration < max_iterations:
iteration += 1
response = client.messages.create(
model="claude-sonnet-4-5-20250514",
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=TOOLS,
messages=messages,
)
# Check if Claude wants to use a tool
if response.stop_reason == "tool_use":
# Process all tool use blocks in the response
assistant_content = response.content
tool_results = []
for block in assistant_content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_calls_made.append(tool_name)
print(f" [Agent] Calling tool: {tool_name}")
print(f" Input: {json.dumps(tool_input, indent=2)[:200]}")
result = execute_tool(tool_name, tool_input, collection)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
# Add assistant message and tool results to conversation
messages.append({"role": "assistant", "content": assistant_content})
messages.append({"role": "user", "content": tool_results})
elif response.stop_reason == "end_turn":
# Claude is done - extract the final text response
final_response = ""
for block in response.content:
if hasattr(block, "text"):
final_response += block.text
print(f"\n [Agent] Tools used: {tool_calls_made if tool_calls_made else 'None (answered from general knowledge)'}")
print(f"\n Answer: {final_response}")
return final_response
return "Agent reached maximum iterations without producing a final answer."
# ============================================================
# STEP 4: Main Execution
# ============================================================
def main():
print("Agentic RAG Knowledge Base")
print("=" * 60)
# Ingest documents
print("\n[1] Ingesting documents...")
chunks = ingest_documents("documents")
print(f" Loaded {len(chunks)} chunks from documents/")
# Set up ChromaDB
print("\n[2] Setting up ChromaDB vector database...")
collection = setup_chromadb(chunks)
# Test basic retrieval
print("\n[3] Testing basic vector search...")
test_results = search_collection(collection, "remote work policy")
print(f" Query: 'remote work policy'")
print(f" Top result from: {test_results[0]['source']}")
print(f" Distance: {test_results[0]['distance']:.4f}")
# Run the agentic system
print("\n[4] Running Agentic RAG queries...")
questions = [
"How many days per week can employees work remotely?",
"What are the pricing tiers for DataSync Pro?",
"What is the meaning of life?",
"What happens during week 2 of onboarding?",
]
for question in questions:
run_agent(question, collection)
print()
if __name__ == "__main__":
main()Your agentic loop is fully built. Next up, you'll run the system and watch Claude autonomously search, evaluate, and synthesize answers from your knowledge base.
Test and Demonstrate the Agentic RAG System
Your entire agentic RAG pipeline is built. You have document ingestion, vector storage, tool definitions, and an autonomous agent loop all wired together in main.py.
Now it is time to run the system and watch the agent think. You will observe how Claude autonomously decides when to search, evaluates its own results, and handles questions that the knowledge base cannot answer.
In this step, get ready to:
- Run the full agentic RAG system against four different test questions.
- Compare agentic retrieval with naive RAG to see why autonomous evaluation matters.
- Read the agent's decision-making trace to understand the tool-use loop in action.
Run queries that demonstrate agentic behavior
The main() function in your script runs four carefully chosen questions through the agent. Each question tests a different behavior pattern.
- Run the full agentic RAG system by executing this command in your terminal:
python main.pyWhat are the four test questions?
Each question in the questions list tests a different agent capability:
- "How many days per week can employees work remotely?" tests basic retrieval. The agent should search and find the answer in company_policies.txt.
- "What are the pricing tiers for DataSync Pro?" tests retrieval from a different document. The agent should find the Starter, Professional, and Enterprise tiers in product_guide.txt.
- "What is the meaning of life?" tests the agent's judgment. It should recognize this is general knowledge and answer directly without searching.
- "What happens during week 2 of onboarding?" tests retrieval from the onboarding guide with specific section-level precision.
- Watch the terminal output as the agent processes each question. The first run may take 10-30 seconds as ChromaDB downloads the embedding model.
You should see structured output for each question: the question itself, tool calls the agent makes, a summary of tools used, and the final answer with source citations.
Seeing an authentication error?
If you see an error about an invalid API key, check that your .env file has the correct value for ANTHROPIC_API_KEY and that your virtual environment is activated.
Help me debug my Anthropic API key error when running the agentic RAG system.
Compare agentic RAG with naive RAG
To appreciate what your agent does, consider what a basic RAG pipeline would do instead. Naive RAG follows a fixed three-step path: embed the query, retrieve the top-3 chunks, and cram them into a prompt with no evaluation.
What makes agentic RAG different from naive RAG?
Naive RAG always retrieves, even when the question does not need the knowledge base. It cannot judge whether the retrieved chunks actually answer the question. If the first search returns irrelevant results, naive RAG is stuck with them.
Your agentic version improves on this in three ways:
- It decides whether to search. For "What is the meaning of life?" it skips retrieval entirely.
- It evaluates results using the evaluate_results tool and can re-search with different terms if the first attempt misses.
- It cites sources and honestly reports when information is not available, instead of hallucinating an answer from irrelevant chunks.
- Look at the "meaning of life" question in your terminal output. Check the Tools used: line for that question.
You should see None (answered from general knowledge) printed for that question. The agent recognized it does not need the knowledge base and answered directly. A naive RAG system would have retrieved three irrelevant chunks and tried to answer from them.
Read the agent's decision-making trace
Your run_agent() function prints every tool call the agent makes. This trace is your window into autonomous AI decision-making.
- Scroll through the terminal output and look for the [Agent] Calling tool: lines. These show you exactly which tools Claude called and what inputs it chose.
- For the remote work question, notice the agent called search_knowledge_base with a query it formulated on its own. You did not tell it what to search for.
- Check whether the agent called evaluate_results after any of its searches. This is the self-correction mechanism in action.
What does the tool-use trace reveal?
The stop_reason field in each API response tells your loop what Claude wants to do next. When it equals tool_use, Claude is requesting a tool call. When it equals end_turn, Claude has finished reasoning and is ready to deliver its final answer.
This loop pattern (send message, check stop reason, execute tool, send result back) is the exact architecture used in production AI agents. The only difference in enterprise systems is scale: more tools, better error handling, and observability dashboards over the traces you are reading in your terminal right now.
Why does the agent sometimes skip the evaluate_results tool?
Claude is an LLM making autonomous decisions. Sometimes it judges the search results as clearly relevant and skips the explicit evaluation step. Other times it calls evaluate_results to reason through relevance before answering. Both behaviors are valid. The system prompt encourages evaluation, but does not force it on every query.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py and that your terminal shows all four questions processed with answers.
ⓧ I'd like to double check the full code
Here is the complete main.py file. Your code should match this exactly:
import os
import json
from dotenv import load_dotenv
import anthropic
import chromadb
load_dotenv()
# ============================================================
# STEP 1: Document Loading and Chunking
# ============================================================
def load_documents(directory: str) -> list[dict]:
"""Load all .txt files from a directory with metadata."""
documents = []
for filename in os.listdir(directory):
if filename.endswith(".txt"):
filepath = os.path.join(directory, filename)
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
documents.append({
"content": content,
"source": filename,
})
return documents
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
if end < len(text):
last_period = chunk.rfind(".")
last_newline = chunk.rfind("\n")
break_point = max(last_period, last_newline)
if break_point > chunk_size * 0.5:
chunk = chunk[: break_point + 1]
end = start + break_point + 1
chunks.append(chunk.strip())
start = end - overlap
return [c for c in chunks if len(c) > 20]
def ingest_documents(directory: str) -> list[dict]:
"""Load, chunk, and prepare documents for embedding."""
documents = load_documents(directory)
all_chunks = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
all_chunks.append({
"id": f"{doc['source']}_chunk_{i}",
"text": chunk,
"metadata": {
"source": doc["source"],
"chunk_index": i,
},
})
return all_chunks
# ============================================================
# STEP 2: ChromaDB Vector Database
# ============================================================
def setup_chromadb(chunks: list[dict]) -> chromadb.Collection:
"""Create/load ChromaDB collection and ingest chunks."""
client = chromadb.PersistentClient(path="./chroma_db")
try:
client.delete_collection("knowledge_base")
except ValueError:
pass
collection = client.create_collection(name="knowledge_base")
collection.add(
ids=[chunk["id"] for chunk in chunks],
documents=[chunk["text"] for chunk in chunks],
metadatas=[chunk["metadata"] for chunk in chunks],
)
print(f"Ingested {len(chunks)} chunks into ChromaDB")
return collection
def search_collection(collection: chromadb.Collection, query: str, n_results: int = 3) -> list[dict]:
"""Search the vector database and return results with metadata."""
results = collection.query(
query_texts=[query],
n_results=n_results,
)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"text": results["documents"][0][i],
"source": results["metadatas"][0][i]["source"],
"chunk_index": results["metadatas"][0][i]["chunk_index"],
"distance": results["distances"][0][i],
})
return formatted_results
# ============================================================
# STEP 3: Agentic Tool-Use Loop
# ============================================================
TOOLS = [
{
"name": "search_knowledge_base",
"description": "Search the company knowledge base for information. Use this tool when you need to find specific facts, policies, procedures, or product details. Returns the most relevant text chunks with their source documents.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. Be specific and use relevant keywords.",
},
"num_results": {
"type": "integer",
"description": "Number of results to return (1-5). Use more for broad questions.",
"default": 3,
},
},
"required": ["query"],
},
},
{
"name": "evaluate_results",
"description": "Evaluate whether search results are relevant enough to answer the user's question. Call this after searching to decide if you need to search again with different terms.",
"input_schema": {
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "The original user question.",
},
"results_summary": {
"type": "string",
"description": "Brief summary of what the search results contain.",
},
"is_relevant": {
"type": "boolean",
"description": "Whether the results contain information that helps answer the question.",
},
},
"required": ["question", "results_summary", "is_relevant"],
},
},
]
SYSTEM_PROMPT = """You are a knowledgeable company assistant with access to an internal knowledge base. Your job is to answer questions accurately using information from the knowledge base.
Your workflow for each question:
1. First, determine if the question requires searching the knowledge base. Questions about company policies, products, procedures, or specific internal details require a search. General knowledge questions do not.
2. When searching, formulate a specific search query with relevant keywords.
3. After receiving results, evaluate whether they actually answer the question. If the results are not relevant, try searching with different terms (up to 2 search attempts total).
4. Synthesize your answer from the retrieved information. Always cite your sources by mentioning the document name.
5. If the knowledge base does not contain relevant information after searching, say so honestly rather than making up an answer.
Important: Always cite the source document when providing information from the knowledge base. Format citations as [Source: filename]."""
def execute_tool(tool_name: str, tool_input: dict, collection: chromadb.Collection) -> str:
"""Execute a tool call and return the result as a string."""
if tool_name == "search_knowledge_base":
results = search_collection(
collection,
tool_input["query"],
tool_input.get("num_results", 3),
)
if not results:
return "No results found for this query."
output = f"Found {len(results)} results:\n\n"
for i, r in enumerate(results, 1):
output += f"Result {i} [Source: {r['source']}]:\n"
output += f"{r['text']}\n"
output += f"(Relevance distance: {r['distance']:.4f})\n\n"
return output
elif tool_name == "evaluate_results":
relevance = "relevant" if tool_input["is_relevant"] else "not relevant"
return f"Evaluation recorded: Results are {relevance} to the question."
return f"Unknown tool: {tool_name}"
def run_agent(question: str, collection: chromadb.Collection) -> str:
"""Run the agentic RAG loop for a single question."""
client = anthropic.Anthropic()
messages = [{"role": "user", "content": question}]
max_iterations = 5
iteration = 0
tool_calls_made = []
print(f"\n{'='*60}")
print(f"Question: {question}")
print(f"{'='*60}")
while iteration < max_iterations:
iteration += 1
response = client.messages.create(
model="claude-sonnet-4-5-20250514",
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=TOOLS,
messages=messages,
)
if response.stop_reason == "tool_use":
assistant_content = response.content
tool_results = []
for block in assistant_content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_calls_made.append(tool_name)
print(f" [Agent] Calling tool: {tool_name}")
print(f" Input: {json.dumps(tool_input, indent=2)[:200]}")
result = execute_tool(tool_name, tool_input, collection)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({"role": "assistant", "content": assistant_content})
messages.append({"role": "user", "content": tool_results})
elif response.stop_reason == "end_turn":
final_response = ""
for block in response.content:
if hasattr(block, "text"):
final_response += block.text
print(f"\n [Agent] Tools used: {tool_calls_made if tool_calls_made else 'None (answered from general knowledge)'}")
print(f"\n Answer: {final_response}")
return final_response
return "Agent reached maximum iterations without producing a final answer."
# ============================================================
# STEP 4: Main Execution
# ============================================================
def main():
print("Agentic RAG Knowledge Base")
print("=" * 60)
print("\n[1] Ingesting documents...")
chunks = ingest_documents("documents")
print(f" Loaded {len(chunks)} chunks from documents/")
print("\n[2] Setting up ChromaDB vector database...")
collection = setup_chromadb(chunks)
print("\n[3] Testing basic vector search...")
test_results = search_collection(collection, "remote work policy")
print(f" Query: 'remote work policy'")
print(f" Top result from: {test_results[0]['source']}")
print(f" Distance: {test_results[0]['distance']:.4f}")
print("\n[4] Running Agentic RAG queries...")
questions = [
"How many days per week can employees work remotely?",
"What are the pricing tiers for DataSync Pro?",
"What is the meaning of life?",
"What happens during week 2 of onboarding?",
]
for question in questions:
run_agent(question, collection)
print()
if __name__ == "__main__":
main()Secret mission
Add Multi-Turn Conversation Memory
Your agent answers single questions well, but real assistants hold conversations. Can you build a chat mode where the agent remembers what you discussed earlier and uses that context to give better follow-up answers?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally. The only cost was Anthropic API calls during your active session. There are no running services that accumulate charges when idle.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Project folder containing main.py, agent_chat.py, .env, documents/, chroma_db/, and venv/.
Keep everything running
No action needed. Choose this if you plan to keep building on your Agentic RAG system or want to experiment with new documents and queries.
- Your project folder stays on your Desktop with all files intact.
- The ChromaDB database in chroma_db/ persists your embedded documents for future runs.
- Your virtual environment stays ready to activate whenever you return.
Pause - I'll come back to this later
Shut down running processes to free up memory. Your files stay on disk for later.
- If your virtual environment is still active (you see (venv) in your terminal prompt), deactivate it by running:
deactivate- Close any terminal windows running the agent.
- Your project folder and all files remain on your Desktop for when you return.
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes your code, documents, virtual environment, and the ChromaDB vector database.
- If your virtual environment is still active, deactivate it first by running:
deactivate- Navigate to your Desktop and delete the project folder by running:
cd ~/Desktop
rm -rf agentic-ragWindows users
Use this command instead to remove the project folder:
cd ~\Desktop
rmdir /s /q agentic-rag- Confirm the folder is gone by running:
ls- You should no longer see your project folder in the output.
Nice Work!
Nice Work!
You just built a complete Agentic RAG Knowledge Base from scratch. No frameworks, no abstractions. Every layer of this system is code you wrote and understand.
You've learned how to:
- Build a document ingestion pipeline that loads, chunks with overlap, embeds, and stores documents in ChromaDB using free local Sentence Transformer embeddings.
- Implement Claude's tool-use pattern to build an autonomous agent loop that decides when to search, formulates its own queries, and evaluates retrieval quality.
- Create a self-correcting agentic RAG system that re-searches with refined terms when results are poor and always cites its sources.
- Added multi-turn conversation memory so the agent builds context across questions and references earlier answers.
Ready to quiz yourself?