#!/usr/bin/env python3 """ LLM Fallback Router Integration Example This example demonstrates how to integrate the LLMFallbackRouter with the existing VoucherBot system as a fallback for the regex-based router. Usage: python llm_fallback_router_example.py """ import os import json from dotenv import load_dotenv from llm_fallback_router import LLMFallbackRouter, InvalidInputError, LLMProcessingError, InvalidLLMResponseError # Import existing components from agent_setup import initialize_caseworker_agent from enhanced_semantic_router_v2 import EnhancedSemanticRouterV2, Intent # Load environment variables load_dotenv() class MockLLMClient: """ Mock LLM client for demonstration purposes. In a real implementation, this would be replaced with actual LLM clients like OpenAI, Anthropic, or the Gemini client used in the project. """ def __init__(self): self.call_count = 0 def generate(self, prompt: str) -> str: """ Generate a mock response based on the prompt content. In production, this would make actual API calls to an LLM. """ self.call_count += 1 # Extract the message from the prompt message_start = prompt.find('Message: "') + 10 message_end = prompt.find('"', message_start) message = prompt[message_start:message_end] if message_start > 9 else "" # Simple rule-based mock responses message_lower = message.lower() if any(word in message_lower for word in ["find", "search", "look for", "apartment", "listing"]): return json.dumps({ "intent": "SEARCH_LISTINGS", "parameters": { "borough": "Brooklyn" if "brooklyn" in message_lower or "bk" in message_lower else None, "bedrooms": 2 if "2" in message or "two" in message_lower else None, "max_rent": 3000 if "$3000" in message or "3000" in message else None, "voucher_type": "Section 8" if "section" in message_lower else None }, "reasoning": "User is looking for apartment listings with specified criteria" }) elif any(word in message_lower for word in ["what about", "try", "instead", "change"]): return json.dumps({ "intent": "REFINE_SEARCH", "parameters": { "borough": "Queens" if "queens" in message_lower else None }, "reasoning": "User wants to modify their existing search parameters" }) elif any(word in message_lower for word in ["violation", "safe", "building", "inspect"]): return json.dumps({ "intent": "CHECK_VIOLATIONS", "parameters": {}, "reasoning": "User wants to check building safety violations" }) elif any(word in message_lower for word in ["help", "assist", "what can you do"]): return json.dumps({ "intent": "HELP_REQUEST", "parameters": {}, "reasoning": "User is requesting help or information about available features" }) else: return json.dumps({ "intent": "UNKNOWN", "parameters": {}, "reasoning": "Unable to determine user intent from the message" }) class TwoTierSemanticRouter: """ Combined router that uses regex-based routing first, then falls back to LLM. This demonstrates the two-tier architecture mentioned in the specification. """ def __init__(self, llm_client=None, debug=False): # Initialize the regex-based router (V2) self.regex_router = EnhancedSemanticRouterV2() # Initialize the LLM fallback router if llm_client is None: llm_client = MockLLMClient() self.llm_router = LLMFallbackRouter(llm_client, debug=debug) self.debug = debug def route(self, message: str, context: dict = None) -> dict: """ Route a message using the two-tier system. Args: message: User message to route context: Optional context dictionary with conversation state Returns: Dictionary with routing results including: - intent: Classified intent - parameters: Extracted parameters - reasoning: Explanation of the classification - router_used: Which router was used ("regex" or "llm") - confidence: Confidence level (if available) """ if self.debug: print(f"\nπŸ” Routing message: '{message}'") # Step 1: Try regex-based routing first try: regex_intent = self.regex_router.classify_intent(message, context) regex_params = self.regex_router.extract_parameters(message) # Check if regex router was successful if regex_intent != Intent.UNCLASSIFIED and (regex_params or regex_intent in [Intent.SHOW_HELP, Intent.CHECK_VIOLATIONS]): if self.debug: print("βœ… Regex router succeeded") return { "intent": regex_intent.value, "parameters": regex_params, "reasoning": f"Classified by regex patterns as {regex_intent.value}", "router_used": "regex", "confidence": 0.95 # Regex patterns are highly confident when they match } except Exception as e: if self.debug: print(f"⚠️ Regex router failed: {e}") # Step 2: Fall back to LLM router if self.debug: print("🧠 Falling back to LLM router") try: # Convert context to string format for LLM context_str = None if context: context_str = f"Previous search: {json.dumps(context)}" llm_result = self.llm_router.route(message, context_str) llm_result["router_used"] = "llm" llm_result["confidence"] = 0.8 # LLM results are generally less confident if self.debug: print("βœ… LLM router succeeded") return llm_result except (InvalidInputError, LLMProcessingError, InvalidLLMResponseError) as e: if self.debug: print(f"❌ LLM router failed: {e}") # Both routers failed - return unknown intent return { "intent": "UNKNOWN", "parameters": {}, "reasoning": f"Both regex and LLM routers failed. Error: {e}", "router_used": "none", "confidence": 0.0 } def demonstrate_integration(): """Demonstrate the LLM Fallback Router integration.""" print("🏠 VoucherBot LLM Fallback Router Integration Demo") print("=" * 60) # Initialize the two-tier router mock_llm = MockLLMClient() router = TwoTierSemanticRouter(mock_llm, debug=True) # Test cases that demonstrate fallback behavior test_cases = [ # Cases that should work with regex router { "message": "Find apartments in Brooklyn with 2 bedrooms", "context": None, "expected_router": "regex" }, { "message": "Show me help", "context": None, "expected_router": "regex" }, # Cases that should fall back to LLM { "message": "I'm looking for a place but not sure where to start", "context": None, "expected_router": "llm" }, { "message": "ΒΏDΓ³nde puedo encontrar apartamentos?", # Spanish "context": None, "expected_router": "llm" }, { "message": "What about trying somewhere else?", "context": {"borough": "Brooklyn", "bedrooms": 2}, "expected_router": "llm" }, # Edge cases { "message": "yo wassup", # Very informal "context": None, "expected_router": "llm" } ] print("\nπŸ“‹ Running Test Cases:") print("-" * 40) for i, test_case in enumerate(test_cases, 1): print(f"\n{i}. Testing: '{test_case['message']}'") result = router.route(test_case["message"], test_case["context"]) print(f" Intent: {result['intent']}") print(f" Router Used: {result['router_used']}") print(f" Confidence: {result['confidence']}") print(f" Parameters: {result['parameters']}") print(f" Reasoning: {result['reasoning']}") # Verify expected router was used if result['router_used'] == test_case['expected_router']: print(" βœ… Expected router used") else: print(f" ⚠️ Expected {test_case['expected_router']}, got {result['router_used']}") def demonstrate_real_integration(): """ Demonstrate how this would integrate with the actual VoucherBot system. """ print("\n\nπŸ”§ Real Integration Example") print("=" * 40) # This is how you would integrate with the actual system print("Integration points:") print("1. Replace MockLLMClient with actual Gemini client from agent_setup.py") print("2. Integrate TwoTierSemanticRouter into email_handler.py") print("3. Update app.py to use the new router for message classification") # Example integration code integration_code = ''' # In email_handler.py - replace the current classification logic from llm_fallback_router import LLMFallbackRouter from agent_setup import initialize_caseworker_agent # Initialize LLM client (use the same one from agent_setup) caseworker_agent = initialize_caseworker_agent() llm_client = caseworker_agent.model # Extract the model # Create the two-tier router two_tier_router = TwoTierSemanticRouter(llm_client) # Use in classification def enhanced_classify_message(message: str, context: dict = None) -> str: result = two_tier_router.route(message, context) return result["intent"] ''' print("\nExample integration code:") print(integration_code) def demonstrate_error_handling(): """Demonstrate robust error handling.""" print("\n\nπŸ›‘οΈ Error Handling Demo") print("=" * 30) # Create router with a failing LLM client class FailingLLMClient: def generate(self, prompt): raise Exception("API timeout") failing_router = TwoTierSemanticRouter(FailingLLMClient(), debug=True) # Test error handling test_messages = [ "", # Empty message "x" * 1001, # Too long message "Normal message" # Should fall back gracefully ] for message in test_messages: print(f"\nTesting error handling for: '{message[:20]}{'...' if len(message) > 20 else ''}'") try: result = failing_router.route(message) print(f"Result: {result['intent']} (Router: {result['router_used']})") except Exception as e: print(f"Error handled: {e}") if __name__ == "__main__": # Run all demonstrations demonstrate_integration() demonstrate_real_integration() demonstrate_error_handling() print("\n\n🎯 Summary") print("=" * 20) print("βœ… LLMFallbackRouter successfully created") print("βœ… Two-tier routing system demonstrated") print("βœ… Error handling validated") print("βœ… Integration path defined") print("\nThe LLMFallbackRouter is ready for integration into VoucherBot!")