RAG Agents
Build agents that combine retrieval-augmented generation with tool use — search, fetch, synthesize, and cite sources.
Real-World Scenario
A technical support agent handles questions about a 1,000-page API documentation. Instead of stuffing all docs into context, it retrieves the most relevant sections for each question, decides if it has enough information, searches again if needed, and answers with cited sources. When documentation is missing, it escalates to a human rather than guessing.
Basic RAG Agent
import anthropic
import json
from dataclasses import dataclass
client = anthropic.Anthropic()
# Simulated document store (in production: ChromaDB, Pinecone, pgvector)
DOCS = [
{
"id": "auth-001",
"title": "Authentication Guide",
"content": "Use Bearer tokens in the Authorization header. Tokens expire after 24 hours. Refresh with POST /auth/refresh.",
},
{
"id": "rate-001",
"title": "Rate Limiting",
"content": "Default rate limit is 100 requests/minute. Headers X-RateLimit-Remaining and X-RateLimit-Reset are included in every response.",
},
{
"id": "err-001",
"title": "Error Codes",
"content": "429: Rate limit exceeded. 401: Invalid or expired token. 403: Insufficient permissions. 404: Resource not found.",
},
{
"id": "ws-001",
"title": "WebSocket API",
"content": "Connect to wss://api.example.com/ws. Send JSON messages with type and payload fields. Ping every 30s to keep connection alive.",
},
]
def keyword_search(query: str, top_k: int = 3) -> list[dict]:
"""Simulate vector search with keyword overlap scoring."""
query_words = set(query.lower().split())
scored = []
for doc in DOCS:
text = (doc["title"] + " " + doc["content"]).lower()
score = sum(1 for w in query_words if w in text)
if score > 0:
scored.append({**doc, "score": score})
scored.sort(key=lambda x: x["score"], reverse=True)
return scored[:top_k]
# Tools the RAG agent can use
RAG_TOOLS = [
{
"name": "search_docs",
"description": "Search the documentation for information relevant to the user's question. Use specific keywords from the question.",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query — use keywords, not full sentences"},
"top_k": {"type": "integer", "description": "Number of results to return (1-5)", "default": 3},
},
"required": ["query"]
}
},
{
"name": "get_document",
"description": "Retrieve the full content of a specific document by its ID.",
"input_schema": {
"type": "object",
"properties": {
"doc_id": {"type": "string", "description": "The document ID from a previous search"}
},
"required": ["doc_id"]
}
},
{
"name": "answer_with_sources",
"description": "Provide the final answer with cited sources. Call this when you have enough information to answer confidently.",
"input_schema": {
"type": "object",
"properties": {
"answer": {"type": "string", "description": "The answer to the user's question"},
"sources": {
"type": "array",
"items": {"type": "string"},
"description": "List of document IDs used to formulate this answer"
},
"confidence": {
"type": "string",
"enum": ["high", "medium", "low"],
"description": "How confident you are in this answer"
}
},
"required": ["answer", "sources", "confidence"]
}
},
]
SYSTEM = """You are a technical support agent with access to API documentation.
Process:
1. Search the documentation for relevant information
2. If the first search isn't enough, search again with different keywords
3. When you have enough information, call answer_with_sources
4. If you cannot find the answer, say so clearly — never guess
Always cite which documents you used. If confidence is low, tell the user."""
def rag_agent(question: str, max_iterations: int = 6) -> dict:
messages = [{"role": "user", "content": question}]
final_ans = None
for iteration in range(max_iterations):
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=SYSTEM,
tools=RAG_TOOLS,
messages=messages,
)
if response.stop_reason == "end_turn":
text = next((b.text for b in response.content if hasattr(b, "text")), "")
return {"answer": text, "sources": [], "confidence": "low", "iterations": iteration + 1}
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type != "tool_use":
continue
if block.name == "search_docs":
results = keyword_search(block.input["query"], block.input.get("top_k", 3))
result_text = json.dumps([
{"id": r["id"], "title": r["title"], "snippet": r["content"][:200]}
for r in results
], indent=2)
print(f" [search] '{block.input['query']}' → {len(results)} results")
elif block.name == "get_document":
doc = next((d for d in DOCS if d["id"] == block.input["doc_id"]), None)
result_text = json.dumps(doc) if doc else f"Document {block.input['doc_id']} not found"
print(f" [fetch] {block.input['doc_id']}")
elif block.name == "answer_with_sources":
final_ans = {**block.input, "iterations": iteration + 1}
result_text = "Answer recorded."
print(f" [answer] confidence={block.input['confidence']}, sources={block.input['sources']}")
else:
result_text = "Unknown tool"
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result_text,
})
messages.append({"role": "user", "content": tool_results})
if final_ans:
return final_ans
return {"answer": "Could not find an answer.", "sources": [], "confidence": "low", "iterations": max_iterations}
# Test
questions = [
"How do I handle a 429 error from the API?",
"How long do authentication tokens last and how do I refresh them?",
"How do I keep a WebSocket connection alive?",
]
for q in questions:
print(f"\nQ: {q}")
result = rag_agent(q)
print(f"A: {result['answer']}")
print(f" Sources: {result['sources']} Confidence: {result['confidence']} Iterations: {result['iterations']}")Iterative Retrieval Agent
import anthropic
import json
client = anthropic.Anthropic()
def multi_hop_rag_agent(complex_question: str) -> str:
"""
Agent that breaks complex questions into sub-questions,
retrieves for each, then synthesizes a final answer.
"""
DECOMPOSE_TOOLS = [
{
"name": "decompose_question",
"description": "Break a complex question into simpler sub-questions that can each be answered with a single search.",
"input_schema": {
"type": "object",
"properties": {
"sub_questions": {
"type": "array",
"items": {"type": "string"},
"description": "List of 2-4 simpler sub-questions"
}
},
"required": ["sub_questions"]
}
}
]
# Step 1: Decompose into sub-questions
decompose_resp = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=512,
system="Break complex questions into simple, searchable sub-questions.",
tools=DECOMPOSE_TOOLS,
messages=[{"role": "user", "content": f"Decompose: {complex_question}"}]
)
sub_questions = []
for block in decompose_resp.content:
if block.type == "tool_use" and block.name == "decompose_question":
sub_questions = block.input["sub_questions"]
break
if not sub_questions:
sub_questions = [complex_question]
print(f"Sub-questions: {sub_questions}")
# Step 2: Retrieve for each sub-question
gathered_context = []
for sub_q in sub_questions:
results = keyword_search(sub_q, top_k=2)
for r in results:
gathered_context.append(f"[{r['id']}] {r['title']}: {r['content']}")
# Deduplicate
seen_ids = set()
unique_context = []
for ctx in gathered_context:
doc_id = ctx.split("]")[0][1:]
if doc_id not in seen_ids:
seen_ids.add(doc_id)
unique_context.append(ctx)
# Step 3: Synthesize final answer
context_str = "\n\n".join(unique_context)
synthesis = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system="Answer questions using only the provided documentation. Cite source IDs.",
messages=[{
"role": "user",
"content": f"Documentation:\n{context_str}\n\nQuestion: {complex_question}"
}]
)
return synthesis.content[0].text
answer = multi_hop_rag_agent(
"My API calls are failing with a 401 error and I'm also seeing high latency. "
"What are the likely causes and how should I debug this?"
)
print(f"\nFinal Answer:\n{answer}")Frequently Asked Questions
What makes a RAG agent different from basic RAG?
Basic RAG retrieves once and generates. A RAG agent can decide when to retrieve, what to search for, how many times to search (iterative refinement), which sources to trust, and when the retrieved context is insufficient. It uses retrieval as a tool in a larger reasoning loop rather than a fixed preprocessing step.
How do I prevent a RAG agent from hallucinating citations?
Two approaches: (1) include the full chunk text in the context and require the model to quote directly from it — hallucinated quotes are easy to detect. (2) Use a post-generation verification step that checks every cited claim against the retrieved text. Always store source metadata (URL, title, chunk ID) so citations are traceable.