turns-00039.parquet:1566
8c0638fb4b0ff4695a9c38f8
turn 1/1gpt-4o-mini-2024-07-18GermanGermany2848 words
degenerate_repetitionAbsentFinal dense release
USER
prüfe dieses skript auf fehler und probleme. Erstelle eine liste mit allen gefundenen problemen.
import sys
import os
import re
import subprocess
import json
import requests
import threading
import sqlite3
import time
import logging
from PyQt5.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLineEdit,
QPushButton, QLabel, QDialog, QFormLayout, QSlider, QMessageBox, QComboBox,
QTextBrowser, QAction, QScrollArea, QListWidget, QListWidgetItem, QProgressBar
)
from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QObject
from PyQt5.QtGui import QTextCursor, QColor, QFont, QTextCharFormat, QRegularExpression, QSyntaxHighlighter
from PyQt5.Qt import QUrl, QClipboard
# Logging-Konfiguration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Database Manager
class DatabaseManager:
def __init__(self, db_name="autollm_context.db", max_context_length=10000):
self.conn = sqlite3.connect(db_name)
self.max_context_length = max_context_length
self.create_table()
def create_table(self):
self.conn.execute('''
CREATE TABLE IF NOT EXISTS context (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sender TEXT NOT NULL,
message TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.commit()
def add_message(self, sender, message):
self.conn.execute("INSERT INTO context (sender, message) VALUES (?, ?)", (sender, message))
self.conn.commit()
self.ensure_context_limit()
def get_context(self, max_length=None):
if max_length is None:
max_length = self.max_context_length
cursor = self.conn.cursor()
cursor.execute("SELECT sender, message FROM context ORDER BY id DESC")
rows = cursor.fetchall()
context = []
current_length = 0
for row in reversed(rows):
msg = f"{row[0]}: {row[1]}"
msg_length = len(msg)
if current_length + msg_length > max_length:
break
context.insert(0, msg)
current_length += msg_length
return "\n".join(context)
def clear_context(self):
self.conn.execute("DELETE FROM context")
self.conn.commit()
def ensure_context_limit(self):
cursor = self.conn.cursor()
cursor.execute("SELECT SUM(LENGTH(message)) FROM context")
total_length = cursor.fetchone()[0] or 0
if total_length > self.max_context_length:
cursor.execute("DELETE FROM context WHERE id IN (SELECT id FROM context ORDER BY id ASC LIMIT 100)")
self.conn.commit()
def close_connection(self):
self.conn.close()
# Context Manager mit Optimierungen
class ContextManager(QObject):
summarization_requested = pyqtSignal(str)
def __init__(self, ollama_controller):
super().__init__()
self.db = DatabaseManager()
self.summary = ""
self.ollama = ollama_controller
self.ollama.summarization_received.connect(self.handle_summarization_response)
self.ollama.response_received.connect(self.receive_summary)
self.summarize_request_event = threading.Event()
self.summarization_result = ""
self.lock = threading.Lock()
def add_message(self, sender, message):
self.db.add_message(sender, message)
self.check_and_summarize_context()
def get_context(self):
with self.lock:
if self.summary:
return self.summary + "\n" + self.db.get_context()
return self.db.get_context()
def clear_context(self):
self.db.clear_context()
with self.lock:
self.summary = ""
def check_and_summarize_context(self):
context = self.db.get_context()
if len(context) > self.db.max_context_length * 0.8:
self.summarize_context(context)
def summarize_context(self, context_text):
summary_prompt = (
"Fasse bitte den folgenden Kontext zusammen, um Platz für neue Informationen zu schaffen. "
"Behalte dabei alle wichtigen Details und wesentlichen Punkte bei:\n\n"
f"{context_text}"
)
self.summarization_result = ""
self.summarize_request_event.clear()
self.ollama.send_request(summary_prompt, {
"temperature": 0.5,
"max_new_tokens": 1000,
"additional_options": {"stream": False, "format": "json"}
})
threading.Thread(target=self.wait_for_summary, daemon=True).start()
def wait_for_summary(self):
if self.summarize_request_event.wait(timeout=60):
if self.summarization_result:
with self.lock:
self.summary = self.summarization_result
self.db.clear_context()
self.db.add_message("System", f"Zusammenfassung des vorherigen Kontexts:\n{self.summary}")
else:
logging.error("Timeout bei der Kontextzusammenfassung.")
self.summarize_request_event.set()
def handle_summarization_response(self, response):
if "response" in response and response["response"]:
summary = response["response"]
if summary:
with self.lock:
self.summarization_result = summary
else:
logging.error("Keine Zusammenfassungsantwort erhalten.")
self.summarize_request_event.set()
def receive_summary(self, response):
if "response" in response and response["response"]:
summary = response["response"]
if summary:
with self.lock:
self.summary = summary
self.db.clear_context()
self.db.add_message("System", f"Zusammenfassung des vorherigen Kontexts:\n{self.summary}")
self.summarize_request_event.set()
# Code Highlighter
class CodeHighlighter(QSyntaxHighlighter):
def __init__(self, parent=None):
super().__init__(parent)
self.highlightingRules = []
format_keyword = QTextCharFormat()
format_keyword.setForeground(QColor("orange"))
keywords = [
"if", "else", "for", "while", "def", "return", "class", "import", "from",
"try", "except", "elif", "with", "as", "and", "or", "not", "in", "is", "lambda"
]
for keyword in keywords:
pattern = QRegularExpression(r"\b" + keyword + r"\b")
self.highlightingRules.append((pattern, format_keyword))
def highlightBlock(self, text):
for pattern, format in self.highlightingRules:
match_iterator = pattern.globalMatch(text)
while match_iterator.hasNext():
match = match_iterator.next()
self.setFormat(match.capturedStart(), match.capturedLength(), format)
# Ollama Controller mit Verbesserungen
class OllamaController(QObject):
server_status_changed = pyqtSignal(bool)
response_received = pyqtSignal(dict)
error_occurred = pyqtSignal(str)
llm_list_updated = pyqtSignal(list)
summarization_received = pyqtSignal(dict)
def __init__(self):
super().__init__()
self.process = None
self.selected_llm = None
self.api_port = 11434
self.server_started = False
self.interrupt_event = threading.Event()
self.thread_lock = threading.Lock()
self.check_server_status()
def check_server_status(self):
try:
response = requests.get(f"http://localhost:{self.api_port}/api/tags", timeout=2)
if response.status_code == 200:
self.server_started = True
self.server_status_changed.emit(True)
self.update_llm_list()
else:
self.server_started = False
self.server_status_changed.emit(False)
self.error_occurred.emit("API-Prüfung fehlgeschlagen.")
except requests.exceptions.RequestException:
self.server_started = False
self.server_status_changed.emit(False)
self.error_occurred.emit("Verbindung zum Server fehlgeschlagen.")
def update_llm_list(self):
llms = self.get_installed_llms()
if llms:
self.llm_list_updated.emit(llms)
if not self.selected_llm:
self.selected_llm = llms[0]
else:
self.llm_list_updated.emit([])
self.selected_llm = None
def start_server(self):
if not self.is_server_running():
try:
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
self.process = subprocess.Popen(
['ollama', 'serve'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
shell=False,
startupinfo=startupinfo,
cwd=os.path.expanduser("~")
)
threading.Thread(target=self.delayed_check, daemon=True).start()
except Exception as e:
self.error_occurred.emit(f"Fehler beim Starten des Servers: {e}")
else: # Server läuft bereits
self.server_started = True
self.server_status_changed.emit(True)
self.update_llm_list()
def delayed_check(self):
time.sleep(5) # Wartezeit, damit der Server ordentlich starten kann
self.check_server_status()
def stop_server(self):
if self.is_server_running() and self.process:
try:
self.interrupt_processing()
self.process.terminate()
try:
self.process.wait(timeout=10)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait(timeout=5)
self.server_started = False
self.server_status_changed.emit(False)
self.process = None
self.selected_llm = None
self.llm_list_updated.emit([])
except Exception as e:
self.error_occurred.emit(f"Fehler beim Stoppen des Servers: {e}")
else:
self.error_occurred.emit("Server läuft nicht oder Prozess ist nicht gültig.")
def get_installed_llms(self):
try:
response = requests.get(f"http://localhost:{self.api_port}/api/tags", timeout=2)
if response.status_code == 200:
data = response.json()
if "models" in data:
return [model["name"] for model in data["models"] if "name" in model]
return []
self.error_occurred.emit(f"API-Status: {response.status_code}")
return []
except requests.exceptions.Timeout:
self.error_occurred.emit("API-Timeout - Server antwortet nicht")
return []
except requests.exceptions.ConnectionError:
self.error_occurred.emit("Keine Verbindung zum Server möglich")
return []
except Exception as e:
self.error_occurred.emit(f"Fehler beim Abrufen der Modelle: {str(e)}")
return []
def is_server_running(self):
try:
response = requests.get(f"http://localhost:{self.api_port}/api/tags", timeout=2)
return response.status_code == 200
except Exception:
return False
def set_selected_llm(self, llm_name):
self.selected_llm = llm_name
def send_request(self, prompt, parameters):
if not self.is_server_running():
self.error_occurred.emit("Ollama-Server läuft nicht.")
return
if not self.selected_llm:
self.error_occurred.emit("Kein LLM ausgewählt.")
return
threading.Thread(target=self._handle_request, args=(prompt, parameters), daemon=True).start()
def _handle_request(self, prompt, parameters):
self.interrupt_event.clear()
try:
url = f"http://localhost:{self.api_port}/api/generate"
payload = {
"model": self.selected_llm,
"prompt": prompt,
"options": {
"temperature": parameters.get("temperature", 0.5),
"num_predict": parameters.get("max_new_tokens", 1500),
**parameters.get("additional_options", {})
},
"stream": parameters.get("additional_options", {}).get("stream", True)
}
response = requests.post(url, json=payload, stream=True)
response.raise_for_status() # Ausnahme bei schlechten Statuscodes auslösen
for line in response.iter_lines():
if self.interrupt_event.is_set():
raise InterruptedError("Anfrage vom Benutzer unterbrochen.")
if line:
decoded_line = line.decode('utf-8')
try:
data = json.loads(decoded_line)
if "error" in data:
raise Exception(f"Ollama API-Fehler: {data['error']}")
self.response_received.emit(data)
except json.JSONDecodeError as e:
self.error_occurred.emit(f"Ungültige JSON-Antwort: {e}")
return
except InterruptedError:
logging.warning("Anfrage abgebrochen.")
self.error_occurred.emit("Anfrage abgebrochen.")
except requests.exceptions.RequestException as e:
self.error_occurred.emit(f"API-Anfrage fehlgeschlagen: {e}")
except Exception as e:
self.error_occurred.emit(f"Fehler bei der Verarbeitung der Anfrage: {e}")
# Task Manager mit Verbesserungen
class TaskManager(QObject):
task_completed = pyqtSignal(str)
task_failed = pyqtSignal(str)
progress_updated = pyqtSignal(int, int) # (completed, total)
final_answer_ready = pyqtSignal(str)
def __init__(self, ollama_controller, context_manager):
super().__init__()
self.ollama = ollama_controller
self.context_manager = context_manager
self.task_queue = []
self.completed_tasks = []
self.failed_tasks = []
self.current_task = None
self.total_tasks = 0
self.completed_count = 0
self.finalizing = False
self.ollama.response_received.connect(self.handle_response)
self.ollama.response_start.connect(self.on_response_start)
self.ollama.response_end.connect(self.on_response_end)
self.ollama.error_occurred.connect(self.on_error)
def add_task(self, user_request):
breakdown_prompt = (
"Erstelle einen detaillierten Plan, um die folgende Anfrage in sinnvolle Teilschritte zu zerlegen. "
"Formuliere jede Teilaufgabe als eine klare, eigenständige Aufgabenstellung, die in einem Bearbeitungsschritt abgeschlossen werden kann. "
"Gib die Schritte in einer nummerierten Liste zurück.\n\n"
f"Anfrage: {user_request}"
)
self.current_user_request = user_request
self.ollama.send_request(breakdown_prompt, {
"temperature": 0.5,
"max_new_tokens": 1500,
"additional_options": {"stream": False, "format": "json"}
})
self.total_tasks = 0
self.completed_count = 0
self.task_queue.clear()
self.completed_tasks.clear()
self.failed_tasks.clear()
self.current_task = None
self.finalizing = False
self.progress_updated.emit(self.completed_count, self.total_tasks)
def handle_response(self, response):
if self.finalizing:
if "response" in response:
final_answer = response["response"]
if final_answer:
self.final_answer_ready.emit(final_answer)
return
if "response" in response:
chunk = response["response"]
steps = self.parse_breakdown(chunk)
if steps:
self.task_queue.extend(steps)
self.total_tasks += len(steps)
self.progress_updated.emit(self.completed_count, self.total_tasks)
self.process_next_task()
else:
self.task_failed.emit("Fehler beim Zerlegen der Anfrage in Teilschritte.")
else:
self.error_occurred.emit("Keine Antwort von LLM erhalten.")
def parse_breakdown(self, breakdown_text):
steps = re.findall(r'^\d+\.\s+(.*)', breakdown_text, re.MULTILINE)
return steps
def process_next_task(self):
if self.current_task is not None:
return
if not self.task_queue:
if not self.finalizing:
self.finalize()
return
self.current_task = self.task_queue.pop(0)
task_prompt = (
f"Bearbeite bitte die folgende Aufgabe in einem Schritt:\n\n{self.current_task}\n\n"
"Gib das Ergebnis in einer klaren und präzisen Antwort zurück."
)
self.ollama.send_request(task_prompt, {
"temperature": 0.5,
"max_new_tokens": 1500,
"additional_options": {"stream": False, "format": "json"}
})
def finalize(self):
if self.finalizing:
return
self.finalizing = True
results = "\n".join([f"{i+1}. {task}" for i, task in enumerate(self.completed_tasks)])
final_prompt = (
f"Fasse bitte alle Ergebnisse der folgenden Aufgaben zusammen, um eine finale Antwort auf die ursprüngliche Anfrage zu erstellen:\n\n"
f"Ursprüngliche Anfrage: {self.current_user_request}\n\n"
f"Ergebnisse der Aufgaben:\n{results}\n\n"
"Erstelle nun eine ausführliche und zusammenhängende Antwort basierend auf diesen Ergebnissen."
)
self.ollama.send_request(final_prompt, {
"temperature": 0.5,
"max_new_tokens": 2000,
"additional_options": {"stream": False, "format": "json"}
})
self.finalizing = True
def on_response_start(self):
pass
def on_response_end(self):
if not self.current_task and not self.task_queue and self.finalizing:
self.finalize()
def on_error(self, error_message):
if self.current_task:
self.failed_tasks.append(self.current_task)
self.completed_count += 1
self.progress_updated.emit(self.completed_count, self.total_tasks)
self.task_failed.emit(f"Fehler bei der Aufgabe '{self.current_task}': {error_message}")
self.current_task = None
self.process_next_task()
else:
self.task_failed.emit(f"Fehler: {error_message}")
def handle_summarization_response(self, response):
if "response" in response:
summary = response["response"]
if summary:
self.context_manager.add_message("LLM", f"Zusammenfassung: {summary}")
self.finalize()
# AutoLLM App mit Verbesserungen
class AutoLLMApp(QWidget):
def __init__(self):
super().__init__()
self.setWindowTitle("AutoLLM")
self.resize(1200, 800)
self.ollama = OllamaController()
self.context_manager = ContextManager(self.ollama)
self.current_parameters = {"temperature": 0.5, "max_new_tokens": 1500, "additional_options": {}}
self.init_ui()
self.ollama.start_server()
self.enable_ui(False)
# Task Manager
self.task_manager = TaskManager(self.ollama, self.context_manager)
self.task_manager.task_completed.connect(self.on_task_completed_step)
self.task_manager.task_failed.connect(self.on_task_failed_step)
self.task_manager.progress_updated.connect(self.update_progress)
self.task_manager.final_answer_ready.connect(self.display_final_answer)
# Verbinden der Signale von Ollama
self.ollama.server_status_changed.connect(self.update_server_status)
self.ollama.response_received.connect(self.handle_llm_response)
self.ollama.error_occurred.connect(self.show_error)
self.ollama.llm_list_updated.connect(self.update_llm_list)
self.ollama.summarization_received.connect(self.handle_summarization_response)
self.ollama.response_start.connect(self.handle_llm_response_start)
self.ollama.response_end.connect(self.handle_llm_response_end)
# Initialisieren des Code-Highlightings nach der Erstellung von chat_area
self.highlighter = CodeHighlighter(self.chat_area.document())
# Für die Speicherung der LLM-Antworten
self.llm_response_buffer = ""
self.current_display_index = 0
# Für blinkende "..."
self.blink_timer = QTimer()
self.blink_timer.timeout.connect(self.toggle_ellipsis)
self.ellipsis_visible = True
def init_ui(self):
main_layout = QHBoxLayout()
# Linke Seite: Chat und Fortschritt
left_layout = QVBoxLayout()
# Chat-Bereich mit Scrollbar
self.chat_area = QTextBrowser()
self.chat_area.setReadOnly(True)
self.chat_area.setFont(QFont("Courier", 10))
self.chat_area.setContextMenuPolicy(Qt.CustomContextMenu)
self.chat_area.customContextMenuRequested.connect(self.handle_context_menu)
chat_scroll_area = QScrollArea()
chat_scroll_area.setWidgetResizable(True)
chat_scroll_area.setWidget(self.chat_area)
left_layout.addWidget(chat_scroll_area)
# Eingabefeld und Sende-Schaltfläche
input_layout = QHBoxLayout()
self.input_field = QLineEdit()
self.input_field.setPlaceholderText("Geben Sie Ihre Nachricht hier ein...")
self.input_field.returnPressed.connect(self.send_message)
self.send_button = QPushButton("Senden")
self.send_button.clicked.connect(self.send_message)
input_layout.addWidget(self.input_field)
input_layout.addWidget(self.send_button)
left_layout.addLayout(input_layout)
# Unterbrechen-Schaltfläche
self.interrupt_button = QPushButton("Unterbrechen")
self.interrupt_button.clicked.connect(self.interrupt_processing)
left_layout.addWidget(self.interrupt_button)
# Server-Überwachungsbereich
server_layout = QHBoxLayout()
self.server_status = QLabel("🔴 Server Gestoppt")
self.server_button = QPushButton("Server Starten")
self.server_button.clicked.connect(self.toggle_server)
server_layout.addWidget(self.server_status)
server_layout.addWidget(self.server_button)
left_layout.addLayout(server_layout)
# LLM-Auswahl
llm_layout = QHBoxLayout()
self.llm_label = QLabel("LLM auswählen:")
self.llm_combo = QComboBox()
self.llm_combo.setEnabled(False)
self.llm_combo.currentTextChanged.connect(self.change_llm)
llm_layout.addWidget(self.llm_label)
llm_layout.addWidget(self.llm_combo)
left_layout.addLayout(llm_layout)
# Einstellungen-Schaltfläche
self.settings_button = QPushButton("Einstellungen")
self.settings_button.clicked.connect(self.open_settings)
left_layout.addWidget(self.settings_button)
# Fortschrittsbereich
self.task_progress = QProgressBar()
self.task_progress.setValue(0)
left_layout.addWidget(self.task_progress)
# Plan- und Schritte-Bereich
plan_layout = QVBoxLayout()
self.plan_label = QLabel("Aktueller Plan:")
self.plan_list = QListWidget()
self.plan_list.setFixedHeight(200)
plan_layout.addWidget(self.plan_label)
plan_layout.addWidget(self.plan_list)
left_layout.addLayout(plan_layout)
# Interaktions-Schaltflächen
interaction_layout = QHBoxLayout()
self.edit_button = QPushButton("Aufgabe Bearbeiten")
self.edit_button.clicked.connect(self.edit_task)
self.skip_button = QPushButton("Aufgabe Überspringen")
self.skip_button.clicked.connect(self.skip_task)
interaction_layout.addWidget(self.edit_button)
interaction_layout.addWidget(self.skip_button)
left_layout.addLayout(interaction_layout)
main_layout.addLayout(left_layout, 3)
# Rechte Seite: Optionaler Kontext- und Log-Bereich (kann erweitert werden)
self.setLayout(main_layout)
def enable_ui(self, enabled):
self.input_field.setEnabled(enabled)
self.send_button.setEnabled(enabled)
self.interrupt_button.setEnabled(enabled)
self.settings_button.setEnabled(enabled)
self.llm_combo.setEnabled(enabled and self.llm_combo.count() > 0)
def update_server_status(self, is_running):
if is_running:
self.server_status.setText("🟢 Server Läuft")
self.server_button.setText("Server Stoppen")
self.enable_ui(True)
else:
if self.ollama.server_started:
self.enable_ui(False)
self.llm_combo.clear()
self.llm_combo.addItem("Keine LLMs Installiert")
self.ollama.set_selected_llm(None)
QMessageBox.warning(self, "Server Gestoppt", "Der Ollama-Server wurde unerwartet gestoppt.")
self.server_status.setText("🔴 Server Gestoppt")
self.server_button.setText("Server Starten")
def update_llm_list(self, llm_list):
self.llm_combo.blockSignals(True)
self.llm_combo.clear()
if llm_list:
self.llm_combo.addItems(llm_list)
self.ollama.set_selected_llm(llm_list[0])
self.llm_combo.setEnabled(True)
else:
self.llm_combo.addItem("Keine LLMs Installiert")
self.ollama.set_selected_llm(None)
self.llm_combo.setEnabled(False)
self.llm_combo.blockSignals(False)
self.enable_ui(bool(llm_list))
def change_llm(self, llm_name):
if llm_name != "Keine LLMs Installiert":
self.ollama.set_selected_llm(llm_name)
else:
self.ollama.set_selected_llm(None)
def show_error(self, message):
logging.error(message)
self.append_message("System", message, message_type="error")
def send_message(self):
user_message = self.input_field.text().strip()
if user_message:
self.context_manager.add_message("Benutzer", user_message)
self.append_message("Benutzer", user_message)
self.input_field.clear()
self.task_manager.add_task(user_message)
def append_message(self, sender, message, message_type="normal"):
if message_type == "typing":
if sender == "LLM":
self.chat_area.append('<p style="color: green;"><b>LLM:</b> <i>...</i></p>')
else:
if sender == "Benutzer":
color = "blue"
prefix = "<b>Benutzer:</b> "
elif sender == "LLM":
color = "green"
prefix = "<b>LLM:</b> "
else:
color = "gray"
prefix = "<b>System:</b> "
formatted_message = self.format_message(message)
if message_type == "error":
html_message = f'<p style="color: red;">{prefix}{formatted_message}</p>'
elif message_type == "warning":
html_message = f'<p style="color: orange;">{prefix}{formatted_message}</p>'
else:
html_message = f'<p style="color: {color};">{prefix}{formatted_message}</p>'
self.chat_area.append(html_message)
self.chat_area.moveCursor(QTextCursor.End)
def format_message(self, message):
# Formatierung von Links
message = re.sub(r'(https?://\S+)', r'<a href="\1">\1</a>', message)
# Formatierung von Codeblöcken
message = re.sub(
r'```(.*?)```',
r'<pre style="background-color:#f0f0f0; padding:5px; font-family: monospace; white-space: pre-wrap;"><code>\1</code></pre>',
message,
flags=re.DOTALL | re.MULTILINE
)
return message
def handle_link_click(self, url):
QDesktopServices.openUrl(QUrl(url))
def handle_context_menu(self, position):
cursor = self.chat_area.cursorForPosition(position)
cursor.select(QTextCursor.WordUnderCursor)
selected_text = cursor.selectedText()
menu = self.chat_area.createStandardContextMenu()
if re.match(r'^https?://', selected_text):
copy_action = QAction("Link kopieren", self)
copy_action.triggered.connect(lambda: self.copy_to_clipboard(selected_text))
menu.addAction(copy_action)
elif selected_text.startswith("```") and selected_text.endswith("```"):
code = selected_text.strip("`")
copy_code_action = QAction("Code kopieren", self)
copy_code_action.triggered.connect(lambda: self.copy_to_clipboard(code))
menu.addAction(copy_code_action)
menu.exec_(self.chat_area.mapToGlobal(position))
def copy_to_clipboard(self, text):
clipboard = QApplication.clipboard()
clipboard.setText(text)
QMessageBox.information(self, "Kopiert", "Text wurde in die Zwischenablage kopiert.")
def interrupt_processing(self):
self.ollama.interrupt_processing()
self.append_message("System", "LLM-Bearbeitung wurde unterbrochen.")
def toggle_server(self):
if self.ollama.is_server_running():
reply = QMessageBox.question(self, 'Server stoppen',
'Möchten Sie den Ollama-Server wirklich stoppen?',
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.Yes:
self.ollama.stop_server()
else:
self.ollama.start_server()
def open_settings(self):
settings_dialog = SettingsDialog(self.current_parameters, self)
settings_dialog.settings_updated.connect(self.update_parameters)
settings_dialog.exec_()
def update_parameters(self, params):
self.current_parameters = params
QMessageBox.information(self, "Einstellungen aktualisiert", "LLM-Parameter wurden aktualisiert.")
def handle_llm_response_start(self):
self.llm_response_buffer = ""
self.current_display_index = 0
self.chat_area.moveCursor(QTextCursor.End)
self.append_message("LLM", "", message_type="typing")
self.start_ellipsis_blink()
def handle_llm_response_char(self, response):
if not response.get("stream", True):
if "response" in response:
final_answer = response["response"]
if final_answer:
self.append_message("LLM", final_answer)
else:
if "response" in response:
char = response["response"]
self.llm_response_buffer += char
cursor = self.chat_area.textCursor()
cursor.movePosition(QTextCursor.End)
cursor.select(QTextCursor.BlockUnderCursor)
cursor.removeSelectedText()
formatted_message = f'<p style="color: green;"><b>LLM:</b> {self.llm_response_buffer}</p>'
self.chat_area.insertHtml(formatted_message)
self.chat_area.moveCursor(QTextCursor.End)
def handle_llm_response_end(self):
self.stop_ellipsis_blink()
def start_ellipsis_blink(self):
self.blink_timer.start(500)
self.ellipsis_visible = True
def toggle_ellipsis(self):
cursor = self.chat_area.textCursor()
cursor.movePosition(QTextCursor.End)
if self.ellipsis_visible:
self.chat_area.insertHtml(" ...")
else:
self.chat_area.insertHtml(" ")
self.ellipsis_visible = not self.ellipsis_visible
def stop_ellipsis_blink(self):
self.blink_timer.stop()
def on_task_completed_step(self, result):
self.task_manager.on_task_completed_step(result)
self.append_message("LLM", f"Ergebnis: {result}")
self.update_plan_list()
def on_task_failed_step(self, error):
self.task_manager.on_task_failed_step(error)
self.append_message("System", f"Fehler: {error}", message_type="error")
self.update_plan_list()
def update_progress(self, completed, total):
if total == 0:
self.task_progress.setValue(0)
else:
progress = int((completed / total) * 100)
self.task_progress.setValue(progress)
self.update_plan_list()
def update_plan_list(self):
self.plan_list.clear()
for i, task in enumerate(self.task_manager.completed_tasks, start=1):
item = QListWidgetItem(f"{i}. {task}")
item.setCheckState(Qt.Checked)
self.plan_list.addItem(item)
if self.task_manager.current_task:
current_index = len(self.task_manager.completed_tasks) + 1
item = QListWidgetItem(f"{current_index}. {self.task_manager.current_task} (läuft)")
item.setForeground(QColor("blue"))
self.plan_list.addItem(item)
for i, task in enumerate(self.task_manager.task_queue, start=len(self.task_manager.completed_tasks) + 2):
item = QListWidgetItem(f"{i}. {task}")
self.plan_list.addItem(item)
for i, task in enumerate(self.task_manager.failed_tasks, start=len(self.task_manager.completed_tasks) + len(self.task_manager.task_queue) + 2):
item = QListWidgetItem(f"{i}. {task} (Fehler)")
item.setForeground(QColor("red"))
self.plan_list.addItem(item)
def display_final_answer(self, answer):
self.append_message("LLM", f"Finale Antwort: {answer}")
def edit_task(self):
if self.task_manager.current_task:
text, ok = QInputDialog.getText(self, "Aufgabe Bearbeiten", "Bearbeite die aktuelle Aufgabe:", QLineEdit.Normal, self.task_manager.current_task)
if ok and text.strip():
self.task_manager.current_task = text.strip()
self.append_message("System", f"Aufgabe bearbeitet: {self.task_manager.current_task}")
def skip_task(self):
if self.task_manager.current_task:
skipped_task = self.task_manager.current_task
self.task_manager.failed_tasks.append(skipped_task)
self.task_manager.completed_count += 1
self.progress_updated.emit(self.task_manager.completed_count, self.task_manager.total_tasks)
self.append_message("System", f"Aufgabe übersprungen: {skipped_task}", message_type="warning")
self.task_manager.current_task = None
self.task_manager.process_next_task()
# Settings Dialog
class SettingsDialog(QDialog):
settings_updated = pyqtSignal(dict)
def __init__(self, current_settings, parent=None):
super().__init__(parent)
self.setWindowTitle("Einstellungen")
self.current_settings = current_settings
self.init_ui()
def init_ui(self):
layout = QFormLayout()
# Temperature Slider
self.temperature_slider = QSlider(Qt.Horizontal)
self.temperature_slider.setRange(0, 100)
self.temperature_slider.setValue(int(self.current_settings.get("temperature", 0.5) * 100))
self.temperature_slider.valueChanged.connect(self.update_temperature_label)
self.temperature_label = QLabel(f"{self.current_settings.get('temperature', 0.5):.2f}")
temp_layout = QHBoxLayout()
temp_layout.addWidget(self.temperature_slider)
temp_layout.addWidget(self.temperature_label)
layout.addRow("Temperatur:", temp_layout)
# Output Length Slider
self.output_slider = QSlider(Qt.Horizontal)
self.output_slider.setRange(100, 2000)
self.output_slider.setValue(self.current_settings.get("max_new_tokens", 1500))
self.output_slider.valueChanged.connect(self.update_output_label)
self.output_label = QLabel(str(self.current_settings.get("max_new_tokens", 1500)))
output_layout = QHBoxLayout()
output_layout.addWidget(self.output_slider)
output_layout.addWidget(self.output_label)
layout.addRow("Ausgabe Länge:", output_layout)
# Zusätzliche Optionen
self.additional_options_box = QLabel("Weitere Optionen (JSON):")
self.additional_options_input = QLineEdit()
self.additional_options_input.setPlaceholderText('z.B. "top_p": 0.9, "frequency_penalty": 1.0')
layout.addRow(self.additional_options_box, self.additional_options_input)
# Save and Cancel Buttons
buttons_layout = QHBoxLayout()
self.save_button = QPushButton("Speichern")
self.save_button.clicked.connect(self.save_settings)
self.cancel_button = QPushButton("Abbrechen")
self.cancel_button.clicked.connect(self.reject)
buttons_layout.addWidget(self.save_button)
buttons_layout.addWidget(self.cancel_button)
layout.addRow(buttons_layout)
self.setLayout(layout)
def update_temperature_label(self, value):
temperature = value / 100
self.temperature_label.setText(f"{temperature:.2f}")
def update_output_label(self, value):
self.output_label.setText(str(value))
def save_settings(self):
try:
additional_text = self.additional_options_input.text().strip()
additional_options = {}
if additional_text:
additional_options = json.loads("{" + additional_text + "}")
temperature = self.temperature_slider.value() / 100
output_length = self.output_slider.value()
settings = {
"temperature": temperature,
"max_new_tokens": output_length,
"additional_options": additional_options
}
self.settings_updated.emit(settings)
self.accept()
except json.JSONDecodeError:
QMessageBox.warning(self, "Ungültiges JSON", "Bitte geben Sie gültiges JSON für die zusätzlichen Optionen ein.")
# Vollständiges Skript-Ende
if __name__ == "__main__":
app = QApplication(sys.argv)
app.setStyleSheet("""
QWidget {
background-color: #2b2b2b;
color: #ffffff;
}
QPushButton {
background-color: #3c3f41;
border: none;
padding: 5px;
color: #ffffff;
border-radius: 3px;
}
QPushButton:hover {
background-color: #4b4e52;
}
QLineEdit {
background-color: #3c3f41;
border: 1px solid #5c5c5c;
padding: 5px;
color: #ffffff;
}
QTextBrowser {
background-color: #2b2b2b;
border: 1px solid #5c5c5c;
padding: 5px;
color: #ffffff;
}
QComboBox {
background-color: #3c3f41;
border: 1px solid #5c5c5c;
padding: 5px;
color: #ffffff;
}
QLabel {
color: #ffffff;
}
QListWidget {
background-color: #2b2b2b;
border: 1px solid #5c5c5c;
color: #ffffff;
}
QProgressBar {
border: 1px solid #5c5c5c;
text-align: center;
color: #ffffff;
}
QProgressBar::chunk {
background-color: #4caf50;
}
""")
window = AutoLLMApp()
window.show()
sys.exit(app.exec_())ASSISTANT
Hier sind einige mögliche Probleme und Anmerkungen, die aus dem gegebenen Skript hervorgehen:
1. **Datenbankverbindung**:
- Es gibt keine Fehlerbehandlung beim Herstellen der Verbindung zur SQLite-Datenbank. Wenn die Datei nicht erstellt werden kann oder ein anderer Fehler auftritt, wird das nicht gefangen.
2. **Thread-Sicherheit**:
- Es gibt mehrere Stellen, an denen auf gemeinsame Ressourcen zugegriffen wird, z. B. beim Speichern und Abrufen von Context in der `DatabaseManager`-Klasse. Obwohl ein Lock verwendet wird, könnten potentielle Race Conditions auftreten.
3. **Sich wiederholende API-Anfragen**:
- Bei der Überprüfung des Serverstatus wird mindestens dreimal hintereinander eine API-Anfrage an denselben Endpunkt gesendet. Es wäre besser, diese Logik zu konsolidieren, um übermäßige Anfragen an den Server zu vermeiden.
4. **Fehlende Fehlerbehandlung**:
- In der `OllamaController`-Klasse gibt es eine Vielzahl von API-Anfragen, aber nicht alle haben umfassende Fehlerbehandlung in Bezug auf spezifische Fehlercodes.
5. **Eingabevalidierung für den JSON-Editor**:
- Der JSON-Parser für zusätzliche Optionen im `SettingsDialog` akzeptiert keine ungültigen JSON-Daten. Dies könnte zu einem Absturz führen, wenn der Benutzer ein ungültiges JSON übergibt.
6. **Unzureichende Rückmeldungen für den Benutzer**:
- In mehreren Situationen, z. B. beim Scheitern einer API-Anfrage oder bei der Unterbrechung einer Anfrage, könnte eine stärkere Kommunikation mit dem Benutzer in Form von Benachrichtigungen oder Warnungen hilfreich sein.
7. **Code-Duplizierung**:
- In der `send_request`-Methode von `OllamaController` und in anderen Stellen wird Funktionalität zur Behandlung von Serverantworten und Fehlern in verschiedenen Methoden dupliziert. Dies könnte z. B. durch Nutzung von Hilfsfunktionen optimiert werden.
8. **Zugriff auf die GUI von mehreren Threads**:
- PyQt5 ist nicht threadsicher. Es sollten Mechanismen wie `pyqtSignal` verwendet werden, um die Kommunikation zwischen Threads (wie beim Abrufen von Serverdaten) und der Haupt-GUI in die Quittungen zu lösen, um potenzielle Race Conditions oder Fehler zu vermeiden.
9. **Dokumentation und Kommentare**:
- Einige Funktionen sind dokumentiert, aber möglicherweise wäre etwas mehr Kontext in Form von Docstrings oder Kommentaren nützlich, um den Code für andere Entwickler verständlicher zu machen.
10. **Fehlende Möglichkeit zur Beendigung des Kontextes**:
- In `ContextManager` gibt es keine Möglichkeit, den letzten genutzten Kontext zu speichern oder ihn über die `clear_context` zu behalten nach einem Zusammenfassungsprozess.
11. **Seltene UI-Updates**:
- Bei der UI-Aktualisierung könnte es gelegentlich zu Verzögerungen kommen. Ein effizientes Design bei der Verwendung von Signalen und Slots wäre wichtig.
12. **Python 2 vs. 3 Kompatibilität**:
- Es könnte unklar sein, ob der Code auch in Python 2 ausgeführt werden kann. Zum Beispiel ist die Verwendung eines `print`-Statements als Funktion in Python 3.
13. **Performance**:
- Bei großen Datenmengen in der Datenbank könnte das Laden aller Nachrichten bei `get_context` ineffizient werden. Eine paginierte Annäherung könnte die Leistung erhöhen.
Diese Punkte sollten bei der Überarbeitung und Wartung des Skripts berücksichtigt werden, um Safeguards zu erhöhen, die Performance zu verbessern und eine bessere Benutzererfahrung zu schaffen.