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. Args: available_agents: Dictionary mit verfügbaren Agenten Returns: Formatierter Prompt für den Moderator """ 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. WICHTIG: Der Workflow darf erst beendet werden, wenn TATSÄCHLICHE ERGEBNISSE geliefert wurden.""" 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]" - Bei Abschluss (nur wenn [STATUS: ERGEBNIS] vorliegt): "Workflow beenden - vollständiges Ergebnis erreicht" WICHTIG: Du darfst den Workflow NUR beenden, wenn mindestens ein Agent ein konkretes [STATUS: ERGEBNIS] geliefert hat! """ return base + agents_list + instructions 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. 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 # Suche nach exakten Phrasen für Workflow-Beendigung if "workflow beenden" in text: # Wenn ein eindeutiges Ergebnis vorliegt oder eine spezifische Phrase gefunden wurde if result_exists or any(phrase in text for phrase in workflow_complete_phrases): return "WORKFLOW_COMPLETE" # Sonst: In den Logs warnen, dass der Moderator versucht, ohne Ergebnis zu beenden else: logger.warning("Moderator versuchte, Workflow ohne vollständiges Ergebnis zu beenden") # 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: 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"