import os import json import logging import re from typing import Dict, Any, List, Optional, Tuple # Logger konfigurieren logger = logging.getLogger(__name__) def get_agent_instructions(agent_type: str, agent: Dict[str, Any] = None, file_contexts: List[Dict[str, Any]] = None) -> str: """ Liefert Anweisungen für einen Agenten basierend auf seinen Attributen. Diese Version fügt explizite Informationen zu Datei-IDs hinzu. Args: agent_type: Typ des Agenten agent: Agent-Konfiguration mit allen Attributen file_contexts: Liste der verfügbaren Dateien Returns: Formatierte Anweisungen für den Agenten """ # Basis-Instruktionen aus dem Agenten-Profil extrahieren base_instructions = "" if agent: if agent.get("instructions"): base_instructions += agent.get("instructions").strip() + "\n\n" if agent.get("description"): base_instructions += "Kontext: " + agent.get("description").strip() + "\n\n" if agent.get("capabilities"): base_instructions += "Deine Fähigkeiten: " + agent.get("capabilities").strip() + "\n\n" # Wenn keine Instruktionen gefunden wurden, verwende eine generische Anweisung if not base_instructions: base_instructions = """ Analysiere die Anfrage gründlich und liefere ein konkretes, nützliches Ergebnis. Strukturiere deine Antwort klar und beantworte alle Aspekte der Anfrage. """ # Anweisung zur Selbstdeklaration des Ergebnisstatus hinzufügen status_declaration = """ WICHTIG: Deklariere am Ende deiner Antwort den Status deines Ergebnisses mit einem der folgenden Tags: [STATUS: ERGEBNIS] - Wenn du ein konkretes, vollständiges Ergebnis geliefert hast [STATUS: TEILWEISE] - Wenn du ein teilweises Ergebnis geliefert hast, das noch ergänzt werden sollte [STATUS: PLAN] - Wenn du hauptsächlich einen Plan oder eine Vorgehensweise vorgeschlagen hast Diese Deklaration hilft dem Moderator zu entscheiden, ob weitere Agentenarbeit erforderlich ist. """ # Konkrete Dateiinformationen bereitstellen file_info = "" if file_contexts: file_info = "Verfügbare Dateien:\n" for file in file_contexts: file_info += f"- {file['name']} (Typ: {file.get('type', 'unbekannt')}, ID: {file['id']})\n" file_info += "\n" # Hinweise zum Dateiaufrufen file_access = f""" {file_info} Um mehr Dateiinhalte anzufordern, verwende einen der folgenden Befehle: [[FILE:load_file(file_id=DATEI_ID, complete=true)]] - für den vollständigen Dateiinhalt [[FILE:load_file(file_id=DATEI_ID, pages=[1,2,3])]] - für spezifische Seiten einer PDF [[FILE:load_file(file_id=DATEI_ID, start=0, end=5000)]] - für Textabschnitte Ersetze DATEI_ID mit der tatsächlichen ID aus der Dateiliste oben (nur die ID-Nummer, keine Anführungszeichen). """ return base_instructions + status_declaration + file_access def get_default_agent_instructions() -> str: """ Gibt Standard-Anweisungen für einen Agenten zurück, wenn keine spezifischen Anweisungen verfügbar sind. 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 WICHTIG: Deklariere am Ende deiner Antwort den Status deines Ergebnisses: [STATUS: ERGEBNIS], [STATUS: TEILWEISE] oder [STATUS: PLAN] """ 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", "") # Kopiere alle Felder vom Original-Agenten und füge used-Status hinzu agent_data = agent.copy() agent_data["used"] = False agent_data["last_result_status"] = None # Für die Speicherung des Ergebnisstatus available_agents[agent_id] = agent_data # Log agent data for debugging logger.debug(f"Initialized agent: {agent_name} (Type: {agent_type})") if "instructions" in agent_data: logger.debug(f"Agent {agent_name} has instructions of length: {len(agent_data['instructions'])}") logger.info(f"Initialized {len(available_agents)} agents for workflow") return available_agents def get_moderator_prompt(available_agents: Dict[str, Dict[str, Any]]) -> str: """ Erstellt einen Moderator-Prompt, der die Status-Deklarationen der Agenten berücksichtigt. Der User Agent wird nur aufgerufen, wenn es Inputs von ihm benötigt. Args: available_agents: Dictionary mit verfügbaren Agenten Returns: Formatierter Prompt für den Moderator """ # Prüfen, ob der User Agent bereits verwendet wurde user_agent_used = False user_agent_confirmed = False if "user_agent" in available_agents: user_agent_used = available_agents["user_agent"].get("used", False) # Prüfe, ob der User Agent eine explizite Bestätigung gegeben hat if user_agent_used: for chat_entry in available_agents["user_agent"].get("chat_entries", []): if any(confirmation in chat_entry.lower() for confirmation in ["ja", "bestätige", "stimme zu", "akzeptiere"]): user_agent_confirmed = True break base = """Du bist Moderator eines Multi-Agent-Systems. Deine Aufgabe ist es, die Agenten zu koordinieren, um die Anfrage vollständig zu erfüllen und ein konkretes Endergebnis zu liefern. Der User wird nur aufgerufen, wenn es Inputs von ihm braucht, welche die anderen Agenten nicht liefern können. Somit immer zuerst die anderen Agenten fragen, zuletzt den User. WICHTIG: Der Workflow darf erst beendet werden, wenn TATSÄCHLICHE ERGEBNISSE geliefert wurden""" # Unterschiedliche Bedingungen für das Beenden des Workflows if not user_agent_confirmed: base += """ UND der User Agent explizit mit einem 'JA' bestätigt hat, dass er mit dem Ergebnis zufrieden ist. KRITISCH WICHTIG: Bevor du den Workflow beendest, MUSST du den User Agent befragen, ob er mit den Ergebnissen zufrieden ist, und er muss EXPLIZIT mit 'JA' oder einer eindeutigen Zustimmung antworten!""" agents_list = "\nVerfügbare Agenten:\n" for agent_id, agent in available_agents.items(): status = "✓ Bereits verwendet" if agent["used"] else "✗ Noch nicht verwendet" result_status = "" if agent["used"] and agent.get("last_result_status"): result_status = f" (Letzte Antwort: {agent.get('last_result_status')})" description = agent.get("description", "") capabilities = agent.get("capabilities", "") agents_list += f"- {agent['name']} ({agent['type']}): {capabilities}\n {description}\n Status: {status}{result_status}\n" instructions = """ Berücksichtige die STATUS-Deklarationen der Agenten bei deiner Entscheidung: - [STATUS: ERGEBNIS] - Der Agent hat ein vollständiges Ergebnis geliefert - [STATUS: TEILWEISE] - Der Agent hat ein teilweises Ergebnis geliefert, weitere Arbeit ist nötig - [STATUS: PLAN] - Der Agent hat einen Plan geliefert, keine konkreten Ergebnisse Mögliche Entscheidungen: - Bei Agent-Auswahl: "Ich wähle [Agentname], um [konkrete Aufgabe]" """ if not user_agent_confirmed: instructions += """- Bei Abschluss (nur wenn [STATUS: ERGEBNIS] vorliegt): DU MUSST ZUERST den User Agent explizit fragen, ob er mit den Ergebnissen zufrieden ist. Der User Agent MUSS mit "JA" oder einer eindeutigen Zustimmung antworten, bevor der Workflow beendet werden kann! WICHTIG: Du darfst den Workflow NICHT beenden, bevor der User Agent explizit mit "JA" bestätigt hat! Stelle dem User Agent eine KLARE, DIREKTE Frage, ob er mit dem Ergebnis zufrieden ist oder ob er weitere Informationen benötigt. """ else: instructions += """- Bei Abschluss (nur wenn [STATUS: ERGEBNIS] vorliegt und der User Agent mit "JA" bestätigt hat): "Workflow beenden - vollständiges Ergebnis erreicht" WICHTIG: Da der User Agent bereits seine Zustimmung gegeben hat, kannst du den Workflow jetzt beenden, wenn ein Agent ein konkretes [STATUS: ERGEBNIS] geliefert hat! """ return base + agents_list + instructions def track_user_agent_response(available_agents: Dict[str, Dict[str, Any]], message_content: str) -> None: """ Verfolgt die Antworten des User Agents, um eine explizite Bestätigung zu erkennen. Args: available_agents: Dictionary mit verfügbaren Agenten message_content: Inhalt der Nachricht des User Agents """ if "user_agent" not in available_agents: return # Initialisiere das chat_entries Array, falls es noch nicht existiert if "chat_entries" not in available_agents["user_agent"]: available_agents["user_agent"]["chat_entries"] = [] # Füge die aktuelle Nachricht hinzu available_agents["user_agent"]["chat_entries"].append(message_content) # Prüfe, ob die Nachricht eine explizite Bestätigung enthält confirmation_phrases = ["ja", "bestätige", "stimme zu", "akzeptiere", "einverstanden", "passt", "okay", "ok"] has_confirmation = any(phrase in message_content.lower() for phrase in confirmation_phrases) # Setze das confirmed-Flag entsprechend available_agents["user_agent"]["confirmed"] = has_confirmation # Loggen der Erkennung if has_confirmation: logger.info("User Agent hat explizit mit 'Ja' bestätigt") else: logger.info("User Agent hat keine eindeutige Bestätigung gegeben") def create_agent_prompt(agent: Dict[str, Any], agent_instructions: str, file_contexts: List[Dict[str, Any]] = None) -> Dict[str, str]: """Verbesserter Agent-Prompt mit Datei-IDs""" # Füge Datei-ID-Infos hinzu, wenn verfügbar file_info = "" if file_contexts: file_info = "\nVerfügbare Dateien:\n" for file in file_contexts: file_info += f"- {file['name']} (ID: {file['id']})\n" content = f""" Du bist Agent {agent['name']} ({agent['type']}). {agent_instructions} {file_info} Format: [Agent: {agent['name']}] Deine Antwort... """.strip() return {"role": "system", "content": content} def find_next_agent(moderator_text: str, available_agents: Dict[str, Dict[str, Any]]) -> Optional[str]: """ Findet den nächsten Agenten basierend auf der Moderator-Entscheidung. Berücksichtigt die Anweisung zum Workflow-Abschluss nur, wenn ein Ergebnis vorliegt und der User Agent explizit mit "JA" bestätigt hat. Args: moderator_text: Text der Moderator-Entscheidung available_agents: Dictionary mit verfügbaren Agenten Returns: ID des nächsten Agenten oder "WORKFLOW_COMPLETE" zum Beenden """ text = moderator_text.lower() # Prüfe, ob der Workflow beendet werden soll - nur wenn vollständige Ergebnisse vorliegen workflow_complete_phrases = [ "workflow beenden - vollständiges ergebnis erreicht", "workflow beenden - vollständiges ergebnis", "vollständiges ergebnis erreicht" ] # Prüfen, ob ein Agent ein Ergebnis geliefert hat result_exists = False for agent_id, agent in available_agents.items(): if agent.get("used") and agent.get("last_result_status") == "ERGEBNIS": result_exists = True break # Prüfen, ob der User Agent bereits befragt wurde und mit "JA" bestätigt hat user_agent_confirmed = False if "user_agent" in available_agents and available_agents["user_agent"].get("used", False): # Prüfe die letzte Nachricht des User Agents auf explizite Bestätigung for chat_entry in available_agents["user_agent"].get("chat_entries", []): if any(confirmation in chat_entry.lower() for confirmation in ["ja", "bestätige", "stimme zu", "akzeptiere"]): user_agent_confirmed = True break # Suche nach exakten Phrasen für Workflow-Beendigung if "workflow beenden" in text: # Wenn ein eindeutiges Ergebnis vorliegt und der User Agent explizit bestätigt hat if result_exists and user_agent_confirmed: if any(phrase in text for phrase in workflow_complete_phrases): return "WORKFLOW_COMPLETE" # Sonst: In den Logs warnen, dass Bedingungen für Beendigung nicht erfüllt sind else: if not result_exists: logger.warning("Moderator versuchte, Workflow ohne vollständiges Ergebnis zu beenden") if not user_agent_confirmed: logger.warning("Moderator versuchte, Workflow ohne explizite Bestätigung des User Agents zu beenden") # Wähle den User Agent aus, um eine explizite Bestätigung zu erhalten if "user_agent" in available_agents: return "user_agent" # Suche nach "ich wähle" Pattern für Agentenwahl if "ich wähle" in text: for agent_id, agent in available_agents.items(): agent_name_lower = agent["name"].lower() if agent_name_lower in text: return agent_id # Fallback: Direktes Name-Matching for agent_id, agent in available_agents.items(): agent_name_lower = agent["name"].lower() if agent_name_lower in text or f"agent {agent_name_lower}" in text: return agent_id # Wenn kein Agent explizit gewählt wurde und User Agent noch nicht befragt wurde # oder keine explizite Bestätigung gegeben hat, priorisiere den User Agent if result_exists and (not "user_agent" in available_agents or not available_agents["user_agent"].get("used", False) or not user_agent_confirmed): if "user_agent" in available_agents: return "user_agent" # Wenn kein Agent explizit gewählt wurde: Wähle den ersten unbenutzten Agenten for agent_id, agent in available_agents.items(): if not agent["used"]: return agent_id # Letzter Ausweg: Ersten Agenten wiederverwenden if available_agents: return list(available_agents.keys())[0] return None def update_agent_results_with_status(content: str) -> Tuple[str, str]: """ Extrahiert den deklarierten Status aus einem Agenten-Ergebnis. Args: content: Der vom Agenten gelieferte Ergebnistext Returns: Tuple mit (bereinigter Text, Status) """ # Standard-Status, falls keine Deklaration gefunden wird status = "UNBEKANNT" # Suche nach Status-Deklaration status_pattern = r'\[STATUS:\s*(ERGEBNIS|TEILWEISE|PLAN)\]' match = re.search(status_pattern, content, re.IGNORECASE) if match: # Extrahiere den Status status = match.group(1).upper() # Entferne die Status-Deklaration aus dem Text content = re.sub(status_pattern, '', content, flags=re.IGNORECASE).strip() return content, status def extract_summary(text: str, max_length: int = 200) -> str: """ Extrahiert eine kurze Zusammenfassung aus einem Text. Args: text: Der zu extrahierende Text max_length: Maximale Länge der Zusammenfassung Returns: Eine kurze Zusammenfassung des Textes """ # Erste Zeilen oder Absätze extrahieren lines = text.split('\n') paragraphs = [] current_para = [] for line in lines: line = line.strip() if not line: if current_para: paragraphs.append(' '.join(current_para)) current_para = [] else: current_para.append(line) if current_para: paragraphs.append(' '.join(current_para)) # Versuche, den ersten oder zweiten Absatz als Zusammenfassung zu verwenden if paragraphs: summary = paragraphs[0] # Falls der erste Absatz zu kurz ist, versuche den zweiten hinzuzufügen if len(summary) < 100 and len(paragraphs) > 1: summary += " " + paragraphs[1] # Kürze auf die maximale Länge if len(summary) > max_length: summary = summary[:max_length-3] + "..." return summary # Fallback auf die ersten Zeichen des Textes if text: return text[:max_length-3] + "..." if len(text) > max_length else text return "Keine Zusammenfassung verfügbar"