import os import json import logging from typing import Dict, Any, List, Optional # Logger konfigurieren logger = logging.getLogger(__name__) def get_agent_instructions(agent_type: str) -> str: """ Gibt agententypspezifische Anweisungen zurück, die aus der agents.json geladen werden. Erweitert um Hinweise zur Dateianforderung. Args: agent_type: Typ des Agenten, für den Anweisungen geladen werden sollen Returns: Die geladenen oder Standard-Anweisungen mit Dateizugriffsinformationen """ try: # Pfad zur agents.json-Datei agents_file = os.path.join(os.path.dirname(__file__), 'data', 'agents.json') # Überprüfen, ob die Datei existiert if not os.path.exists(agents_file): logger.warning(f"Agents-Definitionen nicht gefunden: {agents_file}") instructions = get_default_agent_instructions() else: # Datei lesen with open(agents_file, 'r', encoding='utf-8') as f: agents_data = json.load(f) # Nach dem Agententyp suchen instructions = None for agent in agents_data: if agent.get("type") == agent_type: # Anweisungen zurückgeben, wenn vorhanden instructions = agent.get("instructions") if instructions: logger.debug(f"Anweisungen für Agent-Typ '{agent_type}' aus agents.json geladen") break # Wenn kein passender Agent gefunden wurde, Standardanweisungen verwenden if not instructions: logger.warning(f"Keine Anweisungen für Agent-Typ '{agent_type}' in agents.json gefunden") instructions = get_default_agent_instructions() # Füge Hinweise zur Dateianforderung hinzu file_access_instructions = """ # Weitere Dateiinhalte anfordern Falls du mehr Details aus einer Datei benötigst, kannst du zusätzliche Dateiinhalte mit folgendem Befehl anfordern: [[FILE:load_file(file_id=ID, complete=True/False, start=N, end=M, pages=[1,2,3])]] Parameter: - file_id: Die ID der Datei (erforderlich) - complete: Wenn 'true', wird die gesamte Datei geladen - start, end: Startet und Endposition (in Zeichen) für Textdateien - pages: Liste von Seitennummern für PDFs (z.B. [1,3,5]) Beispiele: [[FILE:load_file(file_id="doc1", complete=true)]] [[FILE:load_file(file_id="doc2", pages=[1,2,3])]] Der angeforderte Dateiinhalt wird dir als Antwort bereitgestellt, bevor du deine Analyse fortsetzen kannst. """ return instructions + file_access_instructions except Exception as e: logger.error(f"Fehler beim Laden der Agent-Anweisungen aus agents.json: {e}") return get_default_agent_instructions() def get_default_agent_instructions() -> str: """ Gibt Standard-Anweisungen für einen Agenten zurück, wenn keine spezifischen Anweisungen in der agents.json gefunden wurden. Diese Funktion gibt generische Anweisungen zurück, unabhängig vom Agententyp. """ return """ Als Agent ist es deine Aufgabe, Anfragen zu analysieren und entsprechend deinen Fähigkeiten zu bearbeiten. Folge diesen allgemeinen Anweisungen: 1. Verstehe die Anfrage gründlich 2. Analysiere relevante Daten und Informationen 3. Liefere präzise und hilfreiche Antworten 4. Strukturiere deine Antwort klar und verständlich In deiner Antwort: - Beginne mit einer Zusammenfassung der Anfrage - Gib gut begründete Antworten oder Empfehlungen - Führe wichtige Erkenntnisse klar auf - Schließe mit konkreten nächsten Schritten oder Empfehlungen ab """ def initialize_agents(agents: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: """ Initialisiert die Agenten mit ihren Fähigkeiten und Status Args: agents: Liste der Agenten aus dem Workflow Returns: Dictionary mit Agent-IDs als Schlüssel und Agent-Informationen """ available_agents = {} for agent in agents: agent_id = agent["id"] agent_name = agent["name"] agent_type = agent["type"] agent_capabilities = agent.get("capabilities", "") available_agents[agent_id] = { "id": agent_id, "name": agent_name, "type": agent_type, "capabilities": agent_capabilities, "used": False } return available_agents def get_moderator_prompt(available_agents: Dict[str, Dict[str, Any]]) -> str: """ Erstellt den Prompt für den Moderator des Multi-Agent-Systems Args: available_agents: Dictionary mit verfügbaren Agenten Returns: Der vollständige Moderator-Prompt """ # Basis-Prompt für den Moderator moderator_prompt_base = """ Du bist der Moderator eines Multi-Agent-Systems. Deine Aufgabe ist es, die Zusammenarbeit zwischen verschiedenen spezialisierten Agenten zu koordinieren, um die Anfrage des Benutzers bestmöglich zu erfüllen. Du sollst: 1. Die Anfrage des Benutzers verstehen und analysieren 2. Den am besten geeigneten Agenten basierend auf seinen Fähigkeiten auswählen 3. Die Antworten der Agenten überwachen und bewerten 4. Falls nötig, weitere Agenten hinzuziehen, um die Anfrage vollständig zu bearbeiten 5. Den Workflow beenden, wenn die Anfrage vollständig erfüllt wurde Für jeden Schritt sollst du begründen, warum du einen bestimmten Agenten auswählst, und zusammenfassen, was bisher erreicht wurde. """ # Dynamischer Teil - Verfügbare Agenten aus den tatsächlich vorhandenen Agenten agents_description = "Verfügbare Agenten:\n" for agent_id, agent in available_agents.items(): agents_description += f"- {agent['name']} (Typ: {agent['type']}): {agent['capabilities']}\n" moderator_prompt_end = """ Beende den Workflow, wenn die Aufgabe erfüllt ist oder keine weiteren Agenten zur Bearbeitung beitragen können. """ # Kombiniere alle Teile return moderator_prompt_base + "\n" + agents_description + "\n" + moderator_prompt_end def create_agent_prompt(agent: Dict[str, Any], agent_instructions: str) -> Dict[str, str]: """ Erstellt den Prompt für einen spezifischen Agenten Args: agent: Agent-Informationen agent_instructions: Anweisungen für den Agententyp Returns: Der formatierte Agent-Prompt als Dictionary """ return { "role": "system", "content": f""" # Aufgabe Du bist ein spezialisierter Agent vom Typ {agent['type']} mit dem Namen {agent['name']}. {agent_instructions} Bitte analysiere den Chatverlauf und die Dateien und beantworte die Anfrage gemäß deiner Rolle. Ausgabeformat: [Agent: {agent['name']}] Deine Antwort... """ } def find_next_agent(moderator_text: str, available_agents: Dict[str, Dict[str, Any]]) -> Optional[str]: """ Findet den vom Moderator ausgewählten Agenten anhand des Moderator-Textes Args: moderator_text: Die Antwort des Moderators available_agents: Dictionary mit verfügbaren Agenten Returns: Die ID des ausgewählten Agenten oder None, wenn kein Agent gefunden wurde """ # Prüfe, ob der Workflow beendet werden soll if any(phrase in moderator_text.lower() for phrase in [ "workflow beenden", "aufgabe erfüllt", "beende den workflow", "workflow abschließen" ]): return "WORKFLOW_COMPLETE" # Versuche, den ausgewählten Agenten aus dem Text zu extrahieren for agent_id, agent in available_agents.items(): if agent["name"] in moderator_text or f"Agent {agent_id}" in moderator_text: return agent_id # Keine direkte Erwähnung gefunden - wähle den ersten nicht verwendeten Agenten for agent_id, agent in available_agents.items(): if not agent["used"]: return agent_id # Wenn alle Agenten bereits verwendet wurden, wähle den Initialisierungs-Agenten for agent_id, agent in available_agents.items(): if agent["type"] == "initialisierung": return agent_id # Als letztes Mittel wähle einfach den ersten Agenten if available_agents: return list(available_agents.keys())[0] # Keine Agenten vorhanden oder keine Auswahl möglich return None