"""Gradio Demo: Reachy Mini Retail Assistant Demonstrates conversational product search with backend telemetry. HuggingFace Spaces ready version. """ import gradio as gr import os from datetime import datetime from pathlib import Path # Load environment variables from parent .env file (local dev only) from dotenv import load_dotenv env_path = Path(__file__).parent.parent / "reachy_edge" / ".env" if env_path.exists(): load_dotenv(env_path) print(f"[DEBUG] Loaded .env from {env_path}") else: print(f"[DEBUG] No .env file found at {env_path} - using environment variables") # Local imports (no sys.path hacks needed) from cache.l2_cache import ProductCache from models import Product # Try to import HuggingFace try: from huggingface_hub import InferenceClient HF_AVAILABLE = True HF_TOKEN = os.environ.get("HF_TOKEN") if HF_TOKEN: # Initialize client without model - specify per request hf_client = InferenceClient(token=HF_TOKEN) print(f"[DEBUG] ✅ HuggingFace client configured with token (length: {len(HF_TOKEN)})") print(f"[DEBUG] Will use model: meta-llama/Llama-3.3-70B-Instruct (free tier)") else: HF_AVAILABLE = False hf_client = None print(f"[DEBUG] ⚠️ No HF_TOKEN found - will use mock responses") except ImportError: HF_AVAILABLE = False hf_client = None print("[DEBUG] huggingface-hub not available") # Initialize cache (regular ProductCache - no thread-safety needed for demo) cache = ProductCache(db_path="./data/cache.db") cache.initialize() # Interaction history for telemetry interaction_history = [] # Mock LLM responses (fallback if no OpenAI API key) RESPONSE_TEMPLATES = { "diesel": "I can help you with diesel! We have premium diesel fuel available at Fuel Islands 1-4. Would you like to know about pricing or DEF fluid as well?", "fuel": "We have several fuel options available. Are you looking for diesel, gasoline, or something specific like DEF fluid?", "shower": "Our shower facilities are available! We offer 30-minute shower credits for $15. They're located at the Service Desk - nice and clean!", "food": "Hungry? We've got fresh hot food at the counter - pizza, burgers, chicken tenders, and breakfast sandwiches. What sounds good?", "coffee": "Need some fuel for yourself? We've got fresh truck stop coffee at the coffee station - large cups are just $2.49!", "cb radio": "Looking for a CB radio? We carry the Cobra 29 LX - it's a solid choice at $129.99. Find it in Aisle 4 (Electronics section).", "safety": "Safety first! We've got high-visibility vests, LED road flares, flashlights, and emergency kits in Aisle 5. What do you need?", "deal": "Great question! While I don't have today's specific deals loaded, I can show you our best everyday prices. Popular items include our coffee ($2.49), hot dogs ($2.99), and bulk snacks. Diesel fuel is competitively priced - check our sign out front for today's rate!", "special": "Great question! While I don't have today's specific deals loaded, I can show you our best everyday prices. Popular items include our coffee ($2.49), hot dogs ($2.99), and bulk snacks. Diesel fuel is competitively priced - check our sign out front for today's rate!", "cheap": "Looking for value? Our best deals are coffee at $2.49, hot dogs at $2.99, and we have great bulk pricing on snacks and energy drinks. What are you interested in?", "parking": "We offer overnight parking for $20 - secure and well-lit. Reserve your spot at the Service Desk!", "wash": "Our full-service truck wash is $125. Gets your rig looking like new! Find us at the Truck Wash Bay.", "tired": "Need a break? We've got showers ($15), hot coffee ($2.49), and energy drinks to keep you going. Or grab some overnight parking ($20) if you need rest!", } def get_mock_response(query: str, results: list = None) -> str: """Generate conversational response based on query and search results.""" query_lower = query.lower() # Check templates first for keyword, response in RESPONSE_TEMPLATES.items(): if keyword in query_lower: return response # Smart fallback based on results if results and len(results) > 0: top_result = results[0] return f"I found some great options! The top match is **{top_result.name}** for ${top_result.price:.2f} - you'll find it at {top_result.location}. Check out the products below!" # Last resort return "I'm searching our inventory for you! Let me show you what we have..." def get_llm_response(query: str, results: list) -> tuple[str, str]: """Generate response using HuggingFace. Returns: (response_text, source_indicator) """ if not HF_AVAILABLE or not hf_client: print(f"[DEBUG] ⚠️ HF unavailable - using mock response instead") return get_mock_response(query, results), "🔴 MOCK (no API token)" print(f"[DEBUG] 🚀 CALLING HF INFERENCE API for query: {query}") print(f"[DEBUG] Model: meta-llama/Llama-3.3-70B-Instruct") try: # Build context about products found if results: products_context = "I found these items in our truck stop:\n" for i, p in enumerate(results[:3], 1): # Check if it's a service or product is_service = p.category == "Services" item_type = "service" if is_service else "product" products_context += f"{i}. {p.name} (${p.price:.2f}) - {item_type} available at {p.location}\n" else: products_context = "I searched our inventory but didn't find any matching items." # Use chat_completion for conversational models messages = [ {"role": "user", "content": f"""You are Reachy, a helpful truck stop assistant. Customer asks: '{query}' {products_context} Respond naturally in 2-3 sentences: - If services found (like showers, parking, truck wash), say "we offer" or "we have" (not "carry") - If products found (like fuel, snacks, CB radios), mention them by name - If nothing found, politely say we don't have it and suggest checking with staff - Be conversational and helpful"""} ] # Call HuggingFace chat_completion API with Meta Llama 3.3 70B (free tier, reliable) response = hf_client.chat_completion( messages=messages, model="meta-llama/Llama-3.3-70B-Instruct", max_tokens=100, temperature=0.7 ) llm_response = response.choices[0].message.content.strip() print(f"[DEBUG] ✅ LLAMA SUCCESS! Response length: {len(llm_response)} chars") print(f"[DEBUG] Response preview: {llm_response[:100]}...") return llm_response, "🟢 Meta Llama 3.3 70B Instruct" except Exception as e: print(f"[ERROR] ❌ NVIDIA NEMOTRON API FAILED: {type(e).__name__}: {e}") print(f"[ERROR] Falling back to mock response") return get_mock_response(query, results), f"🔴 MOCK (API error: {type(e).__name__})" def format_product_card(product: Product) -> str: """Format product as HTML card.""" # Category emoji mapping emoji_map = { "Fuel & Fluids": "⛽", "Trucker Supplies": "🚛", "Electronics": "📻", "Energy & Snacks": "☕", "Hot Food & Beverages": "🍕", "Services": "🚿", "Safety & Lighting": "⚠️", "Convenience": "🛒" } emoji = emoji_map.get(product.category, "📦") relevance_bar = "" if product.relevance_score: score_pct = min(100, product.relevance_score * 10) # Scale for display relevance_bar = f'
' return f"""
{emoji}

