Spaces:
Running
Running
| """ | |
| Simple standalone test for intelligent reply functionality | |
| """ | |
| from typing import Optional, Dict, Any, List | |
| def determine_reply_to_message( | |
| conversation_history: List[Dict[str, Any]], | |
| reply_context: Optional[Dict[str, Any]] = None | |
| ) -> Optional[str]: | |
| """ | |
| Analyze conversation and determine which message AIDA should quote. | |
| Strategy: | |
| 1. If user just replied to a specific message (reply_context exists), | |
| quote the LATEST USER MESSAGE, not the message they swiped | |
| 2. Otherwise, quote the most recent user message | |
| Args: | |
| conversation_history: List of messages with structure: | |
| [{"id": str, "sender_id": str, "content": str, "is_ai": bool, ...}, ...] | |
| reply_context: Optional context of message user replied to | |
| Returns: | |
| Message ID to quote, or None if no appropriate message found | |
| """ | |
| if not conversation_history: | |
| return None | |
| # Find the most recent user message (not AI) | |
| latest_user_message = None | |
| for msg in conversation_history: | |
| if not msg.get("is_ai", False): | |
| latest_user_message = msg | |
| break # conversation_history is sorted newest first | |
| if latest_user_message: | |
| return latest_user_message.get("id") | |
| return None | |
| def build_reply_context_for_response( | |
| conversation_history: List[Dict[str, Any]], | |
| reply_to_id: str | |
| ) -> Optional[Dict[str, Any]]: | |
| """ | |
| Build reply context for AIDA's response. | |
| Returns: | |
| { | |
| "id": message_id, | |
| "sender_name": sender name, | |
| "content": message content | |
| } | |
| """ | |
| if not reply_to_id: | |
| return None | |
| # Find the message to quote | |
| for msg in conversation_history: | |
| if msg.get("id") == reply_to_id: | |
| return { | |
| "id": msg.get("id"), | |
| "sender_name": msg.get("sender_name", "User"), | |
| "content": msg.get("content", ""), | |
| } | |
| return None | |
| def test_determine_reply_to_message(): | |
| """Test the reply message determination logic""" | |
| print("\n" + "="*60) | |
| print("TEST 1: Simple conversation - should quote latest user message") | |
| print("="*60) | |
| conversation_history = [ | |
| { | |
| "id": "msg-003", | |
| "sender_id": "user-123", | |
| "sender_name": "John Doe", | |
| "content": "I need 2 bedrooms", | |
| "is_ai": False, | |
| "timestamp": "2026-02-15T10:03:00Z" | |
| }, | |
| { | |
| "id": "msg-002", | |
| "sender_id": "aida", | |
| "sender_name": "AIDA", | |
| "content": "Hello! How can I help?", | |
| "is_ai": True, | |
| "timestamp": "2026-02-15T10:02:00Z" | |
| }, | |
| { | |
| "id": "msg-001", | |
| "sender_id": "user-123", | |
| "sender_name": "John Doe", | |
| "content": "Hi", | |
| "is_ai": False, | |
| "timestamp": "2026-02-15T10:01:00Z" | |
| } | |
| ] | |
| reply_to_id = determine_reply_to_message(conversation_history) | |
| print(f"\nResult: reply_to_id = {reply_to_id}") | |
| print(f"Expected: msg-003 (latest user message)") | |
| assert reply_to_id == "msg-003", f"Expected msg-003, got {reply_to_id}" | |
| print("PASS: TEST PASSED") | |
| def test_build_reply_context(): | |
| """Test the reply context building logic""" | |
| print("\n" + "="*60) | |
| print("TEST 2: Build reply context for response") | |
| print("="*60) | |
| conversation_history = [ | |
| { | |
| "id": "msg-003", | |
| "sender_id": "user-123", | |
| "sender_name": "John Doe", | |
| "content": "I need 2 bedrooms in Cotonou", | |
| "is_ai": False, | |
| "timestamp": "2026-02-15T10:03:00Z" | |
| }, | |
| { | |
| "id": "msg-002", | |
| "sender_id": "aida", | |
| "sender_name": "AIDA", | |
| "content": "Hello! How can I help?", | |
| "is_ai": True, | |
| "timestamp": "2026-02-15T10:02:00Z" | |
| } | |
| ] | |
| reply_context = build_reply_context_for_response( | |
| conversation_history=conversation_history, | |
| reply_to_id="msg-003" | |
| ) | |
| print(f"\nResult:") | |
| print(f" - id: {reply_context['id']}") | |
| print(f" - sender_name: {reply_context['sender_name']}") | |
| print(f" - content: {reply_context['content']}") | |
| assert reply_context['id'] == "msg-003" | |
| assert reply_context['sender_name'] == "John Doe" | |
| assert reply_context['content'] == "I need 2 bedrooms in Cotonou" | |
| print("\nPASS: TEST PASSED") | |
| def test_empty_conversation(): | |
| """Test with empty conversation history""" | |
| print("\n" + "="*60) | |
| print("TEST 3: Empty conversation - should return None") | |
| print("="*60) | |
| reply_to_id = determine_reply_to_message([]) | |
| print(f"\nResult: reply_to_id = {reply_to_id}") | |
| print(f"Expected: None") | |
| assert reply_to_id is None | |
| print("PASS: TEST PASSED") | |
| def test_only_ai_messages(): | |
| """Test with only AI messages - should return None""" | |
| print("\n" + "="*60) | |
| print("TEST 4: Only AI messages - should return None") | |
| print("="*60) | |
| conversation_history = [ | |
| { | |
| "id": "msg-002", | |
| "sender_id": "aida", | |
| "sender_name": "AIDA", | |
| "content": "Hello!", | |
| "is_ai": True, | |
| "timestamp": "2026-02-15T10:02:00Z" | |
| }, | |
| { | |
| "id": "msg-001", | |
| "sender_id": "aida", | |
| "sender_name": "AIDA", | |
| "content": "How can I help?", | |
| "is_ai": True, | |
| "timestamp": "2026-02-15T10:01:00Z" | |
| } | |
| ] | |
| reply_to_id = determine_reply_to_message(conversation_history) | |
| print(f"\nResult: reply_to_id = {reply_to_id}") | |
| print(f"Expected: None") | |
| assert reply_to_id is None | |
| print("PASS: TEST PASSED") | |
| def test_with_reply_context(): | |
| """Test when user swipes to reply to a specific message""" | |
| print("\n" + "="*60) | |
| print("TEST 5: User swipes to reply - should still quote latest user message") | |
| print("="*60) | |
| conversation_history = [ | |
| { | |
| "id": "msg-005", | |
| "sender_id": "user-123", | |
| "sender_name": "John Doe", | |
| "content": "I need 2 bedrooms", | |
| "is_ai": False, | |
| "timestamp": "2026-02-15T10:05:00Z" | |
| }, | |
| { | |
| "id": "msg-004", | |
| "sender_id": "aida", | |
| "sender_name": "AIDA", | |
| "content": "I can help with that!", | |
| "is_ai": True, | |
| "timestamp": "2026-02-15T10:04:00Z" | |
| }, | |
| { | |
| "id": "msg-003", | |
| "sender_id": "user-123", | |
| "sender_name": "John Doe", | |
| "content": "Show me apartments", | |
| "is_ai": False, | |
| "timestamp": "2026-02-15T10:03:00Z" | |
| }, | |
| { | |
| "id": "msg-002", | |
| "sender_id": "aida", | |
| "sender_name": "AIDA", | |
| "content": "Hello! How can I help?", | |
| "is_ai": True, | |
| "timestamp": "2026-02-15T10:02:00Z" | |
| }, | |
| { | |
| "id": "msg-001", | |
| "sender_id": "user-123", | |
| "sender_name": "John Doe", | |
| "content": "Hi", | |
| "is_ai": False, | |
| "timestamp": "2026-02-15T10:01:00Z" | |
| } | |
| ] | |
| # User swiped "Hi" but the latest user message is "I need 2 bedrooms" | |
| reply_context = {"message_id": "msg-001", "content": "Hi"} | |
| reply_to_id = determine_reply_to_message( | |
| conversation_history=conversation_history, | |
| reply_context=reply_context | |
| ) | |
| print(f"\nResult: reply_to_id = {reply_to_id}") | |
| print(f"Expected: msg-005 (latest user message, NOT msg-001 which was swiped)") | |
| assert reply_to_id == "msg-005", f"Expected msg-005, got {reply_to_id}" | |
| print("PASS: TEST PASSED") | |
| def main(): | |
| """Run all tests""" | |
| print("\n" + "="*60) | |
| print("INTELLIGENT REPLY TESTS") | |
| print("="*60) | |
| try: | |
| test_determine_reply_to_message() | |
| test_build_reply_context() | |
| test_empty_conversation() | |
| test_only_ai_messages() | |
| test_with_reply_context() | |
| print("\n" + "="*60) | |
| print("ALL TESTS PASSED!") | |
| print("="*60 + "\n") | |
| return 0 | |
| except AssertionError as e: | |
| print(f"\nTEST FAILED: {e}\n") | |
| return 1 | |
| except Exception as e: | |
| print(f"\nERROR: {e}\n") | |
| import traceback | |
| traceback.print_exc() | |
| return 1 | |
| if __name__ == "__main__": | |
| import sys | |
| sys.exit(main()) | |