File: //opt/owl/owl/webapp_de.py
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Import from the correct module path
from utils import run_society
import os
import gradio as gr
import time
import json
import logging
import datetime
from typing import Tuple
import importlib
from dotenv import load_dotenv, set_key, find_dotenv, unset_key
import threading
import queue
import re
os.environ["PYTHONIOENCODING"] = "utf-8"
# Logging-System konfigurieren
def setup_logging():
"""Konfiguriert das Logging-System, um Logs in Datei, Speicher-Queue und Konsole auszugeben"""
# Logs-Verzeichnis erstellen (falls nicht vorhanden)
logs_dir = os.path.join(os.path.dirname(__file__), "logs")
os.makedirs(logs_dir, exist_ok=True)
# Log-Dateinamen generieren (unter Verwendung des aktuellen Datums)
current_date = datetime.datetime.now().strftime("%Y-%m-%d")
log_file = os.path.join(logs_dir, f"gradio_log_{current_date}.txt")
# Root-Logger konfigurieren (erfasst alle Logs)
root_logger = logging.getLogger()
# Vorhandene Handler entfernen, um doppelte Logs zu vermeiden
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
root_logger.setLevel(logging.INFO)
# File-Handler erstellen
file_handler = logging.FileHandler(log_file, encoding="utf-8", mode="a")
file_handler.setLevel(logging.INFO)
# Konsolen-Handler erstellen
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# Formatter erstellen
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# Handler zum Root-Logger hinzufügen
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
logging.info("Logging-System initialisiert, Log-Datei: %s", log_file)
return log_file
# Globale Variablen
LOG_FILE = None
LOG_QUEUE: queue.Queue = queue.Queue() # Log-Queue
STOP_LOG_THREAD = threading.Event()
CURRENT_PROCESS = None # Verfolgt den aktuell laufenden Prozess
STOP_REQUESTED = threading.Event() # Markiert, ob ein Stopp angefordert wurde
# Funktionen zum Lesen und Aktualisieren der Logs
def log_reader_thread(log_file):
"""Hintergrundthread, der kontinuierlich die Log-Datei liest und neue Zeilen der Queue hinzufügt"""
try:
with open(log_file, "r", encoding="utf-8") as f:
# Zum Dateiende springen
f.seek(0, 2)
while not STOP_LOG_THREAD.is_set():
line = f.readline()
if line:
LOG_QUEUE.put(line) # Zur Gesprächsaufzeichnungs-Queue hinzufügen
else:
# Keine neuen Zeilen, kurze Pause
time.sleep(0.1)
except Exception as e:
logging.error(f"Fehler im Log-Lesethread: {str(e)}")
def get_latest_logs(max_lines=100, queue_source=None):
"""Liefert die neuesten Log-Zeilen aus der Queue oder liest direkt aus der Datei, wenn die Queue leer ist
Args:
max_lines: Maximale Anzahl der zurückgegebenen Zeilen
queue_source: Gibt an, welche Queue verwendet werden soll, Standard ist LOG_QUEUE
Returns:
str: Log-Inhalt
"""
logs = []
log_queue = queue_source if queue_source else LOG_QUEUE
# Temporäre Queue erstellen, um Logs zu speichern, ohne sie aus der Original-Queue zu entfernen
temp_queue = queue.Queue()
temp_logs = []
try:
# Alle verfügbaren Log-Zeilen aus der Queue abrufen
while not log_queue.empty() and len(temp_logs) < max_lines:
log = log_queue.get_nowait()
temp_logs.append(log)
temp_queue.put(log) # Log wieder in die temporäre Queue einfügen
except queue.Empty:
pass
# Gesprächsaufzeichnungen verarbeiten
logs = temp_logs
# Falls nicht genügend Logs in der Queue sind, letzte Zeilen direkt aus der Datei lesen
if len(logs) < max_lines and LOG_FILE and os.path.exists(LOG_FILE):
try:
with open(LOG_FILE, "r", encoding="utf-8") as f:
all_lines = f.readlines()
# Falls bereits einige Logs in der Queue sind, nur die restlichen Zeilen lesen
remaining_lines = max_lines - len(logs)
file_logs = (
all_lines[-remaining_lines:]
if len(all_lines) > remaining_lines
else all_lines
)
# Datei-Logs vor Queue-Logs einfügen
logs = file_logs + logs
except Exception as e:
error_msg = f"Fehler beim Lesen der Log-Datei: {str(e)}"
logging.error(error_msg)
if not logs: # Fehlernachricht nur hinzufügen, wenn keine Logs vorhanden sind
logs = [error_msg]
# Falls weiterhin keine Logs vorhanden sind, Hinweis zurückgeben
if not logs:
return "Initialisierung läuft..."
# Logs filtern, nur Zeilen mit 'camel.agents.chat_agent - INFO' behalten
filtered_logs = []
for log in logs:
if "camel.agents.chat_agent - INFO" in log:
filtered_logs.append(log)
# Falls nach dem Filtern keine Logs übrig sind, Hinweis zurückgeben
if not filtered_logs:
return "Noch keine Gesprächsaufzeichnungen vorhanden."
# Log-Inhalt verarbeiten, um die neuesten Benutzer- und Agentennachrichten zu extrahieren
simplified_logs = []
# Set zur Verfolgung bereits verarbeiteter Nachrichten, um Duplikate zu vermeiden
processed_messages = set()
def process_message(role, content):
# Einzigartige Kennung zur Verfolgung der Nachrichten erstellen
msg_id = f"{role}:{content}"
if msg_id in processed_messages:
return None
processed_messages.add(msg_id)
content = content.replace("\\n", "\n")
lines = [line.strip() for line in content.split("\n")]
content = "\n".join(lines)
role_emoji = "🙋" if role.lower() == "user" else "🤖"
return f"""### {role_emoji} {role.title()} Agent
{content}"""
for log in filtered_logs:
formatted_messages = []
# Versuchen, das Nachrichtenarray zu extrahieren
messages_match = re.search(
r"Model (.*?), index (\d+), processed these messages: (\[.*\])", log
)
if messages_match:
try:
messages = json.loads(messages_match.group(3))
for msg in messages:
if msg.get("role") in ["user", "assistant"]:
formatted_msg = process_message(
msg.get("role"), msg.get("content", "")
)
if formatted_msg:
formatted_messages.append(formatted_msg)
except json.JSONDecodeError:
pass
# Falls JSON-Parsing fehlschlägt oder kein Nachrichtenarray gefunden wird, direkten Inhalt extrahieren
if not formatted_messages:
user_pattern = re.compile(r"\{'role': 'user', 'content': '(.*?)'\}")
assistant_pattern = re.compile(
r"\{'role': 'assistant', 'content': '(.*?)'\}"
)
for content in user_pattern.findall(log):
formatted_msg = process_message("user", content)
if formatted_msg:
formatted_messages.append(formatted_msg)
for content in assistant_pattern.findall(log):
formatted_msg = process_message("assistant", content)
if formatted_msg:
formatted_messages.append(formatted_msg)
if formatted_messages:
simplified_logs.append("\n\n".join(formatted_messages))
# Formatierung der Log-Ausgabe, trenne einzelne Gesprächsaufzeichnungen angemessen
formatted_logs = []
for i, log in enumerate(simplified_logs):
# Überschüssige Leerzeichen am Anfang und Ende entfernen
log = log.strip()
formatted_logs.append(log)
# Sicherstellen, dass jede Aufzeichnung mit einem Zeilenumbruch endet
if not log.endswith("\n"):
formatted_logs.append("\n")
return "\n".join(formatted_logs)
# Wörterbuch mit Modulbeschreibungen
MODULE_DESCRIPTIONS = {
"run": "Verwendung des OpenAI-Modells mit minimaler Konfiguration zur Aufgabenbearbeitung",
"run_mini": "Verwendung des OpenAI-Modells mit minimaler Konfiguration zur Aufgabenbearbeitung",
"run_gemini": "Verwendung des Gemini-Modells zur Aufgabenbearbeitung",
"run_deepseek_de": "Verwendung des Deepseek-Modells für deutschsprachige Aufgaben",
"lokal": "Verwendung eines lokalen DSGVO-kompatiblen Modells zur Aufgabenbearbeitung",
"run_groq": "Verwendung des Groq-Modells zur Aufgabenbearbeitung",
}
# Standard-Umgebungsvariablenschablone
DEFAULT_ENV_TEMPLATE = """#===========================================
# MODELL & API
# (Siehe https://docs.camel-ai.org/key_modules/models.html#)
#===========================================
# OPENAI API (https://platform.openai.com/api-keys)
OPENAI_API_KEY='Ihr_Schlüssel'
# OPENAI_API_BASE_URL=""
# Azure OpenAI API
# AZURE_OPENAI_BASE_URL=""
# AZURE_API_VERSION=""
# AZURE_OPENAI_API_KEY=""
# AZURE_DEPLOYMENT_NAME=""
# Qwen API (https://help.aliyun.com/zh/model-studio/developer-reference/get-api-key)
QWEN_API_KEY='Ihr_Schlüssel'
# DeepSeek API (https://platform.deepseek.com/api_keys)
DEEPSEEK_API_KEY='Ihr_Schlüssel'
#===========================================
# Tools & Services API
#===========================================
# Google Search API (https://coda.io/@jon-dallas/google-image-search-pack-example/search-engine-id-and-google-api-key-3)
GOOGLE_API_KEY='Ihr_Schlüssel'
SEARCH_ENGINE_ID='Ihre_ID'
# Chunkr API (https://chunkr.ai/)
CHUNKR_API_KEY='Ihr_Schlüssel'
# Firecrawl API (https://www.firecrawl.dev/)
FIRECRAWL_API_KEY='Ihr_Schlüssel'
#FIRECRAWL_API_URL="https://api.firecrawl.dev"
"""
def validate_input(question: str) -> bool:
"""Überprüft, ob die Benutzereingabe gültig ist
Args:
question: Benutzerfrage
Returns:
bool: Ob die Eingabe gültig ist
"""
# Prüfen, ob die Eingabe leer ist oder nur aus Leerzeichen besteht
if not question or question.strip() == "":
return False
return True
def run_eule(question: str, example_module: str) -> Tuple[str, str, str]:
"""Führt das EULE-System aus und liefert Ergebnisse
Args:
question: Benutzerfrage
example_module: Beispielmodulname zum Importieren (z.B. "run_terminal_zh" oder "run_deep")
Returns:
Tuple[...]: Antwort, Tokenanzahl, Status
"""
global CURRENT_PROCESS
# Eingabe validieren
if not validate_input(question):
logging.warning("Benutzer hat eine ungültige Eingabe eingereicht")
return (
"Bitte geben Sie eine gültige Frage ein",
"0",
"❌ Fehler: Ungültige Eingabe",
)
try:
# Sicherstellen, dass Umgebungsvariablen geladen werden
load_dotenv(find_dotenv(), override=True)
logging.info(
f"Verarbeite Frage: '{question}', verwende Modul: {example_module}"
)
# Überprüfen, ob das Modul in MODULE_DESCRIPTIONS enthalten ist
if example_module not in MODULE_DESCRIPTIONS:
logging.error(f"Nicht unterstütztes Modul ausgewählt: {example_module}")
return (
f"Ausgewähltes Modul '{example_module}' wird nicht unterstützt",
"0",
"❌ Fehler: Nicht unterstütztes Modul",
)
# Dynamischen Import des Zielmoduls
module_path = f"examples.{example_module}"
try:
logging.info(f"Importiere Modul: {module_path}")
module = importlib.import_module(module_path)
except ImportError as ie:
logging.error(f"Modul {module_path} konnte nicht importiert werden: {str(ie)}")
return (
f"Modul konnte nicht importiert werden: {module_path}",
"0",
f"❌ Fehler: Modul {example_module} existiert nicht oder kann nicht geladen werden - {str(ie)}",
)
except Exception as e:
logging.error(
f"Fehler beim Import des Moduls {module_path}: {str(e)}"
)
return (
f"Fehler beim Import des Moduls: {module_path}",
"0",
f"❌ Fehler: {str(e)}",
)
# Überprüfen, ob die Funktion construct_society vorhanden ist
if not hasattr(module, "construct_society"):
logging.error(
f"Funktion construct_society wurde im Modul {module_path} nicht gefunden"
)
return (
f"Funktion construct_society wurde im Modul {module_path} nicht gefunden",
"0",
"❌ Fehler: Modul inkompatibel",
)
# Gesellschaftssimulation erstellen
try:
logging.info("Erstelle Gesellschaftssimulation...")
society = module.construct_society(question)
except Exception as e:
logging.error(f"Fehler beim Erstellen der Gesellschaftssimulation: {str(e)}")
return (
f"Fehler beim Erstellen der Gesellschaftssimulation: {str(e)}",
"0",
f"❌ Fehler: Erstellung fehlgeschlagen - {str(e)}",
)
# Gesellschaftssimulation ausführen
try:
logging.info("Führe Gesellschaftssimulation aus...")
answer, chat_history, token_info = run_society(society)
logging.info("Gesellschaftssimulation abgeschlossen")
except Exception as e:
logging.error(f"Fehler bei der Ausführung der Gesellschaftssimulation: {str(e)}")
return (
f"Fehler bei der Ausführung der Gesellschaftssimulation: {str(e)}",
"0",
f"❌ Fehler: Ausführung fehlgeschlagen - {str(e)}",
)
# Tokenanzahl sicher ermitteln
if not isinstance(token_info, dict):
token_info = {}
completion_tokens = token_info.get("completion_token_count", 0)
prompt_tokens = token_info.get("prompt_token_count", 0)
total_tokens = completion_tokens + prompt_tokens
logging.info(
f"Verarbeitung abgeschlossen, Tokenverbrauch: Abschluss={completion_tokens}, Eingabe={prompt_tokens}, Gesamt={total_tokens}"
)
return (
answer,
f"Abschluss-Token: {completion_tokens:,} | Eingabe-Token: {prompt_tokens:,} | Gesamt: {total_tokens:,}",
"✅ Erfolgreich abgeschlossen",
)
except Exception as e:
logging.error(
f"Unbehandelter Fehler bei der Verarbeitung der Frage: {str(e)}"
)
return (f"Fehler: {str(e)}", "0", f"❌ Fehler: {str(e)}")
def update_module_description(module_name: str) -> str:
"""Gibt die Beschreibung des ausgewählten Moduls zurück"""
return MODULE_DESCRIPTIONS.get(module_name, "Keine Beschreibung verfügbar")
# Speichert von der Frontend konfigurierte Umgebungsvariablen
WEB_FRONTEND_ENV_VARS: dict[str, str] = {}
def init_env_file():
"""Initialisiert die .env-Datei, falls sie nicht existiert"""
dotenv_path = find_dotenv()
if not dotenv_path:
with open(".env", "w") as f:
f.write(DEFAULT_ENV_TEMPLATE)
dotenv_path = find_dotenv()
return dotenv_path
def load_env_vars():
"""Lädt Umgebungsvariablen und gibt sie als Wörterbuch zurück
Returns:
dict: Wörterbuch der Umgebungsvariablen, jeder Wert ist ein Tupel (Wert, Quelle)
"""
dotenv_path = init_env_file()
load_dotenv(dotenv_path, override=True)
# Umgebungsvariablen aus der .env-Datei lesen
env_file_vars = {}
with open(dotenv_path, "r") as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
if "=" in line:
key, value = line.split("=", 1)
env_file_vars[key.strip()] = value.strip().strip("\"'")
# Aus den System-Umgebungsvariablen lesen
system_env_vars = {
k: v
for k, v in os.environ.items()
if k not in env_file_vars and k not in WEB_FRONTEND_ENV_VARS
}
# Umgebungsvariablen zusammenführen und Quellen kennzeichnen
env_vars = {}
# System-Umgebungsvariablen hinzufügen (niedrigste Priorität)
for key, value in system_env_vars.items():
env_vars[key] = (value, "System")
# .env-Datei Variablen hinzufügen (mittlere Priorität)
for key, value in env_file_vars.items():
env_vars[key] = (value, ".env-Datei")
# Vom Frontend konfigurierte Umgebungsvariablen hinzufügen (höchste Priorität)
for key, value in WEB_FRONTEND_ENV_VARS.items():
env_vars[key] = (value, "Frontend-Konfiguration")
# Auch die System-Umgebungsvariablen aktualisieren
os.environ[key] = value
return env_vars
def save_env_vars(env_vars):
"""Speichert Umgebungsvariablen in der .env-Datei
Args:
env_vars: Wörterbuch, Schlüssel sind Variablennamen, Werte können Strings oder (Wert, Quelle)-Tupel sein
"""
try:
dotenv_path = init_env_file()
# Jede Umgebungsvariable speichern
for key, value_data in env_vars.items():
if key and key.strip(): # Schlüssel darf nicht leer sein
# Fall behandeln, dass value ein Tupel sein könnte
if isinstance(value_data, tuple):
value = value_data[0]
else:
value = value_data
set_key(dotenv_path, key.strip(), value.strip())
# Umgebungsvariablen neu laden, um die Änderungen zu übernehmen
load_dotenv(dotenv_path, override=True)
return True, "Umgebungsvariablen wurden erfolgreich gespeichert!"
except Exception as e:
return False, f"Fehler beim Speichern der Umgebungsvariablen: {str(e)}"
def add_env_var(key, value, from_frontend=True):
"""Fügt eine einzelne Umgebungsvariable hinzu oder aktualisiert sie
Args:
key: Name der Umgebungsvariable
value: Wert der Umgebungsvariable
from_frontend: Ob sie von der Frontend-Konfiguration stammt, Standard ist True
"""
try:
if not key or not key.strip():
return False, "Variablenname darf nicht leer sein"
key = key.strip()
value = value.strip()
# Falls von Frontend, zur Frontend-Umgebungsvariablen-Dictionary hinzufügen
if from_frontend:
WEB_FRONTEND_ENV_VARS[key] = value
# System-Umgebungsvariablen direkt aktualisieren
os.environ[key] = value
# Auch in der .env-Datei aktualisieren
dotenv_path = init_env_file()
set_key(dotenv_path, key, value)
load_dotenv(dotenv_path, override=True)
return True, f"Umgebungsvariable {key} wurde erfolgreich hinzugefügt/aktualisiert!"
except Exception as e:
return False, f"Fehler beim Hinzufügen der Umgebungsvariablen: {str(e)}"
def delete_env_var(key):
"""Löscht eine Umgebungsvariable"""
try:
if not key or not key.strip():
return False, "Variablenname darf nicht leer sein"
key = key.strip()
# Aus der .env-Datei löschen
dotenv_path = init_env_file()
unset_key(dotenv_path, key)
# Aus dem Frontend-Umgebungsvariablen-Dictionary löschen
if key in WEB_FRONTEND_ENV_VARS:
del WEB_FRONTEND_ENV_VARS[key]
# Auch aus den aktuellen System-Umgebungsvariablen löschen
if key in os.environ:
del os.environ[key]
return True, f"Umgebungsvariable {key} wurde erfolgreich gelöscht!"
except Exception as e:
return False, f"Fehler beim Löschen der Umgebungsvariablen: {str(e)}"
def is_api_related(key: str) -> bool:
"""Bestimmt, ob eine Umgebungsvariable API-bezogen ist
Args:
key: Name der Umgebungsvariablen
Returns:
bool: Ob sie API-bezogen ist
"""
# API-bezogene Schlüsselwörter
api_keywords = [
"api",
"key",
"token",
"secret",
"password",
"openai",
"qwen",
"deepseek",
"google",
"search",
"hf",
"hugging",
"chunkr",
"firecrawl",
]
# Überprüfen, ob eines der Schlüsselwörter (unabhängig von der Groß-/Kleinschreibung) enthalten ist
return any(keyword in key.lower() for keyword in api_keywords)
def get_api_guide(key: str) -> str:
"""Gibt basierend auf dem Namen der Umgebungsvariablen den entsprechenden API-Leitfaden zurück
Args:
key: Name der Umgebungsvariablen
Returns:
str: API-Leitfaden-Link oder Beschreibung
"""
key_lower = key.lower()
if "openai" in key_lower:
return "https://platform.openai.com/api-keys"
elif "qwen" in key_lower or "dashscope" in key_lower:
return "https://help.aliyun.com/zh/model-studio/developer-reference/get-api-key"
elif "deepseek" in key_lower:
return "https://platform.deepseek.com/api_keys"
elif "google" in key_lower:
return "https://coda.io/@jon-dallas/google-image-search-pack-example/search-engine-id-and-google-api-key-3"
elif "search_engine_id" in key_lower:
return "https://coda.io/@jon-dallas/google-image-search-pack-example/search-engine-id-and-google-api-key-3"
elif "chunkr" in key_lower:
return "https://chunkr.ai/"
elif "firecrawl" in key_lower:
return "https://www.firecrawl.dev/"
else:
return ""
def update_env_table():
"""Aktualisiert die Anzeige der Umgebungsvariablen-Tabelle und zeigt nur API-bezogene Variablen"""
env_vars = load_env_vars()
# API-bezogene Umgebungsvariablen filtern
api_env_vars = {k: v for k, v in env_vars.items() if is_api_related(k)}
# In Listenformat umwandeln, um den Anforderungen von Gradio Dataframe zu entsprechen
# Format: [Variablennamen, Wert, Leitfaden-Link]
result = []
for k, v in api_env_vars.items():
guide = get_api_guide(k)
# Falls ein Leitfaden-Link vorhanden ist, klickbaren Link erstellen
guide_link = (
f"<a href='{guide}' target='_blank' class='guide-link'>🔗 Abrufen</a>"
if guide
else ""
)
result.append([k, v[0], guide_link])
return result
def save_env_table_changes(data):
"""Speichert Änderungen an der Umgebungsvariablen-Tabelle
Args:
data: Dataframe-Daten, eventuell ein pandas DataFrame-Objekt
Returns:
str: Statusinformation der Operation, als HTML-formatierte Statusnachricht
"""
try:
logging.info(
f"Starte Verarbeitung der Umgebungsvariablen-Tabelle, Typ: {type(data)}"
)
# Alle aktuellen Umgebungsvariablen abrufen
current_env_vars = load_env_vars()
processed_keys = set() # Verarbeitete Schlüssel speichern, um gelöschte Variablen zu erkennen
# Verarbeitung eines pandas DataFrame-Objekts
import pandas as pd
if isinstance(data, pd.DataFrame):
# Spaltennamen abrufen
columns = data.columns.tolist()
logging.info(f"DataFrame Spalten: {columns}")
# Jede Zeile des DataFrame iterieren
for index, row in data.iterrows():
# Über Spaltennamen auf Daten zugreifen
if len(columns) >= 3:
# Variablennamen und Wert abrufen (Spalte 0 ist Name, Spalte 1 ist Wert)
key = row[0] if isinstance(row, pd.Series) else row.iloc[0]
value = row[1] if isinstance(row, pd.Series) else row.iloc[1]
# Leere Zeile oder gelöschte Variable prüfen
if key and str(key).strip():
logging.info(
f"Verarbeite Umgebungsvariable: {key} = {value}"
)
add_env_var(key, str(value))
processed_keys.add(key)
# Verarbeitung anderer Formate
elif isinstance(data, dict):
logging.info(f"Dictionary-Format, Schlüssel: {list(data.keys())}")
# Falls Dictionary-Format, unterschiedliche Schlüssel prüfen
if "data" in data:
rows = data["data"]
elif "values" in data:
rows = data["values"]
elif "value" in data:
rows = data["value"]
else:
# Direktes Dictionary als Zeilendaten verwenden
rows = []
for key, value in data.items():
if key not in ["headers", "types", "columns"]:
rows.append([key, value])
if isinstance(rows, list):
for row in rows:
if isinstance(row, list) and len(row) >= 2:
key, value = row[0], row[1]
if key and str(key).strip():
add_env_var(key, str(value))
processed_keys.add(key)
elif isinstance(data, list):
# Listenformat
for row in data:
if isinstance(row, list) and len(row) >= 2:
key, value = row[0], row[1]
if key and str(key).strip():
add_env_var(key, str(value))
processed_keys.add(key)
else:
logging.error(f"Unbekanntes Datenformat: {type(data)}")
return f"❌ Speichern fehlgeschlagen: Unbekanntes Datenformat {type(data)}"
# Gelöschte Variablen verarbeiten - prüfen, ob es Variablen gibt, die in der Tabelle nicht erscheinen
api_related_keys = {k for k in current_env_vars.keys() if is_api_related(k)}
keys_to_delete = api_related_keys - processed_keys
# Variablen löschen, die nicht mehr in der Tabelle sind
for key in keys_to_delete:
logging.info(f"Lösche Umgebungsvariable: {key}")
delete_env_var(key)
return "✅ Umgebungsvariablen wurden erfolgreich gespeichert"
except Exception as e:
import traceback
error_details = traceback.format_exc()
logging.error(f"Fehler beim Speichern der Umgebungsvariablen: {str(e)}\n{error_details}")
return f"❌ Speichern fehlgeschlagen: {str(e)}"
def get_env_var_value(key):
"""Gibt den tatsächlichen Wert einer Umgebungsvariablen zurück
Priorität: Frontend-Konfiguration > .env-Datei > System-Umgebungsvariablen
"""
# Zuerst die vom Frontend konfigurierten Umgebungsvariablen prüfen
if key in WEB_FRONTEND_ENV_VARS:
return WEB_FRONTEND_ENV_VARS[key]
# Aus den System-Umgebungsvariablen (einschließlich .env) abrufen
return os.environ.get(key, "")
def create_ui():
"""Erstellt die erweiterte Gradio-Benutzeroberfläche"""
def clear_log_file():
"""Löscht den Inhalt der Log-Datei"""
try:
if LOG_FILE and os.path.exists(LOG_FILE):
# Inhalt der Log-Datei leeren, anstatt die Datei zu löschen
open(LOG_FILE, "w").close()
logging.info("Logdatei wurde geleert")
# Log-Queue leeren
while not LOG_QUEUE.empty():
try:
LOG_QUEUE.get_nowait()
except queue.Empty:
break
return ""
else:
return ""
except Exception as e:
logging.error(f"Fehler beim Leeren der Logdatei: {str(e)}")
return ""
# Echtzeit-Log-Aktualisierungsfunktion erstellen
def process_with_live_logs(question, module_name):
"""Verarbeitet Fragen und aktualisiert Logs in Echtzeit"""
global CURRENT_PROCESS
# Log-Datei leeren
clear_log_file()
# Einen Hintergrundthread zur Verarbeitung der Frage erstellen
result_queue = queue.Queue()
def process_in_background():
try:
result = run_eule(question, module_name)
result_queue.put(result)
except Exception as e:
result_queue.put(
(f"Fehler: {str(e)}", "0", f"❌ Fehler: {str(e)}")
)
# Hintergrundthread starten
bg_thread = threading.Thread(target=process_in_background)
CURRENT_PROCESS = bg_thread # Aktuellen Prozess vermerken
bg_thread.start()
# Während der Verarbeitung einmal pro Sekunde die Logs aktualisieren
while bg_thread.is_alive():
# Gesprächsaufzeichnungen aktualisieren
logs2 = get_latest_logs(100, LOG_QUEUE)
# Status immer aktualisieren
yield (
"0",
"<span class='status-indicator status-running'></span> Verarbeitung läuft...",
logs2,
)
time.sleep(1)
# Verarbeitung abgeschlossen, Ergebnisse abrufen
if not result_queue.empty():
result = result_queue.get()
answer, token_count, status = result
# Letzte Aktualisierung der Gesprächsaufzeichnungen
logs2 = get_latest_logs(100, LOG_QUEUE)
# Statusindikator basierend auf dem Status setzen
if "Fehler" in status:
status_with_indicator = (
f"<span class='status-indicator status-error'></span> {status}"
)
else:
status_with_indicator = (
f"<span class='status-indicator status-success'></span> {status}"
)
yield token_count, status_with_indicator, logs2
else:
logs2 = get_latest_logs(100, LOG_QUEUE)
yield (
"0",
"<span class='status-indicator status-error'></span> Beendet",
logs2,
)
with gr.Blocks(title="EULE", theme=gr.themes.Soft(primary_hue="blue")) as app:
gr.Markdown(
"""
# 🦉 EULE Effiziente Unterstützung für Lernende und Experten
"""
)
# Benutzerdefiniertes CSS hinzufügen
gr.HTML("""
<style>
/* Chat-Container-Stil */
.chat-container .chatbot {
height: 500px;
overflow-y: auto;
border-radius: 10px;
box-shadow: 0 2px 10px rgba(0, 0, 0, 0.1);
}
/* Verbesserter Tab-Stil */
.tabs .tab-nav {
background-color: #f5f5f5;
border-radius: 8px 8px 0 0;
padding: 5px;
}
.tabs .tab-nav button {
border-radius: 5px;
margin: 0 3px;
padding: 8px 15px;
font-weight: 500;
}
.tabs .tab-nav button.selected {
background-color: #2c7be5;
color: white;
}
/* Statusindikator-Stil */
.status-indicator {
display: inline-block;
width: 10px;
height: 10px;
border-radius: 50%;
margin-right: 5px;
}
.status-running {
background-color: #ffc107;
animation: pulse 1.5s infinite;
}
.status-success {
background-color: #28a745;
}
.status-error {
background-color: #dc3545;
}
/* Log-Anzeige-Stil */
.log-display textarea {
height: 400px !important;
max-height: 400px !important;
overflow-y: auto !important;
font-family: monospace;
font-size: 0.9em;
white-space: pre-wrap;
line-height: 1.4;
}
.log-display {
border-radius: 10px;
padding: 15px;
margin-bottom: 20px;
min-height: 50vh;
max-height: 75vh;
}
/* Status-Icon-Stil */
.status-icon-cell {
text-align: center;
font-size: 1.2em;
}
/* Link-Stil */
.guide-link {
color: #2c7be5;
text-decoration: none;
cursor: pointer;
font-weight: 500;
}
.guide-link:hover {
text-decoration: underline;
}
.env-status {
margin-top: 15px;
font-weight: 500;
padding: 10px;
border-radius: 6px;
transition: all 0.3s ease;
}
.env-status-success {
background-color: #d4edda;
color: #155724;
border: 1px solid #c3e6cb;
}
.env-status-error {
background-color: #f8d7da;
color: #721c24;
border: 1px solid #f5c6cb;
}
@keyframes pulse {
0% { opacity: 1; }
50% { opacity: 0.5; }
100% { opacity: 1; }
}
</style>
""")
with gr.Row():
with gr.Column(scale=0.5):
question_input = gr.Textbox(
lines=5,
placeholder="Bitte gib deine Frage ein...",
label="Frage",
elem_id="question_input",
show_copy_button=True,
value="Wie kann ich untersuchen, welche Social Media Plattform an meiner Schule am beliebtesten ist, die Ergebnisse in einem Diagramm präsentieren? Welche Fragen könnten sinnvoll sein?"
)
# Erweiterte Modul-Auswahl Dropdown
module_dropdown = gr.Dropdown(
choices=list(MODULE_DESCRIPTIONS.keys()),
value="run",
label="Wähle Funktionsmodul",
interactive=True,
)
# Textfeld für die Modulbeschreibung
module_description = gr.Textbox(
value=MODULE_DESCRIPTIONS["run"],
label="Modulbeschreibung",
interactive=False,
elem_classes="module-info",
)
with gr.Row():
run_button = gr.Button(
"Ausführen", variant="primary", elem_classes="primary"
)
status_output = gr.HTML(
value="<span class='status-indicator status-success'></span> Bereit",
label="Status",
)
token_count_output = gr.Textbox(
label="Tokenanzahl", interactive=False, elem_classes="token-count"
)
# Beispiel-Fragen (Beispiele für eine Social-Media-Projektwoche in der 10. Klasse)
beispiele = [
"Erstelle einen fließend gesprochenen Hörtext, der die Entstehung und Entwicklung von Social-Media-Plattformen erläutert und deren Einfluss auf Kommunikationsgewohnheiten und gesellschaftliche Veränderungen beleuchtet. Ziel ist es, den Zuhörenden einen klaren Überblick über die historischen Meilensteine und aktuellen Trends zu geben. Hinweis: Achte auf eine natürliche Sprechweise, passende Pausen und Betonungen, damit der Hörtext lebendig und informativ wirkt. Gib den Text letztlich als MP3 Datei aus.",
"Generiere ein strukturiertes PDF-Dokument, das die neuesten Trends im Social Media Marketing vorstellt. Beschreibe, wie Unternehmen Plattformen wie Instagram, Facebook und TikTok nutzen, um ihre Marken zu stärken und welche Erfolgsfaktoren dabei eine Rolle spielen. Hinweis: Integriere erklärende Texte, Diagramme und Statistiken, um die Inhalte anschaulich darzustellen. Verwende klare Überschriften und Absätze, um die Lesbarkeit zu erhöhen.",
"Erstelle eine Excel-Tabelle, in der Daten zur Nutzung und zum Engagement verschiedener Social Media Plattformen gesammelt werden. Die Tabelle soll Spalten beinhalten wie Plattformname, Benutzerzahlen, Interaktionsraten (Likes, Shares, Kommentare) und Wachstumsraten. Ergänze die Tabelle durch ein Diagramm, das den Vergleich zwischen den Plattformen visualisiert. Hinweis: Nutze bedingte Formatierungen zur Hervorhebung von Trends und Aenderungen in den Daten, und sorge für eine klare Beschriftung aller Elemente.",
"Vergleiche die aktuelle Nutzung von Facebookj, Twitter und Snapchat und diskutiere, welche Plattform die meisten Lrnmöglichkeiten bietet. Erstelle ein Microsoft Word Dokument im RTF Format.",
"Konzipiere eine PowerPoint-Präsentation mit 8–10 Folien, die aufzeigt, wie Social Media politische Meinungsbildung, öffentliche Diskurse und gesellschaftliche Bewegungen beeinflusst. Jede Folie sollte einen spezifischen Aspekt behandeln, wie z. B. historische Entwicklungen, Key Player, Chancen und Risiken sowie zukünftige Entwicklungen. Hinweis: Gestalte die Folien mit prägnanten Titeln, kurzen Stichpunkten und passenden Grafiken/Bildern, um eine klare und nachvollziehbare Argumentationskette zu präsentieren."
]
gr.Examples(examples=beispiele, inputs=question_input, label="Beispiele")
gr.HTML("""
<div class="footer" id="about">
<p>© 2025 by Sascha Endlicher, M.A.</p>
</div>
""")
with gr.Tabs():
with gr.TabItem("Konversationsprotokoll"):
# Bereich für die Anzeige der Gesprächsaufzeichnungen hinzufügen
with gr.Box():
log_display2 = gr.Markdown(
value="Noch keine Gesprächsaufzeichnungen vorhanden.",
elem_classes="log-display",
)
with gr.Row():
refresh_logs_button2 = gr.Button("Protokoll aktualisieren")
auto_refresh_checkbox2 = gr.Checkbox(
label="Automatische Aktualisierung", value=True, interactive=True
)
clear_logs_button2 = gr.Button(
"Protokoll löschen", variant="secondary"
)
# Ereignisbindung einrichten
run_button.click(
fn=process_with_live_logs,
inputs=[question_input, module_dropdown],
outputs=[token_count_output, status_output, log_display2],
)
# Modul-Auswahl aktualisiert die Beschreibung
module_dropdown.change(
fn=update_module_description,
inputs=module_dropdown,
outputs=module_description,
)
# Ereignisse für das Konversationsprotokoll
refresh_logs_button2.click(
fn=lambda: get_latest_logs(100, LOG_QUEUE), outputs=[log_display2]
)
clear_logs_button2.click(fn=clear_log_file, outputs=[log_display2])
# Steuerung für automatische Aktualisierung
def toggle_auto_refresh(enabled):
if enabled:
return gr.update(every=3)
else:
return gr.update(every=0)
auto_refresh_checkbox2.change(
fn=toggle_auto_refresh,
inputs=[auto_refresh_checkbox2],
outputs=[log_display2],
)
# Standardmäßig keine automatische Aktualisierung der Logs
return app
# Hauptfunktion
def main():
try:
# Logging-System initialisieren
global LOG_FILE
LOG_FILE = setup_logging()
logging.info("EULE Webanwendung gestartet")
# Log-Lesethread starten
log_thread = threading.Thread(
target=log_reader_thread, args=(LOG_FILE,), daemon=True
)
log_thread.start()
logging.info("Log-Lesethread gestartet")
# .env-Datei initialisieren (falls nicht vorhanden)
init_env_file()
app = create_ui()
app.queue()
app.launch(server_name="0.0.0.0", server_port=8373, share=False, favicon_path="../assets/eule-favicon.ico")
except Exception as e:
logging.error(f"Fehler beim Starten der Anwendung: {str(e)}")
print(f"Fehler beim Starten der Anwendung: {str(e)}")
import traceback
traceback.print_exc()
finally:
# Sicherstellen, dass der Log-Thread beendet wird
STOP_LOG_THREAD.set()
STOP_REQUESTED.set()
logging.info("Anwendung geschlossen")
if __name__ == "__main__":
main()