237 lines
No EOL
9.8 KiB
Python
237 lines
No EOL
9.8 KiB
Python
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, agent: Dict[str, Any] = None) -> str:
|
|
"""
|
|
Gets agent-specific instructions. Prioritizes instructions from the agent data
|
|
if available, falling back to default instructions if needed.
|
|
|
|
Args:
|
|
agent_type: Type of the agent for which to load instructions
|
|
agent: The agent object containing data from the database (optional)
|
|
|
|
Returns:
|
|
The loaded or default instructions with file access information
|
|
"""
|
|
instructions = None
|
|
|
|
# First, try to get instructions directly from the agent data
|
|
if agent and agent.get("instructions"):
|
|
logger.debug(f"Using instructions from agent data for type '{agent_type}'")
|
|
instructions = agent.get("instructions")
|
|
else:
|
|
logger.warning(f"No instructions found for agent type '{agent_type}', using default")
|
|
instructions = get_default_agent_instructions()
|
|
|
|
# Add file access instructions
|
|
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
|
|
|
|
|
|
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
|
|
"""
|
|
|
|
|
|
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
|
|
|
|
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 den Prompt für den Moderator des Multi-Agent-Systems
|
|
"""
|
|
# 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.
|
|
|
|
Als Moderator:
|
|
1. Wähle für jede Runde EINEN Agenten aus, der mit der spezifischen Aufgabe fortfahren soll
|
|
2. Wähle den Agenten basierend auf seinen Fähigkeiten und dem aktuellen Stand der Bearbeitung
|
|
3. Bewerte die bisherigen Ergebnisse kritisch - ist die Anfrage des Benutzers WIRKLICH beantwortet?
|
|
4. Vermeide es, einen Workflow als abgeschlossen zu betrachten, nur weil alle Agenten einmal verwendet wurden
|
|
|
|
WICHTIG: Beende den Workflow NUR, wenn die Benutzeranfrage vollständig und qualitativ hochwertig beantwortet wurde.
|
|
Ein Agent kann mehrfach zum Einsatz kommen, wenn weitere Arbeit nötig ist.
|
|
"""
|
|
|
|
# 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():
|
|
status = "✓ Bereits verwendet" if agent["used"] else "✗ Noch nicht verwendet"
|
|
agents_description += f"- {agent['name']} (Typ: {agent['type']}): {agent['capabilities']}\n Status: {status}\n"
|
|
|
|
moderator_prompt_end = """
|
|
REGELN FÜR DEINE ANTWORT:
|
|
|
|
1. WÄHLE EINEN AGENTEN EXPLIZIT aus, indem du einen dieser Sätze verwendest:
|
|
- "Ich wähle den [Name des Agenten] aus, um..."
|
|
- "Ich empfehle, dass [Name des Agenten] jetzt..."
|
|
|
|
2. ODER BEENDE DEN WORKFLOW, nur wenn alle folgenden Bedingungen erfüllt sind:
|
|
- Die Anfrage des Benutzers wurde vollständig beantwortet
|
|
- Alle erforderlichen Informationen wurden gesammelt
|
|
- Eine hochwertige, strukturierte Antwort liegt vor
|
|
|
|
3. Wenn du den Workflow beenden möchtest, verwende explizit die Phrase:
|
|
"Der Workflow kann jetzt beendet werden, weil die Benutzeranfrage vollständig beantwortet wurde."
|
|
|
|
BEACHTE: Ein Agent kann mehrmals ausgewählt werden, wenn nötig. Die Nutzung aller verfügbaren Agenten ist KEINE Voraussetzung für die Beendigung des Workflows.
|
|
"""
|
|
|
|
# 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]:
|
|
# Create the agent description
|
|
agent_description = f"""
|
|
# Aufgabe
|
|
Du bist ein spezialisierter Agent vom Typ {agent['type']} mit dem Namen {agent['name']}.
|
|
|
|
# Fähigkeiten
|
|
{agent.get('capabilities', 'Keine spezifischen Fähigkeiten angegeben.')}
|
|
|
|
# Anweisungen
|
|
{agent_instructions}
|
|
|
|
Bitte analysiere den Chatverlauf und die Dateien und beantworte die Anfrage gemäß deiner Rolle.
|
|
|
|
Ausgabeformat:
|
|
[Agent: {agent['name']}]
|
|
Deine Antwort...""" # No trailing newline
|
|
|
|
# Make sure there's no trailing whitespace
|
|
content = agent_description.strip()
|
|
|
|
return {
|
|
"role": "system",
|
|
"content": content
|
|
}
|
|
|
|
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
|
|
"""
|
|
# Normalize the moderator text for case-insensitive matching
|
|
moderator_text_lower = moderator_text.lower()
|
|
|
|
# Check for explicit workflow completion signal
|
|
if "workflow kann jetzt beendet werden" in moderator_text_lower or "beende den workflow" in moderator_text_lower:
|
|
logger.info("Moderator decided to complete the workflow")
|
|
return "WORKFLOW_COMPLETE"
|
|
|
|
# Look for explicit agent selection patterns
|
|
for pattern in ["ich wähle den ", "ich empfehle den ", "ich empfehle, dass "]:
|
|
if pattern in moderator_text_lower:
|
|
# Extract text after the pattern
|
|
text_after_pattern = moderator_text_lower.split(pattern)[1]
|
|
|
|
# Check each agent name against the extracted text
|
|
for agent_id, agent in available_agents.items():
|
|
agent_name_lower = agent["name"].lower()
|
|
|
|
# Check if the agent name appears right after the selection pattern
|
|
if text_after_pattern.startswith(agent_name_lower):
|
|
logger.info(f"Moderator explicitly selected agent: {agent['name']}")
|
|
return agent_id
|
|
|
|
# Direct name checking as fallback
|
|
for agent_id, agent in available_agents.items():
|
|
if agent["name"].lower() in moderator_text_lower:
|
|
logger.info(f"Moderator mentioned agent by name: {agent['name']}")
|
|
return agent_id
|
|
|
|
logger.info("No explicit agent selection found, using fallback selection")
|
|
|
|
# Fallback to the first unused agent
|
|
for agent_id, agent in available_agents.items():
|
|
if not agent["used"]:
|
|
logger.info(f"Selecting first unused agent: {agent['name']}")
|
|
return agent_id
|
|
|
|
# If all agents are used, select the Initialisierung agent
|
|
for agent_id, agent in available_agents.items():
|
|
if agent["type"] == "initialisierung":
|
|
logger.info(f"All agents used, falling back to initialization agent: {agent['name']}")
|
|
return agent_id
|
|
|
|
# Last resort: select the first available agent
|
|
if available_agents:
|
|
first_agent_id = list(available_agents.keys())[0]
|
|
logger.info(f"Fallback to first agent: {available_agents[first_agent_id]['name']}")
|
|
return first_agent_id
|
|
|
|
# No agents available
|
|
logger.warning("No agents available to select")
|
|
return None |