Agents

Understanding general AI agents in PandaAGI SDK

What are General AI Agents?

General AI Agents are the core building blocks of the PandaAGI SDK. They are autonomous AI entities that can understand natural language, make decisions, use tools, and interact with their environment to accomplish tasks. PandaAGI general AI agents consist of four key components:

• WebSocket Communication: Real-time bidirectional connection with the AGI backend server
• Tool Handlers: Pluggable modules that provide different capabilities and tool access
• Environment Integration: Isolated workspace for secure file and system operations
• Event Streaming: Real-time transmission of agent thoughts, actions, and results

Creating General AI Agents

Basic General AI Agent

The simplest way to create an agent:

import asyncio
from panda_agi import Agent

async def main():
    agent = Agent()
    
    response = await agent.run("What's the weather like today?")
    print(f"Task completed: {response.output}")

if __name__ == "__main__":
    asyncio.run(main())

General AI Agent with Custom Configuration

For more control over agent behavior:

import uuid
from panda_agi import Agent
from panda_agi.envs import LocalEnv

agent = Agent(
    api_key="your-api-key",      # API key for authentication
    conversation_id=str(uuid.uuid4()), # Must be a UUID4 string
    auto_reconnect=True,         # Automatic reconnection
    reconnect_interval=5.0,      # Reconnection delay
    environment=LocalEnv("./my_workspace")  # Custom workspace
)

Models

PandaAGI offers two powerful agentic models to suit your needs. You can specify the model using the model parameter in the Agent constructor.

Model	Description
annie-pro	Our most powerful and advanced generalist agentic technology. It excels at complex reasoning, multi-step task planning, and sophisticated skill integration. This is the recommended model for production applications that require high accuracy and robust performance.
annie-lite	A super powerful, yet lightweight, generalist agentic model optimized for speed and efficiency, at a fraction of the cost. It’s ideal for rapid prototyping, simple automation tasks, and applications where response time is critical.

By default, the SDK uses annie-pro. You can select annie-lite like this:

agent = Agent(model="annie-lite")

Using Environment Variables

Set your API key in environment variables:

export PANDA_AGI_KEY="your-api-key-here"

from panda_agi import Agent

# Agent will automatically use PANDA_AGI_KEY from environment
agent = Agent()

How do general AI agents work?

Understanding how agents work internally:

1. Initialization

   General AI agent initializes WebSocket client, event manager, and tool handlers

2. Connection

   General AI agent establishes WebSocket connection and waits for initialization

3. Request Processing

   General AI agent sends query and begins streaming events in real-time

4. Event Streaming

   General AI agent streams thoughts, tool usage, and results as they happen

5. Completion

   General AI agent signals task completion and can handle new requests

How to configure general AI agents?

Environment Setup

Configure the agent’s workspace:

from panda_agi.envs import LocalEnv

# Custom workspace directory
env = LocalEnv("./agent_workspace")

agent = Agent(environment=env)

# Change working directory
agent.change_working_directory("/path/to/project")

# Get current directory
current_dir = agent.get_working_directory()

Tool Handlers

Customize tool handling behavior:

from panda_agi.handlers import BaseHandler

class CustomToolHandler(BaseHandler):
    # ... implementation ...
    pass

custom_handlers = {
    "custom_tool": CustomToolHandler(),
    # Add your custom handlers here
}

agent = Agent(tools_handlers=custom_handlers)

Conversation Context

Enable follow-up questions and maintain context across multiple requests:

import asyncio
import uuid
from panda_agi import Agent
from panda_agi.handlers import LogsHandler

async def persistent_conversation():
    # Use same conversation_id for related requests
    # Must be a proper UUID4 string
    conversation_id = str(uuid.uuid4())
    handlers = [LogsHandler(use_colors=True, show_timestamps=True)]

    agent = Agent(conversation_id=conversation_id, event_handlers=handlers)
    
    # First request
    response1 = await agent.run("Tell me about Alice in Wonderland")
    print(f"First response: {response1.output}")
    
    # Follow-up request with context
    response2 = await agent.run("What's the main character's personality like?")
    print(f"Follow-up response: {response2.output}")  # Should mention "Alice"

if __name__ == "__main__":
    asyncio.run(persistent_conversation())

Key Benefits:

• Context Retention: Agent remembers previous questions and answers
• Natural Conversations: Ask follow-up questions without repeating context
• Session Persistence: Maintain conversations across multiple script runs

Important: The conversation_id must be a valid UUID4 string generated with str(uuid.uuid4()). This ensures proper conversation tracking and prevents conflicts.

Multi-Turn Conversations

Build complex interactions with follow-up questions:

import asyncio
import uuid
from panda_agi import Agent, EventType
from panda_agi.handlers import LogsHandler

