Agent Architecture

Agent Architecture (Phase 3)

This document describes the architecture of the autonomous agent system in Adastrea Director.

Overview

The Agent System (Phase 3) provides a flexible, extensible framework for autonomous agents that can proactively monitor, analyze, and assist with game development tasks.

Architecture Diagram

┌─────────────────────────────────────────────────────────────┐
│                     Orchestrator                             │
│  • Agent Lifecycle Management                                │
│  • Event Coordination                                        │
│  • Resource Allocation                                       │
└─────────────┬───────────────────────────────────────────────┘
              │
    ┌─────────┴─────────┐
    │                   │
    ▼                   ▼
┌────────────┐    ┌────────────┐
│ Event Bus  │    │   Shared   │
│            │◄───┤   State    │
│ (Pub/Sub)  │    │  Manager   │
└─────┬──────┘    └────────────┘
      │
      ├──────────┬──────────┬──────────┐
      ▼          ▼          ▼          ▼
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│Performance│ │   Bug    │ │   Code   │ │  Future  │
│  Agent   │ │Detection │ │ Quality  │ │  Agents  │
│          │ │  Agent   │ │  Agent   │ │          │
└──────────┘ └──────────┘ └──────────┘ └──────────┘

Core Components

1. Agent Orchestrator

Purpose: Manages the lifecycle and coordination of all agents.

Responsibilities:

• Start/stop agents
• Monitor agent health
• Handle agent failures
• Resource allocation
• Event coordination

Implementation:

class AgentOrchestrator:
    def __init__(self):
        self.agents = {}
        self.event_bus = EventBus()
        self.shared_state = SharedState()
    
    def start_agent(self, agent_type: str):
        agent = AgentFactory.create(agent_type)
        agent.start()
        self.agents[agent_type] = agent
    
    def stop_agent(self, agent_type: str):
        if agent_type in self.agents:
            self.agents[agent_type].stop()
            del self.agents[agent_type]

2. Event Bus

Purpose: Facilitates communication between agents via publish-subscribe pattern.

Features:

• Asynchronous message passing
• Topic-based subscriptions
• Event filtering
• Event history

Example:

# Agent publishes event
event_bus.publish("performance.warning", {
    "metric": "fps",
    "value": 30,
    "threshold": 60,
    "timestamp": datetime.now()
})

# Other agents subscribe
event_bus.subscribe("performance.*", performance_callback)
event_bus.subscribe("*.warning", alert_callback)

3. Shared State

Purpose: Thread-safe state management across all agents.

Features:

• Atomic operations
• State persistence
• Rollback support
• State queries

Example:

# Agent updates state
shared_state.set("last_performance_check", datetime.now())

# Agent reads state
last_check = shared_state.get("last_performance_check")

# Atomic increment
shared_state.increment("alert_count")

4. Base Agent

All agents inherit from a base agent class:

class BaseAgent(ABC):
    def __init__(self, event_bus, shared_state):
        self.event_bus = event_bus
        self.shared_state = shared_state
        self.running = False
    
    @abstractmethod
    def run(self):
        """Main agent loop"""
        pass
    
    @abstractmethod
    def on_event(self, event):
        """Handle incoming events"""
        pass
    
    def start(self):
        self.running = True
        threading.Thread(target=self.run).start()
    
    def stop(self):
        self.running = False

Agent Types

Performance Agent

Purpose: Monitor and optimize performance metrics.

Capabilities:

• FPS monitoring
• Memory usage tracking
• CPU profiling
• GPU utilization
• Frame time analysis

Events Published:

• performance.warning - Performance issue detected
• performance.report - Periodic performance report
• performance.optimized - Optimization applied

Events Subscribed:

• system.start - Begin monitoring
• system.stop - Stop monitoring

Bug Detection Agent

Purpose: Identify potential bugs and issues.

Capabilities:

• Log analysis
• Crash detection
• Error pattern recognition
• Reproduction step generation

Events Published:

• bug.detected - New bug found
• bug.analyzed - Bug analysis complete
• bug.reproduced - Reproduction steps available

Events Subscribed:

• system.error - System error occurred
• crash.reported - Crash detected

Code Quality Agent

Purpose: Monitor code quality and suggest improvements.

Capabilities:

• Code smell detection
• Complexity analysis
• Style checking
• Best practice validation

Events Published:

• quality.issue - Quality issue found
• quality.suggestion - Improvement suggested
• quality.improved - Quality metric improved

Events Subscribed:

• code.changed - Code was modified
• build.completed - Build finished

Communication Patterns

Event-Driven

Agents communicate via events:

# Pattern 1: Reactive
@subscribe("performance.warning")
def on_performance_warning(event):
    # React to performance issues
    analyze_and_optimize(event.data)

