app.py

"""
Financial AI Chatbot with Smart Routing & RAG - Gradio Frontend

This Gradio application demonstrates a complete GenAI product development workflow,
showcasing smart routing capabilities of an AI chatbot for financial Q&A based on financial reports.

Key Features:
- Smart routing between FAQ, RAG, and LLM responses
- Real-time routing insights and answer quality scoring
- Production-ready architecture with separated backend/frontend
- Interactive examples for different routing scenarios

Backend API: Deployed on Render with FastAPI + LangChain
Frontend UI: This Gradio interface deployed on Hugging Face Spaces
Data: 2024 financial reports from 5 major companies (Apple, Google, Amazon, Tesla, Intel)

For complete technical details and implementation guide, see:
https://huggingface.co/spaces/krinya/smart_routing_with_render_example/blob/main/README.md
"""

import gradio as gr
import requests
import uuid
from datetime import datetime
from typing import Dict, List, Tuple, Optional
import time

API_BASE_URL = "https://gen-ai-demo-rag-bot.onrender.com"
CHAT_ENDPOINT = f"{API_BASE_URL}/chat"
HEALTH_ENDPOINT = f"{API_BASE_URL}/health"
DOCS_ENDPOINT = f"{API_BASE_URL}/docs"

EXAMPLE_QUERIES = {
    "FAQ": "Who is the CEO of Tesla?",
    "RAG": "What was Apple's revenue in 2024?", 
    "LLM": "How do you calculate price-to-earnings ratio?"
}

ROUTING_COLORS = {
    "faq": "🔍 #4CAF50",
    "rag": "📚 #2196F3",
    "llm": "🧠 #FF9800",
    "general": "💭 #9E9E9E"
}

def check_api_health(retries: int = 6, timeout_secs: int = 20, backoff_secs: int = 3) -> Tuple[bool, str]:
    """Check if the API is accessible.

    Uses a small retry loop with exponential-ish backoff to tolerate cold starts
    (Render free tier can take a while on the first request). Returns a
    (bool, message) tuple where bool indicates healthy.
    """
    last_err = None
    for attempt in range(1, retries + 1):
        try:
            response = requests.get(HEALTH_ENDPOINT, timeout=timeout_secs)
            if response.status_code == 200:
                return True, "API is online and healthy"
            else:
                return False, (
                    f"API returned status {response.status_code}. "
                    "The free Render API may take up to 1 minute to start on the first request, check the status on: {DOCS_ENDPOINT}. "
                    "Please wait a minute and try again."
                )
        except requests.exceptions.RequestException as e:
            last_err = e
            if attempt < retries:
                time.sleep(backoff_secs * attempt)
                continue
            return False, (
                f"❌ Cannot connect to API: {str(last_err)}. "
                "The free Render API may take up to 1 minute to start on the first request. , check the status on: {DOCS_ENDPOINT}. "
                "Please wait a minute and try again."
            )

def send_message_to_api(message: str, session_id: str) -> Dict:
    """Send message to the chatbot API"""
    try:
        payload = {
            "message": message,
            "session_id": session_id
        }
        response = requests.post(
            CHAT_ENDPOINT,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=150
        )
        if response.status_code == 200:
            return response.json()
        else:
            return {
                "error": f"API Error {response.status_code}: {response.text}",
                "response": "Sorry, I'm having trouble connecting to the server right now."
            }
    except requests.exceptions.Timeout:
        return {
            "error": "Request timeout",
            "response": "Sorry, the request took too long. Please try again."
        }
    except requests.exceptions.RequestException as e:
        return {
            "error": f"Connection error: {str(e)}",
            "response": "Sorry, I can't connect to the server right now."
        }

def format_routing_info(routing_data: Dict) -> str:
    """Format routing information for display"""
    if not routing_data:
        return "No routing information available"
    
    primary_route = routing_data.get('primary_route', 'unknown')
    answer_quality = routing_data.get('answer_quality', 'unknown')
    color_info = ROUTING_COLORS.get(primary_route.lower(), ROUTING_COLORS['general'])
    icon, color = color_info.split(' ')
    
    info_lines = [
        f"{icon} **Route:** {primary_route.upper()}",
        f"⭐ **Quality:** {answer_quality.title()}"
    ]
    
    rephrase_attempts = routing_data.get('rephrase_attempts', 0)
    if rephrase_attempts > 0:
        info_lines.append(f"🔄 **Rephrase attempts:** {rephrase_attempts}")
    
    failed_sources = routing_data.get('failed_sources', [])
    if failed_sources:
        info_lines.append(f"⚠️ **Failed sources:** {', '.join(failed_sources)}")
    
    return "\n\n".join(info_lines)