{product.name}

SKU: {product.sku}

Category: {product.category}

Location: {product.location}

${product.price:.2f}

{relevance_bar} {f'

Relevance: {product.relevance_score:.2f}

' if product.relevance_score else ''}

{product.description}

""" def format_backend_activity(query: str, results: list, response_time_ms: float) -> str: """Format backend telemetry as HTML.""" timestamp = datetime.now().strftime("%H:%M:%S") activity = f"""
🔍 [{timestamp}] NEW QUERY RECEIVED
query:
"{query}"
💾 L2 CACHE (SQLite FTS5)
// Executing full-text search with BM25 ranking...
cache
.search_products("{query}", max_results=5)
✓ Found {len(results)} products in {response_time_ms:.2f}ms
📊 INTERACTION LOGGED
// Recording customer interaction for analytics...
{{
"timestamp": "{timestamp}",
"query": "{query}",
"results_count": {len(results)},
"response_time_ms": {response_time_ms:.2f}
}}
✓ Logged to analytics database
🧠 LEARNING INSIGHTS
// Updating search patterns for future optimization...
• Customer interest in: {results[0].category if results else 'N/A'}
• Popular query terms indexed
• Response quality will improve over time
""" return activity def chat(message: str, history: list) -> tuple: """Process customer query and return response with products and telemetry.""" import time start_time = time.time() # Search products print(f"[DEBUG] Searching for: {message}") results = cache.search_products(message, max_results=5) print(f"[DEBUG] Found {len(results)} results") if results: print(f"[DEBUG] First result: {results[0].name}") # Calculate response time response_time_ms = (time.time() - start_time) * 1000 # Generate conversational response using LLM (falls back to mock if no API key) print(f"\n{'='*60}") print(f"[QUERY] {message}") response, source_indicator = get_llm_response(message, results) print(f"[RESPONSE] {response[:80]}...") print(f"[SOURCE] {source_indicator}") print(f"{'='*60}\n") # Add source indicator at top source_badge_color = "#4CAF50" if "🟢" in source_indicator else "#FF9800" source_html = f"
{source_indicator}
" # Format products as HTML cards products_html = source_html + "
" if results: products_html += "

📦 Products Found:

" for product in results: products_html += format_product_card(product) else: products_html += "

No products found matching your query.

" products_html += "
" # Format backend activity backend_html = format_backend_activity(message, results, response_time_ms) # Log interaction interaction_history.append({ "timestamp": datetime.now().isoformat(), "query": message, "results_count": len(results), "response_time_ms": response_time_ms }) # Build conversation history in Gradio 6.0 format (list of messages with role/content) history = history or [] history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": response}) return history, products_html, backend_html def clear_chat(): """Clear chat history.""" return [], "", "" # Build Gradio interface with gr.Blocks(theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🤖 Reachy Mini: Retail Assistant Demo **Interaction Loop Demonstration:** - Customer asks questions in natural language - Reachy searches the product database (SQLite FTS5 + BM25 ranking) - **Meta Llama 3.3 70B Instruct** generates conversational responses (via HuggingFace API) - Visual product cards + backend telemetry - All interactions logged for continuous improvement *Try asking: "Where can I get diesel?", "I need a CB radio", "Do you have grape jelly?", "What's on sale?"* """) with gr.Row(): # Left column: Chat interface with gr.Column(scale=2): chatbot = gr.Chatbot( label="💬 Reachy Conversation", height=400, show_label=True ) with gr.Row(): msg = gr.Textbox( label="Ask Reachy...", placeholder="Where can I get diesel fuel?", show_label=False, scale=4 ) submit = gr.Button("Send", variant="primary", scale=1) clear = gr.Button("Clear Chat", variant="secondary") products_display = gr.HTML(label="Products") # Right column: Backend telemetry with gr.Column(scale=1): gr.Markdown("### 🔧 Backend Activity") backend_display = gr.HTML() # Event handlers submit.click( chat, inputs=[msg, chatbot], outputs=[chatbot, products_display, backend_display] ).then( lambda: "", outputs=[msg] ) msg.submit( chat, inputs=[msg, chatbot], outputs=[chatbot, products_display, backend_display] ).then( lambda: "", outputs=[msg] ) clear.click( clear_chat, outputs=[chatbot, products_display, backend_display] ) # Example queries gr.Examples( examples=[ ["Where can I get diesel fuel?"], ["I need a CB radio"], ["Where are the showers?"], ["I'm hungry, what do you have?"], ["Do you sell safety vests?"], ["Do you have grape jelly?"], ], inputs=[msg] ) if __name__ == "__main__": # Load sample products if cache is empty from data.sample_products import load_sample_data # Clear database to avoid duplicates on restart try: cache._get_connection().execute("DELETE FROM products_fts") cache._get_connection().commit() except: pass print("Loading sample products...") load_sample_data(cache) print("✓ 44 products loaded") demo.launch()