""" 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()}