752 lines
No EOL
34 KiB
Python
752 lines
No EOL
34 KiB
Python
import asyncio
|
|
import uuid
|
|
import os
|
|
import json
|
|
import logging
|
|
import base64
|
|
import mimetypes
|
|
from typing import List, Dict, Any, Optional
|
|
from datetime import datetime
|
|
from fastapi import HTTPException
|
|
import configload as configload
|
|
|
|
# Logger konfigurieren
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Konfigurationsdaten laden
|
|
def load_config_data():
|
|
config = configload.load_config()
|
|
result = {
|
|
"application": {
|
|
"debug": config.get('Module_AgentserviceInterface', 'DEBUG'),
|
|
"upload_dir": config.get('Module_AgentserviceInterface', 'UPLOAD_DIR'),
|
|
"results_dir": config.get('Module_AgentserviceInterface', 'RESULTS_DIR'),
|
|
"max_history": config.get('Module_AgentserviceInterface', 'MAX_HISTORY'),
|
|
"ai_provider": config.get('Module_AgentserviceInterface', 'AI_PROVIDER', fallback="openai") # Standard ist OpenAI
|
|
}
|
|
}
|
|
# Debug-Modus einstellen
|
|
if result["application"]["debug"].lower() == "true":
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger.setLevel(logging.DEBUG)
|
|
logger.debug("Debug-Modus aktiviert")
|
|
return result
|
|
|
|
|
|
class AgentService:
|
|
"""
|
|
Service für die Verwaltung und Ausführung von Multi-Agent-Workflows mit verschiedenen Modellen.
|
|
"""
|
|
|
|
def __init__(self, mandate_id: int = None, user_id: int = None):
|
|
"""
|
|
Initialisiert den AgentService.
|
|
|
|
Args:
|
|
mandate_id: ID des aktuellen Mandanten (optional)
|
|
user_id: ID des aktuellen Benutzers (optional)
|
|
"""
|
|
# Mandanten- und Benutzerkontext
|
|
self.mandate_id = mandate_id
|
|
self.user_id = user_id
|
|
|
|
# Konfiguration laden
|
|
self.config = load_config_data()
|
|
|
|
# Verzeichnisse aus der Konfiguration übernehmen
|
|
self.results_dir = self.config["application"]["results_dir"]
|
|
self.upload_dir = self.config["application"]["upload_dir"]
|
|
self.max_history = int(self.config["application"]["max_history"])
|
|
|
|
# AI Provider aus der Konfiguration übernehmen
|
|
self.ai_provider = self.config["application"]["ai_provider"].lower()
|
|
|
|
# Verzeichnisse erstellen
|
|
os.makedirs(self.results_dir, exist_ok=True)
|
|
os.makedirs(self.upload_dir, exist_ok=True)
|
|
|
|
# Connector-Instanzen initialisieren
|
|
if self.ai_provider == "anthropic":
|
|
import connector_aichat_anthropic as service_aichat
|
|
self.service_aichat = service_aichat.ChatService()
|
|
logger.info("Anthropic AI Provider wird verwendet")
|
|
else:
|
|
import connector_aichat_openai as service_aichat
|
|
self.service_aichat = service_aichat.ChatService()
|
|
logger.info("OpenAI AI Provider wird verwendet")
|
|
|
|
import connector_aiweb_webscraping as service_aiscrap
|
|
self.service_aiscrap = service_aiscrap.WebScrapingService()
|
|
|
|
logger.info(f"AgentService initialisiert mit:")
|
|
logger.info(f" - AI Provider: {self.ai_provider}")
|
|
logger.info(f" - Ergebnisverzeichnis: {self.results_dir}")
|
|
logger.info(f" - Upload-Verzeichnis: {self.upload_dir}")
|
|
|
|
# Workflow-Speicher
|
|
self.workflows = {}
|
|
|
|
async def execute_workflow(
|
|
self,
|
|
workflow_id: str,
|
|
prompt: str,
|
|
agents: List[Dict[str, Any]],
|
|
files: List[Dict[str, Any]]
|
|
) -> str:
|
|
"""
|
|
Führt einen Workflow mit den angegebenen Agenten und Dateien aus.
|
|
|
|
Anstatt die Agenten der Reihe nach abzuarbeiten, wird ein AI-Moderator verwendet,
|
|
der die Agenten basierend auf ihren Fähigkeiten und bisherigen Antworten steuert.
|
|
"""
|
|
logger.info(f"Starte Workflow {workflow_id} mit {len(agents)} Agenten und {len(files)} Dateien")
|
|
|
|
# Mandanten- und Benutzerkontext in die Workflow-Daten aufnehmen
|
|
self.workflows[workflow_id] = {
|
|
"id": workflow_id,
|
|
"mandate_id": self.mandate_id,
|
|
"user_id": self.user_id,
|
|
"status": "running",
|
|
"progress": 0.0,
|
|
"started_at": datetime.now().isoformat(),
|
|
"completed_at": None,
|
|
"agent_statuses": {},
|
|
"logs": [],
|
|
"results": []
|
|
}
|
|
|
|
# Log-Eintrag für den Start des Workflows
|
|
self._add_log(workflow_id, "Workflow gestartet", "info")
|
|
self._add_log(workflow_id, f"Verarbeite {len(files)} Dateien...", "info")
|
|
|
|
# Dateikontexte und Inhalte vorbereiten
|
|
file_contexts = []
|
|
file_contents = {}
|
|
|
|
for file in files:
|
|
file_id = file["id"]
|
|
file_name = file["name"]
|
|
file_type = file["type"]
|
|
file_path = file.get("path", "")
|
|
|
|
# Wenn kein Pfad angegeben ist, versuche, ihn aus dem Upload-Verzeichnis abzuleiten
|
|
if not file_path and file_name:
|
|
possible_path = os.path.join(self.upload_dir, file_name)
|
|
if os.path.exists(possible_path):
|
|
file_path = possible_path
|
|
logger.debug(f"Pfad für Datei {file_name} gefunden: {file_path}")
|
|
|
|
file_contexts.append({
|
|
"id": file_id,
|
|
"name": file_name,
|
|
"type": file_type,
|
|
"size": file.get("size", "Unbekannt"),
|
|
"path": file_path
|
|
})
|
|
|
|
# Dateiinhalt lesen, wenn der Pfad verfügbar ist
|
|
if file_path and os.path.exists(file_path):
|
|
try:
|
|
# Text-basierte Dateien direkt lesen
|
|
if file_type == "document":
|
|
# Einfache Textdateien
|
|
if file_name.endswith(('.txt', '.csv', '.md', '.json')):
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
file_contents[file_id] = f.read()
|
|
self._add_log(workflow_id, f"Datei {file_name} gelesen", "info")
|
|
|
|
# Excel-Dateien
|
|
elif file_name.endswith(('.xlsx', '.xls')):
|
|
import pandas as pd
|
|
try:
|
|
df = pd.read_excel(file_path)
|
|
file_contents[file_id] = f"Excel-Datei mit {len(df)} Zeilen und {len(df.columns)} Spalten.\n"
|
|
file_contents[file_id] += f"Spalten: {', '.join(df.columns.tolist())}\n"
|
|
file_contents[file_id] += "Erste 5 Zeilen:\n"
|
|
file_contents[file_id] += df.head(5).to_string()
|
|
self._add_log(workflow_id, f"Excel-Datei {file_name} gelesen", "info")
|
|
except Exception as e:
|
|
self._add_log(workflow_id, f"Fehler beim Lesen der Excel-Datei {file_name}: {str(e)}", "error")
|
|
|
|
# CSV-Dateien
|
|
elif file_name.endswith('.csv'):
|
|
import pandas as pd
|
|
try:
|
|
df = pd.read_csv(file_path)
|
|
file_contents[file_id] = f"CSV-Datei mit {len(df)} Zeilen und {len(df.columns)} Spalten.\n"
|
|
file_contents[file_id] += f"Spalten: {', '.join(df.columns.tolist())}\n"
|
|
file_contents[file_id] += "Erste 5 Zeilen:\n"
|
|
file_contents[file_id] += df.head(5).to_string()
|
|
self._add_log(workflow_id, f"CSV-Datei {file_name} gelesen", "info")
|
|
except Exception as e:
|
|
self._add_log(workflow_id, f"Fehler beim Lesen der CSV-Datei {file_name}: {str(e)}", "error")
|
|
|
|
# PDF-Dateien
|
|
elif file_name.endswith('.pdf'):
|
|
try:
|
|
# Falls PyPDF2 installiert ist
|
|
try:
|
|
from PyPDF2 import PdfReader
|
|
reader = PdfReader(file_path)
|
|
text = ""
|
|
for page in reader.pages:
|
|
text += page.extract_text() + "\n\n"
|
|
file_contents[file_id] = f"PDF mit {len(reader.pages)} Seiten.\nInhalt:\n{text[:2000]}..."
|
|
self._add_log(workflow_id, f"PDF-Datei {file_name} gelesen", "info")
|
|
except ImportError:
|
|
self._add_log(workflow_id, "PyPDF2 nicht installiert. PDF-Inhalt kann nicht extrahiert werden.", "warning")
|
|
file_contents[file_id] = f"PDF-Datei (Inhalt nicht verfügbar, PyPDF2 fehlt)"
|
|
except Exception as e:
|
|
self._add_log(workflow_id, f"Fehler beim Lesen der PDF-Datei {file_name}: {str(e)}", "error")
|
|
|
|
# Andere Dokumenttypen
|
|
else:
|
|
self._add_log(workflow_id, f"Nicht unterstütztes Dokumentformat: {file_name}", "warning")
|
|
file_contents[file_id] = f"Dateiinhalt nicht verfügbar (Nicht unterstütztes Format)"
|
|
|
|
# Bilddateien werden nicht direkt gelesen, nur Metadaten gespeichert
|
|
elif file_type == "image":
|
|
file_contents[file_id] = f"Bilddatei: {file_name} (Inhalt nicht als Text verfügbar)"
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Lesen der Datei {file_name}: {str(e)}")
|
|
self._add_log(workflow_id, f"Fehler beim Lesen der Datei {file_name}: {str(e)}", "error")
|
|
else:
|
|
if file_path:
|
|
self._add_log(workflow_id, f"Datei {file_name} nicht gefunden: {file_path}", "warning")
|
|
else:
|
|
self._add_log(workflow_id, f"Kein Pfad für Datei {file_name} verfügbar", "warning")
|
|
file_contents[file_id] = f"Dateiinhalt nicht verfügbar"
|
|
|
|
# Erstelle einen Kontext mit Dateiliste und Inhalten für leichteren Zugriff
|
|
file_context_text = "Verfügbare Dateien:\n" + "\n".join([f"- {file['name']} ({file['type']}, {file['size']})" for file in file_contexts])
|
|
|
|
# Füge Dateiinhalte hinzu (mit Längenbegrenzung)
|
|
for file_id, content in file_contents.items():
|
|
file_name = next((f['name'] for f in file_contexts if f['id'] == file_id), "Unbekannte Datei")
|
|
file_context_text += f"\n\n==== DATEIINHALT: {file_name} ====\n"
|
|
|
|
# Begrenze den Inhalt, um Token-Limits zu respektieren
|
|
max_content_length = 5000 # Anpassen je nach Anzahl der Dateien und Umfang
|
|
if len(content) > max_content_length:
|
|
file_context_text += content[:max_content_length] + "...\n[Dateiinhalt gekürzt aus Platzgründen]"
|
|
else:
|
|
file_context_text += content
|
|
|
|
self.workflows[workflow_id]["progress"] = 0.1
|
|
|
|
# Initialisiere den Chatverlauf für den Agenten-Dialog
|
|
chat_history = []
|
|
|
|
# Erstelle das Nachrichtenobjekt für die initialen Dateien und den Prompt
|
|
message_content = self.service_aichat.prepare_file_message_content(prompt, file_contexts)
|
|
|
|
# Füge Dateien als Base64-Anhänge hinzu
|
|
for file in file_contexts:
|
|
if file["path"] and os.path.exists(file["path"]):
|
|
try:
|
|
# Datei als Base64 codieren
|
|
with open(file["path"], "rb") as f:
|
|
file_data = f.read()
|
|
base64_data = base64.b64encode(file_data).decode('utf-8')
|
|
|
|
# MIME-Typ bestimmen
|
|
mime_type, _ = mimetypes.guess_type(file["path"])
|
|
if not mime_type:
|
|
mime_type = "application/octet-stream"
|
|
|
|
# Füge die Datei als Anhang hinzu
|
|
message_content.append({
|
|
"type": "file",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": mime_type,
|
|
"data": base64_data
|
|
}
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Hinzufügen der Datei {file['name']} als Anhang: {str(e)}")
|
|
|
|
# Nachrichtenobjekt erstellen
|
|
initial_message = {
|
|
"role": "user",
|
|
"content": message_content
|
|
}
|
|
|
|
# Initialen Prompt zum Chatverlauf hinzufügen
|
|
chat_history.append(initial_message)
|
|
|
|
# Initialisiere die verfügbaren Agenten mit ihren Fähigkeiten
|
|
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
|
|
}
|
|
|
|
# Initialisiere den Status
|
|
self.workflows[workflow_id]["agent_statuses"][agent_id] = "pending"
|
|
|
|
# Initialisiere die Moderator-Rolle - Fester Teil
|
|
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
|
|
moderator_system_prompt = moderator_prompt_base + "\n" + agents_description + "\n" + moderator_prompt_end
|
|
|
|
# Starte den Workflow mit dem Moderator
|
|
self._add_log(workflow_id, "Starte Agenten-Tischrunde mit Moderator", "info")
|
|
|
|
# Maximale Anzahl der Runden zur Vermeidung endloser Schleifen
|
|
max_rounds = 12
|
|
current_round = 0
|
|
workflow_complete = False
|
|
|
|
while current_round < max_rounds and not workflow_complete:
|
|
current_round += 1
|
|
self._add_log(workflow_id, f"Starte Runde {current_round}", "info")
|
|
|
|
# Der Moderator wählt den nächsten Agenten aus
|
|
moderator_prompt = {
|
|
"role": "system",
|
|
"content": moderator_system_prompt
|
|
}
|
|
|
|
# Kopie des Chatverlaufs für den Moderator erstellen
|
|
moderator_chat = [moderator_prompt] + chat_history[-self.max_history:]
|
|
|
|
# Füge eine Zusammenfassung der verfügbaren Agenten hinzu
|
|
agent_info = "Verfügbare Agenten:\n"
|
|
for agent_id, agent in available_agents.items():
|
|
status = "✓ Bereits verwendet" if agent["used"] else "✗ Noch nicht verwendet"
|
|
agent_info += f"- {agent['name']} (Typ: {agent['type']}): {agent['capabilities']}\n Status: {status}\n"
|
|
|
|
moderator_chat.append({
|
|
"role": "system",
|
|
"content": agent_info + "\nWähle den nächsten Agenten aus oder beende den Workflow, wenn die Aufgabe erfüllt ist."
|
|
})
|
|
|
|
# Moderator trifft die Entscheidung
|
|
try:
|
|
moderator_decision = await self.service_aichat.call_api(moderator_chat)
|
|
moderator_text = moderator_decision["choices"][0]["message"]["content"]
|
|
|
|
# Füge die Entscheidung des Moderators zum Chatverlauf hinzu
|
|
chat_history.append({
|
|
"role": "assistant",
|
|
"content": f"[Moderator] {moderator_text}"
|
|
})
|
|
|
|
# Log der Moderator-Entscheidung
|
|
self._add_log(workflow_id, f"Moderator-Entscheidung: {moderator_text[:100]}...", "info")
|
|
|
|
# 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"]):
|
|
self._add_log(workflow_id, "Moderator hat den Workflow beendet", "success")
|
|
workflow_complete = True
|
|
break
|
|
|
|
# Extrahiere den ausgewählten Agenten
|
|
selected_agent_id = None
|
|
|
|
# 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:
|
|
selected_agent_id = agent_id
|
|
break
|
|
|
|
if not selected_agent_id:
|
|
self._add_log(workflow_id, "Moderator konnte keinen Agenten identifizieren", "warning")
|
|
# Wähle den ersten nicht verwendeten Agenten
|
|
for agent_id, agent in available_agents.items():
|
|
if not agent["used"]:
|
|
selected_agent_id = agent_id
|
|
break
|
|
|
|
# Wenn alle Agenten bereits verwendet wurden, wähle den Initialisierungs-Agenten
|
|
if not selected_agent_id:
|
|
for agent_id, agent in available_agents.items():
|
|
if agent["type"] == "initialisierung":
|
|
selected_agent_id = agent_id
|
|
break
|
|
|
|
# Als letztes Mittel wähle einfach den ersten Agenten
|
|
if not selected_agent_id and available_agents:
|
|
selected_agent_id = list(available_agents.keys())[0]
|
|
|
|
if selected_agent_id:
|
|
# Agenten aus der Liste markieren
|
|
selected_agent = available_agents[selected_agent_id]
|
|
selected_agent["used"] = True
|
|
|
|
# Agenten-Status aktualisieren
|
|
self.workflows[workflow_id]["agent_statuses"][selected_agent_id] = "running"
|
|
self._add_log(
|
|
workflow_id,
|
|
f"Agent '{selected_agent['name']}' beginnt mit der Verarbeitung...",
|
|
"start",
|
|
selected_agent_id,
|
|
selected_agent['name']
|
|
)
|
|
|
|
# Agent-spezifische Anweisungen erstellen
|
|
agent_instructions = self._get_agent_instructions(selected_agent["type"])
|
|
|
|
# Agent-Prompt erstellen
|
|
agent_prompt = {
|
|
"role": "system",
|
|
"content": f"""
|
|
# Aufgabe
|
|
Du bist ein spezialisierter Agent vom Typ {selected_agent['type']} mit dem Namen {selected_agent['name']}.
|
|
|
|
{agent_instructions}
|
|
|
|
Bitte analysiere den Chatverlauf und die Dateien und beantworte die Anfrage gemäß deiner Rolle.
|
|
|
|
Ausgabeformat:
|
|
[Agent: {selected_agent['name']}]
|
|
Deine Antwort...
|
|
"""
|
|
}
|
|
|
|
# Kopie des Chatverlaufs für den Agenten erstellen
|
|
agent_chat = [agent_prompt] + chat_history[-self.max_history:]
|
|
|
|
# Falls der Agent ein Webscraper ist und Scraping notwendig ist
|
|
if selected_agent["type"] == "scraper":
|
|
self._add_log(workflow_id, "Führe Web-Scraping durch...", "info", selected_agent_id, selected_agent["name"])
|
|
web_data = await self.service_aiscrap.scrape_web_data(prompt)
|
|
if web_data:
|
|
agent_chat.append({
|
|
"role": "system",
|
|
"content": f"# Gescrapte Web-Daten\n{web_data}"
|
|
})
|
|
self._add_log(workflow_id, "Web-Scraping abgeschlossen", "info", selected_agent_id, selected_agent["name"])
|
|
|
|
# Agent führt seinen Teil aus
|
|
try:
|
|
agent_response = await self.service_aichat.call_api(agent_chat)
|
|
agent_text = agent_response["choices"][0]["message"]["content"]
|
|
|
|
# Füge die Antwort des Agenten zum Chatverlauf hinzu
|
|
chat_history.append({
|
|
"role": "assistant",
|
|
"content": agent_text
|
|
})
|
|
|
|
# Agent-Ergebnis erstellen
|
|
result = self._create_agent_result(
|
|
workflow_id,
|
|
selected_agent,
|
|
len(self.workflows[workflow_id]["results"]),
|
|
prompt,
|
|
file_contexts,
|
|
agent_text
|
|
)
|
|
self.workflows[workflow_id]["results"].append(result)
|
|
|
|
self._add_log(
|
|
workflow_id,
|
|
f"Agent '{selected_agent['name']}' hat die Verarbeitung abgeschlossen",
|
|
"complete",
|
|
selected_agent_id,
|
|
selected_agent["name"]
|
|
)
|
|
|
|
# Agenten-Status aktualisieren
|
|
self.workflows[workflow_id]["agent_statuses"][selected_agent_id] = "completed"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei der Ausführung von Agent '{selected_agent['name']}': {str(e)}")
|
|
self._add_log(
|
|
workflow_id,
|
|
f"Fehler bei der Ausführung: {str(e)}",
|
|
"error",
|
|
selected_agent_id,
|
|
selected_agent["name"]
|
|
)
|
|
self.workflows[workflow_id]["agent_statuses"][selected_agent_id] = "failed"
|
|
|
|
# Füge die Fehlermeldung zum Chatverlauf hinzu
|
|
chat_history.append({
|
|
"role": "assistant",
|
|
"content": f"[Fehler bei Agent '{selected_agent['name']}']: {str(e)}"
|
|
})
|
|
else:
|
|
self._add_log(workflow_id, "Kein Agent ausgewählt. Beende Workflow.", "warning")
|
|
workflow_complete = True
|
|
|
|
# Fortschritt aktualisieren
|
|
self.workflows[workflow_id]["progress"] = min(0.9, 0.1 + (current_round / max_rounds) * 0.8)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler in der Moderator-Phase: {str(e)}")
|
|
self._add_log(workflow_id, f"Fehler in der Moderator-Phase: {str(e)}", "error")
|
|
break
|
|
|
|
# Workflow abschließen
|
|
if workflow_complete:
|
|
self.workflows[workflow_id]["status"] = "completed"
|
|
self._add_log(workflow_id, "Workflow erfolgreich beendet", "success")
|
|
elif current_round >= max_rounds:
|
|
self.workflows[workflow_id]["status"] = "completed"
|
|
self._add_log(workflow_id, f"Workflow nach {max_rounds} Runden automatisch beendet", "info")
|
|
else:
|
|
self.workflows[workflow_id]["status"] = "failed"
|
|
self._add_log(workflow_id, "Workflow mit Fehlern beendet", "error")
|
|
|
|
self.workflows[workflow_id]["progress"] = 1.0
|
|
self.workflows[workflow_id]["completed_at"] = datetime.now().isoformat()
|
|
|
|
# Speichere Ergebnisse in Datei für spätere Verwendung
|
|
self._save_workflow_results(workflow_id)
|
|
|
|
return workflow_id
|
|
|
|
def _get_agent_instructions(self, agent_type: str) -> str:
|
|
"""
|
|
Gibt agententypspezifische Anweisungen zurück, die aus der agents.json geladen werden.
|
|
Falls die Datei nicht existiert oder der Agententyp nicht gefunden wird,
|
|
wird ein Standardagent zurückgegeben.
|
|
"""
|
|
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}")
|
|
return self._get_default_agent_instructions()
|
|
|
|
# Datei lesen
|
|
with open(agents_file, 'r', encoding='utf-8') as f:
|
|
agents_data = json.load(f)
|
|
|
|
# Nach dem Agententyp suchen
|
|
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")
|
|
return instructions
|
|
|
|
# Wenn kein passender Agent gefunden wurde, Standardanweisungen verwenden
|
|
logger.warning(f"Keine Anweisungen für Agent-Typ '{agent_type}' in agents.json gefunden")
|
|
return self._get_default_agent_instructions()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Laden der Agent-Anweisungen aus agents.json: {e}")
|
|
return self._get_default_agent_instructions()
|
|
|
|
def _get_default_agent_instructions(self) -> 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 _add_log(
|
|
self,
|
|
workflow_id: str,
|
|
message: str,
|
|
log_type: str,
|
|
agent_id: Optional[str] = None,
|
|
agent_name: Optional[str] = None
|
|
) -> None:
|
|
"""Fügt einen Protokolleintrag zum Workflow hinzu"""
|
|
log_entry = {
|
|
"id": f"log_{uuid.uuid4()}",
|
|
"mandate_id": self.mandate_id,
|
|
"user_id": self.user_id,
|
|
"message": message,
|
|
"type": log_type,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"agent_id": agent_id,
|
|
"agent_name": agent_name
|
|
}
|
|
|
|
workflow = self.workflows.get(workflow_id)
|
|
if workflow:
|
|
workflow["logs"].append(log_entry)
|
|
logger.info(f"Workflow {workflow_id}: {message}")
|
|
|
|
def _create_agent_result(
|
|
self,
|
|
workflow_id: str,
|
|
agent: Dict[str, Any],
|
|
index: int,
|
|
prompt: str,
|
|
file_contexts: List[Dict[str, Any]],
|
|
content: str
|
|
) -> Dict[str, Any]:
|
|
"""Erstellt ein Ergebnisobjekt basierend auf dem Agententyp und der API-Antwort"""
|
|
agent_type = agent["type"]
|
|
agent_id = agent["id"]
|
|
agent_name = agent["name"]
|
|
|
|
# Grundlegende Ergebnisstruktur
|
|
result = {
|
|
"id": f"result_{workflow_id}_{index}",
|
|
"mandate_id": self.mandate_id,
|
|
"user_id": self.user_id,
|
|
"agent_id": agent_id,
|
|
"agent_name": agent_name,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"type": "text", # Standardtyp
|
|
"metadata": {
|
|
"files_processed": [file["name"] for file in file_contexts],
|
|
"prompt": prompt
|
|
}
|
|
}
|
|
|
|
# Titel und Inhalt basierend auf dem Agententyp anpassen
|
|
if agent_type == "analyzer":
|
|
result.update({
|
|
"title": "Datenanalyse-Ergebnis",
|
|
"content": content,
|
|
})
|
|
elif agent_type == "visualizer":
|
|
result.update({
|
|
"title": "Visualisierungsvorschlag",
|
|
"content": content,
|
|
"type": "chart" # Auch wenn kein echtes Diagramm, markieren wir es als solches
|
|
})
|
|
elif agent_type == "writer":
|
|
result.update({
|
|
"title": "Zusammenfassung und Empfehlungen",
|
|
"content": content,
|
|
})
|
|
elif agent_type == "scraper":
|
|
result.update({
|
|
"title": "Web-Recherche Ergebnisse",
|
|
"content": content,
|
|
})
|
|
elif agent_type == "initialisierung":
|
|
result.update({
|
|
"title": "Direkte Antwort",
|
|
"content": content,
|
|
})
|
|
elif agent_type == "organisator":
|
|
result.update({
|
|
"title": "Aufgabenstrukturierung",
|
|
"content": content,
|
|
})
|
|
elif agent_type == "entwickler":
|
|
result.update({
|
|
"title": "Code und Ausführungsergebnisse",
|
|
"content": content,
|
|
})
|
|
else:
|
|
result.update({
|
|
"title": f"Ergebnis von {agent_name}",
|
|
"content": content,
|
|
})
|
|
|
|
return result
|
|
|
|
def _save_workflow_results(self, workflow_id: str) -> None:
|
|
"""Speichert die Workflow-Ergebnisse in einer Datei"""
|
|
workflow = self.workflows.get(workflow_id)
|
|
if workflow:
|
|
try:
|
|
file_path = os.path.join(self.results_dir, f"workflow_{workflow_id}.json")
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(workflow, f, indent=2, ensure_ascii=False)
|
|
logger.info(f"Workflow-Ergebnisse gespeichert: {file_path}")
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Speichern der Workflow-Ergebnisse: {e}")
|
|
|
|
def get_workflow_status(self, workflow_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Gibt den Status eines Workflows zurück"""
|
|
workflow = self.workflows.get(workflow_id)
|
|
if not workflow:
|
|
return None
|
|
|
|
return {
|
|
"id": workflow["id"],
|
|
"mandate_id": workflow.get("mandate_id"),
|
|
"user_id": workflow.get("user_id"),
|
|
"status": workflow["status"],
|
|
"progress": workflow["progress"],
|
|
"started_at": workflow["started_at"],
|
|
"completed_at": workflow["completed_at"],
|
|
"agent_statuses": workflow["agent_statuses"]
|
|
}
|
|
|
|
def get_workflow_logs(self, workflow_id: str) -> Optional[List[Dict[str, Any]]]:
|
|
"""Gibt die Protokolle eines Workflows zurück"""
|
|
workflow = self.workflows.get(workflow_id)
|
|
if not workflow:
|
|
return None
|
|
|
|
return workflow["logs"]
|
|
|
|
def get_workflow_results(self, workflow_id: str) -> Optional[List[Dict[str, Any]]]:
|
|
"""Gibt die Ergebnisse eines Workflows zurück"""
|
|
workflow = self.workflows.get(workflow_id)
|
|
if not workflow:
|
|
return None
|
|
|
|
return workflow["results"]
|
|
|
|
async def close(self):
|
|
"""Schließt die HTTP-Clients beim Beenden der Anwendung"""
|
|
await self.service_aichat.close()
|
|
await self.service_aiscrap.close()
|
|
|
|
|
|
# Singleton-Factory für AgentService-Instanzen pro Kontext
|
|
_agent_service_instances = {}
|
|
|
|
def get_agentservice_interface(mandate_id: int = None, user_id: int = None) -> AgentService:
|
|
"""
|
|
Gibt eine AgentService-Instanz für den angegebenen Kontext zurück.
|
|
Wiederverwendet bestehende Instanzen.
|
|
"""
|
|
context_key = f"{mandate_id}_{user_id}"
|
|
if context_key not in _agent_service_instances:
|
|
_agent_service_instances[context_key] = AgentService(mandate_id, user_id)
|
|
return _agent_service_instances[context_key] |