Spaces:
Running
Running
| import gradio as gr | |
| from datetime import datetime | |
| import json | |
| import difflib # For better product name matching | |
| from utils import get_product_by_name, get_products_by_category, get_products_and_category | |
| # Dummy moderation function (replace with actual API) | |
| def openai_moderation_check(text): | |
| return {"flagged": False} # Assume no flagging for now | |
| # Dummy AI response function (replace with actual API call to generate responses) | |
| def get_completion_from_messages(messages): | |
| return "This is a placeholder response from the model." # Mock response for now | |
| # Function to log conversation to a JSON file | |
| def log_conversation(user_input, ai_response, metadata=None, history=None): | |
| log_entry = { | |
| "timestamp": str(datetime.now()), | |
| "user_input": user_input, | |
| "ai_response": ai_response, | |
| "metadata": metadata, | |
| "history": history | |
| } | |
| with open("conversation_log.json", "a") as log_file: | |
| log_file.write(json.dumps(log_entry) + "\n") | |
| # Function to handle user queries using utils.py functions | |
| def handle_user_query(user_input, all_messages=[], debug=False): | |
| # Step 1: Moderation check on user input | |
| moderation_result = openai_moderation_check(user_input) | |
| if moderation_result["flagged"]: | |
| return "Sorry, your request is non-compliant.", all_messages | |
| # Step 2: Extract products and categories from user input | |
| products_and_category = get_products_and_category() | |
| # Debugging: Print the structure of products_and_category | |
| if debug: | |
| print("Products and Categories Structure:", products_and_category) | |
| product_name = None | |
| category_name = None | |
| # Flatten the product list to check for partial matches | |
| all_products = [] | |
| for category, products in products_and_category.items(): | |
| all_products.extend(products) | |
| # Use difflib to find the closest match to the user input | |
| product_matches = difflib.get_close_matches(user_input, all_products, n=1, cutoff=0.4) | |
| if product_matches: | |
| product_name = product_matches[0] | |
| # Check for category match if no product match is found | |
| if not product_name: | |
| category_matches = difflib.get_close_matches(user_input, products_and_category.keys(), n=1, cutoff=0.4) | |
| if category_matches: | |
| category_name = category_matches[0] | |
| # Step 3: Generate a response | |
| if product_name: | |
| product_info = get_product_by_name(product_name) | |
| if product_info: | |
| response = f"Product: {product_info['name']}\n" \ | |
| f"Category: {product_info['category']}\n" \ | |
| f"Price: ${product_info['price']}\n" \ | |
| f"Rating: {product_info['rating']} stars\n" \ | |
| f"Features: {', '.join(product_info['features'])}\n" \ | |
| f"Description: {product_info['description']}" | |
| return response, all_messages | |
| else: | |
| return "Sorry, I couldn't find the product you're asking about.", all_messages | |
| elif category_name: | |
| products_in_category = get_products_by_category(category_name) | |
| if products_in_category: | |
| response = f"Category: {category_name}\n" | |
| for product in products_in_category: | |
| response += f"Product: {product['name']}\nPrice: ${product['price']}\n\n" | |
| return response.strip(), all_messages | |
| else: | |
| return "Sorry, I couldn't find products in that category.", all_messages | |
| else: | |
| return "Please provide the name of a product or category you'd like to know about.", all_messages | |
| # Chatbot logic to handle conversation | |
| def handle_chat(user_input, history): | |
| response, updated_history = handle_user_query(user_input, history) | |
| history.append((user_input, response)) # Append the latest user input and response to the history | |
| log_conversation(user_input, response) # Log the interaction | |
| return response, history | |
| # Gradio chatbot UI setup | |
| def chatbot_ui(): | |
| with gr.Blocks() as app: | |
| gr.Markdown("# Store Assistant Chatbot") | |
| chatbot = gr.Chatbot(label="Chat with Store Assistant") | |
| message_input = gr.Textbox(label="Ask about products!") | |
| clear_btn = gr.Button("Clear Chat") | |
| # Initialize conversation history as a stateful variable | |
| conversation_history = gr.State([]) | |
| # Process user input and update the conversation | |
| def on_user_message(user_message, history): | |
| response, updated_history = handle_chat(user_message, history) | |
| return "", updated_history # Removed the extra append | |
| # Clear chat history | |
| def clear_chat(): | |
| return [], [] | |
| # Link input to chatbot | |
| message_input.submit(on_user_message, inputs=[message_input, conversation_history], outputs=[message_input, chatbot]) | |
| clear_btn.click(clear_chat, inputs=[], outputs=[chatbot, conversation_history]) | |
| app.launch() | |
| # Run the chatbot interface | |
| if __name__ == "__main__": | |
| chatbot_ui() |