import sys
import os
import inspect
import time
import random
import requests # Ollamaモデルリスト取得用
import json # Ollamaレスポンス解析用
import unicodedata # 日本語判定用
from PySide6.QtWidgets import (QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QComboBox, QSpinBox, QCheckBox, QScrollArea,
QTextEdit, QGroupBox, QListWidget, QListWidgetItem,
QFileDialog, QMessageBox)
from typing import Any, Optional # Any, Optional をインポート
from PySide6.QtCore import Qt, Signal, QThread, QObject
from PySide6.QtGui import QFont
# paintgui.py から必要なクラスをインポート
try:
from GUI.PaintGUI import ChatPaintWidget
except ImportError as e:
# Fallback if GUI.PaintGUI is not directly in PYTHONPATH, try relative
# This assumes ComparisonGui.py is in the same root as GUI folder
# Adjust the number of os.path.dirname based on your project structure
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from GUI.PaintGUI import ChatPaintWidget
# PaintGUIからコールバックハンドラとストリーミング用関数をインポート
from GUI.PaintGUI import ChatPaintWidget, MyStreamlitCallbackHandler as PaintGUICallbackHandler
# AIAgent と関連ヘルパー
from Agents.AIAgent import AIAgent
# tools.program_called_command_list から WORK_SPACE_DIR, set_work_space, load_ai_agent_name_list をインポート
from tools.program_called_command_list import WORK_SPACE_DIR, set_work_space, load_ai_agent_name_list
from langchain_core.tools import Tool, BaseTool # Toolクラスの型チェック用
from tools import command_list as cl_module # tools.command_list モジュールとしてインポート
from tools.exception import InterruptedException as InterruptedException
# 日本語かどうかを簡易的に判定する関数
def is_predominantly_japanese(text, threshold_ratio=0.3):
"""
テキストが主に日本語であるかどうかを判定します。
ひらがな、カタカナ、漢字の文字が指定された割合以上含まれていればTrueを返します。
"""
if not text:
return False
japanese_char_count = 0
total_significant_chars = 0 # 空白以外の文字数
for char in text:
if char.isspace():
continue
total_significant_chars += 1
# unicodedata.script は Python 3.8+ で利用可能
try:
script = unicodedata.script(char)
if script in ('Hiragana', 'Katakana', 'Han'):
japanese_char_count += 1
except AttributeError: # 古いPythonのためのフォールバック (より単純な判定)
if '\u3040' <= char <= '\u309F' or \
'\u30A0' <= char <= '\u30FF' or \
'\u4E00' <= char <= '\u9FFF':
japanese_char_count += 1
if total_significant_chars == 0:
return False
return (japanese_char_count / total_significant_chars) >= threshold_ratio
# パフォーマンス情報取得のための関数
def get_performance_metrics(start_time, end_time, response_text, llm_output_metadata=None):
processing_time = end_time - start_time
tokens_generated = 0
if llm_output_metadata and isinstance(llm_output_metadata, dict):
token_usage = llm_output_metadata.get('token_usage', {}) # .getで安全にアクセス
if isinstance(token_usage, dict):
if 'completion_tokens' in token_usage: # OpenAI, Gemini
tokens_generated = token_usage.get('completion_tokens', 0)
elif 'output_tokens' in token_usage: # Ollama
tokens_generated = token_usage.get('output_tokens', 0)
if tokens_generated == 0 and response_text: # LLMメタデータから取得できなかった場合のフォールバック
if is_predominantly_japanese(response_text):
tokens_generated = len(response_text) # 日本語の場合は文字数をトークン数とみなす
else:
tokens_generated = len(response_text.split()) # それ以外はスペースで分割した単語数をトークン数とみなす
tokens_per_second = tokens_generated / processing_time if processing_time > 0 else 0
#vram_used_gb = random.uniform(1.0, 8.0) # VRAMはダミー
return {
"time_s": f"{processing_time:.2f}",
"tokens": str(tokens_generated),
"tps": f"{tokens_per_second:.2f}",
# "vram_gb": f"{vram_used_gb:.2f}"
}
# --- AIAgentが期待するGUIインターフェースのためのグローバル参照 ---
_g_active_comparison_gui_instance: Optional['ComparisonGui'] = None
def _get_active_chat_paint_widget() -> Optional[ChatPaintWidget]:
"""現在アクティブなComparisonGuiインスタンスから、最初のモデルビューのChatPaintWidgetを返す（簡易的な実装）"""
global _g_active_comparison_gui_instance
if _g_active_comparison_gui_instance and _g_active_comparison_gui_instance.model_views:
# ここでは単純に最初のモデルビューを対象とする。
# より高度な実装では、どのビューを操作対象とするか明確にする必要がある。
# return _g_active_comparison_gui_instance.model_views[_g_active_comparison_gui_instance.processing_model_index].output_widget
current_processing_idx = _g_active_comparison_gui_instance.processing_model_index
if 0 <= current_processing_idx < len(_g_active_comparison_gui_instance.model_views):
return _g_active_comparison_gui_instance.model_views[current_processing_idx].output_widget
# Fallback or if no specific model is processing, maybe return the first one or handle error
return None
def get_stc_handler():
"""AIAgent用のコールバックハンドラを返す"""
active_widget = _get_active_chat_paint_widget()
if active_widget:
# PaintGUIのMyStreamlitCallbackHandlerをラップするか、
# ChatPaintWidgetを直接渡せるようにPaintGUICallbackHandlerを調整する必要がある。
# ここでは、PaintGUICallbackHandlerがChatPaintWidgetインスタンスをcontainerとして受け取れると仮定。
#my_widget = _get_my_widget_instance()
#return MyStreamlitCallbackHandler(my_widget)
print("active_widget",type(active_widget))
return PaintGUICallbackHandler(active_widget)
print("Warning: comparison_gui.get_stc_handler() called, but no active ChatPaintWidget found.")
return None # またはダミーのハンドラ
def append_node(name: str, text: str):
active_widget = _get_active_chat_paint_widget()
if active_widget:
active_widget.append_node(name, text)
def set_last_node_text(text: str):
active_widget = _get_active_chat_paint_widget()
if active_widget:
active_widget.set_last_node_text(text)
def get_available_llm_models():
# 書式: "provider:model_identifier"
static_models = [
"gemini:gemini-1.5-flash",
# "gemini:gemini-1.5-pro-latest", # Vision対応の可能性
# "hf:microsoft/Phi-3-mini-4k-instruct" # HuggingFaceモデルの例 (別途HuggingFaceLLMの実装が必要)
]
ollama_models = []
try:
response = requests.get("http://localhost:11434/api/tags", timeout=3) # タイムアウトを3秒に設定
response.raise_for_status() # HTTPエラーがあれば例外を発生
models_data = response.json()
if "models" in models_data and isinstance(models_data["models"], list):
for model_info in models_data["models"]:
if "name" in model_info:
ollama_models.append(f"ollama:{model_info['name']}")
if not ollama_models:
print("Warning: No models found in Ollama API response, though connection was successful.")
except requests.exceptions.Timeout:
print("Warning: Timeout when trying to connect to Ollama to get model list.")
except requests.exceptions.ConnectionError:
print("Warning: Could not connect to Ollama server (e.g., server not running).")
except requests.exceptions.RequestException as e:
print(f"Warning: Error connecting to Ollama to get model list: {e}")
except json.JSONDecodeError as e:
print(f"Warning: Could not parse Ollama model list response: {e}")
combined_models = ollama_models + static_models
# Ollamaモデルが一つも取得できなかった場合、静的なOllamaモデルをフォールバックとして追加
if not any(m.startswith("ollama:") for m in combined_models):
print("Warning: No Ollama models fetched dynamically. Adding static Ollama models as fallback.")
static_ollama_fallbacks = [
"ollama:llama3",
"ollama:qwen2.5vl:7b", # OllamaのVisionモデル例
"ollama:gemma3:4b-it-qat",
"ollama:llava", # OllamaのVisionモデル例
]
combined_models.extend(static_ollama_fallbacks)
return sorted(list(set(combined_models))) # 重複除去とソート
def get_tools_from_command_list_module(module):
tools_info = []
# tools.command_list.get_tool_list() は Tool オブジェクトのリストを返す
if not hasattr(module, 'get_tool_list') or not callable(module.get_tool_list):
print(f"Error: Module {module.__name__} does not have a callable 'get_tool_list' function.")
return []
raw_tools = module.get_tool_list()
# print(f"Debug: raw_tools from {module.__name__}.get_tool_list(): {raw_tools}")
for func_tool in raw_tools:
if isinstance(func_tool, BaseTool): # Tool から BaseTool に変更
tools_info.append({
'name': func_tool.name,
'description': func_tool.description,
'function': getattr(func_tool, 'func', None), # StructuredTool には func が直接ない場合があるため getattr を使用
'langchain_tool': func_tool
})
# print(f"Debug: Added tool (BaseTool instance): {func_tool.name}")
# 必要であれば、@toolデコレータが直接関数に属性を付与するケースも考慮
elif inspect.isfunction(func_tool) and hasattr(func_tool, 'name') and hasattr(func_tool, 'description'):
tools_info.append({
'name': func_tool.name,
'description': func_tool.description,
'function': func_tool,
'langchain_tool': Tool(name=func_tool.name, description=func_tool.description, func=func_tool, args_schema=getattr(func_tool, 'args_schema', None))
})
# print(f"Debug: Added tool (decorated function): {func_tool.name}")
else:
print(f"Debug: Skipped item in raw_tools (not a Tool instance or recognized decorated function): {func_tool}")
unique_tools = []
seen_names = set()
for t in tools_info:
if t['name'] not in seen_names:
unique_tools.append(t)
seen_names.add(t['name'])
# print(f"Debug: unique_tools to be returned: {unique_tools}")
return unique_tools
class SingleModelView(QWidget):
def __init__(self, model_id, available_llms, available_tools, parent=None):
super().__init__(parent)
self.model_id = model_id
self.available_llms = available_llms
self.available_tools = available_tools
self.selected_tools_objects = []
self.layout = QVBoxLayout(self)
self.layout.setContentsMargins(5,5,5,5)
# --- Font settings ---
self.default_font_name = "Arial"
self.settings_group = QGroupBox(f"Model {self.model_id + 1} Settings")
self.settings_layout = QVBoxLayout()
self.llm_combo = QComboBox()
self.llm_combo.addItems(self.available_llms)
self.settings_layout.addWidget(QLabel("LLM Model:"))
self.settings_layout.addWidget(self.llm_combo)
# ウォームアップのON/OFFチェックボックスを追加
# self.warmup_checkbox = QCheckBox("Enable Warmup")
# self.warmup_checkbox.setChecked(True) # デフォルトでON
# self.settings_layout.addWidget(self.warmup_checkbox)
self.tools_checkbox = QCheckBox("Use Tools")
self.tools_checkbox.toggled.connect(self.toggle_tools_list)
self.settings_layout.addWidget(self.tools_checkbox)
self.tools_list_widget = QListWidget()
self.tools_list_widget.setSelectionMode(QListWidget.MultiSelection)
for i, tool_info in enumerate(self.available_tools):
item_text = f"{tool_info['name']}"
if tool_info['description']:
item_text += f" - {tool_info['description'][:40]}..." # 説明が長すぎる場合省略
item = QListWidgetItem(item_text)
item.setData(Qt.UserRole, tool_info) # tool_info全体を格納
item.setFlags(item.flags() | Qt.ItemIsUserCheckable) # チェックボックスを有効化
item.setCheckState(Qt.Unchecked) # 初期状態は未チェック
self.tools_list_widget.addItem(item)
# 最後以外のアイテムの後ろに区切り線を追加
if i < len(self.available_tools) - 1:
separator_item = QListWidgetItem("─" * 30)
separator_item.setFlags(Qt.NoItemFlags) # 選択不可、チェック不可
# 区切り線のフォントは update_font_settings で設定される
separator_item.setForeground(Qt.gray) # 区切り線の色をグレーに
self.tools_list_widget.addItem(separator_item)
self.tools_list_widget.setVisible(False)
# itemSelectionChangedはハイライトの変更を検知するが、チェックボックスの変更は検知しない。
# チェックボックスの状態変更を検知するには itemChanged を使う必要がある。
self.tools_list_widget.itemChanged.connect(self._update_selected_tools)
self.settings_layout.addWidget(self.tools_list_widget)
self.settings_group.setLayout(self.settings_layout)
self.layout.addWidget(self.settings_group)
self.output_widget = ChatPaintWidget() # PaintGUIから再利用
self.output_widget.setMinimumHeight(300) # 適当な高さ
self.layout.addWidget(self.output_widget, 1) # ストレッチファクターを1に設定
self.perf_label = QLabel("Perf: TBD")
self.perf_label.setAlignment(Qt.AlignRight)
self.layout.addWidget(self.perf_label)
# 初期メッセージを表示
self.output_widget.append_node("ai", "作業指示をしてください。")
self.update_font_settings(14, 12) # 初期フォントサイズ設定 (全てのウィジェットが作成された後)
def update_font_settings(self, default_size, label_size, separator_size=10):
default_font = QFont(self.default_font_name, default_size)
label_font = QFont(self.default_font_name, label_size)
separator_font = QFont(self.default_font_name, separator_size)
self.settings_group.setFont(default_font)
# QLabel for LLM Model
if self.settings_layout.count() > 1 and isinstance(self.settings_layout.itemAt(0).widget(), QLabel): # 最初のウィジェットがラベルか確認
self.settings_layout.itemAt(0).widget().setFont(label_font)
self.llm_combo.setFont(default_font)
# self.warmup_checkbox.setFont(default_font) # ウォームアップチェックボックスにフォント適用
self.tools_checkbox.setFont(default_font)
self.tools_list_widget.setFont(default_font)
for i in range(self.tools_list_widget.count()):
item = self.tools_list_widget.item(i)
if not (item.flags() & Qt.ItemIsUserCheckable): # 区切り線アイテム
item.setFont(separator_font)
self.perf_label.setFont(label_font)
# ChatPaintWidgetのフォントも更新
if hasattr(self.output_widget, 'update_font_size'): # Changed from update_node_font_size
self.output_widget.update_font_size(default_size)
def toggle_tools_list(self, checked):
self.tools_list_widget.setVisible(checked)
def _update_selected_tools(self):
self.selected_tools_objects = []
for i in range(self.tools_list_widget.count()):
item = self.tools_list_widget.item(i)
if item.flags() & Qt.ItemIsUserCheckable: # チェック可能なアイテムのみ処理
if item.checkState() == Qt.Checked:
tool_info = item.data(Qt.UserRole) # 格納したtool_infoを取得
if tool_info and 'langchain_tool' in tool_info:
self.selected_tools_objects.append(tool_info['langchain_tool'])
def get_settings(self):
return {
"llm_model": self.llm_combo.currentText(),
# "enable_warmup": self.warmup_checkbox.isChecked(), # 新しい設定を保存
"use_image": True, # 画像入力は常に有効とみなす (GUIからチェックボックスを削除するため)
"use_tools": self.tools_checkbox.isChecked(),
"selected_tools": self.selected_tools_objects if self.tools_checkbox.isChecked() else []
}
def display_output(self, title, text, perf_data=None):
self.output_widget.append_node(title, text)
if perf_data:
perf_str = f"Time: {perf_data['time_s']}s, Tokens: {perf_data['tokens']}, TPS: {perf_data['tps']}, VRAM: {perf_data['vram_gb']}GB"
self.perf_label.setText(perf_str)
else:
self.perf_label.setText("Perf: N/A")
def clear_output(self):
# ChatPaintWidgetの表示をクリア
if hasattr(self.output_widget, 'chat_thread') and hasattr(self.output_widget.chat_thread, 'nodes'):
self.output_widget.chat_thread.nodes.clear()
self.output_widget.chat_thread.draw_start_index = 0
# スライダーの状態もリセット (必要に応じて)
if hasattr(self.output_widget, '_ChatPaintWidget__update_slider'): # プライベートメンバアクセス注意
try:
self.output_widget._ChatPaintWidget__update_slider()
except Exception as e:
print(f"Error updating slider during clear: {e}")
self.output_widget.update()
self.perf_label.setText("Perf: TBD")
def start_streaming_display(self, title: str):
"""Prepares the output widget for a new streaming response."""
if hasattr(self.output_widget, 'new_streaming_node'):
self.output_widget.new_streaming_node(title)
else:
print(f"Warning: output_widget for model {self.model_id} is missing 'new_streaming_node' method.")
# Fallback: append a new node with a placeholder message
self.output_widget.append_node(title, "Streaming started...")
def append_streaming_text(self, token: str):
"""Appends a token to the current streaming response in the output widget."""
if hasattr(self.output_widget, 'append_text'):
self.output_widget.append_text(token)
else:
print(f"Warning: output_widget for model {self.model_id} is missing 'append_text' method.")
def finalize_streaming_display(self, final_text: str, perf_data: Optional[dict]):
"""Finalizes the display after streaming, primarily updating performance metrics."""
# The text in ChatPaintWidget should have been built by append_streaming_text.
# Performance data is the main thing to update here.
if perf_data:
# perf_str = f"Time: {perf_data['time_s']}s, Tokens: {perf_data['tokens']}, TPS: {perf_data['tps']}, VRAM: {perf_data['vram_gb']}GB"
perf_str = f"Time: {perf_data['time_s']}s, Tokens: {perf_data['tokens']}, TPS: {perf_data['tps']}"
self.perf_label.setText(perf_str)
else:
self.perf_label.setText("Perf: N/A")
class ComparisonWorker(QObject):
# シグナル定義: model_index, title, text, perf_data
result_ready = Signal(int, str, str, dict) # model_index, type ("warmup" or "main"), text, perf_data
error_occurred = Signal(int, str)
finished = Signal() # このワーカーの全処理（ウォームアップ＋メイン）完了
new_token_ready = Signal(int, str) # model_index, new_token
streaming_started = Signal(int, str, str) # model_index, type ("warmup" or "main"), title
def __init__(self, model_index, agent_settings, common_prompt, image_path=None):
super().__init__()
self.model_index = model_index
self.agent_settings = agent_settings
self.common_prompt = common_prompt
self.image_path = image_path
# self._is_cancelled = False # 中断フラグ
def cancel(self): # 中断メソッド
print(f"Worker {self.model_index}: Cancellation requested.")
InterruptedException.set_cancel(True)
#raise InterruptedException("LLM stream cancelled by worker flag.")
class WorkerCallbackHandler(PaintGUICallbackHandler): # MyStreamlitCallbackHandler を PaintGUICallbackHandler に変更
def __init__(self, worker_ref: 'ComparisonWorker'): # Type hint for worker_ref
super().__init__(container=None) # containerは直接使わない
self.worker_ref = worker_ref # ComparisonWorkerへの参照
self.set_cancel(False) # 中断フラグ
def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
if self.worker_ref:
self.worker_ref.new_token_ready.emit(self.worker_ref.model_index, token)
def on_llm_end(self, response: Any, **kwargs: Any) -> None:
# The base class (PaintGUICallbackHandler/MyStreamlitCallbackHandler)
# would try to use self.container, which is None here.
# The full response is handled by the result_ready signal from the worker.
# So, we don't need to do anything specific here for flushing a buffer to a container.
if self.worker_ref._is_cancelled:
print(f"Worker {self.worker_ref.model_index}: LLM end callback detected cancellation.")
return
# 通常の終了処理は result_ready で行うので、ここでは何もしない
# pass
def _create_agent(self):
# agent_name はユニークにするか、private_memory=Trueなら同じでも問題ない場合がある
agent_name = f"Comparer_{self.model_index}_{self.agent_settings['llm_model'].replace(':', '_')}"
tools_to_use = self.agent_settings.get('selected_tools', [])
if not self.agent_settings.get('use_tools', False):
tools_to_use = []
llm_model_str = self.agent_settings['llm_model']
llm_kwargs = {} # LLMプロバイダへの追加引数用
if llm_model_str.startswith("ollama:"):
# 例: OllamaLLMにbase_urlを渡す場合
llm_kwargs['base_url'] = "http://localhost:11434" # 必要に応じて設定ファイルから
print("agentname",agent_name)
agent = AIAgent(
agent_name=agent_name,
system_prompt="You are a helpful AI assistant.", # ここでは固定
tools=tools_to_use,
private_memory=True, # 各比較は独立したメモリを持つ
llm_model=llm_model_str,
**llm_kwargs
)
return agent
def _get_llm_metadata(self, agent_instance):
# AIAgentインスタンスからLLMの出力メタデータを取得する試み
if agent_instance and agent_instance.llm_provider:
if hasattr(agent_instance.llm_provider, 'llm') and agent_instance.llm_provider.llm:
if hasattr(agent_instance.llm_provider.llm, 'response_metadata'): # AIMessage.response_metadata
return agent_instance.llm_provider.llm.response_metadata
if hasattr(agent_instance.llm_provider.llm, 'llm_output'): # 一部のLLMラッパー
return agent_instance.llm_provider.llm.llm_output
return None
def run(self):
#self.enable_warmup = self.agent_settings.get('enable_warmup', True)
try:
print("InterruptedException.is_cancelled()",InterruptedException.is_cancelled())
if InterruptedException.is_cancelled():
print(f"Worker {self.model_index}: Run started but already cancelled.")
self.finished.emit()
return
# --- 1. ウォームアップ実行 ---
#if self.agent_settings.get('enable_warmup', True): # デフォルトはTrue
#if self.enable_warmup: # ComparisonGui の設定を使用
if self.agent_settings.get('enable_warmup', True): # ワーカー設定から取得
self.streaming_started.emit(self.model_index, "warmup", "AI (Warm-up)")
greeting_prompt = "こんにちは"
start_time_greeting = time.time()
agent_for_greeting = self._create_agent()
response_greeting = agent_for_greeting.get_respons(
greeting_prompt,
image_paths=None # ウォームアップでは画像なし
)
end_time_greeting = time.time()
metadata_greeting = self._get_llm_metadata(agent_for_greeting)
perf_data_greeting = get_performance_metrics(start_time_greeting, end_time_greeting, response_greeting, metadata_greeting)
self.result_ready.emit(self.model_index, "warmup", response_greeting, perf_data_greeting)
# --- 2. メインプロンプト実行 ---
self.streaming_started.emit(self.model_index, "main", "AI (Main)")
start_time_main = time.time()
agent_for_main = self._create_agent() # メインプロンプト用エージェント (履歴を分けるため再作成)
#img_paths_for_agent = []
#if self.image_path: # 画像パスが設定されていれば常に渡す (use_image設定は無視)
# img_paths_for_agent = [self.image_path]
# self.image_path は既にリストなので、そのまま渡す
response_main = agent_for_main.get_respons(
self.common_prompt,
#image_paths=img_paths_for_agent
image_paths=self.image_path # 複数画像パスのリストを直接渡す
)
# end_time_main = time.time()
# metadata_main = self._get_llm_metadata(agent_for_main)
# perf_data_main = get_performance_metrics(start_time_main, end_time_main, response_main, metadata_main)
# self.result_ready.emit(self.model_index, "main", response_main, perf_data_main)
except InterruptedException as ie: # カスタム中断例外をキャッチ
print(f"Worker {self.model_index}: InterruptedException caught - {ie}")
self.error_occurred.emit(self.model_index, "Cancelled by user.")
except Exception as e:
import traceback
# エラーが発生したのがウォームアップ中かメイン処理中かでメッセージを分けることも可能
# ここでは共通のエラーメッセージとする
error_msg = (f"Error in worker {self.model_index} "
f"({self.agent_settings['llm_model']}): {e}\n"
f"{traceback.format_exc()}")
print(error_msg)
self.error_occurred.emit(self.model_index, f"Processing error: {e}")
finally:
self.finished.emit()
end_time_main = time.time()
metadata_main = self._get_llm_metadata(agent_for_main)
if InterruptedException.is_cancelled():
response_main = InterruptedException.get_response()
print("DEBUG run finally", start_time_main ,end_time_main,response_main,metadata_main)
perf_data_main = get_performance_metrics(start_time_main, end_time_main, response_main, metadata_main)
self.result_ready.emit(self.model_index, "main", response_main, perf_data_main)
class ComparisonGui(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
global _g_active_comparison_gui_instance
_g_active_comparison_gui_instance = self # このインスタンスをグローバルに設定
self.setWindowTitle("LLM Comparison Tool")
self.setGeometry(50, 50, 1800, 950) # 広めのウィンドウサイズ
# --- Font settings ---
self.current_default_font_size = 14
self.current_label_font_size = 12
self.current_button_font_size = 13
self.current_groupbox_font_size = 14
self.default_font_name = "Arial"
self.main_layout = QVBoxLayout(self)
self.main_layout.setSpacing(0) # ウィジェット間のデフォルトスペーシングをなくす (コメントアウト解除)
self.main_layout.setContentsMargins(0, 0, 0, 0) # レイアウト自身のマージンもなくす
# グローバル設定 (比較数など)
self.global_settings_layout = QHBoxLayout()
compare_label = QLabel("Number of Models to Compare:")
self.global_settings_layout.addWidget(compare_label)
self.compare_count_spinbox = QSpinBox()
self.compare_count_spinbox.setRange(1, 5) # 1から5モデルまで
self.compare_count_spinbox.setValue(1)
self.compare_count_spinbox.valueChanged.connect(self.update_model_views)
self.global_settings_layout.addWidget(self.compare_count_spinbox)
# グローバルウォームアップ設定をここに追加
self.warmup_checkbox = QCheckBox("Enable Warmup")
self.warmup_checkbox.setChecked(True) # デフォルトでON
self.global_settings_layout.addWidget(self.warmup_checkbox)
self.main_layout.addLayout(self.global_settings_layout)
# グローバルウォームアップもテキストを表示させるためのスペーサーこのラベルがないと正常に文字列が表示されない。
self.global_warmup_layout = QHBoxLayout()
warmup_label = QLabel("")
self.global_warmup_layout.addWidget(warmup_label)
# モデルごとの設定と出力表示エリア (横スクロール)
self.models_area_scroll = QScrollArea()
self.models_area_scroll.setWidgetResizable(True)
# self.models_area_scroll.setFont(default_font) # ScrollArea自体には不要かも
self.models_area_widget = QWidget()
self.models_layout = QHBoxLayout(self.models_area_widget) # ここにSingleModelViewを追加
self.models_area_scroll.setWidget(self.models_area_widget)
self.main_layout.addWidget(self.models_area_scroll, 1) # Stretch factor 1
self.model_views = [] # SingleModelViewのリスト
# SingleModelView 内のフォントは SingleModelView の __init__ で設定
# 共通入力エリア
self.common_input_group = QGroupBox("Common Input")
self.common_input_layout = QVBoxLayout()
self.prompt_input_edit = QTextEdit()
self.prompt_input_edit.setPlaceholderText("Enter your prompt here for all models...")
self.prompt_input_edit.setFixedHeight(100)
self.common_input_layout.addWidget(self.prompt_input_edit)
self.image_input_layout = QHBoxLayout()
self.image_path_label = QLabel("Image (optional): None")
self.image_input_layout.addWidget(self.image_path_label)
self.browse_image_button = QPushButton("Browse Image")
self.browse_image_button.clicked.connect(self.browse_image)
self.image_input_layout.addWidget(self.browse_image_button)
self.clear_image_button = QPushButton("Clear Image")
self.clear_image_button.clicked.connect(self.clear_image)
self.image_input_layout.addWidget(self.clear_image_button)
self.common_input_layout.addLayout(self.image_input_layout)
self.selected_image_path = []
self.control_buttons_layout = QHBoxLayout()
self.send_button = QPushButton("Send to All Models")
self.send_button.clicked.connect(self.send_to_all_models)
self.control_buttons_layout.addWidget(self.send_button)
self.stop_button = QPushButton("Stop All Models")
self.stop_button.clicked.connect(self.stop_all_models)
self.stop_button.setEnabled(False) # Initially disabled
self.control_buttons_layout.addWidget(self.stop_button)
self.clear_outputs_button = QPushButton("Clear All Outputs")
self.clear_outputs_button.clicked.connect(self.clear_all_outputs)
self.control_buttons_layout.addWidget(self.clear_outputs_button)
self.common_input_layout.addLayout(self.control_buttons_layout)
self.common_input_group.setLayout(self.common_input_layout)
self.main_layout.addWidget(self.common_input_group)
# フォントサイズ変更UI
self.font_control_layout = QHBoxLayout()
font_size_label = QLabel("Global Font Size:")
self.font_control_layout.addWidget(font_size_label)
self.font_size_spinbox = QSpinBox()
self.font_size_spinbox.setRange(8, 30) # フォントサイズの範囲
self.font_size_spinbox.setValue(self.current_default_font_size)
self.font_control_layout.addWidget(self.font_size_spinbox)
apply_font_button = QPushButton("Apply Font Size")
apply_font_button.clicked.connect(self.apply_global_font_size)
self.font_control_layout.addWidget(apply_font_button)
self.main_layout.addLayout(self.font_control_layout)
# 初期化
self.available_llms = get_available_llm_models()
self.available_tools = get_tools_from_command_list_module(cl_module) # command_listからツール取得
self.model_views = [] # model_viewsをここで初期化
self.update_model_views() # ここで SingleModelView が作成される
self.threads_and_workers_history = [] # 実行したスレッドとワーカーの記録用
self.processing_model_index = -1
self.model_task_queue = []
self._apply_current_font_settings() # model_views が初期化された後にフォント適用
self.setAcceptDrops(True) # ★★★ ドラッグアンドドロップを有効化 ★★★
def _apply_current_font_settings(self):
default_font = QFont(self.default_font_name, self.current_default_font_size)
label_font = QFont(self.default_font_name, self.current_label_font_size)
button_font = QFont(self.default_font_name, self.current_button_font_size)
groupbox_font = QFont(self.default_font_name, self.current_groupbox_font_size, QFont.Bold)
#self.enable_warmup = True
self.setFont(default_font)
if hasattr(self, 'agent_combo'): # 初期化順序によるエラー回避
self.global_settings_layout.itemAt(0).widget().setFont(label_font) # Compare Label
self.compare_count_spinbox.setFont(default_font)
self.warmup_checkbox.setFont(default_font) # グローバルウォームアップチェックボックスにフォント適用
self.common_input_group.setFont(groupbox_font)
self.prompt_input_edit.setFont(default_font)
self.image_path_label.setFont(label_font)
self.browse_image_button.setFont(button_font)
self.clear_image_button.setFont(button_font)
self.stop_button.setFont(button_font) # Stop button font
self.send_button.setFont(button_font)
self.clear_outputs_button.setFont(button_font)
if hasattr(self, 'font_size_spinbox'): # フォント変更UIが初期化されていれば
self.font_control_layout.itemAt(0).widget().setFont(label_font) # Global Font Size Label
self.font_size_spinbox.setFont(default_font)
# self.font_control_layout.itemAt(2).widget().setFont(button_font) # Apply Font Button
# self.global_settings_layout.itemAt(0).widget().setFont(label_font)
self.font_control_layout.itemAt(2).widget().setFont(button_font)
self.update_model_views_font()
def update_model_views_font(self):
for view in self.model_views:
view.update_font_settings(self.current_default_font_size, self.current_label_font_size)
def update_model_views(self):
num_models = self.compare_count_spinbox.value()
# 古いビューを削除
for view in self.model_views:
self.models_layout.removeWidget(view)
view.deleteLater()
self.model_views = []
# 新しいビューを作成
for i in range(num_models):
view = SingleModelView(i, self.available_llms, self.available_tools)
self.models_layout.addWidget(view)
# view.update_font_settings(self.current_default_font_size, self.current_label_font_size) # ここでは呼ばない
self.model_views.append(view)
# すべてのビューがレイアウトに追加された後に初期メッセージとフォントを設定
for view in self.model_views:
# view.output_widget.append_node(f"Model {view.model_id + 1}", "Ready to receive prompt.") # SingleModelViewの__init__で設定済みの想定
view.update_font_settings(self.current_default_font_size, self.current_label_font_size)
view.output_widget.update() # 明示的な更新を試みる
def browse_image(self):
# WORK_SPACE_DIR が設定されていればそこから開始、なければホームディレクトリ
start_dir = WORK_SPACE_DIR if WORK_SPACE_DIR and os.path.isdir(WORK_SPACE_DIR) else os.path.expanduser("~")
# 複数ファイル選択を許可
filePaths, _ = QFileDialog.getOpenFileNames(self, "Select Images", start_dir, "Images (*.png *.jpg *.jpeg *.bmp *.gif)")
if filePaths:
# # 既存のリストに追加
# self.selected_image_path.extend(filePaths)
# # 重複を排除し、ソートして一貫性を保つ
# self.selected_image_path = sorted(list(set(self.selected_image_path)))
#
# if len(self.selected_image_path) == 1:
# self.image_path_label.setText(f"Image: {os.path.basename(self.selected_image_path[0])}")
# else:
# self.image_path_label.setText(f"Images: {len(self.selected_image_path)} files selected")
self.add_images(filePaths)
def add_images(self, file_paths: list):
"""画像パスのリストを受け取り、UIを更新するヘルパーメソッド"""
self.selected_image_path.extend(file_paths)
# 重複を排除し、ソートして一貫性を保つ
self.selected_image_path = sorted(list(set(self.selected_image_path)))
if len(self.selected_image_path) == 1:
self.image_path_label.setText(f"Image: {os.path.basename(self.selected_image_path[0])}")
elif len(self.selected_image_path) > 1:
self.image_path_label.setText(f"Images: {len(self.selected_image_path)} files selected")
else: # 0件の場合
self.image_path_label.setText("Image (optional): None")
def dragEnterEvent(self, event):
# ドラッグされたデータにURL(ファイルパス)が含まれているかチェック
if event.mimeData().hasUrls():
event.acceptProposedAction() # ドロップ操作を受け入れる
else:
event.ignore() # 受け入れない
def dropEvent(self, event):
# ドロップされたファイルパスを取得
urls = event.mimeData().urls()
image_paths = []
for url in urls:
if url.isLocalFile():
file_path = url.toLocalFile()
if file_path.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp', '.gif')):
image_paths.append(file_path)
if image_paths:
self.add_images(image_paths)
def clear_image(self):
self.selected_image_path = []
self.image_path_label.setText("Image (optional): None")
def send_to_all_models(self):
InterruptedException.clear_cancel()
common_prompt = self.prompt_input_edit.toPlainText()
if not common_prompt.strip():
QMessageBox.warning(self, "Input Error", "Prompt cannot be empty.")
return
if self.processing_model_index != -1 and self.processing_model_index < len(self.model_task_queue):
QMessageBox.information(self, "Busy", "Models are still being processed.")
return
self.send_button.setEnabled(False)
self.clear_outputs_button.setEnabled(False)
self.stop_button.setEnabled(True) # Stop button enabled during processing
enable_warmup_global = self.warmup_checkbox.isChecked() # グローバル設定を取得
self.model_task_queue = []
for i, model_view in enumerate(self.model_views):
model_view.clear_output()
# ユーザー入力の表示 (ウォームアップとメインの両方)
#model_view.display_output("User (Warm-up)", "こんにちは")
model_view.display_output("User", common_prompt)
settings = model_view.get_settings()
settings['enable_warmup'] = enable_warmup_global # グローバル設定を個々の設定に追加
self.model_task_queue.append({'index': i, 'settings': settings})
self.processing_model_index = -1 # リセット
self.threads_and_workers_history.clear() # 古い履歴をクリア
self.execute_next_model_in_queue()
def execute_next_model_in_queue(self):
self.processing_model_index += 1
if self.processing_model_index < len(self.model_task_queue):
task_info = self.model_task_queue[self.processing_model_index]
model_idx = task_info['index']
settings = task_info['settings']
common_prompt = self.prompt_input_edit.toPlainText() # 最新のプロンプトを取得
thread = QThread()
worker = ComparisonWorker(model_idx, settings, common_prompt, self.selected_image_path)
worker.moveToThread(thread)
worker.result_ready.connect(self.handle_worker_result) # 引数変更に対応
worker.error_occurred.connect(self.handle_worker_error) # 引数変更に対応
worker.streaming_started.connect(self.handle_worker_streaming_started)
worker.new_token_ready.connect(self.handle_worker_new_token)
# このモデルの処理が完了したら次のモデルの処理を開始
worker.finished.connect(self.execute_next_model_in_queue)
# スレッドとワーカーのクリーンアップ
worker.finished.connect(thread.quit)
worker.finished.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
# ワーカーのrunメソッドをスレッド開始時に実行するように接続
thread.started.connect(worker.run)
self.threads_and_workers_history.append({'thread': thread, 'worker': worker})
thread.start()
else:
# 全モデルの処理が完了
QMessageBox.information(self, "Completed", "All models have finished processing.")
self.send_button.setEnabled(True)
self.clear_outputs_button.setEnabled(True)
self.processing_model_index = -1 # リセット
# self.threads_and_workers_history は記録として残しても良い
def stop_all_models(self):
print("Stop All Models button clicked.")
InterruptedException.set_cancel(True)
for item in self.threads_and_workers_history:
worker = item.get('worker')
if worker:
worker.cancel() # 各ワーカーにキャンセルを通知
# UIの更新
self.send_button.setEnabled(True)
self.clear_outputs_button.setEnabled(True)
self.stop_button.setEnabled(False)
# 実行キューをクリアし、処理インデックスをリセット
self.model_task_queue.clear()
self.processing_model_index = -1
QMessageBox.information(self, "Stopped", "Processing of all models has been requested to stop.")
def handle_worker_result(self, model_index, type, text, perf_data): # type引数追加
if 0 <= model_index < len(self.model_views):
# ChatPaintWidgetのテキストはストリーミング中に更新されている。
# finalize_streaming_displayは主にパフォーマンスデータを更新する。
# text引数は現状のfinalize_streaming_displayでは直接使われない。
self.model_views[model_index].finalize_streaming_display(text, perf_data)
# 必要であれば、type ("warmup" or "main") に応じてパフォーマンスラベルの表示を調整
def handle_worker_error(self, model_index, error_message):
if 0 <= model_index < len(self.model_views):
# エラーメッセージもタイトルで区別して表示
self.model_views[model_index].display_output(f"Error ({self.model_views[model_index].get_settings()['llm_model']})", error_message)
def handle_worker_streaming_started(self, model_index, type, title): # type引数追加
if 0 <= model_index < len(self.model_views):
# ユーザープロンプトの表示はsend_to_all_modelsで行っているので、
# ここではAIの応答が始まることを示すノード (Warm-up用またはMain用) を開始
self.model_views[model_index].start_streaming_display(title)
def handle_worker_new_token(self, model_index, token):
if 0 <= model_index < len(self.model_views):
self.model_views[model_index].append_streaming_text(token)
# ワーカーがキャンセルされていたらトークンを追加しない
worker_item = next((item for item in self.threads_and_workers_history if item['worker'].model_index == model_index), None)
if worker_item and worker_item['worker']._is_cancelled:
return
self.model_views[model_index].append_streaming_text(token)
def clear_all_outputs(self):
for view in self.model_views:
view.clear_output()
self.prompt_input_edit.clear()
self.clear_image()
def closeEvent(self, event):
# アプリケーション終了時に実行中のスレッドがあれば終了処理
print("ComparisonGui.closeEvent called.")
active_threads_to_wait_for = []
global _g_active_comparison_gui_instance
_g_active_comparison_gui_instance = None # GUI終了時に参照をクリア
# Iterate over a copy of the list in case it's modified,
# though in this specific logic, direct modification of the list
# during this first loop is not happening.
for item in list(self.threads_and_workers_history):
thread = item.get('thread')
worker = item.get('worker') # For logging purposes
if thread is None:
# This should ideally not happen if items are added correctly.
print(f" Skipping item as thread is None (worker: {worker.model_index if worker and hasattr(worker, 'model_index') else 'N/A'}).")
continue
try:
if thread.isRunning():
print(f" Thread for model {worker.model_index if worker and hasattr(worker, 'model_index') else 'N/A'} is running. Attempting to quit.")
thread.quit()
active_threads_to_wait_for.append(thread)
# If not running, it's assumed to be finished or will be handled by deleteLater.
# No specific action needed for non-running threads in this loop part.
except RuntimeError:
# This catches cases where the C++ QThread object has already been deleted.
model_idx_str = "N/A"
if worker and hasattr(worker, 'model_index'):
model_idx_str = str(worker.model_index)
print(f" Warning: Thread object for model {model_idx_str} was already deleted (RuntimeError accessing isRunning).")
if active_threads_to_wait_for:
print(f"Waiting for {len(active_threads_to_wait_for)} active threads to finish...")
for thread_to_wait in active_threads_to_wait_for:
try:
if not thread_to_wait.wait(1000): # Wait for 1 second
print(f" Warning: Thread (object: {thread_to_wait}) did not terminate gracefully. Forcing termination.")
thread_to_wait.terminate()
thread_to_wait.wait() # Wait for termination to complete
else:
print(f" Thread (object: {thread_to_wait}) finished gracefully.")
except RuntimeError:
print(f" Warning: Thread object (object: {thread_to_wait}) became invalid during wait in closeEvent.")
else:
print("No active threads were found to wait for.")
self.threads_and_workers_history.clear() # Clear the history list after processing
print("threads_and_workers_history cleared.")
super().closeEvent(event)
def apply_global_font_size(self):
new_base_size = self.font_size_spinbox.value()
self.current_default_font_size = new_base_size
# 他のフォントサイズもベースサイズに基づいて調整する（例）
self.current_label_font_size = max(8, int(new_base_size * 0.85)) # ラベルは少し小さく
self.current_button_font_size = max(8, int(new_base_size * 0.9)) # ボタンも調整
self.current_groupbox_font_size = new_base_size # グループボックスは同じか少し大きく
self._apply_current_font_settings()
self.update_model_views_font() # 各モデルビューのフォントも更新
if __name__ == '__main__':
# このファイルが直接実行された場合の処理 (通常は main_comparison.py から呼び出される)
# QApplicationインスタンスはメインのエントリポイント(main_comparison.py)で作成される想定
app = QApplication.instance()
if not app: # main_comparison.py を介さずに直接実行された場合
print("Warning: ComparisonGui.py is being run directly. WORK_SPACE_DIR might not be set correctly by an entry point script.")
app = QApplication(sys.argv)
gui = ComparisonGui()
gui.show()
sys.exit(app.exec())
AIによる説明
LLM比較ツールGUI (ComparisonGui.py) リファレンスマニュアル
このドキュメントは、ComparisonGui.py ファイルで定義されているLLM比較ツールGUIの使用方法と内部構造を説明します。このGUIは、複数のLLMモデルを同時に実行し、その結果を比較するために設計されています。
1. GUIの概要
ComparisonGui は、PySide6 を使用して構築されたGUIアプリケーションです。複数のLLMモデルを同時に実行し、それぞれの出力とパフォーマンスメトリクスを比較表示します。主な機能は以下の通りです。
- モデル選択: 利用可能なLLMモデル（Ollama、Geminiなど）とツールを選択できます。Ollamaモデルは、Ollamaサーバーから動的に取得されます。
- プロンプト入力: 全てのモデルに共通のプロンプトを入力できます。
- 画像入力: 複数の画像ファイルをドラッグアンドドロップ、またはファイル選択ダイアログで指定できます。
- 実行制御: “Send to All Models” ボタンで全てのモデルに対してプロンプトを実行します。”Stop All Models” ボタンで実行中のモデルを停止できます。”Clear All Outputs” ボタンで全ての出力結果をクリアします。
- 出力表示: 各モデルの出力は、
ChatPaintWidgetを使用してチャット形式で表示されます。パフォーマンスメトリクス（処理時間、生成トークン数、トークン毎秒）も表示されます。 - フォントサイズ調整: グローバルなフォントサイズをスピンボックスとボタンで調整できます。
- ウォームアップ: 各モデルの実行前にウォームアップを実行するかどうかを選択できます（グローバル設定とモデルごとの設定）。
2. クラス構造
2.1 ComparisonGui
メインのGUIウィンドウを表すクラスです。以下の主要な役割を持ちます。
- ウィジェットの配置:
QVBoxLayoutを使用して、モデル設定エリア、共通入力エリア、出力表示エリアなどを配置します。 - モデルビューの管理:
SingleModelViewインスタンスのリスト (model_views) を管理し、モデルの追加・削除、設定の更新を行います。 - イベント処理: ボタンクリック、フォントサイズ変更、ドラッグアンドドロップなどのイベントを処理します。
- ワーカーの管理:
ComparisonWorkerを使用してLLMモデルを非同期で実行し、結果を処理します。スレッドとワーカーの履歴をthreads_and_workers_historyに保持します。 - エラー処理: LLM実行中のエラーをキャッチし、エラーメッセージを表示します。
- 終了処理: アプリケーション終了時に実行中のスレッドを適切に終了させます。
2.2 SingleModelView
各LLMモデルの設定と出力表示を行うウィジェットを表すクラスです。
- モデル設定: LLMモデル、使用するツールを選択するためのコンボボックスとチェックボックスを提供します。
- 出力表示:
ChatPaintWidgetを使用して、LLMモデルからの出力を表示します。 - パフォーマンス表示:
perf_labelにパフォーマンスメトリクスを表示します。 - フォント設定:
update_font_settingsメソッドでフォントサイズを更新します。 - ストリーミング表示:
start_streaming_display,append_streaming_text,finalize_streaming_displayメソッドでストリーミング出力を処理します。
2.3 ComparisonWorker
各LLMモデルを別スレッドで実行するワーカークラスです。QObject を継承し、シグナル/スロット機構を使用して ComparisonGui と通信します。
- モデル実行:
AIAgentを使用してLLMモデルを実行します。 - ウォームアップ実行:
enable_warmupフラグに基づいてウォームアップを実行します。 - パフォーマンス計測:
get_performance_metrics関数を使用してパフォーマンスメトリクスを計測します。 - 結果送信:
result_readyシグナルで結果をComparisonGuiに送信します。 - エラー処理:
error_occurredシグナルでエラーをComparisonGuiに送信します。 - ストリーミング処理:
new_token_readyシグナルでストリーミング中のトークンをComparisonGuiに送信します。 - 中断処理:
cancelメソッドで実行を中断します。
2.4 WorkerCallbackHandler
ComparisonWorker 内で使用されるコールバックハンドラです。PaintGUICallbackHandler を継承し、LLMからのストリーミング出力を処理します。
3. 関数
is_predominantly_japanese(text, threshold_ratio=0.3): テキストが主に日本語かどうかを判定します。get_performance_metrics(start_time, end_time, response_text, llm_output_metadata=None): LLMモデルのパフォーマンスメトリクスを計測します。get_available_llm_models(): 利用可能なLLMモデルのリストを取得します。Ollamaと静的に定義されたモデルを組み合わせます。get_tools_from_command_list_module(module):tools.command_listモジュールから利用可能なツール情報を取得します。append_node(name: str, text: str): アクティブなChatPaintWidgetにノードを追加します。set_last_node_text(text: str): アクティブなChatPaintWidgetの最後のノードのテキストを更新します。
4. 使用方法
- 環境設定: Ollamaサーバーが実行されていることを確認します。
tools.program_called_command_listモジュールでワークスペースディレクトリを設定する必要があります。 - GUI起動:
main_comparison.py(このファイルには含まれていませんが、このGUIを起動するメインスクリプトが必要です) を実行します。 - モデル選択: 比較するモデルの数と、各モデルの設定（LLM、ツール）を選択します。
- プロンプト入力: 共通のプロンプトを入力します。
- 画像入力 (オプション): 画像ファイルをドラッグアンドドロップするか、”Browse Image” ボタンで選択します。
- 実行: “Send to All Models” ボタンをクリックしてモデルを実行します。
- 結果確認: 各モデルの出力とパフォーマンスメトリクスを確認します。
- 停止: “Stop All Models” ボタンで実行を停止します。
- クリア: “Clear All Outputs” ボタンで出力をクリアします。
5. 注意点
- Ollamaサーバーへの接続に失敗した場合、エラーメッセージが表示されます。
ChatPaintWidgetの内部実装に依存している部分があります。ChatPaintWidgetの変更に合わせてコードを修正する必要がある可能性があります。- エラー処理は、より詳細なエラーメッセージとログ出力を行うように改善できます。
- 大量のモデルを同時に実行すると、リソース消費が大きくなる可能性があります。
このマニュアルが、ComparisonGui.py の理解と使用に役立つことを願っています。