From 14f820e27bc1e6c7f8b97dc87a223c2b4ec22a0e Mon Sep 17 00:00:00 2001 From: valueon Date: Wed, 26 Mar 2025 01:42:28 +0100 Subject: [PATCH] MVP1.0.0 --- changelog.txt | 44 +- gwserver/_database_gateway/users.json | 4 +- gwserver/modules/agentservice_interface.py | 820 +++++++++++++++---- gwserver/modules/agentservice_part_agents.py | 107 ++- gwserver/routes/workflows.py | 103 ++- 5 files changed, 881 insertions(+), 197 deletions(-) diff --git a/changelog.txt b/changelog.txt index 7fdd32fd..c4f3e5bc 100644 --- a/changelog.txt +++ b/changelog.txt @@ -1,27 +1,43 @@ ....................... TASKS +Agentenauswahlfenster einfacher und mit klick auf Namen +----------------------- OPEN + + +DOKUS +Doku des Systems für Investoren (Hi-level Struktur, Integrationsfähigkeit und Skalierbarkeit) +Doku des Systems für Code Integration +Release Notes (was kann das Teil) +Log der Anpassungen +Systemarchitektur (Grundsätze der Architektur, Komponenten und deren Aufbau) + +----------------------- DONE + + +Die Buttons "Workflow starten" und "Zurücksetzen" haben keinen Rahmen. Ist hier ggf. die Style Class falsch oder nicht appliziert? + +Anpassung des Visuals "Ausführung & Ergebnisse": +- Das Ausführungsprotokoll so belassen. Einen Button rechts von den anderen zwei Buttons (alle anzeigen / Details zuklappen) ergänzen, für dies mit dem Ausführungsprotokollfenster: toggle function collapse and restore +- Die Bereiche "Multi-Agent-Chat" und "Ergebnisse" machen so keinen Sinn. Diese beiden Bereiche bitte zusammenlegen in einen grossen Bereich mit dem Namen "Multi-Agent Chat Area". Dort laufend die Messages der Agenten in einer HTML-Ansicht der Messages protokollieren. Jeweils der Name des Agenten im Titel und darunter seine Message. Die letzte Message soll aufgeklappt sein, alle früheren sollen jeweils zugeklappt sein, aber durch den User soll ein toggle pro Message möglich sein, um die Details zu sehen. + + +Kannst Du den Ablauf des Agenten-Chats wie folgt optimieren: +- Bei jedem Chat einen "User Agent" mit dem Namen des eingelogten Benutzers ergänzen. Wenn etwas im Chat nicht klar ist, oder zusätzliche Informationen nötig sind, so fragt er den User Agent. Auch bevor er den Chat beendet, fragt er den User Agent, ob dieser einverstanden ist. +- Wenn der User Agent eine Anfrage erhält, so kann er direkt unter der Chat History im Bereich ereiche "Multi-Agent-Chat" seinen Text in einem mehrzeiligen Textfeld erfassen. Er kann auch zusätzliche Files hochladen. Wenn er "Enter" drückt, werden die zusätzlichen Daten mit den ergänzten Files zur Message ergänzt, das Eingabefenster verschwindet wieder und der Moderator führt den Chat fort. Immer nach einer Benutzereingabe startet der Zähler wieder bei Runde 1. + +Statistik ergänzen: Kannst Du bitte rechtsbündig neben dem Titel des "Ausführungsprotokolls" laufend die Statistik nachführen, wieviele kBytes (kB) Daten über den Connector zum AI-Modell gesendet wurden (dies ist die Datengrösse des Message-Objektes) und wieviele kB an Messages zurückgeliefert wurden. Diese angabe pro Workflow-Durchlauf, also immer beim Start eines neuen Workflows wird der Zähler auf 0 gesetzt. In diesem Format: "^ 250k v 1'250k ", v und ^ durch Pfeile ersetzt. + + +In den Einstellungen des Frontends soll die Sprache des aktiven benutzers gemäss den Listenoptionen in den "...model.py" angepasst werden können. die sprache gilt dann auch für die Attributnamen in einem Formularfeld im "generic-entity.js". eine sprachänderung zieht somit eine anpassung des Users über das API nach sich, indem die Sprache in der Datenbank angepasst wird. + kannst du die ausführungsprotokollierung anpassen? das protokoll soll laufend anzeigen, welcher assistent welches resultat produziert hat und welcher assistent aktuell am arbeiten ist. Prozentzahlen sind keine nötig, diese machen keinen sinn. das polling so beibehalten, aber wenn keine neuen Daten bereitstelen, dann beim letzten Timestamp einfach laufend "." ergänzen, bis die nächste Meldung ausgegeben wird. hast du alle daten, um dies im frontend und im backend anzupassen? Im Ausführungsprotokoll pro Eintrag nur den Titel zeigen und die Details zwar ins Protokoll nehmen, aber ausblenden. Der Benutzer kann dann im Protokoll die zugeklappten Texte aufklappen, um die gewünschten Details gezielt zu sehen. Im Front-End beim Workflow-Modul bitte das Ausführungsprotokoll-Fenster dynamisch in der Grösse anpassbar machen. in der Breite und der Höhe. Dasselbe für das Ergebnis-Fenster. Zudem die Ansicht so gestalten, dass die Fensterteile "Workflow-Konfiguration" und "Ausführung & Ergebnisse" ein- und ausgeblendet werden können, damit jeweils ein Teil die komplette Arbeitsfläche verwenden kann, weil dort viel Text stehen wird. Dies ist für den Benutzer besser. - - ------------------------ OPEN - -TOKENS!!! - Bei Folgeanfragen immer nur den letzten Input mitnehmen oder anders optimieren. AI fragen... - -Erweiterete Parameter aus config.ini einbinden in die Module - -Chat mit Instant message - auch inputs geben während der ausführung - -In den Einstellungen des Frontends soll die Sprache des aktiven benutzers gemäss den Listenoptionen in den "...model.py" angepasst werden können. die sprache gilt dann auch für die Attributnamen in einem Formularfeld im "generic-entity.js". eine sprachänderung zieht somit eine anpassung des Users über das API nach sich, indem die Sprache in der Datenbank angepasst wird. - ------------------------ DONE - nun zu diesem zentralen modul. ich hätte gern, dass die daten als tabellen dargestellt und bearbeitet werden können. für view, add, modify, delete jeweils icon pro datensatz ganz links und zuoberst im header ein "new item" symbol oder text, mach einen vorschlag. ist es möglich, eine checkbox pro datensatz zu machen, um mehrere elemente auszuwählen und oben an der tabelle icons zu haben für mehrfach delete? diff --git a/gwserver/_database_gateway/users.json b/gwserver/_database_gateway/users.json index add197c3..a6611ccf 100644 --- a/gwserver/_database_gateway/users.json +++ b/gwserver/_database_gateway/users.json @@ -2,8 +2,8 @@ { "mandate_id": 1, "username": "admin", - "email": "admin@example.com", - "full_name": "Administrator", + "email": "p.motsch@valueon.ch", + "full_name": "Patrick Motsch | ValueOn AG", "disabled": false, "language": "de", "privilege": "sysadmin", diff --git a/gwserver/modules/agentservice_interface.py b/gwserver/modules/agentservice_interface.py index 4ab7a4a1..64b4f214 100644 --- a/gwserver/modules/agentservice_interface.py +++ b/gwserver/modules/agentservice_interface.py @@ -2,7 +2,8 @@ import asyncio import uuid import os import logging -from typing import List, Dict, Any, Optional +import json +from typing import List, Dict, Any, Optional, Tuple from datetime import datetime from fastapi import HTTPException import configload as configload @@ -87,6 +88,9 @@ class AgentService: # Workflow-Speicher self.workflows = {} + + # Statistiken für die Datenmengen + self.data_stats = {} async def execute_workflow( self, @@ -117,7 +121,12 @@ class AgentService: "completed_at": None, "agent_statuses": {}, "logs": [], - "results": [] + "results": [], + # Statistik für diesen Workflow initialisieren + "data_stats": { + "sent_bytes": 0, + "received_bytes": 0 + } } # Log-Eintrag für den Start des Workflows @@ -158,16 +167,37 @@ class AgentService: # Initialen Prompt zum Chatverlauf hinzufügen chat_history.append(initial_message) + # Datenstatistik aktualisieren - Schätzung der gesendeten Bytes + message_size = len(json.dumps(initial_message)) + self.workflows[workflow_id]["data_stats"]["sent_bytes"] += message_size + # Initialisiere die verfügbaren Agenten mit ihren Fähigkeiten available_agents = agents.initialize_agents(agents_list) + # Füge den User-Agent hinzu + user_info = await self._get_user_info(self.user_id) + user_name = user_info.get("full_name") or user_info.get("username") or f"Benutzer {self.user_id}" + + available_agents["user_agent"] = { + "id": "user_agent", + "name": "User Agent", + "full_name": user_name, + "type": "user", + "capabilities": "Beantwortung von Fragen, Bereitstellung zusätzlicher Informationen, Entscheidungsfindung", + "description": f"Repräsentiert den Benutzer {user_name}", + "used": False + } + # Initialisiere den Status für jeden Agenten for agent_id in available_agents: self.workflows[workflow_id]["agent_statuses"][agent_id] = "pending" - - # Moderator-Prompt erstellen - moderator_system_prompt = agents.get_moderator_prompt(available_agents) - + + # Speichere wichtige Daten im Workflow-Objekt für die spätere Fortsetzung + self.workflows[workflow_id]["chat_history"] = chat_history + self.workflows[workflow_id]["available_agents"] = available_agents + self.workflows[workflow_id]["file_contexts"] = file_contexts + self.workflows[workflow_id]["file_contents"] = file_contents + # Starte den Workflow mit dem Moderator self._add_log(workflow_id, f"Starte Agenten-Tischrunde mit Moderator und {len(available_agents)} Agenten", "info") @@ -175,180 +205,317 @@ class AgentService: max_rounds = 12 current_round = 0 workflow_complete = False + waiting_for_user_input = False + # Hauptschleife für den Workflow while current_round < max_rounds and not workflow_complete: current_round += 1 + self.workflows[workflow_id]["current_round"] = current_round 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_chat = await self._sanitize_messages(moderator_chat) - moderator_decision = await self.service_aichat.call_api(moderator_chat) - moderator_text = moderator_decision["choices"][0]["message"]["content"] - logger.debug(f"Full moderator decision text: {moderator_text}") - - # 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}", "info") - - # Finde den nächsten zu verwendenden Agenten - selected_agent_id = agents.find_next_agent(moderator_text, available_agents) - - # Prüfe, ob der Workflow beendet werden soll - if selected_agent_id == "WORKFLOW_COMPLETE": - self._add_log(workflow_id, "Moderator hat den Workflow beendet", "success") - workflow_complete = True - break - - # Prüfe, ob ein Agent ausgewählt wurde - if not selected_agent_id: - self._add_log(workflow_id, "Kein Agent ausgewählt. Beende Workflow.", "warning") - workflow_complete = True - continue - - # 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 = agents.get_agent_instructions(selected_agent["type"], selected_agent, file_contexts) - - # Agent-Prompt erstellen - agent_prompt = agents.create_agent_prompt(selected_agent, agent_instructions) - - # 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: - # Ensure web_data has no trailing whitespace - web_data = web_data.strip() if isinstance(web_data, str) else web_data - agent_chat.append({ - "role": "system", - "content": f"# Gescrapte Web-Daten\n{web_data}".strip() - }) - self._add_log(workflow_id, "Web-Scraping abgeschlossen", "info", - selected_agent_id, selected_agent["name"]) - - # Agent führt seinen Teil aus - try: - # Da wir keine partielle Dateiladung benötigen, können wir den Agenten direkt aufrufen - agent_chat = await self._sanitize_messages(agent_chat) - agent_response = await self.service_aichat.call_api(agent_chat) - agent_text = agent_response["choices"][0]["message"]["content"] - - # Füge die endgültige Antwort des Agenten zum Chatverlauf hinzu - chat_history.append({ - "role": "assistant", - "content": agent_text - }) - - # Agent-Ergebnis erstellen - result = results.create_agent_result( - workflow_id, - selected_agent, - len(self.workflows[workflow_id]["results"]), - prompt, - file_contexts, - agent_text, - self.mandate_id, - self.user_id - ) - 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)}" - }) - - # 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") + # Falls auf eine Benutzereingabe gewartet wird, nicht fortfahren + if waiting_for_user_input: + self._add_log(workflow_id, "Warte auf Benutzereingabe...", "info") + # Warten wir einfach kurz und beenden dann die Schleife, um im nächsten Polling weiterzumachen + await asyncio.sleep(2) break + + # Führe einen Moderator-Zyklus durch + workflow_complete, waiting_for_user_input, selected_agent_id = await self._run_moderator_cycle( + workflow_id, + chat_history, + available_agents, + file_contexts, + file_contents, + is_user_input_continuation=False # Initialer Aufruf, keine Fortsetzung nach Benutzereingabe + ) + + # Fortschritt aktualisieren + self.workflows[workflow_id]["progress"] = min(0.9, 0.1 + (current_round / max_rounds) * 0.8) - # 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() + # Workflow abschließen, wenn nicht auf Benutzereingabe wartend + if not waiting_for_user_input: + 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 results.save_workflow_results(self.workflows, workflow_id, self.results_dir) return workflow_id + + + async def _run_moderator_cycle( + self, + workflow_id: str, + chat_history: List[Dict[str, Any]], + available_agents: Dict[str, Dict[str, Any]], + file_contexts: List[Dict[str, Any]], + file_contents: Dict[str, str], + is_user_input_continuation: bool = False, + user_message: str = None + ) -> Tuple[bool, bool, str]: + """ + Führt einen Moderator-Zyklus durch: Moderator trifft Entscheidung, Agent wird ausgewählt, + und der ausgewählte Agent wird ausgeführt. + + Args: + workflow_id: ID des Workflows + chat_history: Der bisherige Chat-Verlauf + available_agents: Verfügbare Agenten + file_contexts: Dateikontexte + file_contents: Dateiinhalte + is_user_input_continuation: Gibt an, ob dieser Zyklus eine Fortsetzung nach Benutzereingabe ist + user_message: Die Nachricht des Benutzers (falls is_user_input_continuation=True) + + Returns: + Tuple mit (workflow_complete, waiting_for_user_input, selected_agent_id) + """ + # Initialisiere Rückgabewerte + workflow_complete = False + waiting_for_user_input = False + selected_agent_id = None + + # Moderator-Prompt erstellen + base_prompt = agents.get_moderator_prompt(available_agents) + + # Ergänze mit einem Hinweis zur Benutzereingabe, falls zutreffend + if is_user_input_continuation: + moderator_system_prompt = base_prompt + """ + Wichtig: Der User Agent hat soeben geantwortet. Berücksichtige diese Antwort in deiner Entscheidung. + Wähle nun den nächsten Agenten aus oder beende den Workflow, wenn die Aufgabe erfüllt ist. + """ + else: + moderator_system_prompt = base_prompt + + # 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.get('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_chat = await self._sanitize_messages(moderator_chat) + moderator_decision = await self.service_aichat.call_api(moderator_chat) + moderator_text = moderator_decision["choices"][0]["message"]["content"] + logger.debug(f"Full moderator decision text: {moderator_text}") + + # Datenstatistik aktualisieren + request_size = len(json.dumps(moderator_chat)) + response_size = len(json.dumps(moderator_decision)) + self.workflows[workflow_id]["data_stats"]["sent_bytes"] += request_size + self.workflows[workflow_id]["data_stats"]["received_bytes"] += response_size + + # 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}", "info") + + # Finde den nächsten zu verwendenden Agenten + selected_agent_id = agents.find_next_agent(moderator_text, available_agents) + + # Prüfe, ob der Moderator eine Anfrage an den User Agent stellt + if selected_agent_id != "user_agent": # Nur prüfen, wenn nicht explizit der User Agent ausgewählt wurde + is_user_agent_query = self._check_for_user_agent_query(moderator_text) + if is_user_agent_query: + self._add_log(workflow_id, "Moderator stellt eine Frage an den User Agent", "info") + selected_agent_id = "user_agent" + waiting_for_user_input = True + + # Prüfe, ob der Workflow beendet werden soll + if selected_agent_id == "WORKFLOW_COMPLETE": + self._add_log(workflow_id, "Moderator hat den Workflow beendet", "success") + workflow_complete = True + return workflow_complete, waiting_for_user_input, selected_agent_id + + # Prüfe, ob der User-Agent ausgewählt wurde + if selected_agent_id == "user_agent": + self._add_log(workflow_id, "Warte auf Benutzereingabe für den User Agent", "info", "user_agent", "User Agent") + # Markiere, dass wir auf eine Benutzereingabe warten + waiting_for_user_input = True + self.workflows[workflow_id]["waiting_for_user"] = True + + # Benutzeranfrage zum Chatverlauf hinzufügen + chat_history.append({ + "role": "assistant", + "content": f"[Moderator zu User Agent] {moderator_text}" + }) + + # Chat-Verlauf im Workflow aktualisieren + self.workflows[workflow_id]["chat_history"] = chat_history + + # Workflow-Status speichern + results.save_workflow_results(self.workflows, workflow_id, self.results_dir) + + return workflow_complete, waiting_for_user_input, selected_agent_id + + # Prüfe, ob ein anderer Agent ausgewählt wurde + if not selected_agent_id: + self._add_log(workflow_id, "Kein Agent ausgewählt. Beende Workflow.", "warning") + workflow_complete = True + return workflow_complete, waiting_for_user_input, 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 = agents.get_agent_instructions(selected_agent["type"], selected_agent, file_contexts) + + # Agent-Prompt erstellen + agent_prompt = agents.create_agent_prompt(selected_agent, agent_instructions) + + # 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"]) + + # Verwende die Benutzereingabe als Prompt für Scraping, falls verfügbar + scrape_prompt = user_message if is_user_input_continuation and user_message else \ + (chat_history[0]["content"] if chat_history else "") + + web_data = await self.service_aiscrap.scrape_web_data(scrape_prompt) + if web_data: + # Ensure web_data has no trailing whitespace + web_data = web_data.strip() if isinstance(web_data, str) else web_data + agent_chat.append({ + "role": "system", + "content": f"# Gescrapte Web-Daten\n{web_data}".strip() + }) + self._add_log(workflow_id, "Web-Scraping abgeschlossen", "info", + selected_agent_id, selected_agent["name"]) + + # Agent führt seinen Teil aus + try: + # Da wir keine partielle Dateiladung benötigen, können wir den Agenten direkt aufrufen + agent_chat = await self._sanitize_messages(agent_chat) + agent_response = await self.service_aichat.call_api(agent_chat) + agent_text = agent_response["choices"][0]["message"]["content"] + + # Datenstatistik aktualisieren + request_size = len(json.dumps(agent_chat)) + response_size = len(json.dumps(agent_response)) + self.workflows[workflow_id]["data_stats"]["sent_bytes"] += request_size + self.workflows[workflow_id]["data_stats"]["received_bytes"] += response_size + + # Füge die endgültige Antwort des Agenten zum Chatverlauf hinzu + chat_history.append({ + "role": "assistant", + "content": agent_text + }) + + # Prüfe, ob die Antwort des Agenten eine Anfrage an den User Agent enthält + is_user_agent_query = self._check_for_user_agent_query(agent_text) + if is_user_agent_query: + self._add_log( + workflow_id, + f"Agent '{selected_agent['name']}' stellt eine Frage an den User Agent", + "info" + ) + # Markiere, dass wir auf eine Benutzereingabe warten + waiting_for_user_input = True + self.workflows[workflow_id]["waiting_for_user"] = True + + # Workflow-Status speichern und Chat-Verlauf aktualisieren + self.workflows[workflow_id]["chat_history"] = chat_history + results.save_workflow_results(self.workflows, workflow_id, self.results_dir) + + return workflow_complete, waiting_for_user_input, selected_agent_id + + # Agent-Ergebnis erstellen - Prompt basierend auf Fortsetzungsart wählen + prompt_for_result = user_message if is_user_input_continuation else \ + (chat_history[0]["content"] if chat_history else "") + + result = results.create_agent_result( + workflow_id, + selected_agent, + len(self.workflows[workflow_id]["results"]), + prompt_for_result, + file_contexts, + agent_text, + self.mandate_id, + self.user_id + ) + 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)}" + }) + + 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") + + # Aktualisiere den Chat-Verlauf im Workflow + self.workflows[workflow_id]["chat_history"] = chat_history + + return workflow_complete, waiting_for_user_input, selected_agent_id + def _add_log( self, @@ -372,7 +539,17 @@ class AgentService: def get_workflow_status(self, workflow_id: str) -> Optional[Dict[str, Any]]: """Gibt den Status eines Workflows zurück""" - return results.get_workflow_status(self.workflows, workflow_id) + workflow_status = results.get_workflow_status(self.workflows, workflow_id) + # Füge Datenstatistik hinzu + if workflow_status and workflow_id in self.workflows: + workflow_status["data_stats"] = self.workflows[workflow_id].get("data_stats", { + "sent_bytes": 0, + "received_bytes": 0 + }) + # Füge das waiting_for_user-Flag explizit hinzu + workflow_status["waiting_for_user"] = self.workflows[workflow_id].get("waiting_for_user", False) + return workflow_status + def get_workflow_logs(self, workflow_id: str) -> Optional[List[Dict[str, Any]]]: """Gibt die Protokolle eines Workflows zurück""" @@ -422,6 +599,255 @@ class AgentService: logger.info(f"Workflow {workflow_id} wurde gestoppt") return True + + async def process_user_input(self, workflow_id: str, message: str, additional_files: List[Dict[str, Any]] = None) -> bool: + """ + Verarbeitet eine Benutzereingabe für einen laufenden Workflow. + Ergänzt um die Verfolgung von User Agent Bestätigungen. + + Args: + workflow_id: ID des Workflows + message: Nachricht des Benutzers + additional_files: Liste zusätzlicher Dateien (optional) + + Returns: + bool: True, wenn die Eingabe erfolgreich verarbeitet wurde + """ + if workflow_id not in self.workflows: + logger.warning(f"Workflow {workflow_id} nicht gefunden") + return False + + # Prüfen, ob der Workflow auf eine Benutzereingabe wartet + if not self.workflows[workflow_id].get("waiting_for_user", False): + logger.warning(f"Workflow {workflow_id} wartet nicht auf Benutzereingabe") + return False + + logger.info(f"Verarbeite Benutzereingabe für Workflow {workflow_id}") + + # Benutzerinfos abrufen + user_info = await self._get_user_info(self.user_id) + user_name = user_info.get("full_name") or user_info.get("username") or f"Benutzer {self.user_id}" + + # Log-Eintrag für die Benutzereingabe + self._add_log( + workflow_id, + f"Benutzereingabe empfangen: {message[:50]}{'...' if len(message) > 50 else ''}", + "info", + "user_agent", + user_name + ) + + # Verfolge die User Agent Antwort für Bestätigungsprüfung + # Importiere die Hilfsfunktion aus dem agents-Modul + from modules.agentservice_part_agents import track_user_agent_response + available_agents = self.workflows[workflow_id].get("available_agents", {}) + track_user_agent_response(available_agents, message) + + # Rest der Funktion bleibt unverändert... + # Wenn zusätzliche Dateien vorhanden sind, diese verarbeiten und in den Workflow integrieren + additional_context = "" + if additional_files and len(additional_files) > 0: + file_contexts = file_handling.prepare_file_contexts(additional_files, self.upload_dir) + file_contents = file_handling.read_file_contents( + file_contexts, + self.upload_dir, + workflow_id, + lambda wid, msg, typ, aid=None, aname=None: self._add_log(wid, msg, typ, aid, aname) + ) + + # Formatiere den Dateikontext + file_context_text = file_handling.format_file_context_text(file_contexts, file_contents) + additional_context = f"\n\n### Zusätzliche Dateien:\n{file_context_text}" + + # Log-Eintrag für die zusätzlichen Dateien + self._add_log( + workflow_id, + f"{len(additional_files)} zusätzliche Dateien hinzugefügt", + "info", + "user_agent", + user_name + ) + + # Füge neue Dateien zu den bestehenden Dateikontexten und -inhalten hinzu + existing_file_contexts = self.workflows[workflow_id].get("file_contexts", []) + existing_file_contents = self.workflows[workflow_id].get("file_contents", {}) + + # Alte IDs speichern, um Duplikate zu vermeiden + existing_ids = {fc["id"] for fc in existing_file_contexts} + + # Neue Dateikontexte hinzufügen (nur wenn ID noch nicht existiert) + for file_context in file_contexts: + if file_context["id"] not in existing_ids: + existing_file_contexts.append(file_context) + existing_ids.add(file_context["id"]) + + # Neue Dateiinhalte hinzufügen + existing_file_contents.update(file_contents) + + # Aktualisierte Dateikontexte und -inhalte im Workflow speichern + self.workflows[workflow_id]["file_contexts"] = existing_file_contexts + self.workflows[workflow_id]["file_contents"] = existing_file_contents + + # Kombinierte Nachricht erstellen (Benutzernachricht + ggf. zusätzliche Dateien) + combined_message = message + additional_context + + # Schätzung der gesendeten Bytes + message_size = len(combined_message) + self.workflows[workflow_id]["data_stats"]["sent_bytes"] += message_size + + # Fortsetzung des Workflows einleiten + # Wir starten eine neue Aufgabe, um den Workflow fortzusetzen + asyncio.create_task( + self._continue_workflow_after_user_input( + workflow_id, + combined_message, + user_name + ) + ) + + return True + + + async def _continue_workflow_after_user_input(self, workflow_id: str, user_message: str, user_name: str) -> None: + """ + Setzt einen Workflow nach einer Benutzereingabe fort. + Nutzt die gemeinsame _run_moderator_cycle Methode für die Fortsetzung. + + Args: + workflow_id: ID des Workflows + user_message: Nachricht des Benutzers + user_name: Name des Benutzers + """ + if workflow_id not in self.workflows: + logger.warning(f"Workflow {workflow_id} nicht gefunden") + return + + # Workflow-Status aktualisieren + self.workflows[workflow_id]["status"] = "running" + + # Log-Eintrag für die Fortsetzung + self._add_log( + workflow_id, + "Workflow wird nach Benutzereingabe fortgesetzt", + "info" + ) + + # Hole die benötigten Daten aus dem Workflow + chat_history = self.workflows[workflow_id].get("chat_history", []) + available_agents = self.workflows[workflow_id].get("available_agents", {}) + file_contexts = self.workflows[workflow_id].get("file_contexts", []) + file_contents = self.workflows[workflow_id].get("file_contents", {}) + + # Benutzereingabe zum Chatverlauf hinzufügen + chat_history.append({ + "role": "user", + "content": f"[User Agent: {user_name}] {user_message}" + }) + + # Speichere den aktualisierten Chat-Verlauf im Workflow + self.workflows[workflow_id]["chat_history"] = chat_history + + # User-Antwort als Ergebnis speichern + user_result = results.create_agent_result( + workflow_id, + {"id": "user_agent", "name": "User Agent", "type": "user"}, + len(self.workflows[workflow_id].get("results", [])), + "Benutzereingabe", + file_contexts, # Dateikontexte übergeben + user_message, + self.mandate_id, + self.user_id + ) + + self.workflows[workflow_id]["results"].append(user_result) + + # Markiere den User-Agent als verwendet + if "user_agent" in available_agents: + available_agents["user_agent"]["used"] = True + + # Markiere, dass wir nicht mehr auf eine Benutzereingabe warten + self.workflows[workflow_id]["waiting_for_user"] = False + + # Hole die aktuelle Rundenzahl und maximale Rundenzahl + current_round = self.workflows[workflow_id].get("current_round", 0) + max_rounds = 12 # Gleicher Wert wie in execute_workflow + + # Prüfe, ob wir schon die maximale Anzahl von Runden erreicht haben + if current_round >= max_rounds: + self._add_log(workflow_id, f"Workflow nach {max_rounds} Runden und Benutzereingabe automatisch beendet", "info") + self.workflows[workflow_id]["status"] = "completed" + self.workflows[workflow_id]["progress"] = 1.0 + self.workflows[workflow_id]["completed_at"] = datetime.now().isoformat() + + # Speichere Ergebnisse + results.save_workflow_results(self.workflows, workflow_id, self.results_dir) + return + + # Nächste Runde starten + current_round += 1 + self.workflows[workflow_id]["current_round"] = current_round + self._add_log(workflow_id, f"Starte Runde {current_round} nach Benutzereingabe", "info") + + # Führe einen Moderator-Zyklus durch + workflow_complete, waiting_for_user_input, selected_agent_id = await self._run_moderator_cycle( + workflow_id, + chat_history, + available_agents, + file_contexts, + file_contents, + is_user_input_continuation=True, # Dies ist eine Fortsetzung nach Benutzereingabe + user_message=user_message + ) + + # Workflow abschließen, wenn er vollständig ist oder nicht auf Benutzereingabe wartet + if workflow_complete or not waiting_for_user_input: + # Prüfe, ob wir weitere Runden durchführen sollten + if not workflow_complete and not waiting_for_user_input and current_round < max_rounds: + # Starte eine neue Aufgabe für die nächste Moderator-Runde + asyncio.create_task( + self._continue_workflow_after_user_input( + workflow_id, + "", # Leere Nachricht für die nächste Runde + user_name + ) + ) + else: + # Workflow wurde entweder abgeschlossen oder wartet auf Benutzereingabe + if workflow_complete: + self.workflows[workflow_id]["status"] = "completed" + self._add_log(workflow_id, "Workflow nach Benutzereingabe erfolgreich beendet", "success") + self.workflows[workflow_id]["progress"] = 1.0 + self.workflows[workflow_id]["completed_at"] = datetime.now().isoformat() + + # Fortschritt aktualisieren + if not self.workflows[workflow_id]["completed_at"]: + self.workflows[workflow_id]["progress"] = min(0.9, 0.1 + (current_round / max_rounds) * 0.8) + + # Speichere Ergebnisse + results.save_workflow_results(self.workflows, workflow_id, self.results_dir) + + + async def _get_user_info(self, user_id: int) -> Dict[str, Any]: + """ + Ruft Benutzerinformationen aus der Datenbank ab. + + Args: + user_id: ID des Benutzers + + Returns: + Dict mit Benutzerinformationen + """ + try: + # Hier würden wir normalerweise die Benutzerinformationen aus einer Datenbank abrufen + # Für diese Implementierung verwenden wir Platzhalter + return { + "id": user_id, + "username": f"user_{user_id}", + "full_name": f"Benutzer {user_id}" + } + except Exception as e: + logger.error(f"Fehler beim Abrufen der Benutzerinformationen: {str(e)}") + return {"id": user_id} async def _sanitize_messages(self, messages): """Sanitizes all messages to prevent API errors.""" @@ -455,6 +881,59 @@ class AgentService: ] return content + + def _check_for_user_agent_query(self, message_content: str) -> bool: + """ + Prüft, ob ein Text eine Anfrage an den User-Agent enthält. + + Args: + message_content: Der zu prüfende Text + + Returns: + bool: True, wenn der Text eine Anfrage an den User-Agent enthält + """ + # Keine Prüfung, wenn der Text leer ist + if not message_content: + return False + + # Prüfmuster für User-Agent-Anfragen + user_agent_phrases = [ + 'User Agent', + 'Benutzer', + 'Was denken Sie', + 'Was denkst du', + 'Ihre Meinung', + 'deine Meinung', + 'Sind Sie zufrieden', + 'Gibt es weitere Fragen', + 'Wollen Sie', + 'Möchten Sie', + 'Kannst du', + 'Können Sie', + 'Könnten Sie', + 'Bitte teilen Sie uns mit', + 'Moderator zu User Agent' + ] + + # Text in Kleinbuchstaben umwandeln für case-insensitive Suche + content_lower = message_content.lower() + + # Prüfen, ob ein Fragezeichen enthalten ist + has_question_mark = '?' in message_content + + # Prüfen, ob eine der Phrasen enthalten ist + has_user_agent_phrase = any(phrase.lower() in content_lower for phrase in user_agent_phrases) + + # Spezielle Bedingungen für User-Agent-Anfragen + is_user_agent_query = ( + (has_question_mark and has_user_agent_phrase) or + ('user agent' in content_lower and 'workflow' in content_lower) or + 'moderator zu user agent' in content_lower + ) + return is_user_agent_query + + + # Singleton-Factory für AgentService-Instanzen pro Kontext _agent_service_instances = {} @@ -467,4 +946,5 @@ def get_agentservice_interface(mandate_id: int = None, user_id: int = None) -> A 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] \ No newline at end of file + return _agent_service_instances[context_key] + diff --git a/gwserver/modules/agentservice_part_agents.py b/gwserver/modules/agentservice_part_agents.py index 7e822974..a5ba2810 100644 --- a/gwserver/modules/agentservice_part_agents.py +++ b/gwserver/modules/agentservice_part_agents.py @@ -135,6 +135,7 @@ def initialize_agents(agents: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]] def get_moderator_prompt(available_agents: Dict[str, Dict[str, Any]]) -> str: """ Erstellt einen Moderator-Prompt, der die Status-Deklarationen der Agenten berücksichtigt. + Mit besonderem Fokus auf die explizite Bestätigung durch den User Agent. Args: available_agents: Dictionary mit verfügbaren Agenten @@ -142,10 +143,29 @@ def get_moderator_prompt(available_agents: Dict[str, Dict[str, Any]]) -> str: 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. -WICHTIG: Der Workflow darf erst beendet werden, wenn TATSÄCHLICHE ERGEBNISSE geliefert wurden.""" +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(): @@ -167,14 +187,57 @@ Berücksichtige die STATUS-Deklarationen der Agenten bei deiner Entscheidung: 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! + 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 @@ -199,7 +262,8 @@ def create_agent_prompt(agent: Dict[str, Any], agent_instructions: str, file_con 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. + 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 @@ -224,14 +288,30 @@ def find_next_agent(moderator_text: str, available_agents: Dict[str, Dict[str, A 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 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 + # 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: - logger.warning("Moderator versuchte, Workflow ohne vollständiges Ergebnis zu beenden") + 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: @@ -246,6 +326,14 @@ def find_next_agent(moderator_text: str, available_agents: Dict[str, Dict[str, A 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"]: @@ -257,7 +345,6 @@ def find_next_agent(moderator_text: str, available_agents: Dict[str, Dict[str, A return None - def update_agent_results_with_status(content: str) -> Tuple[str, str]: """ Extrahiert den deklarierten Status aus einem Agenten-Ergebnis. diff --git a/gwserver/routes/workflows.py b/gwserver/routes/workflows.py index 7249d687..93a88b93 100644 --- a/gwserver/routes/workflows.py +++ b/gwserver/routes/workflows.py @@ -174,4 +174,105 @@ async def stop_workflow( "workflow_id": workflow_id, "status": "stopped", "message": "Workflow wurde gestoppt" - } \ No newline at end of file + } + +@router.post("/{workflow_id}/user-input", response_model=Dict[str, Any]) +async def submit_user_input( + workflow_id: str = Path(..., description="ID des Workflows"), + user_input: Dict[str, Any] = Body(...), + current_user: Dict[str, Any] = Depends(get_current_active_user) +): + """ + Ermöglicht es dem Benutzer, Eingaben für einen laufenden Workflow zu senden. + Dies wird verwendet, wenn der User-Agent im Workflow angesprochen wird. + + Der Request-Body sollte folgendes Format haben: + { + "message": "Die Antwort des Benutzers", + "additional_files": [1, 2, 3] // Optional: IDs zusätzlicher Dateien + } + """ + mandate_id, user_id = await get_user_context(current_user) + + # Überprüfen, ob die erforderlichen Felder vorhanden sind + if "message" not in user_input: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Das Feld 'message' ist erforderlich" + ) + + message = user_input["message"] + additional_file_ids = user_input.get("additional_files", []) + + # LucyDOM-Interface mit Benutzerkontext initialisieren + lucy_interface = get_lucydom_interface(mandate_id, user_id) + + # AgentService mit Benutzerkontext initialisieren + agent_service = get_agentservice_interface(mandate_id, user_id) + + # Zusätzliche Dateien einsammeln, wenn vorhanden + additional_files = [] + if additional_file_ids: + for file_id in additional_file_ids: + file = lucy_interface.get_file(file_id) + if not file: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Zusätzliche Datei mit ID {file_id} nicht gefunden" + ) + additional_files.append(file) + + # Benutzereingabe verarbeiten + try: + result = await agent_service.process_user_input(workflow_id, message, additional_files) + if not result: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Workflow {workflow_id} wartet nicht auf Benutzereingabe oder existiert nicht" + ) + + return { + "workflow_id": workflow_id, + "status": "processing", + "message": "Benutzereingabe wurde empfangen und wird verarbeitet" + } + except Exception as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Fehler bei der Verarbeitung der Benutzereingabe: {str(e)}" + ) + +@router.get("/{workflow_id}/data-statistics", response_model=Dict[str, Any]) +async def get_workflow_data_statistics( + workflow_id: str, + current_user: Dict[str, Any] = Depends(get_current_active_user) +): + """ + Gibt Statistiken über die übertragenen Datenmengen für einen Workflow zurück. + """ + mandate_id, user_id = await get_user_context(current_user) + + # AgentService mit Benutzerkontext initialisieren + agent_service = get_agentservice_interface(mandate_id, user_id) + + status = agent_service.get_workflow_status(workflow_id) + if not status: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Workflow mit ID {workflow_id} nicht gefunden" + ) + + # Gib nur die Datenstatistiken zurück + if "data_stats" in status: + return { + "workflow_id": workflow_id, + "data_stats": status["data_stats"] + } + else: + return { + "workflow_id": workflow_id, + "data_stats": { + "sent_bytes": 0, + "received_bytes": 0 + } + } \ No newline at end of file