Your First Agent
This tutorial will guide you through building a complete RAG agent that can answer questions about your documents. You'll learn how to combine retrieval, generation, and evaluation into a working system.
What You'll Build
By the end of this tutorial, you'll have:
- A working RAG agent that answers questions
- A knowledge base with your documents
- Evaluation metrics to measure performance
- A foundation for building more complex agents
Prerequisites
- Completed Setup Environment Guide
- Basic Python knowledge
- OpenAI API key (or another LLM provider)
Step 1: Project Setup
Create Project Structure
mkdir my-first-agent
cd my-first-agent
# Create project files
touch agent.py
touch requirements.txt
touch .env
touch test_agent.py
Install Dependencies
# Create virtual environment
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
# Install RecoAgent
pip install recoagent
# Install additional dependencies
pip install python-dotenv pytest
Set Up Environment
# .env file
OPENAI_API_KEY=your_openai_api_key_here
LANGSMITH_API_KEY=your_langsmith_key_here
LANGSMITH_PROJECT=my-first-agent
Step 2: Create Sample Knowledge Base
Prepare Your Documents
Create a knowledge_base.txt file with sample content:
RecoAgent is an enterprise RAG platform built with LangGraph and LangChain.
Key Features:
- Hybrid retrieval combining BM25 and vector search
- Built-in evaluation with RAGAS metrics
- Safety guardrails using NVIDIA NeMo Guardrails
- Multiple vector store support (OpenSearch, Azure AI Search, Vertex AI)
- LangSmith integration for observability
Architecture:
RecoAgent follows a modular architecture with three main packages:
1. packages.agents - Agent orchestration and workflows
2. packages.rag - Retrieval, reranking, and evaluation
3. packages.observability - Monitoring, tracing, and metrics
Installation:
To install RecoAgent, use pip: pip install recoagent
Configuration:
RecoAgent can be configured through environment variables or configuration files.
Key settings include LLM provider, vector store type, and evaluation metrics.
Use Cases:
- Enterprise knowledge management
- Customer support automation
- Technical documentation Q&A
- Legal document analysis
- Research and development assistance
Step 3: Build the Agent
Basic Agent Implementation
# agent.py
import os
from dotenv import load_dotenv
from recoagent import RecoAgent
from recoagent.evaluators import RAGASEvaluator
# Load environment variables
load_dotenv()
class MyFirstAgent:
def __init__(self):
"""Initialize the RAG agent."""
self.agent = RecoAgent(
llm_provider="openai",
llm_model="gpt-3.5-turbo",
embedding_model="text-embedding-ada-002",
chunk_size=500,
chunk_overlap=50
)
self.evaluator = RAGASEvaluator()
def load_knowledge_base(self, file_path: str):
"""Load documents from a text file."""
with open(file_path, 'r', encoding='utf-8') as file:
content = file.read()
# Split into documents (simple approach)
documents = [doc.strip() for doc in content.split('\n\n') if doc.strip()]
# Add documents to agent
self.agent.add_documents(documents)
print(f"✅ Loaded {len(documents)} documents into knowledge base")
def ask_question(self, question: str):
"""Ask a question and get an answer."""
print(f"\n🤔 Question: {question}")
# Get response from agent
response = self.agent.ask(question)
print(f"💡 Answer: {response.answer}")
print(f"🎯 Confidence: {response.confidence:.2f}")
print(f"📚 Sources: {len(response.sources)} documents")
# Show sources
if response.sources:
print("\n📖 Source Documents:")
for i, source in enumerate(response.sources[:3], 1):
print(f" {i}. {source[:100]}...")
return response
def evaluate_performance(self):
"""Evaluate the agent's performance."""
print("\n📊 Evaluating Agent Performance...")
test_questions = [
"What is RecoAgent?",
"What are the key features of RecoAgent?",
"How do you install RecoAgent?",
"What vector stores does RecoAgent support?",
"What is the architecture of RecoAgent?"
]
results = self.evaluator.evaluate_rag_system(
agent=self.agent,
test_questions=test_questions
)
print(f"\n📈 Evaluation Results:")
print(f" Context Precision: {results.metrics.context_precision:.3f}")
print(f" Context Recall: {results.metrics.context_recall:.3f}")
print(f" Faithfulness: {results.metrics.faithfulness:.3f}")
print(f" Answer Relevancy: {results.metrics.answer_relevancy:.3f}")
return results
def main():
"""Main function to run the agent."""
print("🚀 Starting My First Agent...")
# Initialize agent
agent = MyFirstAgent()
# Load knowledge base
agent.load_knowledge_base("knowledge_base.txt")
# Interactive Q&A session
print("\n" + "="*50)
print("💬 Interactive Q&A Session")
print("="*50)
print("Type 'quit' to exit, 'eval' to run evaluation")
while True:
question = input("\n❓ Ask a question: ").strip()
if question.lower() == 'quit':
break
elif question.lower() == 'eval':
agent.evaluate_performance()
elif question:
agent.ask_question(question)
print("\n👋 Thanks for using My First Agent!")
if __name__ == "__main__":
main()
Step 4: Test the Agent
Create Test Script
# test_agent.py
import pytest
from agent import MyFirstAgent
def test_agent_initialization():
"""Test that agent initializes correctly."""
agent = MyFirstAgent()
assert agent.agent is not None
assert agent.evaluator is not None
print("✅ Agent initialization test passed")
def test_knowledge_base_loading():
"""Test knowledge base loading."""
agent = MyFirstAgent()
agent.load_knowledge_base("knowledge_base.txt")
# Check that documents were loaded
assert len(agent.agent.vector_store.get_all_documents()) > 0
print("✅ Knowledge base loading test passed")
def test_question_answering():
"""Test basic question answering."""
agent = MyFirstAgent()
agent.load_knowledge_base("knowledge_base.txt")
# Test simple question
response = agent.ask_question("What is RecoAgent?")
assert response.answer is not None
assert len(response.answer) > 0
assert response.confidence > 0
print("✅ Question answering test passed")
def test_evaluation():
"""Test evaluation functionality."""
agent = MyFirstAgent()
agent.load_knowledge_base("knowledge_base.txt")
results = agent.evaluate_performance()
assert results.metrics.context_precision >= 0
assert results.metrics.context_recall >= 0
assert results.metrics.faithfulness >= 0
assert results.metrics.answer_relevancy >= 0
print("✅ Evaluation test passed")
if __name__ == "__main__":
test_agent_initialization()
test_knowledge_base_loading()
test_question_answering()
test_evaluation()
print("\n🎉 All tests passed!")
Run Tests
python test_agent.py
Step 5: Run Your Agent
Start Interactive Session
python agent.py
Example Session:
🚀 Starting My First Agent...
✅ Loaded 8 documents into knowledge base
==================================================
💬 Interactive Q&A Session
==================================================
Type 'quit' to exit, 'eval' to run evaluation
❓ Ask a question: What is RecoAgent?
🤔 Question: What is RecoAgent?
💡 Answer: RecoAgent is an enterprise RAG platform built with LangGraph and LangChain. It provides hybrid retrieval, built-in evaluation with RAGAS metrics, safety guardrails, and support for multiple vector stores.
🎯 Confidence: 0.92
📚 Sources: 2 documents
📖 Source Documents:
1. RecoAgent is an enterprise RAG platform built with LangGraph and LangChain...
2. Key Features: - Hybrid retrieval combining BM25 and vector search...
❓ Ask a question: How do I install it?
🤔 Question: How do I install it?
💡 Answer: To install RecoAgent, you can use pip: pip install recoagent. This will install the core package and its dependencies.
🎯 Confidence: 0.88
📚 Sources: 1 documents
❓ Ask a question: eval
📊 Evaluating Agent Performance...
📈 Evaluation Results:
Context Precision: 0.85
Context Recall: 0.78
Faithfulness: 0.91
Answer Relevancy: 0.87
❓ Ask a question: quit
👋 Thanks for using My First Agent!
Step 6: Enhance Your Agent
Add Custom Tools
# enhanced_agent.py
from recoagent.tools import Tool
import requests
import json
class EnhancedAgent(MyFirstAgent):
def __init__(self):
super().__init__()
self._add_custom_tools()
def _add_custom_tools(self):
"""Add custom tools to the agent."""
# Web search tool
def web_search(query: str) -> str:
"""Search the web for current information."""
# This is a placeholder - in practice, you'd use a real search API
return f"Web search results for: {query}"
# Calculator tool
def calculator(expression: str) -> str:
"""Evaluate mathematical expressions safely."""
try:
result = eval(expression)
return str(result)
except:
return "Invalid mathematical expression"
# Add tools to agent
self.agent.add_tool(Tool(name="web_search", function=web_search))
self.agent.add_tool(Tool(name="calculator", function=calculator))
print("✅ Added custom tools: web_search, calculator")
Add Memory and Context
# Add conversation memory
class ConversationalAgent(MyFirstAgent):
def __init__(self):
super().__init__()
self.conversation_history = []
def ask_question(self, question: str):
"""Ask question with conversation context."""
# Add context from conversation history
context = ""
if self.conversation_history:
context = "\n".join([
f"Previous Q: {q}\nPrevious A: {a}"
for q, a in self.conversation_history[-3:] # Last 3 exchanges
])
# Enhance question with context
enhanced_question = f"{question}\n\nContext from conversation:\n{context}" if context else question
# Get response
response = super().ask_question(enhanced_question)
# Store in history
self.conversation_history.append((question, response.answer))
return response
Add Error Handling
# Add robust error handling
class RobustAgent(MyFirstAgent):
def ask_question(self, question: str):
"""Ask question with error handling."""
try:
response = super().ask_question(question)
# Check response quality
if response.confidence < 0.5:
print("⚠️ Low confidence answer. Consider rephrasing your question.")
return response
except Exception as e:
print(f"❌ Error processing question: {e}")
return None
def evaluate_performance(self):
"""Evaluate with error handling."""
try:
return super().evaluate_performance()
except Exception as e:
print(f"❌ Evaluation failed: {e}")
return None
Step 7: Deploy Your Agent
Create API Wrapper
# api.py
from flask import Flask, request, jsonify
from agent import MyFirstAgent
app = Flask(__name__)
agent = MyFirstAgent()
@app.route('/health', methods=['GET'])
def health_check():
"""Health check endpoint."""
return jsonify({"status": "healthy", "agent": "ready"})
@app.route('/ask', methods=['POST'])
def ask_question():
"""Ask a question endpoint."""
try:
data = request.get_json()
question = data.get('question', '')
if not question:
return jsonify({"error": "No question provided"}), 400
response = agent.ask_question(question)
return jsonify({
"question": question,
"answer": response.answer,
"confidence": response.confidence,
"sources": response.sources
})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route('/evaluate', methods=['POST'])
def evaluate():
"""Evaluate agent performance."""
try:
results = agent.evaluate_performance()
return jsonify({
"metrics": {
"context_precision": results.metrics.context_precision,
"context_recall": results.metrics.context_recall,
"faithfulness": results.metrics.faithfulness,
"answer_relevancy": results.metrics.answer_relevancy
}
})
except Exception as e:
return jsonify({"error": str(e)}), 500
if __name__ == '__main__':
# Load knowledge base
agent.load_knowledge_base("knowledge_base.txt")
# Start API server
app.run(host='0.0.0.0', port=5000, debug=True)
Test API
# Start API server
python api.py
# Test in another terminal
curl -X POST http://localhost:5000/ask \
-H "Content-Type: application/json" \
-d '{"question": "What is RecoAgent?"}'
Step 8: Monitor and Improve
Add Logging
import logging
from datetime import datetime
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('agent.log'),
logging.StreamHandler()
]
)
class LoggingAgent(MyFirstAgent):
def ask_question(self, question: str):
"""Ask question with logging."""
start_time = datetime.now()
# Log question
logging.info(f"Question received: {question}")
try:
response = super().ask_question(question)
# Log response
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
logging.info(f"Response generated in {duration:.2f}s with confidence {response.confidence:.2f}")
return response
except Exception as e:
logging.error(f"Error processing question: {e}")
raise
Performance Monitoring
# Add performance metrics
class MonitoredAgent(MyFirstAgent):
def __init__(self):
super().__init__()
self.metrics = {
"total_questions": 0,
"average_confidence": 0.0,
"average_response_time": 0.0,
"error_count": 0
}
def ask_question(self, question: str):
"""Ask question with metrics tracking."""
import time
self.metrics["total_questions"] += 1
start_time = time.time()
try:
response = super().ask_question(question)
# Update metrics
end_time = time.time()
response_time = end_time - start_time
# Update running averages
total = self.metrics["total_questions"]
self.metrics["average_confidence"] = (
(self.metrics["average_confidence"] * (total - 1) + response.confidence) / total
)
self.metrics["average_response_time"] = (
(self.metrics["average_response_time"] * (total - 1) + response_time) / total
)
return response
except Exception as e:
self.metrics["error_count"] += 1
raise
def get_metrics(self):
"""Get current performance metrics."""
return self.metrics.copy()
Troubleshooting
Common Issues
Import Error: No module named 'recoagent'
# Ensure virtual environment is activated
source venv/bin/activate
pip install recoagent
API Key Issues
# Check environment variables
echo $OPENAI_API_KEY
# Verify in Python
python -c "import os; print('API key set:', bool(os.getenv('OPENAI_API_KEY')))"
Low Performance Scores
# Improve chunking
agent = RecoAgent(
chunk_size=300, # Smaller chunks
chunk_overlap=100 # More overlap
)
# Add more documents
agent.add_documents(additional_documents)
Memory Issues
# Use streaming for large documents
agent = RecoAgent(
streaming=True,
batch_size=50
)
Next Steps
Congratulations! You've built your first RAG agent. Now you can:
- 🔧 Hybrid Retrieval - Learn advanced retrieval techniques
- 📚 Browse Examples - See more complex implementations
- 📖 API Reference - Deep dive into all features
- 🚀 How-To Guides - Solve specific problems
Summary
You've successfully:
- ✅ Built a complete RAG agent with retrieval and generation
- ✅ Created a knowledge base from your documents
- ✅ Implemented evaluation to measure performance
- ✅ Added error handling and monitoring
- ✅ Created an API for external access
- ✅ Learned best practices for agent development
Your agent can now answer questions about your documents with source citations and confidence scores. You have a solid foundation for building more sophisticated agents!
Ready for more? Check out Hybrid Retrieval to learn advanced techniques!