File: //opt/owl/streamlit_app.py
import streamlit as st
import os
import sys
import importlib
from dotenv import load_dotenv
import re
# Stelle sicher, dass das 'owl' Verzeichnis und das 'examples' Verzeichnis im Python-Pfad sind
# Passe die Pfade ggf. an deine Projektstruktur an
base_path = os.path.abspath(os.path.dirname(__file__))
owl_module_path = os.path.join(base_path, 'owl')
examples_path = os.path.join(base_path, 'examples') # Hinzufügen für dynamischen Import
if owl_module_path not in sys.path:
sys.path.insert(0, owl_module_path)
if examples_path not in sys.path:
sys.path.insert(0, examples_path) # examples auch hinzufügen
# Lade Umgebungsvariablen aus .env (im owl-Unterverzeichnis)
# Die construct_society Funktionen in den examples-Modulen sollten dies ggf. auch tun
dotenv_path = os.path.join(owl_module_path, '.env') # Pfad: /opt/owl/owl/.env
load_dotenv(dotenv_path=dotenv_path, override=True) # override=True kann helfen
# Entferne die Debug-Ausgabe
# st.write(f"DEBUG: DEEPSEEK_API_KEY loaded: {os.getenv('DEEPSEEK_API_KEY')}")
# Importiere notwendige Funktionen/Klassen aus dem owl-Projekt
try:
# OwlRolePlaying wird jetzt innerhalb von construct_society verwendet
# run_society wird weiterhin benötigt
from utils.enhanced_role_playing import run_society
# ModelFactory und Typen werden in den example-Modulen importiert
except ImportError as e:
st.error(f"Fehler beim Importieren von owl-Modulen (run_society): {e}")
st.stop()
# --- Modul-Beschreibungen (vereinfacht) ---
# Diese Liste sollte den Dateinamen in examples/ entsprechen (ohne .py)
# Die Beschreibungen können aus webapp.py übernommen werden
AVAILABLE_MODULES = {
"run": "Default mode: Using OpenAI model's default agent collaboration mode...",
"run_mini": "Using OpenAI model with minimal configuration...",
"run_deepseek_de": "Using deepseek model to process German tasks...",
"run_openai_compatible_model": "Using openai compatible model...",
"run_ollama": "Using local ollama model...",
"run_qwen_mini_zh": "Using qwen model with minimal configuration...",
# Füge hier weitere Module hinzu, die in examples/ existieren
}
# --- Streamlit UI ---
st.set_page_config(layout="wide")
st.title("🦉 Owl Streamlit Alternative (dynamischer Import)")
st.sidebar.header("Einstellungen")
# Modulauswahl
module_name = st.sidebar.selectbox(
"Wähle ein Modul:",
options=list(AVAILABLE_MODULES.keys()),
format_func=lambda name: f"{name} - {AVAILABLE_MODULES.get(name, 'Keine Beschreibung')}" # Verwende .get für Sicherheit
)
# Eingabe der Frage/Aufgabe
question = st.text_area("Gib deine Frage oder Aufgabe ein:", height=150)
# Button zum Starten
start_button = st.button("Starte Verarbeitung")
st.divider()
# --- Verarbeitungslogik ---
if start_button and question:
if not module_name:
st.error("Bitte wähle ein Modul aus.")
else:
st.info(f"Verwende Modul: {module_name}")
st.info("Verarbeitung gestartet...")
final_answer_placeholder = st.empty()
history_placeholder = st.expander("Konversationsverlauf anzeigen", expanded=False)
info_placeholder = st.expander("Zusätzliche Informationen", expanded=False)
try:
# 1. Dynamisch das example-Modul importieren
module_path_import = f"examples.{module_name}" # Korrigierter Importpfad
st.write(f"DEBUG: Versuche Modul zu importieren: {module_path_import}") # Debug
try:
module = importlib.import_module(module_path_import)
st.write(f"DEBUG: Modul {module_path_import} erfolgreich importiert.") # Debug
except ImportError as ie:
st.error(f"Fehler beim Importieren des Moduls '{module_path_import}': {ie}")
st.error("Stelle sicher, dass die Datei 'examples/{module_name}.py' existiert und keine Importfehler enthält.")
st.stop()
except Exception as e:
st.error(f"Unerwarteter Fehler beim Importieren des Moduls '{module_path_import}': {e}")
st.exception(e)
st.stop()
# 2. Prüfen und Aufrufen von construct_society im importierten Modul
if not hasattr(module, "construct_society"):
st.error(f"Die Funktion 'construct_society' wurde im Modul '{module_path_import}' nicht gefunden.")
st.error("Jedes Beispiel-Modul muss diese Funktion definieren, um die Agenten zu erstellen.")
st.stop()
st.write(f"DEBUG: Rufe construct_society im Modul {module_path_import} auf.") # Debug
try:
# Hier werden die API-Keys etc. innerhalb von construct_society verwendet
society = module.construct_society(question)
st.write(f"DEBUG: Society-Objekt erfolgreich erstellt.") # Debug
except Exception as e:
st.error(f"Fehler beim Ausführen von 'construct_society' in '{module_path_import}':")
st.exception(e)
st.stop()
# 3. Führe die Society aus (wie zuvor)
# round_limit kann angepasst werden
st.write(f"DEBUG: Rufe run_society auf.") # Debug
final_answer, history, info = run_society(society, round_limit=25)
st.write(f"DEBUG: run_society beendet.") # Debug
# 4. Zeige die Ergebnisse an (wie zuvor)
final_answer_placeholder.success("Verarbeitung abgeschlossen!")
final_answer_placeholder.markdown("**Antwort:**")
final_answer_placeholder.markdown(final_answer) # Oder st.write(final_answer)
history_placeholder.markdown("### Konversationsverlauf")
for i, message_dict in enumerate(history):
# Versuche, die Rolle zu bestimmen (Priorität: 'role', dann 'role_name', dann 'role_type')
role = "Unbekannt"
if isinstance(message_dict, dict):
if "role" in message_dict:
role = message_dict["role"].title()
elif "role_name" in message_dict:
role = message_dict["role_name"].title()
elif "role_type" in message_dict:
# Extrahiere aus Enum-ähnlicher Darstellung wie <RoleType.ASSISTANT: 'assistant'>
role_match = re.search(r"RoleType\.([A-Z_]+)", str(message_dict['role_type']))
if role_match:
role = role_match.group(1).replace("_", " ").title()
content = message_dict.get('content', '')
tool_calls = message_dict.get('tool_calls')
tool_call_id = message_dict.get('tool_call_id')
# Emoji und Chat-Nachricht basierend auf der Rolle
if role.lower() == "user":
emoji = "👤"
name = "User"
elif role.lower() == "assistant":
emoji = "🤖"
name = "Assistant"
elif role.lower() == "tool":
emoji = "🛠️"
name = "Tool"
elif role.lower() == "system": # Explizit System-Rolle behandeln
emoji = "⚙️"
name = "System"
else:
emoji = "❓"
name = role # Fallback auf den extrahierten Rollennamen
with history_placeholder.chat_message(name=name):
st.markdown(f"**{emoji} {name}**")
if content:
st.markdown(content)
# Zeige Tool-Aufrufe an, falls vorhanden (innerhalb der Assistant-Nachricht)
if tool_calls and isinstance(tool_calls, list):
st.markdown("**Tool Calls:**")
# Debug-Ausgaben für die UI-Anzeige
st.write(f"DEBUG [UI]: Found tool_calls in message: {tool_calls}")
for tool_call in tool_calls:
st.write(f"DEBUG [UI]: Processing tool_call item: {tool_call}")
# Sicherstellen, dass tool_call ein dict ist
if isinstance(tool_call, dict):
# Sicherer Zugriff auf verschachtelte Strukturen
function_dict = tool_call.get('function', {})
st.write(f"DEBUG [UI]: Extracted function_dict: {function_dict}")
tool_name = function_dict.get('name', 'Unbekanntes Tool')
st.write(f"DEBUG [UI]: Extracted tool_name: '{tool_name}'")
tool_args = function_dict.get('arguments', '{}')
tool_id = tool_call.get('id', 'Keine ID')
# Wiederhergestellte UI-Anzeige für Tool Calls
st.code(f"ID: {tool_id}\nTool: {tool_name}\nArguments: {tool_args}", language="json")
else:
st.warning(f"Unerwartetes Format für tool_call: {tool_call}")
# Markiere Tool-Antworten
if role.lower() == "tool" and tool_call_id:
st.markdown(f"_(Antwort auf Tool Call ID: {tool_call_id})_")
else:
# Fallback, wenn history kein dict ist
with history_placeholder.chat_message(name="System"):
st.warning(f"Unerwartetes Format im Verlaufseintrag {i}: {message_dict}")
info_placeholder.markdown("### Zusätzliche Informationen")
info_placeholder.json(info)
# Debug-Ausgabe vor dem Download-Bereich
st.write("DEBUG: Ergebnisse angezeigt. Beginne Suche nach Download-Dateien.")
# --- Download-Bereich ---
st.markdown("---")
downloadable_files = set() # Verwende ein Set, um doppelte Buttons zu vermeiden
# Funktion zum Suchen und Validieren von Dateien
def find_and_validate_files(text_to_search):
st.write(f"DEBUG [find_and_validate]: Suche in Text: '{text_to_search[:100]}...'") # Debug
found_files = set()
# Regex, um potenzielle Dateinamen zu finden (etwas breiter gefasst)
potential_filenames = re.findall(r"\b([\w.-]+\.(?:mp3|wav|ogg|pdf|csv|txt|zip|py|json|html|png|jpg|jpeg|gif|docx|xlsx|pptx))\b", text_to_search, re.IGNORECASE)
st.write(f"DEBUG [find_and_validate]: Regex 1 fand: {potential_filenames}") # Debug
for filename in potential_filenames:
# Annahme: Dateien werden relativ zum base_path (/opt/owl) gespeichert
filepath = os.path.abspath(os.path.join(base_path, filename))
st.write(f"DEBUG [find_and_validate]: Prüfe Pfad: {filepath}") # Debug
exists = os.path.exists(filepath)
st.write(f"DEBUG [find_and_validate]: Existiert? {exists}") # Debug
if exists:
found_files.add(filepath)
# Suche auch nach Mustern wie save('filename') im Text (oft in Code-Argumenten)
save_matches = re.findall(r"save\(['\"]([^'\"]+\.(?:mp3|wav|ogg|pdf|csv|txt|zip|py|json|html|png|jpg|jpeg|gif|docx|xlsx|pptx))['\"]\)", text_to_search, re.IGNORECASE)
st.write(f"DEBUG [find_and_validate]: Regex 2 (save) fand: {save_matches}") # Debug
for filename in save_matches:
filepath = os.path.abspath(os.path.join(base_path, filename))
st.write(f"DEBUG [find_and_validate]: Prüfe Pfad (aus save): {filepath}") # Debug
exists = os.path.exists(filepath)
st.write(f"DEBUG [find_and_validate]: Existiert? {exists}") # Debug
if exists:
found_files.add(filepath)
st.write(f"DEBUG [find_and_validate]: Gefundene valide Dateien: {found_files}") # Debug
return found_files
# 1. Suche in der finalen Antwort
downloadable_files.update(find_and_validate_files(final_answer))
# 2. Suche in der History (insbesondere Assistant-Antworten und Tool-Code/Output)
st.write("DEBUG [Download]: Starte Suche in History für Downloads.") # Debug
for message_dict in history:
if isinstance(message_dict, dict):
content = message_dict.get('content', '')
# Suche im normalen Content der Nachricht
st.write(f"DEBUG [Download]: Suche in content: '{content[:50]}...'", ) # Debug
downloadable_files.update(find_and_validate_files(content))
# Suche *zusätzlich* spezifisch in den Argumenten von Code-Ausführungs-Tools
tool_calls = message_dict.get('tool_calls')
if tool_calls and isinstance(tool_calls, list):
st.write(f"DEBUG [Download]: Found tool_calls in message: {tool_calls}") # Debug
for tool_call in tool_calls:
st.write(f"DEBUG [Download]: Processing tool_call item: {tool_call}") # Debug
if isinstance(tool_call, dict):
# Extrahiere Tool-Name und Argumente gemäß der beobachteten Struktur
tool_name = tool_call.get('tool_name', '')
st.write(f"DEBUG [Download]: Extracted tool_name: '{tool_name}'") # Debug
tool_args_dict = tool_call.get('args', {})
st.write(f"DEBUG [Download]: Extracted tool_args_dict: {tool_args_dict}") # Debug
# Prüfe, ob es ein Code-Execution-Tool ist
if 'execute_code' in tool_name.lower(): # Prüfe den korrekt extrahierten Namen
# Extrahiere den Code-String aus dem Argumente-Dictionary
code_string = tool_args_dict.get('code', '')
if code_string:
st.write(f"DEBUG [Download]: Suche in Code-Argumenten für '{tool_name}': '{code_string[:100]}...'", ) # Debug
# Wende die Dateisuche auf den Code-String an
downloadable_files.update(find_and_validate_files(code_string))
else:
st.write(f"DEBUG [Download]: Kein 'code'-Schlüssel in args für {tool_name} gefunden.") # Debug
else:
st.write(f"DEBUG [Download]: Tool '{tool_name}' ist kein Code-Execution-Tool.") # Debug
st.write(f"DEBUG [Download]: Finale Liste herunterladbarer Dateien: {downloadable_files}") # Debug
# 3. Zeige Download-Buttons an
if downloadable_files:
st.markdown("### Dateien zum Herunterladen:")
for filepath in sorted(list(downloadable_files)):
filename = os.path.basename(filepath)
st.write(f"DEBUG [Download Button]: Versuche Button für {filename} ({filepath}) zu erstellen.")
try:
with open(filepath, "rb") as fp:
st.download_button(
label=f"Download {filename}",
data=fp,
file_name=filename,
mime=None # Streamlit versucht, den Mime-Typ automatisch zu erkennen
)
except Exception as e:
st.error(f"Fehler beim Lesen der Datei {filename} für den Download: {e}")
# --- Ende Download-Bereich ---
except Exception as e:
st.error(f"Ein unerwarteter Fehler ist während der Verarbeitung aufgetreten:")
st.exception(e)
elif start_button and not question:
st.warning("Bitte gib eine Frage oder Aufgabe ein.")
st.sidebar.markdown("---")
st.sidebar.markdown("Entwickelt als Alternative zur Gradio-App.")