""" Chat Agent Registry Modul. Stellt ein zentrales Registry-System für alle verfügbaren Agenten bereit. """ import os import logging import importlib from typing import Dict, Any, List, Optional logger = logging.getLogger(__name__) class AgentRegistry: """Zentrale Registry für alle verfügbaren Agenten im System.""" _instance = None @classmethod def get_instance(cls): """Gibt eine Singleton-Instanz der Agent-Registry zurück.""" if cls._instance is None: cls._instance = cls() return cls._instance def __init__(self): """Initialisiert die Agent-Registry.""" if AgentRegistry._instance is not None: raise RuntimeError("Singleton-Instanz existiert bereits - verwende get_instance()") self.agents = {} self.ai_service = None self._load_agents() def _load_agents(self): """Lädt alle verfügbaren Agenten aus den Modulen.""" logger.info("Lade Agent-Module...") # Liste der zu ladenden Agent-Module agent_modules = [] agent_dir = os.path.dirname(__file__) # Durchsuche das Verzeichnis nach Agent-Modulen for filename in os.listdir(agent_dir): if filename.startswith("chat_agent_") and filename.endswith(".py"): agent_modules.append(filename[:-3]) # Entferne .py-Endung if not agent_modules: logger.warning("Keine Agent-Module gefunden") return logger.info(f"{len(agent_modules)} Agent-Module gefunden") # Lade jedes Agent-Modul for module_name in agent_modules: try: # Importiere das Modul module = importlib.import_module(f"modules.{module_name}") # Suche nach der Agent-Klasse oder einer get_*_agent-Funktion agent_name= module_name.split('_')[-1] class_name = f"Agent{agent_name.capitalize()}" getter_name = f"get_{agent_name}_agent" agent = None # Versuche, den Agenten über die get_*_agent-Funktion zu erhalten if hasattr(module, getter_name): getter_func = getattr(module, getter_name) agent = getter_func() logger.info(f"Agent '{agent.name}' über {getter_name}() geladen") # Alternativ versuche, den Agenten direkt zu instanziieren elif hasattr(module, class_name): agent_class = getattr(module, class_name) agent = agent_class() logger.info(f"Agent '{agent.name}' (Typ: {agent.name}) direkt instanziert") if agent: # Registriere den Agenten self.register_agent(agent) else: logger.warning(f"Keine Agent-Klasse oder Getter-Funktion in Modul {module_name} gefunden") except ImportError as e: logger.error(f"Modul {module_name} konnte nicht importiert werden: {e}") except Exception as e: logger.error(f"Fehler beim Laden des Agenten aus Modul {module_name}: {e}") def set_ai_service(self, ai_service): self.ai_service = ai_service self.update_agent_dependencies() def update_agent_dependencies(self): """Aktualisiert die Abhängigkeiten für alle registrierten Agenten.""" for agent_id, agent in self.agents.items(): if hasattr(agent, 'set_dependencies'): agent.set_dependencies(ai_service=self.ai_service) def register_agent(self, agent): """ Registriert einen Agenten in der Registry. Args: agent: Der zu registrierende Agent """ agent_id = getattr(agent, 'name', "unknown_agent") # Initialisiere Agenten mit Abhängigkeiten if hasattr(agent, 'set_dependencies'): agent.set_dependencies(ai_service=self.ai_service) self.agents[agent_id] = agent logger.debug(f"Agent '{agent.name}' (Typ: {agent_id}, Name: {agent_id}) registriert") def get_agent(self, agent_identifier: str): """ Gibt eine Agenten-Instanz zurück Args: agent_identifier: ID oder Typ des gewünschten Agenten Returns: Agenten-Instanz oder None, falls nicht gefunden """ if agent_identifier in self.agents: return self.agents[agent_identifier] logger.error(f"Agent mit Kennung '{agent_identifier}' nicht gefunden") return None def get_all_agents(self) -> Dict[str, Any]: """Gibt alle registrierten Agenten zurück.""" return self.agents def get_agent_infos(self) -> List[Dict[str, Any]]: """Gibt Informationen über alle registrierten Agenten zurück.""" agent_infos = [] seen_agents = set() for agent in self.agents.values(): if agent not in seen_agents: # Verwende get_agent_info oder erstelle manuell die Info if hasattr(agent, 'get_agent_info'): agent_infos.append(agent.get_agent_info()) else: agent_infos.append({ "name": agent.name, "capabilities": getattr(agent, 'capabilities', ""), }) logger.error(f"Agent '{agent.name}' does not show profile.") seen_agents.add(agent) return agent_infos # Basis-Agent-Klasse class AgentBase: """ Basis-Klasse für alle Chat-Agenten. Definiert die grundlegende Schnittstelle und Funktionalität. """ def __init__(self): """Initialisiere den Basis-Agenten.""" self.name = "Basis-Agent" self.capabilities = "Grundlegende Agentenfunktionen" self.ai_service = None def set_dependencies(self, ai_service=None): self.ai_service = ai_service def get_config(self) -> Dict[str, Any]: return { "name": self.name, "capabilities": self.capabilities, } async def process_message(self, message: Dict[str, Any], context: Dict[str, Any] = None) -> Dict[str, Any]: # Basisimplementierung - sollte von spezialisierten Agenten überschrieben werden if not self.ai_service: logger.warning(f"Agent {self.id} hat keinen konfigurierten AI-Service") return { "role": "assistant", "content": f"Ich bin {self.name}, aber ich bin nicht richtig konfiguriert. Bitte den AI-Service einrichten.", "agent_name": self.name, } # Einfachen Prompt erstellen prompt = message.get("content", "") # Antwort generieren try: response_content = self.ai_service.call_api([ {"role": "system", "content": f"Du bist {self.name}, ein spezialisierter {self.name}-Agent mit Fähigkeiten in: {self.capabilities}"}, {"role": "user", "content": prompt} ]) return { "role": "assistant", "content": response_content, "agent_name": self.name, } except Exception as e: logger.error(f"Fehler in Agent {self.id}: {str(e)}") return { "role": "assistant", "content": f"Ich habe einen Fehler festgestellt: {str(e)}", "agent_name": self.name, } # Singleton-Factory für die Agent-Registry def get_agent_registry(): return AgentRegistry.get_instance()