Chapter 6: Building Your First AI Agent
Building Your First AI Agent
We've explored what AI agents are, how they work, and their various types. Now it's time to roll up our sleeves and build one. This chapter bridges theory and application by guiding you through creating your own functional AI agent.
Whether you're a developer looking to add AI capabilities to your projects or simply curious about how these systems come together, this hands-on approach will demystify the process. You'll start with simple implementations and gradually add sophisticated capabilities like memory, planning, and tool use.
By the end of this chapter, you'll have built a working AI agent and understand the fundamental patterns that apply across different frameworks and contexts.
Setting Up Your Development Environment
Before writing any agent code, you need to establish a suitable development environment. The right setup varies based on your preferred framework, but we'll focus on a Python-based approach that works for most agent development scenarios.
Essential Components
A complete agent development environment typically includes:
- Python: The most common language for AI development due to its readability and extensive libraries
- Virtual Environment: To manage dependencies and keep your projects isolated
- Key Libraries: Libraries for AI functionality, including potentially:
- A language model library (like OpenAI's API or Hugging Face's Transformers)
- Tools for agent design patterns (LangChain, AutoGPT, etc.)
- Domain-specific libraries for your agent's tasks
Let's walk through setting up this environment step by step.
Step-by-Step Environment Setup
First, ensure you have Python installed (preferably version 3.9+). Then create and activate a virtual environment:
# Create a directory for your agent project
mkdir myfirstagent
cd myfirstagent
# Create a virtual environment
python -m venv venv
# Activate the virtual environment
# On Windows:
venv\Scripts\activate
# On macOS/Linux:
source venv/bin/activate
Next, install the necessary libraries with specific versions for compatibility:
# Core libraries for agent development
pip install langchain==0.0.267 openai==0.27.8 requests==2.31.0 python-dotenv==1.0.0
# Optional: Additional libraries for specific capabilities
pip install pandas==2.0.3 numpy==1.24.3 matplotlib==3.7.2
Alternatively, create a requirements.txt file:
# requirements.txt
langchain==0.0.267
openai==0.27.8
requests==2.31.0
python-dotenv==1.0.0
pandas==2.0.3
numpy==1.24.3
matplotlib==3.7.2
Then install using:
pip install -r requirements.txt
Create a basic project structure:
myfirstagent/
├── venv/
├── .env # For API keys and configuration
├── agent.py # Main agent implementation
├── tools/ # For custom tools and integrations
│ └── init.py
├── requirements.txt # Dependencies list
├── tests/ # For test files
│ └── init.py
└── README.md # Documentation
In your .env file, you'll store any API keys needed (for language models, etc.):
OPENAIAPIKEY=yourkeyhere
WEATHERAPIKEY=yourweatherapikeyhere
# Other API keys as needed
Framework Selection
While you can build agents from scratch, using an established framework accelerates development. Popular options include:
- LangChain: Excellent for language model-powered agents with tools and memory
- AutoGPT: Good for autonomous agents that can perform multi-step reasoning
- RASA: Specialized for conversational agents and chatbots
- Microsoft Bot Framework: Enterprise-focused with Azure integration
Your choice depends on your specific goals. For this tutorial, we'll use LangChain as it offers flexibility while hiding unnecessary complexity.
Did You Know?
Before diving into agent development, many experienced developers sketch their agent's architecture on paper. This planning step helps clarify what capabilities your agent needs, what tools it should access, and how complex its decision-making process should be. A simple diagram showing inputs, processing steps, and outputs can prevent major redesigns later.
Simple Reflex Agent Implementation
Let's start with the simplest agent type: a reflex agent that follows condition-action rules. This agent will respond immediately to input without considering past states or planning ahead.
Example: Weather Response Agent
We'll create a simple agent that responds to weather queries with appropriate advice:
# agent.py
import os
import logging
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='agent_debug.log'
)
# Load environment variables
load_dotenv()
class SimpleWeatherAgent:
"""A simple reflex agent that responds to weather conditions."""
def init(self):
"""Initialize the weather agent with rules and language model."""
try:
# Initialize the language model
self.llm = OpenAI(temperature=0.7)
# Define the condition-action rules
self.rules = {
"rain": "It's raining! Don't forget your umbrella.",
"snow": "It's snowing! Wear warm clothes and be careful on the roads.",
"sunny": "It's sunny! Don't forget sunscreen if you're going out.",
"cloudy": "It's cloudy. You might want to take a light jacket.",
"windy": "It's windy! Secure any loose items if you're outdoors."
}
# For more complex responses beyond our rules
self.prompt = PromptTemplate(
input_variables=["weather", "location"],
template="The weather is {weather} in {location}. What should I wear and what activities are appropriate?"
)
self.chain = LLMChain(llm=self.llm, prompt=self.prompt)
logging.debug("SimpleWeatherAgent initialized successfully")
except Exception as e:
logging.error(f"Failed to initialize agent: {e}")
raise
def respond(self, weather_condition: str, location: str = None) -> str:
"""
Generate a response based on the current weather condition.
Args:
weather_condition: A string describing the weather
location: Optional location context for the weather
Returns:
A string response with weather advice
"""
if not weather_condition:
return "Please provide a weather condition."
logging.debug(f"Processing request - Weather: {weather_condition}, Location: {location}")
# Convert to lowercase for matching
weathercondition = weathercondition.lower()
# Check for matching conditions
matched_responses = []
for condition, response in self.rules.items():
if condition in weather_condition:
matched_responses.append(response)
# Return the first match if any found
if matched_responses:
logging.debug(f"Found matching rule: {matched_responses[0]}")
return matched_responses[0]
# If no rule matches and we have a location, use the language model
if location:
try:
logging.debug("No rule matched. Using language model for response.")
return self.chain.run(weather=weather_condition, location=location)
except Exception as e:
logging.error(f"Error using language model: {e}")
return "I'm having trouble processing that request. Please try again."
# Default response if nothing else applies
logging.debug("No rule matched and no location provided. Using default response.")
return "I'm not sure about that weather condition. Please check a forecast."
# Example usage
if name == "main":
try:
agent = SimpleWeatherAgent()
print(agent.respond("rain"))
print(agent.respond("partly cloudy with a chance of rain", "Seattle"))
except Exception as e:
logging.error(f"Error in example usage: {e}")
print(f"An error occurred: {e}")
This simple agent illustrates several key concepts:
- Condition-Action Rules: The agent has predefined responses for specific weather conditions
- Perception: It perceives weather through the input parameter
- Action: It takes action by returning appropriate advice
- Fallback Mechanism: For unknown conditions, it either uses a language model or provides a default response
- Error Handling: The code includes proper exception handling and logging
- Type Hints: Python type annotations clarify expected inputs and outputs
Run the agent with:
python agent.py
Understanding the Implementation
Notice how this reflex agent operates:
- It has no memory of past interactions
- It makes decisions purely based on current input
- It follows a simple mapping from conditions to actions
- It includes robust error handling to manage failures gracefully
While simple, this type of agent is useful for straightforward scenarios with clear rules. It responds quickly and predictably, making it suitable for applications where the environment states are well-defined and limited.
Try It Yourself
Expand the simple weather agent by adding more weather conditions and corresponding advice. Then, try creating a different reflex agent that handles another domain, such as a basic customer service agent that responds to common queries with predefined answers. What condition-action rules would you define?
Adding Memory and Planning Capabilities
A purely reactive agent is limited. Let's enhance our agent with memory to track past states and events, making it a model-based agent.
Implementing Memory
We'll extend our weather agent to remember previous queries and weather conditions:
# agent.py (extended with memory)
import os
import json
import logging
import threading
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='agent_debug.log'
)
# Load environment variables
load_dotenv()
class WeatherAgentWithMemory:
"""A weather agent that maintains memory of past interactions."""
def init(self, memoryfile: str = "agentmemory.json"):
"""
Initialize the weather agent with memory capabilities.
Args:
memory_file: File path to store persistent memory
"""
try:
self.llm = OpenAI(temperature=0.7)
# Same rules as before
self.rules = {
"rain": "It's raining! Don't forget your umbrella.",
"snow": "It's snowing! Wear warm clothes and be careful on the roads.",
"sunny": "It's sunny! Don't forget sunscreen if you're going out.",
"cloudy": "It's cloudy. You might want to take a light jacket.",
"windy": "It's windy! Secure any loose items if you're outdoors."
}
# Memory configuration
self.memoryfile = memoryfile
self.memory_lock = threading.Lock() # For thread safety
self.memory = self.loadmemory()
# Enhanced prompt that includes memory
self.prompt = PromptTemplate(
input_variables=["weather", "location", "history"],
template="""
The weather is {weather} in {location}.
Previous weather observations:
{history}
Given the current weather and previous observations, what should I wear and
what activities are appropriate? Also, mention any notable weather changes.
"""
)
self.chain = LLMChain(llm=self.llm, prompt=self.prompt)
logging.debug("WeatherAgentWithMemory initialized successfully")
except Exception as e:
logging.error(f"Failed to initialize agent with memory: {e}")
raise
def loadmemory(self) -> List[Dict[str, Any]]:
"""Load memory from file if available."""
try:
with open(self.memory_file, 'r') as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.info(f"No existing memory found or error reading memory: {e}. Starting with empty memory.")
return []
def savememory(self) -> None:
"""Save memory to persistent storage."""
try:
with open(self.memory_file, 'w') as f:
json.dump(self.memory, f)
except Exception as e:
logging.error(f"Failed to save memory: {e}")
def remember(self, weather: str, location: str) -> None:
"""
Add the current weather to memory.
Args:
weather: The current weather condition
location: The location of the weather observation
"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
with self.memory_lock: # Thread safety for multi-user scenarios
self.memory.append({
"timestamp": timestamp,
"weather": weather,
"location": location
})
# Keep memory from growing too large (retain last 5 entries)
if len(self.memory) > 5:
self.memory = self.memory[-5:]
# Save to persistent storage
self.savememory()
logging.debug(f"Added to memory: {weather} in {location} at {timestamp}")
def format_history(self) -> str:
"""
Format the memory for inclusion in the prompt.
Returns:
Formatted history string for the prompt
"""
if not self.memory:
return "No previous observations."
history = ""
for entry in self.memory:
history += f"- At {entry['timestamp']}: {entry['weather']} in {entry['location']}\n"
return history
def respond(self, weather_condition: str, location: str) -> str:
"""
Generate a response based on current weather and memory.
Args:
weather_condition: The current weather condition
location: The location to check weather for
Returns:
A response string with weather advice
"""
# Input validation
if not weather_condition:
return "Please provide a weather condition."
if not location:
return "Please provide a location."
try:
weathercondition = weathercondition.lower()
# Add to memory before responding
self.remember(weather_condition, location)
# Check for simple reflex responses
matched_responses = []
for condition, response in self.rules.items():
if condition in weather_condition:
matched_responses.append(response)
# If we have a match and memory, enhance the response
if matched_responses and len(self.memory) > 1:
history = self.format_history()
try:
context_response = self.chain.run(
weather=weather_condition,
location=location,
history=history
)
return f"{matchedresponses[0]}\n\nBased on recent weather patterns: {contextresponse}"
except Exception as e:
logging.error(f"Error enhancing response with context: {e}")
return matched_responses[0] # Fallback to simple response
elif matched_responses:
return matched_responses[0]
# For conditions not in our rules, use the LLM with memory context
history = self.format_history()
return self.chain.run(
weather=weather_condition,
location=location,
history=history
)
except Exception as e:
logging.error(f"Error generating response: {e}")
return "I'm sorry, I encountered an error while processing your request."
# Example usage
if name == "main":
try:
agent = WeatherAgentWithMemory()
print(agent.respond("sunny", "Los Angeles"))
print("\n---\n")
print(agent.respond("rainy", "Los Angeles")) # Weather changed
print("\n---\n")
print(agent.respond("rainy and windy", "Los Angeles")) # Weather worsened
except Exception as e:
logging.error(f"Error in example usage: {e}")
print(f"An error occurred: {e}")
Key enhancements in this version:
- The agent maintains a memory of past weather conditions and timestamps
- Memory is persistent across runs using file storage
- Thread safety is implemented with locks for multi-user scenarios
- The agent includes this history in generating responses, allowing for comparisons
- The agent can notice weather changes and adapt its recommendations accordingly
- Proper error handling ensures the agent degrades gracefully when issues occur
- Type hints clarify the expected data structures and function signatures
Adding Goal-Based Planning
Now let's incorporate planning to make our agent goal-oriented. We'll create a trip planning agent that determines the best activities based on weather forecasts:
# tripplanningagent.py
import os
import json
import logging
from typing import List, Dict, Any, Optional, Union
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='tripagentdebug.log'
)
# Load environment variables
load_dotenv()
class TripPlanningAgent:
"""A goal-based agent that plans activities based on weather forecasts."""
def init(self):
"""Initialize the trip planning agent with activity preferences and LLM."""
try:
self.llm = OpenAI(temperature=0.7)
# Define activity preferences by weather
self.activity_options = {
"sunny": ["beach visit", "hiking", "outdoor dining", "sightseeing"],
"cloudy": ["museum visit", "shopping", "city tour", "park walk"],
"rainy": ["museum visit", "shopping", "indoor dining", "spa day"],
"snow": ["skiing", "snowboarding", "cozy cafe visit", "snowshoeing"]
}
# Planning prompt
self.planning_prompt = PromptTemplate(
input_variables=["location", "duration", "forecast", "preferences", "activities"],
template="""
You are planning a trip to {location} for {duration} days.
The weather forecast is: {forecast}
Traveler preferences: {preferences}
Available activities based on weather: {activities}
Create a day-by-day itinerary that:
1. Maximizes enjoyment based on weather conditions
2. Respects traveler preferences
3. Provides a variety of activities
4. Includes practical logistics (best times, transportation tips)
Format as a daily schedule with morning, afternoon, and evening activities.
"""
)
self.planningchain = LLMChain(llm=self.llm, prompt=self.planningprompt)
logging.debug("TripPlanningAgent initialized successfully")
except Exception as e:
logging.error(f"Failed to initialize trip planning agent: {e}")
raise
def validateforecast(self, forecast: List[Dict[str, Any]]) -> bool:
"""
Validate forecast data structure.
Args:
forecast: The weather forecast data to validate
Returns:
True if valid, raises ValueError otherwise
"""
if not isinstance(forecast, list):
raise ValueError("Forecast should be a list of daily weather data")
for day in forecast:
if not isinstance(day, dict):
raise ValueError("Each forecast day should be a dictionary")
required_keys = ["day", "condition", "high", "low"]
missingkeys = [key for key in requiredkeys if key not in day]
if missing_keys:
raise ValueError(f"Forecast day missing required keys: {missing_keys}")
return True
def getavailableactivities(self, forecast: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Determine available activities based on forecast.
Args:
forecast: List of daily weather forecasts
Returns:
List of activities available for each day
"""
available_activities = []
for day in forecast:
day_weather = day["condition"].lower()
day_activities = []
# Check weather conditions and add appropriate activities
for weathertype, activities in self.activityoptions.items():
if weathertype in dayweather:
day_activities.extend(activities)
# If no specific weather matched, add cloudy activities as default
if not day_activities:
dayactivities = self.activityoptions["cloudy"]
available_activities.append({
"day": day["day"],
"weather": day["condition"],
"activities": day_activities
})
return available_activities
def formatforecast(self, forecast: List[Dict[str, Any]]) -> str:
"""
Format forecast data for the prompt.
Args:
forecast: The weather forecast data
Returns:
Formatted forecast string
"""
forecast_text = ""
for day in forecast:
forecast_text += f"Day {day['day']}: {day['condition']}, High: {day['high']}°F, Low: {day['low']}°F\n"
return forecast_text
def plan_trip(self, location: str, duration: int, forecast: List[Dict[str, Any]],
preferences: str) -> str:
"""
Generate a trip plan based on goals and constraints.
Args:
location: The trip destination
duration: Number of days for the trip
forecast: Weather forecast for the duration
preferences: Traveler preferences and constraints
Returns:
A complete trip itinerary
"""
try:
# Validate forecast data
self.validateforecast(forecast)
# Process the forecast data
availableactivities = self.getavailable_activities(forecast)
# Format activities for the prompt
activities_text = ""
for dayactivities in availableactivities:
activitiestext += f"Day {dayactivities['day']} ({day_activities['weather']}): "
activitiestext += ", ".join(dayactivities['activities']) + "\n"
# Generate the plan
plan = self.planning_chain.run(
location=location,
duration=str(duration),
forecast=self.formatforecast(forecast),
preferences=preferences,
activities=activities_text
)
logging.debug(f"Generated trip plan for {location} ({duration} days)")
return plan
except ValueError as e:
logging.error(f"Invalid input data: {e}")
return f"I couldn't create a trip plan: {str(e)}"
except Exception as e:
logging.error(f"Error generating trip plan: {e}")
return f"I'm sorry, I encountered an error while planning your trip: {str(e)}"
# Example usage
if name == "main":
try:
agent = TripPlanningAgent()
# Example forecast data
forecast = [
{"day": 1, "condition": "Sunny", "high": 75, "low": 65},
{"day": 2, "condition": "Partly Cloudy", "high": 72, "low": 63},
{"day": 3, "condition": "Rainy", "high": 68, "low": 60}
]
# Generate a trip plan
plan = agent.plan_trip(
location="San Francisco",
duration=3,
forecast=forecast,
preferences="Enjoys outdoor activities, cultural experiences, and local cuisine. Prefers to avoid crowds when possible."
)
print(plan)
except Exception as e:
logging.error(f"Error in example usage: {e}")
print(f"An error occurred: {e}")
This goal-based agent demonstrates several advanced capabilities:
- Goal Orientation: The agent aims to create an optimal trip plan considering multiple factors
- Planning: It maps out activities across multiple days based on forecasted conditions
- Constraint Handling: It accounts for weather limitations and traveler preferences
- Sequential Decision-Making: The agent makes interconnected decisions that form a coherent plan
- Input Validation: It carefully validates input data to prevent processing errors
- Error Handling: Various error conditions are properly managed with informative messages
The trip planning agent takes a multi-step approach:
- It analyzes weather forecasts to determine available activities
- It combines these options with traveler preferences
- It creates a comprehensive plan that optimizes enjoyment over multiple days
This represents a significantly more sophisticated agent than our initial reflex example, capable of planning over time and balancing multiple considerations.
Integrating External Tools
A powerful AI agent often needs to interact with external tools and services. Let's enhance our agent by adding tool integration capabilities.
The Tool Use Pattern
The tool use pattern enables agents to extend their capabilities by leveraging external functionalities. We'll implement a more versatile weather agent that can fetch real weather data:
# weathertoolagent.py
import os
import requests
import json
import logging
from typing import Dict, Any, List, Optional
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='weathertooldebug.log'
)
# Load environment variables
load_dotenv()
class WeatherTool:
"""Tool for fetching weather information from an API."""
def init(self, api_key: Optional[str] = None):
"""
Initialize the weather tool with API credentials.
Args:
api_key: API key for weather service
"""
self.apikey = apikey or os.getenv("WEATHERAPIKEY")
if not self.api_key:
raise ValueError("Weather API key is required. Provide it as a parameter or set WEATHERAPIKEY environment variable.")
self.base_url = "https://api.weatherapi.com/v1"
logging.debug("WeatherTool initialized")
def getcurrentweather(self, location: str) -> Dict[str, Any]:
"""
Fetch current weather for a location.
Args:
location: City or location name
Returns:
Dictionary with weather data or error information
"""
if not location or not isinstance(location, str):
return {"error": "Invalid location provided"}
endpoint = f"{self.base_url}/current.json"
params = {
"key": self.api_key,
"q": location
}
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raiseforstatus()
data = response.json()
# Extract relevant weather information
current = data["current"]
result = {
"condition": current["condition"]["text"],
"temperaturec": current["tempc"],
"temperaturef": current["tempf"],
"humidity": current["humidity"],
"windkph": current["windkph"],
"feelslikec": current["feelslike_c"],
"feelslikef": current["feelslike_f"]
}
logging.debug(f"Successfully retrieved current weather for {location}")
return result
except requests.exceptions.Timeout:
logging.error(f"Request timed out for {location}")
return {"error": "Request timed out. Please try again."}
except requests.exceptions.HTTPError as e:
statuscode = e.response.statuscode if hasattr(e, 'response') else "unknown"
logging.error(f"HTTP error {status_code} for {location}")
return {"error": f"HTTP error occurred: {status_code}"}
except requests.exceptions.ConnectionError:
logging.error(f"Connection error for {location}")
return {"error": "Connection error. Please check your internet connection."}
except KeyError as e:
logging.error(f"Unexpected API response format: missing {str(e)}")
return {"error": f"Unexpected API response format: missing {str(e)}"}
except Exception as e:
logging.error(f"Unexpected error retrieving weather: {str(e)}")
return {"error": f"An unexpected error occurred: {str(e)}"}
def get_forecast(self, location: str, days: int = 3) -> List[Dict[str, Any]]:
"""
Fetch weather forecast for a location.
Args:
location: City or location name
days: Number of days to forecast (1-10)
Returns:
List of daily forecasts or error dictionary
"""
if not location or not isinstance(location, str):
return {"error": "Invalid location provided"}
if not isinstance(days, int) or days < 1 or days > 10:
return {"error": "Days must be an integer between 1 and 10"}
endpoint = f"{self.base_url}/forecast.json"
params = {
"key": self.api_key,
"q": location,
"days": days
}
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raiseforstatus()
data = response.json()
# Extract forecast information
forecast_days = []
for day in data["forecast"]["forecastday"]:
forecast_days.append({
"date": day["date"],
"condition": day["day"]["condition"]["text"],
"maxtempc": day["day"]["maxtemp_c"],
"mintempc": day["day"]["mintemp_c"],
"chanceofrain": day["day"]["dailychanceof_rain"]
})
logging.debug(f"Successfully retrieved {days}-day forecast for {location}")
return forecast_days
except requests.exceptions.Timeout:
logging.error(f"Forecast request timed out for {location}")
return {"error": "Request timed out. Please try again."}
except requests.exceptions.HTTPError as e:
statuscode = e.response.statuscode if hasattr(e, 'response') else "unknown"
logging.error(f"HTTP error {status_code} for forecast request")
return {"error": f"HTTP error occurred: {status_code}"}
except requests.exceptions.ConnectionError:
logging.error(f"Connection error for forecast request")
return {"error": "Connection error. Please check your internet connection."}
except KeyError as e:
logging.error(f"Unexpected API response format in forecast: missing {str(e)}")
return {"error": f"Unexpected API response format: missing {str(e)}"}
except Exception as e:
logging.error(f"Unexpected error retrieving forecast: {str(e)}")
return {"error": f"An unexpected error occurred: {str(e)}"}
class WeatherAssistantAgent:
"""An agent that uses external weather data to provide recommendations."""
def init(self, weatherapikey: Optional[str] = None):
"""
Initialize the weather assistant agent.
Args:
weatherapikey: Optional API key for weather service
"""
try:
self.weatherapikey = weatherapikey or os.getenv("WEATHERAPIKEY")
if not self.weatherapikey:
raise ValueError("Weather API key is required. Provide it as a parameter or set WEATHERAPIKEY environment variable.")
self.llm = OpenAI(temperature=0.7)
self.weathertool = WeatherTool(apikey=self.weatherapikey)
# Memory for recent queries
self.memory = []
# Recommendation prompt
self.recommendation_prompt = PromptTemplate(
inputvariables=["location", "currentweather", "forecast", "query"],
template="""
You are a helpful weather assistant that provides personalized recommendations.
Location: {location}
Current Weather:
{current_weather}
3-Day Forecast:
{forecast}
User Query: {query}
Based on the weather data above, provide a helpful response to the user's query.
Include specific recommendations based on current conditions and the forecast.
If the query isn't weather-related, politely explain that you focus on weather assistance.
"""
)
self.recommendationchain = LLMChain(llm=self.llm, prompt=self.recommendationprompt)
logging.debug("WeatherAssistantAgent initialized successfully")
except Exception as e:
logging.error(f"Failed to initialize WeatherAssistantAgent: {e}")
raise
def formatcurrentweather(self, weatherdata: Dict[str, Any]) -> str:
"""
Format current weather data for the prompt.
Args:
weather_data: Current weather information
Returns:
Formatted weather string
"""
if "error" in weather_data:
return f"Error: {weather_data['error']}"
return f"""
Condition: {weather_data['condition']}
Temperature: {weatherdata['temperaturec']}°C / {weatherdata['temperaturef']}°F
Feels like: {weatherdata['feelslikec']}°C / {weatherdata['feelslikef']}°F
Humidity: {weather_data['humidity']}%
Wind: {weatherdata['windkph']} km/h
"""
def formatforecast(self, forecast_data: Union[List[Dict[str, Any]], Dict[str, Any]]) -> str:
"""
Format forecast data for the prompt.
Args:
forecast_data: Forecast data from weather API
Returns:
Formatted forecast string
"""
if isinstance(forecastdata, dict) and "error" in forecastdata:
return f"Error: {forecast_data['error']}"
result = ""
for day in forecast_data:
result += f"""
Date: {day['date']}
Condition: {day['condition']}
Temperature Range: {day['mintempc']}°C to {day['maxtempc']}°C
Chance of Rain: {day['chanceofrain']}%
"""
return result
def respond(self, location: str, query: str) -> str:
"""
Generate a response using weather data and language model.
Args:
location: City or location name
query: User's weather-related question
Returns:
Personalized weather recommendation
"""
# Input validation
if not location or not isinstance(location, str):
return "Please provide a valid location."
if not query or not isinstance(query, str):
return "Please provide a valid query."
try:
logging.debug(f"Processing request for {location}: {query}")
# Fetch weather data
currentweather = self.weathertool.getcurrentweather(location)
if "error" in current_weather:
return f"I couldn't get current weather information: {current_weather['error']}"
forecast = self.weathertool.getforecast(location)
if isinstance(forecast, dict) and "error" in forecast:
return f"I couldn't get forecast information: {forecast['error']}"
# Store in memory
self.memory.append({
"location": location,
"query": query,
"currentweather": currentweather
})
# Keep memory manageable
if len(self.memory) > 5:
self.memory = self.memory[-5:]
# Format data for the prompt
currentweathertext = self.formatcurrentweather(currentweather)
forecasttext = self.format_forecast(forecast)
# Generate recommendation
response = self.recommendation_chain.run(
location=location,
currentweather=currentweather_text,
forecast=forecast_text,
query=query
)
logging.debug(f"Generated response for {location}")
return response
except Exception as e:
logging.error(f"Error generating weather recommendation: {e}")
return f"I'm sorry, I encountered an error while getting your weather information: {str(e)}"
# Example usage
if name == "main":
try:
WEATHERAPIKEY = os.getenv("WEATHERAPIKEY")
if not WEATHERAPIKEY:
print("Please set your WEATHERAPIKEY in the .env file")
exit(1)
agent = WeatherAssistantAgent(WEATHERAPIKEY)
response = agent.respond(
location="London",
query="I'm planning to go sightseeing tomorrow. What should I wear and should I bring an umbrella?"
)
print(response)
except Exception as e:
logging.error(f"Error in example usage: {e}")
print(f"An error occurred: {e}")
This tool-integrated agent demonstrates:
- External API Integration: It connects to a real weather service to fetch live data
- Tool Abstraction: The weather functionality is encapsulated in a dedicated tool class
- Data Processing: It formats API responses into a structure useful for the agent
- Enhanced Context: The agent uses live data to provide meaningful recommendations
- Comprehensive Error Handling: Each API call and processing step includes proper error handling
- Input Validation: All user inputs are validated before processing
- Type Hinting: Python type annotations clarify the expected data structures
The agent now follows a more sophisticated workflow:
- Receives a user query about a location
- Retrieves real-time weather data from an external API
- Processes and formats this data
- Passes the processed information to a language model for interpretation
- Returns personalized recommendations based on actual weather conditions
This approach is much more powerful than our previous examples, as the agent can access real-world data rather than relying solely on predefined rules or simulated information.
Testing and Debugging Agent Behavior
Building an agent is only half the battle; ensuring it works correctly and handles edge cases is equally important. Let's explore strategies for testing and debugging AI agents.
Unit Testing Agent Components
First, create tests for each component of your agent. Here's a comprehensive test file for our weather agent:
# testweatheragent.py
import unittest
from unittest.mock import patch, MagicMock
import json
import os
import sys
from weathertoolagent import WeatherTool, WeatherAssistantAgent
class TestWeatherTool(unittest.TestCase):
"""Tests for the WeatherTool class."""
@patch('weathertoolagent.requests.get')
def testgetcurrentweathersuccess(self, mock_get):
"""Test successful weather retrieval."""
# Set up mock response
mock_response = MagicMock()
mockresponse.json.returnvalue = {
"current": {
"condition": {"text": "Sunny"},
"temp_c": 25,
"temp_f": 77,
"humidity": 65,
"wind_kph": 10,
"feelslike_c": 26,
"feelslike_f": 78.8
}
}
mockresponse.raiseforstatus.returnvalue = None
mockget.returnvalue = mock_response
# Create the tool and fetch weather
tool = WeatherTool(apikey="fakekey")
weather = tool.getcurrentweather("London")
# Verify results
self.assertEqual(weather["condition"], "Sunny")
self.assertEqual(weather["temperature_c"], 25)
self.assertEqual(weather["humidity"], 65)
@patch('weathertoolagent.requests.get')
def testgetcurrentweathererror(self, mock_get):
"""Test error handling for weather retrieval."""
# Set up mock to raise an exception
mockget.sideeffect = Exception("API error")
# Create the tool and fetch weather
tool = WeatherTool(apikey="fakekey")
result = tool.getcurrentweather("London")
# Verify error is handled
self.assertIn("error", result)
self.assertIn("API error", result["error"])
@patch('weathertoolagent.requests.get')
def testgetforecastsuccess(self, mockget):
"""Test successful forecast retrieval."""
# Set up mock response
mock_response = MagicMock()
mockresponse.json.returnvalue = {
"forecast": {
"forecastday": [
{
"date": "2023-01-01",
"day": {
"condition": {"text": "Sunny"},
"maxtemp_c": 28,
"mintemp_c": 20,
"dailychanceof_rain": 10
}
}
]
}
}
mockresponse.raiseforstatus.returnvalue = None
mockget.returnvalue = mock_response
# Create the tool and fetch forecast
tool = WeatherTool(apikey="fakekey")
forecast = tool.get_forecast("London", days=1)
# Verify results
self.assertEqual(len(forecast), 1)
self.assertEqual(forecast[0]["date"], "2023-01-01")
self.assertEqual(forecast[0]["condition"], "Sunny")
def testinvalidinputs(self):
"""Test handling of invalid inputs."""
tool = WeatherTool(apikey="fakekey")
# Test empty location
result = tool.getcurrentweather("")
self.assertIn("error", result)
# Test invalid days parameter
result = tool.get_forecast("London", days=0)
self.assertIn("error", result)
result = tool.get_forecast("London", days=11)
self.assertIn("error", result)
class TestWeatherAssistantAgent(unittest.TestCase):
"""Tests for the WeatherAssistantAgent class."""
@patch('weathertoolagent.WeatherTool')
@patch('weathertoolagent.OpenAI')
def testrespondsuccess(self, mockopenai, mockweather_tool):
"""Test successful response generation."""
# Set up mocks
mocktoolinstance = MagicMock()
mocktoolinstance.getcurrentweather.return_value = {
"condition": "Sunny",
"temperature_c": 25,
"temperature_f": 77,
"humidity": 65,
"wind_kph": 10,
"feelslikec": 26,
"feelslikef": 78.8
}
mocktoolinstance.getforecast.returnvalue = [
{
"date": "2023-01-01",
"condition": "Sunny",
"maxtempc": 28,
"mintempc": 20,
"chanceofrain": 10
}
]
mockweathertool.returnvalue = mocktool_instance
# Set up LLM mock
mockllminstance = MagicMock()
mock_chain = MagicMock()
mockchain.run.returnvalue = "It's sunny today, perfect for sightseeing! Wear light clothes and don't forget sunscreen."
mockllminstance.returnvalue = mockllm_instance
# Create a mock for the chain
with patch('weathertoolagent.LLMChain', returnvalue=mockchain):
# Create agent and get response
agent = WeatherAssistantAgent(weatherapikey="fake_key")
response = agent.respond(
location="London",
query="I'm planning to go sightseeing today. What should I wear?"
)
# Verify correct response
self.assertIn("sunny", response.lower())
self.assertIn("sightseeing", response.lower())
@patch('weathertoolagent.WeatherTool')
def testweatherapierror(self, mockweather_tool):
"""Test handling of weather API errors."""
# Set up mock to return error
mocktoolinstance = MagicMock()
mocktoolinstance.getcurrentweather.return_value = {
"error": "API connection error"
}
mockweathertool.returnvalue = mocktool_instance
# Create agent and get response
with patch('weathertoolagent.OpenAI'):
with patch('weathertoolagent.LLMChain'):
agent = WeatherAssistantAgent(weatherapikey="fake_key")
response = agent.respond(
location="London",
query="What's the weather like?"
)
# Verify error message is returned
self.assertIn("couldn't get current weather", response.lower())
self.assertIn("api connection error", response.lower())
def testemptylocation(self):
"""Test handling of empty location."""
with patch('weathertoolagent.OpenAI'):
with patch('weathertoolagent.LLMChain'):
agent = WeatherAssistantAgent(weatherapikey="fake_key")
response = agent.respond(
location="",
query="What's the weather like?"
)
# Verify error message
self.assertIn("provide a valid location", response.lower())
def testemptyquery(self):
"""Test handling of empty query."""
with patch('weathertoolagent.OpenAI'):
with patch('weathertoolagent.LLMChain'):
agent = WeatherAssistantAgent(weatherapikey="fake_key")
response = agent.respond(
location="London",
query=""
)
# Verify error message
self.assertIn("provide a valid query", response.lower())
if name == "main":
unittest.main()
Agent Evaluation Framework
Beyond unit tests, it's important to evaluate your agent's overall performance. Create an evaluation harness to test various scenarios:
# agent_evaluator.py
import json
import logging
from typing import List, Dict, Any, Optional
from weathertoolagent import WeatherAssistantAgent
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='evaluator.log'
)
class AgentEvaluator:
"""Evaluates agent performance across different scenarios."""
def init(self, agent: Any):
"""
Initialize the evaluator.
Args:
agent: The agent to evaluate
"""
self.agent = agent
self.test_cases = []
self.results = []
def loadtestcases(self, file_path: str) -> None:
"""
Load test cases from a JSON file.
Args:
file_path: Path to the test cases file
"""
try:
with open(file_path, 'r') as file:
self.test_cases = json.load(file)
logging.info(f"Loaded {len(self.testcases)} test cases from {filepath}")
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.error(f"Error loading test cases: {e}")
self.test_cases = []
def run_evaluation(self) -> None:
"""Run all test cases and record results."""
self.results = []
for i, case in enumerate(self.test_cases):
logging.info(f"Running test case {i+1}/{len(self.test_cases)}: {case.get('description', 'No description')}")
try:
# Get agent response
response = self.agent.respond(
location=case["location"],
query=case["query"]
)
# Check if expected error and got error
if case.get("expected_error", False):
if any(err in response.lower() for err in ["error", "sorry", "couldn't", "invalid"]):
status = "success" # Expected an error and got one
reason = "Successfully handled expected error case"
else:
status = "failure" # Expected an error but didn't get one
reason = "Expected error response but got success"
else:
# Validate response contains expected elements
expectedelements = case.get("expectedelements", [])
missing_elements = []
for element in expected_elements:
if element.lower() not in response.lower():
missing_elements.append(element)
if missing_elements:
status = "failure"
reason = f"Response missing expected elements: {missing_elements}"
else:
status = "success"
reason = "All expected elements found in response"
# Record result
self.results.append({
"test_case": case,
"response": response,
"status": status,
"reason": reason if "reason" in locals() else ""
})
except Exception as e:
# Record failure
error_msg = str(e)
logging.error(f"Error in test case {i+1}: {error_msg}")
self.results.append({
"test_case": case,
"error": error_msg,
"status": "success" if case.get("expected_error", False) else "failure",
"reason": "Exception occurred as expected" if case.get("expectederror", False) else f"Unexpected exception: {errormsg}"
})
logging.info(f"Evaluation complete. {len(self.results)} tests run.")
def saveresults(self, filepath: str) -> None:
"""
Save evaluation results to a JSON file.
Args:
file_path: Path to save results
"""
try:
with open(file_path, 'w') as file:
json.dump(self.results, file, indent=2)
logging.info(f"Results saved to {file_path}")
except Exception as e:
logging.error(f"Error saving results: {e}")
def print_summary(self) -> None:
"""Print a summary of evaluation results."""
total = len(self.results)
successes = sum(1 for result in self.results if result["status"] == "success")
failures = total - successes
print(f"Evaluation Summary:")
print(f"Total test cases: {total}")
print(f"Successes: {successes} ({successes/total*100:.1f}%)")
print(f"Failures: {failures} ({failures/total*100:.1f}%)")
if failures > 0:
print("\nFailed cases:")
for i, result in enumerate(self.results):
if result["status"] == "failure":
print(f"- Case {i+1}: {result['test_case'].get('description', 'No description')}")
print(f" Location: {result['testcase']['location']}, Query: {result['testcase']['query']}")
print(f" Reason: {result.get('reason', 'No reason provided')}")
if "error" in result:
print(f" Error: {result['error']}")
print()
# Example usage
if name == "main":
import os
from dotenv import load_dotenv
load_dotenv()
# Check for API key
WEATHERAPIKEY = os.getenv("WEATHERAPIKEY")
if not WEATHERAPIKEY:
print("Error: WEATHERAPIKEY environment variable not set!")
exit(1)
try:
# Create the agent
agent = WeatherAssistantAgent(WEATHERAPIKEY)
# Create evaluator
evaluator = AgentEvaluator(agent)
# Load test cases
evaluator.loadtestcases("test_cases.json")
# Run evaluation
evaluator.run_evaluation()
# Print summary
evaluator.print_summary()
# Save detailed results
evaluator.saveresults("evaluationresults.json")
except Exception as e:
logging.error(f"Error in evaluation: {e}")
print(f"An error occurred: {e}")
Create a JSON file with comprehensive test cases:
[
{
"description": "Basic umbrella query",
"location": "London",
"query": "Should I bring an umbrella today?",
"expected_elements": ["umbrella", "rain", "weather"]
},
{
"description": "Outdoor activities query",
"location": "New York",
"query": "What should I wear for outdoor activities this weekend?",
"expected_elements": ["wear", "activities", "temperature"]
},
{
"description": "Sightseeing query",
"location": "Tokyo",
"query": "Is it a good day for sightseeing?",
"expected_elements": ["sightseeing", "weather", "recommend"]
},
{
"description": "Sun protection query",
"location": "Paris",
"query": "Will I need sunscreen tomorrow?",
"expected_elements": ["sun", "protection"]
},
{
"description": "Invalid location test",
"location": "Invalid Location XYZ",
"query": "How's the weather?",
"expected_error": true
},
{
"description": "Empty location test",
"location": "",
"query": "What's the weather like?",
"expected_error": true
},
{
"description": "Empty query test",
"location": "London",
"query": "",
"expected_error": true
},
{
"description": "Very long query test",
"location": "Seattle",
"query": "I'm planning to go outside tomorrow for a very long walk in the park with my friends and I'm wondering if the weather will be suitable and if I should bring an umbrella or sunscreen or perhaps a light jacket depending on the conditions. Can you please give me detailed advice? " + "More detail please. " * 10,
"expected_elements": ["weather", "recommend"]
},
{
"description": "Non-weather query test",
"location": "Boston",
"query": "What's the best restaurant in town?",
"expected_elements": ["weather", "assist", "focus"]
}
]
Debugging Techniques for AI Agents
Debugging AI agents presents unique challenges. Here are effective strategies:
Verbose Logging: Add detailed logging throughout your agent implementation:
import logging
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='agent_debug.log'
)
# Then in your agent code
def respond(self, location, query):
logging.debug(f"Received request - Location: {location}, Query: {query}")
# Fetch weather data
logging.debug(f"Fetching current weather for {location}")
currentweather = self.weathertool.getcurrentweather(location)
logging.debug(f"Weather data received: {current_weather}")
# ...rest of the method
Step-by-Step Execution: Implement a debug mode that shows each step of the agent's decision process
class WeatherAssistantAgent:
# ...existing code...
def respond(self, location, query, debug_mode=False):
"""Generate a response with optional debug output."""
if debug_mode:
print(f"Step 1: Validating inputs - Location: {location}, Query: {query}")
# Input validation
if not location or not isinstance(location, str):
return "Please provide a valid location."
if not query or not isinstance(query, str):
return "Please provide a valid query."
try:
if debug_mode:
print(f"Step 2: Fetching current weather for {location}")
# Fetch weather data
currentweather = self.weathertool.getcurrentweather(location)
if debug_mode:
print(f"Weather data received: {json.dumps(current_weather, indent=2)}")
print(f"Step 3: Fetching forecast for {location}")
# ... and so on for each step
Prompt Inspection: Examine the exact prompts being sent to language models
def respond(self, location, query, debug_mode=False):
# ...existing code...
# Format data for the prompt
currentweathertext = self.formatcurrentweather(currentweather)
forecasttext = self.format_forecast(forecast)
if debug_mode:
print("\nPROMPT SENT TO LANGUAGE MODEL:")
print(self.recommendation_prompt.template)
print("\nINPUTS:")
print(f"Location: {location}")
print(f"Current Weather:\n{currentweathertext}")
print(f"Forecast:\n{forecast_text}")
print(f"Query: {query}")
# Generate recommendation
response = self.recommendation_chain.run(
location=location,
currentweather=currentweather_text,
forecast=forecast_text,
query=query
)
Response Analysis: Save raw API responses for analysis
def getcurrentweather(self, location):
# ...existing code...
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raiseforstatus()
raw_data = response.json()
# Save raw response for debugging
if not os.path.exists('debug'):
os.makedirs('debug')
with open(f'debug/weatherresponse{location}{datetime.now().strftime("%Y%m%d%H%M%S")}.json', 'w') as f:
json.dump(raw_data, f, indent=2)
# ...process response as normal
These techniques help identify where issues originate—whether in the agent logic, external API calls, or language model responses.
Try It Yourself
Run your agent with edge cases to see how it handles unusual inputs. Try queries with misspelled locations, extremely long inputs, or requests about locations with unusual weather patterns. Does your agent provide adequate responses or does it break? Use the debugging techniques we've discussed to identify and fix any issues.
Key Learnings & Takeaways
Building your first AI agent involves multiple steps, from environment setup to implementation of increasingly sophisticated capabilities. Key insights include:
- Start Simple: Begin with a reflex agent to establish basic functionality before adding complexity. This incremental approach makes debugging easier and ensures a solid foundation.
- Modular Design: Separate your agent's components (perception, memory, decision-making, actions) to make the code more maintainable and easier to test. This approach follows good software engineering practices.
- Memory Matters: Adding memory transforms a simple agent into one that can consider context and history, dramatically improving its utility for many real-world tasks.
- Planning Capabilities: Goal-based planning enables your agent to solve complex problems by considering sequences of actions toward objectives, not just immediate responses.
- Tool Integration: External tools extend your agent's capabilities beyond what it can do alone. By connecting to APIs and services, your agent can access real-world data and perform concrete actions.
- Testing is Crucial: Systematic testing helps identify edge cases and ensures your agent behaves as expected across different scenarios. Both unit tests and integrated testing frameworks are valuable.
- Error Handling: Robust error handling makes your agent resilient to unexpected inputs or external service failures. Never assume that tool calls will always succeed.
Each capability you add to your agent—memory, planning, or tool use—exponentially increases its potential usefulness while also introducing new complexity. Managing this complexity through good design patterns and thorough testing is essential for successful agent development.
Did You Know?
The "tool use" pattern we implemented represents one of the most powerful recent developments in AI agent design. When large language models are combined with external tools, they can overcome many of their inherent limitations. This pattern has been formalized by frameworks like LangChain and AutoGPT, and is sometimes called an "agentic workflow" because it allows AI to act more autonomously in the real world.
Conclusion: Your Agent Journey
In this chapter, we've taken you from theory to practice by building increasingly sophisticated AI agents. Starting with a simple reflex agent, we added memory, planning capabilities, and tool integration to create a more powerful and useful system.
The progression we followed—from reflex to model-based to goal-based agents—mirrors the evolution of AI agent technology over time. Each step adds capabilities but also complexity. The key is finding the right level of sophistication for your specific application.
As you continue developing AI agents, remember that the fundamentals remain consistent across frameworks and applications: agents perceive, reason, and act to achieve goals. The details of implementation may vary, but these core principles will serve you well in any agent development project.
In the next chapter, we'll explore even more advanced agent capabilities, including natural language understanding, multi-agent collaboration, and learning strategies. But with the foundation you've built here, you already have the skills to create useful, functional AI agents for a wide range of applications.
Recommended Next Steps
Ready to go further? Here are some recommended activities to continue your journey:
- Enhance Your Agent: Add more sophisticated capabilities to your weather agent, such as location detection, user preference tracking, or integration with calendar tools.
- Explore Different Domains: Apply the agent patterns you've learned to a different domain, such as a financial advisor agent or a recipe recommendation agent.
- Study Advanced Frameworks: Dive deeper into LangChain, AutoGPT, or other agent frameworks to leverage their more advanced features.
- Experiment with Multi-Agent Systems: Create two simple agents that can communicate with each other to solve a problem collaboratively.
- Contribute to the Community: Share your agent implementations on GitHub or contribute to open-source agent frameworks.
Remember, the best way to learn is by doing. As you build more agents, you'll develop intuition for what approaches work best in different scenarios and how to troubleshoot common issues that arise during development. Happy building!
