"""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'
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}
No products found matching your query.
" products_html += "