Build an Agentic RAG Assistant
Build a RAG pipeline and AI agent with Gemini and ChromaDB.
Introduction
30 Second Summary
AI chatbots are great for general questions, but they fall short when you need answers from your own notes, research, or documentation. What if your AI could actually read your files and decide on its own how to help you?
In this project, you will build a terminal-based AI research assistant that retrieves answers from your own documents and autonomously decides which tools to use. You will combine two of the most in-demand AI engineering patterns: Retrieval-Augmented Generation (RAG) and AI agents.
What You'll Build
A Python-based agentic RAG assistant that ingests documents into a local vector database and uses Gemini's function calling to search your knowledge base, perform calculations, or respond directly.
By the end of this project, you'll have:
- A complete RAG pipeline that loads, chunks, embeds, and retrieves documents from ChromaDB.
- An AI agent with custom tool definitions that autonomously decides when to search documents or perform calculations.
- A multi-turn chat interface with conversation history running in your terminal.
- Secret Mission: Add a web search tool using Google Search grounding for questions beyond your loaded documents.
Are there any prerequisites?
You should be comfortable writing Python (functions, loops, dictionaries) and have a basic understanding of how APIs work. No prior experience with RAG, vector databases, or AI agents is required.
Set Up Your Environment and API Key
Before you can build a RAG pipeline or an AI agent, you need the right tools in place. Your assistant will rely on the Gemini API for language generation and ChromaDB for storing document embeddings locally.
In this step, you will create a Python project, install the required packages, connect to the Gemini API, and prepare sample documents for your knowledge base.
In this step, get ready to:
- Create a project directory with a virtual environment and install dependencies.
- Get a free Gemini API key and verify your connection.
- Create sample markdown documents for your RAG knowledge base.
Create your project directory and install packages
You need a dedicated folder for this project and an isolated Python environment to keep your dependencies clean.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir rag-assistant
cd rag-assistant- Confirm the folder exists by running:
lsYou should see the rag-assistant folder listed if you look at your Desktop.
Now create a virtual environment to isolate your project's packages from your system Python.
- Create and activate your virtual environment by following the instructions for your operating system:
macOS
- Create the virtual environment by running:
python3 -m venv venv- Activate the virtual environment by running:
source venv/bin/activateYou should see (venv) appear at the start of your terminal prompt.
Windows
- Create the virtual environment by running:
python -m venv venv- Activate the virtual environment by running:
venv\Scripts\activateYou should see (venv) appear at the start of your terminal prompt.
What is a virtual environment?
A virtual environment is an isolated Python installation that keeps your project's packages separate from your system Python. This prevents version conflicts between different projects on your machine.
- Install the three packages your project needs by running:
pip install google-genai chromadb python-dotenvWhat are these packages?
google-genai is Google's official SDK for the Gemini API. It handles authentication and model calls.
chromadb is a local vector database that stores your document embeddings and handles semantic search.
python-dotenv loads environment variables from a .env file so you can keep your API key out of your code.
The install will download several dependencies. This may take a minute or two.
- Verify the installation succeeded by running:
python -c "from google import genai; import chromadb; print('Setup complete')"You should see Setup complete printed in your terminal.
Seeing a ModuleNotFoundError?
Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. If not, run the activate command again from the platform tab above.
If the error persists, try running pip install google-genai chromadb python-dotenv again.
Help me troubleshoot my package installation
Get your Gemini API key and create the test script
The Gemini API gives your assistant the ability to understand and generate text. You need an API key to authenticate your requests.
- Open your browser and navigate to:
https://aistudio.google.com/app/apikey- Sign in with your Google account if prompted.
- Click Create API Key.
- Select a Google Cloud project (or let it create one for you) and click Create API key in existing project.
- Copy the generated API key. You will only see it once at creation time, so copy it now.
Why is this free?
Google offers a free tier for the Gemini API that provides free input and output tokens with no credit card required. The gemini-2.5-flash model used in this project is available on the free tier.
Now store your API key safely in a .env file so your code can load it without hardcoding secrets.
- Create a .env file in your rag-assistant folder with the following content:
GEMINI_API_KEY=your-api-key-here- Replace your-api-key-here with the API key you copied from Google AI Studio.
Keep your .env file secret
Never commit your .env file to version control. If you use Git, add .env to your .gitignore file.
Next, create the test script that verifies your API connection.
- Create a main.py file in your rag-assistant folder and add the following code:
import os
from dotenv import load_dotenv
from google import genai
load_dotenv()
api_key = os.environ.get('GEMINI_API_KEY')
client = genai.Client(api_key=api_key)
response = client.models.generate_content(
model='gemini-2.5-flash',
contents='Say hello and confirm you are working.'
)
print(response.text)What does this code do?
load_dotenv() reads the .env file and loads GEMINI_API_KEY into your environment variables.
genai.Client(api_key=api_key) creates a client that authenticates all your requests to the Gemini API.
client.models.generate_content() sends a prompt to gemini-2.5-flash and returns the model's response.
- Save main.py and run it:
python main.pyYou should see a friendly response from Gemini printed in your terminal, confirming your API key and SDK are working.
Seeing an error about GEMINI_API_KEY?
Double check that your .env file is in the same folder as main.py (your rag-assistant folder). Make sure there are no spaces around the = sign in your .env file.
If you see a permission error from Google, verify you copied the full API key without any trailing spaces.
Help me debug my Gemini API connection
✔️ Awesome, I've got everything!
Great. Make sure you have saved main.py and your .env file before moving on.
ⓧ I'd like to double check the full code
Here is the complete main.py file:
import os
from dotenv import load_dotenv
from google import genai
load_dotenv()
api_key = os.environ.get('GEMINI_API_KEY')
client = genai.Client(api_key=api_key)
response = client.models.generate_content(
model='gemini-2.5-flash',
contents='Say hello and confirm you are working.'
)
print(response.text)And your .env file should look like this (with your actual key):
GEMINI_API_KEY=your-api-key-hereCreate your document knowledge base
Your RAG assistant needs documents to search through. You will create a docs/ folder with three markdown files covering distinct topics.
- Create the docs folder inside your rag-assistant directory by running:
mkdir docs- Create docs/python_best_practices.md with the following content:
# Python Best Practices
## Code Style
Follow PEP 8 for consistent code formatting. Use meaningful variable names that describe the purpose of the variable. Keep functions small and focused on a single responsibility.
## Error Handling
Always use specific exception types rather than catching generic Exception. Use try/except blocks around code that might fail, especially I/O operations and network calls. Log errors with context information.
## Virtual Environments
Always use virtual environments to isolate project dependencies. Create a requirements.txt or pyproject.toml to document dependencies. Use pip freeze to capture exact versions for reproducibility.
## Testing
Write unit tests for all public functions. Use pytest as your testing framework. Aim for at least 80% code coverage. Mock external dependencies in tests to keep them fast and reliable.- Create docs/cloud_computing.md with the following content:
# Cloud Computing Fundamentals
## What is Cloud Computing
Cloud computing delivers computing services over the internet including servers, storage, databases, networking, and software. The three main service models are IaaS, PaaS, and SaaS.
## Key Benefits
Cloud computing offers scalability, cost efficiency through pay-as-you-go pricing, high availability with redundant systems, and global reach through distributed data centers.
## Major Providers
The three largest cloud providers are AWS (Amazon Web Services), Microsoft Azure, and Google Cloud Platform. Each offers hundreds of services spanning compute, storage, AI, and more.
## Serverless Computing
Serverless computing lets you run code without managing servers. Functions execute in response to events and scale automatically. You only pay for actual compute time used. Popular services include AWS Lambda, Azure Functions, and Google Cloud Functions.- Create docs/data_engineering.md with the following content:
# Data Engineering Concepts
## ETL vs ELT
ETL (Extract, Transform, Load) transforms data before loading into the target. ELT (Extract, Load, Transform) loads raw data first, then transforms in place. ELT is preferred for modern cloud data warehouses because they have powerful compute for transformations.
## Data Warehouses
A data warehouse is a centralized repository for structured, filtered data. It serves analytical queries and business intelligence. Modern examples include Snowflake, BigQuery, and Redshift.
## Medallion Architecture
The medallion architecture organizes data into Bronze (raw), Silver (cleaned), and Gold (business-ready) layers. Bronze holds raw ingested data. Silver applies cleaning and conforming. Gold contains aggregated, business-level tables ready for analytics.
## Orchestration
Data orchestration tools manage the execution order and dependencies of data pipelines. Apache Airflow is the most popular open-source orchestrator. It uses DAGs (Directed Acyclic Graphs) to define workflow dependencies.Why three separate topics?
Having documents on distinct subjects (Python, cloud, data engineering) makes it easy to verify that your RAG pipeline retrieves the correct document for a given question. If you ask about "medallion architecture," only the data engineering file should come back.
- Verify your docs folder has all three files by running:
ls docs/You should see cloud_computing.md, data_engineering.md, and python_best_practices.md listed.
Don't see all three files?
Make sure you saved each file inside the docs/ folder (not in the root rag-assistant folder). Check the filenames match exactly, including the .md extension.
Help me check my docs folder structure
Your environment is ready and your knowledge base has content to search. Next up, you will build the RAG pipeline that ingests these documents, splits them into chunks, and stores their embeddings in ChromaDB for semantic search.
Build the RAG Pipeline
Your Gemini connection is live. Now it's time to give your assistant something to be smart about.
Right now, if you ask Gemini a question about your documents, it has no idea they exist. Retrieval-Augmented Generation (RAG) solves this by feeding relevant snippets from your own files into the prompt so the model can answer with grounded, accurate context.
In this step, get ready to:
- Load and chunk your markdown documents into smaller pieces.
- Store those chunks as vector embeddings in a local ChromaDB database.
- Search the database with a natural language query and retrieve relevant results.
Load and chunk your documents
Large language models work best when they receive focused, relevant context. Feeding an entire document into a prompt wastes tokens and dilutes the answer.
The solution is chunking: splitting documents into smaller overlapping pieces so only the most relevant piece gets retrieved later.
- Create a new file called rag.py in your project directory (the same folder that contains main.py).
- Add the imports and write the load_documents() and chunk_text() functions by pasting this code:
import os
import chromadb
def load_documents(docs_dir="docs"):
"""Load all markdown files from the docs directory."""
documents = []
for filename in os.listdir(docs_dir):
if filename.endswith(".md"):
filepath = os.path.join(docs_dir, 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, chunk_size=500, overlap=50):
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start += chunk_size - overlap
return chunksWhat does this code do?
load_documents() scans the docs/ folder for any file ending in .md. It reads each file's content and stores it alongside the filename as metadata so you know where each piece of text originated.
chunk_text() slides a 500-character window across the text, stepping forward 450 characters each time. The 50-character overlap ensures no sentence gets cut off at a boundary without context on the other side.
You won't see output yet. These two helper functions will be called by the ingestion function you write next.
Ingest documents into ChromaDB
ChromaDB is a vector database that stores text as numerical representations called embeddings. When you search later, it compares your query's embedding against stored embeddings to find the closest matches.
- Add the ingest_documents() function below chunk_text() in rag.py:
def ingest_documents(docs_dir="docs"):
"""Load, chunk, and store documents in ChromaDB."""
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(name="research_docs")
documents = load_documents(docs_dir)
all_chunks = []
all_ids = []
all_metadatas = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
chunk_id = f"{doc['source']}_chunk_{i}"
all_chunks.append(chunk)
all_ids.append(chunk_id)
all_metadatas.append({"source": doc["source"], "chunk_index": i})
collection.upsert(documents=all_chunks, ids=all_ids, metadatas=all_metadatas)
print(f"Ingested {len(all_chunks)} chunks from {len(documents)} documents.")
return collectionWhat does this code do?
- chromadb.PersistentClient(path="./chroma_db") creates a local database in a chroma_db folder. Data persists between script runs so you don't re-embed every time.
- get_or_create_collection(name="research_docs") creates a named collection (like a table) for your document chunks. If it already exists, it reuses it.
- collection.upsert() inserts new chunks or updates existing ones by ID. ChromaDB automatically generates embeddings using its default local model (all-MiniLM-L6-v2).
You won't see output yet. The next substep adds a search function and a test runner that exercises everything.
Search the vector database
The final piece is semantic search. Instead of matching keywords, ChromaDB converts your question into an embedding and finds the stored chunks closest to it in meaning.
- Add the search_documents() function and the test block below ingest_documents() in rag.py:
def search_documents(query, n_results=3):
"""Search the document collection for relevant chunks."""
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(name="research_docs")
results = collection.query(query_texts=[query], n_results=n_results)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append(
{
"content": results["documents"][0][i],
"source": results["metadatas"][0][i]["source"],
"distance": results["distances"][0][i],
}
)
return formatted_results
if __name__ == "__main__":
ingest_documents()
print("\nTesting search...")
results = search_documents("What is medallion architecture?")
for r in results:
print(f"\n[{r['source']}] (distance: {r['distance']:.4f})")
print(r["content"][:200])What does this code do?
search_documents() takes a plain English query, converts it to an embedding behind the scenes, and returns the 3 closest matching chunks from your collection. Each result includes the chunk text, source filename, and a distance score (lower means more relevant).
The if __name__ == "__main__" block at the bottom runs only when you execute rag.py directly. It ingests documents first, then searches for "What is medallion architecture?" to prove retrieval works.
- Save rag.py.
- Run the RAG pipeline by executing this command in your terminal:
python rag.pyYou should see an ingestion summary followed by search results. The top result should come from data_engineering.md with a low distance score, since that file contains the medallion architecture content.
First run takes longer
On the first run, ChromaDB downloads a small embedding model (all-MiniLM-L6-v2) to your machine. This happens automatically and only needs to happen once. Subsequent runs will be faster.
Not seeing search results?
- Make sure your docs/ folder contains .md files and that they have actual text content inside.
- Check that you're running the command from the project directory (the folder containing both rag.py and the docs/ folder).
- If you see a permissions error about the chroma_db folder, make sure no other process has it locked.
Still stuck? Help me debug why rag.py isn't returning search results.
✔️ Awesome, I've got everything!
Great. Double check you've saved rag.py and confirm the output shows chunks ingested from 3 documents with relevant search results.
ⓧ I'd like to double check the full code
import os
import chromadb
def load_documents(docs_dir="docs"):
"""Load all markdown files from the docs directory."""
documents = []
for filename in os.listdir(docs_dir):
if filename.endswith(".md"):
filepath = os.path.join(docs_dir, 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, chunk_size=500, overlap=50):
"""Split text into overlapping chunks."""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start += chunk_size - overlap
return chunks
def ingest_documents(docs_dir="docs"):
"""Load, chunk, and store documents in ChromaDB."""
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(name="research_docs")
documents = load_documents(docs_dir)
all_chunks = []
all_ids = []
all_metadatas = []
for doc in documents:
chunks = chunk_text(doc["content"])
for i, chunk in enumerate(chunks):
chunk_id = f"{doc['source']}_chunk_{i}"
all_chunks.append(chunk)
all_ids.append(chunk_id)
all_metadatas.append({"source": doc["source"], "chunk_index": i})
collection.upsert(documents=all_chunks, ids=all_ids, metadatas=all_metadatas)
print(f"Ingested {len(all_chunks)} chunks from {len(documents)} documents.")
return collection
def search_documents(query, n_results=3):
"""Search the document collection for relevant chunks."""
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(name="research_docs")
results = collection.query(query_texts=[query], n_results=n_results)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append(
{
"content": results["documents"][0][i],
"source": results["metadatas"][0][i]["source"],
"distance": results["distances"][0][i],
}
)
return formatted_results
if __name__ == "__main__":
ingest_documents()
print("\nTesting search...")
results = search_documents("What is medallion architecture?")
for r in results:
print(f"\n[{r['source']}] (distance: {r['distance']:.4f})")
print(r["content"][:200])Your RAG pipeline is working. Documents go in, relevant chunks come back out. Next up, you'll build an AI agent that decides when to use this search tool versus answering from its own knowledge.
Build the Agent with Tool Definitions
Your RAG pipeline can retrieve relevant document chunks. Now you need something that decides when to search those documents versus doing something else entirely.
An AI agent is a model that can choose which tools to call based on the user's question. In this step, you will define tool declarations that tell Gemini what capabilities are available, write the Python functions that execute when those tools are called, and wire everything into a loop that handles the back-and-forth between the model and your code.
In this step, get ready to:
- Define tool declarations that describe your document search and calculator to Gemini.
- Write executor functions that run when the model calls each tool.
- Build the agent loop that handles function calling and returns the final response.
Define the tool declarations
Gemini's function calling feature lets you describe tools as JSON schemas. The model reads these descriptions and decides which tool to call based on the user's message. It never executes code directly. It returns a structured request that your code handles.
- Create a new file called agent.py in your project folder.
- Add the imports and the first tool declaration by pasting this code:
import os
from google import genai
from google.genai import types
from rag import search_documents, ingest_documents
# Tool declarations for Gemini
search_tool = {
"name": "search_knowledge_base",
"description": "Search the loaded documents for information about a topic. Use this when the user asks about Python, cloud computing, or data engineering concepts that might be in the knowledge base.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query to find relevant documents.",
},
},
"required": ["query"],
},
}What does this code do?
The imports bring in the Gemini SDK's types module (for constructing tool configs) and your RAG functions from the previous step.
The search_tool dictionary is a schema that tells Gemini: "there is a tool called search_knowledge_base that accepts a query string." Gemini reads the description field to decide when this tool is relevant to the user's question.
- Add the second tool declaration directly below search_tool:
calculator_tool = {
"name": "calculator",
"description": "Evaluate a mathematical expression. Use this for any calculations the user requests.",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to evaluate, e.g. '2 + 2' or '1024 * 768'.",
},
},
"required": ["expression"],
},
}What does this code do?
The calculator_tool dictionary follows the same schema pattern. It tells Gemini there is a calculator that accepts a math expression string. The description and example values help the model understand what format to pass.
- Save agent.py.
Why dictionaries instead of actual functions?
These tool declarations are metadata only. They describe what each tool does and what parameters it expects. Gemini uses this information to decide which tool to call and what arguments to pass. Your actual Python functions (coming next) are what execute when the model makes a choice.
Write the executor functions
When Gemini decides to call a tool, it sends back a structured response containing the tool name and arguments. Your code needs real Python functions that actually perform the work and return a result.
- Add the execute_search function below your tool declarations in agent.py:
def execute_search(query):
"""Execute document search and format results."""
results = search_documents(query)
if not results:
return {"answer": "No relevant documents found."}
context = ""
sources = []
for r in results:
context += f"\n\n[Source: {r['source']}]\n{r['content']}"
sources.append(r["source"])
return {"context": context, "sources": list(set(sources))}What does this code do?
This function calls your search_documents() function from rag.py and formats the results into a dictionary. It concatenates the matching chunks with their source filenames, and returns a deduplicated list of sources so the model can cite them in its answer.
- Add the execute_calculator function directly below execute_search:
def execute_calculator(expression):
"""Safely evaluate a math expression."""
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return {"error": "Invalid expression. Only numbers and basic operators allowed."}
try:
result = eval(expression)
return {"result": str(result)}
except Exception as e:
return {"error": f"Calculation failed: {str(e)}"}What does this code do?
This function validates that the expression only contains safe characters (digits and basic math operators) before passing it to eval(). This prevents code injection. If anything goes wrong, it returns an error message that Gemini can relay to the user.
Both executor functions return dictionaries. Gemini receives these dictionaries as the function result and uses them to formulate its final response to the user.
- Save agent.py.
Seeing a NameError for search_documents?
Make sure your import at the top of agent.py says from rag import search_documents, ingest_documents. Both function names must match exactly.
Help me debug my imports
Build the agent loop
The agent loop is where everything comes together. It sends the user's message to Gemini along with the tool declarations. If Gemini decides to call a tool, your code executes the matching function and sends the result back. This cycle repeats until Gemini has enough information to respond directly.
- Add the run_agent function and the entry point below your executor functions in agent.py:
def run_agent():
"""Run the agentic RAG assistant."""
# Initialize
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
print("Error: GEMINI_API_KEY not found. Add it to your .env file.")
return
client = genai.Client(api_key=api_key)
# Ingest documents on startup
print("Ingesting documents...")
ingest_documents()
# Configure tools
tools = types.Tool(function_declarations=[search_tool, calculator_tool])
config = types.GenerateContentConfig(
tools=[tools],
system_instruction="You are a research assistant with access to a knowledge base of documents about Python, cloud computing, and data engineering. Use the search_knowledge_base tool when users ask about topics that might be in your documents. Use the calculator tool for math. Answer directly for general knowledge questions. When you use the search tool, cite the source files in your response.",
)
contents = []
print("\nAgentic RAG Assistant ready! Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not user_input:
continue
contents.append(
types.Content(role="user", parts=[types.Part(text=user_input)])
)
response = client.models.generate_content(
model="gemini-2.5-flash", contents=contents, config=config
)
# Handle function calls
while response.candidates[0].content.parts[0].function_call:
function_call = response.candidates[0].content.parts[0].function_call
print(f" [Using tool: {function_call.name}]")
# Execute the function
if function_call.name == "search_knowledge_base":
result = execute_search(function_call.args["query"])
if "sources" in result:
print(f" [Sources: {', '.join(result['sources'])}]")
elif function_call.name == "calculator":
result = execute_calculator(function_call.args["expression"])
else:
result = {"error": f"Unknown function: {function_call.name}"}
# Send function result back to the model
contents.append(response.candidates[0].content)
fn_response_part = types.Part.from_function_response(
name=function_call.name, response=result, id=function_call.id
)
contents.append(
types.Content(role="user", parts=[fn_response_part])
)
response = client.models.generate_content(
model="gemini-2.5-flash", contents=contents, config=config
)
# Print final response
assistant_message = response.candidates[0].content
contents.append(assistant_message)
print(f"\nAssistant: {response.text}\n")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"\nError: {e}\n")
if __name__ == "__main__":
run_agent()What does this code do?
The function starts by loading your API key and ingesting documents into ChromaDB. It then configures a types.Tool object with both function declarations and a GenerateContentConfig with a system instruction that tells Gemini how to behave.
The while True loop reads user input, appends it to a contents list (which tracks the full conversation), and sends it to Gemini. If the response contains a function_call, the code executes the matching function and sends the result back using types.Part.from_function_response() with the function's id.
This inner while loop handles cases where Gemini calls tools multiple times before giving a final text answer. Once no more function calls remain, the assistant's text response is printed.
- Save agent.py.
- Run the agent by executing this command:
python agent.py- When you see Agentic RAG Assistant ready! Type 'quit' to exit., type a document question like:
What does the documentation say about serverless computing?- You should see [Using tool: search_knowledge_base] followed by source file names and then Gemini's answer citing those sources.
- Now test the calculator by typing:
What is 1024 * 768?- You should see [Using tool: calculator] followed by the computed result in the answer.
- Type quit to exit the agent.
Agent not calling the tools?
If Gemini answers directly without calling search_knowledge_base, try a more specific question like "What do my documents say about virtual environments?" The model sometimes answers general knowledge questions from memory. Mentioning "documents" or "knowledge base" nudges it toward the tool.
If you see a KeyError or attribute error, make sure your .env file contains a valid GEMINI_API_KEY and that rag.py is in the same folder as agent.py.
Help me debug my agent
✔️ Awesome, I've got everything!
Great. Double check you have saved agent.py and confirmed both tools work.
ⓧ I'd like to double check the full code
import os
from google import genai
from google.genai import types
from rag import search_documents, ingest_documents
# Tool declarations for Gemini
search_tool = {
"name": "search_knowledge_base",
"description": "Search the loaded documents for information about a topic. Use this when the user asks about Python, cloud computing, or data engineering concepts that might be in the knowledge base.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query to find relevant documents.",
},
},
"required": ["query"],
},
}
calculator_tool = {
"name": "calculator",
"description": "Evaluate a mathematical expression. Use this for any calculations the user requests.",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to evaluate, e.g. '2 + 2' or '1024 * 768'.",
},
},
"required": ["expression"],
},
}
def execute_search(query):
"""Execute document search and format results."""
results = search_documents(query)
if not results:
return {"answer": "No relevant documents found."}
context = ""
sources = []
for r in results:
context += f"\n\n[Source: {r['source']}]\n{r['content']}"
sources.append(r["source"])
return {"context": context, "sources": list(set(sources))}
def execute_calculator(expression):
"""Safely evaluate a math expression."""
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return {"error": "Invalid expression. Only numbers and basic operators allowed."}
try:
result = eval(expression)
return {"result": str(result)}
except Exception as e:
return {"error": f"Calculation failed: {str(e)}"}
def run_agent():
"""Run the agentic RAG assistant."""
# Initialize
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
print("Error: GEMINI_API_KEY not found. Add it to your .env file.")
return
client = genai.Client(api_key=api_key)
# Ingest documents on startup
print("Ingesting documents...")
ingest_documents()
# Configure tools
tools = types.Tool(function_declarations=[search_tool, calculator_tool])
config = types.GenerateContentConfig(
tools=[tools],
system_instruction="You are a research assistant with access to a knowledge base of documents about Python, cloud computing, and data engineering. Use the search_knowledge_base tool when users ask about topics that might be in your documents. Use the calculator tool for math. Answer directly for general knowledge questions. When you use the search tool, cite the source files in your response.",
)
contents = []
print("\nAgentic RAG Assistant ready! Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not user_input:
continue
contents.append(
types.Content(role="user", parts=[types.Part(text=user_input)])
)
response = client.models.generate_content(
model="gemini-2.5-flash", contents=contents, config=config
)
# Handle function calls
while response.candidates[0].content.parts[0].function_call:
function_call = response.candidates[0].content.parts[0].function_call
print(f" [Using tool: {function_call.name}]")
# Execute the function
if function_call.name == "search_knowledge_base":
result = execute_search(function_call.args["query"])
if "sources" in result:
print(f" [Sources: {', '.join(result['sources'])}]")
elif function_call.name == "calculator":
result = execute_calculator(function_call.args["expression"])
else:
result = {"error": f"Unknown function: {function_call.name}"}
# Send function result back to the model
contents.append(response.candidates[0].content)
fn_response_part = types.Part.from_function_response(
name=function_call.name, response=result, id=function_call.id
)
contents.append(
types.Content(role="user", parts=[fn_response_part])
)
response = client.models.generate_content(
model="gemini-2.5-flash", contents=contents, config=config
)
# Print final response
assistant_message = response.candidates[0].content
contents.append(assistant_message)
print(f"\nAssistant: {response.text}\n")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"\nError: {e}\n")
if __name__ == "__main__":
run_agent()Your agent can now autonomously decide whether to search your documents or fire up the calculator. Next, you will turn this into a proper multi-turn chat interface that remembers the full conversation across questions.
Build the Multi-Turn Chat Interface
Your agent can search documents and run calculations. But right now, each interaction is a one-shot exchange.
A real assistant remembers what you talked about. In this step, you'll understand how the contents list gives your agent conversational memory, explore the system instruction that guides tool selection, and test it all by having a multi-turn conversation.
In this step, get ready to:
- Understand how the conversation history list preserves context across turns.
- Run the agent and have a multi-turn conversation with tool use.
- Verify the agent routes to the right tool and remembers previous answers.
Understand how conversation history works
For an AI assistant to handle follow-up questions, it needs to "remember" the entire conversation. Your run_agent() function already does this through the contents list.
- In agent.py, find the contents = [] line inside run_agent().
contents = []
print("\nAgentic RAG Assistant ready! Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not user_input:
continue
contents.append(
types.Content(role="user", parts=[types.Part(text=user_input)])
)What does this code do?
The contents list is your conversation memory. Every message gets appended to it and never removed.
- Each user message is appended as a types.Content(role="user", ...) object.
- Each model response (including function calls) is appended as the model's content.
- Function results are also appended. When generate_content is called, the entire list is sent, giving Gemini full context of the conversation.
Why does conversation memory matter for agents?
Without the full history, asking "tell me more about that" after a document search would fail. The model would not know what "that" refers to.
By accumulating every turn (user messages, tool calls, tool results, and model replies), Gemini can resolve references like "summarize that in a table" or "what about the second point?" across multiple exchanges.
Now look at the system instruction that guides the model's tool selection.
- Scroll up to the config variable inside run_agent().
config = types.GenerateContentConfig(
tools=[tools],
system_instruction="You are a research assistant with access to a knowledge base of documents about Python, cloud computing, and data engineering. Use the search_knowledge_base tool when users ask about topics that might be in your documents. Use the calculator tool for math. Answer directly for general knowledge questions. When you use the search tool, cite the source files in your response.",
)What does the system instruction do?
The system_instruction tells Gemini exactly when to use each tool. Without it, the model might use document search for general questions or skip it entirely for document-specific ones.
The instruction also asks the model to cite source files. This means when the agent retrieves from your documents, it will tell you which file the answer came from.
Run the assistant and test multi-turn conversations
Everything is wired up. Time to run your agentic assistant and have a real conversation with it.
- Start the agent by running this command in your terminal:
python agent.pyYou should see the documents being ingested, followed by the prompt Agentic RAG Assistant ready! Type 'quit' to exit.
Seeing an import error or GEMINI_API_KEY not found?
- Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
- Confirm your .env file is in the same directory as agent.py and contains your GEMINI_API_KEY.
- Check that python-dotenv is installed. If not, run pip install python-dotenv.
Help me debug why my agent.py is failing to start
- Once the assistant is ready, type a question about your documents. Try this:
What is the medallion architecture?You should see [Using tool: search_knowledge_base] printed, followed by the source files used, and then the model's answer citing those sources.
- Now ask a follow-up question that references the previous answer:
Can you explain the Bronze layer in more detail?What should I see?
The agent should understand that "Bronze layer" refers to the medallion architecture you just asked about. It will likely call search_knowledge_base again with a more specific query, then provide a detailed answer drawing from your data_engineering.md file.
This proves the conversation history is working. The model resolves "the Bronze layer" because it sees your earlier exchange in the contents list.
Test tool routing and error handling
Your agent decides which tool to use based on the system instruction. Let's verify it routes correctly across different question types.
- Test the calculator tool by typing:
What is 1024 * 768?You should see [Using tool: calculator] followed by the computed result.
- Now test direct answering with a general knowledge question:
What year was Python created?The model should answer directly without calling any tool. No [Using tool: ...] message should appear.
How does the agent decide which tool to use?
Gemini reads the system_instruction and the description field on each tool declaration. It matches the user's intent to the best tool. If no tool fits, it answers from its own knowledge.
This is what makes your code an "agent" rather than just a chatbot. The model autonomously decides its actions at each turn instead of following a fixed script.
- Finally, test graceful exit by pressing Ctrl+C or typing quit.
Both should print Goodbye! and exit cleanly without a traceback.
Why catch KeyboardInterrupt separately?
When a user presses Ctrl+C, Python raises a KeyboardInterrupt exception. Without an explicit handler, the terminal would show a noisy traceback.
The general except Exception block catches everything else (network errors, API rate limits, unexpected responses) so the loop continues instead of crashing.
Agent not calling the right tool?
- Make sure your question clearly relates to a topic in your documents. Asking "what's in my docs about Python?" is clearer than just "tell me about Python."
- Check that your docs/ folder has the sample markdown files from Step 2. If the collection is empty, the search tool returns nothing useful.
- The model's tool choice is non-deterministic. Occasionally it may answer directly instead of searching. Try rephrasing your question to be more specific.
Help me debug why my agent picks the wrong tool
✔️ Awesome, I've got everything!
Great. Double check you've saved agent.py and that your multi-turn conversation worked correctly.
ⓧ I'd like to double check the full code
Here is the complete agent.py for comparison:
import os
from google import genai
from google.genai import types
from rag import search_documents, ingest_documents
# Tool declarations for Gemini
search_tool = {
"name": "search_knowledge_base",
"description": "Search the loaded documents for information about a topic. Use this when the user asks about Python, cloud computing, or data engineering concepts that might be in the knowledge base.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query to find relevant documents.",
},
},
"required": ["query"],
},
}
calculator_tool = {
"name": "calculator",
"description": "Evaluate a mathematical expression. Use this for any calculations the user requests.",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to evaluate, e.g. '2 + 2' or '1024 * 768'.",
},
},
"required": ["expression"],
},
}
def execute_search(query):
"""Execute document search and format results."""
results = search_documents(query)
if not results:
return {"answer": "No relevant documents found."}
context = ""
sources = []
for r in results:
context += f"\n\n[Source: {r['source']}]\n{r['content']}"
sources.append(r["source"])
return {"context": context, "sources": list(set(sources))}
def execute_calculator(expression):
"""Safely evaluate a math expression."""
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return {"error": "Invalid expression. Only numbers and basic operators allowed."}
try:
result = eval(expression)
return {"result": str(result)}
except Exception as e:
return {"error": f"Calculation failed: {str(e)}"}
def run_agent():
"""Run the agentic RAG assistant."""
# Initialize
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
print("Error: GEMINI_API_KEY not found. Add it to your .env file.")
return
client = genai.Client(api_key=api_key)
# Ingest documents on startup
print("Ingesting documents...")
ingest_documents()
# Configure tools
tools = types.Tool(function_declarations=[search_tool, calculator_tool])
config = types.GenerateContentConfig(
tools=[tools],
system_instruction="You are a research assistant with access to a knowledge base of documents about Python, cloud computing, and data engineering. Use the search_knowledge_base tool when users ask about topics that might be in your documents. Use the calculator tool for math. Answer directly for general knowledge questions. When you use the search tool, cite the source files in your response.",
)
contents = []
print("\nAgentic RAG Assistant ready! Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if not user_input:
continue
contents.append(
types.Content(role="user", parts=[types.Part(text=user_input)])
)
response = client.models.generate_content(
model="gemini-2.5-flash", contents=contents, config=config
)
# Handle function calls
while response.candidates[0].content.parts[0].function_call:
function_call = response.candidates[0].content.parts[0].function_call
print(f" [Using tool: {function_call.name}]")
# Execute the function
if function_call.name == "search_knowledge_base":
result = execute_search(function_call.args["query"])
if "sources" in result:
print(f" [Sources: {', '.join(result['sources'])}]")
elif function_call.name == "calculator":
result = execute_calculator(function_call.args["expression"])
else:
result = {"error": f"Unknown function: {function_call.name}"}
# Send function result back to the model
contents.append(response.candidates[0].content)
fn_response_part = types.Part.from_function_response(
name=function_call.name, response=result, id=function_call.id
)
contents.append(
types.Content(role="user", parts=[fn_response_part])
)
response = client.models.generate_content(
model="gemini-2.5-flash", contents=contents, config=config
)
# Print final response
assistant_message = response.candidates[0].content
contents.append(assistant_message)
print(f"\nAssistant: {response.text}\n")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"\nError: {e}\n")
if __name__ == "__main__":
run_agent()Your agentic RAG assistant is fully operational. It searches documents, computes math, remembers conversations, and decides autonomously which tool to use. Next up, you'll extend it with web search capabilities in the Secret Mission.
Secret mission
Add Web Search with Google Search Grounding
Your agent can search local documents and do math. But what about questions on current events or topics not in your knowledge base? In this mission, you'll combine Gemini's built-in Google Search grounding with your custom tools to create a multi-source research agent.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally with a free-tier API key, so there are no ongoing costs. Decide whether to keep your resources for future experimentation, pause them to come back later, or delete them entirely.
Resources you used:
- Project folder on your Desktop (contains main.py, rag.py, agent.py, .env, docs/, and chroma_db/).
- Python virtual environment (inside your project folder).
- Gemini API key (in Google AI Studio / Google Cloud Console).
Keep everything running
No action needed. Choose this if you plan to keep experimenting with your RAG assistant, adding more documents, or extending the agent with new tools.
- Your project folder stays on your Desktop.
- Your Gemini API key remains active on the free tier at no cost.
- The chroma_db/ folder persists your vector embeddings so you do not need to re-ingest documents on the next run.
Pause - I'll come back to this later
Shut down any running Python processes to free up memory. Your files and API key stay in place for when you return.
- If the agent is still running in your terminal, press Ctrl+C to stop it.
- Deactivate your virtual environment by running:
deactivate- Your project folder, chroma_db/ data, and API key all remain intact for next time.
Delete - I don't want to use this again
Remove all project resources and revoke your API key for a clean slate.
- If the agent is still running, press Ctrl+C to stop it.
- Deactivate your virtual environment by running:
deactivate- Delete the entire project folder by running:
macOS
cd ~/Desktop
rm -rf agentic-rag-assistantWindows
cd ~\Desktop
rmdir /s /q agentic-rag-assistant- Delete your Gemini API key from the Google Cloud Console:
- Open the Google Cloud API Credentials page.
- Find the API key you created for this project and click the Actions icon (three dots).
- Select Delete API key.
- In the Delete credential modal, select Delete to confirm.
That's a Wrap!
That's a Wrap!
Nice work! You just built an agentic RAG assistant from scratch using two of the most in-demand AI engineering patterns in 2026.
You've learned how to:
- Build a complete RAG pipeline that ingests documents, chunks text, creates vector embeddings with ChromaDB, and retrieves semantically relevant context to answer questions.
- Implement an AI agent with custom function calling that autonomously decides which tools to use based on the user's question.
- Create a multi-turn chat interface with conversation history that preserves context across an entire session.
- Secret Mission: Combined built-in Google Search grounding with custom tool declarations to create a multi-tool agent that routes between local documents and live web results.
Ready to quiz yourself?