#!/usr/bin/env python3
"""
OmniTech Customer Support Chatbot - Gradio Interface
═══════════════════════════════════════════════════════════════════════════════
OVERVIEW
--------
This file creates the web-based user interface for the OmniTech Customer Support
chatbot using Gradio (https://gradio.app). It provides:
1. CHAT TAB - Customer-facing chat interface where users ask questions
2. AGENT DASHBOARD TAB - Shows RAG pipeline metrics and debug info (Developer Mode)
3. MCP MONITOR TAB - Displays MCP server status and tool activity (Developer Mode)
4. KNOWLEDGE SEARCH TAB - Direct search of the knowledge base (Developer Mode)
5. TICKETS TAB - View and filter support tickets (Developer Mode)
KEY CONCEPTS
------------
- Gradio Blocks: The main container for building custom UIs with full control
- Gradio Tabs: Used to organize different views/screens
- gr.HTML: Renders custom HTML for rich displays and styling
- Event Handlers: Connect UI components (buttons, inputs) to Python functions
ARCHITECTURE
------------
┌─────────────────────────────────────────────────────────────┐
│ Gradio Web Interface │
├─────────────────────────────────────────────────────────────┤
│ [Chat] [Agent Dashboard] [MCP Monitor] [Knowledge] [Tickets]
│ │ │
│ AppState │
│ (manages agent) │
│ │ │
│ SyncAgent │
│ (from rag_agent.py) │
│ │ │
│ MCP Server Tools │
│ (from mcp_server.py) │
└─────────────────────────────────────────────────────────────┘
DEVELOPER MODE
--------------
Toggle "Developer Mode" checkbox to reveal debug tabs:
- Agent Dashboard: See RAG metrics, LLM prompts, and responses
- MCP Monitor: View tool calls, timing, and server stats
- Knowledge Search: Query the vector database directly
- Tickets: View support tickets created by the system
This integrates with the RAG agent (rag_agent.py) and MCP server (mcp_server.py).
"""
# ═══════════════════════════════════════════════════════════════════════════════
# IMPORTS
# ═══════════════════════════════════════════════════════════════════════════════
import gradio as gr # Gradio library for building web UIs
import json # JSON parsing for debug displays
from datetime import datetime # Timestamps for chat messages
from typing import Dict, List, Any # Type hints for better code clarity
# ─────────────────────────────────────────────────────────────────────────────
# RAG Agent Import
# Try to import the RAG agent - if it fails, the UI runs in "demo mode"
# This allows testing the UI without the full backend running
# ─────────────────────────────────────────────────────────────────────────────
try:
from rag_agent import SyncAgent
AGENT_AVAILABLE = True
except ImportError:
AGENT_AVAILABLE = False
print("Warning: rag_agent not found. Running in demo mode.")
# ╔══════════════════════════════════════════════════════════════════════════╗
# ║ SECTION 1: APPLICATION STATE ║
# ║ ║
# ║ Purpose: Centralized state management for the entire application ║
# ║ ║
# ║ The AppState class holds: ║
# ║ - The RAG agent instance (initialized lazily on first query) ║
# ║ - Conversation history for display in the chat ║
# ║ - Session metrics (queries, resolutions, tickets) ║
# ║ - Debug information from the last query (for Agent Dashboard) ║
# ║ ║
# ║ WHY A CLASS? Using a class instead of global variables provides: ║
# ║ 1. Encapsulation - all state in one place ║
# ║ 2. Clear initialization - __init__ sets up defaults ║
# ║ 3. Methods - related functionality grouped together ║
# ╚══════════════════════════════════════════════════════════════════════════╝
class AppState:
"""Global application state."""
def __init__(self):
self.agent = None
self.conversation_history: List[Dict] = []
self.metrics = {
'total_queries': 0,
'resolved_queries': 0,
'tickets_created': 0
}
# Store last debug info for Agent Dashboard
self.last_prompt = ""
self.last_response = ""
def initialize_agent(self) -> bool:
"""Initialize the MCP agent."""
if not AGENT_AVAILABLE:
return False
if self.agent is None:
try:
self.agent = SyncAgent()
print("Agent initialized successfully")
return True
except Exception as e:
print(f"Failed to initialize agent: {e}")
return False
return True
def process_query(self, query: str, email: str) -> Dict[str, Any]:
"""Process a query through the agent."""
if self.agent:
return self.agent.process_query(query, email)
else:
# Demo mode response
return {
"response": f"[Demo Mode] Received: {query}",
"workflow": "demo",
"confidence": 0.5
}
def get_mcp_stats(self) -> Dict[str, Any]:
"""Get MCP server statistics."""
if self.agent:
return self.agent.get_server_stats()
return {"status": "Demo mode - no server"}
def search_knowledge(self, query: str, max_results: int = 3) -> List[Dict]:
"""Search knowledge base directly via MCP."""
if self.agent:
try:
# Call the search_knowledge tool via the agent's internal method
# SyncAgent.agent = OmniTechAgent, SyncAgent.loop = event loop
result = self.agent.loop.run_until_complete(
self.agent.agent.call_tool("search_knowledge", {
"query": query,
"max_results": max_results
})
)
return result.get("matches", [])
except Exception as e:
print(f"Knowledge search error: {e}")
return []
def get_tickets(self, customer_email: str = None, status: str = None) -> List[Dict]:
"""Get tickets via MCP tool."""
if self.agent:
try:
args = {"limit": 50}
if customer_email:
args["customer_email"] = customer_email
if status:
args["status"] = status
result = self.agent.loop.run_until_complete(
self.agent.agent.call_tool("get_tickets", args)
)
return result.get("tickets", [])
except Exception as e:
print(f"Get tickets error: {e}")
return []
def get_security_log(self) -> List[Dict]:
"""Get security event log from agent."""
if self.agent:
return self.agent.get_security_log()
return []
def clear_security_log(self):
"""Clear the security log."""
if self.agent:
self.agent.clear_security_log()
app_state = AppState()
# ╔══════════════════════════════════════════════════════════════════════════╗
# ║ SECTION 2: CUSTOM CSS STYLES ║
# ║ ║
# ║ Purpose: Define custom styling for the Gradio interface ║
# ║ ║
# ║ Gradio allows injecting CSS to customize the look and feel. We inject ║
# ║ this CSS via gr.HTML() since Gradio 6.0 doesn't support the css= ║
# ║ parameter on gr.Blocks(). ║
# ║ ║
# ║ KEY CSS CLASSES: ║
# ║ .metric-card - Styled cards for displaying metrics/info ║
# ║ .chat-message-* - Styling for chat bubbles (user vs agent) ║
# ║ .tool-card - Styling for MCP tool displays ║
# ║ .nav-button - Navigation button styling ║
# ╚══════════════════════════════════════════════════════════════════════════╝
CUSTOM_CSS = """
@import url('https://fonts.googleapis.com/css2?family=Inter:wght@300;400;500;600;700&display=swap');
.gradio-container {
font-family: 'Inter', -apple-system, BlinkMacSystemFont, sans-serif !important;
}
/* Debug toggle styling */
.debug-toggle {
margin-top: 1rem;
padding: 0.5rem 1rem;
background: #f1f5f9;
border-radius: 8px;
font-size: 0.875rem;
}
.debug-toggle label {
cursor: pointer;
}
/* Typing indicator animation */
@keyframes typing-dot {
0%, 60%, 100% { opacity: 0.3; }
30% { opacity: 1; }
}
.typing-indicator {
display: inline-flex;
gap: 4px;
padding: 0.75rem 1rem;
background: #f1f5f9;
border-radius: 12px;
margin: 0.5rem 0;
}
.typing-indicator span {
width: 8px;
height: 8px;
background: #64748b;
border-radius: 50%;
animation: typing-dot 1.4s infinite;
}
.typing-indicator span:nth-child(2) { animation-delay: 0.2s; }
.typing-indicator span:nth-child(3) { animation-delay: 0.4s; }
.nav-button {
background: #ffffff;
border: 1.5px solid #e2e8f0 !important;
border-radius: 10px;
padding: 0.875rem 1.25rem;
margin: 0.375rem 0;
transition: all 0.2s ease;
font-weight: 500;
color: #475569;
}
.nav-button:hover {
background: #f8fafc;
border-color: #cbd5e1 !important;
transform: translateX(4px);
}
.metric-card {
background: #ffffff;
border: 1px solid #e2e8f0;
border-radius: 10px;
padding: 1.25rem;
margin: 0.5rem 0;
transition: all 0.2s ease;
}
.metric-card:hover {
border-color: #cbd5e1;
box-shadow: 0 4px 8px -2px rgba(0, 0, 0, 0.08);
}
.chat-message-user {
background: #ffffff;
border: 1px solid #e2e8f0;
border-left: 3px solid #3b82f6;
border-radius: 10px;
padding: 1rem;
margin: 0.5rem 0;
}
.chat-message-agent {
background: #f8fafc;
border: 1px solid #e2e8f0;
border-left: 3px solid #10b981;
border-radius: 10px;
padding: 1rem;
margin: 0.5rem 0;
}
.tool-card {
background: #f8fafc;
border: 1px solid #e2e8f0;
border-radius: 8px;
padding: 0.75rem 1rem;
margin: 0.375rem 0;
font-family: 'SF Mono', Monaco, monospace;
font-size: 0.875rem;
}
"""
# ╔══════════════════════════════════════════════════════════════════════════╗
# ║ SECTION 3: HELPER FUNCTIONS ║
# ║ ║
# ║ Purpose: Utility functions for formatting and generating HTML content ║
# ║ ║
# ║ FUNCTIONS IN THIS SECTION: ║
# ║ format_message() - Format chat messages as styled HTML ║
# ║ process_query_handler() - Main handler for user query submission ║
# ║ generate_agent_dashboard() - Generate Agent Dashboard HTML ║
# ║ generate_mcp_monitor() - Generate MCP Monitor HTML ║
# ║ generate_tickets_display() - Generate Tickets tab HTML ║
# ║ clear_chat() - Reset conversation and metrics ║
# ║ get_status() - Get system status string ║
# ║ search_knowledge_direct() - Search knowledge base directly ║
# ╚══════════════════════════════════════════════════════════════════════════╝
def format_message(sender: str, content: str, timestamp: str) -> str:
"""Format a chat message as HTML."""
msg_class = "chat-message-agent" if sender == "agent" else "chat-message-user"
sender_name = "AI Agent" if sender == "agent" else "You"
return f"""
{sender_name}
{timestamp}
{content}
"""
def process_query_handler(query: str, customer_email: str, history: str):
"""Process a query and update chat history."""
if not query.strip():
return history, "", "", ""
timestamp = datetime.now().strftime("%H:%M:%S")
history += format_message("customer", query, timestamp)
# Initialize agent if needed
if not app_state.agent:
app_state.initialize_agent()
# Process query
result = app_state.process_query(query, customer_email)
response = result.get("response", "Error processing query")
# Add to history
app_state.conversation_history.append({
'sender': 'customer',
'content': query,
'timestamp': timestamp
})
app_state.conversation_history.append({
'sender': 'agent',
'content': response,
'timestamp': datetime.now().strftime("%H:%M:%S"),
'metadata': result
})
# Update metrics
app_state.metrics['total_queries'] += 1
if result.get('confidence', 0) > 0.7:
app_state.metrics['resolved_queries'] += 1
if result.get('ticket_created'):
app_state.metrics['tickets_created'] += 1
history += format_message("agent", response, datetime.now().strftime("%H:%M:%S"))
# Get prompt and response for debug display
prompt = result.get('llm_prompt', 'No prompt available')
response_json = json.dumps({k: v for k, v in result.items() if k != 'llm_prompt'}, indent=2)
# Store in app_state for Agent Dashboard
app_state.last_prompt = prompt
app_state.last_response = response_json
return history, "", prompt, response_json
def generate_agent_dashboard() -> str:
"""Generate the agent dashboard HTML."""
metrics = app_state.metrics
total = metrics['total_queries']
resolved = metrics['resolved_queries']
tickets = metrics['tickets_created']
rate = f"{(resolved/max(total,1)*100):.1f}%"
html = """
Agent Performance Dashboard
Real-time analytics and RAG pipeline metrics
"""
# Metrics row
html += f"""
TICKETS CREATED
{tickets}
"""
# Recent queries with RAG details
agent_msgs = [m for m in app_state.conversation_history if m['sender'] == 'agent']
if agent_msgs:
html += "Recent RAG Operations
"
for i, msg in enumerate(agent_msgs[-5:], 1):
meta = msg.get('metadata', {})
workflow = meta.get('workflow', 'unknown')
category = meta.get('classification', {}).get('category', 'N/A')
sources = meta.get('sources', [])
confidence = meta.get('confidence', 0)
html += f"""
Query {i} - {msg.get('timestamp', 'N/A')}
Workflow: {workflow}
Category: {category}
Confidence: {confidence:.2%}
Sources: {', '.join(sources) if sources else 'None'}
"""
else:
html += "No queries processed yet. Start chatting to see RAG analytics.
"
return html
def generate_mcp_monitor() -> str:
"""Generate the MCP monitor HTML."""
stats = app_state.get_mcp_stats()
html = """
MCP Protocol Monitor
Server status and tool activity
"""
if isinstance(stats, dict) and 'error' not in stats:
# Server metrics
html += f"""
TOTAL REQUESTS
{stats.get('total_requests', 0)}
KNOWLEDGE DOCS
{stats.get('knowledge_documents', 0)}
CUSTOMERS
{stats.get('customers_in_db', 0)}
"""
# Recent MCP calls (shown first)
if app_state.agent:
mcp_log = app_state.agent.get_mcp_log()
if mcp_log:
html += "Recent MCP Calls
"
for entry in mcp_log[-10:]:
html += f"""
{entry['tool']}
{entry['duration_ms']}ms
{entry['timestamp']} - {'✓' if entry['success'] else '✗'}
"""
# Available tools
tools = stats.get('tools_available', [])
if tools:
html += "Available MCP Tools
"
tool_descriptions = {
'classify_query': 'Classify customer queries into support categories',
'get_query_template': 'Get prompt templates for categories',
'list_categories': 'List all support categories',
'search_knowledge': 'Search the knowledge base',
'get_knowledge_for_query': 'Get knowledge for a category',
'lookup_customer': 'Look up customer information',
'create_support_ticket': 'Create support tickets',
'get_server_stats': 'Get server statistics'
}
for tool in tools:
desc = tool_descriptions.get(tool, 'MCP tool')
html += f"""
"""
html += "
Server statistics not available. Initialize the agent first.
"
return html
def generate_tickets_display(customer_filter: str = "", status_filter: str = "") -> str:
"""Generate HTML display for tickets."""
# Ensure agent is initialized
if not app_state.agent:
app_state.initialize_agent()
# Apply filters
customer = customer_filter if customer_filter and customer_filter != "All" else None
status = status_filter if status_filter and status_filter != "All" else None
tickets = app_state.get_tickets(customer_email=customer, status=status)
html = """
Support Tickets
View and track customer support tickets
"""
if not app_state.agent:
html += """
Agent not initialized
Send a chat message first to initialize the system.
"""
return html
if not tickets:
html += """
No tickets found
Tickets will appear here when customers submit support requests.
"""
return html
# Summary stats
open_count = len([t for t in tickets if t.get("status") == "Open"])
closed_count = len(tickets) - open_count
html += f"""
TOTAL TICKETS
{len(tickets)}
"""
# Tickets list
html += ""
for ticket in tickets:
status = ticket.get("status", "Unknown")
priority = ticket.get("priority", "medium")
# Status badge color
status_color = "#f59e0b" if status == "Open" else "#10b981"
status_bg = "#fef3c7" if status == "Open" else "#d1fae5"
# Priority badge
priority_colors = {
"high": ("#dc2626", "#fee2e2"),
"medium": ("#f59e0b", "#fef3c7"),
"low": ("#3b82f6", "#dbeafe")
}
pri_color, pri_bg = priority_colors.get(priority, ("#64748b", "#f1f5f9"))
html += f"""
{ticket.get('id', 'N/A')}
{status}
{priority}
{ticket.get('created_at', 'N/A')[:16]}
Customer: {ticket.get('customer_email', 'N/A')}
Issue: {ticket.get('issue_type', 'N/A')}
{ticket.get('description', 'No description')[:200]}{'...' if len(ticket.get('description', '')) > 200 else ''}
"""
html += "
Security Monitor
Track potential prompt injection and goal-hijacking attempts
"""
if not app_state.agent:
html += """
Agent not initialized
Send a chat message first to initialize the system.
"""
return html
# Summary stats
high_count = len([e for e in events if e.get("severity") == "high"])
medium_count = len([e for e in events if e.get("severity") == "medium"])
low_count = len([e for e in events if e.get("severity") == "low"])
html += f"""
TOTAL EVENTS
{len(events)}
HIGH SEVERITY
{high_count}
MEDIUM SEVERITY
{medium_count}
"""
# Detection patterns info
html += """
Monitored Patterns
The agent monitors for common prompt injection patterns including:
ignore instructions, role changes,
fake system prompts, reveal prompt attempts, and more.
"""
if not events:
html += """
✓ No suspicious activity detected
All queries have passed security inspection.
"""
return html
# Events list (most recent first)
html += "Security Events
"
html += ""
for event in reversed(events[-20:]): # Show last 20, most recent first
severity = event.get("severity", "low")
event_type = event.get("event_type", "unknown")
details = event.get("details", "No details")
query = event.get("query", "")
timestamp = event.get("timestamp", "")[:19] # Trim microseconds
customer = event.get("customer_email", "N/A")
# Severity badge colors
severity_colors = {
"high": ("#dc2626", "#fee2e2"),
"medium": ("#f59e0b", "#fef3c7"),
"low": ("#3b82f6", "#dbeafe")
}
sev_color, sev_bg = severity_colors.get(severity, ("#64748b", "#f1f5f9"))
html += f"""
{event_type}
{severity}
{timestamp}
Details: {details}
Customer: {customer}
{f'
{query[:150]}{"..." if len(query) > 150 else ""}
' if query else ''}
"""
html += "
How can we help you today?
Ask about orders, products, accounts, or technical support.
"""
return initial_html, "", ""
def get_status() -> str:
"""Get system status."""
if app_state.agent:
tools = app_state.agent.get_available_tools()
return f"**System Online**\n\nMCP Tools: {len(tools)}\n\nReady to assist"
else:
return "**Initializing...**\n\nClick 'Send' to start"
def search_knowledge_direct(search_query: str, max_results: int) -> str:
"""Direct knowledge base search and format results as HTML."""
if not search_query.strip():
return """
Enter a search query to search the knowledge base.
"""
results = app_state.search_knowledge(search_query, int(max_results))
if not results:
return """
No results found. Try different search terms.
"""
html = f"""
Search Results for: "{search_query}"
Found {len(results)} documents
"""
for i, doc in enumerate(results, 1):
similarity = doc.get('similarity', 0)
category = doc.get('category', 'unknown')
content = doc.get('content', '')[:500]
source = doc.get('source', 'Unknown')
# Visual similarity bar
bar_filled = int(similarity * 10) if similarity > 0 else 0
similarity_bar = "█" * bar_filled + "░" * (10 - bar_filled)
html += f"""
Result {i}: {category.replace('_', ' ').title()}
Source: {source}
Similarity: {similarity:.3f} {similarity_bar}
"""
return html
# ╔══════════════════════════════════════════════════════════════════════════╗
# ║ SECTION 4: GRADIO INTERFACE DEFINITION ║
# ║ ║
# ║ Purpose: Define the complete Gradio UI layout and event handlers ║
# ║ ║
# ║ STRUCTURE: ║
# ║ 1. gr.Blocks() - Main container with title ║
# ║ 2. Header Row - Title banner + Developer Mode toggle ║
# ║ 3. gr.Tabs() containing all tabs: ║
# ║ - Chat Tab (always visible) - Main customer interface ║
# ║ - Agent Dashboard Tab (Developer Mode) - RAG metrics & debug ║
# ║ - MCP Monitor Tab (Developer Mode) - Server stats & tools ║
# ║ - Knowledge Search Tab (Developer Mode) - Direct KB search ║
# ║ - Tickets Tab (Developer Mode) - View support tickets ║
# ║ 4. Footer - Branding and copyright ║
# ║ 5. Event Handlers - Connect UI components to Python functions ║
# ║ ║
# ║ KEY GRADIO CONCEPTS: ║
# ║ - gr.Blocks: Container for custom layouts (vs gr.Interface) ║
# ║ - gr.Row/Column: Layout containers for organizing components ║
# ║ - gr.Tab: Tab panels within gr.Tabs ║
# ║ - gr.HTML: Render custom HTML content ║
# ║ - gr.Button: Clickable buttons with variants (primary/secondary) ║
# ║ - gr.Textbox: Text input fields ║
# ║ - gr.Dropdown: Selection dropdowns ║
# ║ - gr.Checkbox: Toggle switches ║
# ║ - .click()/.change()/.submit(): Event handler decorators ║
# ║ - gr.update(): Return value to update component properties ║
# ╚══════════════════════════════════════════════════════════════════════════╝
with gr.Blocks(title="OmniTech Support") as demo:
# Inject custom CSS
gr.HTML(f"")
# Header with debug toggle
with gr.Row():
with gr.Column(scale=20):
gr.HTML("""
OmniTech Customer Support
AI-Powered Support Assistant
""")
with gr.Column(scale=1, min_width=180):
debug_mode = gr.Checkbox(
label="Developer Mode",
value=False
)
# Status display (only visible in debug mode)
status_display = gr.Markdown(get_status(), visible=False)
# Use Tabs for navigation (Gradio 6.0 compatible)
with gr.Tabs() as tabs:
# Customer Chat Tab
with gr.Tab("Chat", id="chat_tab"):
with gr.Row():
with gr.Column(scale=4):
customer_email = gr.Dropdown(
label="Customer",
choices=["john.doe@email.com", "sarah.smith@email.com", "mike.johnson@email.com", "guest@example.com"],
value="john.doe@email.com",
container=True
)
with gr.Column(scale=1):
clear_btn = gr.Button("Clear Chat", size="sm")
chat_display = gr.HTML(
value="""
How can we help you today?
Ask about orders, products, accounts, or technical support.
"""
)
query_input = gr.Textbox(
label="Message",
placeholder="Type your question here...",
lines=2,
show_label=False
)
with gr.Row():
send_btn = gr.Button("Send Message", variant="primary", scale=3)
gr.Markdown("**Try asking about:**", elem_classes=["quick-actions-label"])
with gr.Row():
q1 = gr.Button("🔑 Password Reset", size="sm", variant="secondary")
q2 = gr.Button("🔧 Device Issue", size="sm", variant="secondary")
q3 = gr.Button("📦 Return Policy", size="sm", variant="secondary")
q4 = gr.Button("🚚 Track Order", size="sm", variant="secondary")
# Agent Dashboard Tab (hidden by default)
with gr.Tab("Agent Dashboard", id="agent_tab", visible=False) as agent_tab:
agent_dashboard = gr.HTML(generate_agent_dashboard())
refresh_agent_btn = gr.Button("Refresh Dashboard")
# Debug Info Section - LLM Prompt and Full Response
gr.Markdown("---")
gr.Markdown("### Last Query Debug Info")
with gr.Row():
with gr.Column():
prompt_display = gr.Textbox(label="LLM Prompt", lines=10, interactive=False)
with gr.Column():
response_display = gr.Textbox(label="Full Response", lines=10, interactive=False)
# MCP Monitor Tab (hidden by default)
with gr.Tab("MCP Monitor", id="mcp_tab", visible=False) as mcp_tab:
mcp_monitor = gr.HTML(generate_mcp_monitor())
refresh_mcp_btn = gr.Button("Refresh Monitor")
# Knowledge Search Tab (hidden by default)
with gr.Tab("Knowledge Search", visible=False) as kb_tab:
gr.Markdown("## Knowledge Base Search")
gr.Markdown("Search the OmniTech product documentation directly.")
with gr.Row():
with gr.Column(scale=3):
search_input = gr.Textbox(
label="Search Query",
placeholder="Enter search terms (e.g., 'password reset', 'warranty', 'smart home')",
lines=1
)
with gr.Column(scale=1):
search_results_slider = gr.Slider(
label="Max Results",
minimum=1,
maximum=10,
value=3,
step=1
)
search_btn = gr.Button("Search Knowledge Base", variant="primary")
knowledge_results = gr.HTML(
value="""
Enter a search query to explore the OmniTech knowledge base.
Try queries like: "password reset", "device troubleshooting", "return policy", "warranty information"
"""
)
gr.Markdown("---")
gr.Markdown("### Knowledge Base Categories")
gr.HTML("""
Account Security
Password reset, 2FA, account recovery
Device Support
Troubleshooting, setup, compatibility
Shipping & Returns
Order tracking, return policy, refunds
""")
# Tickets Tab (hidden by default)
with gr.Tab("Tickets", id="tickets_tab", visible=False) as tickets_tab:
with gr.Row():
with gr.Column(scale=2):
ticket_customer_filter = gr.Dropdown(
label="Filter by Customer",
choices=["All", "john.doe@email.com", "sarah.smith@email.com", "mike.johnson@email.com"],
value="All"
)
with gr.Column(scale=2):
ticket_status_filter = gr.Dropdown(
label="Filter by Status",
choices=["All", "Open", "Closed"],
value="All"
)
with gr.Column(scale=1):
refresh_tickets_btn = gr.Button("Refresh", variant="secondary")
tickets_display = gr.HTML(generate_tickets_display())
# Security Log Tab (hidden by default)
with gr.Tab("Security", id="security_tab", visible=False) as security_tab:
security_display = gr.HTML(generate_security_log_display())
with gr.Row():
refresh_security_btn = gr.Button("Refresh", variant="secondary")
clear_security_btn = gr.Button("Clear Log", variant="secondary")
# Footer
gr.HTML("""
OmniTech Customer Support • Enterprise AI Accelerator Capstone
""")
# Debug mode toggle handler
def toggle_debug_mode(enabled):
"""Toggle visibility of debug elements."""
return (
gr.update(visible=enabled), # agent_tab
gr.update(visible=enabled), # mcp_tab
gr.update(visible=enabled), # kb_tab
gr.update(visible=enabled), # tickets_tab
gr.update(visible=enabled), # security_tab
gr.update(visible=enabled), # status_display
gr.update(selected="chat_tab"), # tabs - always select chat tab on toggle
)
debug_mode.change(
toggle_debug_mode,
inputs=[debug_mode],
outputs=[agent_tab, mcp_tab, kb_tab, tickets_tab, security_tab, status_display, tabs]
)
# Chat handlers
send_btn.click(
process_query_handler,
inputs=[query_input, customer_email, chat_display],
outputs=[chat_display, query_input, prompt_display, response_display]
)
query_input.submit(
process_query_handler,
inputs=[query_input, customer_email, chat_display],
outputs=[chat_display, query_input, prompt_display, response_display]
)
clear_btn.click(clear_chat, outputs=[chat_display, prompt_display, response_display])
# Quick action handlers
q1.click(lambda: "How do I reset my password?", outputs=query_input)
q2.click(lambda: "My device won't turn on", outputs=query_input)
q3.click(lambda: "What is your return policy?", outputs=query_input)
q4.click(lambda: "How can I track my order?", outputs=query_input)
# Refresh handlers
refresh_agent_btn.click(lambda: generate_agent_dashboard(), outputs=agent_dashboard)
refresh_mcp_btn.click(lambda: generate_mcp_monitor(), outputs=mcp_monitor)
# Auto-refresh dashboards when switching tabs
def refresh_agent_tab():
"""Refresh agent dashboard and debug info."""
return generate_agent_dashboard(), app_state.last_prompt, app_state.last_response
agent_tab.select(refresh_agent_tab, outputs=[agent_dashboard, prompt_display, response_display])
mcp_tab.select(lambda: generate_mcp_monitor(), outputs=mcp_monitor)
# Knowledge search handler
search_btn.click(
search_knowledge_direct,
inputs=[search_input, search_results_slider],
outputs=knowledge_results
)
search_input.submit(
search_knowledge_direct,
inputs=[search_input, search_results_slider],
outputs=knowledge_results
)
# Tickets handlers
def refresh_tickets(customer, status):
return generate_tickets_display(customer, status)
refresh_tickets_btn.click(
refresh_tickets,
inputs=[ticket_customer_filter, ticket_status_filter],
outputs=tickets_display
)
ticket_customer_filter.change(
refresh_tickets,
inputs=[ticket_customer_filter, ticket_status_filter],
outputs=tickets_display
)
ticket_status_filter.change(
refresh_tickets,
inputs=[ticket_customer_filter, ticket_status_filter],
outputs=tickets_display
)
def refresh_tickets_on_select():
"""Refresh tickets and status when switching to tickets tab."""
return generate_tickets_display(), get_status()
tickets_tab.select(
refresh_tickets_on_select,
outputs=[tickets_display, status_display]
)
# Security log handlers
def refresh_security():
return generate_security_log_display()
def clear_security():
app_state.clear_security_log()
return generate_security_log_display()
refresh_security_btn.click(refresh_security, outputs=security_display)
clear_security_btn.click(clear_security, outputs=security_display)
security_tab.select(refresh_security, outputs=security_display)
# Initialize on load
demo.load(lambda: (app_state.initialize_agent(), get_status())[1], outputs=status_display)
# ╔══════════════════════════════════════════════════════════════════════════╗
# ║ 5. Main Entry Point ║
# ╚══════════════════════════════════════════════════════════════════════════╝
if __name__ == "__main__":
print("=" * 60)
print("OmniTech Customer Support Chatbot")
print("=" * 60)
print(f"Agent available: {AGENT_AVAILABLE}")
print("Starting Gradio interface...")
print("=" * 60)
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=True,
footer_links=[
{"text": "© 2026 Tech Skills Transformations", "url": "https://getskillsnow.com"}
]
)