Download app.py from krinya/smart_routing_with_render_example: direct link, hf CLI and curl.
- Browser
- Download file 14.7 kB
-
https://huggingface.co/spaces/krinya/smart_routing_with_render_example/resolve/main/app.py
- Command line
-
hf download hf://spaces/krinya/smart_routing_with_render_example/app.py
-
curl -L -o app.py https://huggingface.co/spaces/krinya/smart_routing_with_render_example/resolve/main/app.py
14.7 kB
| """ | |
| 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 | |
| ) | |