def format_chat_message(message: str, is_user: bool, routing_info: Optional[Dict] = None) -> str:
    timestamp = datetime.now().strftime("%H:%M")
    if is_user:
        return f"**👤 You** *({timestamp})*\n{message}"
    else:
        route_indicator = ""
        if routing_info:
            primary_route = routing_info.get('primary_route', 'general').lower()
            color_info = ROUTING_COLORS.get(primary_route, ROUTING_COLORS['general'])
            icon = color_info.split(' ')[0]
            route_indicator = f" {icon}"
        return f"**🤖 Assistant{route_indicator}** *({timestamp})*\n{message}"

def chat_with_bot(message: str, history: List[Dict[str, str]], session_id: str, show_routing: bool) -> Tuple[List[Dict[str, str]], str, str, str]:
    """Main chat function"""
    if not message.strip():
        return history, "", "", session_id
    
    # Send message to API with persistent session ID
    api_response = send_message_to_api(message, session_id)
    
    # Extract response and routing info
    bot_response = api_response.get('response', 'Sorry, I encountered an error.')
    metadata = api_response.get('metadata', {})
    routing_info = metadata.get('routing_info', {})
    
    # Format routing information
    routing_display = ""
    if show_routing and routing_info:
        routing_display = format_routing_info(routing_info)
    
    # Add to chat history using messages format
    history.append({"role": "user", "content": message})
    history.append({"role": "assistant", "content": bot_response})
    
    return history, "", routing_display, session_id

def load_example(example_text: str) -> str:
    """Load an example query into the input box"""
    return example_text

