# chatbot_demo.py import gradio as gr import logging from datetime import datetime from typing import List, Tuple, Optional import os import socket from kallam.app.chatbot_manager import ChatbotManager mgr = ChatbotManager(log_level="INFO") logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # INLINE SVG for icons CABBAGE_SVG = """ """ # ----------------------- # Core handlers # ----------------------- def _session_status(session_id: str) -> str: """Get current session status using mgr.get_session()""" if not session_id: return "🔴 **No Active Session** - Click **New Session** to start" try: # Use same method as simple app now = datetime.now() s = mgr.get_session(session_id) or {} ts = s.get("timestamp", now.strftime("%d %b %Y | %I:%M %p")) model = s.get("model_used", "Orchestrated SEA-Lion") total = s.get("total_messages", 0) saved_memories = s.get("saved_memories") or "General consultation" return f""" 🟢 **Session:** `{session_id[:8]}...` 🏥 **Profile:** {saved_memories[:50]}{"..." if len(saved_memories) > 50 else ""} 📅 **Created:** {ts} 💬 **Messages:** {total} 🤖 **Model:** {model} """.strip() except Exception as e: logger.error(f"Error getting session status: {e}") return f"❌ **Error loading session:** {session_id[:8]}..." def start_new_session(health_profile: str = ""): """Create new session using mgr - same as simple app""" try: sid = mgr.start_session(saved_memories=health_profile.strip() or None) status = _session_status(sid) # Initial welcome message welcome_msg = { "role": "assistant", "content": """Hello! I'm KaLLaM 🌿, your caring AI health advisor 💖 I can communicate in both **Thai** and **English**. I'm here to support your health and well-being with personalized advice. How are you feeling today? 😊 สวัสดีค่ะ! ฉันชื่อกะหล่ำ 🌿 เป็นที่ปรึกษาด้านสุขภาพ AI ที่จะคอยดูแลคุณ 💖 ฉันสามารถสื่อสารได้ทั้งภาษาไทยและภาษาอังกฤษ วันนี้รู้สึกยังไงบ้างคะ? 😊""" } history = [welcome_msg] result_msg = f"✅ **New Session Created Successfully!**\n\n🆔 Session ID: `{sid}`" if health_profile.strip(): result_msg += f"\n🏥 **Health Profile:** Applied successfully" return sid, history, "", status, result_msg except Exception as e: logger.error(f"Error creating new session: {e}") return "", [], "", "❌ **Failed to create session**", f"❌ **Error:** {e}" def send_message(user_msg: str, history: list, session_id: str): """Send message using mgr - same as simple app""" # Defensive: auto-create session if missing (same as simple app) if not session_id: logger.warning("No session found, auto-creating...") sid, history, _, status, _ = start_new_session("") history.append({"role": "assistant", "content": "🔄 **New session created automatically.** You can now continue chatting!"}) return history, "", sid, status if not user_msg.strip(): return history, "", session_id, _session_status(session_id) try: # Add user message history = history + [{"role": "user", "content": user_msg}] # Get bot response using mgr (same as simple app) bot_response = mgr.handle_message( session_id=session_id, user_message=user_msg ) # Add bot response history = history + [{"role": "assistant", "content": bot_response}] return history, "", session_id, _session_status(session_id) except Exception as e: logger.error(f"Error processing message: {e}") error_msg = {"role": "assistant", "content": f"❌ **Error:** Unable to process your message. Please try again.\n\nDetails: {e}"} history = history + [error_msg] return history, "", session_id, _session_status(session_id) def update_health_profile(session_id: str, health_profile: str): """Update health profile for current session using mgr's database access""" if not session_id: return "❌ **No active session**", _session_status(session_id) if not health_profile.strip(): return "❌ **Please provide health information**", _session_status(session_id) try: # Use mgr's database path (same pattern as simple app would use) from kallam.infra.db import sqlite_conn with sqlite_conn(str(mgr.db_path)) as conn: conn.execute( "UPDATE sessions SET saved_memories = ?, last_activity = ? WHERE session_id = ?", (health_profile.strip(), datetime.now().isoformat(), session_id), ) result = f"✅ **Health Profile Updated Successfully!**\n\n📝 **Updated Information:** {health_profile.strip()[:100]}{'...' if len(health_profile.strip()) > 100 else ''}" return result, _session_status(session_id) except Exception as e: logger.error(f"Error updating health profile: {e}") return f"❌ **Error updating profile:** {e}", _session_status(session_id) def clear_session(session_id: str): """Clear current session using mgr""" if not session_id: return "", [], "", "🔴 **No active session to clear**", "❌ **No active session**" try: # Check if mgr has delete_session method, otherwise handle gracefully if hasattr(mgr, 'delete_session'): mgr.delete_session(session_id) else: # Fallback: just clear the session data if method doesn't exist logger.warning("delete_session method not available, clearing session state only") return "", [], "", "🔴 **Session cleared - Create new session to continue**", f"✅ **Session `{session_id[:8]}...` cleared successfully**" except Exception as e: logger.error(f"Error clearing session: {e}") return session_id, [], "", _session_status(session_id), f"❌ **Error clearing session:** {e}" def force_summary(session_id: str): """Force summary using mgr (same as simple app)""" if not session_id: return "❌ No active session." try: if hasattr(mgr, 'summarize_session'): s = mgr.summarize_session(session_id) return f"📋 Summary updated:\n\n{s}" else: return "❌ Summarize function not available." except Exception as e: return f"❌ Failed to summarize: {e}" def lock_inputs(): """Lock inputs during processing (same as simple app)""" return gr.update(interactive=False), gr.update(interactive=False) def unlock_inputs(): """Unlock inputs after processing (same as simple app)""" return gr.update(interactive=True), gr.update(interactive=True) # ----------------------- # UI with improved architecture and greenish cream styling - LIGHT MODE DEFAULT # ----------------------- def create_app() -> gr.Blocks: # Enhanced CSS with greenish cream color scheme, fixed positioning, and light mode defaults custom_css = """ :root { --kallam-primary: #659435; --kallam-secondary: #5ea0bd; --kallam-accent: #b8aa54; --kallam-light: #f8fdf5; --kallam-dark: #2d3748; --kallam-cream: #f5f7f0; --kallam-green-cream: #e8f4e0; --kallam-border-cream: #d4e8c7; --shadow-soft: 0 4px 6px -1px rgba(0, 0, 0, 0.1), 0 2px 4px -1px rgba(0, 0, 0, 0.06); --shadow-medium: 0 10px 15px -3px rgba(0, 0, 0, 0.1), 0 4px 6px -2px rgba(0, 0, 0, 0.05); --border-radius: 12px; --transition: all 0.3s ease; } /* Force light mode styles - Override any dark mode defaults */ body, .gradio-container, .app { background-color: #ffffff !important; color: #2d3748 !important; } /* Ensure light backgrounds for all major containers */ .block, .form, .gap { background-color: #ffffff !important; color: #2d3748 !important; } /* Light mode for input elements */ input, textarea, select { background-color: #ffffff !important; border: 1px solid #d1d5db !important; color: #2d3748 !important; } input:focus, textarea:focus, select:focus { border-color: var(--kallam-primary) !important; box-shadow: 0 0 0 3px rgba(101, 148, 53, 0.1) !important; } /* Ensure dark mode styles don't override in light mode */ html:not(.dark) .dark { display: none !important; } .gradio-container { max-width: 100% !important; width: 100% !important; margin: 0 auto !important; min-height: 100vh; background-color: #ffffff !important; } .main-layout { display: flex !important; min-height: calc(100vh - 2rem) !important; gap: 1.5rem !important; } .fixed-sidebar { width: 320px !important; min-width: 320px !important; max-width: 320px !important; background: #ffffff !important; backdrop-filter: blur(10px) !important; border-radius: var(--border-radius) !important; border: 3px solid var(--kallam-primary) !important; box-shadow: var(--shadow-soft) !important; padding: 1.5rem !important; height: fit-content !important; position: sticky !important; top: 1rem !important; overflow: visible !important; } .main-content { flex: 1 !important; min-width: 0 !important; } .kallam-header { background: linear-gradient(135deg, var(--kallam-secondary) 0%, var(--kallam-primary) 50%, var(--kallam-accent) 100%); border-radius: var(--border-radius); padding: 2rem; margin-bottom: 1.5rem; text-align: center; box-shadow: var(--shadow-medium); position: relative; overflow: hidden; } .kallam-header h1 { color: white !important; font-size: 2.5rem !important; font-weight: 700 !important; margin: 0 !important; text-shadow: 0 2px 4px rgba(0,0,0,0.2); position: relative; z-index: 1; } .kallam-subtitle { color: rgba(255,255,255,0.9) !important; font-size: 1.1rem !important; margin-top: 0.5rem !important; position: relative; z-index: 1; } .btn { border-radius: 8px !important; font-weight: 600 !important; padding: 0.75rem 1.5rem !important; transition: var(--transition) !important; border: none !important; box-shadow: var(--shadow-soft) !important; cursor: pointer !important; } .btn:hover { transform: translateY(-2px) !important; box-shadow: var(--shadow-medium) !important; } .btn.btn-primary { background: linear-gradient(135deg, var(--kallam-primary) 0%, var(--kallam-secondary) 100%) !important; color: white !important; } .btn.btn-secondary { background: #f8f9fa !important; color: #2d3748 !important; border: 1px solid #d1d5db !important; } .chat-container { background: var(--kallam-green-cream) !important; border-radius: var(--border-radius) !important; border: 2px solid var(--kallam-border-cream) !important; box-shadow: var(--shadow-medium) !important; overflow: hidden !important; } .session-status-container .markdown { margin: 0 !important; padding: 0 !important; font-size: 0.85rem !important; line-height: 1.4 !important; overflow-wrap: break-word !important; word-break: break-word !important; } @media (max-width: 1200px) { .main-layout { flex-direction: column !important; } .fixed-sidebar { width: 100% !important; min-width: 100% !important; max-width: 100% !important; position: static !important; } } """ # Create a light theme with explicit light mode settings light_theme = gr.themes.Soft( # type: ignore primary_hue="green", secondary_hue="blue", neutral_hue="slate" ).set( # Force light mode colors body_background_fill="white", body_text_color="#2d3748", background_fill_primary="white", background_fill_secondary="#f8f9fa", border_color_primary="#d1d5db", border_color_accent="#659435", button_primary_background_fill="#659435", button_primary_text_color="white", button_secondary_background_fill="#f8f9fa", button_secondary_text_color="#2d3748" ) with gr.Blocks( title="🥬 KaLLaM - Thai Motivational Therapeutic Advisor", theme=light_theme, css=custom_css, js=""" function() { // Force light mode on load by removing any dark classes and setting light preferences document.documentElement.classList.remove('dark'); document.body.classList.remove('dark'); // Set data attributes for light mode document.documentElement.setAttribute('data-theme', 'light'); // Override any system preferences for dark mode const style = document.createElement('style'); style.textContent = ` @media (prefers-color-scheme: dark) { :root { color-scheme: light !important; } body, .gradio-container { background-color: white !important; color: #2d3748 !important; } } `; document.head.appendChild(style); } """ ) as app: # State management - same as simple app session_id = gr.State(value="") # Header gr.HTML(f"""
{CABBAGE_SVG}

