Single agent AI systems represent the foundational building block of autonomous AI. While multi-agent systems get a lot of attention, understanding how to build robust, capable single agents is crucial for any AI engineer. Let's dive deep into the architecture, patterns, and best practices for creating truly autonomous AI agents.
What is a Single Agent AI System?
A single agent AI system is an autonomous entity that can:
- Perceive: Understand its environment and context
- Reason: Make decisions based on available information
- Act: Execute actions through tools and APIs
- Learn: Improve performance over time
Unlike simple chatbots that just respond to inputs, autonomous agents have agency - they can plan, execute multi-step tasks, and adapt to changing situations.
Core Architecture Components
1. The Reasoning Engine
The brain of your agent, typically powered by an LLM:
from anthropic import Anthropic
from openai import OpenAI
class AgentReasoning:
def __init__(self, provider: str = "anthropic"):
if provider == "anthropic":
self.client = Anthropic()
self.model = "claude-3-5-sonnet-20241022"
else:
self.client = OpenAI()
self.model = "gpt-4-turbo-preview"
def reason(self, context: str, task: str) -> dict:
"""Core reasoning function"""
prompt = f"""
Context: {context}
Task: {task}
Think step by step:
1. What is the goal?
2. What information do I have?
3. What information do I need?
4. What actions should I take?
5. What is the expected outcome?
Provide your reasoning and action plan.
"""
response = self.client.messages.create(
model=self.model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}]
)
return self.parse_reasoning(response.content[0].text)
def parse_reasoning(self, text: str) -> dict:
"""Extract structured reasoning from response"""
# Parse the reasoning into actionable steps
return {
"goal": self.extract_goal(text),
"analysis": self.extract_analysis(text),
"actions": self.extract_actions(text),
"expected_outcome": self.extract_outcome(text)
}
2. Tool Integration System
Enable your agent to interact with the world:
from typing import Callable, Dict, Any
import inspect
class ToolRegistry:
def __init__(self):
self.tools: Dict[str, Callable] = {}
self.tool_descriptions: Dict[str, str] = {}
def register(self, name: str, description: str):
"""Decorator to register tools"""
def decorator(func: Callable):
self.tools[name] = func
self.tool_descriptions[name] = description
# Get function signature for schema
sig = inspect.signature(func)
params = {
name: {
"type": param.annotation.__name__,
"required": param.default == inspect.Parameter.empty
}
for name, param in sig.parameters.items()
}
return func
return decorator
def get_tool_schema(self) -> list:
"""Generate tool schema for LLM"""
return [
{
"name": name,
"description": desc,
"input_schema": self.get_tool_params(name)
}
for name, desc in self.tool_descriptions.items()
]
def execute_tool(self, tool_name: str, **kwargs) -> Any:
"""Execute a registered tool"""
if tool_name not in self.tools:
raise ValueError(f"Tool {tool_name} not found")
try:
return self.tools[tool_name](**kwargs)
except Exception as e:
return f"Error executing {tool_name}: {str(e)}"
# Example tools
tools = ToolRegistry()
@tools.register("web_search", "Search the web for information")
def web_search(query: str) -> str:
"""Search the web and return results"""
# Implement web search
return f"Search results for: {query}"
@tools.register("calculate", "Perform mathematical calculations")
def calculate(expression: str) -> float:
"""Safely evaluate mathematical expressions"""
try:
# Use safe evaluation
return eval(expression, {"__builtins__": {}})
except:
return "Invalid expression"
@tools.register("read_file", "Read contents of a file")
def read_file(filepath: str) -> str:
"""Read and return file contents"""
with open(filepath, 'r') as f:
return f.read()
@tools.register("write_file", "Write content to a file")
def write_file(filepath: str, content: str) -> str:
"""Write content to a file"""
with open(filepath, 'w') as f:
f.write(content)
return f"Successfully wrote to {filepath}"
3. Action Planning and Execution
The agent's ability to plan and execute multi-step tasks:
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional
class ActionStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Action:
tool_name: str
parameters: dict
status: ActionStatus = ActionStatus.PENDING
result: Optional[Any] = None
error: Optional[str] = None
class ActionPlanner:
def __init__(self, reasoning_engine: AgentReasoning, tools: ToolRegistry):
self.reasoning = reasoning_engine
self.tools = tools
def create_plan(self, task: str, context: str) -> List[Action]:
"""Generate action plan for a task"""
reasoning_result = self.reasoning.reason(context, task)
# Convert reasoning into concrete actions
plan = []
for action_desc in reasoning_result["actions"]:
action = self.parse_action(action_desc)
if action:
plan.append(action)
return plan
def parse_action(self, action_desc: str) -> Optional[Action]:
"""Parse action description into Action object"""
# Use LLM to extract structured action
prompt = f"""
Parse this action description into a tool call:
{action_desc}
Available tools: {list(self.tools.tools.keys())}
Return JSON with:
{{
"tool_name": "tool_name",
"parameters": {{"param1": "value1"}}
}}
"""
# Get structured output from LLM
# Implementation depends on your LLM provider
# For now, simplified:
return Action(
tool_name="example_tool",
parameters={}
)
def execute_plan(self, plan: List[Action]) -> List[Action]:
"""Execute action plan"""
for action in plan:
action.status = ActionStatus.IN_PROGRESS
try:
result = self.tools.execute_tool(
action.tool_name,
**action.parameters
)
action.result = result
action.status = ActionStatus.COMPLETED
except Exception as e:
action.error = str(e)
action.status = ActionStatus.FAILED
break # Stop on failure
return plan
4. Memory and State Management
Enable your agent to maintain context:
from datetime import datetime
from collections import deque
class AgentMemory:
def __init__(self, max_short_term: int = 10):
self.short_term = deque(maxlen=max_short_term)
self.long_term = []
self.working_context = {}
def add_to_short_term(self, item: dict):
"""Add to short-term memory"""
item["timestamp"] = datetime.now()
self.short_term.append(item)
def add_to_long_term(self, item: dict):
"""Add to long-term memory"""
item["timestamp"] = datetime.now()
self.long_term.append(item)
def update_context(self, key: str, value: Any):
"""Update working context"""
self.working_context[key] = value
def get_context_summary(self) -> str:
"""Generate context summary for the agent"""
summary = []
# Recent interactions
if self.short_term:
summary.append("Recent interactions:")
for item in list(self.short_term)[-5:]:
summary.append(f"- {item.get('type', 'interaction')}: {item.get('content', '')}")
# Working context
if self.working_context:
summary.append("\nCurrent context:")
for key, value in self.working_context.items():
summary.append(f"- {key}: {value}")
return "\n".join(summary)
def search_memory(self, query: str, limit: int = 5) -> List[dict]:
"""Search through memories"""
# Simple keyword search (in production, use vector search)
results = []
for memory in reversed(self.long_term):
content = str(memory.get("content", ""))
if query.lower() in content.lower():
results.append(memory)
if len(results) >= limit:
break
return results
Building a Complete Agent
Now let's put it all together:
class AutonomousAgent:
def __init__(self, name: str, role: str, goal: str):
self.name = name
self.role = role
self.goal = goal
# Initialize components
self.reasoning = AgentReasoning()
self.tools = ToolRegistry()
self.planner = ActionPlanner(self.reasoning, self.tools)
self.memory = AgentMemory()
# Agent state
self.current_task = None
self.task_history = []
def process_task(self, task: str) -> str:
"""Main task processing loop"""
self.current_task = task
# Add task to memory
self.memory.add_to_short_term({
"type": "task",
"content": task
})
# Get context
context = self.build_context()
# Create action plan
plan = self.planner.create_plan(task, context)
# Execute plan
executed_plan = self.planner.execute_plan(plan)
# Analyze results
result = self.analyze_results(executed_plan)
# Update memory
self.memory.add_to_long_term({
"type": "completed_task",
"task": task,
"plan": executed_plan,
"result": result
})
self.task_history.append({
"task": task,
"result": result,
"timestamp": datetime.now()
})
return result
def build_context(self) -> str:
"""Build context for reasoning"""
context_parts = [
f"I am {self.name}, a {self.role}.",
f"My goal is: {self.goal}",
"\n" + self.memory.get_context_summary(),
f"\nAvailable tools: {list(self.tools.tools.keys())}"
]
if self.task_history:
recent = self.task_history[-3:]
context_parts.append("\nRecent tasks:")
for item in recent:
context_parts.append(f"- {item['task']}: {item['result']}")
return "\n".join(context_parts)
def analyze_results(self, plan: List[Action]) -> str:
"""Analyze execution results and generate summary"""
successful = [a for a in plan if a.status == ActionStatus.COMPLETED]
failed = [a for a in plan if a.status == ActionStatus.FAILED]
if not plan:
return "No actions were planned"
if failed:
return f"Task partially completed. {len(successful)}/{len(plan)} actions succeeded. Failures: {[a.error for a in failed]}"
# Generate summary using LLM
results_text = "\n".join([
f"{a.tool_name}: {a.result}" for a in successful
])
summary_prompt = f"""
Summarize the results of these actions for the task: {self.current_task}
Actions taken:
{results_text}
Provide a concise summary of what was accomplished.
"""
response = self.reasoning.client.messages.create(
model=self.reasoning.model,
max_tokens=500,
messages=[{"role": "user", "content": summary_prompt}]
)
return response.content[0].text
def reflect(self) -> str:
"""Agent reflects on its performance"""
if not self.task_history:
return "No tasks completed yet"
reflection_prompt = f"""
Review these recent tasks and reflect on performance:
{self.task_history[-5:]}
Consider:
1. What worked well?
2. What could be improved?
3. Are there patterns in successes/failures?
4. What should I do differently next time?
Provide insights and recommendations.
"""
response = self.reasoning.client.messages.create(
model=self.reasoning.model,
max_tokens=1000,
messages=[{"role": "user", "content": reflection_prompt}]
)
return response.content[0].text
Advanced Agent Patterns
1. ReAct (Reasoning + Acting)
Interleave reasoning and acting for better decision-making:
class ReActAgent(AutonomousAgent):
def react_loop(self, task: str, max_iterations: int = 5):
"""ReAct loop: Reason, Act, Observe, Repeat"""
context = self.build_context()
observations = []
for i in range(max_iterations):
# Thought: Reason about next action
thought = self.reason_next_step(task, context, observations)
# Action: Execute the action
action_result = self.execute_action(thought["action"])
# Observation: Record result
observation = {
"iteration": i + 1,
"thought": thought["reasoning"],
"action": thought["action"],
"result": action_result
}
observations.append(observation)
# Check if task is complete
if self.is_task_complete(task, observations):
break
return self.synthesize_result(observations)
def reason_next_step(self, task: str, context: str, observations: list) -> dict:
"""Reason about the next step"""
prompt = f"""
Task: {task}
Context: {context}
Previous observations:
{observations}
Think: What should I do next to complete this task?
Provide your reasoning and the next action to take.
"""
# Get reasoning from LLM
# Return structured thought and action
pass
2. Chain-of-Thought Reasoning
Enable deeper reasoning with step-by-step thinking:
class ChainOfThoughtAgent(AutonomousAgent):
def solve_with_cot(self, problem: str) -> str:
"""Solve problem using chain-of-thought"""
cot_prompt = f"""
Problem: {problem}
Let's solve this step by step:
Step 1: Understand the problem
- What is being asked?
- What information do I have?
Step 2: Break down the solution
- What are the sub-problems?
- What's the approach?
Step 3: Execute the solution
- Solve each sub-problem
- Combine results
Step 4: Verify the answer
- Does it make sense?
- Did I answer the question?
Think through each step carefully.
"""
response = self.reasoning.client.messages.create(
model=self.reasoning.model,
max_tokens=2048,
messages=[{"role": "user", "content": cot_prompt}]
)
return response.content[0].text
3. Self-Correction and Validation
Build agents that can verify and correct their work:
class SelfCorrectingAgent(AutonomousAgent):
def execute_with_validation(self, task: str) -> str:
"""Execute task with self-validation"""
# Initial attempt
result = self.process_task(task)
# Validate result
validation = self.validate_result(task, result)
if not validation["valid"]:
# Attempt correction
corrected = self.correct_result(
task,
result,
validation["issues"]
)
return corrected
return result
def validate_result(self, task: str, result: str) -> dict:
"""Validate if result satisfies task requirements"""
validation_prompt = f"""
Task: {task}
Result: {result}
Validate if the result properly addresses the task:
1. Is it complete?
2. Is it accurate?
3. Does it address all requirements?
Return validation status and any issues found.
"""
# Get validation from LLM
# Return structured validation result
pass
def correct_result(self, task: str, result: str, issues: list) -> str:
"""Correct result based on identified issues"""
correction_prompt = f"""
Original task: {task}
Previous result: {result}
Issues identified: {issues}
Provide a corrected result that addresses these issues.
"""
# Get corrected result from LLM
pass
Best Practices
1. Clear Agent Identity
Define clear roles and capabilities:
agent = AutonomousAgent(
name="DataAnalyst",
role="Senior Data Analyst specialized in financial data",
goal="Analyze data and provide actionable insights"
)
2. Robust Error Handling
def safe_execute(self, action: Action) -> Any:
"""Execute action with error handling"""
try:
result = self.tools.execute_tool(
action.tool_name,
**action.parameters
)
return result
except Exception as e:
# Log error
logger.error(f"Action failed: {action.tool_name}, Error: {e}")
# Attempt recovery
recovery_action = self.plan_recovery(action, e)
if recovery_action:
return self.safe_execute(recovery_action)
raise
3. Monitoring and Observability
class ObservableAgent(AutonomousAgent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.metrics = {
"tasks_completed": 0,
"tasks_failed": 0,
"total_actions": 0,
"average_task_time": 0
}
def process_task(self, task: str) -> str:
start_time = time.time()
try:
result = super().process_task(task)
self.metrics["tasks_completed"] += 1
return result
except Exception as e:
self.metrics["tasks_failed"] += 1
raise
finally:
execution_time = time.time() - start_time
self.update_metrics(execution_time)
def get_metrics(self) -> dict:
return self.metrics
Real-World Applications
Personal Assistant Agent
personal_assistant = AutonomousAgent(
name="PersonalAssistant",
role="Personal productivity assistant",
goal="Help manage tasks, schedule, and information"
)
# Register relevant tools
@personal_assistant.tools.register("schedule_meeting", "Schedule a meeting")
def schedule_meeting(title: str, time: str, attendees: list):
# Integration with calendar API
pass
@personal_assistant.tools.register("send_email", "Send an email")
def send_email(to: str, subject: str, body: str):
# Integration with email API
pass
result = personal_assistant.process_task(
"Schedule a meeting with the engineering team for tomorrow at 2 PM to discuss the new feature"
)
Code Assistant Agent
code_assistant = AutonomousAgent(
name="CodeAssistant",
role="Senior software engineer",
goal="Help with code review, debugging, and development"
)
# Code-specific tools
@code_assistant.tools.register("analyze_code", "Analyze code for issues")
def analyze_code(filepath: str):
# Static analysis
pass
@code_assistant.tools.register("run_tests", "Execute test suite")
def run_tests(test_path: str):
# Run tests
pass
Research Agent
research_assistant = AutonomousAgent(
name="ResearchAssistant",
role="Research analyst",
goal="Conduct thorough research and provide comprehensive reports"
)
# Research tools
@research_assistant.tools.register("web_search", "Search the web")
def web_search(query: str):
pass
@research_assistant.tools.register("summarize_paper", "Summarize research paper")
def summarize_paper(url: str):
pass
Conclusion
Single agent AI systems are powerful building blocks for autonomous AI applications. By combining reasoning, planning, tool use, and memory, you can create agents that:
- Solve complex, multi-step problems
- Learn and improve over time
- Interact with external systems
- Make autonomous decisions
The key to success is:
- Clear design: Define roles, goals, and capabilities
- Robust architecture: Implement proper error handling and validation
- Effective tools: Provide the right capabilities for the task
- Good memory: Maintain relevant context
- Continuous improvement: Monitor, reflect, and optimize
Start with simple agents and gradually add complexity. Test thoroughly, monitor performance, and iterate based on real-world usage.
Build your first autonomous agent today. Start with a clear goal, add essential tools, and let your agent learn and grow in capability.