|
|
|
|
|
import gradio as gr |
|
|
|
|
|
class MobileUXHandlers: |
|
|
def __init__(self, orchestrator): |
|
|
self.orchestrator = orchestrator |
|
|
self.mobile_state = {} |
|
|
|
|
|
async def handle_mobile_submit(self, message, chat_history, session_id, |
|
|
show_reasoning, show_agent_trace, request: gr.Request): |
|
|
""" |
|
|
Mobile-optimized submission handler with enhanced UX |
|
|
""" |
|
|
|
|
|
user_agent = request.headers.get("user-agent", "").lower() |
|
|
is_mobile = any(device in user_agent for device in ['mobile', 'android', 'iphone']) |
|
|
|
|
|
|
|
|
if is_mobile: |
|
|
return await self._mobile_optimized_processing( |
|
|
message, chat_history, session_id, show_reasoning, show_agent_trace |
|
|
) |
|
|
else: |
|
|
return await self._desktop_processing( |
|
|
message, chat_history, session_id, show_reasoning, show_agent_trace |
|
|
) |
|
|
|
|
|
async def _mobile_optimized_processing(self, message, chat_history, session_id, |
|
|
show_reasoning, show_agent_trace): |
|
|
""" |
|
|
Mobile-specific processing with enhanced UX feedback |
|
|
""" |
|
|
try: |
|
|
|
|
|
yield { |
|
|
"chatbot": chat_history + [[message, "Thinking..."]], |
|
|
"message_input": "", |
|
|
"reasoning_display": {"status": "processing"}, |
|
|
"performance_display": {"status": "processing"} |
|
|
} |
|
|
|
|
|
|
|
|
result = await self.orchestrator.process_request( |
|
|
session_id=session_id, |
|
|
user_input=message, |
|
|
mobile_optimized=True, |
|
|
max_tokens=800 |
|
|
) |
|
|
|
|
|
|
|
|
formatted_response = self._format_for_mobile( |
|
|
result['final_response'], |
|
|
show_reasoning and result.get('reasoning_chain'), |
|
|
show_agent_trace and result.get('agent_trace') |
|
|
) |
|
|
|
|
|
|
|
|
updated_history = chat_history + [[message, formatted_response]] |
|
|
|
|
|
yield { |
|
|
"chatbot": updated_history, |
|
|
"message_input": "", |
|
|
"reasoning_display": result.get('reasoning_chain', {}), |
|
|
"performance_display": result.get('performance_metrics', {}) |
|
|
} |
|
|
|
|
|
except Exception as e: |
|
|
|
|
|
error_response = self._get_mobile_friendly_error(e) |
|
|
yield { |
|
|
"chatbot": chat_history + [[message, error_response]], |
|
|
"message_input": message, |
|
|
"reasoning_display": {"error": "Processing failed"}, |
|
|
"performance_display": {"error": str(e)} |
|
|
} |
|
|
|
|
|
def _format_for_mobile(self, response, reasoning_chain, agent_trace): |
|
|
""" |
|
|
Format response for optimal mobile readability |
|
|
""" |
|
|
|
|
|
if len(response) > 400: |
|
|
paragraphs = self._split_into_paragraphs(response, max_length=300) |
|
|
response = "\n\n".join(paragraphs) |
|
|
|
|
|
|
|
|
formatted = f""" |
|
|
<div class="mobile-response"> |
|
|
{response} |
|
|
</div> |
|
|
""" |
|
|
|
|
|
|
|
|
if reasoning_chain: |
|
|
formatted += f""" |
|
|
<div class="reasoning-mobile" style="margin-top: 15px; padding: 10px; background: #f5f5f5; border-radius: 8px; font-size: 14px;"> |
|
|
<strong>Reasoning:</strong> {reasoning_chain[:200]}... |
|
|
</div> |
|
|
""" |
|
|
|
|
|
return formatted |
|
|
|
|
|
def _get_mobile_friendly_error(self, error): |
|
|
""" |
|
|
User-friendly error messages for mobile |
|
|
""" |
|
|
error_messages = { |
|
|
"timeout": "⏱️ Taking longer than expected. Please try a simpler question.", |
|
|
"network": "📡 Connection issue. Check your internet and try again.", |
|
|
"rate_limit": "🚦 Too many requests. Please wait a moment.", |
|
|
"default": "❌ Something went wrong. Please try again." |
|
|
} |
|
|
|
|
|
error_type = "default" |
|
|
if "timeout" in str(error).lower(): |
|
|
error_type = "timeout" |
|
|
elif "network" in str(error).lower() or "connection" in str(error).lower(): |
|
|
error_type = "network" |
|
|
elif "rate" in str(error).lower(): |
|
|
error_type = "rate_limit" |
|
|
|
|
|
return error_messages[error_type] |
|
|
|
|
|
async def _desktop_processing(self, message, chat_history, session_id, |
|
|
show_reasoning, show_agent_trace): |
|
|
""" |
|
|
Desktop processing without mobile optimizations |
|
|
""" |
|
|
|
|
|
return { |
|
|
"chatbot": chat_history, |
|
|
"message_input": "", |
|
|
"reasoning_display": {}, |
|
|
"performance_display": {} |
|
|
} |
|
|
|
|
|
def _split_into_paragraphs(self, text, max_length=300): |
|
|
""" |
|
|
Split text into mobile-friendly paragraphs |
|
|
""" |
|
|
|
|
|
words = text.split() |
|
|
paragraphs = [] |
|
|
current_para = [] |
|
|
|
|
|
for word in words: |
|
|
current_para.append(word) |
|
|
if len(' '.join(current_para)) > max_length: |
|
|
paragraphs.append(' '.join(current_para[:-1])) |
|
|
current_para = [current_para[-1]] |
|
|
|
|
|
if current_para: |
|
|
paragraphs.append(' '.join(current_para)) |
|
|
|
|
|
return paragraphs |
|
|
|
|
|
|