KaLLaM

Thai Motivational Therapeutic Advisor

""") # Main layout with gr.Row(elem_classes=["main-layout"]): # Sidebar with enhanced styling with gr.Column(scale=1, elem_classes=["fixed-sidebar"]): gr.HTML("""

Controls

Manage session and health profile

""") with gr.Group(): new_session_btn = gr.Button("➕ New Session", variant="primary", size="lg", elem_classes=["btn", "btn-primary"]) health_profile_btn = gr.Button("👤 Custom Health Profile", variant="secondary", elem_classes=["btn", "btn-secondary"]) clear_session_btn = gr.Button("🗑️ Clear Session", variant="secondary", elem_classes=["btn", "btn-secondary"]) # Hidden health profile section with gr.Column(visible=False) as health_profile_section: gr.HTML('

') health_context = gr.Textbox( label="🏥 Patient's Health Information", placeholder="e.g., Patient's name, age, medical conditions (high blood pressure, diabetes), current symptoms, medications, lifestyle factors, mental health status...", lines=5, max_lines=8, info="This information helps KaLLaM provide more personalized and relevant health advice. All data is kept confidential within your session." ) with gr.Row(): update_profile_btn = gr.Button("💾 Update Health Profile", variant="primary", elem_classes=["btn", "btn-primary"]) back_btn = gr.Button("⏪ Back", variant="secondary", elem_classes=["btn", "btn-secondary"]) gr.HTML('

') # Session status session_status = gr.Markdown(value="🔄 **Initializing...**") # Main chat area with gr.Column(scale=3, elem_classes=["main-content"]): gr.HTML("""

