Spaces:
Sleeping
Sleeping
File size: 2,617 Bytes
f7e620e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
"""
Agent manager for EmotionMirror application.
Coordinates communication between different agents.
"""
import logging
from typing import Dict, Any, Optional
from agent_framework.base_agent import BaseAgent
from agent_framework.visual_agent import VisualAgent
# Configure logging
logger = logging.getLogger(__name__)
class AgentManager:
"""Manages agents and coordinates their communication"""
_instance = None
def __new__(cls):
"""Singleton pattern implementation"""
if cls._instance is None:
cls._instance = super(AgentManager, cls).__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
"""Initialize the agent manager"""
if self._initialized:
return
self._initialized = True
logger.info("Initializing AgentManager")
# Dictionary to store agents
self.agents = {}
# Initialize agents
self._initialize_agents()
def _initialize_agents(self) -> None:
"""Initialize available agents"""
# In Phase 1, we only have the VisualAgent
self.register_agent(VisualAgent())
def register_agent(self, agent: BaseAgent) -> None:
"""
Register an agent with the manager.
Args:
agent: Agent instance to register
"""
self.agents[agent.name] = agent
logger.info(f"Agent registered: {agent.name}")
def get_agent(self, agent_name: str) -> Optional[BaseAgent]:
"""
Get an agent by name.
Args:
agent_name: Name of the agent
Returns:
Agent instance or None if not found
"""
return self.agents.get(agent_name)
def process_visual(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Process visual data using the VisualAgent.
Args:
data: Input data for processing
Returns:
Processing results
"""
visual_agent = self.get_agent("VisualAgent")
if visual_agent:
return visual_agent.process(data)
else:
logger.error("VisualAgent not found")
return {"error": "VisualAgent not available"}
def list_agents(self) -> Dict[str, str]:
"""
List all registered agents.
Returns:
Dictionary of agent names and descriptions
"""
return {name: agent.description for name, agent in self.agents.items()}
|