Build a Production-Grade RAG Agent
Build a RAG agent with tool calling, evaluation, and CloudWatch observability.
Introduction
30 Second Summary
Every company has a pile of internal documents that employees need answers from. Finding the right policy or procedure usually means digging through dozens of files and hoping you land on the right one.
In this project, you will build a production-grade RAG agent that retrieves company documents and answers questions using Claude's tool-calling API. You will also wire in an evaluation pipeline, observability with AWS CloudWatch, and PII detection guardrails.
What You'll Build
A company knowledge base assistant that uses Claude's tool-calling API to search a ChromaDB vector store, generate grounded answers with citations, and pass through evaluation and observability layers before reaching the user.
By the end of this project, you'll have:
- A tool-calling RAG agent that searches a ChromaDB knowledge base and generates answers using Claude's agentic loop.
- An evaluation pipeline that scores every response for faithfulness and relevance using Claude-as-judge.
- Production observability with structured metrics pushed to AWS CloudWatch, including latency, token usage, and error rates.
- A PII detection guardrail that scans inputs and outputs before they reach the model or the user.
- Secret Mission: Add confidence-based routing that detects low retrieval quality and logs those queries separately for review.
Are there any prerequisites?
You should be comfortable with Python and familiar with REST APIs. You will also need an Anthropic API key and an AWS account with CLI access configured. Step 1 will walk you through verifying both.
Before We Start
Before We Start...
Set Up Your Environment and Install Dependencies
Your RAG agent uses three Python packages to work. One talks to Claude's API, another stores and searches documents in ChromaDB, and the third pushes production metrics to AWS CloudWatch.
In this step, you'll create a project directory with a Python virtual environment, install these packages, and verify your API credentials are ready.
In this step, get ready to:
- Create a project directory with a Python virtual environment.
- Install anthropic, chromadb, and boto3.
- Verify your Anthropic API key and AWS CLI credentials.
Create your project directory and virtual environment
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
- Confirm Python 3.9 or higher is installed by running:
python3 --versionI see a version number
You're all set. Any version of 3.9 or above works for this project.
Command not found
You need Python installed before continuing.
- Download the latest Python installer from python.org/downloads.
- Run the installer. On Windows, check the Add python.exe to PATH box before clicking Install Now.
- Close and reopen your terminal, then run python3 --version again to confirm.
Pro tip: Windows Python command
On Windows, if python3 is not recognized, use python instead throughout this project.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project directory and move into it by running these commands:
mkdir rag-agent
cd rag-agent- Create a Python virtual environment by running:
python3 -m venv venv- Confirm the virtual environment was created by running:
lsYou should see a venv folder in the output.
- Activate the virtual environment by running:
macOS
source venv/bin/activateWindows
.\venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
Don't see (venv) in your prompt?
- Make sure you ran the activation command for your operating system. macOS uses source venv/bin/activate and Windows uses .\venv\Scripts\activate.
- On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell, then try activating again.
- Check that you're inside the rag-agent directory where you created the venv.
Help me activate my Python virtual environment.
Install the required packages
With your virtual environment active, any packages you install stay isolated to this project. They won't interfere with other Python projects on your machine.
- Install all three packages by running:
pip install anthropic chromadb boto3What are these packages?
- anthropic is the official Python SDK for Claude's API. It handles authentication, message creation, and tool calling.
- chromadb is a vector database that stores and searches documents using embeddings. Your agent's knowledge base lives here.
- boto3 is AWS's Python SDK. You'll use it to push observability metrics to CloudWatch.
The installation may take a minute. chromadb pulls in several dependencies including its default embedding model.
- Verify all packages installed correctly by running:
python3 -c "import anthropic; import chromadb; import boto3; print('All packages imported successfully')"You should see All packages imported successfully printed to your terminal.
Seeing a ModuleNotFoundError?
- Check that (venv) still appears in your terminal prompt. If it disappeared, reactivate the virtual environment before running pip install again.
- If one specific package failed, try installing it individually (e.g. pip install chromadb) to see the full error message.
Help me fix my package installation errors.
Verify your API credentials
The Anthropic SDK reads your API key from an environment variable called ANTHROPIC_API_KEY. Your AWS CLI also needs valid credentials to push metrics to CloudWatch later.
- Set your Anthropic API key as an environment variable by running the command below.
- Replace your-api-key-here with your actual Anthropic API key before running it.
macOS
export ANTHROPIC_API_KEY=your-api-key-hereWindows
$env:ANTHROPIC_API_KEY="your-api-key-here"Where do I find my Anthropic API key?
Go to console.anthropic.com/settings/keys and create a new key if you don't have one yet. Copy the key value immediately. Anthropic only shows it once.
- Verify the API key is set correctly by running:
python3 -c "import os; key = os.environ.get('ANTHROPIC_API_KEY', ''); print(f'Key is set: {key[:8]}...') if key else print('ERROR: Key not set')"You should see the first 8 characters of your key printed. If you see ERROR: Key not set, run the export command again and make sure you pasted your key correctly.
- Verify your AWS CLI is configured by running:
aws sts get-caller-identityYou should see a JSON response containing your Account, UserId, and Arn fields.
Seeing "Unable to locate credentials"?
- Your AWS CLI is not configured yet. Run aws configure and enter your AWS Access Key ID, Secret Access Key, default region (e.g. us-east-1), and output format (json).
- If you don't have AWS access keys, create them in the AWS Console under IAM > Users > your user > Security credentials > Create access key.
Help me configure my AWS CLI credentials.
💡 Why does this project need AWS?
You won't deploy anything to AWS. The agent runs entirely on your machine. You use AWS CloudWatch only to push observability metrics (latency, token usage, error rates) so you can monitor the agent like a production system.
Your environment is set up with all dependencies installed and both sets of credentials verified. Next up, you'll build the knowledge base that your RAG agent will search.
Build the Knowledge Base with ChromaDB
Your environment is locked and loaded. Now it's time to give your agent something to actually search through.
Every RAG system needs a knowledge base. In this step, you'll create a set of simulated company documents, load them into a ChromaDB vector store, and build a search function that retrieves the most relevant documents for any query.
In this step, get ready to:
- Define a collection of enterprise documents covering HR, finance, and IT departments.
- Build ingestion and search functions powered by ChromaDB.
- Test the knowledge base with sample queries to confirm it returns relevant results.
Define the enterprise documents
Your RAG agent needs a realistic set of documents to search through. You'll define eight simulated enterprise documents spanning three departments: HR policies like PTO and remote work, finance procedures like budget approvals, and IT guides like VPN setup.
Each document carries metadata including its department, doc_type, and last_updated date. This metadata lets the search function filter results by department later.
- Create a new file called knowledge_base.py inside your rag-agent folder.
- Paste the following code into knowledge_base.py to define the import and document collection:
import chromadb
# Simulated enterprise documents
COMPANY_DOCUMENTS = [
{
"id": "hr-001",
"document": "PTO Policy: Full-time employees accrue 15 days of paid time off per year during their first 3 years. After 3 years, accrual increases to 20 days per year. PTO requests must be submitted at least 2 weeks in advance for absences of 3 or more days. Unused PTO can be carried over up to a maximum of 5 days into the next calendar year.",
"metadata": {"department": "hr", "doc_type": "policy", "last_updated": "2025-01-15"},
},
{
"id": "hr-002",
"document": "Remote Work Policy: Employees may work remotely up to 3 days per week with manager approval. Fully remote arrangements require VP-level approval and a signed remote work agreement. Remote employees must be available during core hours (10am-3pm local time) and maintain a reliable internet connection. Equipment stipends of $500 are provided annually for home office setup.",
"metadata": {"department": "hr", "doc_type": "policy", "last_updated": "2025-03-01"},
},
{
"id": "hr-003",
"document": "Expense Reimbursement Policy: Business expenses must be submitted within 30 days of the transaction. Receipts are required for all expenses over $25. Meals during business travel are reimbursed up to $75 per day. Hotel accommodations are reimbursed at the GSA per diem rate for the travel location. All expenses over $500 require pre-approval from a department director.",
"metadata": {"department": "hr", "doc_type": "policy", "last_updated": "2025-02-10"},
},
{
"id": "finance-001",
"document": "Budget Approval Process: Purchase requests under $5,000 require manager approval only. Requests between $5,000 and $25,000 require director approval and a brief justification memo. Requests over $25,000 require VP approval, a full business case document, and CFO sign-off. Emergency purchases up to $10,000 can be made with retroactive approval within 48 hours.",
"metadata": {"department": "finance", "doc_type": "procedure", "last_updated": "2025-01-20"},
},
{
"id": "finance-002",
"document": "Vendor Payment Terms: Standard payment terms are Net 30 from invoice date. Vendors may request Net 15 terms with a 2% early payment discount. All new vendors must complete a W-9 form and pass a compliance screening before first payment. Vendor invoices over $50,000 require two authorized signatures for payment approval.",
"metadata": {"department": "finance", "doc_type": "procedure", "last_updated": "2025-04-01"},
},
{
"id": "it-001",
"document": "VPN Setup Instructions: Download the FortiClient VPN application from the company software portal. Use your corporate email as the username and your Active Directory password. The VPN gateway address is vpn.company.internal. Two-factor authentication is required using the company-issued authenticator app. Contact the IT help desk at help@company.internal if you encounter connection issues.",
"metadata": {"department": "it", "doc_type": "procedure", "last_updated": "2025-03-15"},
},
{
"id": "it-002",
"document": "Password Reset Procedure: Passwords must be changed every 90 days. Use the self-service portal at passwords.company.internal to reset your password. New passwords must be at least 12 characters with a mix of uppercase, lowercase, numbers, and special characters. If locked out, contact the IT help desk. Account lockout occurs after 5 failed login attempts.",
"metadata": {"department": "it", "doc_type": "procedure", "last_updated": "2025-02-28"},
},
{
"id": "it-003",
"document": "Software Request Process: Submit software requests through the IT service portal. Standard software (listed in the approved catalog) is provisioned within 2 business days. Non-standard software requires security review, which takes 5-10 business days. All software must be licensed through the company. Personal software installations on company devices are prohibited.",
"metadata": {"department": "it", "doc_type": "procedure", "last_updated": "2025-01-30"},
},
]What does this code do?
- import chromadb brings in the ChromaDB library, which handles vector embeddings and similarity search.
- COMPANY_DOCUMENTS is a list of eight dictionaries. Each one has a unique id, a document string with the full policy text, and a metadata dict for filtering.
- The documents cover three departments: HR (PTO, remote work, expenses), finance (budget approvals, vendor payments), and IT (VPN, passwords, software requests).
- Save knowledge_base.py.
- You won't see any output yet. The next chunk adds the functions that ingest these documents into ChromaDB.
Build the ingestion and search functions
ChromaDB is an in-memory vector store. When you add documents to it, ChromaDB automatically converts each one into a numerical embedding using the Sentence Transformers all-MiniLM-L6-v2 model. These embeddings let you search by meaning instead of exact keyword match.
- In knowledge_base.py, add the following two functions below the COMPANY_DOCUMENTS list:
def create_knowledge_base():
"""Create and populate the ChromaDB knowledge base."""
client = chromadb.Client()
collection = client.get_or_create_collection(name="company_docs")
collection.add(
ids=[doc["id"] for doc in COMPANY_DOCUMENTS],
documents=[doc["document"] for doc in COMPANY_DOCUMENTS],
metadatas=[doc["metadata"] for doc in COMPANY_DOCUMENTS],
)
return collection
def search_knowledge_base(collection, query, n_results=3, department=None):
"""Search the knowledge base and return results with metadata."""
where_filter = {"department": department} if department else None
results = collection.query(
query_texts=[query],
n_results=n_results,
where=where_filter,
include=["documents", "metadatas", "distances"],
)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append(
{
"id": results["ids"][0][i],
"document": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i],
}
)
return formatted_resultsWhat do these functions do?
- create_knowledge_base() spins up an in-memory ChromaDB client and creates a collection called company_docs. It then adds all eight documents with their IDs, text, and metadata. ChromaDB auto-embeds each document behind the scenes.
- search_knowledge_base() takes a natural-language query, embeds it, and returns the closest matching documents. The optional department filter narrows results to a single department using ChromaDB's where clause.
- Each result includes a distance score. ChromaDB uses L2 (Euclidean) distance by default, so lower values mean higher similarity.
- Save knowledge_base.py.
- You won't see a change yet. The next chunk adds a test block so you can run the file and see real search results.
Test the knowledge base
Time to verify that documents go in and relevant results come out. You'll add a test block that creates the knowledge base, runs a sample query, and prints results with their distance scores.
- Add the following test block at the bottom of knowledge_base.py:
if __name__ == "__main__":
collection = create_knowledge_base()
results = search_knowledge_base(collection, "What is the PTO policy?")
for r in results:
print(f"[{r['distance']:.4f}] {r['id']}: {r['document'][:100]}...")What does this test block do?
- if __name__ == "__main__": means this code only runs when you execute the file directly. It won't run when other files import from this module.
- It creates the knowledge base, searches for "What is the PTO policy?", and prints the top 3 results with their L2 distance scores.
- Save knowledge_base.py.
- Run the file by executing this command in your terminal (make sure your virtual environment is active):
python knowledge_base.pyFirst run takes a moment
The first time ChromaDB runs, it downloads the Sentence Transformers all-MiniLM-L6-v2 embedding model (~80MB). This is a one-time download that happens automatically. On a slow connection, it can take a minute or two.
You should see three results printed to your terminal, each with a distance score in square brackets, a document ID, and a truncated preview of the document text. The hr-001 PTO policy document should appear first with the lowest distance score.
Not seeing any output?
- Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. If not, activate it again with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).
- If you see ModuleNotFoundError: No module named 'chromadb', the package is not installed in your active environment. Run pip install chromadb again.
- If the script hangs, it is likely downloading the embedding model. Give it a minute or two to finish.
Why isn't my knowledge_base.py producing any output?
Awesome, I've got everything!
Great. Double check you've saved knowledge_base.py before moving on.
I'd like to double check the full code
import chromadb
# Simulated enterprise documents
COMPANY_DOCUMENTS = [
{
"id": "hr-001",
"document": "PTO Policy: Full-time employees accrue 15 days of paid time off per year during their first 3 years. After 3 years, accrual increases to 20 days per year. PTO requests must be submitted at least 2 weeks in advance for absences of 3 or more days. Unused PTO can be carried over up to a maximum of 5 days into the next calendar year.",
"metadata": {"department": "hr", "doc_type": "policy", "last_updated": "2025-01-15"},
},
{
"id": "hr-002",
"document": "Remote Work Policy: Employees may work remotely up to 3 days per week with manager approval. Fully remote arrangements require VP-level approval and a signed remote work agreement. Remote employees must be available during core hours (10am-3pm local time) and maintain a reliable internet connection. Equipment stipends of $500 are provided annually for home office setup.",
"metadata": {"department": "hr", "doc_type": "policy", "last_updated": "2025-03-01"},
},
{
"id": "hr-003",
"document": "Expense Reimbursement Policy: Business expenses must be submitted within 30 days of the transaction. Receipts are required for all expenses over $25. Meals during business travel are reimbursed up to $75 per day. Hotel accommodations are reimbursed at the GSA per diem rate for the travel location. All expenses over $500 require pre-approval from a department director.",
"metadata": {"department": "hr", "doc_type": "policy", "last_updated": "2025-02-10"},
},
{
"id": "finance-001",
"document": "Budget Approval Process: Purchase requests under $5,000 require manager approval only. Requests between $5,000 and $25,000 require director approval and a brief justification memo. Requests over $25,000 require VP approval, a full business case document, and CFO sign-off. Emergency purchases up to $10,000 can be made with retroactive approval within 48 hours.",
"metadata": {"department": "finance", "doc_type": "procedure", "last_updated": "2025-01-20"},
},
{
"id": "finance-002",
"document": "Vendor Payment Terms: Standard payment terms are Net 30 from invoice date. Vendors may request Net 15 terms with a 2% early payment discount. All new vendors must complete a W-9 form and pass a compliance screening before first payment. Vendor invoices over $50,000 require two authorized signatures for payment approval.",
"metadata": {"department": "finance", "doc_type": "procedure", "last_updated": "2025-04-01"},
},
{
"id": "it-001",
"document": "VPN Setup Instructions: Download the FortiClient VPN application from the company software portal. Use your corporate email as the username and your Active Directory password. The VPN gateway address is vpn.company.internal. Two-factor authentication is required using the company-issued authenticator app. Contact the IT help desk at help@company.internal if you encounter connection issues.",
"metadata": {"department": "it", "doc_type": "procedure", "last_updated": "2025-03-15"},
},
{
"id": "it-002",
"document": "Password Reset Procedure: Passwords must be changed every 90 days. Use the self-service portal at passwords.company.internal to reset your password. New passwords must be at least 12 characters with a mix of uppercase, lowercase, numbers, and special characters. If locked out, contact the IT help desk. Account lockout occurs after 5 failed login attempts.",
"metadata": {"department": "it", "doc_type": "procedure", "last_updated": "2025-02-28"},
},
{
"id": "it-003",
"document": "Software Request Process: Submit software requests through the IT service portal. Standard software (listed in the approved catalog) is provisioned within 2 business days. Non-standard software requires security review, which takes 5-10 business days. All software must be licensed through the company. Personal software installations on company devices are prohibited.",
"metadata": {"department": "it", "doc_type": "procedure", "last_updated": "2025-01-30"},
},
]
def create_knowledge_base():
"""Create and populate the ChromaDB knowledge base."""
client = chromadb.Client()
collection = client.get_or_create_collection(name="company_docs")
collection.add(
ids=[doc["id"] for doc in COMPANY_DOCUMENTS],
documents=[doc["document"] for doc in COMPANY_DOCUMENTS],
metadatas=[doc["metadata"] for doc in COMPANY_DOCUMENTS],
)
return collection
def search_knowledge_base(collection, query, n_results=3, department=None):
"""Search the knowledge base and return results with metadata."""
where_filter = {"department": department} if department else None
results = collection.query(
query_texts=[query],
n_results=n_results,
where=where_filter,
include=["documents", "metadatas", "distances"],
)
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append(
{
"id": results["ids"][0][i],
"document": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i],
}
)
return formatted_results
if __name__ == "__main__":
collection = create_knowledge_base()
results = search_knowledge_base(collection, "What is the PTO policy?")
for r in results:
print(f"[{r['distance']:.4f}] {r['id']}: {r['document'][:100]}...")Your knowledge base is loaded and searchable. Next up, you'll build the RAG agent that uses Claude's tool-calling API to decide when to search this knowledge base and generate grounded answers from the results.
Build the Tool-Calling RAG Agent
Your knowledge base is loaded and searchable. Now you need something that can actually use it to answer questions.
In this step, you will build a RAG agent that connects Claude to your ChromaDB knowledge base using Anthropic's tool-calling API. Instead of just dumping documents into a prompt, the agent will decide when to search, execute the search itself, and then generate an answer grounded in the results.
In this step, get ready to:
- Define a tool schema and system prompt for Claude's tool-use API.
- Implement the agentic loop that handles tool calls and tool results.
- Test the agent with queries that exercise different response paths.
Define the tool schema and system prompt
Claude's tool-use API lets you describe tools as JSON schemas. When Claude receives a user question along with a tool definition, it can choose to call that tool instead of answering directly.
- Create a new file called agent.py inside your rag-agent folder.
- Add the imports, tool schema, and system prompt by copying this code into agent.py:
import json
import time
import anthropic
from knowledge_base import create_knowledge_base, search_knowledge_base
# Tool definition that tells Claude how to search our knowledge base
SEARCH_TOOL = {
"name": "search_knowledge_base",
"description": "Search the company knowledge base for information about HR policies, finance procedures, and IT guidelines. Always use this tool before answering a question.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query to find relevant documents.",
},
"department": {
"type": "string",
"enum": ["hr", "finance", "it"],
"description": "Optional department filter to narrow search results.",
},
},
"required": ["query"],
},
}
# Instructions that shape how the agent behaves
SYSTEM_PROMPT = """You are a company knowledge base assistant. Your role is to answer employee questions using information from the company knowledge base.
Rules:
1. ALWAYS search the knowledge base before answering. Never answer from general knowledge.
2. Base your answer ONLY on the retrieved documents. Do not make up information.
3. Cite the document ID (e.g., [hr-001]) when referencing specific policies or procedures.
4. If the retrieved documents do not contain the answer, say "I don't have information about that in the knowledge base."
5. Be concise and direct in your answers."""What does this code do?
- SEARCH_TOOL is a JSON schema that tells Claude what the tool does, what parameters it accepts, and which ones are required. Claude reads this schema to decide when and how to call the tool.
- The department parameter is optional, with an enum constraint that limits it to hr, finance, or it. Claude will use this filter when the question is clearly about one department.
- SYSTEM_PROMPT sets strict rules: always search before answering, only use retrieved documents, cite document IDs, and admit when information is missing.
- Save agent.py.
You won't see a change yet. The tool schema and system prompt are just configuration that the agent class will use next.
Implement the agentic loop
An agentic loop is the core pattern behind tool-calling agents. You send a message to Claude, and instead of always getting a text reply, Claude can respond with a tool_use request. Your code executes that tool, sends the result back, and Claude continues until it has enough information to give a final answer.
You will build this loop inside a RAGAgent class. Start with the class definition, the initializer, and a helper method that dispatches tool calls to your knowledge base.
- In agent.py, add the RAGAgent class below SYSTEM_PROMPT by copying this code:
class RAGAgent:
def __init__(self, collection):
# Initialize the Anthropic client (reads ANTHROPIC_API_KEY from env)
self.client = anthropic.Anthropic()
self.collection = collection
self.model = "claude-sonnet-4-5-20250929"
# Track retrieved contexts and metrics for observability later
self.last_retrieved_contexts = []
self.last_metrics = {}
def _execute_tool(self, tool_name, tool_input):
"""Execute a tool call and return the result."""
if tool_name == "search_knowledge_base":
results = search_knowledge_base(
self.collection,
query=tool_input["query"],
department=tool_input.get("department"),
)
# Store contexts so we can use them for evaluation later
self.last_retrieved_contexts = results
return {
"results": [
{
"id": r["id"],
"document": r["document"],
"department": r["metadata"]["department"],
"distance": r["distance"],
}
for r in results
]
}
raise ValueError(f"Unknown tool: {tool_name}")What does this code do?
- __init__ creates an Anthropic client that reads the ANTHROPIC_API_KEY environment variable automatically. It also stores the ChromaDB collection and sets the model to claude-sonnet-4-5-20250929.
- _execute_tool is the dispatcher. When Claude asks to call a tool, this method checks the tool name and runs the matching function. For search_knowledge_base, it calls the function you built in the previous step and formats the results for Claude.
- last_retrieved_contexts and last_metrics are stored on the class instance so the evaluation and observability layers can access them later.
- Save agent.py.
You won't see a change yet. The class needs its query method before it can do anything useful.
Now add the query method. This is the agentic loop itself. It sends the user's question to Claude, watches for tool-call responses, executes the tools, feeds results back, and repeats until Claude gives a final text answer.
- In agent.py, add the query method inside the RAGAgent class, between __init__ and _execute_tool:
def query(self, user_question):
"""Run a query through the RAG agent with tool calling."""
start_time = time.time()
total_input_tokens = 0
total_output_tokens = 0
self.last_retrieved_contexts = []
messages = [{"role": "user", "content": user_question}]
# First API call: send the question with the tool definition
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=[SEARCH_TOOL],
messages=messages,
)
total_input_tokens += response.usage.input_tokens
total_output_tokens += response.usage.output_tokens
# Agentic loop: keep going while Claude wants to use tools
while response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
try:
result = self._execute_tool(block.name, block.input)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result),
}
)
except Exception as exc:
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": str(exc),
"is_error": True,
}
)
# Append the assistant's response and tool results to the conversation
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
# Send the tool results back to Claude
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=[SEARCH_TOOL],
messages=messages,
)
total_input_tokens += response.usage.input_tokens
total_output_tokens += response.usage.output_tokens
# Extract the final text response
final_text = next(
(block.text for block in response.content if block.type == "text"),
"No response generated.",
)
latency_ms = (time.time() - start_time) * 1000
# Store metrics for observability
self.last_metrics = {
"latency_ms": latency_ms,
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
"retrieval_count": len(self.last_retrieved_contexts),
"avg_retrieval_distance": (
sum(c["distance"] for c in self.last_retrieved_contexts)
/ len(self.last_retrieved_contexts)
if self.last_retrieved_contexts
else 0
),
}
return final_textWhat's happening in the agentic loop?
The loop follows Claude's tool-calling protocol:
- You send the user's question to Claude with the SEARCH_TOOL definition. Claude decides whether to call the tool or answer directly.
- If stop_reason is "tool_use", the response contains a tool call with a name, input parameters, and a unique ID.
- Your code executes the tool and sends the result back as a tool_result message, matching the tool_use_id so Claude knows which call it corresponds to.
- If the tool call fails, you send is_error: True so Claude can handle the failure gracefully instead of hallucinating.
- The loop exits when stop_reason is "end_turn", meaning Claude has enough information to give a final answer.
💡 Why build the loop by hand instead of using the Tool Runner SDK?
Anthropic offers a Tool Runner SDK that abstracts the loop away. Building it manually teaches you the tool_use/tool_result lifecycle end-to-end. Once you understand the protocol, you can confidently debug and customize agent behavior in production.
- Save agent.py.
You won't see a change yet. The next chunk adds a test script so you can run the agent end-to-end.
Awesome, I've got everything!
Double check you have saved agent.py.
I'd like to double check the full code
import json
import time
import anthropic
from knowledge_base import create_knowledge_base, search_knowledge_base
SEARCH_TOOL = {
"name": "search_knowledge_base",
"description": "Search the company knowledge base for information about HR policies, finance procedures, and IT guidelines. Always use this tool before answering a question.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query to find relevant documents.",
},
"department": {
"type": "string",
"enum": ["hr", "finance", "it"],
"description": "Optional department filter to narrow search results.",
},
},
"required": ["query"],
},
}
SYSTEM_PROMPT = """You are a company knowledge base assistant. Your role is to answer employee questions using information from the company knowledge base.
Rules:
1. ALWAYS search the knowledge base before answering. Never answer from general knowledge.
2. Base your answer ONLY on the retrieved documents. Do not make up information.
3. Cite the document ID (e.g., [hr-001]) when referencing specific policies or procedures.
4. If the retrieved documents do not contain the answer, say "I don't have information about that in the knowledge base."
5. Be concise and direct in your answers."""
class RAGAgent:
def __init__(self, collection):
self.client = anthropic.Anthropic()
self.collection = collection
self.model = "claude-sonnet-4-5-20250929"
self.last_retrieved_contexts = []
self.last_metrics = {}
def query(self, user_question):
"""Run a query through the RAG agent with tool calling."""
start_time = time.time()
total_input_tokens = 0
total_output_tokens = 0
self.last_retrieved_contexts = []
messages = [{"role": "user", "content": user_question}]
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=[SEARCH_TOOL],
messages=messages,
)
total_input_tokens += response.usage.input_tokens
total_output_tokens += response.usage.output_tokens
while response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
try:
result = self._execute_tool(block.name, block.input)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result),
}
)
except Exception as exc:
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": str(exc),
"is_error": True,
}
)
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=[SEARCH_TOOL],
messages=messages,
)
total_input_tokens += response.usage.input_tokens
total_output_tokens += response.usage.output_tokens
final_text = next(
(block.text for block in response.content if block.type == "text"),
"No response generated.",
)
latency_ms = (time.time() - start_time) * 1000
self.last_metrics = {
"latency_ms": latency_ms,
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
"retrieval_count": len(self.last_retrieved_contexts),
"avg_retrieval_distance": (
sum(c["distance"] for c in self.last_retrieved_contexts)
/ len(self.last_retrieved_contexts)
if self.last_retrieved_contexts
else 0
),
}
return final_text
def _execute_tool(self, tool_name, tool_input):
"""Execute a tool call and return the result."""
if tool_name == "search_knowledge_base":
results = search_knowledge_base(
self.collection,
query=tool_input["query"],
department=tool_input.get("department"),
)
self.last_retrieved_contexts = results
return {
"results": [
{
"id": r["id"],
"document": r["document"],
"department": r["metadata"]["department"],
"distance": r["distance"],
}
for r in results
]
}
raise ValueError(f"Unknown tool: {tool_name}")Test the agent with sample queries
A good agent test covers three scenarios: a straightforward lookup, a cross-document question, and a question the knowledge base cannot answer. This verifies that the agent calls the tool, uses the context, and fails gracefully.
- Add a test block at the bottom of agent.py, outside the class, by copying this code:
if __name__ == "__main__":
collection = create_knowledge_base()
agent = RAGAgent(collection)
# Three test cases: direct lookup, cross-department, and out-of-scope
test_queries = [
"What is the PTO policy for new employees?",
"How do I get approval for a $15,000 purchase?",
"What are the company's stock trading hours?",
]
for q in test_queries:
print(f"\nQ: {q}")
answer = agent.query(q)
print(f"A: {answer}")
print(f"Metrics: {agent.last_metrics}")What does each test query check?
- "What is the PTO policy for new employees?" should return a clear answer from hr-001 with the 15-day accrual detail.
- "How do I get approval for a $15,000 purchase?" should pull from the finance department and reference the director approval threshold.
- "What are the company's stock trading hours?" has no answer in the knowledge base. The agent should say it does not have that information.
- Save agent.py.
- Run the agent tests by running this command in your terminal:
python agent.pyYou should see three question-answer pairs printed, each with a Metrics line showing latency, token counts, and retrieval statistics. The first two queries should return grounded answers with document citations. The third should say the information is not in the knowledge base.
Why does the first run take a bit longer?
ChromaDB downloads its default embedding model (all-MiniLM-L6-v2, about 80MB) the first time it runs. Subsequent runs reuse the cached model and are much faster.
🙋♀️ Seeing an error when running the agent?
- If you see AuthenticationError, your ANTHROPIC_API_KEY environment variable is missing or incorrect. Re-export it in your terminal session.
- If you see ModuleNotFoundError: No module named 'anthropic', make sure your virtual environment is activated. You should see (venv) in your terminal prompt.
- If you see ImportError related to knowledge_base, confirm that knowledge_base.py is saved in the same rag-agent folder as agent.py.
Help me debug my RAG agent error.
Your RAG agent is alive. It searches the knowledge base, uses the retrieved documents, cites its sources, and knows when to say "I don't know." Next up, you will wire in production observability so every query is tracked with latency, token counts, and retrieval scores in CloudWatch.
Add Production Observability with CloudWatch
In the last step, you built a RAG agent that searches your knowledge base and answers questions with citations. The agent already tracks metrics like latency and token counts internally through last_metrics.
But those metrics vanish the moment the process stops. In production, you need a persistent record of how your agent performs over time. In this step, you will build an observability module that pushes structured metrics to AWS CloudWatch and logs every query as a JSON audit trail.
In this step, get ready to:
- Create a MetricsCollector class that pushes custom metrics to CloudWatch.
- Record per-query metrics including latency, token usage, and retrieval scores.
- Add structured JSON logging and error tracking for a full audit trail.
Create the MetricsCollector class and CloudWatch client
CloudWatch is AWS's monitoring service. You send it custom metrics (numbers with names and dimensions), and it stores them so you can graph trends, set alarms, and trace individual requests.
- Create a new file called observability.py in your rag-agent folder.
- Add the imports and the MetricsCollector class skeleton by copying this code into the file:
import json
import logging
import time
import uuid
import boto3
class MetricsCollector:
def __init__(self, namespace="RAGAgent"):
self.cloudwatch = boto3.client("cloudwatch")
self.namespace = namespace
self.logger = self._setup_logger()
def _setup_logger(self):
"""Set up structured JSON logging."""
logger = logging.getLogger("rag_agent")
logger.setLevel(logging.INFO)
if not logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
return loggerWhat does this code do?
- boto3.client("cloudwatch") creates an AWS CloudWatch client using the credentials you configured in Step 1.
- The namespace groups all your agent's metrics under "RAGAgent" in the CloudWatch console, keeping them separate from other AWS services.
- _setup_logger() configures Python's built-in logging module with a minimal formatter. The %(message)s format outputs raw JSON strings without extra prefixes.
You won't see any output yet. The next chunk adds the method that actually pushes metrics.
Push query metrics to CloudWatch
Your agent already collects latency, token counts, and retrieval distances in last_metrics. The record_query method takes those metrics and pushes them to CloudWatch so they persist beyond the process lifetime.
- In observability.py, find the _setup_logger method, then add the record_query method below it:
def record_query(self, request_id, question, answer, metrics, tool_calls=None):
"""Record metrics for a single query to CloudWatch and structured logs."""
metric_data = [
{
"MetricName": "Latency",
"Value": metrics["latency_ms"],
"Unit": "Milliseconds",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
{
"MetricName": "InputTokens",
"Value": metrics["input_tokens"],
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
{
"MetricName": "OutputTokens",
"Value": metrics["output_tokens"],
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
{
"MetricName": "RetrievalCount",
"Value": metrics["retrieval_count"],
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
]
if metrics.get("avg_retrieval_distance", 0) > 0:
metric_data.append(
{
"MetricName": "AvgRetrievalDistance",
"Value": metrics["avg_retrieval_distance"],
"Unit": "None",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
}
)
self.cloudwatch.put_metric_data(
Namespace=self.namespace, MetricData=metric_data
)
log_entry = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"request_id": request_id,
"question": question,
"answer_length": len(answer),
"latency_ms": round(metrics["latency_ms"], 2),
"input_tokens": metrics["input_tokens"],
"output_tokens": metrics["output_tokens"],
"retrieval_count": metrics["retrieval_count"],
"avg_retrieval_distance": round(metrics.get("avg_retrieval_distance", 0), 4),
"tool_calls": tool_calls or [],
}
self.logger.info(json.dumps(log_entry))What does this code do?
- Each metric in metric_data has a name, a value, a unit, and a RequestId dimension. The dimension lets you trace a specific query in the CloudWatch console later.
- put_metric_data sends all five metrics (Latency, InputTokens, OutputTokens, RetrievalCount, and optionally AvgRetrievalDistance) to CloudWatch in a single API call.
- The log_entry dictionary creates a structured JSON log line. Structured logs are machine-parseable, so tools like CloudWatch Logs Insights can query them directly.
- AvgRetrievalDistance is only pushed when it's greater than 0. A value of 0 means no documents were retrieved, so there is no meaningful distance to record.
💡 Why structured logging instead of plain print statements?
Plain print() output is human-readable but impossible to query at scale. Structured JSON logs let you filter by request_id, search for slow queries by latency_ms, or find high-token queries. Every production system uses structured logging for exactly this reason.
You won't see output yet. The next chunk completes the module with error tracking and a request ID generator.
Add error tracking and a request ID generator
When a query fails, you need to track that separately. An ErrorCount metric in CloudWatch lets you spot spikes in failures without digging through logs.
- In observability.py, add the record_error method below record_query, then add the standalone generate_request_id function outside the class:
def record_error(self, request_id, question, error_message):
"""Record an error metric and log entry."""
self.cloudwatch.put_metric_data(
Namespace=self.namespace,
MetricData=[
{
"MetricName": "ErrorCount",
"Value": 1,
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
}
],
)
log_entry = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"request_id": request_id,
"question": question,
"error": error_message,
"level": "ERROR",
}
self.logger.error(json.dumps(log_entry))
def generate_request_id():
"""Generate a unique request ID for tracing."""
return str(uuid.uuid4())[:8]What does this code do?
- record_error pushes an ErrorCount metric of 1 to CloudWatch each time a query fails. In production, you would set a CloudWatch alarm on this metric to get notified when errors spike.
- The error log entry includes the question and the error message, tagged with "level": "ERROR" so log analysis tools can filter for failures.
- generate_request_id() creates an 8-character UUID prefix. Every query gets a unique ID so you can trace it from the log to the CloudWatch metric and back.
- Save observability.py.
- Verify the module loads correctly by running this command in your terminal:
python -c "from observability import MetricsCollector, generate_request_id; print(generate_request_id())"You should see an 8-character string like a3f2b1c4 printed to the terminal. This confirms the file imports cleanly and the ID generator works.
Seeing a ModuleNotFoundError for boto3?
- Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
- If boto3 is missing, reinstall it by running pip install boto3.
- If you see a NoCredentialsError, your AWS CLI is not configured. Run aws sts get-caller-identity to verify your credentials are set up.
Help me debug this boto3 import error in my observability module.
💡 Will CloudWatch cost me anything?
CloudWatch's free tier includes 10 custom metrics and 5GB of log ingestion per month. This project uses 6 metrics total, so you will stay well within the free tier. No charges are expected.
Awesome, I've got everything!
Great. Double check you have saved observability.py.
I'd like to double check the full code
import json
import logging
import time
import uuid
import boto3
class MetricsCollector:
def __init__(self, namespace="RAGAgent"):
self.cloudwatch = boto3.client("cloudwatch")
self.namespace = namespace
self.logger = self._setup_logger()
def _setup_logger(self):
"""Set up structured JSON logging."""
logger = logging.getLogger("rag_agent")
logger.setLevel(logging.INFO)
if not logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
return logger
def record_query(self, request_id, question, answer, metrics, tool_calls=None):
"""Record metrics for a single query to CloudWatch and structured logs."""
metric_data = [
{
"MetricName": "Latency",
"Value": metrics["latency_ms"],
"Unit": "Milliseconds",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
{
"MetricName": "InputTokens",
"Value": metrics["input_tokens"],
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
{
"MetricName": "OutputTokens",
"Value": metrics["output_tokens"],
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
{
"MetricName": "RetrievalCount",
"Value": metrics["retrieval_count"],
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
},
]
if metrics.get("avg_retrieval_distance", 0) > 0:
metric_data.append(
{
"MetricName": "AvgRetrievalDistance",
"Value": metrics["avg_retrieval_distance"],
"Unit": "None",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
}
)
self.cloudwatch.put_metric_data(
Namespace=self.namespace, MetricData=metric_data
)
log_entry = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"request_id": request_id,
"question": question,
"answer_length": len(answer),
"latency_ms": round(metrics["latency_ms"], 2),
"input_tokens": metrics["input_tokens"],
"output_tokens": metrics["output_tokens"],
"retrieval_count": metrics["retrieval_count"],
"avg_retrieval_distance": round(metrics.get("avg_retrieval_distance", 0), 4),
"tool_calls": tool_calls or [],
}
self.logger.info(json.dumps(log_entry))
def record_error(self, request_id, question, error_message):
"""Record an error metric and log entry."""
self.cloudwatch.put_metric_data(
Namespace=self.namespace,
MetricData=[
{
"MetricName": "ErrorCount",
"Value": 1,
"Unit": "Count",
"Dimensions": [{"Name": "RequestId", "Value": request_id}],
}
],
)
log_entry = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"request_id": request_id,
"question": question,
"error": error_message,
"level": "ERROR",
}
self.logger.error(json.dumps(log_entry))
def generate_request_id():
"""Generate a unique request ID for tracing."""
return str(uuid.uuid4())[:8]Your observability module is ready. Next up, you will build an evaluation pipeline that scores your agent's answers for faithfulness and relevance using Claude as the judge.
Build the Evaluation Pipeline with Claude-as-Judge
Your agent answers questions and CloudWatch tracks performance metrics like latency and token counts. But those numbers don't tell you whether the answers are actually correct.
In this step, you will build an evaluation pipeline that tests your agent against a golden dataset of known questions and answers. A second LLM (Claude Haiku) acts as the judge, scoring each response for faithfulness and relevance.
In this step, get ready to:
- Define a golden dataset of 10 test cases covering HR, finance, and IT questions.
- Implement faithfulness and relevance scoring functions using Claude as the judge.
- Run the full evaluation pipeline and check quality gates.
Create the golden dataset
A golden dataset is a set of test cases where you already know the correct answer. By running your agent against these known-good questions, you can measure whether it retrieves the right context and generates accurate responses.
- Create a new file called evaluate.py inside your rag-agent folder.
- Add the imports and golden dataset by pasting the following code:
import json
import anthropic
from agent import RAGAgent
from knowledge_base import create_knowledge_base
# 10 test cases with known-good answers spanning all departments
GOLDEN_DATASET = [
{
"question": "How many PTO days do new employees get per year?",
"expected_answer": "15 days per year during the first 3 years",
"expected_department": "hr",
},
{
"question": "How many days per week can I work remotely?",
"expected_answer": "Up to 3 days per week with manager approval",
"expected_department": "hr",
},
{
"question": "What is the meal reimbursement limit for business travel?",
"expected_answer": "$75 per day",
"expected_department": "hr",
},
{
"question": "What approval do I need for a $15,000 purchase?",
"expected_answer": "Director approval and a brief justification memo",
"expected_department": "finance",
},
{
"question": "What are the standard vendor payment terms?",
"expected_answer": "Net 30 from invoice date",
"expected_department": "finance",
},
{
"question": "What VPN client should I download?",
"expected_answer": "FortiClient VPN from the company software portal",
"expected_department": "it",
},
{
"question": "How often must passwords be changed?",
"expected_answer": "Every 90 days",
"expected_department": "it",
},
{
"question": "How long does non-standard software approval take?",
"expected_answer": "5-10 business days for security review",
"expected_department": "it",
},
{
"question": "What is the maximum PTO carryover?",
"expected_answer": "5 days into the next calendar year",
"expected_department": "hr",
},
{
"question": "What is the equipment stipend for remote workers?",
"expected_answer": "$500 annually",
"expected_department": "hr",
},
]What does this code do?
- The imports bring in your existing RAGAgent and create_knowledge_base so the evaluation pipeline can run queries against your agent.
- GOLDEN_DATASET defines 10 test cases, each with a question, an expected_answer (ground truth), and an expected_department.
- The test cases span all three departments (HR, finance, IT) and include both straightforward lookups and questions requiring synthesis from multiple documents.
- Save evaluate.py.
You won't see output yet. The scoring functions you build next will use this dataset.
Implement the scoring functions
The evaluation pipeline needs to answer two questions about every agent response. First, is the response faithful to the retrieved context? Second, does the response actually answer the question that was asked?
You will use Claude Haiku 4.5 as the judge for both checks. Each function sends a structured prompt to Haiku and parses the score from its JSON response.
Why Claude-as-judge instead of an evaluation framework?
Popular evaluation frameworks like RAGAS depend on OpenAI by default and pull in heavy dependencies like LangChain. Using Claude Haiku directly keeps your project focused on one provider, reduces costs, and teaches the LLM-as-judge pattern that production teams use widely.
Haiku 4.5 costs $1/$5 per million input/output tokens. Since the scoring prompts are short and simple, the total evaluation cost stays well under a few cents.
- In evaluate.py, add the two scoring functions below the GOLDEN_DATASET list:
def score_faithfulness(client, response, contexts):
"""Score whether claims in the response are supported by the context."""
context_text = "\n---\n".join(contexts)
scoring_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
messages=[
{
"role": "user",
"content": f"""Score the faithfulness of the Response based on the Context.
Faithfulness measures whether every claim in the response is supported by the context.
Context:
{context_text}
Response:
{response}
Reply with ONLY a JSON object in this format:
{{"score": <float between 0 and 1>, "reason": "<brief explanation>"}}""",
}
],
)
text = scoring_response.content[0].text
try:
result = json.loads(text)
return result["score"], result["reason"]
except (json.JSONDecodeError, KeyError):
return 0.0, f"Failed to parse scoring response: {text}"
def score_relevance(client, response, question):
"""Score whether the response actually answers the question."""
scoring_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
messages=[
{
"role": "user",
"content": f"""Score the relevance of the Response to the Question.
Relevance measures whether the response actually answers what was asked.
Question:
{question}
Response:
{response}
Reply with ONLY a JSON object in this format:
{{"score": <float between 0 and 1>, "reason": "<brief explanation>"}}""",
}
],
)
text = scoring_response.content[0].text
try:
result = json.loads(text)
return result["score"], result["reason"]
except (json.JSONDecodeError, KeyError):
return 0.0, f"Failed to parse scoring response: {text}"What do these functions do?
- score_faithfulness sends the agent's response and the retrieved context to Claude Haiku. Haiku scores whether every claim in the response is actually supported by the context (0 to 1).
- score_relevance measures the opposite dimension. It checks whether the response actually answers the question that was asked (0 to 1).
- Both functions use claude-haiku-4-5-20251001 to keep evaluation costs low. The prompt asks Haiku to reply with a JSON object containing a score and a reason.
- If Haiku returns something that cannot be parsed as JSON, the function returns a score of 0.0 with an error message instead of crashing.
- Save evaluate.py.
You won't see output yet. The evaluation pipeline in the next section calls these functions for every test case.
Run the evaluation pipeline
The run_evaluation function ties everything together. It sends each golden dataset question through the RAG agent, collects the retrieved contexts, scores both faithfulness and relevance, and prints a report with a pass/fail quality gate.
- In evaluate.py, add the run_evaluation function and the __main__ block below your scoring functions:
def run_evaluation():
"""Run the full evaluation pipeline."""
collection = create_knowledge_base()
agent = RAGAgent(collection)
judge_client = anthropic.Anthropic()
results = []
total_faithfulness = 0
total_relevance = 0
print("=" * 70)
print("RAG Agent Evaluation Report")
print("=" * 70)
for i, test_case in enumerate(GOLDEN_DATASET):
question = test_case["question"]
print(f"\n[{i + 1}/{len(GOLDEN_DATASET)}] {question}")
# Run the question through the agent
answer = agent.query(question)
contexts = [c["document"] for c in agent.last_retrieved_contexts]
# Score faithfulness and relevance using Claude as judge
faith_score, faith_reason = score_faithfulness(judge_client, answer, contexts)
rel_score, rel_reason = score_relevance(judge_client, answer, question)
total_faithfulness += faith_score
total_relevance += rel_score
# Quality gate: both metrics must be >= 0.7
passed = faith_score >= 0.7 and rel_score >= 0.7
status = "PASS" if passed else "FAIL"
print(f" Answer: {answer[:120]}...")
print(f" Faithfulness: {faith_score:.2f} - {faith_reason}")
print(f" Relevance: {rel_score:.2f} - {rel_reason}")
print(f" Status: [{status}]")
results.append(
{
"question": question,
"answer": answer,
"faithfulness": faith_score,
"relevance": rel_score,
"passed": passed,
}
)
# Print summary report
n = len(GOLDEN_DATASET)
avg_faithfulness = total_faithfulness / n
avg_relevance = total_relevance / n
pass_count = sum(1 for r in results if r["passed"])
print("\n" + "=" * 70)
print("Summary")
print("=" * 70)
print(f" Total test cases: {n}")
print(f" Passed: {pass_count}/{n}")
print(f" Avg Faithfulness: {avg_faithfulness:.2f}")
print(f" Avg Relevance: {avg_relevance:.2f}")
print(f" Quality Gate: {'PASS' if pass_count == n else 'FAIL'}")
print("=" * 70)
return results
if __name__ == "__main__":
run_evaluation()What does this code do?
- run_evaluation creates a fresh knowledge base and agent instance, plus a separate Anthropic client for the judge calls.
- For each test case, it runs the question through the agent, extracts the retrieved context documents from agent.last_retrieved_contexts, then scores both faithfulness and relevance.
- The quality gate threshold is 0.7 for both metrics. Any response scoring below that on either metric is flagged as FAIL.
- The summary at the end shows aggregate averages and an overall quality gate that only passes if every test case passes.
- Save evaluate.py.
- Run the evaluation pipeline by running this command:
python evaluate.pyThe pipeline sends 10 questions through your agent and scores each one. This takes 1-2 minutes because each test case involves multiple API calls (one to the agent, two to the judge).
You should see a per-question report with faithfulness and relevance scores, followed by a summary showing how many test cases passed the quality gate.
Seeing an API error or low scores?
- If you see an authentication error, make sure your ANTHROPIC_API_KEY environment variable is still set in this terminal session.
- If you see "Failed to parse scoring response" warnings, the judge model returned text that was not valid JSON. This can happen occasionally. Re-run the pipeline and the results should improve.
- If several test cases show FAIL, check that knowledge_base.py has all 8 documents. Missing documents mean the agent cannot find relevant context.
- Still stuck? Help me debug my evaluation pipeline. I'm seeing this error when I run python evaluate.py.
💡 What should I see?
Your terminal should show 10 test cases printing one at a time, each with an answer snippet, faithfulness score, relevance score, and a PASS or FAIL status. At the end, a summary block shows aggregate averages. Most test cases should pass with scores above 0.7.
Now double check that your full evaluate.py file matches the complete code.
Awesome, I've got everything!
Great. Double check you have saved evaluate.py.
I'd like to double check the full code
import json
import anthropic
from agent import RAGAgent
from knowledge_base import create_knowledge_base
GOLDEN_DATASET = [
{
"question": "How many PTO days do new employees get per year?",
"expected_answer": "15 days per year during the first 3 years",
"expected_department": "hr",
},
{
"question": "How many days per week can I work remotely?",
"expected_answer": "Up to 3 days per week with manager approval",
"expected_department": "hr",
},
{
"question": "What is the meal reimbursement limit for business travel?",
"expected_answer": "$75 per day",
"expected_department": "hr",
},
{
"question": "What approval do I need for a $15,000 purchase?",
"expected_answer": "Director approval and a brief justification memo",
"expected_department": "finance",
},
{
"question": "What are the standard vendor payment terms?",
"expected_answer": "Net 30 from invoice date",
"expected_department": "finance",
},
{
"question": "What VPN client should I download?",
"expected_answer": "FortiClient VPN from the company software portal",
"expected_department": "it",
},
{
"question": "How often must passwords be changed?",
"expected_answer": "Every 90 days",
"expected_department": "it",
},
{
"question": "How long does non-standard software approval take?",
"expected_answer": "5-10 business days for security review",
"expected_department": "it",
},
{
"question": "What is the maximum PTO carryover?",
"expected_answer": "5 days into the next calendar year",
"expected_department": "hr",
},
{
"question": "What is the equipment stipend for remote workers?",
"expected_answer": "$500 annually",
"expected_department": "hr",
},
]
def score_faithfulness(client, response, contexts):
"""Score whether claims in the response are supported by the context."""
context_text = "\n---\n".join(contexts)
scoring_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
messages=[
{
"role": "user",
"content": f"""Score the faithfulness of the Response based on the Context.
Faithfulness measures whether every claim in the response is supported by the context.
Context:
{context_text}
Response:
{response}
Reply with ONLY a JSON object in this format:
{{"score": <float between 0 and 1>, "reason": "<brief explanation>"}}""",
}
],
)
text = scoring_response.content[0].text
try:
result = json.loads(text)
return result["score"], result["reason"]
except (json.JSONDecodeError, KeyError):
return 0.0, f"Failed to parse scoring response: {text}"
def score_relevance(client, response, question):
"""Score whether the response actually answers the question."""
scoring_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
messages=[
{
"role": "user",
"content": f"""Score the relevance of the Response to the Question.
Relevance measures whether the response actually answers what was asked.
Question:
{question}
Response:
{response}
Reply with ONLY a JSON object in this format:
{{"score": <float between 0 and 1>, "reason": "<brief explanation>"}}""",
}
],
)
text = scoring_response.content[0].text
try:
result = json.loads(text)
return result["score"], result["reason"]
except (json.JSONDecodeError, KeyError):
return 0.0, f"Failed to parse scoring response: {text}"
def run_evaluation():
"""Run the full evaluation pipeline."""
collection = create_knowledge_base()
agent = RAGAgent(collection)
judge_client = anthropic.Anthropic()
results = []
total_faithfulness = 0
total_relevance = 0
print("=" * 70)
print("RAG Agent Evaluation Report")
print("=" * 70)
for i, test_case in enumerate(GOLDEN_DATASET):
question = test_case["question"]
print(f"\n[{i + 1}/{len(GOLDEN_DATASET)}] {question}")
answer = agent.query(question)
contexts = [c["document"] for c in agent.last_retrieved_contexts]
faith_score, faith_reason = score_faithfulness(judge_client, answer, contexts)
rel_score, rel_reason = score_relevance(judge_client, answer, question)
total_faithfulness += faith_score
total_relevance += rel_score
passed = faith_score >= 0.7 and rel_score >= 0.7
status = "PASS" if passed else "FAIL"
print(f" Answer: {answer[:120]}...")
print(f" Faithfulness: {faith_score:.2f} - {faith_reason}")
print(f" Relevance: {rel_score:.2f} - {rel_reason}")
print(f" Status: [{status}]")
results.append(
{
"question": question,
"answer": answer,
"faithfulness": faith_score,
"relevance": rel_score,
"passed": passed,
}
)
n = len(GOLDEN_DATASET)
avg_faithfulness = total_faithfulness / n
avg_relevance = total_relevance / n
pass_count = sum(1 for r in results if r["passed"])
print("\n" + "=" * 70)
print("Summary")
print("=" * 70)
print(f" Total test cases: {n}")
print(f" Passed: {pass_count}/{n}")
print(f" Avg Faithfulness: {avg_faithfulness:.2f}")
print(f" Avg Relevance: {avg_relevance:.2f}")
print(f" Quality Gate: {'PASS' if pass_count == n else 'FAIL'}")
print("=" * 70)
return results
if __name__ == "__main__":
run_evaluation()Your agent now has a systematic way to prove its answers are grounded in real documents. Next up, you will add a PII detection guardrail that scans both inputs and outputs before they reach the model or the user.
Add PII Detection Guardrails and Wire Everything Together
Your agent can answer questions, get evaluated by Claude-as-judge, and push metrics to CloudWatch. But there is a gap in the pipeline: nothing stops sensitive information from flowing through the system.
In this step, you will build a PII detection guardrail that scans both inputs and outputs. Then you will create the main entry point that wires every module together into one working application.
In this step, get ready to:
- Build a regex-based PII detector that catches emails, phone numbers, SSNs, and credit card numbers.
- Create the main entry point that runs queries through PII scanning, the agent, response redaction, and metrics collection.
- Test the full end-to-end pipeline with a PII-containing query.
Create the PII detection module
PII (Personally Identifiable Information) like email addresses, phone numbers, and Social Security numbers can appear in user queries or even in the knowledge base documents that Claude retrieves. A guardrail intercepts these before they reach the model or the user.
Why regex instead of a machine learning model?
Regex-based PII detection is fast, deterministic, and has zero external dependencies. For structured PII patterns like SSNs, credit cards, and phone numbers, regex catches the vast majority of cases. A production system would layer ML-based detection on top for unstructured PII (names, addresses), but regex is the right starting point.
- Create a new file called guardrails.py in your rag-agent folder.
- Add the PII regex patterns and safe-list configuration by pasting this code:
import re
# Regex patterns for common PII types
PII_PATTERNS = {
"email": re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"),
"phone": re.compile(r"\b\d{3}[-.?]\d{3}[-.?]\d{4}\b"),
"ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
"credit_card": re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"),
}
# Known safe patterns to exclude from PII detection
SAFE_PATTERNS = {
"email": {
"help@company.internal",
"support@acme.com",
},
}What does this code do?
- PII_PATTERNS is a dictionary mapping PII type names to compiled regex patterns. Each pattern matches a common format: emails with the @ symbol, 10-digit phone numbers, SSNs in XXX-XX-XXXX format, and 16-digit credit card numbers.
- SAFE_PATTERNS is a set of known-safe values that should not trigger PII alerts. Your knowledge base documents contain emails like help@company.internal that are company references, not personal data.
You will not see a change yet. The next chunk uses these patterns in a class.
- Add the PIIDetector class below the SAFE_PATTERNS dictionary by pasting this code:
class PIIDetector:
def __init__(self):
self.patterns = PII_PATTERNS
self.safe_patterns = SAFE_PATTERNS
def scan(self, text):
"""Scan text for PII and return a list of findings."""
findings = []
for pii_type, pattern in self.patterns.items():
matches = pattern.finditer(text)
for match in matches:
value = match.group()
# Skip known safe patterns like company emails
safe_set = self.safe_patterns.get(pii_type, set())
if value not in safe_set:
findings.append(
{
"type": pii_type,
"value": value,
"start": match.start(),
"end": match.end(),
}
)
return findings
def redact(self, text):
"""Redact PII from text, replacing with type labels."""
findings = self.scan(text)
# Sort in reverse order so positions stay valid as we replace
findings.sort(key=lambda f: f["start"], reverse=True)
redacted = text
for finding in findings:
replacement = f"[REDACTED_{finding['type'].upper()}]"
redacted = (
redacted[: finding["start"]]
+ replacement
+ redacted[finding["end"] :]
)
return redactedWhat does this code do?
- scan(text) iterates through every PII pattern, finds all matches in the input text, and filters out safe patterns. It returns a list of findings with the PII type, the matched value, and its position in the string.
- redact(text) calls scan() to find all PII, then replaces each match with a label like [REDACTED_EMAIL]. It sorts findings in reverse order so that earlier positions in the string stay valid as replacements happen.
You will not see a change yet. The next chunk adds a test script so you can verify the detector works.
- Add a test block at the bottom of guardrails.py to verify the detector by pasting this code:
if __name__ == "__main__":
detector = PIIDetector()
test_texts = [
"My email is john.doe@gmail.com and my SSN is 123-45-6789",
"Call me at 555-123-4567",
"Contact help@company.internal for support",
]
for text in test_texts:
findings = detector.scan(text)
print(f"Text: {text}")
print(f" Findings: {findings}")
print(f" Redacted: {detector.redact(text)}")
print()What does this test do?
It runs three test strings through the detector. The first contains an email and SSN (both should be detected). The second has a phone number. The third contains help@company.internal, which is in the safe list and should NOT be flagged.
- Save guardrails.py.
- Run the test by executing this command:
python guardrails.py- You should see output showing that john.doe@gmail.com and 123-45-6789 are detected in the first string, 555-123-4567 is detected in the second, and help@company.internal is skipped in the third.
- The redacted output for the first string should show [REDACTED_EMAIL] and [REDACTED_SSN] replacing the original values.
Not seeing any findings?
- Check that the regex strings in PII_PATTERNS use raw strings (prefix r"..."). Without the r prefix, backslashes get interpreted as escape characters.
- Make sure the import re statement is at the top of the file.
- Why is my PII detector not finding any matches?
💡 Is regex enough for production PII detection?
Regex catches structured PII patterns reliably (emails, SSNs, credit cards, phone numbers). It will not catch unstructured PII like names or addresses. A production system would layer ML-based detection (such as AWS Comprehend or Presidio) on top. This project demonstrates the guardrail pattern, which is the important part.
Awesome, I've got everything!
Double check you have saved your file before moving on.
I'd like to double check the full code
import re
PII_PATTERNS = {
"email": re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"),
"phone": re.compile(r"\b\d{3}[-.?]\d{3}[-.?]\d{4}\b"),
"ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
"credit_card": re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"),
}
# Known safe patterns to exclude from PII detection
SAFE_PATTERNS = {
"email": {
"help@company.internal",
"support@acme.com",
},
}
class PIIDetector:
def __init__(self):
self.patterns = PII_PATTERNS
self.safe_patterns = SAFE_PATTERNS
def scan(self, text):
"""Scan text for PII and return a list of findings."""
findings = []
for pii_type, pattern in self.patterns.items():
matches = pattern.finditer(text)
for match in matches:
value = match.group()
safe_set = self.safe_patterns.get(pii_type, set())
if value not in safe_set:
findings.append(
{
"type": pii_type,
"value": value,
"start": match.start(),
"end": match.end(),
}
)
return findings
def redact(self, text):
"""Redact PII from text, replacing with type labels."""
findings = self.scan(text)
findings.sort(key=lambda f: f["start"], reverse=True)
redacted = text
for finding in findings:
replacement = f"[REDACTED_{finding['type'].upper()}]"
redacted = (
redacted[: finding["start"]]
+ replacement
+ redacted[finding["end"] :]
)
return redacted
if __name__ == "__main__":
detector = PIIDetector()
test_texts = [
"My email is john.doe@gmail.com and my SSN is 123-45-6789",
"Call me at 555-123-4567",
"Contact help@company.internal for support",
]
for text in test_texts:
findings = detector.scan(text)
print(f"Text: {text}")
print(f" Findings: {findings}")
print(f" Redacted: {detector.redact(text)}")
print()Integrate PII detection into the query pipeline
The PII detector works in isolation. Now you need to wire it into the agent's query flow so that every input gets scanned and every output gets redacted before reaching the user.
- Create a new file called main.py in your rag-agent folder.
- Add the imports and the run_query function by pasting this code:
import sys
from agent import RAGAgent
from evaluate import run_evaluation
from guardrails import PIIDetector
from knowledge_base import create_knowledge_base
from observability import MetricsCollector, generate_request_id
def run_query(agent, metrics_collector, pii_detector, question):
"""Run a single query through the full pipeline."""
request_id = generate_request_id()
# Scan the user's input for PII before it reaches the model
input_findings = pii_detector.scan(question)
if input_findings:
print(f" [WARN] PII detected in input: {[f['type'] for f in input_findings]}")
try:
# Send the query through the RAG agent
answer = agent.query(question)
# Redact any PII from the agent's response
clean_answer = pii_detector.redact(answer)
# Push metrics to CloudWatch
metrics_collector.record_query(
request_id=request_id,
question=question,
answer=clean_answer,
metrics=agent.last_metrics,
)
return clean_answer, request_id
except Exception as exc:
metrics_collector.record_error(request_id, question, str(exc))
return f"Error processing query: {exc}", request_idWhat does this code do?
- The imports pull together every module you have built so far: the agent, the evaluator, the guardrails, the knowledge base, and the observability layer.
- run_query() is the single function that orchestrates one query through the entire pipeline. It generates a unique request ID, scans the input for PII, runs the agent, redacts the output, and pushes metrics to CloudWatch.
- If anything goes wrong, the except block records an error metric to CloudWatch so failed queries are tracked and visible.
You will not see a change yet. The next chunk adds the interactive loop that calls this function.
Wire everything together and test end-to-end
The main() function is the entry point for your entire application. It initializes every component and starts an interactive loop where you can ask the agent questions.
- Add the main() function and entry point below run_query in main.py by pasting this code:
def main():
# Handle the --evaluate flag to run the evaluation pipeline
if "--evaluate" in sys.argv:
print("Running evaluation pipeline...\n")
run_evaluation()
return
# Initialize all components
collection = create_knowledge_base()
agent = RAGAgent(collection)
metrics_collector = MetricsCollector()
pii_detector = PIIDetector()
print("Company Knowledge Base Assistant")
print("Type 'quit' to exit.\n")
# Interactive query loop
while True:
question = input("You: ").strip()
if question.lower() in ("quit", "exit", "q"):
break
if not question:
continue
answer, request_id = run_query(agent, metrics_collector, pii_detector, question)
print(f"\nAssistant: {answer}")
print(f" [Request ID: {request_id} | Latency: {agent.last_metrics.get('latency_ms', 0):.0f}ms | Tokens: {agent.last_metrics.get('input_tokens', 0)}in/{agent.last_metrics.get('output_tokens', 0)}out]\n")
if __name__ == "__main__":
main()What does this code do?
- The --evaluate flag check lets you run the full evaluation pipeline from Step 5 by passing a command-line argument. This gives you two modes: interactive chat and batch evaluation.
- The while True loop reads user input, runs it through the full pipeline (PII scan, agent query, redaction, metrics), and prints the response with request tracing metadata.
- Each response shows a request ID, latency in milliseconds, and token counts. This is the same data being pushed to CloudWatch, giving you immediate visibility.
- Save main.py.
- Run the full application by executing this command:
python main.py- You should see the prompt Company Knowledge Base Assistant followed by You: waiting for input.
What should I see?
ChromaDB will download its embedding model on first run if it has not already (about 80MB). This may take a minute on slower connections. After that, the interactive prompt appears.
Now test the PII guardrail by asking a question that includes an email address.
- At the You: prompt, type this query and press Enter:
My email is test.user@gmail.com, what is the PTO policy?- You should see a [WARN] PII detected in input: ['email'] warning printed before the agent's response.
- The agent's answer should describe the PTO policy (15 days per year, 20 after 3 years) with a citation like [hr-001].
- Below the answer you should see a request ID, latency, and token counts in square brackets.
- Type quit to exit the interactive loop.
Seeing an import error or connection error?
- If you see ModuleNotFoundError, make sure your virtual environment is activated and all packages are installed (anthropic, chromadb, boto3).
- If you see an Anthropic authentication error, verify your ANTHROPIC_API_KEY environment variable is set correctly.
- If CloudWatch metrics fail with a credentials error, verify your AWS CLI is configured by running aws sts get-caller-identity.
- Why is my main.py throwing an error when I run it?
Awesome, I've got everything!
Double check you have saved your file before moving on.
I'd like to double check the full code
import sys
from agent import RAGAgent
from evaluate import run_evaluation
from guardrails import PIIDetector
from knowledge_base import create_knowledge_base
from observability import MetricsCollector, generate_request_id
def run_query(agent, metrics_collector, pii_detector, question):
"""Run a single query through the full pipeline."""
request_id = generate_request_id()
input_findings = pii_detector.scan(question)
if input_findings:
print(f" [WARN] PII detected in input: {[f['type'] for f in input_findings]}")
try:
answer = agent.query(question)
clean_answer = pii_detector.redact(answer)
metrics_collector.record_query(
request_id=request_id,
question=question,
answer=clean_answer,
metrics=agent.last_metrics,
)
return clean_answer, request_id
except Exception as exc:
metrics_collector.record_error(request_id, question, str(exc))
return f"Error processing query: {exc}", request_id
def main():
if "--evaluate" in sys.argv:
print("Running evaluation pipeline...\n")
run_evaluation()
return
collection = create_knowledge_base()
agent = RAGAgent(collection)
metrics_collector = MetricsCollector()
pii_detector = PIIDetector()
print("Company Knowledge Base Assistant")
print("Type 'quit' to exit.\n")
while True:
question = input("You: ").strip()
if question.lower() in ("quit", "exit", "q"):
break
if not question:
continue
answer, request_id = run_query(agent, metrics_collector, pii_detector, question)
print(f"\nAssistant: {answer}")
print(f" [Request ID: {request_id} | Latency: {agent.last_metrics.get('latency_ms', 0):.0f}ms | Tokens: {agent.last_metrics.get('input_tokens', 0)}in/{agent.last_metrics.get('output_tokens', 0)}out]\n")
if __name__ == "__main__":
main()Your production-grade RAG agent is complete. Every query now flows through PII detection, the tool-calling agent, response redaction, and CloudWatch metrics. Next up, tackle the Secret Mission to add confidence-based routing.
Secret mission
Confidence-Based Routing with Low-Confidence Logging
One of the most important production patterns is knowing when your system does not know. In this challenge, you'll add confidence-based routing that skips the LLM call entirely when retrieval scores are too low. You'll also log these low-confidence queries separately so you can review them later and improve your knowledge base.
Before you go
This project runs almost entirely locally. ChromaDB runs in-memory and is discarded every time the script exits. The only cloud resource is the CloudWatch custom metrics you pushed to the RAGAgent namespace.
Cost warning
CloudWatch's free tier includes 10 custom metrics and 5GB of log ingestion per month. This project stays well within those limits. If you continue pushing metrics beyond the free tier, costs are approximately $0.30 per metric per month.
Resources you used:
- CloudWatch custom metrics under the RAGAgent namespace (AWS).
- Local project directory rag-agent/ with Python scripts and virtual environment.
- Sentence Transformers embedding model cache (downloaded by ChromaDB on first run).
- ANTHROPIC_API_KEY environment variable.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Keep everything running
No action needed. Choose this if you plan to keep experimenting with the agent or extending the evaluation pipeline.
- Your CloudWatch metrics will persist in the RAGAgent namespace and remain viewable in the AWS Console.
- Your local project files and virtual environment stay on your machine ready to use.
- CloudWatch metrics within the free tier cost nothing.
Pause - I'll come back to this later
There are no running processes to shut down since everything in this project runs on-demand. Your local files are safe to leave as-is.
- Deactivate your virtual environment by running deactivate in your terminal.
- When you come back, reactivate the virtual environment and all your packages will still be installed.
- CloudWatch metrics stay in your AWS account at no additional cost within the free tier.
Delete - I don't want to use this again
Remove all project resources and start fresh.
- Delete the CloudWatch custom metrics by opening the AWS Console, navigating to CloudWatch, selecting All metrics, and removing the RAGAgent namespace. CloudWatch custom metrics expire automatically after 15 months of no new data if you prefer to just leave them.
- Deactivate your virtual environment by running deactivate in your terminal.
- Delete the project directory by running rm -rf ~/Desktop/rag-agent (macOS/Linux) or removing the rag-agent folder from your Desktop (Windows).
- Unset your Anthropic API key environment variable if you no longer need it. Run unset ANTHROPIC_API_KEY in your terminal or remove it from your shell profile.
Mission accomplished
Nice work! You just built a production-grade RAG agent with tool orchestration, evaluation, and observability.
You've learned how to:
- Build a ChromaDB knowledge base and a tool-calling RAG agent with a full agentic loop using Claude's tool_use and tool_result protocol.
- Wire in production observability by pushing latency, token counts, retrieval scores, and error rates to AWS CloudWatch with structured logging.
- Build an evaluation pipeline with Claude-as-judge scoring for faithfulness and relevance, plus PII detection guardrails for input and output scanning.
- Secret Mission: Added confidence-based routing that knows when the system does not know, with separate monitoring for low-confidence queries.
Ready to quiz yourself?