💬 Health Consultation Chat

Chat with your AI health advisor in Thai or English

""") chatbot = gr.Chatbot( label="Chat with KaLLaM", height=500, show_label=False, type="messages", elem_classes=["chat-container"] ) with gr.Row(): with gr.Column(scale=5): msg = gr.Textbox( label="Message", placeholder="Ask about your health in Thai or English...", lines=1, max_lines=4, show_label=False, elem_classes=["chat-container"] ) with gr.Column(scale=1, min_width=120): send_btn = gr.Button("➤", variant="primary", size="lg", elem_classes=["btn", "btn-primary"]) # Result display result_display = gr.Markdown(visible=False) # Footer gr.HTML("""
Built with ❤️ by:
👨‍💻 Nopnatee Trivoravong
📧 nopnatee.triv@gmail.com GitHub
|
👨‍💻 Khamic Srisutrapon
📧 khamic.sk@gmail.com GitHub
|
👩‍💻 Napas Siripala
📧 millynapas@gmail.com GitHub
""") # ====== EVENT HANDLERS - Same pattern as simple app ====== # Auto-initialize on page load (same as simple app) def _init(): sid, history, _, status, note = start_new_session("") return sid, history, status, note app.load( fn=_init, inputs=None, outputs=[session_id, chatbot, session_status, result_display] ) # New session new_session_btn.click( fn=lambda: start_new_session(""), inputs=None, outputs=[session_id, chatbot, msg, session_status, result_display] ) # Show/hide health profile section def show_health_profile(): return gr.update(visible=True) def hide_health_profile(): return gr.update(visible=False) health_profile_btn.click( fn=show_health_profile, outputs=[health_profile_section] ) back_btn.click( fn=hide_health_profile, outputs=[health_profile_section] ) # Update health profile update_profile_btn.click( fn=update_health_profile, inputs=[session_id, health_context], outputs=[result_display, session_status] ).then( fn=hide_health_profile, outputs=[health_profile_section] ) # Send message with lock/unlock pattern (inspired by simple app) send_btn.click( fn=lock_inputs, inputs=None, outputs=[send_btn, msg], queue=False, # lock applies instantly ).then( fn=send_message, inputs=[msg, chatbot, session_id], outputs=[chatbot, msg, session_id, session_status], ).then( fn=unlock_inputs, inputs=None, outputs=[send_btn, msg], queue=False, ) # Enter/submit flow: same treatment msg.submit( fn=lock_inputs, inputs=None, outputs=[send_btn, msg], queue=False, ).then( fn=send_message, inputs=[msg, chatbot, session_id], outputs=[chatbot, msg, session_id, session_status], ).then( fn=unlock_inputs, inputs=None, outputs=[send_btn, msg], queue=False, ) # Clear session clear_session_btn.click( fn=clear_session, inputs=[session_id], outputs=[session_id, chatbot, msg, session_status, result_display] ) return app def main(): app = create_app() # Resolve bind address and port server_name = os.getenv("GRADIO_SERVER_NAME", "0.0.0.0") server_port = int(os.getenv("PORT", os.getenv("GRADIO_SERVER_PORT", 8080))) # Basic health log to confirm listening address try: hostname = socket.gethostname() ip_addr = socket.gethostbyname(hostname) except Exception: hostname = "unknown" ip_addr = "unknown" logger.info( "Starting Gradio app | bind=%s:%s | host=%s ip=%s", server_name, server_port, hostname, ip_addr, ) logger.info( "Env: PORT=%s GRADIO_SERVER_NAME=%s GRADIO_SERVER_PORT=%s", os.getenv("PORT"), os.getenv("GRADIO_SERVER_NAME"), os.getenv("GRADIO_SERVER_PORT"), ) # Secrets presence check (mask values) def _mask(v: str | None) -> str: if not v: return "" return f"set(len={len(v)})" logger.info( "Secrets: SEA_LION_API_KEY=%s GEMINI_API_KEY=%s", _mask(os.getenv("SEA_LION_API_KEY")), _mask(os.getenv("GEMINI_API_KEY")), ) app.launch( share=True, server_name=server_name, # cloud: 0.0.0.0, local: 127.0.0.1 server_port=server_port, # cloud: $PORT, local: 7860/8080 debug=False, show_error=True, inbrowser=True ) if __name__ == "__main__": main()