USER
Diese anwendung soll vereinfacht gesagt einen chat mit einem lokalen llm das über ollama ausgeführt wird ermöglichen. Dieser chat soll aber nicht wie gewöhnlich verwendet werden und nur nach dem schema, "1 nachricht vom nutzer gefolgt von einer antwort des LLM" funktionieren. Das LLM soll wenn es notwendig ist, auch mehr als eine nachricht, bzw einen bearbeitungsschritt durchführen um zu antworten. komplexere anfragen vom nutzer sollen durch einen bearbeitungsprozess beantwortet werden der automatisiert abläuft bis ein abschließendes ergebnis als antwort erstellt und presentiert wurde. Ich möchte das du das skript überprüfst und nach fehlern oder problemen suchst die verhindern das die anwendung funktioniert und genutzt werden kann. Zum beispiel scheitert es zurzeit sogar noch daran das keine antwort im chat fenster zu sehen ist. Zudem viel mir auf das die farbe des textes vom nutzer im chatfenster in dunkel blau kaum lesbar ist. Erstelle mir eine liste mit allen problemen die du findest, die behoben werden sollten damit die anwendung genutzt werden kann.
import sys
import os
import re
import subprocess
import json
import threading
import sqlite3
import time
import logging
import requests
from PyQt5.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLineEdit,
QPushButton, QLabel, QDialog, QFormLayout, QSlider, QMessageBox, QComboBox,
QTextBrowser, QAction, QScrollArea, QListWidget, QListWidgetItem, QProgressBar, QInputDialog
)
from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QObject, QRegularExpression
from PyQt5.QtGui import QTextCursor, QColor, QFont, QTextCharFormat, QSyntaxHighlighter
# Logging configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def find_ollama_executable():
if os.name == 'nt': # Windows
# Exakter korrekter Pfad aus where-Befehl
direct_path = "C:\\Users\\Admin\\AppData\\Local\\Programs\\Ollama\\ollama.exe"
if os.path.exists(direct_path):
return direct_path
# Alternative Schreibweise
alt_path = os.path.join("C:", "Users", "Admin", "AppData", "Local", "Programs", "Ollama", "ollama.exe")
if os.path.exists(alt_path):
return alt_path
return None
# Database Manager
class DatabaseManager:
def __init__(self, db_name="autollm_context.db", max_context_length=10000):
try:
self.conn = sqlite3.connect(db_name, check_same_thread=False)
self.max_context_length = max_context_length
self.create_table()
except sqlite3.Error as e:
logging.error(f"Database connection error: {e}")
def create_table(self):
with self.conn:
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
)
''')
def add_message(self, sender, message):
with self.conn:
self.conn.execute("INSERT INTO context (sender, message) VALUES (?, ?)", (sender, message))
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):
with self.conn:
self.conn.execute("DELETE FROM context")
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):
if self.conn:
self.conn.close()
# Context Manager with Improvements
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.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 = (
"Please summarize the following context to make space for new information. "
"Retain all important details and key points:\n\n"
f"{context_text}"
)
self.summarization_result = ""
self.summarize_request_event.clear()
self.ollama.send_request("/api/generate", {
"model": self.ollama.selected_llm,
"prompt": summary_prompt,
"options": {
"temperature": 0.5,
"max_new_tokens": 1000,
"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"Summary of previous context:\n{self.summary}")
else:
logging.error("Timeout during context summarization.")
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("No response received for summarization.")
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 with API Compliance
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(f"API check failed: {response.status_code}")
except requests.exceptions.RequestException:
self.server_started = False
self.server_status_changed.emit(False)
self.error_occurred.emit("Failed to connect to the server.")
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:
ollama_path = find_ollama_executable()
if not ollama_path:
# Debug-Ausgabe zum Pfad-Check
test_path = "C:\\Users\\Admin\\AppData\\Local\\Ollama\\ollama.exe"
exists = os.path.exists(test_path)
self.error_occurred.emit(f"Debug: Checking path {test_path}, exists: {exists}")
return
startupinfo = None
if os.name == 'nt':
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
self.process = subprocess.Popen(
[ollama_path, '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"Detailed error: {str(e)}")
self.error_occurred.emit(f"Attempted path: {ollama_path}")
else:
self.server_started = True
self.server_status_changed.emit(True)
self.update_llm_list()
def delayed_check(self):
time.sleep(5)
self.check_server_status()
def stop_server(self):
if self.is_server_running() and self.process:
try:
self.interrupt_event.set()
self.process.terminate()
self.process.wait(timeout=10)
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"Error stopping the server: {e}")
def get_installed_llms(self):
try:
response = requests.get(f"http://localhost:{self.api_port}/api/tags", timeout=2)
response.raise_for_status()
data = response.json()
return [model["name"] for model in data.get("models", [])]
except requests.exceptions.RequestException as e:
self.error_occurred.emit(f"Error fetching models: {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, endpoint, data):
if not self.is_server_running():
self.error_occurred.emit("Ollama server is not running.")
return
threading.Thread(target=self._handle_request, args=(endpoint, data), daemon=True).start()
def _handle_request(self, endpoint, data):
self.interrupt_event.clear()
try:
url = f"http://localhost:{self.api_port}{endpoint}"
response = requests.post(url, json=data, stream=True)
response.raise_for_status()
for line in response.iter_lines():
if self.interrupt_event.is_set():
raise InterruptedError("Request interrupted by user.")
if line:
decoded_line = line.decode('utf-8')
try:
data = json.loads(decoded_line)
if "error" in data:
raise Exception(f"Ollama API error: {data['error']}")
self.response_received.emit(data)
except json.JSONDecodeError as e:
self.error_occurred.emit(f"Invalid JSON response: {e}")
return
except InterruptedError:
self.error_occurred.emit("Request interrupted.")
except requests.exceptions.RequestException as e:
self.error_occurred.emit(f"API request failed: {e}")
except Exception as e:
self.error_occurred.emit(f"Error handling request: {e}")
# Task Manager with Improvements
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.error_occurred.connect(self.on_error)
def add_task(self, user_request):
breakdown_prompt = (
"Create a detailed plan to break down the following request into meaningful subtasks. "
"List each subtask as a clear, independent task that can be accomplished in one step. "
"Return the steps in a numbered list.\n\n"
f"Request: {user_request}"
)
self.current_user_request = user_request
self.ollama.send_request("/api/generate", {
"model": self.ollama.selected_llm,
"prompt": breakdown_prompt,
"options": {
"temperature": 0.5,
"max_new_tokens": 1500,
"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("Error parsing the request into subtasks.")
else:
self.task_failed.emit("No response received from the LLM.")
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"Please complete the following task in one step:\n\n{self.current_task}\n\n"
"Return the result in a clear and concise answer."
)
self.ollama.send_request("/api/generate", {
"model": self.ollama.selected_llm,
"prompt": task_prompt,
"options": {
"temperature": 0.5,
"max_new_tokens": 1500,
"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"Please summarize all the results of the following tasks to create a final response to the original request:\n\n"
f"Original Request: {self.current_user_request}\n\n"
f"Task Results:\n{results}\n\n"
"Now create a detailed and cohesive response based on these results."
)
self.ollama.send_request("/api/generate", {
"model": self.ollama.selected_llm,
"prompt": final_prompt,
"options": {
"temperature": 0.5,
"max_new_tokens": 2000,
"stream": False,
"format": "json"
}
})
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"Error on task '{self.current_task}': {error_message}")
self.current_task = None
self.process_next_task()
else:
self.task_failed.emit(f"Error: {error_message}")
# AutoLLM App with API and UI Improvements
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)
# Connect signals from the Ollama controller
self.ollama.server_status_changed.connect(self.update_server_status)
self.ollama.error_occurred.connect(self.show_error)
self.ollama.llm_list_updated.connect(self.update_llm_list)
# Initialize code highlighting after chat_area is created
self.highlighter = CodeHighlighter(self.chat_area.document())
# For storing LLM responses
self.llm_response_buffer = ""
self.current_display_index = 0
# For blinking "..."
self.blink_timer = QTimer()
self.blink_timer.timeout.connect(self.toggle_ellipsis)
self.ellipsis_visible = True
def init_ui(self):
main_layout = QHBoxLayout()
# Left side: chat and progress
left_layout = QVBoxLayout()
# Chat area with scrollbar
self.chat_area = QTextBrowser()
self.chat_area.setReadOnly(True)
self.chat_area.setFont(QFont("Courier", 10))
self.chat_area.setOpenExternalLinks(True)
chat_scroll_area = QScrollArea()
chat_scroll_area.setWidgetResizable(True)
chat_scroll_area.setWidget(self.chat_area)
left_layout.addWidget(chat_scroll_area)
# Input field and send button
input_layout = QHBoxLayout()
self.input_field = QLineEdit()
self.input_field.setPlaceholderText("Enter your message here...")
self.input_field.returnPressed.connect(self.send_message)
self.send_button = QPushButton("Send")
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)
# Interrupt button
self.interrupt_button = QPushButton("Interrupt")
self.interrupt_button.clicked.connect(self.interrupt_processing)
left_layout.addWidget(self.interrupt_button)
# Server monitoring area
server_layout = QHBoxLayout()
self.server_status = QLabel("🔴 Server Stopped")
self.server_button = QPushButton("Start Server")
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 selection
llm_layout = QHBoxLayout()
self.llm_label = QLabel("Select LLM:")
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)
# Settings button
self.settings_button = QPushButton("Settings")
self.settings_button.clicked.connect(self.open_settings)
left_layout.addWidget(self.settings_button)
# Progress area
self.task_progress = QProgressBar()
self.task_progress.setValue(0)
left_layout.addWidget(self.task_progress)
# Plan and steps area
plan_layout = QVBoxLayout()
self.plan_label = QLabel("Current 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)
# Interaction buttons
interaction_layout = QHBoxLayout()
self.edit_button = QPushButton("Edit Task")
self.edit_button.clicked.connect(self.edit_task)
self.skip_button = QPushButton("Skip Task")
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)
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 toggle_ellipsis(self):
if self.ellipsis_visible:
self.chat_area.moveCursor(QTextCursor.End)
cursor = self.chat_area.textCursor()
cursor.movePosition(QTextCursor.Left, QTextCursor.KeepAnchor, 3)
cursor.removeSelectedText()
self.ellipsis_visible = False
else:
self.chat_area.moveCursor(QTextCursor.End)
self.chat_area.insertPlainText("...")
self.ellipsis_visible = True
def start_typing_animation(self):
self.append_message("LLM", "", message_type="typing")
self.blink_timer.start(500) # Blinken alle 500ms
def stop_typing_animation(self):
self.blink_timer.stop()
# Entferne die letzte Typing-Nachricht
cursor = self.chat_area.textCursor()
cursor.movePosition(QTextCursor.End)
cursor.movePosition(QTextCursor.Up, QTextCursor.KeepAnchor)
cursor.removeSelectedText()
cursor.deletePreviousChar() # Entferne zusätzlichen Zeilenumbruch
def closeEvent(self, event):
# Wird beim Schließen der Anwendung aufgerufen
self.ollama.stop_server()
self.context_manager.db.close_connection()
event.accept()
def keyPressEvent(self, event):
# Behandelt Tastatureingaben
if event.key() == Qt.Key_Escape:
self.interrupt_processing()
def update_server_status(self, is_running):
if is_running:
self.server_status.setText("🟢 Server Running")
self.server_button.setText("Stop Server")
self.enable_ui(True)
else:
self.enable_ui(False)
self.llm_combo.clear()
self.llm_combo.addItem("No LLMs Installed")
self.ollama.set_selected_llm(None)
self.server_status.setText("🔴 Server Stopped")
self.server_button.setText("Start Server")
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("No LLMs Installed")
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 != "No LLMs Installed":
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("User", user_message)
self.append_message("User", user_message)
self.input_field.clear()
self.task_manager.add_task(user_message)
def append_message(self, sender, message, message_type="normal"):
color = "gray"
if message_type == "typing":
if sender == "LLM":
self.chat_area.append('<p style="color: green;"><b>LLM:</b> <i>...</i></p>')
else:
if sender == "User":
color = "blue"
prefix = "<b>User:</b> "
elif sender == "LLM":
color = "green"
prefix = "<b>LLM:</b> "
else:
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):
message = re.sub(r'(https?://\S+)', r'<a href="\1">\1</a>', message)
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 interrupt_processing(self):
self.ollama.interrupt_event.set()
self.append_message("System", "LLM processing has been interrupted.")
def toggle_server(self):
if self.ollama.is_server_running():
reply = QMessageBox.question(self, 'Stop Server',
'Are you sure you want to stop the Ollama server?',
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, "Settings Updated", "LLM parameters have been updated.")
def on_task_completed_step(self, result):
self.append_message("LLM", f"Result: {result}")
self.update_plan_list()
def on_task_failed_step(self, error):
self.append_message("System", f"Error: {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} (running)")
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} (error)")
item.setForeground(QColor("red"))
self.plan_list.addItem(item)
def display_final_answer(self, answer):
self.append_message("LLM", f"Final Answer: {answer}")
def edit_task(self):
if self.task_manager.current_task:
text, ok = QInputDialog.getText(self, "Edit Task", "Edit the current task:", QLineEdit.Normal, self.task_manager.current_task)
if ok and text.strip():
self.task_manager.current_task = text.strip()
self.append_message("System", f"Task edited: {self.task_manager.current_task}")
def skip_task(self):
if self.task_manager.current_task:
self.append_message("System", f"Task skipped: {self.task_manager.current_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("Settings")
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("Temperature:", 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("Output Length:", output_layout)
# Additional options
self.additional_options_box = QLabel("Additional Options (JSON):")
self.additional_options_input = QLineEdit()
self.additional_options_input.setPlaceholderText('e.g., "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("Save")
self.save_button.clicked.connect(self.save_settings)
self.cancel_button = QPushButton("Cancel")
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, "Invalid JSON", "Please enter valid JSON for additional options.")
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_())