Spaces:
Running
Running
| import gradio as gr | |
| import requests | |
| import json | |
| import logging | |
| import os | |
| from dotenv import load_dotenv | |
| from typing import Dict, List, Tuple | |
| # Load environment variables | |
| load_dotenv() | |
| # Configure logging | |
| logging.basicConfig( | |
| filename='gradio_frontend.log', | |
| level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s' | |
| ) | |
| # API Configuration | |
| API_BASE_URL=os.getenv("API_BASE_URL") | |
| async def process_query(message: str, language: str, mode: str, chat_history: List[Dict]) -> Tuple[str, str]: | |
| """Make API request to get chatbot response""" | |
| try: | |
| payload = { | |
| "message": message, | |
| "language": language, | |
| "mode": mode | |
| } | |
| response = requests.post( | |
| f"{API_BASE_URL}/api/query", | |
| json=payload | |
| ) | |
| response.raise_for_status() | |
| result = response.json() | |
| return result["answer"], result.get("id") | |
| except requests.RequestException as e: | |
| logging.error(f"API request failed: {str(e)}") | |
| return f"Sorry, there was an error processing your request: {str(e)}", None | |
| async def submit_feedback(feedback_type: str, full_chat_history: List[Dict], language_choice: str, mode: str) -> str: | |
| """Submit feedback via API""" | |
| try: | |
| if mode == "Private": | |
| return "Feedback is disabled in private mode." | |
| if not full_chat_history or len(full_chat_history) < 2: | |
| return "No recent interaction to provide feedback for." | |
| # Get the last assistant message with an ID | |
| assistant_messages = [msg for msg in full_chat_history | |
| if msg.get("role") == "assistant" and "id" in msg] | |
| if not assistant_messages: | |
| logging.error("No assistant messages with ID found in chat history") | |
| return "No response found to provide feedback for." | |
| last_message = assistant_messages[-1] | |
| question_id = last_message.get("id") | |
| if not question_id: | |
| logging.error(f"No ID found in last message: {last_message}") | |
| return "Question ID not found. Please try asking another question." | |
| logging.info(f"Submitting feedback for question ID: {question_id}") | |
| # Submit feedback | |
| payload = { | |
| "question_id": question_id, | |
| "feedback_type": feedback_type, | |
| "language": language_choice, | |
| "mode": mode | |
| } | |
| response = requests.post( | |
| f"{API_BASE_URL}/api/feedback", | |
| json=payload | |
| ) | |
| response.raise_for_status() | |
| return "Thanks for your feedback! Your response has been recorded." | |
| except requests.RequestException as e: | |
| logging.error(f"Failed to submit feedback: {str(e)}") | |
| return f"Sorry, there was an error submitting your feedback: {str(e)}" | |
| except Exception as e: | |
| logging.error(f"Unexpected error in submit_feedback: {str(e)}") | |
| return "An unexpected error occurred. Please try again." | |
| # Create Gradio interface | |
| with gr.Blocks(title="RRA FAQ Chatbot") as demo: | |
| full_chat_history = gr.State([]) # Add this line | |
| gr.Markdown( | |
| """ | |
| # RRA FAQ Chatbot | |
| Ask tax-related questions in English or Kinyarwanda | |
| """ | |
| ) | |
| # Add mode selector | |
| with gr.Row(): | |
| interaction_mode = gr.Radio( | |
| choices=["Normal", "Private"], | |
| value="Normal", | |
| label="Interaction Mode", | |
| info="Normal: Stores interactions to improve service | Private: No data storage" | |
| ) | |
| with gr.Row(): | |
| gr.Markdown( | |
| """ | |
| > 📝 **Data Storage Notice:** | |
| > - Normal Mode: Questions and interactions are stored to improve our service | |
| > - Private Mode: No data is stored, feedback feature is disabled | |
| """ | |
| ) | |
| # Add language selector | |
| language = gr.Radio( | |
| choices=["English", "Kinyarwanda"], | |
| value="English", | |
| label="Select Language / Hitamo Ururimi" | |
| ) | |
| chatbot = gr.Chatbot( | |
| value=[], | |
| show_label=False, | |
| height=400 | |
| ) | |
| with gr.Row(): | |
| msg = gr.Textbox( | |
| label="Ask your question", | |
| placeholder="Type your tax-related question here...", | |
| show_label=False | |
| ) | |
| submit = gr.Button("Send") | |
| # Add feedback section | |
| feedback_container = gr.Row(visible=True) | |
| with feedback_container: | |
| with gr.Column(scale=2): | |
| feedback_label = gr.Markdown("Was this response helpful?") | |
| with gr.Column(scale=1): | |
| feedback_positive = gr.Button("👍 Helpful") | |
| with gr.Column(scale=1): | |
| feedback_negative = gr.Button("👎 Not Helpful") | |
| # Add feedback status message | |
| feedback_status = gr.Markdown("") | |
| async def respond(message, lang, mode, chat_history, full_chat_history): | |
| """Process a user message and update chat history""" | |
| try: | |
| if chat_history is None: | |
| chat_history = [] | |
| if full_chat_history is None: | |
| full_chat_history = [] | |
| # Get response from API | |
| bot_message, question_id = await process_query(message, lang, mode, chat_history) | |
| if not bot_message: | |
| return "", chat_history, full_chat_history | |
| # Build new messages | |
| user_message = { | |
| "content": message, | |
| "role": "user" | |
| } | |
| assistant_message = { | |
| "content": bot_message, | |
| "role": "assistant", | |
| "id": question_id # Store ID in the message | |
| } | |
| # Append to full chat history | |
| new_full_history = full_chat_history + [user_message, assistant_message] | |
| # Prepare messages for chatbot display | |
| new_chat_history = chat_history + [[message, bot_message]] | |
| return "", new_chat_history, new_full_history | |
| except Exception as e: | |
| logging.error(f"Error in respond function: {e}") | |
| return "", chat_history, full_chat_history | |
| def update_mode(mode: str): | |
| """Update UI when mode changes""" | |
| is_private = (mode == "Private") | |
| return { | |
| feedback_container: gr.update(visible=not is_private), | |
| feedback_status: gr.update(value="" if not is_private else "Feedback is disabled in private mode") | |
| } | |
| # Connect feedback buttons | |
| async def submit_positive_feedback(full_chat_history, language_choice, mode): | |
| return await submit_feedback("positive", full_chat_history, language_choice, mode) | |
| async def submit_negative_feedback(full_chat_history, language_choice, mode): | |
| return await submit_feedback("negative", full_chat_history, language_choice, mode) | |
| feedback_positive.click( | |
| fn=submit_positive_feedback, | |
| inputs=[full_chat_history, language, interaction_mode], | |
| outputs=feedback_status | |
| ) | |
| feedback_negative.click( | |
| fn=submit_negative_feedback, | |
| inputs=[full_chat_history, language, interaction_mode], | |
| outputs=feedback_status | |
| ) | |
| # Update UI when mode changes | |
| interaction_mode.change( | |
| fn=update_mode, | |
| inputs=[interaction_mode], | |
| outputs=[feedback_container, feedback_status] | |
| ) | |
| # Example questions | |
| with gr.Row() as english_examples_row: | |
| gr.Examples( | |
| examples=[ | |
| "What is VAT in Rwanda?", | |
| "How do I register for taxes?", | |
| "What are the tax payment deadlines?", | |
| "How can I get a TIN number?", | |
| "How do I get purchase code?" | |
| ], | |
| inputs=msg, | |
| label="English Examples" | |
| ) | |
| with gr.Row(visible=False) as kinyarwanda_examples_row: | |
| gr.Examples( | |
| examples=[ | |
| "Ese VAT ni iki mu Rwanda?", | |
| "Nabona TIN number nte?", | |
| "Ni ryari tugomba kwishyura imisoro?", | |
| "Ese nandikwa nte ku musoro?", | |
| "Ni gute nabone kode yo kugura?" | |
| ], | |
| inputs=msg, | |
| label="Kinyarwanda Examples" | |
| ) | |
| def toggle_language_interface(language_choice): | |
| if language_choice == "English": | |
| placeholder_text = "Type your tax-related question here..." | |
| return { | |
| msg: gr.update(placeholder=placeholder_text), | |
| english_examples_row: gr.update(visible=True), | |
| kinyarwanda_examples_row: gr.update(visible=False) | |
| } | |
| else: | |
| placeholder_text = "Andika ibibazo bijyanye n'umusoro hano" | |
| return { | |
| msg: gr.update(placeholder=placeholder_text), | |
| english_examples_row: gr.update(visible=False), | |
| kinyarwanda_examples_row: gr.update(visible=True) | |
| } | |
| # Connect user inputs | |
| msg.submit( | |
| respond, | |
| [msg, language, interaction_mode, chatbot, full_chat_history], | |
| [msg, chatbot, full_chat_history] | |
| ) | |
| submit.click( | |
| respond, | |
| [msg, language, interaction_mode, chatbot, full_chat_history], | |
| [msg, chatbot, full_chat_history] | |
| ) | |
| # Update interface on language change | |
| language.change( | |
| fn=toggle_language_interface, | |
| inputs=language, | |
| outputs=[msg, english_examples_row, kinyarwanda_examples_row] | |
| ) | |
| gr.Markdown( | |
| """ | |
| ### About | |
| - Created by: [Cedric](mailto:[email protected]) | |
| - Data source: [RRA Website FAQ](https://www.rra.gov.rw/en/domestic-tax-services/faqs) | |
| **Disclaimer:** This chatbot provides general tax information. For official guidance, | |
| consult RRA or call 3004. | |
| """ | |
| ) | |
| if __name__ == "__main__": | |
| try: | |
| demo.launch(share=False) | |
| logging.info("Gradio app launched successfully.") | |
| except Exception as launch_error: | |
| logging.critical(f"Failed to launch Gradio app: {launch_error}") | |
| raise | |