mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-05-07 02:51:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Modularize frontend

Browse source
This commit is contained in:
Jai Suphavadeeprasit 2025-10-13 11:53:13 -04:00
parent bb5eab2645
commit c2d5a28d15
8 changed files with 1353 additions and 597 deletions
Download patch file Download diff file Expand all files Collapse all files

1
mock_web_tools.py
View file

@ -55,7 +55,6 @@ def mock_web_search(query: str, delay: int = 2) -> str:
}
}
print(f"✅ [MOCK] Search completed with {len(result['data']['web'])} results")
return json.dumps(result, indent=2)

527
output.txt Normal file
View file

File diff suppressed because one or more lines are too long

6
run_agent.py
View file

@ -398,15 +398,9 @@ class AIAgent:
Returns:
Dict: Complete conversation result with final response and message history
"""
# ============================================================
# WEBSOCKET LOGGING: Session Initialization
# ============================================================
# Generate unique session ID for this agent execution (or use provided one)
# This ID will be used to link all events together in the log file
if session_id is None:
session_id = str(uuid.uuid4())
# Initialize WebSocket logger if enabled (via --enable_websocket_logging flag)
# Uses synchronous API - no event loop management in agent layer
if self.enable_websocket_logging:
try:

23
ui/__init__.py Normal file
View file

@ -0,0 +1,23 @@
"""
Hermes Agent UI Package
A modular PySide6 UI for the Hermes AI Agent with real-time event streaming.
Modules:
- websocket_client: WebSocket communication
- event_widgets: Event display components
- main_window: Main application window
- hermes_ui: Application entry point
"""
from .websocket_client import WebSocketClient
from .event_widgets import CollapsibleEventWidget, InteractiveEventDisplayWidget
from .main_window import HermesMainWindow
__all__ = [
'WebSocketClient',
'CollapsibleEventWidget',
'InteractiveEventDisplayWidget',
'HermesMainWindow',
]

334
ui/event_widgets.py Normal file
View file

@ -0,0 +1,334 @@
"""
Event display widgets for Hermes Agent UI.
This module provides widgets for displaying and managing real-time agent events
in a collapsible, filterable interface.
"""
import json
from datetime import datetime
from typing import Dict, Any
from PySide6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QCheckBox, QGroupBox, QFrame, QScrollArea
)
from PySide6.QtCore import Qt, QTimer
from PySide6.QtGui import QFont
class CollapsibleEventWidget(QFrame):
"""
A single collapsible event with expand/collapse functionality.
"""
def __init__(self, event: Dict[str, Any], parent=None):
super().__init__(parent)
self.event = event
self.is_expanded = False
self.event_type = event.get("event_type", "unknown")
self.setFrameStyle(QFrame.Box | QFrame.Raised)
self.setLineWidth(1)
self.setup_ui()
def setup_ui(self):
"""Initialize UI components."""
layout = QVBoxLayout()
layout.setContentsMargins(8, 8, 8, 8)
layout.setSpacing(4)
# Header (clickable)
self.header_widget = QWidget()
header_layout = QHBoxLayout()
header_layout.setContentsMargins(0, 0, 0, 0)
self.expand_indicator = QLabel("▶")
self.expand_indicator.setFixedWidth(20)
header_layout.addWidget(self.expand_indicator)
self.summary_label = QLabel()
self.summary_label.setFont(QFont("Arial", 10, QFont.Bold))
self.update_summary()
header_layout.addWidget(self.summary_label, 1)
# Timestamp
timestamp = self.event.get("timestamp", datetime.now().isoformat())
time_str = datetime.fromisoformat(timestamp.replace('Z', '+00:00')).strftime("%H:%M:%S")
time_label = QLabel(time_str)
time_label.setStyleSheet("color: #888;")
header_layout.addWidget(time_label)
self.header_widget.setLayout(header_layout)
self.header_widget.mousePressEvent = lambda e: self.toggle_expand()
self.header_widget.setCursor(Qt.PointingHandCursor)
layout.addWidget(self.header_widget)
# Details (collapsible)
self.details_widget = QWidget()
self.details_layout = QVBoxLayout()
self.details_layout.setContentsMargins(25, 5, 5, 5)
self.populate_details()
self.details_widget.setLayout(self.details_layout)
self.details_widget.setVisible(False)
layout.addWidget(self.details_widget)
self.setLayout(layout)
self.apply_colors()
def apply_colors(self):
"""Apply color scheme based on event type."""
colors = {
"query": "#E8F5E9", # Light green
"api_call": "#E3F2FD", # Light blue
"response": "#F3E5F5", # Light purple
"tool_call": "#FFF3E0", # Light orange
"tool_result": "#E8F5E9", # Light green
"complete": "#E8F5E9", # Light green
"error": "#FFEBEE", # Light red
"session_start": "#F5F5F5" # Light gray
}
bg_color = colors.get(self.event_type, "#FAFAFA")
self.setStyleSheet(f"""
CollapsibleEventWidget {{
background-color: {bg_color};
border: 1px solid #ddd;
border-radius: 4px;
}}
""")
def update_summary(self):
"""Update the summary label with event type."""
self.summary_label.setText(f"- {self.event_type.upper()}")
def populate_details(self):
"""Populate the details section with event data."""
data = self.event.get("data", {})
# Clear existing details
while self.details_layout.count():
item = self.details_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
self.add_detail("Raw Data", json.dumps(data, indent=2), multiline=True)
def add_detail(self, label: str, value: str, multiline: bool = True):
"""Add a detail row to the details section."""
detail_widget = QWidget()
detail_layout = QVBoxLayout() if multiline else QHBoxLayout()
detail_layout.setContentsMargins(0, 2, 0, 2)
label_widget = QLabel(f"<b>{label}:</b>")
label_widget.setTextFormat(Qt.RichText)
value_widget = QLabel(value)
value_widget.setWordWrap(True)
value_widget.setTextInteractionFlags(Qt.TextSelectableByMouse)
if multiline:
font = QFont()
font.setStyleHint(QFont.Monospace)
font.setPointSize(9)
value_widget.setFont(font)
value_widget.setStyleSheet("background-color: #f5f5f5; padding: 5px; border-radius: 3px;")
detail_layout.addWidget(label_widget)
detail_layout.addWidget(value_widget)
else:
detail_layout.addWidget(label_widget)
detail_layout.addWidget(value_widget, 1)
detail_widget.setLayout(detail_layout)
self.details_layout.addWidget(detail_widget)
def toggle_expand(self):
"""Toggle expanded/collapsed state."""
self.is_expanded = not self.is_expanded
self.details_widget.setVisible(self.is_expanded)
self.expand_indicator.setText("▼" if self.is_expanded else "▶")
class InteractiveEventDisplayWidget(QWidget):
"""
Interactive widget for displaying real-time agent events.
Features:
- Collapsible event items
- Event type filtering
- Expand/collapse all
- Auto-scroll to latest events
"""
def __init__(self):
super().__init__()
self.events = []
self.event_widgets = []
self.current_session = None
self.filters = {
"query": True,
"api_call": True,
"response": True,
"tool_call": True,
"tool_result": True,
"complete": True,
"error": True,
"session_start": True
}
self.init_ui()
def init_ui(self):
"""Initialize the UI components."""
layout = QVBoxLayout()
layout.setContentsMargins(5, 5, 5, 5)
# Header with controls
header_layout = QHBoxLayout()
title = QLabel("📡 Real-time Event Stream")
title.setFont(QFont("Arial", 12, QFont.Bold))
header_layout.addWidget(title)
header_layout.addStretch()
# Expand/Collapse All buttons
expand_all_btn = QPushButton("Expand All")
expand_all_btn.clicked.connect(self.expand_all)
header_layout.addWidget(expand_all_btn)
collapse_all_btn = QPushButton("Collapse All")
collapse_all_btn.clicked.connect(self.collapse_all)
header_layout.addWidget(collapse_all_btn)
# Clear button
clear_btn = QPushButton("🗑️ Clear")
clear_btn.clicked.connect(self.clear_events)
header_layout.addWidget(clear_btn)
layout.addLayout(header_layout)
# Filter controls
filter_group = QGroupBox("Event Filters (Show/Hide)")
filter_layout = QHBoxLayout()
filter_layout.setSpacing(10)
self.filter_checkboxes = {}
filter_configs = [
("query", "📝 Queries"),
("api_call", "🔄 API Calls"),
("response", "🤖 Responses"),
("tool_call", "🔧 Tool Calls"),
("tool_result", "✅ Results"),
("complete", "🎉 Complete"),
("error", "❌ Errors"),
]
for event_type, label in filter_configs:
checkbox = QCheckBox(label)
checkbox.setChecked(True)
checkbox.stateChanged.connect(lambda state, et=event_type: self.toggle_filter(et, state))
self.filter_checkboxes[event_type] = checkbox
filter_layout.addWidget(checkbox)
filter_group.setLayout(filter_layout)
layout.addWidget(filter_group)
# Scroll area for events
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
# Container for event widgets
self.events_container = QWidget()
self.events_layout = QVBoxLayout()
self.events_layout.setSpacing(5)
self.events_layout.addStretch() # Push events to top
self.events_container.setLayout(self.events_layout)
scroll_area.setWidget(self.events_container)
layout.addWidget(scroll_area)
self.setLayout(layout)
def clear_events(self):
"""Clear all displayed events."""
self.events.clear()
self.event_widgets.clear()
# Remove all widgets
while self.events_layout.count() > 1: # Keep the stretch
item = self.events_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
self.current_session = None
def add_event(self, event: Dict[str, Any]):
"""Add an event to the display."""
event_type = event.get("event_type", "unknown")
session_id = event.get("session_id", "")
# Track session changes - add session start event
if self.current_session != session_id:
self.current_session = session_id
session_event = {
"event_type": "session_start",
"session_id": session_id,
"timestamp": event.get("timestamp", datetime.now().isoformat()),
"data": {
"session_id": session_id,
"start_time": event.get("timestamp", datetime.now().isoformat())
}
}
self._add_event_widget(session_event)
# Add the actual event
self._add_event_widget(event)
def _add_event_widget(self, event: Dict[str, Any]):
"""Internal method to add event widget."""
event_widget = CollapsibleEventWidget(event)
# Apply filter visibility
event_type = event.get("event_type", "unknown")
event_widget.setVisible(self.filters.get(event_type, True))
# Insert before the stretch
self.events_layout.insertWidget(self.events_layout.count() - 1, event_widget)
self.events.append(event)
self.event_widgets.append(event_widget)
# Auto-scroll to bottom after widget is rendered
QTimer.singleShot(50, self._scroll_to_bottom)
def _scroll_to_bottom(self):
"""Scroll to the bottom of the events list."""
scroll_area = self.events_container.parent()
if isinstance(scroll_area, QScrollArea):
scroll_bar = scroll_area.verticalScrollBar()
scroll_bar.setValue(scroll_bar.maximum())
def expand_all(self):
"""Expand all event widgets."""
for widget in self.event_widgets:
if not widget.is_expanded:
widget.toggle_expand()
def collapse_all(self):
"""Collapse all event widgets."""
for widget in self.event_widgets:
if widget.is_expanded:
widget.toggle_expand()
def toggle_filter(self, event_type: str, state: int):
"""Toggle visibility of events by type."""
self.filters[event_type] = bool(state)
# Update visibility of existing widgets
for event, widget in zip(self.events, self.event_widgets):
if event.get("event_type") == event_type:
widget.setVisible(self.filters[event_type])

593
ui/hermes_ui.py
View file

@ -18,602 +18,16 @@ Usage:
"""
import sys
import json
import signal
import os
import requests
from datetime import datetime
from typing import Dict, Any, List, Optional
# Suppress Qt logging warnings BEFORE importing Qt
os.environ['QT_LOGGING_RULES'] = 'qt.qpa.*=false'
from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QTextEdit, QPushButton, QLabel, QLineEdit, QComboBox, QCheckBox,
QGroupBox, QSplitter, QListWidget, QListWidgetItem,
QTextBrowser, QSpinBox, QMessageBox
)
from PySide6.QtCore import (
Qt, Signal, Slot, QObject, QTimer
)
from PySide6.QtGui import (
QFont, QTextCursor
)
from PySide6.QtWidgets import QApplication
from PySide6.QtCore import QTimer
# WebSocket imports
import websocket
import threading
class WebSocketClient(QObject):
"""
WebSocket client for receiving real-time agent events.
Runs in a separate thread and emits Qt signals when events arrive.
"""
# Signals for event communication
event_received = Signal(dict) # Emits parsed event data
connected = Signal()
disconnected = Signal()
error = Signal(str)
def __init__(self, url: str = "ws://localhost:8000/ws"):
super().__init__()
self.url = url
self.ws = None
self.running = False
self.thread = None
def connect(self):
"""Start WebSocket connection in background thread."""
if self.running:
return
self.running = True
self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start()
def disconnect(self):
"""Stop WebSocket connection."""
self.running = False
if self.ws:
try:
self.ws.close()
except Exception as e:
print(f"Error closing WebSocket: {e}")
def _run(self):
"""WebSocket event loop (runs in background thread)."""
try:
self.ws = websocket.WebSocketApp(
self.url,
on_open=self._on_open,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close
)
# Run forever with reconnection
self.ws.run_forever(ping_interval=300, ping_timeout=60)
except Exception as e:
self.error.emit(f"WebSocket error: {str(e)}")
def _on_open(self, ws):
"""Called when WebSocket connection is established."""
print("🔌 WebSocket connected")
self.connected.emit()
def _on_message(self, ws, message):
"""Called when a message is received from the server."""
try:
data = json.loads(message)
self.event_received.emit(data)
except json.JSONDecodeError as e:
print(f"❌ Failed to parse WebSocket message: {e}")
def _on_error(self, ws, error):
"""Called when an error occurs."""
print(f"❌ WebSocket error: {error}")
self.error.emit(str(error))
def _on_close(self, ws, close_status_code, close_msg):
"""Called when WebSocket connection is closed."""
print(f"🔌 WebSocket disconnected: {close_status_code} - {close_msg}")
self.disconnected.emit()
class EventDisplayWidget(QWidget):
"""
Widget for displaying real-time agent events in a formatted view.
Shows events in chronological order with color coding and formatting.
"""
def __init__(self):
super().__init__()
self.init_ui()
self.current_session = None
def init_ui(self):
"""Initialize the UI components."""
layout = QVBoxLayout()
# Header
header = QLabel("📡 Real-time Event Stream")
header.setFont(QFont("Arial", 12, QFont.Bold))
layout.addWidget(header)
# Event display (rich text browser) - use system monospace font
self.event_display = QTextBrowser()
self.event_display.setOpenExternalLinks(False)
# Use system monospace font instead of hardcoded "Monaco" or "Courier"
font = QFont()
font.setStyleHint(QFont.Monospace)
font.setPointSize(10)
self.event_display.setFont(font)
layout.addWidget(self.event_display)
# Clear button
clear_btn = QPushButton("🗑️ Clear Events")
clear_btn.clicked.connect(self.clear_events)
layout.addWidget(clear_btn)
self.setLayout(layout)
def clear_events(self):
"""Clear all displayed events."""
self.event_display.clear()
self.current_session = None
def add_event(self, event: Dict[str, Any]):
"""
Add an event to the display with formatting.
Args:
event: Event data from WebSocket
"""
event_type = event.get("event_type", "unknown")
session_id = event.get("session_id", "")
data = event.get("data", {})
timestamp = event.get("timestamp", datetime.now().isoformat())
# Track session changes
if self.current_session != session_id:
self.current_session = session_id
self.event_display.append(f"\n{'='*80}")
self.event_display.append(f"<b>🆕 New Session: {session_id[:8]}...</b>")
self.event_display.append(f"{'='*80}\n")
# Format based on event type
if event_type == "query":
query = data.get("query", "")
self.event_display.append(f"<b style='color: #4CAF50;'>📝 QUERY</b>")
self.event_display.append(f" <i>{query}</i>")
self.event_display.append(f" Model: {data.get('model', 'N/A')}")
self.event_display.append(f" Toolsets: {', '.join(data.get('toolsets', []) or ['all'])}")
elif event_type == "api_call":
call_num = data.get("call_number", 0)
self.event_display.append(f"\n<b style='color: #2196F3;'>🔄 API CALL #{call_num}</b>")
self.event_display.append(f" Messages: {data.get('message_count', 0)}")
elif event_type == "response":
content = data.get("content", "")[:200]
self.event_display.append(f"<b style='color: #9C27B0;'>🤖 RESPONSE</b>")
if content:
self.event_display.append(f" {content}...")
self.event_display.append(f" Tool calls: {data.get('tool_call_count', 0)}")
self.event_display.append(f" Duration: {data.get('duration', 0):.2f}s")
elif event_type == "tool_call":
tool_name = data.get("tool_name", "unknown")
params = data.get("parameters", {})
self.event_display.append(f"<b style='color: #FF9800;'>🔧 TOOL CALL: {tool_name}</b>")
self.event_display.append(f" Parameters: {json.dumps(params, indent=2)[:100]}...")
elif event_type == "tool_result":
tool_name = data.get("tool_name", "unknown")
result = data.get("result", "")[:200]
duration = data.get("duration", 0)
error = data.get("error")
if error:
self.event_display.append(f"<b style='color: #F44336;'>❌ TOOL ERROR: {tool_name}</b>")
self.event_display.append(f" {error}")
else:
self.event_display.append(f"<b style='color: #4CAF50;'>✅ TOOL RESULT: {tool_name}</b>")
self.event_display.append(f" Duration: {duration:.2f}s")
if result:
self.event_display.append(f" Result preview: {result}...")
elif event_type == "complete":
final_response = data.get("final_response", "")[:300]
total_calls = data.get("total_calls", 0)
completed = data.get("completed", False)
status_icon = "🎉" if completed else "⚠️"
self.event_display.append(f"\n<b style='color: #4CAF50;'>{status_icon} SESSION COMPLETE</b>")
self.event_display.append(f" Total API calls: {total_calls}")
self.event_display.append(f" Status: {'Success' if completed else 'Failed/Incomplete'}")
if final_response:
self.event_display.append(f" Final response: {final_response}...")
self.event_display.append(f"\n{'='*80}\n")
elif event_type == "error":
error_msg = data.get("error_message", "Unknown error")
self.event_display.append(f"<b style='color: #F44336;'>❌ ERROR</b>")
self.event_display.append(f" {error_msg}")
else:
# Unknown event type
self.event_display.append(f"<b>⚠️ {event_type.upper()}</b>")
self.event_display.append(f" {json.dumps(data, indent=2)[:200]}...")
self.event_display.append("") # Blank line
# Auto-scroll to bottom
cursor = self.event_display.textCursor()
cursor.movePosition(QTextCursor.End)
self.event_display.setTextCursor(cursor)
class HermesMainWindow(QMainWindow):
"""
Main window for Hermes Agent UI.
Provides interface for:
- Submitting queries
- Configuring agent settings
- Viewing real-time events
- Managing sessions
"""
def __init__(self):
super().__init__()
self.api_base_url = "http://localhost:8000"
self.ws_client = None
self.current_session_id = None
self.available_toolsets = []
self.is_closing = False # Flag to prevent reconnection during shutdown
self.init_ui()
self.setup_websocket()
self.load_available_tools()
def init_ui(self):
"""Initialize the user interface."""
self.setWindowTitle("Hermes Agent - AI Assistant UI")
self.setGeometry(100, 100, 1400, 900)
# Central widget
central_widget = QWidget()
self.setCentralWidget(central_widget)
# Main layout (horizontal split)
main_layout = QHBoxLayout()
# Left panel: Controls
left_panel = self.create_control_panel()
# Right panel: Event display
right_panel = self.create_event_panel()
# Splitter for resizable panels
splitter = QSplitter(Qt.Horizontal)
splitter.addWidget(left_panel)
splitter.addWidget(right_panel)
splitter.setStretchFactor(0, 1) # Control panel
splitter.setStretchFactor(1, 2) # Event panel (larger)
main_layout.addWidget(splitter)
central_widget.setLayout(main_layout)
# Status bar
self.statusBar().showMessage("Ready")
def create_control_panel(self) -> QWidget:
"""Create the left control panel."""
panel = QWidget()
layout = QVBoxLayout()
# Title
title = QLabel("🤖 Hermes Agent Control")
title.setFont(QFont("Arial", 14, QFont.Bold))
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
# Query input group
query_group = QGroupBox("Query Input")
query_layout = QVBoxLayout()
self.query_input = QTextEdit()
self.query_input.setPlaceholderText("Enter your query here...")
self.query_input.setMaximumHeight(150)
query_layout.addWidget(self.query_input)
self.submit_btn = QPushButton("🚀 Submit Query")
self.submit_btn.setFont(QFont("Arial", 11, QFont.Bold))
self.submit_btn.setStyleSheet("QPushButton { background-color: #4CAF50; color: white; padding: 10px; }")
self.submit_btn.clicked.connect(self.submit_query)
query_layout.addWidget(self.submit_btn)
query_group.setLayout(query_layout)
layout.addWidget(query_group)
# Model configuration group
model_group = QGroupBox("Model Configuration")
model_layout = QVBoxLayout()
# Model selection
model_layout.addWidget(QLabel("Model:"))
self.model_combo = QComboBox()
self.model_combo.addItems([
"claude-sonnet-4-5-20250929",
"claude-opus-4-20250514",
"gpt-4",
"gpt-4-turbo"
])
model_layout.addWidget(self.model_combo)
# API Base URL
model_layout.addWidget(QLabel("API Base URL:"))
self.base_url_input = QLineEdit("https://api.anthropic.com/v1/")
model_layout.addWidget(self.base_url_input)
# Max turns
model_layout.addWidget(QLabel("Max Turns:"))
self.max_turns_spin = QSpinBox()
self.max_turns_spin.setMinimum(1)
self.max_turns_spin.setMaximum(50)
self.max_turns_spin.setValue(10)
model_layout.addWidget(self.max_turns_spin)
model_group.setLayout(model_layout)
layout.addWidget(model_group)
# Tools configuration group
tools_group = QGroupBox("Tools & Toolsets")
tools_layout = QVBoxLayout()
tools_layout.addWidget(QLabel("Select Toolsets:"))
self.toolsets_list = QListWidget()
self.toolsets_list.setSelectionMode(QListWidget.MultiSelection)
self.toolsets_list.setMaximumHeight(150)
tools_layout.addWidget(self.toolsets_list)
tools_group.setLayout(tools_layout)
layout.addWidget(tools_group)
# Options group
options_group = QGroupBox("Options")
options_layout = QVBoxLayout()
self.mock_mode_checkbox = QCheckBox("Mock Web Tools (Testing)")
options_layout.addWidget(self.mock_mode_checkbox)
self.verbose_checkbox = QCheckBox("Verbose Logging")
options_layout.addWidget(self.verbose_checkbox)
options_layout.addWidget(QLabel("Mock Delay (seconds):"))
self.mock_delay_spin = QSpinBox()
self.mock_delay_spin.setMinimum(1)
self.mock_delay_spin.setMaximum(300)
self.mock_delay_spin.setValue(60)
options_layout.addWidget(self.mock_delay_spin)
options_group.setLayout(options_layout)
layout.addWidget(options_group)
# Connection status
self.connection_status = QLabel("🔴 Disconnected")
self.connection_status.setAlignment(Qt.AlignCenter)
self.connection_status.setStyleSheet("QLabel { padding: 5px; background-color: #F44336; color: white; border-radius: 3px; }")
layout.addWidget(self.connection_status)
# Add stretch to push everything to top
layout.addStretch()
panel.setLayout(layout)
return panel
def create_event_panel(self) -> QWidget:
"""Create the right event display panel."""
panel = QWidget()
layout = QVBoxLayout()
# Event display widget
self.event_widget = EventDisplayWidget()
layout.addWidget(self.event_widget)
panel.setLayout(layout)
return panel
def setup_websocket(self):
"""Setup WebSocket connection for real-time events."""
self.ws_client = WebSocketClient("ws://localhost:8000/ws")
# Connect signals
self.ws_client.connected.connect(self.on_ws_connected)
self.ws_client.disconnected.connect(self.on_ws_disconnected)
self.ws_client.error.connect(self.on_ws_error)
self.ws_client.event_received.connect(self.on_event_received)
# Start connection
self.ws_client.connect()
@Slot()
def on_ws_connected(self):
"""Called when WebSocket connection is established."""
self.connection_status.setText("🟢 Connected")
self.connection_status.setStyleSheet("QLabel { padding: 5px; background-color: #4CAF50; color: white; border-radius: 3px; }")
self.statusBar().showMessage("WebSocket connected")
@Slot()
def on_ws_disconnected(self):
"""Called when WebSocket connection is lost."""
# Don't attempt reconnection if we're closing the application
if self.is_closing:
return
self.connection_status.setText("🔴 Disconnected")
self.connection_status.setStyleSheet("QLabel { padding: 5px; background-color: #F44336; color: white; border-radius: 3px; }")
self.statusBar().showMessage("WebSocket disconnected - attempting reconnect...")
# Attempt reconnect after 5 seconds
QTimer.singleShot(5000, self.ws_client.connect)
@Slot(str)
def on_ws_error(self, error: str):
"""Called when WebSocket error occurs."""
self.statusBar().showMessage(f"WebSocket error: {error}")
@Slot(dict)
def on_event_received(self, event: Dict[str, Any]):
"""
Called when an event is received from WebSocket.
Args:
event: Event data from server
"""
self.event_widget.add_event(event)
# Update status for specific events
event_type = event.get("event_type")
if event_type == "query":
self.statusBar().showMessage("Query received - agent processing...")
elif event_type == "complete":
self.statusBar().showMessage("Agent completed!")
self.submit_btn.setEnabled(True)
def load_available_tools(self):
"""Load available toolsets from the API."""
try:
response = requests.get(f"{self.api_base_url}/tools", timeout=5)
if response.status_code == 200:
data = response.json()
toolsets = data.get("toolsets", [])
self.available_toolsets = toolsets
self.toolsets_list.clear()
for toolset in toolsets:
name = toolset.get("name", "")
description = toolset.get("description", "")
tool_count = toolset.get("tool_count", 0)
item_text = f"{name} ({tool_count} tools) - {description}"
item = QListWidgetItem(item_text)
item.setData(Qt.UserRole, name) # Store toolset name
self.toolsets_list.addItem(item)
self.statusBar().showMessage(f"Loaded {len(toolsets)} toolsets")
else:
self.statusBar().showMessage("Failed to load toolsets from API")
except requests.exceptions.RequestException as e:
self.statusBar().showMessage(f"Error loading toolsets: {str(e)}")
# Add some default toolsets
default_toolsets = ["web", "vision", "terminal", "research"]
for ts in default_toolsets:
item = QListWidgetItem(f"{ts} (default)")
item.setData(Qt.UserRole, ts)
self.toolsets_list.addItem(item)
@Slot()
def submit_query(self):
"""Submit query to the agent API."""
query = self.query_input.toPlainText().strip()
if not query:
QMessageBox.warning(self, "No Query", "Please enter a query first!")
return
# Get selected toolsets
selected_toolsets = []
for i in range(self.toolsets_list.count()):
item = self.toolsets_list.item(i)
if item.isSelected():
toolset_name = item.data(Qt.UserRole)
selected_toolsets.append(toolset_name)
# Build request payload
payload = {
"query": query,
"model": self.model_combo.currentText(),
"base_url": self.base_url_input.text(),
"max_turns": self.max_turns_spin.value(),
"enabled_toolsets": selected_toolsets if selected_toolsets else None,
"mock_web_tools": self.mock_mode_checkbox.isChecked(),
"mock_delay": self.mock_delay_spin.value(),
"verbose": self.verbose_checkbox.isChecked()
}
# Disable submit button during execution
self.submit_btn.setEnabled(False)
self.submit_btn.setText("⏳ Running...")
self.statusBar().showMessage("Submitting query to agent...")
# Submit to API
try:
response = requests.post(
f"{self.api_base_url}/agent/run",
json=payload,
timeout=10
)
if response.status_code == 200:
result = response.json()
session_id = result.get("session_id", "")
self.current_session_id = session_id
self.statusBar().showMessage(f"Agent started! Session: {session_id[:8]}...")
# Clear event display for new session
# (or keep history - user preference)
# self.event_widget.clear_events()
else:
QMessageBox.warning(
self,
"API Error",
f"Failed to start agent: {response.status_code}\n{response.text}"
)
self.submit_btn.setEnabled(True)
self.submit_btn.setText("🚀 Submit Query")
except requests.exceptions.RequestException as e:
QMessageBox.critical(
self,
"Connection Error",
f"Failed to connect to API server:\n{str(e)}\n\nMake sure the server is running:\npython logging_server.py"
)
self.submit_btn.setEnabled(True)
self.submit_btn.setText("🚀 Submit Query")
# Re-enable button after short delay (UI feedback)
QTimer.singleShot(2000, lambda: self.submit_btn.setText("🚀 Submit Query"))
def cleanup(self):
"""Clean up resources before exit."""
print("Cleaning up resources...")
self.is_closing = True
if self.ws_client:
try:
self.ws_client.disconnect()
except Exception as e:
print(f"Error disconnecting WebSocket: {e}")
def closeEvent(self, event):
"""Handle window close event - ensures clean shutdown."""
print("Closing application...")
self.cleanup()
event.accept()
from main_window import HermesMainWindow
def setup_signal_handlers(app: QApplication) -> QTimer:
@ -636,7 +50,6 @@ def setup_signal_handlers(app: QApplication) -> QTimer:
print("\n🛑 Interrupt received, shutting down gracefully...")
app.quit()
# Register signal handlers
signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # Termination signal

