Introduction
In this tutorial, we'll explore the core components of nanobot, a lightweight personal AI agent framework developed by HKUDS. Rather than simply using it as a black box, we'll manually recreate its essential subsystems including tool integration, memory management, skills, subagents, and cron scheduling. This hands-on approach will give you a deep understanding of how AI agents are structured and function, which is crucial for building custom AI solutions.
Prerequisites
- Basic understanding of Python programming
- Python 3.8 or higher installed
- Familiarity with AI concepts like LLMs and agent architectures
- Install required packages:
openai,schedule,sqlite3,json,datetime
Step-by-step Instructions
1. Setting Up the Basic Agent Structure
1.1 Create the Core Agent Class
We'll start by creating the foundational agent class that will manage all subsystems.
import openai
import json
import sqlite3
import datetime
class NanobotAgent:
def __init__(self, api_key):
openai.api_key = api_key
self.memory = []
self.tools = {}
self.skills = []
self.subagents = {}
self.schedule = []
def add_tool(self, name, tool_func):
self.tools[name] = tool_func
def add_memory(self, memory_item):
self.memory.append(memory_item)
def execute_tool(self, tool_name, **kwargs):
if tool_name in self.tools:
return self.tools[tool_name](**kwargs)
else:
raise ValueError(f"Tool {tool_name} not found")
Why this step? This creates the foundation of our agent. The class encapsulates all subsystems and provides methods to manage them. We're setting up the basic structure that will hold our tools, memory, and other components.
1.2 Initialize the Database for Memory Storage
Next, we'll set up SQLite for persistent memory storage.
def init_memory_db(self):
conn = sqlite3.connect('agent_memory.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()
conn.close()
# Add this method to the class
self.init_memory_db = init_memory_dbWhy this step? Persistent memory is crucial for agent longevity. Using SQLite ensures that our agent can remember past interactions across restarts, which is essential for learning and context retention.
2. Implementing Tool Integration
2.1 Create a Sample Tool Function
Let's implement a simple tool for retrieving weather information.
import requests
def get_weather(city):
# This is a mock implementation
# In practice, you'd use a real weather API
return f"Weather in {city}: Sunny, 22°C"
# Register the tool
agent.add_tool('weather', get_weather)Why this step? Tools are the agent's interface to external systems. They allow the agent to perform actions beyond pure reasoning, like fetching data or executing commands.
2.2 Add Tool Execution Logic
We'll enhance our agent to handle tool execution with error handling.
def execute_tool_safe(self, tool_name, **kwargs):
try:
result = self.execute_tool(tool_name, **kwargs)
self.add_memory(f"Executed tool {tool_name} with result: {result}")
return result
except Exception as e:
error_msg = f"Error executing tool {tool_name}: {str(e)}"
self.add_memory(error_msg)
return error_msgWhy this step? Real-world tools often fail. Adding error handling ensures our agent can gracefully manage failures and log them for debugging.
3. Building Memory Management
3.1 Implement Memory Persistence
Enhance the memory system to save and load from our database.
def save_memory(self, content):
conn = sqlite3.connect('agent_memory.db')
cursor = conn.cursor()
cursor.execute("INSERT INTO memories (content) VALUES (?)", (content,))
conn.commit()
conn.close()
def load_memory(self):
conn = sqlite3.connect('agent_memory.db')
cursor = conn.cursor()
cursor.execute("SELECT content FROM memories ORDER BY timestamp")
memories = cursor.fetchall()
conn.close()
return [mem[0] for mem in memories]Why this step? Memory persistence allows our agent to maintain context over time. This is crucial for agents that need to remember past interactions or learn from experience.
3.2 Add Memory Retrieval Logic
Implement a method to retrieve recent memories for context.
def get_recent_memories(self, count=5):
memories = self.load_memory()
return memories[-count:] if memories else []Why this step? Agents need to access recent context to make informed decisions. This method allows the agent to pull in relevant memories when processing new tasks.
4. Creating Skills and Subagents
4.1 Define a Skill System
Skills are higher-level capabilities built from tools and memory.
class Skill:
def __init__(self, name, description, function):
self.name = name
self.description = description
self.function = function
def execute(self, agent, **kwargs):
return self.function(agent, **kwargs)
# Add to agent class
self.skills = []
def add_skill(self, skill):
self.skills.append(skill)Why this step? Skills abstract complex operations into reusable components. This modular approach makes our agent more maintainable and extensible.
4.2 Create a Subagent Structure
Subagents are specialized agents that can be orchestrated by the main agent.
def create_subagent(self, name, agent_class, **kwargs):
subagent = agent_class(**kwargs)
self.subagents[name] = subagent
return subagent
def delegate_to_subagent(self, subagent_name, task):
if subagent_name in self.subagents:
return self.subagents[subagent_name].process(task)
else:
raise ValueError(f"Subagent {subagent_name} not found")Why this step? Subagents enable complex problem-solving by breaking tasks into specialized components. This hierarchical structure allows for more sophisticated agent behavior.
5. Implementing Cron Scheduling
5.1 Add Scheduling Framework
Integrate the schedule library to enable time-based task execution.
import schedule
import threading
import time
self.scheduler = schedule
self.running = False
def start_scheduler(self):
self.running = True
def run_scheduler():
while self.running:
self.scheduler.run_pending()
time.sleep(1)
thread = threading.Thread(target=run_scheduler)
thread.daemon = True
thread.start()
def add_scheduled_task(self, interval, task_func):
self.scheduler.every(interval).seconds.do(task_func)Why this step? Scheduled tasks allow agents to perform routine operations automatically. This is essential for agents that need to monitor systems, fetch updates, or perform maintenance.
5.2 Create a Sample Scheduled Task
def check_weather_daily():
print("Checking weather for today")
# Implementation would use our weather tool
# Register the scheduled task
agent.add_scheduled_task(86400, check_weather_daily) # Run dailyWhy this step? Regular tasks ensure our agent remains active and responsive. Daily weather checks, system monitoring, and data updates are common examples of scheduled operations.
6. Putting It All Together
6.1 Create a Complete Agent Workflow
Let's build a complete example that demonstrates all components working together.
# Initialize the agent
agent = NanobotAgent(api_key="your-api-key")
# Add tools
agent.add_tool('weather', get_weather)
# Add skills
skill = Skill("weather_check", "Check weather for a city", lambda a, city: a.execute_tool_safe('weather', city=city))
agent.add_skill(skill)
# Start scheduler
agent.start_scheduler()
# Add a scheduled task
agent.add_scheduled_task(3600, lambda: agent.add_memory("Hourly check"))
# Process a task
result = agent.execute_tool_safe('weather', city='London')
print(result)
Why this step? This integration demonstrates how all components work together. The agent can execute tools, remember results, schedule tasks, and maintain context through memory.
Summary
In this tutorial, we've manually reconstructed the core subsystems of the nanobot framework. We've built a foundation that includes tool integration, persistent memory management, skill definition, subagent orchestration, and cron scheduling. This hands-on approach gives you a deep understanding of how AI agents are structured and how to extend their capabilities. While this is a simplified version of nanobot, it demonstrates the key architectural decisions and patterns used in real-world AI agent frameworks.
