File size: 5,332 Bytes
8f4d405 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
"""
Event handlers for connecting UI to backend
"""
import logging
import uuid
from typing import Dict, Any
logger = logging.getLogger(__name__)
class EventHandlers:
def __init__(self, components: Dict[str, Any]):
self.components = components
self.sessions = {} # In-memory session storage
async def handle_message_submit(self, message: str, chat_history: list,
session_id: str, show_reasoning: bool,
show_agent_trace: bool, request):
"""Handle user message submission"""
try:
# Ensure session exists
if session_id not in self.sessions:
self.sessions[session_id] = {
'history': [],
'context': {},
'created_at': uuid.uuid4().hex
}
# Add user message to history
chat_history.append((message, None)) # None for pending response
# Generate response based on available components
if self.components.get('mock_mode'):
response = self._generate_mock_response(message)
else:
response = await self._generate_ai_response(message, session_id)
# Update chat history with response
chat_history[-1] = (message, response)
# Prepare additional data for UI
reasoning_data = {}
performance_data = {}
if show_reasoning:
reasoning_data = {
"chain_of_thought": {
"step_1": {
"hypothesis": "Mock reasoning for demonstration",
"evidence": ["Mock mode active", f"User input: {message[:50]}..."],
"confidence": 0.5,
"reasoning": "Demonstration mode - enhanced reasoning chain not available"
}
},
"alternative_paths": [],
"uncertainty_areas": [
{
"aspect": "System mode",
"confidence": 0.5,
"mitigation": "Mock mode - full reasoning chain not available"
}
],
"evidence_sources": [],
"confidence_calibration": {"overall_confidence": 0.5, "mock_mode": True}
}
if show_agent_trace:
performance_data = {"agents_used": ["intent", "synthesis", "safety"]}
return "", chat_history, reasoning_data, performance_data
except Exception as e:
logger.error(f"Error handling message: {e}")
error_response = "I apologize, but I'm experiencing technical difficulties. Please try again."
chat_history.append((message, error_response))
return "", chat_history, {"error": str(e)}, {"status": "error"}
def _generate_mock_response(self, message: str) -> str:
"""Generate mock response for demonstration"""
mock_responses = [
f"I understand you're asking about: {message}. This is a mock response while the AI system initializes.",
f"Thank you for your question: '{message}'. The research assistant is currently in demonstration mode.",
f"Interesting question about {message}. In a full implementation, I would analyze this using multiple AI agents.",
f"I've received your query: '{message}'. The system is working properly in mock mode."
]
import random
return random.choice(mock_responses)
async def _generate_ai_response(self, message: str, session_id: str) -> str:
"""Generate AI response using orchestrator"""
try:
if 'orchestrator' in self.components:
result = await self.components['orchestrator'].process_request(
session_id=session_id,
user_input=message
)
return result.get('final_response', 'No response generated')
else:
return "Orchestrator not available. Using mock response."
except Exception as e:
logger.error(f"AI response generation failed: {e}")
return f"AI processing error: {str(e)}"
def handle_new_session(self):
"""Handle new session creation"""
new_session_id = uuid.uuid4().hex[:8] # Short session ID for display
self.sessions[new_session_id] = {
'history': [],
'context': {},
'created_at': uuid.uuid4().hex
}
return new_session_id, [] # New session ID and empty history
def handle_settings_toggle(self, current_visibility: bool):
"""Toggle settings panel visibility"""
return not current_visibility
def handle_tab_change(self, tab_name: str):
"""Handle tab changes in mobile interface"""
return tab_name, False # Return tab name and hide mobile nav
# Factory function
def create_event_handlers(components: Dict[str, Any]):
return EventHandlers(components)
|