133 lines
No EOL
4.2 KiB
Python
133 lines
No EOL
4.2 KiB
Python
"""
|
|
User-Agent für die Interaktion mit dem Benutzer im Agentservice.
|
|
Repräsentiert den menschlichen Benutzer im Workflow und ermöglicht die Kommunikation.
|
|
Angepasste Version für das neue User-Input-Handling.
|
|
"""
|
|
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from modules.agentservice_base import BaseAgent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class UserAgent(BaseAgent):
|
|
"""Agent für die Interaktion mit dem Benutzer"""
|
|
|
|
_instance = None
|
|
|
|
@classmethod
|
|
def get_instance(cls):
|
|
"""Gibt eine Singleton-Instanz zurück"""
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
"""Initialisiert den User-Agenten"""
|
|
super().__init__()
|
|
self.id = "user_agent"
|
|
self.name = "User Agent"
|
|
self.type = "user"
|
|
self.description = "Repräsentiert den Benutzer im Workflow"
|
|
self.capabilities = "Beantwortung von Fragen, Bereitstellung zusätzlicher Informationen, Entscheidungsfindung"
|
|
|
|
def get_agent_info(self) -> Dict[str, Any]:
|
|
"""
|
|
Gibt Informationen über den User-Agenten zurück.
|
|
|
|
Returns:
|
|
Dict mit Agenten-Informationen
|
|
"""
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"type": self.type,
|
|
"description": self.description,
|
|
"capabilities": self.capabilities,
|
|
"used": False,
|
|
"last_result_status": None
|
|
}
|
|
|
|
async def process_message(self, message: Dict[str, Any], context: Dict[str, Any] = None) -> Dict[str, Any]:
|
|
"""
|
|
Verarbeitet eine Nachricht und bereitet die Frontend-Interaktion vor.
|
|
|
|
Args:
|
|
message: Die zu verarbeitende Nachricht
|
|
context: Zusätzlicher Kontext
|
|
|
|
Returns:
|
|
Die Nachricht an den Benutzer
|
|
"""
|
|
# Task aus dem Kontext extrahieren (falls vorhanden)
|
|
task = context.get("task", "") if context else ""
|
|
|
|
# Aus der Nachricht übernehmen, falls kein Kontext
|
|
if not task and message and "content" in message:
|
|
task = message["content"]
|
|
|
|
# Explizites Moderator-Prefix für die Nachricht
|
|
moderator_prefix = "[Moderator zu User Agent] "
|
|
formatted_task = moderator_prefix + task.strip()
|
|
|
|
# Erweiterte Nachricht mit expliziten Flags für das Frontend
|
|
user_request = {
|
|
"role": "assistant",
|
|
"content": formatted_task,
|
|
"agent_type": "moderator", # Explizit als Moderator markieren
|
|
"agent_id": "moderator",
|
|
"agent_name": "Moderator",
|
|
}
|
|
|
|
# Log-Eintrag
|
|
logger.info(f"User-Agent wird aufgerufen, Workflow wartet auf Benutzereingabe: {task[:50]}...")
|
|
|
|
return user_request
|
|
|
|
def extract_status(self, content: str) -> tuple:
|
|
"""
|
|
Bei User-Agent wird kein Status extrahiert, da der Benutzer keinen Status angibt.
|
|
|
|
Args:
|
|
content: Inhalt der Nachricht
|
|
|
|
Returns:
|
|
Tuple mit (Originalinhalt, "USER_INPUT")
|
|
"""
|
|
# User-Agent setzt immer Status "USER_INPUT"
|
|
return content, "USER_INPUT"
|
|
|
|
def format_user_response(self, message: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Formatiert eine Benutzerantwort für die Workflow-Verarbeitung.
|
|
|
|
Args:
|
|
message: Die vom Benutzer gesendete Nachricht
|
|
|
|
Returns:
|
|
Die formatierte Nachricht
|
|
"""
|
|
# Basisformat für die Antwort
|
|
formatted_response = {
|
|
"role": "user",
|
|
"content": message.get("content", ""),
|
|
"agent_type": "user",
|
|
"agent_id": self.id,
|
|
"agent_name": self.name
|
|
}
|
|
|
|
# Dateien übernehmen, falls vorhanden
|
|
if "documents" in message:
|
|
formatted_response["documents"] = message["documents"]
|
|
|
|
return formatted_response
|
|
|
|
# Singleton-Instanz
|
|
_user_agent = None
|
|
|
|
def get_user_agent():
|
|
"""Gibt eine Singleton-Instanz des User-Agenten zurück"""
|
|
global _user_agent
|
|
if _user_agent is None:
|
|
_user_agent = UserAgent()
|
|
return _user_agent |