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