Introduction
In the rapidly evolving world of AI, autonomous agents are becoming more powerful and capable of performing complex tasks on our behalf. However, with great power comes great responsibility. This tutorial will guide you through creating a basic AI agent infrastructure that can be 'parented' using a control layer, similar to what Runta aims to achieve. You'll learn how to build a system that monitors and controls AI agent activities, ensuring they operate within defined boundaries.
Prerequisites
- Basic understanding of Python programming
- Intermediate knowledge of AI/ML concepts
- Python 3.8 or higher installed
- Required packages:
openai,langchain,pydantic,requests - OpenAI API key (for demonstration purposes)
Step-by-Step Instructions
Step 1: Set Up Your Development Environment
Install Required Packages
First, create a virtual environment and install the necessary packages:
python -m venv ai_agent_env
source ai_agent_env/bin/activate # On Windows: ai_agent_env\Scripts\activate
pip install openai langchain pydantic requestsThis creates an isolated environment to prevent package conflicts and ensures you have all the tools needed for building AI agent controls.
Step 2: Create the Base Agent Class
Define the Core Agent Structure
Create a file called agent.py to define your base agent:
from pydantic import BaseModel
from typing import List, Dict, Any
import openai
class AgentConfig(BaseModel):
name: str
capabilities: List[str]
max_budget: float = 100.0
allowed_domains: List[str] = []
class BaseAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.history = []
self.budget = config.max_budget
def execute_task(self, task: str) -> Dict[str, Any]:
# This is where the agent would perform its task
# For now, we'll simulate execution
result = {
"task": task,
"status": "completed",
"cost": 1.0,
"timestamp": "2023-01-01T00:00:00Z"
}
self.history.append(result)
return resultThis establishes the basic structure of an agent with configuration parameters and a task execution method. The configuration allows you to define what the agent can do and its limitations.
Step 3: Implement the Parent Control Layer
Create a Control System to Monitor Agents
Create a file called parent_control.py to implement the control system:
from agent import BaseAgent, AgentConfig
from typing import List, Dict, Any
import json
class ParentControl:
def __init__(self):
self.agents = {}
self.activity_log = []
def register_agent(self, agent: BaseAgent):
self.agents[agent.config.name] = agent
print(f"Agent {agent.config.name} registered")
def check_budget(self, agent_name: str, cost: float) -> bool:
agent = self.agents.get(agent_name)
if not agent:
return False
if agent.budget >= cost:
agent.budget -= cost
return True
else:
print(f"Budget exceeded for {agent_name}")
return False
def monitor_activity(self, agent_name: str, activity: Dict[str, Any]):
self.activity_log.append({
"agent": agent_name,
"activity": activity,
"timestamp": "2023-01-01T00:00:00Z"
})
# Log the activity for review
print(f"Activity logged for {agent_name}: {activity}")This control layer acts as the 'parent' that monitors agent activities and enforces budget limits, similar to how Runta aims to protect users from potentially harmful AI actions.
Step 4: Create a Task Executor with Safety Checks
Implement the Agent Controller
Create a file called agent_controller.py:
from parent_control import ParentControl
from agent import BaseAgent, AgentConfig
import time
class AgentController:
def __init__(self):
self.control = ParentControl()
def create_agent(self, name: str, capabilities: List[str], budget: float = 100.0) -> BaseAgent:
config = AgentConfig(name=name, capabilities=capabilities, max_budget=budget)
agent = BaseAgent(config)
self.control.register_agent(agent)
return agent
def execute_with_control(self, agent_name: str, task: str) -> Dict[str, Any]:
agent = self.control.agents.get(agent_name)
if not agent:
return {"error": "Agent not found"}
# Check if agent has capability for task
if not self._has_capability(agent, task):
return {"error": "Agent lacks required capability"}
# Execute task with budget check
result = agent.execute_task(task)
# Check budget and log activity
if self.control.check_budget(agent_name, result["cost"]):
self.control.monitor_activity(agent_name, result)
return result
else:
return {"error": "Budget exceeded"}
def _has_capability(self, agent: BaseAgent, task: str) -> bool:
# Simple capability check - in reality, this would be more sophisticated
for capability in agent.config.capabilities:
if capability.lower() in task.lower():
return True
return FalseThis controller ensures that agents only perform tasks within their capabilities and stay within budget limits, providing a safety net similar to what Runta is trying to build.
Step 5: Demonstrate the System
Run a Sample Implementation
Create a file called demo.py to test your implementation:
from agent_controller import AgentController
# Initialize the controller
controller = AgentController()
# Create agents with different capabilities and budgets
travel_agent = controller.create_agent(
name="TravelPlanner",
capabilities=["booking", "travel", "flights"],
budget=50.0
)
code_agent = controller.create_agent(
name="CodeAssistant",
capabilities=["coding", "programming", "development"],
budget=100.0
)
# Execute tasks with safety checks
result1 = controller.execute_with_control("TravelPlanner", "Book a flight from NYC to LA")
print("Travel result:", result1)
result2 = controller.execute_with_control("CodeAssistant", "Write a Python function to sort a list")
print("Code result:", result2)
# Try to exceed budget
result3 = controller.execute_with_control("TravelPlanner", "Book an expensive luxury hotel")
print("Budget exceeded result:", result3)This demonstration shows how the control system prevents agents from exceeding their budgets and ensures they only perform tasks within their defined capabilities.
Step 6: Enhance with Real AI Integration
Integrate with OpenAI API
Update your agent.py file to include real AI execution:
from pydantic import BaseModel
from typing import List, Dict, Any
import openai
# ... previous code ...
class AIEnabledAgent(BaseAgent):
def __init__(self, config: AgentConfig, api_key: str):
super().__init__(config)
openai.api_key = api_key
def execute_task(self, task: str) -> Dict[str, Any]:
# Simulate real AI execution
try:
response = openai.Completion.create(
engine="text-davinci-003",
prompt=f"{task}. Respond with a concise answer.",
max_tokens=100
)
result = {
"task": task,
"status": "completed",
"cost": 0.01,
"response": response.choices[0].text.strip(),
"timestamp": "2023-01-01T00:00:00Z"
}
except Exception as e:
result = {
"task": task,
"status": "failed",
"error": str(e),
"cost": 0.0
}
self.history.append(result)
return resultThis enhanced version demonstrates how real AI capabilities can be integrated while maintaining the control layer for safety and budget management.
Summary
In this tutorial, you've built a foundational AI agent control system similar to what Runta is developing. You've learned how to create agents with defined capabilities and budgets, implement a control layer that monitors their activities, and ensure they operate within safe parameters. This system provides the infrastructure needed to 'parent' AI agents, preventing them from performing unauthorized or excessive actions. The modular design allows for easy expansion with more sophisticated safety measures, activity logging, and integration with various AI services.
The key concepts you've learned include: agent configuration and registration, budget management, capability checking, activity monitoring, and real AI integration. These components work together to create a robust system that balances the power of AI agents with the need for control and safety.
