Agents Basics - Building Autonomous AI Systems
Learn to create intelligent agents that can reason, plan, and execute actions using tools to solve complex problems autonomously
🤖 Understanding AI Agents
AI agents are autonomous systems that can perceive their environment, reason about goals, make decisions, and take actions using available tools. They represent a paradigm shift from simple question-answering to dynamic problem-solving.
🧠 The Agent Mindset
text
🤖 AGENT VS TRADITIONAL AI 🤖
(From reactive to proactive)
┌─────────────────────────────────────────────────────────────────┐
│ TRADITIONAL AI SYSTEMS │
│ (Reactive approach) │
│ │
│ Input → Process → Output │
│ │
│ ❌ Single-turn interaction │
│ ❌ No memory or planning │
│ ❌ Limited to training data │
│ ❌ Can't use external tools │
└─────────────────────┬───────────────────────────────────────────┘
│
▼ EVOLUTION TO AGENTS
┌─────────────────────────────────────────────────────────────────┐
│ AI AGENT SYSTEMS │
│ (Autonomous approach) │
│ │
│ 🎯 Goal → 🧠 Plan → 🔧 Execute → 📊 Observe → 🔄 Adapt │
│ │
│ ✅ Multi-turn conversations │
│ ✅ Memory and learning │
│ ✅ Dynamic tool usage │
│ ✅ Goal-oriented behavior │
│ ✅ Error recovery and adaptation │
└─────────────────────────────────────────────────────────────────┘🏗️ Core Agent Components
🧠 Agent Architecture
python
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import Tool
from langchain.memory import ConversationBufferMemory
from typing import List, Dict, Any
# Basic agent architecture components
class AgentComponents:
"""Understand the core components of an AI agent"""
def __init__(self):
self.llm = ChatOpenAI(temperature=0.1)
self.memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
self.tools = []
def demonstrate_components(self):
"""Show how each component works"""
print("🧠 LLM (Language Model):")
print("- The 'brain' that reasons and makes decisions")
print("- Processes instructions and generates responses")
print("- Example: GPT-4, Claude, or local models")
print("\n🔧 Tools:")
print("- External functions the agent can call")
print("- Examples: web search, calculator, database queries")
print("- Bridge between AI reasoning and real-world actions")
print("\n💭 Memory:")
print("- Stores conversation history and context")
print("- Enables multi-turn conversations")
print("- Maintains state across interactions")
print("\n🎯 Prompt/Instructions:")
print("- Defines agent personality and capabilities")
print("- Provides guidance on tool usage")
print("- Sets behavioral guidelines")
print("\n🔄 Executor:")
print("- Orchestrates the agent's decision-making loop")
print("- Manages tool calling and response generation")
print("- Handles errors and retries")
# Demonstrate components
components = AgentComponents()
components.demonstrate_components()🛠️ Creating Basic Tools
python
import requests
import math
from datetime import datetime
# Simple calculation tool
def calculator(expression: str) -> str:
"""
Perform mathematical calculations safely.
Args:
expression: Mathematical expression to evaluate (e.g., "2 + 3 * 4")
Returns:
The result of the calculation as a string
"""
try:
# Safe evaluation - only allow basic math operations
allowed_chars = set('0123456789+-*/()., ')
if not all(c in allowed_chars for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression)
return f"The result of {expression} is {result}"
except Exception as e:
return f"Error calculating {expression}: {str(e)}"
# Web search simulation tool
def web_search(query: str) -> str:
"""
Search the web for information.
Args:
query: Search query string
Returns:
Simulated search results
"""
# In practice, you'd use a real search API like Google, Bing, or DuckDuckGo
simulated_results = {
"python": "Python is a high-level programming language known for its readability and versatility.",
"weather": "Current weather varies by location. For accurate weather, specify your city.",
"news": "Latest news includes updates on technology, politics, and world events.",
"langchain": "LangChain is a framework for developing applications powered by language models."
}
# Simple keyword matching for simulation
for keyword, result in simulated_results.items():
if keyword.lower() in query.lower():
return f"Search results for '{query}':\n{result}"
return f"Search results for '{query}':\nNo specific information found. Try a different search term."
# Current time tool
def get_current_time() -> str:
"""
Get the current date and time.
Returns:
Current date and time as a formatted string
"""
now = datetime.now()
return f"Current date and time: {now.strftime('%Y-%m-%d %H:%M:%S')}"
# File operations tool
def file_operations(operation: str, filename: str = "", content: str = "") -> str:
"""
Perform file operations like read, write, or list.
Args:
operation: Type of operation ('read', 'write', 'list')
filename: Name of the file (for read/write operations)
content: Content to write (for write operations)
Returns:
Result of the file operation
"""
import os
try:
if operation == "list":
files = os.listdir(".")
return f"Files in current directory: {', '.join(files[:10])}" # Limit to first 10
elif operation == "read":
if not filename:
return "Error: filename required for read operation"
# Safety check - only allow reading from current directory
if not os.path.exists(filename) or ".." in filename:
return f"Error: File {filename} not found or access denied"
with open(filename, 'r', encoding='utf-8') as f:
content = f.read()[:1000] # Limit to first 1000 characters
return f"Content of {filename}:\n{content}"
elif operation == "write":
if not filename or not content:
return "Error: Both filename and content required for write operation"
# Safety check - only allow writing to current directory
if ".." in filename:
return "Error: Access denied for security reasons"
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
return f"Successfully wrote to {filename}"
else:
return f"Error: Unknown operation '{operation}'. Use 'read', 'write', or 'list'."
except Exception as e:
return f"Error performing {operation}: {str(e)}"
# Convert functions to LangChain tools
calculator_tool = Tool(
name="calculator",
description="Perform mathematical calculations. Input should be a mathematical expression like '2 + 3 * 4'.",
func=calculator
)
search_tool = Tool(
name="web_search",
description="Search for information on the web. Input should be a search query string.",
func=web_search
)
time_tool = Tool(
name="current_time",
description="Get the current date and time. No input required.",
func=get_current_time
)
file_tool = Tool(
name="file_operations",
description="Perform file operations. Input format: 'operation|filename|content' where operation is 'read', 'write', or 'list'.",
func=lambda input_str: file_operations(*input_str.split('|', 2)) if '|' in input_str else file_operations(input_str)
)
# Tool list for agent
basic_tools = [calculator_tool, search_tool, time_tool, file_tool]
print("Created basic tools:")
for tool in basic_tools:
print(f"- {tool.name}: {tool.description}")🎯 Agent Prompt Templates
python
# System prompt for the agent
system_prompt = """You are a helpful AI assistant with access to various tools.
Your capabilities:
- Perform calculations using the calculator tool
- Search for information using the web search tool
- Get current date/time information
- Read and write files (with safety restrictions)
Guidelines:
1. Use tools when they would be helpful to answer the user's question
2. Explain your reasoning before using tools
3. If a tool fails, try alternative approaches
4. Be honest about limitations and uncertainties
5. Provide clear, helpful responses based on tool results
Remember: Always think step by step and use the most appropriate tool for each task."""
# Create agent prompt template
agent_prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
print("Agent prompt template created with:")
print("- System instructions for behavior")
print("- Chat history placeholder for memory")
print("- User input placeholder")
print("- Agent scratchpad for tool usage tracking")🤖 Building Your First Agent
🚀 Simple Agent Implementation
python
# Create a basic agent
def create_basic_agent():
"""Create a simple agent with basic tools"""
# Initialize LLM
llm = ChatOpenAI(temperature=0.1, model="gpt-3.5-turbo")
# Create the agent
agent = create_openai_functions_agent(
llm=llm,
tools=basic_tools,
prompt=agent_prompt
)
# Create agent executor
agent_executor = AgentExecutor(
agent=agent,
tools=basic_tools,
verbose=True, # Show reasoning steps
handle_parsing_errors=True,
max_iterations=5 # Prevent infinite loops
)
return agent_executor
# Test the basic agent
basic_agent = create_basic_agent()
print("🤖 Basic Agent Created!")
print("Testing with a calculation request...")
# Test calculation
try:
response = basic_agent.invoke({
"input": "What is 25 * 17 + 100, and what time is it now?"
})
print(f"Agent response: {response['output']}")
except Exception as e:
print(f"Error: {e}")
# Test with multiple tools
print("\nTesting with multiple tool usage...")
try:
response = basic_agent.invoke({
"input": "Search for information about Python programming, then calculate 15% of 250"
})
print(f"Agent response: {response['output']}")
except Exception as e:
print(f"Error: {e}")🧠 Agent with Memory
python
from langchain.memory import ConversationBufferWindowMemory
class MemoryEnabledAgent:
"""Agent with conversation memory capabilities"""
def __init__(self, tools: List[Tool], memory_window: int = 10):
self.llm = ChatOpenAI(temperature=0.1)
self.tools = tools
# Memory with sliding window
self.memory = ConversationBufferWindowMemory(
k=memory_window, # Keep last k interactions
memory_key="chat_history",
return_messages=True
)
# Enhanced prompt with memory awareness
self.prompt = ChatPromptTemplate.from_messages([
("system", """You are a helpful AI assistant with access to tools and memory of our conversation.
Available tools: {tool_names}
Instructions:
1. Use conversation history to provide contextual responses
2. Reference previous interactions when relevant
3. Use tools when they would help answer questions
4. Be consistent with information provided in earlier responses
5. If the user refers to "it", "that", or "this", use context to understand what they mean
Think step by step and use tools appropriately."""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# Create agent with memory
self.agent = create_openai_functions_agent(
llm=self.llm,
tools=self.tools,
prompt=self.prompt
)
# Agent executor with memory
self.agent_executor = AgentExecutor(
agent=self.agent,
tools=self.tools,
memory=self.memory,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
def chat(self, message: str) -> str:
"""Have a conversation with the agent"""
try:
# Get tool names for prompt
tool_names = [tool.name for tool in self.tools]
response = self.agent_executor.invoke({
"input": message,
"tool_names": ", ".join(tool_names)
})
return response["output"]
except Exception as e:
return f"I encountered an error: {str(e)}"
def get_conversation_history(self) -> List[str]:
"""Get formatted conversation history"""
history = []
for message in self.memory.chat_memory.messages:
role = "Human" if message.type == "human" else "Assistant"
history.append(f"{role}: {message.content}")
return history
# Create memory-enabled agent
memory_agent = MemoryEnabledAgent(basic_tools, memory_window=6)
print("🧠 Memory-Enabled Agent Created!")
# Test conversation flow
print("\n=== Conversation Test ===")
# First interaction
response1 = memory_agent.chat("Calculate 15 * 24 and remember this number")
print(f"Response 1: {response1}")
# Second interaction referring to previous
response2 = memory_agent.chat("Now divide that result by 3")
print(f"Response 2: {response2}")
# Third interaction with context
response3 = memory_agent.chat("What was the original calculation I asked for?")
print(f"Response 3: {response3}")
# Show conversation history
print("\n=== Conversation History ===")
history = memory_agent.get_conversation_history()
for entry in history:
print(entry)🎯 Specialized Agent Types
🔍 Research Agent
python
class ResearchAgent:
"""Specialized agent for research and analysis tasks"""
def __init__(self):
self.llm = ChatOpenAI(temperature=0.1)
self.research_tools = self._create_research_tools()
self.agent_executor = self._create_research_agent()
def _create_research_tools(self) -> List[Tool]:
"""Create tools specific to research tasks"""
def analyze_topic(topic: str) -> str:
"""Analyze a topic and suggest research approaches"""
analysis_prompt = f"""
Analyze the research topic: {topic}
Provide:
1. Key aspects to investigate
2. Potential sources of information
3. Important questions to explore
4. Methodology suggestions
Analysis:"""
response = self.llm.invoke(analysis_prompt)
return response.content
def create_research_plan(topic: str) -> str:
"""Create a structured research plan"""
plan_prompt = f"""
Create a research plan for: {topic}
Include:
1. Research objectives
2. Step-by-step methodology
3. Timeline considerations
4. Expected outcomes
Research Plan:"""
response = self.llm.invoke(plan_prompt)
return response.content
def synthesize_information(sources: str) -> str:
"""Synthesize information from multiple sources"""
synthesis_prompt = f"""
Synthesize the following information sources:
{sources}
Provide:
1. Key findings summary
2. Common themes
3. Conflicting information (if any)
4. Conclusions
Synthesis:"""
response = self.llm.invoke(synthesis_prompt)
return response.content
return [
Tool(
name="analyze_topic",
description="Analyze a research topic to identify key aspects and approaches. Input: topic name",
func=analyze_topic
),
Tool(
name="create_research_plan",
description="Create a structured research plan for a topic. Input: topic name",
func=create_research_plan
),
Tool(
name="synthesize_information",
description="Synthesize information from multiple sources. Input: source information",
func=synthesize_information
),
search_tool, # Include web search
file_tool # Include file operations
]
def _create_research_agent(self) -> AgentExecutor:
"""Create the research agent"""
research_prompt = ChatPromptTemplate.from_messages([
("system", """You are a specialized research assistant with expertise in:
- Analyzing research topics
- Creating research plans
- Synthesizing information from multiple sources
- Conducting systematic investigations
Your approach:
1. Break down complex topics into manageable parts
2. Use systematic research methodologies
3. Synthesize information from multiple sources
4. Provide evidence-based conclusions
5. Identify gaps in knowledge or conflicting information
Available tools: {tool_names}
Use tools strategically to conduct thorough research."""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_functions_agent(
llm=self.llm,
tools=self.research_tools,
prompt=research_prompt
)
return AgentExecutor(
agent=agent,
tools=self.research_tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=8 # More iterations for complex research
)
def research(self, topic: str) -> str:
"""Conduct research on a topic"""
tool_names = [tool.name for tool in self.research_tools]
response = self.agent_executor.invoke({
"input": f"Conduct comprehensive research on: {topic}",
"tool_names": ", ".join(tool_names)
})
return response["output"]
# Test research agent
research_agent = ResearchAgent()
print("🔍 Research Agent Created!")
print("\nConducting research on machine learning...")
research_result = research_agent.research(
"The impact of transformer models on natural language processing"
)
print(f"Research Result: {research_result}")🛠️ Problem-Solving Agent
python
class ProblemSolvingAgent:
"""Agent specialized in breaking down and solving complex problems"""
def __init__(self):
self.llm = ChatOpenAI(temperature=0.1)
self.problem_tools = self._create_problem_tools()
self.agent_executor = self._create_problem_agent()
def _create_problem_tools(self) -> List[Tool]:
"""Create tools for problem-solving"""
def break_down_problem(problem: str) -> str:
"""Break a complex problem into smaller subproblems"""
breakdown_prompt = f"""
Break down this complex problem into smaller, manageable subproblems:
Problem: {problem}
Provide:
1. Main problem components
2. Dependencies between components
3. Suggested order of solving
4. Potential challenges for each component
Problem Breakdown:"""
response = self.llm.invoke(breakdown_prompt)
return response.content
def evaluate_solution(problem: str, solution: str) -> str:
"""Evaluate a proposed solution"""
evaluation_prompt = f"""
Evaluate this solution for the given problem:
Problem: {problem}
Solution: {solution}
Evaluate:
1. Completeness (does it address all aspects?)
2. Feasibility (is it practical to implement?)
3. Potential issues or edge cases
4. Suggestions for improvement
Evaluation:"""
response = self.llm.invoke(evaluation_prompt)
return response.content
def generate_alternatives(problem: str, current_solution: str) -> str:
"""Generate alternative approaches to solving a problem"""
alternatives_prompt = f"""
Generate alternative approaches to solve this problem:
Problem: {problem}
Current Solution: {current_solution}
Provide:
1. 3-5 alternative approaches
2. Pros and cons of each
3. When each approach would be most suitable
4. Hybrid approaches combining methods
Alternative Approaches:"""
response = self.llm.invoke(alternatives_prompt)
return response.content
return [
Tool(
name="break_down_problem",
description="Break a complex problem into smaller subproblems. Input: problem description",
func=break_down_problem
),
Tool(
name="evaluate_solution",
description="Evaluate a proposed solution. Input: 'problem|solution' separated by |",
func=lambda input_str: evaluate_solution(*input_str.split('|', 1)) if '|' in input_str else "Error: Need problem|solution format"
),
Tool(
name="generate_alternatives",
description="Generate alternative solutions. Input: 'problem|current_solution' separated by |",
func=lambda input_str: generate_alternatives(*input_str.split('|', 1)) if '|' in input_str else "Error: Need problem|solution format"
),
calculator_tool, # Include calculator for numerical problems
search_tool # Include search for additional information
]
def _create_problem_agent(self) -> AgentExecutor:
"""Create the problem-solving agent"""
problem_prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert problem-solving agent with a systematic approach:
Problem-Solving Methodology:
1. Understand the problem thoroughly
2. Break down complex problems into subproblems
3. Analyze each component
4. Generate multiple solution approaches
5. Evaluate solutions objectively
6. Provide clear, actionable recommendations
Your strengths:
- Logical reasoning and analysis
- Creative problem-solving approaches
- Risk assessment and mitigation
- Step-by-step solution development
Use tools to enhance your problem-solving process."""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_functions_agent(
llm=self.llm,
tools=self.problem_tools,
prompt=problem_prompt
)
return AgentExecutor(
agent=agent,
tools=self.problem_tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=10
)
def solve(self, problem: str) -> str:
"""Solve a complex problem using systematic approach"""
response = self.agent_executor.invoke({
"input": f"Help me solve this problem systematically: {problem}"
})
return response["output"]
# Test problem-solving agent
problem_agent = ProblemSolvingAgent()
print("🛠️ Problem-Solving Agent Created!")
print("\nSolving a complex problem...")
problem_result = problem_agent.solve("""
I need to optimize the performance of a web application that's experiencing slow response times.
The app serves 10,000 users daily, uses a PostgreSQL database, and is built with Python/Django.
Users report pages taking 5-10 seconds to load, especially during peak hours (9 AM - 5 PM).
""")
print(f"Problem Solution: {problem_result}")🔄 Agent Coordination and Multi-Agent Systems
👥 Multi-Agent Collaboration
python
class MultiAgentSystem:
"""System for coordinating multiple specialized agents"""
def __init__(self):
self.agents = {
"research": ResearchAgent(),
"problem_solver": ProblemSolvingAgent(),
"general": MemoryEnabledAgent(basic_tools)
}
self.coordinator = self._create_coordinator()
def _create_coordinator(self):
"""Create agent coordinator"""
def route_to_agent(task: str) -> str:
"""Route task to appropriate agent"""
routing_prompt = f"""
Analyze this task and determine which specialized agent should handle it:
Task: {task}
Available agents:
- research: Best for research, analysis, and information gathering
- problem_solver: Best for complex problem-solving and systematic analysis
- general: Best for general tasks, calculations, and file operations
Choose the most appropriate agent and explain why.
Format: agent_name|explanation
"""
llm = ChatOpenAI(temperature=0.1)
response = llm.invoke(routing_prompt)
if '|' in response.content:
agent_name, explanation = response.content.split('|', 1)
return f"{agent_name.strip()}|{explanation.strip()}"
else:
return f"general|{response.content}"
return Tool(
name="route_to_agent",
description="Route a task to the most appropriate specialized agent",
func=route_to_agent
)
def execute_task(self, task: str) -> Dict[str, Any]:
"""Execute task using appropriate agent"""
# Route to appropriate agent
routing_result = self.coordinator.func(task)
agent_name, explanation = routing_result.split('|', 1)
print(f"🎯 Routing to {agent_name} agent: {explanation}")
# Execute with selected agent
if agent_name in self.agents:
agent = self.agents[agent_name]
if agent_name == "research":
result = agent.research(task)
elif agent_name == "problem_solver":
result = agent.solve(task)
else: # general agent
result = agent.chat(task)
return {
"agent_used": agent_name,
"routing_explanation": explanation,
"result": result
}
else:
# Fallback to general agent
result = self.agents["general"].chat(task)
return {
"agent_used": "general",
"routing_explanation": "Fallback to general agent",
"result": result
}
def collaborative_task(self, tasks: List[str]) -> List[Dict[str, Any]]:
"""Execute multiple related tasks using different agents"""
results = []
for i, task in enumerate(tasks):
print(f"\n--- Task {i+1}/{len(tasks)} ---")
result = self.execute_task(task)
results.append(result)
# Add delay between tasks to avoid rate limits
import time
time.sleep(1)
return results
# Test multi-agent system
multi_agent_system = MultiAgentSystem()
print("👥 Multi-Agent System Created!")
# Test with different types of tasks
tasks = [
"Research the latest trends in artificial intelligence",
"How do I optimize database queries for better performance?",
"Calculate the compound interest on $10,000 invested at 5% annually for 10 years"
]
print("\n=== Multi-Agent Task Execution ===")
results = multi_agent_system.collaborative_task(tasks)
for i, result in enumerate(results):
print(f"\nTask {i+1} Result:")
print(f"Agent Used: {result['agent_used']}")
print(f"Routing Explanation: {result['routing_explanation']}")
print(f"Result: {result['result'][:200]}...") # Truncate for readability🔧 Advanced Agent Patterns
🎯 Goal-Oriented Agent
python
class GoalOrientedAgent:
"""Agent that can plan and execute towards specific goals"""
def __init__(self):
self.llm = ChatOpenAI(temperature=0.1)
self.current_goal = None
self.plan = []
self.progress = []
self.tools = basic_tools + self._create_planning_tools()
def _create_planning_tools(self) -> List[Tool]:
"""Create tools for planning and goal management"""
def create_plan(goal: str) -> str:
"""Create a step-by-step plan to achieve a goal"""
planning_prompt = f"""
Create a detailed plan to achieve this goal: {goal}
Provide:
1. Clear objective statement
2. Success criteria
3. Step-by-step action plan
4. Potential obstacles and mitigation strategies
5. Timeline estimation
6. Required resources
Plan:"""
response = self.llm.invoke(planning_prompt)
return response.content
def track_progress(current_step: str, completed_steps: str) -> str:
"""Track progress towards goal completion"""
progress_prompt = f"""
Analyze progress towards goal completion:
Current Step: {current_step}
Completed Steps: {completed_steps}
Assess:
1. Percentage completion
2. Next immediate actions
3. Any adjustments needed to plan
4. Estimated time to completion
Progress Analysis:"""
response = self.llm.invoke(progress_prompt)
return response.content
return [
Tool(
name="create_plan",
description="Create a detailed plan to achieve a specific goal. Input: goal description",
func=create_plan
),
Tool(
name="track_progress",
description="Track progress towards goal. Input: 'current_step|completed_steps' separated by |",
func=lambda input_str: track_progress(*input_str.split('|', 1)) if '|' in input_str else "Error: Need current_step|completed_steps format"
)
]
def set_goal(self, goal: str) -> str:
"""Set a new goal and create initial plan"""
self.current_goal = goal
self.progress = []
# Create initial plan
agent = create_openai_functions_agent(
llm=self.llm,
tools=self.tools,
prompt=ChatPromptTemplate.from_messages([
("system", "You are a goal-oriented planning agent. Create detailed plans and track progress."),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
)
agent_executor = AgentExecutor(
agent=agent,
tools=self.tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
response = agent_executor.invoke({
"input": f"Create a comprehensive plan to achieve this goal: {goal}"
})
self.plan = response["output"]
return f"Goal set: {goal}\n\nInitial Plan:\n{self.plan}"
def execute_next_step(self) -> str:
"""Execute the next step towards the goal"""
if not self.current_goal:
return "No goal set. Please set a goal first."
progress_summary = "\n".join(self.progress) if self.progress else "No steps completed yet"
agent = create_openai_functions_agent(
llm=self.llm,
tools=self.tools,
prompt=ChatPromptTemplate.from_messages([
("system", f"""You are working towards this goal: {self.current_goal}
Current plan: {self.plan}
Progress so far: {progress_summary}
Execute the next logical step towards achieving the goal."""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
)
agent_executor = AgentExecutor(
agent=agent,
tools=self.tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
response = agent_executor.invoke({
"input": "What is the next step I should take, and execute it if possible?"
})
# Record progress
self.progress.append(f"Step {len(self.progress) + 1}: {response['output']}")
return response["output"]
def get_status(self) -> Dict[str, Any]:
"""Get current goal status"""
return {
"current_goal": self.current_goal,
"plan": self.plan,
"steps_completed": len(self.progress),
"progress": self.progress
}
# Test goal-oriented agent
goal_agent = GoalOrientedAgent()
print("🎯 Goal-Oriented Agent Created!")
# Set a goal
goal_result = goal_agent.set_goal(
"Learn the basics of machine learning and create a simple project"
)
print(f"Goal Setting Result:\n{goal_result}")
# Execute a few steps
print("\n=== Executing Steps ===")
for i in range(3):
print(f"\n--- Step {i+1} ---")
step_result = goal_agent.execute_next_step()
print(f"Step Result: {step_result}")
# Check status
status = goal_agent.get_status()
print(f"\n=== Current Status ===")
print(f"Goal: {status['current_goal']}")
print(f"Steps Completed: {status['steps_completed']}")🔄 Error Handling and Recovery
🛡️ Robust Agent Implementation
python
class RobustAgent:
"""Agent with comprehensive error handling and recovery"""
def __init__(self, tools: List[Tool], max_retries: int = 3):
self.llm = ChatOpenAI(temperature=0.1)
self.tools = tools
self.max_retries = max_retries
self.error_log = []
self.agent_executor = self._create_robust_agent()
def _create_robust_agent(self) -> AgentExecutor:
"""Create agent with error handling"""
robust_prompt = ChatPromptTemplate.from_messages([
("system", """You are a robust AI assistant with error handling capabilities.
When encountering errors:
1. Don't panic - analyze what went wrong
2. Try alternative approaches
3. Use different tools if the first choice fails
4. Provide partial results if complete solution isn't possible
5. Explain any limitations or failures clearly
Your goal is to be as helpful as possible even when things don't go perfectly."""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_functions_agent(
llm=self.llm,
tools=self.tools,
prompt=robust_prompt
)
return AgentExecutor(
agent=agent,
tools=self.tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=8,
early_stopping_method="generate" # Stop early if possible
)
def execute_with_recovery(self, task: str) -> Dict[str, Any]:
"""Execute task with automatic error recovery"""
for attempt in range(self.max_retries):
try:
print(f"🔄 Attempt {attempt + 1}/{self.max_retries}")
response = self.agent_executor.invoke({"input": task})
# Success
return {
"success": True,
"result": response["output"],
"attempts": attempt + 1,
"errors": []
}
except Exception as e:
error_msg = str(e)
self.error_log.append({
"attempt": attempt + 1,
"error": error_msg,
"task": task,
"timestamp": datetime.now()
})
print(f"❌ Attempt {attempt + 1} failed: {error_msg}")
if attempt < self.max_retries - 1:
print("🔄 Retrying with modified approach...")
# Modify task slightly for retry
task = f"Please try again with a different approach: {task}"
else:
# Final failure
return {
"success": False,
"result": f"Failed after {self.max_retries} attempts. Last error: {error_msg}",
"attempts": self.max_retries,
"errors": [log["error"] for log in self.error_log if log["task"] == task]
}
def get_error_summary(self) -> Dict[str, Any]:
"""Get summary of errors encountered"""
if not self.error_log:
return {"total_errors": 0, "error_patterns": []}
# Analyze error patterns
error_types = {}
for log in self.error_log:
error_type = log["error"].split(":")[0] if ":" in log["error"] else "Unknown"
error_types[error_type] = error_types.get(error_type, 0) + 1
return {
"total_errors": len(self.error_log),
"error_patterns": error_types,
"recent_errors": self.error_log[-5:] # Last 5 errors
}
# Test robust agent
robust_agent = RobustAgent(basic_tools)
print("🛡️ Robust Agent Created!")
# Test with potentially problematic tasks
test_tasks = [
"Calculate the square root of -1", # Math error
"Search for information about 'nonexistent topic xyz123'", # No results
"Read a file called 'definitely_not_exists.txt'", # File error
"What is 15 * 23?" # Should work fine
]
print("\n=== Testing Error Recovery ===")
for task in test_tasks:
print(f"\n--- Testing: {task} ---")
result = robust_agent.execute_with_recovery(task)
if result["success"]:
print(f"✅ Success after {result['attempts']} attempts")
print(f"Result: {result['result'][:200]}...")
else:
print(f"❌ Failed after {result['attempts']} attempts")
print(f"Errors: {result['errors']}")
# Get error summary
error_summary = robust_agent.get_error_summary()
print(f"\n=== Error Summary ===")
print(f"Total errors: {error_summary['total_errors']}")
print(f"Error patterns: {error_summary['error_patterns']}")🔗 Integration and Production Considerations
🚀 Production-Ready Agent System
python
class ProductionAgentSystem:
"""Production-ready agent system with monitoring and logging"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.setup_logging()
self.setup_monitoring()
self.agents = self._initialize_agents()
def setup_logging(self):
"""Setup comprehensive logging"""
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('agent_system.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger('AgentSystem')
def setup_monitoring(self):
"""Setup monitoring and metrics"""
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"average_response_time": 0,
"tool_usage_count": {},
"error_types": {}
}
def _initialize_agents(self) -> Dict[str, Any]:
"""Initialize all available agents"""
agents = {}
# Basic conversational agent
agents["conversational"] = MemoryEnabledAgent(
basic_tools,
memory_window=self.config.get("memory_window", 10)
)
# Research agent
agents["research"] = ResearchAgent()
# Problem-solving agent
agents["problem_solver"] = ProblemSolvingAgent()
# Goal-oriented agent
agents["goal_oriented"] = GoalOrientedAgent()
self.logger.info(f"Initialized {len(agents)} agents")
return agents
def process_request(self, user_id: str, message: str, agent_type: str = "conversational") -> Dict[str, Any]:
"""Process user request with full monitoring"""
start_time = time.time()
request_id = f"{user_id}_{int(time.time())}"
self.logger.info(f"Processing request {request_id} from user {user_id}")
try:
# Update metrics
self.metrics["total_requests"] += 1
# Route to appropriate agent
if agent_type not in self.agents:
agent_type = "conversational" # Fallback
agent = self.agents[agent_type]
# Process request based on agent type
if agent_type == "research":
result = agent.research(message)
elif agent_type == "problem_solver":
result = agent.solve(message)
elif agent_type == "goal_oriented":
if "set goal:" in message.lower():
goal = message.replace("set goal:", "").strip()
result = agent.set_goal(goal)
else:
result = agent.execute_next_step()
else: # conversational
result = agent.chat(message)
# Calculate metrics
response_time = time.time() - start_time
self._update_metrics(response_time, success=True)
self.logger.info(f"Successfully processed request {request_id}")
return {
"success": True,
"result": result,
"response_time": response_time,
"agent_used": agent_type,
"request_id": request_id
}
except Exception as e:
# Handle errors
response_time = time.time() - start_time
error_type = type(e).__name__
self._update_metrics(response_time, success=False, error_type=error_type)
self.logger.error(f"Failed to process request {request_id}: {str(e)}")
return {
"success": False,
"error": str(e),
"error_type": error_type,
"response_time": response_time,
"agent_used": agent_type,
"request_id": request_id
}
def _update_metrics(self, response_time: float, success: bool, error_type: str = None):
"""Update system metrics"""
if success:
self.metrics["successful_requests"] += 1
else:
self.metrics["failed_requests"] += 1
if error_type:
self.metrics["error_types"][error_type] = (
self.metrics["error_types"].get(error_type, 0) + 1
)
# Update average response time
total_requests = self.metrics["total_requests"]
current_avg = self.metrics["average_response_time"]
self.metrics["average_response_time"] = (
(current_avg * (total_requests - 1) + response_time) / total_requests
)
def get_system_status(self) -> Dict[str, Any]:
"""Get comprehensive system status"""
success_rate = (
self.metrics["successful_requests"] / self.metrics["total_requests"]
if self.metrics["total_requests"] > 0 else 0
)
return {
"status": "healthy" if success_rate > 0.9 else "degraded" if success_rate > 0.7 else "unhealthy",
"metrics": self.metrics,
"success_rate": success_rate,
"available_agents": list(self.agents.keys()),
"uptime": "System running" # In practice, track actual uptime
}
# Example production configuration
production_config = {
"memory_window": 15,
"max_retries": 3,
"timeout_seconds": 30,
"rate_limit_per_user": 100
}
# Initialize production system
production_system = ProductionAgentSystem(production_config)
print("🚀 Production Agent System Initialized!")
# Test the system
test_requests = [
("user123", "Calculate 25 * 17 + 100", "conversational"),
("user456", "Research the latest AI trends", "research"),
("user789", "Help me solve a database performance issue", "problem_solver"),
]
print("\n=== Testing Production System ===")
for user_id, message, agent_type in test_requests:
result = production_system.process_request(user_id, message, agent_type)
print(f"\nUser: {user_id}")
print(f"Message: {message}")
print(f"Agent: {agent_type}")
print(f"Success: {result['success']}")
print(f"Response Time: {result['response_time']:.2f}s")
if result['success']:
print(f"Result: {result['result'][:200]}...")
else:
print(f"Error: {result['error']}")
# Get system status
status = production_system.get_system_status()
print(f"\n=== System Status ===")
print(f"Status: {status['status']}")
print(f"Success Rate: {status['success_rate']:.2%}")
print(f"Total Requests: {status['metrics']['total_requests']}")
print(f"Average Response Time: {status['metrics']['average_response_time']:.2f}s")🔗 Next Steps
Ready to expand your agent capabilities? Continue with:
- Tools - Build custom tools and integrate external services
- Memory Basics - Advanced memory patterns for agents
- Callbacks and Monitoring - Monitor and debug agent behavior
Key Agent Takeaways:
- Agents are autonomous - They can plan, execute, and adapt independently
- Tools extend capabilities - Agents become powerful through tool integration
- Memory enables context - Conversation history improves agent responses
- Specialization improves performance - Different agents for different tasks
- Error handling is crucial - Robust agents recover gracefully from failures
- Production requires monitoring - Track performance and behavior in real systems
- Multi-agent systems enable complex problem-solving through collaboration