diff --git a/README.md b/README.md index 12b7a21..8c94f62 100644 --- a/README.md +++ b/README.md @@ -27,10 +27,15 @@ The Lex Helper library is an extensive collection of functions and classes that - [Core Features](#core-features) - [Dialog Utilities](#dialog-utilities) - [Message Management](#message-management) + - [Smart Disambiguation](#smart-disambiguation) - [Bedrock Integration](#bedrock-integration) - [Bedrock Usage Examples](#bedrock-usage-examples) - [Basic InvokeModel API](#basic-invokemodel-api) - [Converse API with System Prompt](#converse-api-with-system-prompt) +- [Smart Disambiguation](#smart-disambiguation-1) + - [Basic Setup](#basic-setup) + - [AI-Powered Disambiguation](#ai-powered-disambiguation) + - [Example Interaction](#example-interaction) - [Examples](#examples) - [Documentation](#documentation) - [Development Setup](#development-setup) @@ -56,6 +61,8 @@ The Lex Helper library is an extensive collection of functions and classes that - **Reduced Boilerplate**: Common Lex operations like transitioning between intents, handling dialog states, and managing session attributes are simplified into clean, intuitive methods. +- **Smart Disambiguation**: Automatically handle ambiguous user input with intelligent clarification prompts. Optional AI-powered responses using Amazon Bedrock create natural, contextual disambiguation messages that improve user experience. + - **Developer Experience**: Get the benefits of modern Python features like type hints, making your code more maintainable and easier to understand. Full IDE support means better autocomplete and fewer runtime errors. ## Installation @@ -171,6 +178,13 @@ your_project/ - Supports `messages_{localeId}.yaml` files (e.g., `messages_en_US.yaml`, `messages_es_ES.yaml`) - Automatic fallback to `messages.yaml` for missing locales +### Smart Disambiguation +- **Intelligent Intent Resolution**: Automatically detects ambiguous user input and presents clarifying options +- **AI-Powered Responses**: Optional Bedrock integration for contextual, natural language disambiguation messages +- **Configurable Thresholds**: Fine-tune when disambiguation triggers based on confidence scores and similarity +- **Multi-Selection Support**: Users can choose via text, numbers, letters, or button clicks +- **Graceful Fallbacks**: Seamless fallback to static messages if AI services are unavailable + ### Bedrock Integration - **invoke_bedrock**: Direct integration with Amazon Bedrock models - Supports multiple model families (Claude, Titan, Jurassic, Cohere, Llama) @@ -216,6 +230,60 @@ response = invoke_bedrock( print(response['text']) ``` +## Smart Disambiguation + +Handle ambiguous user input intelligently with automatic clarification prompts: + +### Basic Setup +```python +from lex_helper import Config, LexHelper +from lex_helper.core.disambiguation.types import DisambiguationConfig + +# Enable disambiguation with default settings +config = Config( + session_attributes=CustomSessionAttributes(), + enable_disambiguation=True, + disambiguation_config=DisambiguationConfig( + confidence_threshold=0.5, # Trigger when confidence < 50% + max_candidates=2, # Show up to 2 options + ) +) +``` + +### AI-Powered Disambiguation +```python +from lex_helper.core.disambiguation.types import BedrockDisambiguationConfig + +# Enable Bedrock for intelligent, contextual responses +bedrock_config = BedrockDisambiguationConfig( + enabled=True, + model_id="anthropic.claude-3-haiku-20240307-v1:0", + system_prompt="You are a helpful assistant that creates clear, " + "friendly disambiguation messages for users." +) + +disambiguation_config = DisambiguationConfig( + confidence_threshold=0.5, + bedrock_config=bedrock_config, # AI-powered responses +) +``` + +### Example Interaction +``` +User: "I need help with my booking" + +Static Response: +"I can help you with several things. What would you like to do?" +Buttons: ["Book Flight", "Change Flight", "Cancel Flight"] + +AI-Powered Response: +"I'd be happy to help with your booking! Are you looking to make +changes to an existing reservation or book a new flight?" +Buttons: ["Modify existing booking", "Book new flight"] +``` + +For detailed configuration options, see [Smart Disambiguation Documentation](docs/smart-disambiguation.md). + ## Examples - **Basic Example**: See `examples/basic_handler/` for a simple implementation diff --git a/docs/smart-disambiguation.md b/docs/smart-disambiguation.md new file mode 100644 index 0000000..f6979c4 --- /dev/null +++ b/docs/smart-disambiguation.md @@ -0,0 +1,613 @@ +# Smart Disambiguation for lex-helper + +## Overview + +Smart Disambiguation is an intelligent feature in lex-helper that helps resolve ambiguous user input by presenting clear options when Amazon Lex cannot confidently determine the user's intent. Instead of falling back to "I didn't understand," the system analyzes confidence scores and presents relevant choices to guide users to their desired outcome. + +## Table of Contents + +- [How It Works](#how-it-works) +- [When Disambiguation Triggers](#when-disambiguation-triggers) +- [Configuration](#configuration) +- [Message Localization](#message-localization) +- [Integration](#integration) +- [Examples](#examples) +- [Best Practices](#best-practices) +- [Troubleshooting](#troubleshooting) + +## How It Works + +### The Problem + +Traditional chatbots often respond with generic fallback messages when they can't determine user intent: + +``` +User: "I need help with my booking" +Bot: "I didn't understand that. Could you please rephrase your request?" +``` + +### The Solution + +Smart Disambiguation analyzes Lex's confidence scores and presents clear options: + +``` +User: "I need help with my booking" +Bot: "I can help you with a couple of things. Which would you like to do?" + [Book a Flight] [Change Flight] +``` + +### Architecture + +The disambiguation system consists of three main components: + +1. **DisambiguationAnalyzer** - Analyzes confidence scores and determines when to disambiguate +2. **DisambiguationHandler** - Generates user-friendly clarification responses +3. **Handler Pipeline Integration** - Seamlessly integrates with existing lex-helper flow + +## When Disambiguation Triggers + +Disambiguation triggers in two scenarios: + +### 1. Low Confidence Scenario +When the top intent has low confidence and multiple candidates exist: +```python +# Example: All intents have low confidence +{ + "TrackBaggage": 0.25, + "ChangeFlight": 0.23, + "CancelFlight": 0.19 +} +# Result: Triggers disambiguation +``` + +### 2. Close Scores Scenario +When multiple intents have similar confidence scores: +```python +# Example: Two intents are very close +{ + "BookFlight": 0.45, + "ChangeFlight": 0.42, + "CancelFlight": 0.10 +} +# Result: Triggers disambiguation between BookFlight and ChangeFlight +``` + +### When It Doesn't Trigger +When there's a clear winner: +```python +# Example: Clear winner +{ + "TrackBaggage": 0.75, + "ChangeFlight": 0.15, + "Authenticate": 0.10 +} +# Result: Proceeds directly to TrackBaggage +``` + +## Configuration + +### Basic Configuration + +Enable disambiguation with default settings: + +```python +from lex_helper import Config, LexHelper +from lex_helper.core.disambiguation.types import DisambiguationConfig + +config = Config( + session_attributes=MySessionAttributes(), + enable_disambiguation=True # Enable with defaults +) + +lex_helper = LexHelper(config=config) +``` + +### Advanced Configuration + +Customize disambiguation behavior: + +```python +disambiguation_config = DisambiguationConfig( + # Core thresholds + confidence_threshold=0.4, # Trigger when top score < 0.4 + similarity_threshold=0.15, # Trigger when top scores within 0.15 + max_candidates=2, # Show max 2 options + min_candidates=2, # Need at least 2 candidates + + # Intent groupings for better messages + custom_intent_groups={ + "booking": ["BookFlight", "ChangeFlight", "CancelFlight"], + "status": ["FlightDelayUpdate", "TrackBaggage"], + "account": ["Authenticate"] + }, + + # Custom message keys for localization + custom_messages={ + # Direct intent pair mappings + "BookFlight_ChangeFlight": "disambiguation.airline.book_or_change", + + # Intent group mappings + "disambiguation.booking": "disambiguation.airline.booking_options", + + # General mappings + "disambiguation.two_options": "disambiguation.airline.two_options" + } +) + +config = Config( + session_attributes=MySessionAttributes(), + enable_disambiguation=True, + disambiguation_config=disambiguation_config +) +``` + +### Bedrock-Powered Disambiguation + +For even more intelligent and contextual disambiguation, enable Amazon Bedrock integration: + +```python +from lex_helper.core.disambiguation.types import ( + BedrockDisambiguationConfig, + DisambiguationConfig +) + +# Configure Bedrock for intelligent text generation +bedrock_config = BedrockDisambiguationConfig( + enabled=True, + model_id="anthropic.claude-3-haiku-20240307-v1:0", + region_name="us-east-1", + max_tokens=150, + temperature=0.3, + system_prompt=( + "You are a helpful assistant that creates clear, concise " + "disambiguation messages for chatbot users. Be friendly and natural." + ), + fallback_to_static=True, # Graceful fallback if Bedrock fails +) + +# Configure disambiguation with Bedrock +disambiguation_config = DisambiguationConfig( + confidence_threshold=0.5, + max_candidates=2, + bedrock_config=bedrock_config, # Enable Bedrock integration +) +``` + +**Benefits of Bedrock Integration:** +- **Contextual messages**: Acknowledges user's specific input +- **Natural language**: More conversational than static templates +- **Smart button labels**: Generates intuitive action text +- **Adaptive responses**: Tailored to your domain and use case + +**Example comparison:** + +*Static disambiguation:* +``` +User: "I need help with my flight" +Bot: "I can help you with several things. What would you like to do?" +Buttons: ["Book Flight", "Change Flight", "Cancel Flight"] +``` + +*Bedrock-powered disambiguation:* +``` +User: "I need help with my flight" +Bot: "I'd be happy to help with your flight! Are you looking to make changes to an existing booking or book a new flight?" +Buttons: ["Modify existing booking", "Book new flight"] +``` + +### Configuration Parameters + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `confidence_threshold` | float | 0.6 | Minimum confidence to avoid disambiguation | +| `similarity_threshold` | float | 0.15 | Max difference between top scores to trigger | +| `max_candidates` | int | 3 | Maximum options to show users | +| `min_candidates` | int | 2 | Minimum candidates needed to trigger | +| `custom_intent_groups` | dict | {} | Related intent groupings | +| `custom_messages` | dict | {} | Custom message key mappings | +| `bedrock_config` | BedrockDisambiguationConfig | disabled | Bedrock integration settings | + +#### Bedrock Configuration Parameters + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `enabled` | bool | False | Enable Bedrock text generation | +| `model_id` | str | claude-3-haiku | Bedrock model to use | +| `region_name` | str | us-east-1 | AWS region for Bedrock | +| `max_tokens` | int | 200 | Maximum tokens for responses | +| `temperature` | float | 0.3 | Randomness (0.0-1.0, lower = more deterministic) | +| `system_prompt` | str | default | System prompt for the model | +| `fallback_to_static` | bool | True | Fall back to static messages if Bedrock fails | + +## Message Localization + +### Message Key Mapping + +The system uses a hierarchical approach to find the right message: + +#### 1. Direct Intent Pair Mapping (Highest Priority) +```python +# For intents: ["BookFlight", "ChangeFlight"] +# Creates key: "BookFlight_ChangeFlight" (alphabetically sorted) +custom_messages = { + "BookFlight_ChangeFlight": "disambiguation.airline.book_or_change" +} +``` + +#### 2. Intent Group Mapping (Medium Priority) +```python +# For intents in the "booking" group +# Creates key: "disambiguation.booking" +custom_messages = { + "disambiguation.booking": "disambiguation.airline.booking_options" +} +``` + +#### 3. Default Mapping (Lowest Priority) +```python +# Falls back to default keys +custom_messages = { + "disambiguation.two_options": "disambiguation.airline.two_options" +} +``` + +### Message Files + +Add disambiguation messages to your localization files: + +**messages_en_US.yaml:** +```yaml +disambiguation: + # Default messages + two_options: "I can help you with two things. Which would you like to do?" + multiple_options: "I can help you with several things. What would you like to do?" + + # Custom airline messages + airline: + two_options: "I can help you with a couple of things. Which would you like to do?" + booking_options: "I can help you with flight bookings. Would you like to book, change, or cancel a flight?" + book_or_change: "Would you like to book a new flight or change an existing one?" +``` + +**messages_es_ES.yaml:** +```yaml +disambiguation: + two_options: "Puedo ayudarte con dos cosas. ¿Cuál te gustaría hacer?" + multiple_options: "Puedo ayudarte con varias cosas. ¿Qué te gustaría hacer?" + + airline: + two_options: "Puedo ayudarte con un par de cosas. ¿Cuál te gustaría hacer?" + booking_options: "Puedo ayudarte con reservas de vuelos. ¿Te gustaría reservar, cambiar o cancelar un vuelo?" + book_or_change: "¿Te gustaría reservar un nuevo vuelo o cambiar uno existente?" +``` + +## Integration + +### Handler Pipeline Integration + +Disambiguation integrates seamlessly into the lex-helper pipeline: + +```python +# When disambiguation is enabled, the handler pipeline becomes: +handlers = [ + disambiguation_intent_handler, # Added automatically + regular_intent_handler # Existing handler +] +``` + +### Processing Flow + +1. **Request Analysis** - Analyzer examines Lex confidence scores +2. **Disambiguation Decision** - Determines if disambiguation is needed +3. **Response Generation** - Creates user-friendly options with buttons +4. **User Selection** - Processes user's choice and routes to correct intent + +### Conversation Flow Example + +``` +User: "I need help with my booking" + +Bot: "I can help you with a couple of things. Which would you like to do?" + [Book Flight] [Change Flight] + +User clicks [Change Flight] +→ User input appears as: "Change Flight" (not "ChangeFlight") + +Bot: "What is your reservation number?" +``` + +This natural conversation flow ensures users see human-readable text throughout their interaction. + +### Response Format + +Disambiguation responses include both text and interactive buttons: + +```json +{ + "messages": [ + { + "content": "I can help you with a couple of things. Which would you like to do?", + "contentType": "PlainText" + }, + { + "contentType": "ImageResponseCard", + "imageResponseCard": { + "title": "Please choose an option:", + "subtitle": "Select what you'd like to do", + "buttons": [ + {"text": "Track Baggage", "value": "Track Baggage"}, + {"text": "Change Flight", "value": "Change Flight"} + ] + } + } + ] +} +``` + +**Note**: Button values use human-readable display names (e.g., "Track Baggage") rather than technical intent names (e.g., "TrackBaggage"). This ensures that when users click buttons, they see natural language as their input, creating a more conversational experience. +``` + +## Examples + +### Example 1: Basic Setup + +```python +# Minimal setup - just enable disambiguation +config = Config( + session_attributes=MySessionAttributes(), + enable_disambiguation=True +) + +lex_helper = LexHelper(config=config) +``` + +### Example 2: Airline Bot Setup (Complete Implementation) + +The `examples/sample_airline_bot/` directory contains a complete working example with both static and Bedrock-powered disambiguation: + +```python +# Environment-based configuration for flexibility +enable_bedrock = os.getenv("ENABLE_BEDROCK_DISAMBIGUATION", "false").lower() == "true" + +# Bedrock configuration (optional) +bedrock_config = BedrockDisambiguationConfig( + enabled=enable_bedrock, + model_id="anthropic.claude-3-haiku-20240307-v1:0", + system_prompt="You are a helpful airline customer service assistant...", + fallback_to_static=True, +) + +# Full airline bot configuration +disambiguation_config = DisambiguationConfig( + confidence_threshold=0.4, + max_candidates=2, + custom_intent_groups={ + "booking": ["BookFlight", "ChangeFlight", "CancelFlight"], + "status": ["FlightDelayUpdate", "TrackBaggage"] + }, + custom_messages={ + "disambiguation.booking": "disambiguation.airline.booking_options", + "disambiguation.status": "disambiguation.airline.status_options", + "BookFlight_ChangeFlight": "disambiguation.airline.book_or_change" + }, + bedrock_config=bedrock_config, # AI-powered enhancement +) + +config = Config( + session_attributes=AirlineBotSessionAttributes(), + package_name="fulfillment_function", + enable_disambiguation=True, + disambiguation_config=disambiguation_config +) +``` + +**Usage:** +```bash +# Static disambiguation +python lambda_function.py + +# Bedrock-powered disambiguation +ENABLE_BEDROCK_DISAMBIGUATION=true python lambda_function.py +``` + +### Example 3: E-commerce Bot Setup + +```python +disambiguation_config = DisambiguationConfig( + confidence_threshold=0.5, + custom_intent_groups={ + "shopping": ["SearchProducts", "AddToCart", "Checkout"], + "account": ["Login", "Register", "ViewOrders"], + "support": ["ContactSupport", "ReturnItem", "TrackOrder"] + }, + custom_messages={ + "disambiguation.shopping": "ecommerce.shopping_options", + "disambiguation.account": "ecommerce.account_options", + "SearchProducts_AddToCart": "ecommerce.search_or_add" + } +) +``` + +## Best Practices + +### 1. Threshold Configuration + +- **Conservative (0.6-0.8)**: Only disambiguate when really unsure +- **Moderate (0.4-0.5)**: Good balance for most bots +- **Aggressive (0.2-0.3)**: Catch more ambiguous cases + +### 2. Intent Grouping + +Group related intents for better user experience: + +```python +# Good grouping - related functionality +custom_intent_groups = { + "booking": ["BookFlight", "ChangeFlight", "CancelFlight"], + "status": ["FlightStatus", "BaggageStatus"] +} + +# Avoid - unrelated intents +custom_intent_groups = { + "mixed": ["BookFlight", "Weather", "Authenticate"] # Don't do this +} +``` + +### 3. Message Design + +- Keep messages **concise and clear** +- Use **action-oriented language** +- Provide **specific options** rather than generic choices +- Test with **real user scenarios** + +### 4. Localization + +- Always use **message keys** instead of hardcoded text +- Provide translations for **all supported locales** +- Test disambiguation in **each language** +- Consider **cultural differences** in phrasing + +### 5. Testing + +Test disambiguation with various scenarios: + +```python +# Test cases to verify +test_cases = [ + # Low confidence scenario + {"TrackBaggage": 0.25, "ChangeFlight": 0.23}, + + # Close scores scenario + {"BookFlight": 0.45, "ChangeFlight": 0.42}, + + # Clear winner (should not disambiguate) + {"TrackBaggage": 0.75, "ChangeFlight": 0.15}, + + # Single candidate (should not disambiguate) + {"BookFlight": 0.30} +] +``` + +## Troubleshooting + +### Common Issues + +#### 1. Disambiguation Not Triggering + +**Problem**: Expected disambiguation but got fallback instead. + +**Solutions**: +- Check if `enable_disambiguation=True` in config +- Verify confidence scores are within triggering range +- Ensure minimum candidates requirement is met +- Check if disambiguation components are properly imported + +#### 2. Wrong Message Displayed + +**Problem**: Generic message instead of custom message. + +**Solutions**: +- Verify message keys exist in localization files +- Check custom_messages mapping is correct +- Ensure intent names match exactly (case-sensitive) +- Verify locale is set correctly + +#### 3. Disambiguation Triggering Too Often + +**Problem**: Disambiguation shows for clear intent matches. + +**Solutions**: +- Increase `confidence_threshold` (try 0.6 instead of 0.4) +- Increase `similarity_threshold` (try 0.2 instead of 0.15) +- Check Lex training data quality +- Review intent utterance overlap + + + +### Debug Information + +Enable detailed logging to troubleshoot: + +```python +disambiguation_config = DisambiguationConfig( + enable_logging=True, # Enable detailed logs + # ... other config +) +``` + +Check logs for: +- Confidence scores extracted from Lex +- Disambiguation decision reasoning +- Message key resolution + +### Performance Considerations + +- Disambiguation adds minimal latency (~10-50ms) +- Message localization is cached by lex-helper +- Button rendering is handled by Lex UI + +## Migration Guide + +### From Regular lex-helper + +1. **Update imports**: +```python +from lex_helper.core.disambiguation.types import DisambiguationConfig +``` + +2. **Add configuration**: +```python +config = Config( + # ... existing config + enable_disambiguation=True +) +``` + +3. **Add message keys** to localization files + +4. **Test thoroughly** with existing intents + +### Backward Compatibility + +- Disambiguation is **disabled by default** +- Existing code works **without changes** +- No breaking changes to existing APIs +- Graceful fallback if disambiguation components unavailable + +## API Reference + +### DisambiguationConfig + +```python +@dataclass +class DisambiguationConfig: + confidence_threshold: float = 0.6 + max_candidates: int = 3 + fallback_to_original: bool = True + min_candidates: int = 2 + similarity_threshold: float = 0.15 + enable_logging: bool = True + custom_intent_groups: dict[str, list[str]] = field(default_factory=dict) + custom_messages: dict[str, str] = field(default_factory=dict) +``` + +### Config Integration + +```python +class Config: + # ... existing fields + enable_disambiguation: bool = False + disambiguation_config: DisambiguationConfig | None = None +``` + +## Conclusion + +Smart Disambiguation transforms ambiguous user interactions into clear, actionable choices. By leveraging Lex's confidence scores, it significantly improves user experience while maintaining the simplicity and power of lex-helper. + +The feature is designed to be: +- **Easy to integrate** - Just set `enable_disambiguation=True` +- **Highly configurable** - Customize thresholds, messages, and behavior +- **Fully localized** - Support for multiple languages out of the box +- **Backward compatible** - No impact on existing implementations + +Start with the basic configuration and gradually customize based on your bot's specific needs and user feedback. diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/pyproject.toml b/examples/sample_airline_bot/lambdas/fulfillment_function/pyproject.toml index 84a9fba..b42b36e 100644 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/pyproject.toml +++ b/examples/sample_airline_bot/lambdas/fulfillment_function/pyproject.toml @@ -35,4 +35,4 @@ include = [ allow-direct-references = true [tool.uv.sources] -lex-helper = { path = "lex_helper-0.0.1-py3-none-any.whl" } +lex-helper = { path = "lex_helper-0.0.2-py3-none-any.whl" } diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/config_test_script.py b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/config_test_script.py deleted file mode 100644 index b69d84d..0000000 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/config_test_script.py +++ /dev/null @@ -1,75 +0,0 @@ -# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. -# SPDX-License-Identifier: Apache-2.0 -""" -Test script for the configuration utilities. - -This script tests the configuration utilities to ensure they work correctly -in both local development and simulated Lambda environments. -""" - -import logging -import os -import sys - -# Configure logging -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") -logger = logging.getLogger(__name__) - -# Add the current directory to the Python path -sys.path.append(os.path.dirname(os.path.abspath(__file__))) - -# Import our configuration utility -from utils.config import initialize_message_manager - - -def test_message_manager_initialization(): - """Test MessageManager initialization.""" - logger.info("Testing MessageManager initialization...") - - # Mock LexRequest object - class MockBot: - localeId = "en_US" - - class MockLexRequest: - bot = MockBot() - - try: - # Test initialization - mock_request = MockLexRequest() - initialize_message_manager(mock_request) - logger.info("MessageManager initialized successfully") - - # Test with different locale - mock_request.bot.localeId = "it_IT" - initialize_message_manager(mock_request) - logger.info("MessageManager initialized with Italian locale") - - except Exception as e: - logger.error("MessageManager initialization failed: %s", e) - raise - - -def test_messages_directory(): - """Test that messages directory exists.""" - current_dir = os.path.dirname(os.path.abspath(__file__)) - messages_dir = os.path.join(current_dir, "messages") - - if os.path.exists(messages_dir): - logger.info("Messages directory exists at %s", messages_dir) - - # Check for message files - for locale in ["en_US", "it_IT"]: - msg_file = os.path.join(messages_dir, f"messages_{locale}.yaml") - if os.path.exists(msg_file): - logger.info("Found message file: %s", msg_file) - else: - logger.warning("Message file not found: %s", msg_file) - else: - logger.warning(f"Messages directory does not exist: {messages_dir}") - - -if __name__ == "__main__": - logger.info("Testing configuration utilities...") - test_message_manager_initialization() - test_messages_directory() - logger.info("All tests passed!") diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/lambda_function.py b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/lambda_function.py index 58ea4ab..7ef2a0e 100644 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/lambda_function.py +++ b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/lambda_function.py @@ -4,7 +4,15 @@ Main Lambda handler for the Airline-Bot fulfillment function. This is the entry point for the AWS Lambda function that handles Amazon Lex bot requests. -It uses the lex_helper framework to simplify request processing and intent routing. +It uses the lex_helper framework with Smart Disambiguation to provide intelligent handling +of ambiguous user input, with optional Bedrock-powered contextual responses. + +Environment Variables: + ENABLE_BEDROCK_DISAMBIGUATION: Set to 'true' to enable AI-powered disambiguation + BEDROCK_MODEL_ID: Bedrock model ID (default: anthropic.claude-3-haiku-20240307-v1:0) + BEDROCK_REGION: AWS region for Bedrock (default: us-east-1) + BEDROCK_MAX_TOKENS: Maximum tokens for responses (default: 150) + BEDROCK_TEMPERATURE: Temperature for text generation (default: 0.3) """ import json @@ -36,11 +44,20 @@ # Configure logging for Lambda environment logger = logging.getLogger(__name__) +# Configure logging level from environment variable +log_level = os.getenv("LOG_LEVEL", "INFO").upper() +numeric_level = getattr(logging, log_level, logging.INFO) + # Configure basic logging for Lambda if not already configured if not logger.handlers: - logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") + logging.basicConfig(level=numeric_level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") + +# Also set the root logger level to ensure all lex_helper logs are captured +logging.getLogger("lex_helper").setLevel(numeric_level) +logging.getLogger().setLevel(numeric_level) from lex_helper import Config, LexHelper +from lex_helper.core.disambiguation.types import BedrockDisambiguationConfig, DisambiguationConfig # Use absolute import instead of relative import for Lambda compatibility try: @@ -65,23 +82,94 @@ def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]: Returns: Dict[str, Any]: The Lex response formatted for Amazon Lex service """ - logger.debug("Initializing Airline-Bot fulfillment Lambda") + logger.info("🚀 Initializing Airline-Bot fulfillment Lambda") + logger.debug( + "📊 Environment variables: LOG_LEVEL=%s, ENABLE_BEDROCK_DISAMBIGUATION=%s", + os.getenv("LOG_LEVEL", "INFO"), + os.getenv("ENABLE_BEDROCK_DISAMBIGUATION", "false"), + ) # Initialize the session attributes with default values session_attributes = AirlineBotSessionAttributes() - logger.debug("Initialized session attributes") + logger.debug("✅ Initialized session attributes") + + # Configure Bedrock for intelligent disambiguation (optional) + # Set ENABLE_BEDROCK_DISAMBIGUATION=true environment variable to enable + enable_bedrock = os.getenv("ENABLE_BEDROCK_DISAMBIGUATION", "false").lower() == "true" + + bedrock_config = BedrockDisambiguationConfig( + enabled=enable_bedrock, + model_id=os.getenv("BEDROCK_MODEL_ID", "anthropic.claude-3-haiku-20240307-v1:0"), + region_name=os.getenv("BEDROCK_REGION", "us-east-1"), + max_tokens=int(os.getenv("BEDROCK_MAX_TOKENS", "150")), + temperature=float(os.getenv("BEDROCK_TEMPERATURE", "0.3")), + system_prompt=( + "You are a helpful airline customer service assistant. " + "Create clear, friendly disambiguation messages that help travelers " + "choose between flight-related options. Be concise and professional. " + "Use airline industry terminology appropriately. " + "Always acknowledge what the customer said and provide clear next steps." + ), + fallback_to_static=True, # Always fall back gracefully + ) - # Create the lex_helper configuration with automatic error handling + # Configure Smart Disambiguation with Bedrock integration and message keys for localization + disambiguation_config = DisambiguationConfig( + confidence_threshold=0.4, # Threshold for low confidence scenarios + max_candidates=2, # Keep it simple with 2 options + similarity_threshold=0.15, # Only trigger if top scores are within 0.15 of each other + # Define custom intent groups for related airline operations + custom_intent_groups={ + "booking": ["BookFlight", "ChangeFlight", "CancelFlight"], + "status": ["FlightDelayUpdate", "TrackBaggage"], + "account": ["Authenticate"], + }, + # Use message keys instead of hardcoded text for localization + custom_messages={ + # General disambiguation messages (these are message keys) + "disambiguation.two_options": "disambiguation.airline.two_options", + "disambiguation.multiple_options": "disambiguation.airline.multiple_options", + # Specific intent group messages + "disambiguation.booking": "disambiguation.airline.booking_options", + "disambiguation.status": "disambiguation.airline.status_options", + # Specific intent pair messages + "BookFlight_ChangeFlight": "disambiguation.airline.book_or_change", + "ChangeFlight_CancelFlight": "disambiguation.airline.change_or_cancel", + "FlightDelayUpdate_TrackBaggage": "disambiguation.airline.flight_or_baggage", + }, + # Bedrock configuration for intelligent text generation + bedrock_config=bedrock_config, + ) + + # Create the lex_helper configuration with disambiguation enabled config = Config( session_attributes=session_attributes, package_name="fulfillment_function", auto_handle_exceptions=True, # Automatically handle exceptions error_message="general.error_generic", # Custom error message key + enable_disambiguation=True, # Enable Smart Disambiguation + disambiguation_config=disambiguation_config, ) # Initialize the LexHelper with our configuration lex_helper = LexHelper(config=config) - logger.debug("Initialized LexHelper") + + if enable_bedrock: + logger.info("🤖 Bedrock-powered disambiguation enabled with model: %s", bedrock_config.model_id) + logger.info("🌍 Bedrock region: %s", bedrock_config.region_name) + logger.info( + "🎛️ Bedrock settings: max_tokens=%d, temperature=%.1f", bedrock_config.max_tokens, bedrock_config.temperature + ) + else: + logger.info("📝 Using static disambiguation messages (Bedrock disabled)") + + logger.info( + "🎯 Disambiguation config: threshold=%.2f, max_candidates=%d", + disambiguation_config.confidence_threshold, + disambiguation_config.max_candidates, + ) + + logger.debug("Initialized LexHelper with Smart Disambiguation") # Process the Lex request through the framework (exceptions handled automatically) response = lex_helper.handler(event, context) @@ -91,3 +179,21 @@ def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]: logger.debug("Response: %s", json.dumps(response, default=str)) return response + + +# Usage Examples: +# +# 1. Basic disambiguation (default): +# No environment variables needed. Uses static message templates. +# +# 2. Bedrock-powered disambiguation: +# Set environment variable: ENABLE_BEDROCK_DISAMBIGUATION=true +# Optionally configure: BEDROCK_MODEL_ID, BEDROCK_REGION, etc. +# +# 3. Custom Bedrock model: +# ENABLE_BEDROCK_DISAMBIGUATION=true +# BEDROCK_MODEL_ID=anthropic.claude-3-sonnet-20240229-v1:0 +# BEDROCK_TEMPERATURE=0.2 +# +# The system gracefully falls back to static messages if Bedrock fails, +# ensuring your bot always works even if AI services are unavailable. diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/message_manager_test_script.py b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/message_manager_test_script.py deleted file mode 100644 index d73e83e..0000000 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/message_manager_test_script.py +++ /dev/null @@ -1,79 +0,0 @@ -# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. -# SPDX-License-Identifier: Apache-2.0 -#!/usr/bin/env python3 -""" -Test script to verify MessageManager initialization and message retrieval. -""" - -import os -import sys - -# Add the layer path to Python path for local development -if not os.getenv("AWS_EXECUTION_ENV"): - project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - layer_path = os.path.join(project_root, "layers", "lex_helper", "python") - if os.path.exists(layer_path): - sys.path.append(layer_path) - - for item in os.listdir(layer_path): - item_path = os.path.join(layer_path, item) - if os.path.isdir(item_path) and item.startswith("lex_helper"): - sys.path.append(item_path) - - -def test_message_manager(): - """Test MessageManager with mock LexRequest.""" - # Change to the fulfillment_function directory - original_cwd = os.getcwd() - fulfillment_dir = os.path.join(original_cwd, "lambdas", "fulfillment_function") - os.chdir(fulfillment_dir) - - print(f"Changed working directory to: {os.getcwd()}") - print(f"Messages directory exists: {os.path.exists('messages')}") - - if os.path.exists("messages/messages_en_US.yaml"): - print("Found messages_en_US.yaml") - else: - print("messages_en_US.yaml not found") - - try: - # Try to import without lex_helper first - print("Testing basic YAML loading...") - import yaml - - with open("messages/messages_en_US.yaml") as f: - messages = yaml.safe_load(f) - print( - f"YAML loaded successfully: {messages.get('track_baggage', {}).get('elicit_reservation_number', 'KEY NOT FOUND')}" - ) - except Exception as e: - print(f"YAML test failed: {e}") - - # Restore original directory - os.chdir(original_cwd) - - try: - from utils.config import initialize_message_manager - - # Mock LexRequest object - class MockBot: - localeId = "en_US" - - class MockLexRequest: - bot = MockBot() - - # Initialize MessageManager - mock_request = MockLexRequest() - initialize_message_manager(mock_request) - - print("MessageManager initialized, testing message retrieval...") - - except Exception as e: - print(f"Test failed: {e}") - import traceback - - traceback.print_exc() - - -if __name__ == "__main__": - test_message_manager() diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_en_US.yaml b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_en_US.yaml index 6604d2d..432d595 100644 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_en_US.yaml +++ b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_en_US.yaml @@ -54,3 +54,20 @@ validation: invalid_city: "Please provide a valid city name like 'New York' or 'Los Angeles'." invalid_date: "Please provide a valid date in the format MM/DD/YYYY." invalid_passenger_count: "Please provide a valid number of passengers (1-9)." + +# Smart Disambiguation Messages +disambiguation: + # Default disambiguation messages + two_options: "I can help you with either of these. Which would you like to do?" + multiple_options: "I can help you with several things. What would you like to do?" + fallback: "I'm not sure what you're looking for. Could you be more specific?" + + # Airline-specific disambiguation messages + airline: + two_options: "I can help you with a couple of things. Which would you like to do?" + multiple_options: "I can help you with several things. What would you like to do?" + booking_options: "I can help you with flight bookings. Would you like to book, change, or cancel a flight?" + status_options: "I can help you check status information. Would you like flight updates or baggage tracking?" + book_or_change: "Would you like to book a new flight or change an existing one?" + change_or_cancel: "Would you like to change your flight or cancel it?" + flight_or_baggage: "Would you like to check flight status or track your baggage?" diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_it_IT.yaml b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_it_IT.yaml index 8a7d7e7..f051102 100644 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_it_IT.yaml +++ b/examples/sample_airline_bot/lambdas/fulfillment_function/src/fulfillment_function/messages/messages_it_IT.yaml @@ -54,3 +54,20 @@ validation: invalid_city: "Fornisca un nome di città valido come 'Roma' o 'Milano'." invalid_date: "Fornisca una data valida nel formato GG/MM/AAAA." invalid_passenger_count: "Fornisca un numero valido di passeggeri (1-9)." + +# Messaggi di Disambiguazione Intelligente +disambiguation: + # Messaggi di disambiguazione predefiniti + two_options: "Posso aiutarla con due cose. Quale desidera fare?" + multiple_options: "Posso aiutarla con diverse cose. Cosa desidera fare?" + fallback: "Non sono sicuro di cosa stia cercando. Potrebbe essere più specifico?" + + # Messaggi di disambiguazione specifici per compagnie aeree + airline: + two_options: "Posso aiutarla con un paio di cose. Quale desidera fare?" + multiple_options: "Posso aiutarla con diverse cose. Cosa desidera fare?" + booking_options: "Posso aiutarla con le prenotazioni di voli. Desidera prenotare, modificare o cancellare un volo?" + status_options: "Posso aiutarla a controllare le informazioni di stato. Desidera aggiornamenti sui voli o tracciamento bagagli?" + book_or_change: "Desidera prenotare un nuovo volo o modificare uno esistente?" + change_or_cancel: "Desidera modificare il suo volo o cancellarlo?" + flight_or_baggage: "Desidera controllare lo stato del volo o tracciare il suo bagaglio?" diff --git a/examples/sample_airline_bot/lambdas/fulfillment_function/uv.lock b/examples/sample_airline_bot/lambdas/fulfillment_function/uv.lock index 83ba742..84119c1 100644 --- a/examples/sample_airline_bot/lambdas/fulfillment_function/uv.lock +++ b/examples/sample_airline_bot/lambdas/fulfillment_function/uv.lock @@ -44,30 +44,30 @@ wheels = [ [[package]] name = "boto3" -version = "1.40.26" +version = "1.40.30" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "botocore" }, { name = "jmespath" }, { name = "s3transfer" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/66/68/40902312de023458edae9bb42c503f2aafda5009079cead1ab693f2350a6/boto3-1.40.26.tar.gz", hash = "sha256:9a71684825cfd4548027f254eadf4dafb7fccc7523f20e2a1cb74033f4d74a6b", size = 111549, upload-time = "2025-09-08T19:51:09.917Z" } +sdist = { url = "https://files.pythonhosted.org/packages/77/a7/3fde131d2431d1801e3f16f1b428cf9b8c6677996716c5286a72eb43ecb7/boto3-1.40.30.tar.gz", hash = "sha256:e95db539c938710917f4cb4fc5915f71b27f2c836d949a1a95df7895d2e9ec8b", size = 111636, upload-time = "2025-09-12T19:23:22.625Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/e5/b7/7b6f803dc62f57ed7c8ac108b7aef36fb27ca0aaccd197e3f83bdf57a68a/boto3-1.40.26-py3-none-any.whl", hash = "sha256:8272deb4b82c4a0faa1231c2cd5c6d267d71ed6265abef545c1d5b7f0aa936d8", size = 139324, upload-time = "2025-09-08T19:51:07.876Z" }, + { url = "https://files.pythonhosted.org/packages/3f/43/f1865e3e2aa91c1aa54db90a82ed17b8c0dc60c354045adf1c2134e5cbd8/boto3-1.40.30-py3-none-any.whl", hash = "sha256:04e89abf61240857bf7dec160e22f097eec68c502509b2bb3c5010a22cb91052", size = 139343, upload-time = "2025-09-12T19:23:20.728Z" }, ] [[package]] name = "botocore" -version = "1.40.26" +version = "1.40.30" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "jmespath" }, { name = "python-dateutil" }, { name = "urllib3" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/b7/d7/e42337570f38405a99fa9f9d5f1379fd52de412b1d4c65351d688c461b5d/botocore-1.40.26.tar.gz", hash = "sha256:f8f46b3978b7c324f4c0bef03505870c4c5240c736bfb63318da091942a29710", size = 14330031, upload-time = "2025-09-08T19:50:58.402Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c5/be/086ff6f031c407540e8226b3a4921dd18a05688224324c2df60457f9bcc0/botocore-1.40.30.tar.gz", hash = "sha256:8a74f77cfe5c519826d22f7613f89544cbb8491a1a49d965031bd997f89a8e3f", size = 14349135, upload-time = "2025-09-12T19:23:12.57Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/a4/8b/1dadb6b391346a811ee44b1f36159c376e536e7851c2c1348b44d718da76/botocore-1.40.26-py3-none-any.whl", hash = "sha256:c3e89787b1a360d0fd30f9066864415df02d54b07691cabc34a6b1a01c3d2549", size = 14003429, upload-time = "2025-09-08T19:50:54.466Z" }, + { url = "https://files.pythonhosted.org/packages/ad/a8/3644f482b7b319f3fda87d4583f7b073c0cdf4a6d1b58e5a92555fe3e2e3/botocore-1.40.30-py3-none-any.whl", hash = "sha256:1d87874ad81234bec3e83f9de13618f67ccdfefd08d6b8babc041cd45007447e", size = 14022003, upload-time = "2025-09-12T19:23:09.163Z" }, ] [[package]] @@ -231,7 +231,7 @@ dependencies = [ requires-dist = [ { name = "aws-lambda-powertools", extras = ["tracer"], specifier = ">=3.5.0,<4.0.0" }, { name = "cryptography", specifier = ">=44.0.0,<45.0.0" }, - { name = "lex-helper", path = "lex_helper-0.0.1-py3-none-any.whl" }, + { name = "lex-helper", path = "lex_helper-0.0.2-py3-none-any.whl" }, { name = "pydantic", specifier = ">=2.10.6,<3.0.0" }, { name = "pydantic-core", specifier = ">=2.28.0,<3.0.0" }, { name = "pydantic-settings", specifier = ">=2.7.1,<3.0.0" }, @@ -258,8 +258,8 @@ wheels = [ [[package]] name = "lex-helper" -version = "0.0.1" -source = { path = "lex_helper-0.0.1-py3-none-any.whl" } +version = "0.0.2" +source = { path = "lex_helper-0.0.2-py3-none-any.whl" } dependencies = [ { name = "boto3" }, { name = "botocore" }, @@ -267,7 +267,7 @@ dependencies = [ { name = "pyyaml" }, ] wheels = [ - { filename = "lex_helper-0.0.1-py3-none-any.whl", hash = "sha256:12a569eca746bcdbc68d8edea6daddd67265de9f8520dbcdc01b83a0af919021" }, + { filename = "lex_helper-0.0.2-py3-none-any.whl", hash = "sha256:fad8ac9d327ab86186cd6989afd1e081a9c559b4ee4058a68ac83d93ebef5335" }, ] [package.metadata] @@ -299,7 +299,7 @@ wheels = [ [[package]] name = "pydantic" -version = "2.11.7" +version = "2.11.9" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "annotated-types" }, @@ -307,9 +307,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "typing-inspection" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/00/dd/4325abf92c39ba8623b5af936ddb36ffcfe0beae70405d456ab1fb2f5b8c/pydantic-2.11.7.tar.gz", hash = "sha256:d989c3c6cb79469287b1569f7447a17848c998458d49ebe294e975b9baf0f0db", size = 788350, upload-time = "2025-06-14T08:33:17.137Z" } +sdist = { url = "https://files.pythonhosted.org/packages/ff/5d/09a551ba512d7ca404d785072700d3f6727a02f6f3c24ecfd081c7cf0aa8/pydantic-2.11.9.tar.gz", hash = "sha256:6b8ffda597a14812a7975c90b82a8a2e777d9257aba3453f973acd3c032a18e2", size = 788495, upload-time = "2025-09-13T11:26:39.325Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/6a/c0/ec2b1c8712ca690e5d61979dee872603e92b8a32f94cc1b72d53beab008a/pydantic-2.11.7-py3-none-any.whl", hash = "sha256:dde5df002701f6de26248661f6835bbe296a47bf73990135c7d07ce741b9623b", size = 444782, upload-time = "2025-06-14T08:33:14.905Z" }, + { url = "https://files.pythonhosted.org/packages/3e/d3/108f2006987c58e76691d5ae5d200dd3e0f532cb4e5fa3560751c3a1feba/pydantic-2.11.9-py3-none-any.whl", hash = "sha256:c42dd626f5cfc1c6950ce6205ea58c93efa406da65f479dcb4029d5934857da2", size = 444855, upload-time = "2025-09-13T11:26:36.909Z" }, ] [[package]] @@ -432,14 +432,14 @@ wheels = [ [[package]] name = "s3transfer" -version = "0.13.1" +version = "0.14.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "botocore" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/6d/05/d52bf1e65044b4e5e27d4e63e8d1579dbdec54fce685908ae09bc3720030/s3transfer-0.13.1.tar.gz", hash = "sha256:c3fdba22ba1bd367922f27ec8032d6a1cf5f10c934fb5d68cf60fd5a23d936cf", size = 150589, upload-time = "2025-07-18T19:22:42.31Z" } +sdist = { url = "https://files.pythonhosted.org/packages/62/74/8d69dcb7a9efe8baa2046891735e5dfe433ad558ae23d9e3c14c633d1d58/s3transfer-0.14.0.tar.gz", hash = "sha256:eff12264e7c8b4985074ccce27a3b38a485bb7f7422cc8046fee9be4983e4125", size = 151547, upload-time = "2025-09-09T19:23:31.089Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/6d/4f/d073e09df851cfa251ef7840007d04db3293a0482ce607d2b993926089be/s3transfer-0.13.1-py3-none-any.whl", hash = "sha256:a981aa7429be23fe6dfc13e80e4020057cbab622b08c0315288758d67cabc724", size = 85308, upload-time = "2025-07-18T19:22:40.947Z" }, + { url = "https://files.pythonhosted.org/packages/48/f0/ae7ca09223a81a1d890b2557186ea015f6e0502e9b8cb8e1813f1d8cfa4e/s3transfer-0.14.0-py3-none-any.whl", hash = "sha256:ea3b790c7077558ed1f02a3072fb3cb992bbbd253392f4b6e9e8976941c7d456", size = 85712, upload-time = "2025-09-09T19:23:30.041Z" }, ] [[package]] diff --git a/examples/sample_airline_bot/lib/lex-at-scale-stack.ts b/examples/sample_airline_bot/lib/lex-at-scale-stack.ts index 94bc96e..2ee2596 100644 --- a/examples/sample_airline_bot/lib/lex-at-scale-stack.ts +++ b/examples/sample_airline_bot/lib/lex-at-scale-stack.ts @@ -320,7 +320,13 @@ function fulfillmentFunction(stack: cdk.Stack): cdk_alpha.PythonFunction { role: fulfillmentLambdaRole, environment: { MESSAGES_YAML_PATH: "/var/task/fulfillment_function/messages", - LOG_LEVEL: "INFO" + LOG_LEVEL: "DEBUG", // Enable debug logging + // Bedrock Disambiguation Configuration + ENABLE_BEDROCK_DISAMBIGUATION: "true", // Enable AI-powered disambiguation + BEDROCK_MODEL_ID: "anthropic.claude-3-haiku-20240307-v1:0", // Fast, cost-effective model + BEDROCK_REGION: stack.region, // Use the same region as the stack + BEDROCK_MAX_TOKENS: "150", // Concise responses + BEDROCK_TEMPERATURE: "0.3", // More deterministic for consistent UX }, bundling: { assetExcludes: ["*.pyc", "__pycache__", '.venv', 'tests'] diff --git a/lex_helper/__init__.py b/lex_helper/__init__.py index 082a852..bd37e5e 100644 --- a/lex_helper/__init__.py +++ b/lex_helper/__init__.py @@ -21,6 +21,13 @@ from lex_helper.channels.lex import LexChannel from lex_helper.channels.sms import SMSChannel from lex_helper.core import dialog +from lex_helper.core.disambiguation import ( + BedrockDisambiguationConfig, + BedrockDisambiguationGenerator, + DisambiguationConfig, + DisambiguationResult, + IntentCandidate, +) from lex_helper.core.handler import Config, LexHelper from lex_helper.core.invoke_bedrock import ( BedrockInvocationError, @@ -55,6 +62,8 @@ __all__ = [ "__version__", + "BedrockDisambiguationConfig", + "BedrockDisambiguationGenerator", "BedrockInvocationError", "Bot", "Button", @@ -62,11 +71,14 @@ "Config", "DialogAction", "dialog", + "DisambiguationConfig", + "DisambiguationResult", "format_for_channel", "get_message", "handle_exceptions", "ImageResponseCard", "Intent", + "IntentCandidate", "Interpretation", "invoke_bedrock", "invoke_bedrock_converse", diff --git a/lex_helper/core/dialog.py b/lex_helper/core/dialog.py index 2a72638..08f01d8 100644 --- a/lex_helper/core/dialog.py +++ b/lex_helper/core/dialog.py @@ -143,7 +143,7 @@ def close[T: SessionAttributes](lex_request: LexRequest[T], messages: LexMessage messages=messages, ) - logger.debug("FF-LAMBDA :: DIALOG-CLOSE") + logger.debug("Dialog closed") return response @@ -168,7 +168,7 @@ def elicit_intent[T: SessionAttributes](messages: LexMessages, lex_request: LexR session_attributes.previous_slot_to_elicit = "" session_attributes.options_provided = get_provided_options(messages) - logger.debug("FF-LAMBDA :: Elicit-Intent") + logger.debug("Elicit-Intent") return LexResponse( sessionState=SessionState( @@ -240,7 +240,7 @@ def delegate[T: SessionAttributes](lex_request: LexRequest[T]) -> LexResponse[T] Returns: LexResponse: The response object to be sent back to Lex. """ - logger.debug("IN DELEGATE") + logger.debug("Delegating") updated_active_contexts = remove_inactive_context(lex_request) lex_request.sessionState.intent.state = "ReadyForFulfillment" @@ -253,8 +253,6 @@ def delegate[T: SessionAttributes](lex_request: LexRequest[T]) -> LexResponse[T] dialogAction=DialogAction(type="Delegate"), ) - logger.debug("FF-LAMBDA :: DELEGATE") - return LexResponse(sessionState=updated_session_state, requestAttributes={}, messages=[]) @@ -278,7 +276,7 @@ def get_provided_options(messages: LexMessages) -> str: if isinstance(message, LexImageResponseCard) for button in message.imageResponseCard.buttons ] - logger.debug("FF-LAMBDA :: OPTS-PVD :: %s", options) + logger.debug("Get provided options :: %s", options) return json.dumps(options, cls=PydanticEncoder) @@ -443,7 +441,7 @@ def set_subslot( # Logging for debugging purposes logger.debug("Setting subslot %s in composite slot %s", subslot_name, composite_slot_name) - logger.debug("RESULTING INTENT: %s", json.dumps(intent, cls=PydanticEncoder)) + logger.debug("Resulting intent: %s", json.dumps(intent, cls=PydanticEncoder)) return intent @@ -556,7 +554,6 @@ def get_request_components[T: SessionAttributes]( active_contexts = get_active_contexts(lex_request) session_attributes = lex_request.sessionState.sessionAttributes invocation_label = get_invocation_label(lex_request) - logger.debug("DLG-UTL :: INV-LBL: %s", invocation_label) return intent, active_contexts, session_attributes, invocation_label @@ -624,17 +621,17 @@ def handle_any_unknown_slot_choice[T: SessionAttributes](lex_request: LexRequest """ intent, _, session_attributes, _ = get_request_components(lex_request) - logger.debug("FF-LAMBDA :: Handle_Any_Unknown_Choice :: %s", session_attributes) + logger.debug("Handle_Any_Unknown_Choice :: %s", session_attributes) intent = get_intent(lex_request) previous_slot_to_elicit = session_attributes.previous_slot_to_elicit logger.debug("Unparsed slot name: " + (previous_slot_to_elicit or "")) slot_name = previous_slot_to_elicit - logger.debug("IDENTIFIER FOR BAD SLOT %s", slot_name) + logger.debug("Identifier for bad slot: %s", slot_name) choice = get_slot(slot_name or "", intent, preference="interpretedValue") - logger.debug("BAD CHOICE IS %s", choice) + logger.debug("Bad choice is %s", choice) if not isinstance(choice, str): logger.debug("Bad slot choice") return unknown_choice_handler(lex_request=lex_request, choice=choice) @@ -712,12 +709,12 @@ def callback_original_intent_handler[T: SessionAttributes]( Returns: LexResponse[T]: _description_ """ - logger.debug("CALLING BACK ORIGINAL HANDLER") + logger.debug("Calling back original handler") callback_event = lex_request.sessionState.sessionAttributes.callback_event callback_handler = lex_request.sessionState.sessionAttributes.callback_handler or "" if not callback_event and not callback_handler: - logger.debug("NO CALLBACK EVENT OR HANDLER") + logger.debug("No callback event or handler") lex_request.sessionState.intent.name = "greeting" return call_handler_for_file("greeting", lex_request) @@ -728,7 +725,7 @@ def callback_original_intent_handler[T: SessionAttributes]( del lex_request.sessionState.sessionAttributes.callback_handler lex_payload: LexRequest[T] = LexRequest(**callback_request) - logger.debug("MERGING SESSION ATTRIBUTES") + logger.debug("Merging session attributes") merged_attrs = lex_payload.sessionState.sessionAttributes.model_dump() merged_attrs.update( {k: v for k, v in lex_request.sessionState.sessionAttributes.model_dump().items() if v is not None} @@ -757,7 +754,7 @@ def reprompt_slot[T: SessionAttributes](lex_request: LexRequest[T]) -> LexRespon Returns: LexResponse: The response object to be sent back to Lex. """ - logger.debug("FFL-DLG :: REPROMPT-SLOT :: START") + logger.debug("Reprompting slot") session_attributes = lex_request.sessionState.sessionAttributes previous_slot_to_elicit = session_attributes.previous_slot_to_elicit @@ -765,7 +762,6 @@ def reprompt_slot[T: SessionAttributes](lex_request: LexRequest[T]) -> LexRespon return delegate(lex_request) logger.debug("Unparsed slot name: " + previous_slot_to_elicit) slot_name = previous_slot_to_elicit - logger.debug("IDENTIFIER 3 %s", slot_name) messages = [] logger.debug("Reprompt-Messages :: %s", messages) @@ -797,18 +793,18 @@ def load_messages(messages: str) -> LexMessages: case _: res.append(msg) - logger.debug("PARSED-PREV-MSG :: %s", res) + logger.debug("Previous Message :: %s", res) return res def parse_req_sess_attrs[T: SessionAttributes](lex_payload: LexRequest[T]) -> LexRequest[T]: - logger.debug("LEX-PAYLOAD: %s", lex_payload.model_dump_json(exclude_none=True)) + logger.debug("Lex-Payload: %s", lex_payload.model_dump_json(exclude_none=True)) # parsing core_data from session-state from 2nd messages channel_string = "" if lex_payload.requestAttributes: - logger.debug("FFL :: CREATING NEW SESS ATTRS") + logger.debug("Creating new session attributes") if "channel" in lex_payload.requestAttributes: channel_string = lex_payload.requestAttributes["channel"] logger.info("User passed in channel: %s", channel_string) @@ -876,7 +872,6 @@ def transition_to_intent[T: SessionAttributes]( if clear_slots: _clear_slots(intent_name=intent_name, lex_request=lex_request, invocation_label=invocation_label) - # logger.debug(f"TRANSITION :: SESS-STATE : {lex_request.sessionState.sessionAttributes}") # Call the intent handler and get its response response = call_handler_for_file(intent_name=intent_name, lex_request=lex_request) diff --git a/lex_helper/core/disambiguation/__init__.py b/lex_helper/core/disambiguation/__init__.py new file mode 100644 index 0000000..a9e4149 --- /dev/null +++ b/lex_helper/core/disambiguation/__init__.py @@ -0,0 +1,30 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Smart Disambiguation module for lex-helper. + +This module provides intelligent handling of ambiguous user input by analyzing +intent confidence scores, considering conversation context, and presenting +targeted clarifying questions to users. +""" + +from .analyzer import DisambiguationAnalyzer +from .bedrock_generator import BedrockDisambiguationGenerator +from .handler import DisambiguationHandler +from .types import ( + BedrockDisambiguationConfig, + DisambiguationConfig, + DisambiguationResult, + IntentCandidate, +) + +__all__ = [ + "BedrockDisambiguationConfig", + "BedrockDisambiguationGenerator", + "DisambiguationAnalyzer", + "DisambiguationHandler", + "DisambiguationConfig", + "DisambiguationResult", + "IntentCandidate", +] diff --git a/lex_helper/core/disambiguation/analyzer.py b/lex_helper/core/disambiguation/analyzer.py new file mode 100644 index 0000000..d238d20 --- /dev/null +++ b/lex_helper/core/disambiguation/analyzer.py @@ -0,0 +1,263 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +DisambiguationAnalyzer for intelligent intent disambiguation. + +This module provides the core analysis functionality for determining when +disambiguation should occur and which intent candidates to present to users. +""" + +import logging +from typing import Any + +from lex_helper.core.types import LexRequest, SessionAttributes + +from .types import ( + DisambiguationConfig, + DisambiguationResult, + IntentCandidate, + IntentScores, +) + +logger = logging.getLogger(__name__) + +# Constants +MIN_MEANINGFUL_SCORE = 0.1 # Minimum score to consider an intent meaningful +MIN_REASONABLE_SCORE = 0.15 # Minimum score to consider for similarity comparison + + +class DisambiguationAnalyzer: + """ + Analyzes user input to determine disambiguation candidates. + + The analyzer uses Lex's NLU confidence scores to identify potential + intent matches and determine when disambiguation should be triggered + based on configurable thresholds. + """ + + def __init__(self, config: DisambiguationConfig | None = None): + """ + Initialize the DisambiguationAnalyzer. + + Args: + config: Configuration options for disambiguation behavior. + If None, uses default configuration. + """ + self.config = config or DisambiguationConfig() + + def analyze_request( + self, + lex_request: LexRequest[SessionAttributes], + ) -> DisambiguationResult: + """ + Analyze Lex request and return disambiguation candidates. + + Args: + lex_request: The Lex request containing interpretations with confidence scores + + Returns: + DisambiguationResult containing analysis results and candidates + """ + logger.debug("Analyzing Lex request for disambiguation: %s", lex_request.inputTranscript) + + # Extract confidence scores from Lex interpretations + confidence_scores = self.extract_intent_scores(lex_request) + + # Determine if disambiguation should occur + should_disambiguate = self.should_disambiguate(confidence_scores, self.config.confidence_threshold) + + # Generate candidates if disambiguation is needed + candidates = [] + if should_disambiguate: + candidates = self._generate_candidates(confidence_scores, lex_request) + + result = DisambiguationResult( + should_disambiguate=should_disambiguate, + candidates=candidates, + confidence_scores=confidence_scores, + ) + + logger.debug( + "Disambiguation analysis complete: should_disambiguate=%s, candidates=%d", should_disambiguate, len(candidates) + ) + + return result + + def extract_intent_scores(self, lex_request: LexRequest[SessionAttributes]) -> IntentScores: + """ + Extract confidence scores from Lex interpretations. + + Args: + lex_request: The Lex request containing interpretations + + Returns: + Dictionary mapping intent names to confidence scores (0.0-1.0) + """ + scores: dict[str, float] = {} + + for interpretation in lex_request.interpretations: + intent_name = interpretation.intent.name + confidence = interpretation.nluConfidence or 0.0 + scores[intent_name] = confidence + + logger.debug("Extracted intent scores from Lex: %s", scores) + return scores + + def should_disambiguate(self, scores: IntentScores, threshold: float) -> bool: + """ + Determine if disambiguation is needed based on confidence scores. + + Args: + scores: Dictionary of intent names to confidence scores + threshold: Minimum confidence threshold to avoid disambiguation + + Returns: + True if disambiguation should be triggered, False otherwise + """ + if not scores: + return False + + # Get the highest scoring intents + sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True) + + # Filter out very low scores + meaningful_scores = [score for _, score in sorted_scores if score > MIN_MEANINGFUL_SCORE] + if len(meaningful_scores) < self.config.min_candidates: + return False + + # Get top scores for comparison + if len(sorted_scores) < 2: + return False + + top_score = sorted_scores[0][1] + second_score = sorted_scores[1][1] + + # Case 1: Top score is very low (below threshold) AND we have multiple candidates + if top_score < threshold and len(meaningful_scores) >= self.config.min_candidates: + return True + + # Case 2: Multiple scores are close to each other (ambiguous case) + # Only disambiguate if the difference between top scores is small + score_difference = top_score - second_score + + # If the top two scores are within similarity_threshold of each other, consider it ambiguous + # AND both scores are reasonably high + similarity_threshold = self.config.similarity_threshold + if ( + score_difference <= similarity_threshold + and top_score >= MIN_REASONABLE_SCORE + and second_score >= MIN_REASONABLE_SCORE + and len(meaningful_scores) >= self.config.min_candidates + ): + return True + + return False + + def _generate_candidates( + self, scores: IntentScores, lex_request: LexRequest[SessionAttributes] + ) -> list[IntentCandidate]: + """ + Generate intent candidates for disambiguation. + + Args: + scores: Intent confidence scores + lex_request: The Lex request containing interpretation details + + Returns: + List of IntentCandidate objects for presentation to user + """ + # Sort intents by score and take top candidates + sorted_intents = sorted(scores.items(), key=lambda x: x[1], reverse=True) + + # Filter to meaningful scores and limit to max candidates + candidates: list[IntentCandidate] = [] + for intent, score in sorted_intents[: self.config.max_candidates]: + if score > MIN_MEANINGFUL_SCORE: + # Find the corresponding interpretation for slot information + interpretation = self._find_interpretation_by_intent(lex_request, intent) + + candidate = IntentCandidate( + intent_name=intent, + confidence_score=score, + display_name=self._get_display_name(intent), + description=self._get_intent_description(intent), + required_slots=self._get_required_slots_from_interpretation(interpretation), + ) + candidates.append(candidate) + + return candidates + + def _find_interpretation_by_intent(self, lex_request: LexRequest[SessionAttributes], intent_name: str): + """ + Find the interpretation that matches the given intent name. + + Args: + lex_request: The Lex request containing interpretations + intent_name: The intent name to find + + Returns: + The matching interpretation or None if not found + """ + for interpretation in lex_request.interpretations: + if interpretation.intent.name == intent_name: + return interpretation + return None + + def _get_display_name(self, intent: str) -> str: + """ + Get user-friendly display name for an intent. + + Converts technical intent names to user-friendly format. + + Args: + intent: Technical intent name + + Returns: + User-friendly display name + """ + # Convert CamelCase and snake_case to readable format + # Replace underscores with spaces and add spaces before capital letters + import re + + # Handle snake_case + readable = intent.replace("_", " ") + + # Handle CamelCase - add space before capital letters + readable = re.sub(r"([a-z])([A-Z])", r"\1 \2", readable) + + # Capitalize each word + return readable.title() + + def _get_intent_description(self, intent: str) -> str: + """ + Get description for an intent. + + Generates a generic description based on the intent name. + + Args: + intent: Intent name + + Returns: + Brief description of what the intent does + """ + # Generate a generic description based on intent name + display_name = self._get_display_name(intent).lower() + return f"Handle requests related to {display_name}" + + def _get_required_slots_from_interpretation(self, interpretation: Any) -> list[str]: + """ + Get required slots from the interpretation. + + Args: + interpretation: The Lex interpretation object + + Returns: + List of slot names from the interpretation + """ + if not interpretation or not interpretation.intent: + return [] + + # Return the slot names from the intent + slot_names: list[str] = list(interpretation.intent.slots.keys()) if interpretation.intent.slots else [] + return slot_names diff --git a/lex_helper/core/disambiguation/bedrock_generator.py b/lex_helper/core/disambiguation/bedrock_generator.py new file mode 100644 index 0000000..ad629bf --- /dev/null +++ b/lex_helper/core/disambiguation/bedrock_generator.py @@ -0,0 +1,230 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Bedrock-powered text generation for Smart Disambiguation. + +This module provides intelligent, contextual disambiguation message generation +using Amazon Bedrock models to create more natural and helpful clarification +responses based on user input and available intent candidates. +""" + +import json +import logging +from typing import Any + +from lex_helper.core.invoke_bedrock import BedrockInvocationError, invoke_bedrock_simple_converse + +from .types import BedrockDisambiguationConfig, IntentCandidate + +logger = logging.getLogger(__name__) + + +class BedrockDisambiguationGenerator: + """ + Generates disambiguation messages using Amazon Bedrock models. + + This class creates contextual, intelligent disambiguation messages by + analyzing user input and available intent candidates, then using Bedrock + to generate natural language clarification text and button labels. + """ + + def __init__(self, config: BedrockDisambiguationConfig): + """ + Initialize the Bedrock disambiguation generator. + + Args: + config: Configuration for Bedrock text generation + """ + self.config = config + + def generate_clarification_message( + self, user_input: str, candidates: list[IntentCandidate], context: dict[str, Any] | None = None + ) -> str: + """ + Generate a contextual clarification message using Bedrock. + + Args: + user_input: The original user input that was ambiguous + candidates: List of intent candidates to choose from + context: Optional context information (session data, etc.) + + Returns: + Generated clarification message text + """ + if not self.config.enabled: + return self._get_fallback_message(candidates) + + try: + prompt = self._build_clarification_prompt(user_input, candidates, context) + + response = invoke_bedrock_simple_converse( + prompt=prompt, + model_id=self.config.model_id, + system_prompt=self.config.system_prompt, + max_tokens=self.config.max_tokens, + temperature=self.config.temperature, + region_name=self.config.region_name, + ) + + generated_text = response["text"].strip() + logger.debug("Generated clarification message: %s", generated_text) + + return generated_text + + except BedrockInvocationError as e: + logger.warning("Bedrock clarification generation failed: %s", e) + if self.config.fallback_to_static: + return self._get_fallback_message(candidates) + raise + except Exception as e: + logger.error("Unexpected error in Bedrock clarification generation: %s", e) + if self.config.fallback_to_static: + return self._get_fallback_message(candidates) + raise + + def generate_button_labels(self, candidates: list[IntentCandidate], user_input: str | None = None) -> list[str]: + """ + Generate improved button labels using Bedrock. + + Args: + candidates: List of intent candidates + user_input: Optional user input for context + + Returns: + List of generated button labels + """ + if not self.config.enabled: + return [candidate.display_name for candidate in candidates] + + try: + prompt = self._build_button_labels_prompt(candidates, user_input) + + response = invoke_bedrock_simple_converse( + prompt=prompt, + model_id=self.config.model_id, + system_prompt=self.config.system_prompt, + max_tokens=self.config.max_tokens, + temperature=self.config.temperature, + region_name=self.config.region_name, + ) + + # Parse the JSON response to get button labels + generated_text = response["text"].strip() + + # Try to parse as JSON first + try: + if generated_text.startswith("[") and generated_text.endswith("]"): + parsed_labels: Any = json.loads(generated_text) + if isinstance(parsed_labels, list) and len(parsed_labels) == len(candidates): + # Ensure all items are strings + labels: list[str] = [str(item) for item in parsed_labels] # type: ignore[misc] + logger.debug("Generated button labels: %s", labels) + return labels + except json.JSONDecodeError: + pass + + # If JSON parsing fails, try to extract labels from text + extracted_labels = self._extract_labels_from_text(generated_text, len(candidates)) + if extracted_labels: + logger.debug("Extracted button labels: %s", extracted_labels) + return extracted_labels + + # Fallback to original display names + logger.warning("Could not parse generated button labels, using fallback") + return [candidate.display_name for candidate in candidates] + + except BedrockInvocationError as e: + logger.warning("Bedrock button label generation failed: %s", e) + if self.config.fallback_to_static: + return [candidate.display_name for candidate in candidates] + raise + except Exception as e: + logger.error("Unexpected error in Bedrock button label generation: %s", e) + if self.config.fallback_to_static: + return [candidate.display_name for candidate in candidates] + raise + + def _build_clarification_prompt( + self, user_input: str, candidates: list[IntentCandidate], context: dict[str, Any] | None = None + ) -> str: + """Build the prompt for clarification message generation.""" + candidate_descriptions = [] + for i, candidate in enumerate(candidates, 1): + candidate_descriptions.append(f"{i}. {candidate.display_name}: {candidate.description}") + + context_info = "" + if context: + # Add relevant context information + if "session_attributes" in context: + context_info = f"\nContext: {context['session_attributes']}" + + prompt = f"""The user said: "{user_input}" + +This input is ambiguous and could match multiple intents. Here are the possible options: + +{chr(10).join(candidate_descriptions)} + +Generate a friendly, natural clarification message that: +1. Acknowledges what the user said +2. Explains that there are multiple ways to help +3. Asks them to choose which option they want +4. Is conversational and helpful (not robotic) +5. Is concise (1-2 sentences maximum) + +Do not include numbered lists or bullet points in your response. Just provide the clarification message text.{context_info}""" + + return prompt + + def _build_button_labels_prompt(self, candidates: list[IntentCandidate], user_input: str | None = None) -> str: + """Build the prompt for button label generation.""" + candidate_info = [] + for candidate in candidates: + candidate_info.append(f"- {candidate.intent_name}: {candidate.description}") + + user_context = "" + if user_input: + user_context = f'\nUser\'s original input: "{user_input}"' + + prompt = f"""Generate improved button labels for these intent options: + +{chr(10).join(candidate_info)}{user_context} + +Create short, clear, action-oriented button labels (2-4 words each) that users would naturally click. +Make them more conversational and user-friendly than the technical intent names. + +Return your response as a JSON array of strings, like: ["Label 1", "Label 2", "Label 3"] + +Only return the JSON array, nothing else.""" + + return prompt + + def _extract_labels_from_text(self, text: str, expected_count: int) -> list[str] | None: + """Extract button labels from generated text if JSON parsing fails.""" + lines = [line.strip() for line in text.split("\n") if line.strip()] + + # Try to find lines that look like labels + labels: list[str] = [] + for line in lines: + # Remove common prefixes/suffixes and clean up + cleaned = line.strip("- •*\"'[](){}").strip() + # Skip lines that look like headers or explanations + if ( + cleaned + and len(cleaned) <= 50 # Reasonable button label length + and not cleaned.lower().startswith(("here", "the", "options", "choose")) + ): + labels.append(cleaned) + + # Return if we have the right number of labels + if len(labels) == expected_count: + return labels + + return None + + def _get_fallback_message(self, candidates: list[IntentCandidate]) -> str: + """Get fallback message when Bedrock is not available or fails.""" + if len(candidates) == 2: + return "I can help you with two things. Which would you like to do?" + else: + return "I can help you with several things. What would you like to do?" diff --git a/lex_helper/core/disambiguation/handler.py b/lex_helper/core/disambiguation/handler.py new file mode 100644 index 0000000..9c0f228 --- /dev/null +++ b/lex_helper/core/disambiguation/handler.py @@ -0,0 +1,483 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Disambiguation Handler for the Smart Disambiguation feature. + +This module provides the DisambiguationHandler class that orchestrates the +disambiguation process, generates user-friendly clarification responses, +and processes user selections to route to the appropriate intent. +""" + +import json +import logging +from typing import TypeVar + +from lex_helper.core.dialog import ( + close, + elicit_intent, +) +from lex_helper.core.message_manager import get_message +from lex_helper.core.types import ( + LexImageResponseCard, + LexMessages, + LexPlainText, + LexRequest, + LexResponse, + SessionAttributes, +) +from lex_helper.formatters.buttons import Button + +from .bedrock_generator import BedrockDisambiguationGenerator +from .types import ( + DisambiguationConfig, + IntentCandidate, +) + +logger = logging.getLogger(__name__) + +T = TypeVar("T", bound=SessionAttributes) + + +class DisambiguationHandler: + """ + Handles disambiguation response generation and user selection processing. + + This class is responsible for creating user-friendly clarification messages + when multiple intents are possible matches, and processing user responses + to route them to the correct intent handler. + """ + + def __init__(self, config: DisambiguationConfig | None = None): + """ + Initialize the disambiguation handler. + + Args: + config: Configuration options for disambiguation behavior + """ + self.config = config or DisambiguationConfig() + + # Initialize Bedrock generator if enabled + self.bedrock_generator = None + if self.config.bedrock_config.enabled: + try: + self.bedrock_generator = BedrockDisambiguationGenerator(self.config.bedrock_config) + logger.debug("Bedrock disambiguation generator initialized") + except Exception as e: + logger.warning("Failed to initialize Bedrock generator: %s", e) + if not self.config.bedrock_config.fallback_to_static: + raise + + def handle_disambiguation(self, lex_request: LexRequest[T], candidates: list[IntentCandidate]) -> LexResponse[T]: + """ + Generate disambiguation response with clarifying questions. + + Creates a response that presents the user with options to clarify + their intent, using buttons for easy selection. + + Args: + lex_request: The original Lex request + candidates: List of intent candidates to present + + Returns: + LexResponse with disambiguation options + """ + logger.info("Generating disambiguation response with %d candidates", len(candidates)) + + # Limit candidates to configured maximum + limited_candidates = candidates[: self.config.max_candidates] + + # Store disambiguation state in session + self._store_disambiguation_state(lex_request, limited_candidates) + + # Create clarification messages with user input context + messages = self._create_clarification_messages(limited_candidates, lex_request.inputTranscript) + + # Use elicit_intent to get user's clarification + return elicit_intent(messages, lex_request) + + def process_disambiguation_response(self, lex_request: LexRequest[T]) -> LexResponse[T] | None: + """ + Process user's response to disambiguation and route to selected intent. + + Analyzes the user's input to determine which intent they selected + and updates the request to route to that intent. + + Args: + lex_request: The Lex request with user's disambiguation response + + Returns: + None if this isn't a disambiguation response, otherwise routes + to the selected intent by updating the request + """ + # Check if this is a disambiguation response + if not self._is_disambiguation_response(lex_request): + return None + + logger.debug("Processing disambiguation response") + + # Get stored candidates from session + candidates = self._get_stored_candidates(lex_request) + if not candidates: + logger.warning("No stored disambiguation candidates found") + return self._create_fallback_response(lex_request) + + # Determine selected intent + selected_intent = self._determine_selected_intent(lex_request.inputTranscript, candidates) + + if not selected_intent: + logger.warning("Could not determine selected intent from input: %s", lex_request.inputTranscript) + return self._create_fallback_response(lex_request) + + # Update request to route to selected intent + self._update_request_for_selected_intent(lex_request, selected_intent) + + # Clear disambiguation state + self._clear_disambiguation_state(lex_request) + + logger.info("Routed disambiguation to intent: %s", selected_intent) + + # Return None to let the regular handler process the updated request + return None + + def _create_clarification_messages( + self, candidates: list[IntentCandidate], user_input: str | None = None + ) -> LexMessages: + """ + Create user-friendly clarification messages with intent options. + + Args: + candidates: List of intent candidates to present + user_input: Optional user input for context + + Returns: + List of messages including text and buttons + """ + # Get the main clarification message (potentially from Bedrock) + main_message = self._get_clarification_text(candidates, user_input) + + # Generate button labels (potentially from Bedrock) + button_labels = self._get_button_labels(candidates, user_input) + + # Create buttons for each candidate + buttons = [] + for i, candidate in enumerate(candidates): + button_text = button_labels[i] if i < len(button_labels) else candidate.display_name + button = Button( + text=button_text, + value=button_text, # Use button text as value for natural conversation flow + ) + buttons.append(button) + + # Create image response card with buttons + from lex_helper.core.types import ImageResponseCard + from lex_helper.formatters.buttons import buttons_to_dicts + + if buttons: + # Convert buttons to the format expected by ImageResponseCard + button_dicts = buttons_to_dicts(buttons) + + image_card = ImageResponseCard( + title="Please choose an option:", + subtitle="Select what you'd like to do", + buttons=[Button(text=b["text"], value=b["value"]) for b in button_dicts], + ) + + image_response = LexImageResponseCard(imageResponseCard=image_card) + + messages: LexMessages = [LexPlainText(content=main_message), image_response] + else: + messages: LexMessages = [LexPlainText(content=main_message)] + + return messages + + def _get_clarification_text(self, candidates: list[IntentCandidate], user_input: str | None = None) -> str: + """ + Get the main clarification text based on candidates. + + Args: + candidates: List of intent candidates + user_input: Optional user input for context + + Returns: + Clarification message text + """ + # Try Bedrock generation first if enabled + if self.bedrock_generator and user_input: + try: + return self.bedrock_generator.generate_clarification_message(user_input, candidates) + except Exception as e: + logger.warning("Bedrock clarification generation failed, falling back to static: %s", e) + # Continue to static message generation + + # Try to get custom message for specific scenario + custom_message = self._get_custom_clarification_message(candidates) + if custom_message: + return custom_message + + # Use default message based on number of candidates + if len(candidates) == 2: + base_message_key = "disambiguation.two_options" + default = "I can help you with two things. Which would you like to do?" + else: + base_message_key = "disambiguation.multiple_options" + default = "I can help you with several things. What would you like to do?" + + # Check if there's a custom message key configured for this scenario + custom_key = self.config.custom_messages.get(base_message_key) + if custom_key: + # Use the custom message key + localized_message = get_message(custom_key, None) + if localized_message: + return localized_message + # If custom key doesn't resolve, use it as fallback + return custom_key + + # Use the default message key + return get_message(base_message_key, default) + + def _get_custom_clarification_message(self, candidates: list[IntentCandidate]) -> str | None: + """ + Get custom clarification message for specific intent combinations. + + Args: + candidates: List of intent candidates + + Returns: + Custom message if available, None otherwise + """ + # Check for custom messages in config (treat as message keys) + intent_names = [c.intent_name for c in candidates] + intent_key = "_".join(sorted(intent_names)) + + custom_message_key = self.config.custom_messages.get(intent_key) + if custom_message_key: + # Try to get localized message using the key + localized_message = get_message(custom_message_key, None) + if localized_message: + return localized_message + # If no localized message found, use the key as fallback + return custom_message_key + + # Check for intent group messages + for group_name, group_intents in self.config.custom_intent_groups.items(): + if all(intent in group_intents for intent in intent_names): + group_message_key = f"disambiguation.{group_name}" + + # First check if there's a custom message key for this group + custom_group_key = self.config.custom_messages.get(group_message_key) + if custom_group_key: + localized_message = get_message(custom_group_key, None) + if localized_message: + return localized_message + + # Try to get from message manager with default group key + localized_message = get_message(group_message_key, None) + if localized_message: + return localized_message + + return None + + def _get_button_labels(self, candidates: list[IntentCandidate], user_input: str | None = None) -> list[str]: + """ + Get button labels for the candidates, potentially using Bedrock generation. + + Args: + candidates: List of intent candidates + user_input: Optional user input for context + + Returns: + List of button labels + """ + # Try Bedrock generation first if enabled + if self.bedrock_generator: + try: + return self.bedrock_generator.generate_button_labels(candidates, user_input) + except Exception as e: + logger.warning("Bedrock button label generation failed, falling back to display names: %s", e) + # Continue to fallback + + # Fallback to display names + return [candidate.display_name for candidate in candidates] + + def _store_disambiguation_state(self, lex_request: LexRequest[T], candidates: list[IntentCandidate]) -> None: + """ + Store disambiguation state in session attributes. + + Args: + lex_request: The Lex request to update + candidates: List of candidates to store + """ + session_attrs = lex_request.sessionState.sessionAttributes + + # Get button labels for storage + button_labels = self._get_button_labels(candidates, lex_request.inputTranscript) + + # Store candidates as JSON with button labels + candidates_data = [ + { + "intent_name": c.intent_name, + "display_name": c.display_name, + "button_label": button_labels[i] if i < len(button_labels) else c.display_name, + "confidence_score": c.confidence_score, + "description": c.description, + } + for i, c in enumerate(candidates) + ] + + # Store disambiguation state in session attributes + session_attrs.disambiguation_candidates = json.dumps(candidates_data) + session_attrs.disambiguation_active = True + + def _is_disambiguation_response(self, lex_request: LexRequest[T]) -> bool: + """ + Check if this request is a response to disambiguation. + + Args: + lex_request: The Lex request to check + + Returns: + True if this is a disambiguation response + """ + session_attrs = lex_request.sessionState.sessionAttributes + + # Check for disambiguation state + return session_attrs.disambiguation_active + + def _get_stored_candidates(self, lex_request: LexRequest[T]) -> list[IntentCandidate] | None: + """ + Retrieve stored disambiguation candidates from session. + + Args: + lex_request: The Lex request containing session state + + Returns: + List of stored candidates or None if not found + """ + session_attrs = lex_request.sessionState.sessionAttributes + + # Get candidates JSON + candidates_json = session_attrs.disambiguation_candidates + + if not candidates_json: + return None + + try: + candidates_data = json.loads(candidates_json) + return [ + IntentCandidate( + intent_name=c["intent_name"], + display_name=c["display_name"], + confidence_score=c["confidence_score"], + description=c["description"], + # Store button label in required_slots for now (we can extend IntentCandidate later if needed) + required_slots=[c.get("button_label", c["display_name"])], + ) + for c in candidates_data + ] + except (json.JSONDecodeError, KeyError) as e: + logger.error("Failed to parse stored disambiguation candidates: %s", e) + return None + + def _determine_selected_intent(self, user_input: str, candidates: list[IntentCandidate]) -> str | None: + """ + Determine which intent the user selected from their input. + + Args: + user_input: The user's input text + candidates: List of available candidates + + Returns: + Selected intent name or None if not determined + """ + user_input_lower = user_input.lower().strip() + + # Try exact match with intent names + for candidate in candidates: + if candidate.intent_name.lower() == user_input_lower: + return candidate.intent_name + + # Try exact match with display names + for candidate in candidates: + if candidate.display_name.lower() == user_input_lower: + return candidate.intent_name + + # Try exact match with button labels (stored in required_slots[0]) + for candidate in candidates: + if candidate.required_slots and candidate.required_slots[0].lower() == user_input_lower: + return candidate.intent_name + + # Try number selection (1, 2, 3, etc.) + try: + selection_num = int(user_input_lower) + if 1 <= selection_num <= len(candidates): + return candidates[selection_num - 1].intent_name + except ValueError: + pass + + # Try letter selection (A, B, C, etc.) - do this before partial match + if len(user_input_lower) == 1 and user_input_lower.isalpha(): + letter_index = ord(user_input_lower) - ord("a") + if 0 <= letter_index < len(candidates): + return candidates[letter_index].intent_name + + # Try partial match with display names + for candidate in candidates: + if user_input_lower in candidate.display_name.lower(): + return candidate.intent_name + + # Try partial match with button labels + for candidate in candidates: + if candidate.required_slots and user_input_lower in candidate.required_slots[0].lower(): + return candidate.intent_name + + return None + + def _update_request_for_selected_intent(self, lex_request: LexRequest[T], selected_intent: str) -> None: + """ + Update the Lex request to route to the selected intent. + + Args: + lex_request: The request to update + selected_intent: The intent name to route to + """ + # Update the intent in the session state + lex_request.sessionState.intent.name = selected_intent + lex_request.sessionState.intent.state = "InProgress" + + # Clear any existing slots since this is a new intent + lex_request.sessionState.intent.slots = {} + + def _clear_disambiguation_state(self, lex_request: LexRequest[T]) -> None: + """ + Clear disambiguation state from session attributes. + + Args: + lex_request: The request to update + """ + session_attrs = lex_request.sessionState.sessionAttributes + + # Clear disambiguation state + session_attrs.disambiguation_candidates = None + session_attrs.disambiguation_active = False + + def _create_fallback_response(self, lex_request: LexRequest[T]) -> LexResponse[T]: + """ + Create a fallback response when disambiguation fails. + + Args: + lex_request: The original request + + Returns: + Fallback response + """ + fallback_message = get_message( + "disambiguation.fallback", "I'm not sure what you're looking for. Could you be more specific?" + ) + + messages: LexMessages = [LexPlainText(content=fallback_message)] + + # Clear disambiguation state + self._clear_disambiguation_state(lex_request) + + return close(lex_request, messages) diff --git a/lex_helper/core/disambiguation/types.py b/lex_helper/core/disambiguation/types.py new file mode 100644 index 0000000..c14b575 --- /dev/null +++ b/lex_helper/core/disambiguation/types.py @@ -0,0 +1,140 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Type definitions for the Smart Disambiguation feature. + +This module contains all the data classes and type definitions needed for +intelligent disambiguation of ambiguous user input in lex-helper. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class IntentCandidate: + """ + Represents a potential intent match for disambiguation. + + Contains the intent information along with confidence scoring and + user-friendly display information for presenting options to users. + """ + + intent_name: str + """Technical name of the intent (e.g., 'BookFlight')""" + + confidence_score: float + """Confidence score between 0.0 and 1.0 for this intent match""" + + display_name: str + """User-friendly name for display (e.g., 'Book a Flight')""" + + description: str + """Brief description of what this intent does""" + + required_slots: list[str] = field(default_factory=lambda: []) + """List of required slot names for this intent""" + + +@dataclass +class DisambiguationResult: + """ + Result of disambiguation analysis. + + Contains the decision on whether disambiguation should occur and + all the information needed to present options to the user. + """ + + should_disambiguate: bool + """Whether disambiguation should be triggered based on analysis""" + + candidates: list[IntentCandidate] = field(default_factory=lambda: []) + """List of intent candidates to present to the user""" + + confidence_scores: dict[str, float] = field(default_factory=lambda: {}) + """Raw confidence scores for all analyzed intents""" + + +@dataclass +class BedrockDisambiguationConfig: + """ + Configuration for Bedrock-powered disambiguation text generation. + + Allows using Amazon Bedrock models to generate contextual and intelligent + disambiguation messages and button text based on the user's input and + available intent candidates. + """ + + enabled: bool = False + """Whether to use Bedrock for generating disambiguation text""" + + model_id: str = "anthropic.claude-3-haiku-20240307-v1:0" + """Bedrock model ID to use for text generation""" + + region_name: str = "us-east-1" + """AWS region for Bedrock service""" + + max_tokens: int = 200 + """Maximum tokens for generated response""" + + temperature: float = 0.3 + """Temperature for text generation (0.0-1.0, lower = more deterministic)""" + + system_prompt: str = field( + default_factory=lambda: ( + "You are a helpful assistant that creates clear, concise disambiguation messages " + "for chatbot users. When users provide ambiguous input, help them choose between " + "available options with friendly, natural language." + ) + ) + """System prompt for the Bedrock model""" + + fallback_to_static: bool = True + """Whether to fall back to static messages if Bedrock fails""" + + +@dataclass +class DisambiguationConfig: + """ + Configuration options for the disambiguation system. + + Allows developers to customize disambiguation behavior including + thresholds, candidate limits, and custom intent groupings. + """ + + confidence_threshold: float = 0.6 + """Minimum confidence score to avoid disambiguation (0.0-1.0)""" + + max_candidates: int = 3 + """Maximum number of intent candidates to present to users""" + + fallback_to_original: bool = True + """Whether to fall back to original behavior if disambiguation fails""" + + min_candidates: int = 2 + """Minimum number of candidates required to trigger disambiguation""" + + similarity_threshold: float = 0.15 + """Maximum difference between top scores to trigger disambiguation (0.0-1.0)""" + + enable_logging: bool = True + """Whether to enable detailed logging of disambiguation events""" + + custom_intent_groups: dict[str, list[str]] = field(default_factory=lambda: {}) + """Custom groupings of related intents for better disambiguation""" + + custom_messages: dict[str, str] = field(default_factory=lambda: {}) + """Custom clarification messages for specific disambiguation scenarios""" + + bedrock_config: BedrockDisambiguationConfig = field(default_factory=BedrockDisambiguationConfig) + """Configuration for Bedrock-powered text generation""" + + +# Type aliases for better code readability +IntentScores = dict[str, float] +"""Type alias for intent name to confidence score mapping""" + +DisambiguationMessages = dict[str, str] +"""Type alias for disambiguation message templates""" diff --git a/lex_helper/core/handler.py b/lex_helper/core/handler.py index 2c3fcf0..8828f54 100644 --- a/lex_helper/core/handler.py +++ b/lex_helper/core/handler.py @@ -3,7 +3,7 @@ import logging from collections.abc import Callable -from typing import Any, TypeVar +from typing import Any, TypeVar, cast from pydantic import BaseModel @@ -25,6 +25,19 @@ logger = logging.getLogger(__name__) +# Import disambiguation components (lazy import to avoid circular dependencies) +try: + from lex_helper.core.disambiguation.analyzer import DisambiguationAnalyzer + from lex_helper.core.disambiguation.handler import DisambiguationHandler + from lex_helper.core.disambiguation.types import DisambiguationConfig + + disambiguation_available = True +except ImportError: + disambiguation_available = False + DisambiguationConfig = None # type: ignore + DisambiguationHandler = None # type: ignore + DisambiguationAnalyzer = None # type: ignore + T = TypeVar("T", bound=SessionAttributes) @@ -36,12 +49,26 @@ class Config[T: SessionAttributes](BaseModel): auto_initialize_messages: bool = True # Automatically initialize MessageManager with locale from request auto_handle_exceptions: bool = True # Automatically handle exceptions and return error responses error_message: str | None = None # Custom error message or message key for exceptions + enable_disambiguation: bool = False # Enable smart disambiguation for ambiguous input + disambiguation_config: Any | None = None # Configuration for disambiguation behavior class LexHelper[T: SessionAttributes]: def __init__(self, config: Config[T]): self.config = config + # Initialize disambiguation components if enabled + self.disambiguation_handler = None + self.disambiguation_analyzer = None + + if self.config.enable_disambiguation and disambiguation_available: + disambiguation_config = self.config.disambiguation_config or DisambiguationConfig() # type: ignore + self.disambiguation_handler = DisambiguationHandler(disambiguation_config) # type: ignore + self.disambiguation_analyzer = DisambiguationAnalyzer(disambiguation_config) # type: ignore + logger.debug("Disambiguation components initialized") + elif self.config.enable_disambiguation and not disambiguation_available: + logger.warning("Disambiguation requested but components not available - falling back to regular behavior") + def handler(self, event: dict[str, Any], context: Any) -> dict[str, Any]: """ Primary entry point for the lex_helper library. @@ -117,9 +144,14 @@ def _main_handler(self, lex_payload: LexRequest[T]) -> dict[str, Any]: logger.debug("Lex-Intent: %s", lex_intent_name) # Handlers is a list of functions that take a LexRequest and return a LexResponse - handlers: list[Callable[[LexRequest[T]], LexResponse[T] | None]] = [ - self.regular_intent_handler, - ] + handlers: list[Callable[[LexRequest[T]], LexResponse[T] | None]] = [] + + # Add disambiguation handler first if enabled + if self.disambiguation_handler: + handlers.append(self.disambiguation_intent_handler) + + # Add regular intent handler + handlers.append(self.regular_intent_handler) try: response = None @@ -148,7 +180,7 @@ def _main_handler(self, lex_payload: LexRequest[T]) -> dict[str, Any]: messages += response.messages if response.requestAttributes and "callback" in response.requestAttributes: callback_name = response.requestAttributes["callback"] - logger.debug("CALLBACK FOUND: %s", callback_name) + logger.debug("Callback found: %s", callback_name) response.requestAttributes.pop("callback") response = call_handler_for_file( intent_name=callback_name, lex_request=lex_payload, package_name=self.config.package_name @@ -172,6 +204,33 @@ def _main_handler(self, lex_payload: LexRequest[T]) -> dict[str, Any]: logger.exception(e) raise e + def disambiguation_intent_handler(self, lex_payload: LexRequest[T]) -> LexResponse[T] | None: + """ + Handle disambiguation responses and trigger disambiguation when needed. + + This handler processes user responses to disambiguation questions and + triggers new disambiguation when confidence is low. + """ + if not self.disambiguation_handler: + return None + + # First, check if this is a response to a previous disambiguation + disambiguation_response = self.disambiguation_handler.process_disambiguation_response(lex_payload) + if disambiguation_response is not None: + return disambiguation_response + + # If not a disambiguation response, check if we need to trigger disambiguation + if self.disambiguation_analyzer: + # Analyze the request for disambiguation + analysis_result = self.disambiguation_analyzer.analyze_request(cast(Any, lex_payload)) + + if analysis_result.should_disambiguate and analysis_result.candidates: + logger.info("Triggering disambiguation with %d candidates", len(analysis_result.candidates)) + return self.disambiguation_handler.handle_disambiguation(lex_payload, analysis_result.candidates) + + # No disambiguation needed, let regular handler process + return None + def regular_intent_handler(self, lex_payload: LexRequest[T]) -> LexResponse[T] | None: """ Route the incoming request based on intent. diff --git a/lex_helper/core/types.py b/lex_helper/core/types.py index de5991d..36494a9 100644 --- a/lex_helper/core/types.py +++ b/lex_helper/core/types.py @@ -158,6 +158,10 @@ class SessionAttributes(BaseModel): channel: str = "lex" + # Disambiguation attributes + disambiguation_candidates: str | None = None # JSON string of disambiguation candidates + disambiguation_active: bool = False # Whether disambiguation is currently active + def to_cmd_response(self): response = "" self_dict = self.model_dump() diff --git a/lex_helper/formatters/format_buttons.py b/lex_helper/formatters/format_buttons.py index fc75a28..95b20a0 100644 --- a/lex_helper/formatters/format_buttons.py +++ b/lex_helper/formatters/format_buttons.py @@ -1,7 +1,7 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 -from lex_helper import Button +from lex_helper.formatters.buttons import Button def format_buttons(buttons: list[Button]) -> list[Button]: diff --git a/pyproject.toml b/pyproject.toml index 4ad79e4..ebafded 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,6 +77,7 @@ exclude = [ "tests/", "examples/", "docs/", + "tools/", ".github/", ".gitignore", ".pre-commit-config.yaml", diff --git a/tests/test_bedrock_disambiguation.py b/tests/test_bedrock_disambiguation.py new file mode 100644 index 0000000..9d93790 --- /dev/null +++ b/tests/test_bedrock_disambiguation.py @@ -0,0 +1,221 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Tests for Bedrock-powered disambiguation functionality. +""" + +from unittest.mock import patch + +import pytest + +from lex_helper.core.disambiguation.bedrock_generator import BedrockDisambiguationGenerator +from lex_helper.core.disambiguation.types import BedrockDisambiguationConfig, IntentCandidate + + +class TestBedrockDisambiguationGenerator: + """Test the Bedrock disambiguation generator.""" + + @pytest.fixture + def bedrock_config(self): + """Create a test Bedrock configuration.""" + return BedrockDisambiguationConfig( + enabled=True, + model_id="anthropic.claude-3-haiku-20240307-v1:0", + max_tokens=150, + temperature=0.3, + fallback_to_static=True, + ) + + @pytest.fixture + def disabled_bedrock_config(self): + """Create a disabled Bedrock configuration.""" + return BedrockDisambiguationConfig(enabled=False) + + @pytest.fixture + def sample_candidates(self): + """Create sample intent candidates.""" + return [ + IntentCandidate( + intent_name="BookFlight", + confidence_score=0.4, + display_name="Book Flight", + description="Book a new flight reservation", + ), + IntentCandidate( + intent_name="ChangeFlight", + confidence_score=0.35, + display_name="Change Flight", + description="Modify an existing flight booking", + ), + ] + + def test_generator_initialization_enabled(self, bedrock_config): + """Test generator initialization when enabled.""" + generator = BedrockDisambiguationGenerator(bedrock_config) + assert generator.config == bedrock_config + + def test_generator_initialization_disabled(self, disabled_bedrock_config): + """Test generator initialization when disabled.""" + generator = BedrockDisambiguationGenerator(disabled_bedrock_config) + assert generator.config == disabled_bedrock_config + + def test_generate_clarification_message_disabled(self, disabled_bedrock_config, sample_candidates): + """Test clarification message generation when Bedrock is disabled.""" + generator = BedrockDisambiguationGenerator(disabled_bedrock_config) + + result = generator.generate_clarification_message("I need help", sample_candidates) + + assert result == "I can help you with two things. Which would you like to do?" + + @patch("lex_helper.core.disambiguation.bedrock_generator.invoke_bedrock_simple_converse") + def test_generate_clarification_message_success(self, mock_invoke, bedrock_config, sample_candidates): + """Test successful clarification message generation.""" + mock_invoke.return_value = { + "text": "I can help you book a new flight or change your existing booking. Which would you prefer?", + "usage": {}, + } + + generator = BedrockDisambiguationGenerator(bedrock_config) + result = generator.generate_clarification_message("I need help with my flight", sample_candidates) + + assert "book a new flight or change" in result + mock_invoke.assert_called_once() + + @patch("lex_helper.core.disambiguation.bedrock_generator.invoke_bedrock_simple_converse") + def test_generate_clarification_message_bedrock_error_with_fallback( + self, mock_invoke, bedrock_config, sample_candidates + ): + """Test clarification message generation with Bedrock error and fallback enabled.""" + from lex_helper.core.invoke_bedrock import BedrockInvocationError + + mock_invoke.side_effect = BedrockInvocationError("Model not available") + + generator = BedrockDisambiguationGenerator(bedrock_config) + result = generator.generate_clarification_message("I need help", sample_candidates) + + # Should fall back to static message + assert result == "I can help you with two things. Which would you like to do?" + + @patch("lex_helper.core.disambiguation.bedrock_generator.invoke_bedrock_simple_converse") + def test_generate_clarification_message_bedrock_error_no_fallback(self, mock_invoke, sample_candidates): + """Test clarification message generation with Bedrock error and no fallback.""" + from lex_helper.core.invoke_bedrock import BedrockInvocationError + + config = BedrockDisambiguationConfig(enabled=True, fallback_to_static=False) + mock_invoke.side_effect = BedrockInvocationError("Model not available") + + generator = BedrockDisambiguationGenerator(config) + + with pytest.raises(BedrockInvocationError): + generator.generate_clarification_message("I need help", sample_candidates) + + def test_generate_button_labels_disabled(self, disabled_bedrock_config, sample_candidates): + """Test button label generation when Bedrock is disabled.""" + generator = BedrockDisambiguationGenerator(disabled_bedrock_config) + + result = generator.generate_button_labels(sample_candidates) + + assert result == ["Book Flight", "Change Flight"] + + @patch("lex_helper.core.disambiguation.bedrock_generator.invoke_bedrock_simple_converse") + def test_generate_button_labels_success_json(self, mock_invoke, bedrock_config, sample_candidates): + """Test successful button label generation with JSON response.""" + mock_invoke.return_value = { + "text": '["Book new flight", "Modify booking"]', + "usage": {}, + } + + generator = BedrockDisambiguationGenerator(bedrock_config) + result = generator.generate_button_labels(sample_candidates, "I need help with my flight") + + assert result == ["Book new flight", "Modify booking"] + mock_invoke.assert_called_once() + + @patch("lex_helper.core.disambiguation.bedrock_generator.invoke_bedrock_simple_converse") + def test_generate_button_labels_success_text_extraction(self, mock_invoke, bedrock_config, sample_candidates): + """Test button label generation with text extraction fallback.""" + mock_invoke.return_value = { + "text": "Here are the options:\n- Book new flight\n- Modify booking", + "usage": {}, + } + + generator = BedrockDisambiguationGenerator(bedrock_config) + result = generator.generate_button_labels(sample_candidates) + + assert result == ["Book new flight", "Modify booking"] + + @patch("lex_helper.core.disambiguation.bedrock_generator.invoke_bedrock_simple_converse") + def test_generate_button_labels_parsing_failure(self, mock_invoke, bedrock_config, sample_candidates): + """Test button label generation when parsing fails.""" + mock_invoke.return_value = { + "text": "Some unparseable response that doesn't match expected format", + "usage": {}, + } + + generator = BedrockDisambiguationGenerator(bedrock_config) + result = generator.generate_button_labels(sample_candidates) + + # Should fall back to display names + assert result == ["Book Flight", "Change Flight"] + + def test_extract_labels_from_text_success(self, bedrock_config): + """Test successful label extraction from text.""" + generator = BedrockDisambiguationGenerator(bedrock_config) + + text = "- Book new flight\n- Modify booking" + result = generator._extract_labels_from_text(text, 2) + + assert result == ["Book new flight", "Modify booking"] + + def test_extract_labels_from_text_wrong_count(self, bedrock_config): + """Test label extraction with wrong number of labels.""" + generator = BedrockDisambiguationGenerator(bedrock_config) + + text = "- Book new flight" # Only one label, expecting two + result = generator._extract_labels_from_text(text, 2) + + assert result is None + + def test_build_clarification_prompt(self, bedrock_config, sample_candidates): + """Test clarification prompt building.""" + generator = BedrockDisambiguationGenerator(bedrock_config) + + prompt = generator._build_clarification_prompt("I need help", sample_candidates) + + assert "I need help" in prompt + assert "Book Flight" in prompt + assert "Change Flight" in prompt + assert "ambiguous" in prompt + + def test_build_button_labels_prompt(self, bedrock_config, sample_candidates): + """Test button labels prompt building.""" + generator = BedrockDisambiguationGenerator(bedrock_config) + + prompt = generator._build_button_labels_prompt(sample_candidates, "I need help") + + assert "BookFlight" in prompt + assert "ChangeFlight" in prompt + assert "I need help" in prompt + assert "JSON array" in prompt + + def test_get_fallback_message_two_candidates(self, bedrock_config, sample_candidates): + """Test fallback message for two candidates.""" + generator = BedrockDisambiguationGenerator(bedrock_config) + + result = generator._get_fallback_message(sample_candidates) + + assert result == "I can help you with two things. Which would you like to do?" + + def test_get_fallback_message_multiple_candidates(self, bedrock_config): + """Test fallback message for multiple candidates.""" + candidates = [ + IntentCandidate("Intent1", 0.3, "Display 1", "Desc 1"), + IntentCandidate("Intent2", 0.3, "Display 2", "Desc 2"), + IntentCandidate("Intent3", 0.3, "Display 3", "Desc 3"), + ] + + generator = BedrockDisambiguationGenerator(bedrock_config) + result = generator._get_fallback_message(candidates) + + assert result == "I can help you with several things. What would you like to do?" diff --git a/tests/test_disambiguation_analyzer.py b/tests/test_disambiguation_analyzer.py new file mode 100644 index 0000000..30dd7fe --- /dev/null +++ b/tests/test_disambiguation_analyzer.py @@ -0,0 +1,273 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for DisambiguationAnalyzer. + +Tests the core functionality of analyzing Lex requests for disambiguation +including confidence score extraction, threshold-based decisions, and +candidate generation. +""" + +from lex_helper.core.disambiguation import ( + DisambiguationAnalyzer, + DisambiguationConfig, +) +from lex_helper.core.types import ( + Intent, + Interpretation, + LexRequest, + SessionAttributes, +) + + +class TestSessionAttributes(SessionAttributes): + """Test session attributes class.""" + + pass + + +class TestDisambiguationAnalyzer: + """Test cases for DisambiguationAnalyzer.""" + + def test_init_with_default_config(self): + """Test analyzer initialization with default configuration.""" + analyzer = DisambiguationAnalyzer() + + assert analyzer.config is not None + assert analyzer.config.confidence_threshold == 0.6 + assert analyzer.config.max_candidates == 3 + assert analyzer.config.min_candidates == 2 + + def test_init_with_custom_config(self): + """Test analyzer initialization with custom configuration.""" + config = DisambiguationConfig( + confidence_threshold=0.8, + max_candidates=5, + min_candidates=3, + ) + analyzer = DisambiguationAnalyzer(config) + + assert analyzer.config.confidence_threshold == 0.8 + assert analyzer.config.max_candidates == 5 + assert analyzer.config.min_candidates == 3 + + def test_extract_intent_scores_basic(self): + """Test extracting intent scores from Lex interpretations.""" + analyzer = DisambiguationAnalyzer() + + # Create test interpretations + interpretations = [ + Interpretation(intent=Intent(name="BookFlight"), nluConfidence=0.8), + Interpretation(intent=Intent(name="CancelFlight"), nluConfidence=0.3), + Interpretation(intent=Intent(name="ChangeFlight"), nluConfidence=0.1), + ] + + lex_request = LexRequest[TestSessionAttributes]( + inputTranscript="I want to book a flight", interpretations=interpretations + ) + + scores = analyzer.extract_intent_scores(lex_request) + + assert scores["BookFlight"] == 0.8 + assert scores["CancelFlight"] == 0.3 + assert scores["ChangeFlight"] == 0.1 + + def test_extract_intent_scores_missing_confidence(self): + """Test extracting scores when nluConfidence is None.""" + analyzer = DisambiguationAnalyzer() + + interpretations = [ + Interpretation(intent=Intent(name="BookFlight"), nluConfidence=None), + ] + + lex_request = LexRequest[TestSessionAttributes](interpretations=interpretations) + + scores = analyzer.extract_intent_scores(lex_request) + + assert scores["BookFlight"] == 0.0 + + def test_should_disambiguate_high_confidence(self): + """Test that high confidence scores don't trigger disambiguation.""" + analyzer = DisambiguationAnalyzer() + + scores = { + "BookFlight": 0.9, + "CancelFlight": 0.1, + "ChangeFlight": 0.05, + } + + result = analyzer.should_disambiguate(scores, 0.6) + + assert result is False + + def test_should_disambiguate_low_confidence(self): + """Test that low confidence scores trigger disambiguation.""" + analyzer = DisambiguationAnalyzer() + + scores = { + "BookFlight": 0.4, + "CancelFlight": 0.3, + "ChangeFlight": 0.2, + } + + result = analyzer.should_disambiguate(scores, 0.6) + + assert result is True + + def test_should_disambiguate_similar_high_scores(self): + """Test that similar high scores trigger disambiguation.""" + analyzer = DisambiguationAnalyzer() + + scores = { + "BookFlight": 0.7, + "CancelFlight": 0.65, + "ChangeFlight": 0.1, + } + + result = analyzer.should_disambiguate(scores, 0.6) + + assert result is True + + def test_should_disambiguate_insufficient_candidates(self): + """Test that insufficient candidates don't trigger disambiguation.""" + analyzer = DisambiguationAnalyzer() + + scores = { + "BookFlight": 0.4, + } + + result = analyzer.should_disambiguate(scores, 0.6) + + assert result is False + + def test_should_disambiguate_empty_scores(self): + """Test behavior with empty scores.""" + analyzer = DisambiguationAnalyzer() + + scores = {} + + result = analyzer.should_disambiguate(scores, 0.6) + + assert result is False + + def test_analyze_request_no_disambiguation_needed(self): + """Test full analysis when no disambiguation is needed.""" + analyzer = DisambiguationAnalyzer() + + interpretations = [ + Interpretation(intent=Intent(name="BookFlight"), nluConfidence=0.9), + Interpretation(intent=Intent(name="CancelFlight"), nluConfidence=0.1), + ] + + lex_request = LexRequest[TestSessionAttributes]( + inputTranscript="I want to book a flight", interpretations=interpretations + ) + + result = analyzer.analyze_request(lex_request) + + assert result.should_disambiguate is False + assert len(result.candidates) == 0 + assert result.confidence_scores["BookFlight"] == 0.9 + assert result.confidence_scores["CancelFlight"] == 0.1 + + def test_analyze_request_disambiguation_needed(self): + """Test full analysis when disambiguation is needed.""" + analyzer = DisambiguationAnalyzer() + + interpretations = [ + Interpretation(intent=Intent(name="BookFlight", slots={"OriginCity": None}), nluConfidence=0.4), + Interpretation(intent=Intent(name="CancelFlight", slots={"ReservationNumber": None}), nluConfidence=0.3), + ] + + lex_request = LexRequest[TestSessionAttributes]( + inputTranscript="I need help with my flight", interpretations=interpretations + ) + + result = analyzer.analyze_request(lex_request) + + assert result.should_disambiguate is True + assert len(result.candidates) == 2 + assert result.candidates[0].intent_name == "BookFlight" + assert result.candidates[0].confidence_score == 0.4 + assert result.candidates[1].intent_name == "CancelFlight" + assert result.candidates[1].confidence_score == 0.3 + + def test_generate_candidates_with_slots(self): + """Test candidate generation includes slot information.""" + analyzer = DisambiguationAnalyzer() + + scores = { + "BookFlight": 0.4, + "CancelFlight": 0.3, + } + + interpretations = [ + Interpretation( + intent=Intent(name="BookFlight", slots={"OriginCity": None, "DestinationCity": None}), nluConfidence=0.4 + ), + Interpretation(intent=Intent(name="CancelFlight", slots={"ReservationNumber": None}), nluConfidence=0.3), + ] + + lex_request = LexRequest[TestSessionAttributes](interpretations=interpretations) + + candidates = analyzer._generate_candidates(scores, lex_request) + + assert len(candidates) == 2 + assert candidates[0].required_slots == ["OriginCity", "DestinationCity"] + assert candidates[1].required_slots == ["ReservationNumber"] + + def test_get_display_name_camel_case(self): + """Test display name generation for CamelCase intents.""" + analyzer = DisambiguationAnalyzer() + + display_name = analyzer._get_display_name("BookFlight") + + assert display_name == "Book Flight" + + def test_get_display_name_snake_case(self): + """Test display name generation for snake_case intents.""" + analyzer = DisambiguationAnalyzer() + + display_name = analyzer._get_display_name("book_flight") + + assert display_name == "Book Flight" + + def test_get_display_name_mixed_case(self): + """Test display name generation for mixed case intents.""" + analyzer = DisambiguationAnalyzer() + + display_name = analyzer._get_display_name("BookFlight_Request") + + assert display_name == "Book Flight Request" + + def test_find_interpretation_by_intent(self): + """Test finding interpretation by intent name.""" + analyzer = DisambiguationAnalyzer() + + interpretations = [ + Interpretation(intent=Intent(name="BookFlight"), nluConfidence=0.8), + Interpretation(intent=Intent(name="CancelFlight"), nluConfidence=0.3), + ] + + lex_request = LexRequest[TestSessionAttributes](interpretations=interpretations) + + interpretation = analyzer._find_interpretation_by_intent(lex_request, "CancelFlight") + + assert interpretation is not None + assert interpretation.intent.name == "CancelFlight" + assert interpretation.nluConfidence == 0.3 + + def test_find_interpretation_by_intent_not_found(self): + """Test finding interpretation when intent doesn't exist.""" + analyzer = DisambiguationAnalyzer() + + interpretations = [ + Interpretation(intent=Intent(name="BookFlight"), nluConfidence=0.8), + ] + + lex_request = LexRequest[TestSessionAttributes](interpretations=interpretations) + + interpretation = analyzer._find_interpretation_by_intent(lex_request, "NonExistentIntent") + + assert interpretation is None diff --git a/tests/test_disambiguation_handler.py b/tests/test_disambiguation_handler.py new file mode 100644 index 0000000..f1396dd --- /dev/null +++ b/tests/test_disambiguation_handler.py @@ -0,0 +1,350 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for the DisambiguationHandler class. + +Tests cover response generation, message formatting, user selection processing, +and integration with the lex-helper dialog system. +""" + +import json +from unittest.mock import Mock, patch + +import pytest + +from lex_helper.core.disambiguation.handler import DisambiguationHandler +from lex_helper.core.disambiguation.types import ( + DisambiguationConfig, + IntentCandidate, +) +from lex_helper.core.types import ( + Bot, + Intent, + LexPlainText, + LexRequest, + SessionAttributes, + SessionState, +) + + +class TestSessionAttributes(SessionAttributes): + """Test session attributes class.""" + + pass + + +@pytest.fixture +def sample_candidates(): + """Sample intent candidates for testing.""" + return [ + IntentCandidate( + intent_name="BookFlight", + confidence_score=0.7, + display_name="Book a Flight", + description="Book a new flight reservation", + ), + IntentCandidate( + intent_name="ChangeFlight", + confidence_score=0.6, + display_name="Change Flight", + description="Modify an existing flight reservation", + ), + IntentCandidate( + intent_name="CancelFlight", + confidence_score=0.5, + display_name="Cancel Flight", + description="Cancel a flight reservation", + ), + ] + + +@pytest.fixture +def sample_lex_request(): + """Sample LexRequest for testing.""" + return LexRequest( + sessionId="test-session", + inputTranscript="I want to book a flight", + bot=Bot(localeId="en_US"), + sessionState=SessionState(intent=Intent(name="FallbackIntent"), sessionAttributes=TestSessionAttributes()), + ) + + +@pytest.fixture +def disambiguation_handler(): + """DisambiguationHandler instance for testing.""" + config = DisambiguationConfig( + confidence_threshold=0.6, + max_candidates=3, + custom_messages={ + "disambiguation.booking": "I can help you book, change, or cancel a flight. Which would you like to do?" + }, + custom_intent_groups={"booking": ["BookFlight", "ChangeFlight", "CancelFlight"]}, + ) + return DisambiguationHandler(config) + + +class TestDisambiguationHandler: + """Test cases for DisambiguationHandler.""" + + def test_init_with_default_config(self): + """Test handler initialization with default config.""" + handler = DisambiguationHandler() + assert handler.config is not None + assert handler.config.confidence_threshold == 0.6 + assert handler.config.max_candidates == 3 + + def test_init_with_custom_config(self): + """Test handler initialization with custom config.""" + config = DisambiguationConfig(confidence_threshold=0.5, max_candidates=2) + handler = DisambiguationHandler(config) + assert handler.config.confidence_threshold == 0.5 + assert handler.config.max_candidates == 2 + + @patch("lex_helper.core.disambiguation.handler.elicit_intent") + @patch("lex_helper.core.disambiguation.handler.get_message") + def test_handle_disambiguation_basic( + self, mock_get_message, mock_elicit_intent, disambiguation_handler, sample_lex_request, sample_candidates + ): + """Test basic disambiguation response generation.""" + mock_get_message.return_value = "I can help you with several things. What would you like to do?" + mock_response = Mock() + mock_elicit_intent.return_value = mock_response + + result = disambiguation_handler.handle_disambiguation(sample_lex_request, sample_candidates) + + assert result == mock_response + mock_elicit_intent.assert_called_once() + + # Check that disambiguation state was stored + session_attrs = sample_lex_request.sessionState.sessionAttributes + assert session_attrs.disambiguation_active is True + + @patch("lex_helper.core.disambiguation.handler.elicit_intent") + def test_handle_disambiguation_limits_candidates(self, mock_elicit_intent, sample_lex_request, sample_candidates): + """Test that handler limits candidates to max_candidates.""" + config = DisambiguationConfig(max_candidates=2) + handler = DisambiguationHandler(config) + + mock_elicit_intent.return_value = Mock() + + handler.handle_disambiguation(sample_lex_request, sample_candidates) + + # Check stored candidates are limited + session_attrs = sample_lex_request.sessionState.sessionAttributes + candidates_json = session_attrs.disambiguation_candidates + stored_candidates = json.loads(candidates_json) + assert len(stored_candidates) == 2 + + @patch("lex_helper.core.disambiguation.handler.get_message") + def test_create_clarification_messages(self, mock_get_message, disambiguation_handler, sample_candidates): + """Test clarification message creation.""" + # Mock the message system to return the expected message for the booking group + mock_get_message.return_value = "I can help you book, change, or cancel a flight. Which would you like to do?" + + messages = disambiguation_handler._create_clarification_messages(sample_candidates) + + # Should have at least one plain text message + assert len(messages) >= 1 + assert isinstance(messages[0], LexPlainText) + # Should use the mocked message + assert messages[0].content == "I can help you book, change, or cancel a flight. Which would you like to do?" + + @patch("lex_helper.core.disambiguation.handler.get_message") + def test_get_custom_clarification_message(self, mock_get_message, disambiguation_handler, sample_candidates): + """Test custom clarification message retrieval.""" + # Mock the message system to return the expected message + mock_get_message.return_value = "I can help you book, change, or cancel a flight. Which would you like to do?" + + # Test with custom intent group message - should use the configured custom message + result = disambiguation_handler._get_custom_clarification_message(sample_candidates) + assert result == "I can help you book, change, or cancel a flight. Which would you like to do?" + + def test_store_and_retrieve_disambiguation_state(self, disambiguation_handler, sample_lex_request, sample_candidates): + """Test storing and retrieving disambiguation state.""" + # Store state + disambiguation_handler._store_disambiguation_state(sample_lex_request, sample_candidates) + + # Check state is stored + assert disambiguation_handler._is_disambiguation_response(sample_lex_request) + + # Retrieve candidates + retrieved = disambiguation_handler._get_stored_candidates(sample_lex_request) + assert retrieved is not None + assert len(retrieved) == len(sample_candidates) + assert retrieved[0].intent_name == sample_candidates[0].intent_name + + def test_determine_selected_intent_exact_match(self, sample_candidates): + """Test intent selection with exact matches.""" + handler = DisambiguationHandler() + + # Test exact intent name match + result = handler._determine_selected_intent("BookFlight", sample_candidates) + assert result == "BookFlight" + + # Test exact display name match + result = handler._determine_selected_intent("Book a Flight", sample_candidates) + assert result == "BookFlight" + + def test_determine_selected_intent_partial_match(self, sample_candidates): + """Test intent selection with partial matches.""" + handler = DisambiguationHandler() + + # Test partial display name match + result = handler._determine_selected_intent("book", sample_candidates) + assert result == "BookFlight" + + def test_determine_selected_intent_number_selection(self, sample_candidates): + """Test intent selection with number input.""" + handler = DisambiguationHandler() + + # Test number selection + result = handler._determine_selected_intent("1", sample_candidates) + assert result == "BookFlight" + + result = handler._determine_selected_intent("2", sample_candidates) + assert result == "ChangeFlight" + + def test_determine_selected_intent_letter_selection(self, sample_candidates): + """Test intent selection with letter input.""" + handler = DisambiguationHandler() + + # Test letter selection (a=0, b=1, c=2) + result = handler._determine_selected_intent("a", sample_candidates) + assert result == "BookFlight" + + result = handler._determine_selected_intent("b", sample_candidates) + assert result == "ChangeFlight" + + def test_determine_selected_intent_no_match(self, sample_candidates): + """Test intent selection with no match.""" + handler = DisambiguationHandler() + + result = handler._determine_selected_intent("invalid input", sample_candidates) + assert result is None + + def test_update_request_for_selected_intent(self, sample_lex_request): + """Test updating request for selected intent.""" + handler = DisambiguationHandler() + + handler._update_request_for_selected_intent(sample_lex_request, "BookFlight") + + assert sample_lex_request.sessionState.intent.name == "BookFlight" + assert sample_lex_request.sessionState.intent.state == "InProgress" + assert sample_lex_request.sessionState.intent.slots == {} + + def test_clear_disambiguation_state(self, disambiguation_handler, sample_lex_request, sample_candidates): + """Test clearing disambiguation state.""" + # First store state + disambiguation_handler._store_disambiguation_state(sample_lex_request, sample_candidates) + assert disambiguation_handler._is_disambiguation_response(sample_lex_request) + + # Clear state + disambiguation_handler._clear_disambiguation_state(sample_lex_request) + assert not disambiguation_handler._is_disambiguation_response(sample_lex_request) + + @patch("lex_helper.core.disambiguation.handler.close") + @patch("lex_helper.core.disambiguation.handler.get_message") + def test_create_fallback_response(self, mock_get_message, mock_close, disambiguation_handler, sample_lex_request): + """Test fallback response creation.""" + mock_get_message.return_value = "I'm not sure what you're looking for." + mock_response = Mock() + mock_close.return_value = mock_response + + result = disambiguation_handler._create_fallback_response(sample_lex_request) + + assert result == mock_response + mock_close.assert_called_once() + + def test_process_disambiguation_response_not_disambiguation(self, disambiguation_handler, sample_lex_request): + """Test processing when request is not a disambiguation response.""" + result = disambiguation_handler.process_disambiguation_response(sample_lex_request) + assert result is None + + def test_process_disambiguation_response_no_candidates(self, disambiguation_handler, sample_lex_request): + """Test processing when no stored candidates exist.""" + # Set disambiguation active but no candidates + session_attrs = sample_lex_request.sessionState.sessionAttributes + session_attrs.disambiguation_active = True + + with patch.object(disambiguation_handler, "_create_fallback_response") as mock_fallback: + mock_response = Mock() + mock_fallback.return_value = mock_response + + result = disambiguation_handler.process_disambiguation_response(sample_lex_request) + assert result == mock_response + + def test_process_disambiguation_response_success(self, disambiguation_handler, sample_lex_request, sample_candidates): + """Test successful disambiguation response processing.""" + # Store disambiguation state + disambiguation_handler._store_disambiguation_state(sample_lex_request, sample_candidates) + + # Set user input to select first option + sample_lex_request.inputTranscript = "1" + + result = disambiguation_handler.process_disambiguation_response(sample_lex_request) + + # Should return None to let regular handler process + assert result is None + + # Should have updated intent + assert sample_lex_request.sessionState.intent.name == "BookFlight" + + # Should have cleared disambiguation state + assert not disambiguation_handler._is_disambiguation_response(sample_lex_request) + + def test_process_disambiguation_response_invalid_selection( + self, disambiguation_handler, sample_lex_request, sample_candidates + ): + """Test disambiguation response with invalid selection.""" + # Store disambiguation state + disambiguation_handler._store_disambiguation_state(sample_lex_request, sample_candidates) + + # Set invalid user input + sample_lex_request.inputTranscript = "invalid" + + with patch.object(disambiguation_handler, "_create_fallback_response") as mock_fallback: + mock_response = Mock() + mock_fallback.return_value = mock_response + + result = disambiguation_handler.process_disambiguation_response(sample_lex_request) + assert result == mock_response + + def test_get_clarification_text_two_options(self, disambiguation_handler): + """Test clarification text for two options.""" + candidates = [ + IntentCandidate("Intent1", 0.7, "Option 1", "Description 1"), + IntentCandidate("Intent2", 0.6, "Option 2", "Description 2"), + ] + + with patch("lex_helper.core.disambiguation.handler.get_message") as mock_get_message: + mock_get_message.return_value = "Two options message" + + result = disambiguation_handler._get_clarification_text(candidates) + + mock_get_message.assert_called_with( + "disambiguation.two_options", "I can help you with two things. Which would you like to do?" + ) + assert result == "Two options message" + + def test_get_clarification_text_multiple_options(self, disambiguation_handler): + """Test clarification text for multiple options.""" + candidates = [ + IntentCandidate("Intent1", 0.7, "Option 1", "Description 1"), + IntentCandidate("Intent2", 0.6, "Option 2", "Description 2"), + IntentCandidate("Intent3", 0.5, "Option 3", "Description 3"), + ] + + with patch("lex_helper.core.disambiguation.handler.get_message") as mock_get_message: + mock_get_message.return_value = "Multiple options message" + + result = disambiguation_handler._get_clarification_text(candidates) + + mock_get_message.assert_called_with( + "disambiguation.multiple_options", "I can help you with several things. What would you like to do?" + ) + assert result == "Multiple options message" + + +if __name__ == "__main__": + pytest.main([__file__]) diff --git a/tests/test_disambiguation_integration.py b/tests/test_disambiguation_integration.py new file mode 100644 index 0000000..87dc98e --- /dev/null +++ b/tests/test_disambiguation_integration.py @@ -0,0 +1,179 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Tests for disambiguation integration with LexHelper handler pipeline. +""" + +from unittest.mock import patch + +from lex_helper.core.disambiguation.types import DisambiguationConfig +from lex_helper.core.handler import Config, LexHelper +from lex_helper.core.types import Bot, DialogAction, Intent, Interpretation, LexRequest, SessionAttributes, SessionState + + +class TestSessionAttributes(SessionAttributes): + """Test session attributes class.""" + + pass + + +class TestDisambiguationIntegration: + """Test disambiguation integration with LexHelper.""" + + def test_disambiguation_disabled_by_default(self): + """Test that disambiguation is disabled by default.""" + config = Config(session_attributes=TestSessionAttributes()) + helper = LexHelper(config) + + assert helper.config.enable_disambiguation is False + assert helper.disambiguation_handler is None + assert helper.disambiguation_analyzer is None + + def test_disambiguation_can_be_enabled(self): + """Test that disambiguation can be enabled with configuration.""" + disambiguation_config = DisambiguationConfig(confidence_threshold=0.5, max_candidates=2) + + config = Config( + session_attributes=TestSessionAttributes(), + enable_disambiguation=True, + disambiguation_config=disambiguation_config, + ) + helper = LexHelper(config) + + assert helper.config.enable_disambiguation is True + assert helper.disambiguation_handler is not None + assert helper.disambiguation_analyzer is not None + assert helper.disambiguation_analyzer.config.confidence_threshold == 0.5 + assert helper.disambiguation_analyzer.config.max_candidates == 2 + + def test_disambiguation_with_default_config(self): + """Test that disambiguation works with default configuration.""" + config = Config(session_attributes=TestSessionAttributes(), enable_disambiguation=True) + helper = LexHelper(config) + + assert helper.disambiguation_handler is not None + assert helper.disambiguation_analyzer.config.confidence_threshold == 0.6 # default + assert helper.disambiguation_analyzer.config.max_candidates == 3 # default + + def test_handler_pipeline_includes_disambiguation(self): + """Test that the handler pipeline includes disambiguation when enabled.""" + config = Config(session_attributes=TestSessionAttributes(), enable_disambiguation=True) + helper = LexHelper(config) + + # Check that disambiguation handler method exists + assert hasattr(helper, "disambiguation_intent_handler") + assert callable(helper.disambiguation_intent_handler) + + def test_handler_pipeline_without_disambiguation(self): + """Test that the handler pipeline works without disambiguation.""" + config = Config(session_attributes=TestSessionAttributes()) + helper = LexHelper(config) + + # Should still have regular handler + assert hasattr(helper, "regular_intent_handler") + assert callable(helper.regular_intent_handler) + + def create_test_lex_request(self, interpretations=None): + """Create a test LexRequest with interpretations.""" + if interpretations is None: + interpretations = [ + Interpretation(intent=Intent(name="BookFlight", slots={}), nluConfidence=0.4), + Interpretation(intent=Intent(name="ChangeFlight", slots={}), nluConfidence=0.3), + ] + + return LexRequest( + sessionId="test-session", + inputTranscript="I want to change my booking", + interpretations=interpretations, + bot=Bot(name="TestBot", localeId="en_US"), + sessionState=SessionState( + intent=Intent(name="BookFlight"), + sessionAttributes=TestSessionAttributes(), + dialogAction=DialogAction(type="ElicitIntent"), + ), + ) + + def test_disambiguation_handler_can_be_called(self): + """Test that disambiguation handler can be called without errors.""" + config = Config(session_attributes=TestSessionAttributes(), enable_disambiguation=True) + helper = LexHelper(config) + + lex_request = self.create_test_lex_request() + + # Should be able to call the handler (may return None if no disambiguation needed) + result = helper.disambiguation_intent_handler(lex_request) + # Result can be None (no disambiguation) or a LexResponse + assert result is None or hasattr(result, "sessionState") + + def test_confidence_analysis_integration(self): + """Test that confidence analysis works through the integration.""" + config = Config(session_attributes=TestSessionAttributes(), enable_disambiguation=True) + helper = LexHelper(config) + + # Test high confidence - should not disambiguate + high_confidence_request = self.create_test_lex_request( + [Interpretation(intent=Intent(name="BookFlight", slots={}), nluConfidence=0.9)] + ) + + analysis = helper.disambiguation_analyzer.analyze_request(high_confidence_request) + + assert not analysis.should_disambiguate + + # Test low confidence - should disambiguate + low_confidence_request = self.create_test_lex_request( + [ + Interpretation(intent=Intent(name="BookFlight", slots={}), nluConfidence=0.4), + Interpretation(intent=Intent(name="ChangeFlight", slots={}), nluConfidence=0.3), + ] + ) + + analysis = helper.disambiguation_analyzer.analyze_request(low_confidence_request) + + assert analysis.should_disambiguate + assert len(analysis.candidates) > 0 + + def test_fallback_when_disambiguation_unavailable(self): + """Test that system falls back gracefully when disambiguation is unavailable.""" + # Mock the import to simulate disambiguation not being available + with patch("lex_helper.core.handler.disambiguation_available", False): + config = Config( + session_attributes=TestSessionAttributes(), + enable_disambiguation=True, # Request disambiguation but it's not available + ) + helper = LexHelper(config) + + # Should fall back to no disambiguation + assert helper.disambiguation_handler is None + assert helper.disambiguation_analyzer is None + + def test_seamless_fallback_to_existing_behavior(self): + """Test that when disambiguation is disabled, behavior is unchanged.""" + # Create two helpers - one with and one without disambiguation + config_without = Config(session_attributes=TestSessionAttributes()) + helper_without = LexHelper(config_without) + + config_with = Config(session_attributes=TestSessionAttributes(), enable_disambiguation=True) + helper_with = LexHelper(config_with) + + # Both should have regular intent handler + assert hasattr(helper_without, "regular_intent_handler") + assert hasattr(helper_with, "regular_intent_handler") + + # Only the enabled one should have disambiguation components + assert helper_without.disambiguation_handler is None + assert helper_with.disambiguation_handler is not None + + def test_config_validation(self): + """Test that configuration is properly validated.""" + # Test with custom disambiguation config + custom_config = DisambiguationConfig(confidence_threshold=0.8, max_candidates=5) + + config = Config( + session_attributes=TestSessionAttributes(), enable_disambiguation=True, disambiguation_config=custom_config + ) + helper = LexHelper(config) + + # Verify the custom config is used + assert helper.disambiguation_analyzer.config.confidence_threshold == 0.8 + assert helper.disambiguation_analyzer.config.max_candidates == 5