375
ui/main_window.py Normal file
View file

@ -0,0 +1,375 @@
"""
Main window for Hermes Agent UI.
This module provides the main application window with controls for
submitting queries, configuring settings, and viewing real-time events.
"""
import requests
from typing import Dict, Any
from PySide6.QtWidgets import (
QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QTextEdit,
QPushButton, QLabel, QLineEdit, QComboBox, QCheckBox,
QGroupBox, QSplitter, QListWidget, QListWidgetItem,
QSpinBox, QMessageBox
)
from PySide6.QtCore import Qt, Slot, QTimer
from PySide6.QtGui import QFont
from .websocket_client import WebSocketClient
from .event_widgets import InteractiveEventDisplayWidget
class HermesMainWindow(QMainWindow):
"""
Main window for Hermes Agent UI.
Provides interface for:
- Submitting queries
- Configuring agent settings
- Viewing real-time events
- Managing sessions
"""
def __init__(self):
super().__init__()
self.api_base_url = "http://localhost:8000"
self.ws_client = None
self.current_session_id = None
self.available_toolsets = []
self.is_closing = False # Flag to prevent reconnection during shutdown
self.init_ui()
self.setup_websocket()
self.load_available_tools()
def init_ui(self):
"""Initialize the user interface."""
self.setWindowTitle("Hermes Agent - AI Assistant UI")
self.setGeometry(100, 100, 1400, 900)
# Central widget
central_widget = QWidget()
self.setCentralWidget(central_widget)
# Main layout (horizontal split)
main_layout = QHBoxLayout()
# Left panel: Controls
left_panel = self.create_control_panel()
# Right panel: Event display
right_panel = self.create_event_panel()
# Splitter for resizable panels
splitter = QSplitter(Qt.Horizontal)
splitter.addWidget(left_panel)
splitter.addWidget(right_panel)
splitter.setStretchFactor(0, 1) # Control panel
splitter.setStretchFactor(1, 2) # Event panel (larger)
main_layout.addWidget(splitter)
central_widget.setLayout(main_layout)
# Status bar
self.statusBar().showMessage("Ready")
def create_control_panel(self) -> QWidget:
"""Create the left control panel."""
panel = QWidget()
layout = QVBoxLayout()
# Title
title = QLabel("🤖 Hermes Agent Control")
title.setFont(QFont("Arial", 14, QFont.Bold))
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
# Query input group
query_group = QGroupBox("Query Input")
query_layout = QVBoxLayout()
self.query_input = QTextEdit()
self.query_input.setPlaceholderText("Enter your query here...")
self.query_input.setMaximumHeight(150)
query_layout.addWidget(self.query_input)
self.submit_btn = QPushButton("🚀 Submit Query")
self.submit_btn.setFont(QFont("Arial", 11, QFont.Bold))
self.submit_btn.setStyleSheet("QPushButton { background-color: #4CAF50; color: white; padding: 10px; }")
self.submit_btn.clicked.connect(self.submit_query)
query_layout.addWidget(self.submit_btn)
query_group.setLayout(query_layout)
layout.addWidget(query_group)
# Model configuration group
model_group = QGroupBox("Model Configuration")
model_layout = QVBoxLayout()
# Model selection
model_layout.addWidget(QLabel("Model:"))
self.model_combo = QComboBox()
self.model_combo.addItems([
"claude-sonnet-4-5-20250929",
"claude-opus-4-20250514",
"gpt-4",
"gpt-4-turbo"
])
model_layout.addWidget(self.model_combo)
# API Base URL
model_layout.addWidget(QLabel("API Base URL:"))
self.base_url_input = QLineEdit("https://api.anthropic.com/v1/")
model_layout.addWidget(self.base_url_input)
# Max turns
model_layout.addWidget(QLabel("Max Turns:"))
self.max_turns_spin = QSpinBox()
self.max_turns_spin.setMinimum(1)
self.max_turns_spin.setMaximum(50)
self.max_turns_spin.setValue(10)
model_layout.addWidget(self.max_turns_spin)
model_group.setLayout(model_layout)
layout.addWidget(model_group)
# Tools configuration group
tools_group = QGroupBox("Tools & Toolsets")
tools_layout = QVBoxLayout()
tools_layout.addWidget(QLabel("Select Toolsets:"))
self.toolsets_list = QListWidget()
self.toolsets_list.setSelectionMode(QListWidget.MultiSelection)
self.toolsets_list.setMaximumHeight(150)
tools_layout.addWidget(self.toolsets_list)
tools_group.setLayout(tools_layout)
layout.addWidget(tools_group)
# Options group
options_group = QGroupBox("Options")
options_layout = QVBoxLayout()
self.mock_mode_checkbox = QCheckBox("Mock Web Tools (Testing)")
options_layout.addWidget(self.mock_mode_checkbox)
self.verbose_checkbox = QCheckBox("Verbose Logging")
options_layout.addWidget(self.verbose_checkbox)
options_layout.addWidget(QLabel("Mock Delay (seconds):"))
self.mock_delay_spin = QSpinBox()
self.mock_delay_spin.setMinimum(1)
self.mock_delay_spin.setMaximum(300)
self.mock_delay_spin.setValue(60)
options_layout.addWidget(self.mock_delay_spin)
options_group.setLayout(options_layout)
layout.addWidget(options_group)
# Connection status
self.connection_status = QLabel("🔴 Disconnected")
self.connection_status.setAlignment(Qt.AlignCenter)
self.connection_status.setStyleSheet("QLabel { padding: 5px; background-color: #F44336; color: white; border-radius: 3px; }")
layout.addWidget(self.connection_status)
# Add stretch to push everything to top
layout.addStretch()
panel.setLayout(layout)
return panel
def create_event_panel(self) -> QWidget:
"""Create the right event display panel."""
panel = QWidget()
layout = QVBoxLayout()
# Event display widget
self.event_widget = InteractiveEventDisplayWidget()
layout.addWidget(self.event_widget)
panel.setLayout(layout)
return panel
def setup_websocket(self):
"""Setup WebSocket connection for real-time events."""
self.ws_client = WebSocketClient("ws://localhost:8000/ws")
# Connect signals
self.ws_client.connected.connect(self.on_ws_connected)
self.ws_client.disconnected.connect(self.on_ws_disconnected)
self.ws_client.error.connect(self.on_ws_error)
self.ws_client.event_received.connect(self.on_event_received)
# Start connection
self.ws_client.connect()
@Slot()
def on_ws_connected(self):
"""Called when WebSocket connection is established."""
self.connection_status.setText("🟢 Connected")
self.connection_status.setStyleSheet("QLabel { padding: 5px; background-color: #4CAF50; color: white; border-radius: 3px; }")
self.statusBar().showMessage("WebSocket connected")
@Slot()
def on_ws_disconnected(self):
"""Called when WebSocket connection is lost."""
# Don't attempt reconnection if we're closing the application
if self.is_closing:
return
self.connection_status.setText("🔴 Disconnected")
self.connection_status.setStyleSheet("QLabel { padding: 5px; background-color: #F44336; color: white; border-radius: 3px; }")
self.statusBar().showMessage("WebSocket disconnected - attempting reconnect...")
# Attempt reconnect after 5 seconds
QTimer.singleShot(5000, self.ws_client.connect)
@Slot(str)
def on_ws_error(self, error: str):
"""Called when WebSocket error occurs."""
self.statusBar().showMessage(f"WebSocket error: {error}")
@Slot(dict)
def on_event_received(self, event: Dict[str, Any]):
"""
Called when an event is received from WebSocket.
Args:
event: Event data from server
"""
self.event_widget.add_event(event)
# Update status for specific events
event_type = event.get("event_type")
if event_type == "query":
self.statusBar().showMessage("Query received - agent processing...")
elif event_type == "complete":
self.statusBar().showMessage("Agent completed!")
self.submit_btn.setEnabled(True)
def load_available_tools(self):
"""Load available toolsets from the API."""
try:
response = requests.get(f"{self.api_base_url}/tools", timeout=5)
if response.status_code == 200:
data = response.json()
toolsets = data.get("toolsets", [])
self.available_toolsets = toolsets
self.toolsets_list.clear()
for toolset in toolsets:
name = toolset.get("name", "")
description = toolset.get("description", "")
tool_count = toolset.get("tool_count", 0)
item_text = f"{name} ({tool_count} tools) - {description}"
item = QListWidgetItem(item_text)
item.setData(Qt.UserRole, name) # Store toolset name
self.toolsets_list.addItem(item)
self.statusBar().showMessage(f"Loaded {len(toolsets)} toolsets")
else:
self.statusBar().showMessage("Failed to load toolsets from API")
except requests.exceptions.RequestException as e:
self.statusBar().showMessage(f"Error loading toolsets: {str(e)}")
# Add some default toolsets
default_toolsets = ["web", "vision", "terminal", "research"]
for ts in default_toolsets:
item = QListWidgetItem(f"{ts} (default)")
item.setData(Qt.UserRole, ts)
self.toolsets_list.addItem(item)
@Slot()
def submit_query(self):
"""Submit query to the agent API."""
query = self.query_input.toPlainText().strip()
if not query:
QMessageBox.warning(self, "No Query", "Please enter a query first!")
return
# Get selected toolsets
selected_toolsets = []
for i in range(self.toolsets_list.count()):
item = self.toolsets_list.item(i)
if item.isSelected():
toolset_name = item.data(Qt.UserRole)
selected_toolsets.append(toolset_name)
# Build request payload
payload = {
"query": query,
"model": self.model_combo.currentText(),
"base_url": self.base_url_input.text(),
"max_turns": self.max_turns_spin.value(),
"enabled_toolsets": selected_toolsets if selected_toolsets else None,
"mock_web_tools": self.mock_mode_checkbox.isChecked(),
"mock_delay": self.mock_delay_spin.value(),
"verbose": self.verbose_checkbox.isChecked()
}
# Disable submit button during execution
self.submit_btn.setEnabled(False)
self.submit_btn.setText("⏳ Running...")
self.statusBar().showMessage("Submitting query to agent...")
# Submit to API
try:
response = requests.post(
f"{self.api_base_url}/agent/run",
json=payload,
timeout=10
)
if response.status_code == 200:
result = response.json()
session_id = result.get("session_id", "")
self.current_session_id = session_id
self.statusBar().showMessage(f"Agent started! Session: {session_id[:8]}...")
# Clear event display for new session (optional)
# self.event_widget.clear_events()
else:
QMessageBox.warning(
self,
"API Error",
f"Failed to start agent: {response.status_code}\n{response.text}"
)
self.submit_btn.setEnabled(True)
self.submit_btn.setText("🚀 Submit Query")
except requests.exceptions.RequestException as e:
QMessageBox.critical(
self,
"Connection Error",
f"Failed to connect to API server:\n{str(e)}\n\nMake sure the server is running:\npython logging_server.py"
)
self.submit_btn.setEnabled(True)
self.submit_btn.setText("🚀 Submit Query")
# Re-enable button after short delay (UI feedback)
QTimer.singleShot(2000, lambda: self.submit_btn.setText("🚀 Submit Query"))
def cleanup(self):
"""Clean up resources before exit."""
print("Cleaning up resources...")
self.is_closing = True
if self.ws_client:
try:
self.ws_client.disconnect()
except Exception as e:
print(f"Error disconnecting WebSocket: {e}")
def closeEvent(self, event):
"""Handle window close event - ensures clean shutdown."""
print("Closing application...")
self.cleanup()
event.accept()