def create_gradio_interface():
    """Create and configure the Gradio interface"""
    
    # Check API health at startup
    is_healthy, health_status = check_api_health()
    
    with gr.Blocks(
        title="AI Chatbot with Smart Routing",
        theme=gr.themes.Default(primary_hue="blue", secondary_hue="purple")
    ) as interface:
        
        # Header
        gr.Markdown("""
        # 🤖 Financial AI Chatbot with Smart Routing & RAG
        
        **A demo GenAI app that demonstrates smart routing using LangChain - showing how to create a complete GenAI product**
        
        ## 🎯 What This Demonstrates
        
        This project showcases **a GenAI development workflow** from backend to frontend deployment we created an API running on Render and a frontend UI using Gradio on Hugging Face Spaces.:
        
        ### 🧠 Smart Routing with LangChain
        Intelligently routes financial questions about **5 major companies** (Apple, Google, Amazon, Tesla, Intel):
        - 🔍 **FAQ Route**: Quick facts (CEO names, founding dates, basic company info)
        - 📚 **RAG Route**: Detailed financial data from 2024 annual reports (revenue, profits, growth metrics)  
        - 🧠 **LLM Route**: General explanations and complex financial concepts
        
        ### 📊 RAG Implementation
        - **Vector Storage**: ChromaDB with processed financial documents (full annual reports)
        - **Retrieval System**: Semantic search for relevant information
        - **Smart Fallbacks**: Multiple sources with quality scoring
        
        ### 🏗️ Backend and Frontend Architecture
        - **Backend**: Python FastAPI with LangChain, deployed on Render
        - **Frontend**: Gradio UI deployed on Hugging Face Spaces using Docker containerization
        - **Separation**: Backend API + Frontend UI
        
        **🔧 Tech Stack**: OpenAI GPT-5-mini + LangChain orchestration, Python FastAPI, ChromaDB vector database, Docker containerization
        
        **📖 Learn More**: [README with technical details](https://huggingface.co/spaces/krinya/smart_routing_with_render_example/blob/main/README.md)  
        **💻 Backend API Code**: [GitHub Repository](https://github.com/krinya/gen_ai_demo_rag_bot/tree/main)
        """)
        
        # Workflow Architecture Diagram
        gr.Markdown("### 📊 Chatbot Workflow Architecture")
        gr.Image(
            value="chatbot_workflow_graph.png",
            label="Chatbot Workflow Architecture Diagram",
            show_label=True,
            container=True,
            height=400,
            width=800,
            interactive=False
        )
        
        # API Health Status
        with gr.Row():
            if is_healthy:
                gr.Markdown(f"✅ **Status**: {health_status}", container=True)
            else:
                gr.Markdown(f"❌ **Status**: {health_status}", container=True)
        
        # Hidden session ID state (persistent across interactions)
        session_state = gr.State(value=str(uuid.uuid4()))
        
        # Chat interface (full width)
        chatbot = gr.Chatbot(
            value=[],
            label="Chat History",
            height=500,
            show_label=True,
            type="messages",
            latex_delimiters=[
                {"left": "$$", "right": "$$", "display": True},
                {"left": "\\[", "right": "\\]", "display": True},
                {"left": "\\(", "right": "\\)", "display": False}
            ]
        )
        
        with gr.Row():
            msg_input = gr.Textbox(
                placeholder="Ask about Apple, Google, Amazon, Tesla, or Intel financials...",
                label="Your Financial Question",
                scale=4,
                lines=1
            )
            send_btn = gr.Button("Send 📤", scale=1, variant="primary")
        
        # Example queries below chat interface
        gr.Markdown("### 💡 Try These Examples")
        
        with gr.Row():
            for route_type, example in EXAMPLE_QUERIES.items():
                color_info = ROUTING_COLORS.get(route_type.lower(), ROUTING_COLORS['general'])
                icon, color = color_info.split(' ')
                
                example_btn = gr.Button(
                    f"{icon} {example}",
                    size="sm"
                )
                example_btn.click(
                    fn=load_example,
                    inputs=[gr.State(example)],
                    outputs=[msg_input]
                )
        
        # Settings and controls
        with gr.Row():
            show_routing = gr.Checkbox(
                value=True,
                label="Show routing insights",
                info="Display how the AI routes your questions"
            )
            clear_btn = gr.Button("🗑️ Clear Chat", variant="secondary")
            new_session_btn = gr.Button("🔄 New Session", variant="secondary")
            session_indicator = gr.Markdown("💾 **Memory Active** - I'll remember our conversation")
        
        # Routing insights at the bottom
        routing_info = gr.Markdown(
            value="*Routing information will appear here after sending a message*",
            label="🧭 Routing Insights"
        )
        
        # Footer with deployment info
        gr.Markdown("""
        ---
        **🚀 Deployment Info**: This prototype is powered by a FastAPI backend deployed on [Render](https://render.com), 
        showcasing full-stack development knowledge.
        
        **🛠️ Tech Stack**: LangChain • OpenAI GPT-5-mini • ChromaDB • FastAPI • Render • Gradio • Hugging Face Spaces • CI/CD
        """)
        
        # Event handlers
        def clear_chat():
            return [], ""
        
        def new_session():
            return str(uuid.uuid4()), [], ""
        
        # Button click events
        clear_btn.click(
            fn=clear_chat,
            outputs=[chatbot, routing_info]
        )
        
        new_session_btn.click(
            fn=new_session,
            outputs=[session_state, chatbot, routing_info]
        )
        
        # Chat submission events with loading
        def chat_wrapper(message, history, session_id, show_routing):
            # Show loading message
            if message.strip():
                # Add user message and loading response immediately
                loading_history = history + [
                    {"role": "user", "content": message},
                    {"role": "assistant", "content": "🤔 Thinking... be patient, free servers are slow."}
                ]
                yield loading_history, "", "🔄 Processing your message...", session_id
                
                # Get actual response
                result_history, empty_input, routing_info, updated_session = chat_with_bot(message, history, session_id, show_routing)
                yield result_history, "", routing_info, updated_session
            else:
                yield history, "", "", session_id
        
        send_btn.click(
            fn=chat_wrapper,
            inputs=[msg_input, chatbot, session_state, show_routing],
            outputs=[chatbot, msg_input, routing_info, session_state]
        )
        
        msg_input.submit(
            fn=chat_wrapper,
            inputs=[msg_input, chatbot, session_state, show_routing],
            outputs=[chatbot, msg_input, routing_info, session_state]
        )
        
    return interface

if __name__ == "__main__":
    # Create and launch the interface
    interface = create_gradio_interface()
    
    print("🚀 Starting Gradio Chat Interface...")
    print(f"🔗 API Endpoint: {API_BASE_URL}")
    # Launch with Hugging Face Spaces configuration
    interface.launch(
        server_name="0.0.0.0", 
        server_port=7860,
        share=False,            
        show_error=True,
        favicon_path='robot_favicon.png',
        auth=None
    )