# Pattern 2: Proactive
def agent_loop():
    while running:
        metrics = collect_metrics()
        if metrics.fps < threshold:
            event_bus.publish("performance.warning", metrics)
        time.sleep(interval)

Request-Response

For synchronous operations:

# Agent requests action
result = orchestrator.request_action("analyze_code", {
    "file": "player.cpp",
    "lines": [100, 150]
})

# Orchestrator handles request
def request_action(action_type, params):
    if action_type == "analyze_code":
        return code_analyzer.analyze(params)

State Sharing

Agents share state via shared state manager:

# Agent A sets state
shared_state.set("last_optimization", {
    "timestamp": datetime.now(),
    "improvement": 15.3  # % FPS improvement
})

# Agent B reads state
last_opt = shared_state.get("last_optimization")
if last_opt and is_recent(last_opt["timestamp"]):
    # Consider recent optimization in analysis
    pass

Agent Lifecycle

Initialization

Create Agent → Register with Orchestrator → Subscribe to Events → Initialize State

Running

Event Loop:
  1. Check for events
  2. Process events
  3. Perform periodic tasks
  4. Publish results
  5. Update state

Shutdown

Stop Signal → Cleanup → Unsubscribe → Final Report → Terminate

Design Patterns

1. Observer Pattern

Agents observe events via the event bus:

• Loose coupling
• Scalable
• Easy to add new agents

2. Singleton Pattern

Shared state and event bus are singletons:

• Single source of truth
• Consistent state
• Easy access

3. Factory Pattern

Agent creation via factory:

• Encapsulated creation logic
• Easy to add new agent types
• Configuration-driven

4. Template Method Pattern

Base agent defines workflow:

• Consistent agent structure
• Reusable base functionality
• Easy to extend

Scalability

Current Scale

• Agents: Up to 10 concurrent
• Events: 1000+ events/second
• State: Thread-safe for concurrent access

Future Scaling

Distributed Agents:

Orchestrator → Message Queue → Agent Workers
                 (RabbitMQ)     (Multiple processes)

Agent Clustering:

Load Balancer → Agent Cluster 1
             → Agent Cluster 2
             → Agent Cluster 3

Integration Points

With RAG System (P1)

Agents use RAG for context:

# Agent queries documentation
context = rag_system.query("How to optimize rendering?")
optimization = analyze_with_context(metrics, context)

With Planning System (P2)

Agents create plans:

# Agent generates optimization plan
plan = planner.create_plan("Optimize rendering for target FPS")
execute_plan(plan)

With Remote Control API

External systems control agents:

# HTTP API
POST /agents/start
POST /agents/stop
GET /agents/status
GET /agents/events

Testing

Unit Tests

Test individual agent components:

def test_performance_agent_detects_low_fps():
    agent = PerformanceAgent(mock_event_bus, mock_state)
    metrics = {"fps": 30}
    agent.analyze(metrics)
    assert mock_event_bus.published("performance.warning")

Integration Tests

Test agent interactions:

def test_agents_coordinate_optimization():
    orchestrator = AgentOrchestrator()
    orchestrator.start_agent("performance")
    orchestrator.start_agent("code_quality")
    
    # Trigger performance issue
    simulate_low_fps()
    
    # Verify agents coordinated
    assert check_optimization_applied()

End-to-End Tests

Test complete workflows:

def test_full_optimization_workflow():
    # Start system
    orchestrator.start_all()
    
    # Simulate issue
    trigger_performance_issue()
    
    # Verify resolution
    wait_for_resolution()
    assert performance_improved()

Security Considerations

Sandboxing

Agents run in restricted environment:

• Limited file system access
• No arbitrary code execution
• Resource limits

Authorization

Agents require permissions:

@requires_permission("file.read")
def read_code_file(path):
    return read_file(path)

@requires_permission("system.modify")
def apply_optimization(optimization):
    modify_system(optimization)

Future Enhancements

Planned Agent Types

1. Asset Management Agent

   • Optimize asset sizes
   • Detect unused assets
   • Suggest LOD improvements

2. Testing Agent

   • Generate test cases
   • Run automated tests
   • Report coverage

3. Documentation Agent

   • Generate documentation
   • Keep docs up-to-date
   • Suggest improvements

Advanced Features

• Machine Learning: Learn from past optimizations
• Multi-Agent Collaboration: Agents work together on complex tasks
• Adaptive Behavior: Agents adapt to project patterns

Related Documentation

• System Architecture
• Phase 3 Guide
• Autonomous Agents Usage
• Remote Control API

---

← Back to Architecture | Deployment Modes →