async def multi_turn_conversation():
    # Consistent conversation_id enables follow-up questions
    # Must be a proper UUID4 string
    conversation_id = str(uuid.uuid4())
    handlers = [LogsHandler(use_colors=True, show_timestamps=True)]

    agent = Agent(conversation_id=conversation_id, event_handlers=handlers)
    
    # Initial request
    print("=== Initial Question ===")
    response1 = await agent.run("What are the main types of neural networks?")
    print(f"Response: {response1.output}")
    
    # Follow-up question - no need to repeat context
    print("\n=== Follow-up Question ===")
    response2 = await agent.run("Which one is best for image recognition?")
    print(f"Response: {response2.output}")  # Agent knows we're talking about neural networks
    
    # Another follow-up
    print("\n=== Another Follow-up ===")
    response3 = await agent.run("Can you show me a code example?")
    print(f"Response: {response3.output}")  # Agent will provide CNN code example

if __name__ == "__main__":
    asyncio.run(multi_turn_conversation())

Available Event Types

General AI agents emit various types of events during execution. For complete details on all events, see the Events documentation.

• File Operations: FILE_READ, FILE_WRITE, FILE_REPLACE, FILE_FIND, FILE_EXPLORE
• Web Operations: WEB_SEARCH, WEB_NAVIGATION, WEB_SEARCH_RESULT, WEB_NAVIGATION_RESULT
• Shell Operations: SHELL_EXEC, SHELL_VIEW, SHELL_WRITE
• Communication: AGENT_CONNECTION_SUCCESS, USER_NOTIFICATION, USER_QUESTION, COMPLETED_TASK
• Media Generation: IMAGE_GENERATION
• Skill Execution: USE_SKILL, USE_SKILL_RESULT

File System Integration

General AI agents can work with files in their environment:

import asyncio
from panda_agi import Agent, EventType
from panda_agi.handlers import BaseHandler, LogsHandler
from panda_agi.client.models import BaseStreamEvent

async def file_operations():
    agent = Agent()
    
    # Get current file system structure
    fs_structure = await agent.current_file_system
    print(f"Current directory: {fs_structure['directory']}")
    print(f"File structure: {fs_structure['structure']}")
    
    # Change working directory
    agent.change_working_directory("./documents")
    
    # Create custom handler to monitor file operations
    class FileMonitorHandler(BaseHandler):
        def process(self, event: BaseStreamEvent) -> None:
            if event.type == EventType.FILE_EXPLORE:
                print(f"Exploring files: {event.data}")
            elif event.type == EventType.FILE_READ:
                print(f"Reading file: {event.data}")
            elif event.type == EventType.FILE_WRITE:
                print(f"Organizing file: {event.data}")
            elif event.type == EventType.USER_NOTIFICATION:
                print(f"Organization update: {event.data}")

    handlers = [FileMonitorHandler("FileMonitor"), LogsHandler()]
    response = await agent.run("Organize these files by type", event_handlers=handlers)
    print(f"File organization completed: {response.output}")

if __name__ == "__main__":
    asyncio.run(file_operations())

Cross-Session Persistence

Maintain conversations across different script executions:

# session1.py
import asyncio
import uuid
from panda_agi import Agent, EventType
from panda_agi.handlers import LogsHandler

# Generate and save conversation ID for reuse
CONVERSATION_ID = str(uuid.uuid4())
print(f"Conversation ID: {CONVERSATION_ID}")
# Save this ID to reuse in session2.py

async def start_research():
    # Start a research session
    agent = Agent(conversation_id=CONVERSATION_ID)
    
    handlers = [LogsHandler(use_colors=True, show_timestamps=True)]

    response = await agent.run("Please analyze the Python project in my workspace")
    print(f"Analysis complete: {response.output}")
    print("Check session2.py for follow-ups.")

if __name__ == "__main__":
    asyncio.run(start_research())

# session2.py - Run this later with the same conversation ID
import asyncio
from panda_agi import Agent, EventType
from panda_agi.handlers import LogsHandler

# Use the same conversation ID from session1.py
CONVERSATION_ID = "your-uuid-from-session1"  # Replace with actual UUID

async def continue_research():
    handlers = [LogsHandler(use_colors=True, show_timestamps=True)]
    
    # Continue the same conversation
    agent = Agent(conversation_id=CONVERSATION_ID, event_handlers=handlers)
    
    # Agent remembers the previous analysis
    response = await agent.run("Based on your analysis, what improvements would you suggest?")
    print(f"Improvement suggestions completed: {response.output}")

if __name__ == "__main__":
    asyncio.run(continue_research())

Resource Management

Properly manage connections and resources:

# Use context managers when possible
async with Agent() as agent:
    # Work with agent
    pass

# Or ensure proper cleanup
agent = Agent()
try:
    # Work with agent
    pass
finally:
    await agent.disconnect()

Workspace Isolation

Set up appropriate workspace environments for security:

from panda_agi.envs import LocalEnv

# Isolated workspace for each task
env = LocalEnv("./task_specific_workspace")
agent = Agent(environment=env)

# Ensure proper permissions
agent.change_working_directory("/safe/working/directory")

Next Steps

Events

Learn about agent event streaming in detail

Environments

Understand agent workspace environments