Base Agent

The Base Agent is the foundation of all agents in AgoraAI. It provides core functionality and interfaces that all specialized agents inherit and build upon.

Overview

from agoraai.agent import Agent

agent = Agent(
    name="MyAgent",
    agent_type="custom",
    capabilities=["process", "analyze"]
)

Core Properties

Required Properties

• name: Unique identifier for the agent

• agent_type: Type classification of the agent

• capabilities: List of agent capabilities

Optional Properties

• metadata: Additional agent information

• state: Current agent state

• resource_limits: Resource usage constraints

Agent Lifecycle

stateDiagram-v2
    [*] --> Initialized : Create
    Initialized --> Ready : Register
    Ready --> Active : Start
    Active --> Paused : Pause
    Paused --> Active : Resume
    Active --> Stopped : Stop
    Stopped --> [*] : Destroy

Core Methods

Request Handling

@agent.on_request
async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]:
    """Handle incoming requests"""
    task_type = request.get("type")
    if task_type == "process":
        return await process_data(request.get("data"))
    return {"status": "error", "message": "Unknown task type"}

Task Execution

@agent.on_task
async def execute_task(task: Dict[str, Any]) -> Dict[str, Any]:
    """Execute specific tasks"""
    try:
        result = await perform_task(task)
        return {
            "status": "success",
            "result": result
        }
    except Exception as e:
        return {
            "status": "error",
            "error": str(e)
        }

State Management

# Update agent state
await agent.update_state({
    "status": "processing",
    "current_task": "data_analysis",
    "load": 0.75
})

# Get current state
state = agent.get_state()

Event System

Event Registration

@agent.on_event("startup")
async def handle_startup():
    """Handle agent startup"""
    await initialize_resources()

@agent.on_event("shutdown")
async def handle_shutdown():
    """Handle agent shutdown"""
    await cleanup_resources()

Custom Events

@agent.on_event("custom_event")
async def handle_custom_event(event_data):
    """Handle custom events"""
    await process_custom_event(event_data)

# Emit custom event
await agent.emit_event("custom_event", {
    "type": "notification",
    "data": "Something happened"
})

Resource Management

Setting Limits

agent.set_resource_limits({
    "max_memory": "2G",
    "max_cpu": 0.8,
    "max_tasks": 100,
    "timeout": 30
})

Monitoring Usage

# Get current resource usage
usage = agent.get_resource_usage()

# Monitor specific metrics
@agent.on_metric("memory_usage")
async def handle_memory_alert(usage):
    if usage > 0.9:  # 90% usage
        await agent.emit_event("resource_warning", {
            "type": "memory",
            "usage": usage
        })

Error Handling

Basic Error Handling

@agent.on