91
ui/websocket_client.py Normal file
View file

@ -0,0 +1,91 @@
"""
WebSocket client for real-time event streaming from Hermes Agent.
This module provides a WebSocket client that runs in a separate thread
and emits Qt signals when events are received from the server.
"""
import json
import threading
import websocket
from PySide6.QtCore import QObject, Signal
class WebSocketClient(QObject):
"""
WebSocket client for receiving real-time agent events.
Runs in a separate thread and emits Qt signals when events arrive.
"""
# Signals for event communication
event_received = Signal(dict) # Emits parsed event data
connected = Signal()
disconnected = Signal()
error = Signal(str)
def __init__(self, url: str = "ws://localhost:8000/ws"):
super().__init__()
self.url = url
self.ws = None
self.running = False
self.thread = None
def connect(self):
"""Start WebSocket connection in background thread."""
if self.running:
return
self.running = True
self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start()
def disconnect(self):
"""Stop WebSocket connection."""
self.running = False
if self.ws:
try:
self.ws.close()
except Exception as e:
print(f"Error closing WebSocket: {e}")
def _run(self):
"""WebSocket event loop (runs in background thread)."""
try:
self.ws = websocket.WebSocketApp(
self.url,
on_open=self._on_open,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close
)
# Run forever with reconnection
self.ws.run_forever(ping_interval=300, ping_timeout=60)
except Exception as e:
self.error.emit(f"WebSocket error: {str(e)}")
def _on_open(self, ws):
"""Called when WebSocket connection is established."""
print("WebSocket connected")
self.connected.emit()
def _on_message(self, ws, message):
"""Called when a message is received from the server."""
try:
data = json.loads(message)
self.event_received.emit(data)
except json.JSONDecodeError as e:
print(f" Failed to parse WebSocket message: {e}")
def _on_error(self, ws, error):
"""Called when an error occurs."""
print(f"WebSocket error: {error}")
self.error.emit(str(error))
def _on_close(self, ws, close_status_code, close_msg):
"""Called when WebSocket connection is closed."""
print(f"🔌 WebSocket disconnected: {close_status_code} - {close_msg}")
self.disconnected.emit()