diff --git a/TESTING.md b/TESTING.md index edadea8..fb83603 100644 --- a/TESTING.md +++ b/TESTING.md @@ -28,7 +28,6 @@ uv run pytest tests_integ/memory/test_memory_client.py -xvs # Client and developer experience uv run pytest tests_integ/memory/test_memory_client.py -xvs -uv run pytest tests_integ/memory/test_devex.py -xvs ``` ## Identity diff --git a/tests_integ/memory/test_devex.py b/tests_integ/memory/test_devex.py deleted file mode 100644 index 720d9b5..0000000 --- a/tests_integ/memory/test_devex.py +++ /dev/null @@ -1,756 +0,0 @@ -"""Comprehensive developer experience evaluation for Bedrock AgentCore Memory SDK.""" - -import os -import sys - -sys.path.append(os.path.join(os.path.dirname(__file__), "../../src")) - -import json -import logging -import time -from datetime import datetime - -from bedrock_agentcore.memory import MemoryClient - -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") -logger = logging.getLogger(__name__) - - -def print_developer_journey(): - """Print the developer journey to understand the improvements.""" - - logger.info("=" * 80) - logger.info("DEVELOPER EXPERIENCE JOURNEY") - logger.info("=" * 80) - - logger.info("\nšŸ“– STORY: Building a Customer Support Agent") - logger.info("A developer wants to build an AI agent that:") - logger.info("- Handles customer inquiries") - logger.info("- Can explore different response strategies") - logger.info("- Escalates to human agents when needed") - logger.info("- Learns from interactions") - - logger.info("- save_conversation() handles any message pattern") - logger.info("- Full branch management (list, navigate, visualize)") - logger.info("- Flexible roles for tools and system messages") - logger.info("- Memory extraction for learning") - - -def test_complete_agent_workflow(client: MemoryClient, memory_id: str): - """Test a complete customer support agent workflow.""" - - logger.info("\n%s", "=" * 80) - logger.info("COMPLETE AGENT WORKFLOW TEST") - logger.info("=" * 80) - - actor_id = "customer-%s" % datetime.now().strftime("%Y%m%d%H%M%S") - session_id = "support-%s" % datetime.now().strftime("%Y%m%d%H%M%S") - - logger.info("\n1. Memory strategies already configured during creation") - - # Helper function for retries with exponential backoff - def save_with_retry(memory_id, actor_id, session_id, messages, branch=None, max_retries=5): - wait_time = 2 # Start with 2 seconds - attempt = 0 - - while attempt < max_retries: - try: - return client.save_conversation( - memory_id=memory_id, actor_id=actor_id, session_id=session_id, messages=messages, branch=branch - ) - except Exception as e: - if "ThrottledException" in str(e) and attempt < max_retries - 1: - attempt += 1 - logger.info( - "Rate limit hit, retrying in %d seconds (attempt %d/%d)...", wait_time, attempt, max_retries - ) - time.sleep(wait_time) - wait_time *= 2 # Exponential backoff - else: - raise # Re-raise if it's not a throttling error or max retries reached - - # Phase 1: Initial inquiry with context switching - logger.info("\n2. Customer makes initial inquiry...") - - initial = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("Hi, I'm having trouble with my order #12345", "USER"), - ("I'm sorry to hear that. Let me look up your order.", "ASSISTANT"), - ("lookup_order(order_id='12345')", "TOOL"), - ("I see your order was shipped 3 days ago. What specific issue are you experiencing?", "ASSISTANT"), - ("Actually, before that - I also want to change my email address", "USER"), - ( - "Of course! I can help with both. Let's start with updating your email. What's your new email?", - "ASSISTANT", - ), - ("newemail@example.com", "USER"), - ("update_customer_email(old='old@example.com', new='newemail@example.com')", "TOOL"), - ("Email updated successfully! Now, about your order issue?", "ASSISTANT"), - ("The package arrived damaged", "USER"), - ], - ) - logger.info("āœ“ Handled context switch naturally") - - # Phase 2: A/B test different resolution approaches - logger.info("\n3. Testing different resolution strategies...") - - # MODIFIED: Create refund branch with first message only - _refund_branch = client.fork_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - root_event_id=initial["eventId"], - branch_name="immediate-refund", - new_messages=[ - ("I'm very sorry about the damaged package. I'll process an immediate refund.", "ASSISTANT"), - ], - ) - - # Continue the refund branch with additional messages - with longer delays and retries - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("process_refund(order_id='12345', reason='damaged', amount='full')", "TOOL"), - ], - branch={"name": "immediate-refund", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("Refund processed! You'll see it in 3-5 business days. Is there anything else?", "ASSISTANT"), - ("That was fast, thank you!", "USER"), - ], - branch={"name": "immediate-refund", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("You're welcome! I've also added a 10% discount to your account for next purchase.", "ASSISTANT"), - ], - branch={"name": "immediate-refund", "rootEventId": initial["eventId"]}, - ) - - # MODIFIED: Create replacement branch with first message only - time.sleep(5) # Increased delay - _replacement_branch = client.fork_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - root_event_id=initial["eventId"], - branch_name="replacement-offer", - new_messages=[ - ("I apologize for the damaged item. Would you prefer a replacement or refund?", "ASSISTANT"), - ], - ) - - # Continue the replacement branch with additional messages - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("How fast can you send a replacement?", "USER"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("check_inventory(item='ORD-12345-ITEM')", "TOOL"), - ("We have it in stock! I can send a replacement with express shipping - arrives in 2 days.", "ASSISTANT"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("That works for me", "USER"), - ("create_replacement_order(original='12345', shipping='express')", "TOOL"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("Perfect! Replacement ordered with express shipping. You'll get tracking info shortly.", "ASSISTANT"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - # MODIFIED: Create escalation branch with first message only - time.sleep(5) # Increased delay - _escalation_branch = client.fork_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - root_event_id=initial["eventId"], - branch_name="escalation-required", - new_messages=[ - ("I understand this is frustrating. Let me connect you with a specialist who can help.", "ASSISTANT"), - ], - ) - - # Continue the escalation branch with additional messages - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("This is the third time this has happened!", "USER"), - ], - branch={"name": "escalation-required", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("check_customer_history(customer_id='cust-123')", "TOOL"), - ( - "I see you've had multiple issues. I'm escalating this to our senior support team immediately.", - "ASSISTANT", - ), - ], - branch={"name": "escalation-required", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("create_escalation_ticket(priority='high', history='multiple_damages')", "TOOL"), - ("ticket_created: ESC-78901", "TOOL"), - ], - branch={"name": "escalation-required", "rootEventId": initial["eventId"]}, - ) - - time.sleep(5) # Increased delay - save_with_retry( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ( - "I've created high-priority ticket ESC-78901. A senior specialist will contact you within 1 hour.", - "ASSISTANT", - ), - ], - branch={"name": "escalation-required", "rootEventId": initial["eventId"]}, - ) - - logger.info("āœ“ Created 3 different resolution branches") - - # Phase 3: Analyze branches - logger.info("\n4. Analyzing branch outcomes...") - - branches = client.list_branches(memory_id, actor_id, session_id) - logger.info("\nFound %d total branches:", len(branches)) - - for branch in branches: - logger.info("\n Branch: %s", branch["name"]) - logger.info(" Events: %d", branch["eventCount"]) - - if branch["name"] != "main": - messages = client.merge_branch_context( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - branch_name=branch["name"], - include_parent=False, - ) - - if messages: - last_customer = None - last_agent = None - - for msg in reversed(messages): - if msg["role"] == "USER" and not last_customer: - last_customer = msg["content"] - elif msg["role"] == "ASSISTANT" and not last_agent: - last_agent = msg["content"] - - if last_customer and last_agent: - break - - logger.info(" Customer sentiment: %s", last_customer[:50] if last_customer else "N/A") - logger.info(" Final resolution: %s", last_agent[:80] + "..." if last_agent else "N/A") - - # Phase 4: Continue in best branch - logger.info("\n5. Continuing conversation in best branch...") - - # MODIFIED: Split follow-up into smaller batches - time.sleep(1) - client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("I got the replacement - it's perfect! Thank you so much!", "USER"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - time.sleep(1) - client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("Wonderful! I'm glad we could resolve this quickly.", "ASSISTANT"), - ("save_positive_feedback(case_id='12345', rating=5, branch='replacement')", "TOOL"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - time.sleep(1) - _followup = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("Is there anything else I can help you with today?", "ASSISTANT"), - ("No, that's all. Great service!", "USER"), - ("Thank you! Have a great day!", "ASSISTANT"), - ], - branch={"name": "replacement-offer", "rootEventId": initial["eventId"]}, - ) - - logger.info("āœ“ Continued conversation in successful branch") - - # Phase 5: Wait for memory extraction - logger.info("\n6. Waiting for memory extraction...") - logger.info("Note: After creating events, extraction + vector indexing typically takes 2-3 minutes") - - logger.info("Waiting 30 seconds for extraction to trigger...") - time.sleep(30) - - namespace = "support/facts/%s/" % session_id - if client.wait_for_memories(memory_id, namespace, max_wait=180): - logger.info("āœ“ Memories extracted and indexed successfully") - - memories = client.retrieve_memories( - memory_id=memory_id, namespace=namespace, query="customer order issues damaged package", top_k=5 - ) - - logger.info("Retrieved %d relevant memories", len(memories)) - for i, mem in enumerate(memories[:3]): - logger.info(" [%d] %s", i + 1, mem.get("content", {}).get("text", "")[:100]) - else: - logger.info("āš ļø Memory extraction/indexing still in progress") - logger.info("This can take 3-5 minutes total. Try retrieving memories manually later.") - - # Phase 6: Visualize complete conversation - logger.info("\n7. Visualizing conversation structure...") - - tree = client.get_conversation_tree(memory_id, actor_id, session_id) - - def print_tree(branch_data, indent=0): - prefix = " " * indent - events = branch_data.get("events", []) - - if events: - logger.info("%sMain flow: %d events", prefix, len(events)) - for event in events[:2]: - for msg in event.get("messages", []): - logger.info("%s - %s: %s", prefix, msg["role"], msg["text"]) - - for branch_name, sub_branch in branch_data.get("branches", {}).items(): - logger.info("%sā””ā”€ Branch '%s': %d events", prefix, branch_name, len(sub_branch.get("events", []))) - if sub_branch.get("events"): - for msg in sub_branch["events"][0].get("messages", []): - logger.info("%s - %s: %s", prefix, msg["role"], msg["text"]) - - print_tree(tree["main_branch"]) - - -def test_bedrock_integration(client: MemoryClient, memory_id: str): - """Test AgentCore Memory with Amazon Bedrock integration.""" - - logger.info("\n%s", "=" * 80) - logger.info("TESTING BEDROCK INTEGRATION") - logger.info("=" * 80) - - import boto3 - - try: - bedrock = boto3.client("bedrock-runtime", region_name="us-east-1") - except Exception as e: - logger.error("Failed to initialize Bedrock client: %s", e) - logger.info("Skipping Bedrock test - ensure AWS credentials are configured") - return - - actor_id = "bedrock-test-%s" % datetime.now().strftime("%Y%m%d%H%M%S") - session_id = "bedrock-session-%s" % datetime.now().strftime("%Y%m%d%H%M%S") - - # Create initial context - logger.info("\n1. Creating initial conversation context...") - - _initial_events = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("I'm planning a trip to Japan in April", "USER"), - ("That's exciting! April is cherry blossom season. What cities are you planning to visit?", "ASSISTANT"), - ("Tokyo and Kyoto for sure. I love photography", "USER"), - ("Perfect for photography! The cherry blossoms in Maruyama Park in Kyoto are stunning.", "ASSISTANT"), - ], - ) - - # Wait for extraction - logger.info("\n2. Waiting for memory extraction...") - time.sleep(60) - - # New user query - user_query = "What camera equipment should I bring for cherry blossom photography?" - logger.info("\n3. New user query: %s", user_query) - - # Retrieve relevant memories - logger.info("\n4. Retrieving relevant context...") - namespace = "support/facts/%s/" % session_id - memories = client.retrieve_memories(memory_id=memory_id, namespace=namespace, query=user_query, top_k=5) - - context = "" - if memories: - context = "\n".join([m.get("content", {}).get("text", "") for m in memories]) - logger.info("Found %d relevant memories", len(memories)) - - # Call Bedrock with context - logger.info("\n5. Calling Claude 3.5 Sonnet with context...") - - messages = [] - if context: - messages.append( - {"role": "assistant", "content": "Here's what I know from our previous conversation:\n%s" % context} - ) - - messages.append({"role": "user", "content": user_query}) - - try: - response = bedrock.invoke_model( - modelId="anthropic.claude-3-5-sonnet-20241022-v2:0", - contentType="application/json", - accept="application/json", - body=json.dumps( - { - "anthropic_version": "bedrock-2023-05-31", - "max_tokens": 1000, - "messages": messages, - "temperature": 0.7, - } - ), - ) - - response_body = json.loads(response["body"].read()) - llm_response = response_body["content"][0]["text"] - - logger.info("\n6. Claude's response:") - logger.info("%s...", llm_response[:200]) - - # Save the new turn - logger.info("\n7. Saving conversation turn...") - _new_event = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[(user_query, "USER"), (llm_response, "ASSISTANT")], - ) - - logger.info("āœ“ Successfully integrated Memory with Bedrock!") - - except Exception as e: - logger.error("Bedrock call failed: %s", e) - logger.info("Make sure you have access to Claude 3.5 Sonnet v2 in Bedrock") - - -def test_developer_productivity_metrics(client: MemoryClient, memory_id: str): - """Measure developer productivity improvements.""" - - logger.info("\n%s", "=" * 80) - logger.info("DEVELOPER PRODUCTIVITY METRICS") - logger.info("=" * 80) - - _actor_id = "metrics-test" - _session_id = "metrics-session" - - logger.info("\n1. Lines of Code Comparison") - logger.info("\nFlexible conversation handling:") - logger.info(" event = client.save_conversation(messages=[") - logger.info(" ('Question 1', 'USER'),") - logger.info(" ('Question 2', 'USER'),") - logger.info(" ('Checking...', 'ASSISTANT'),") - logger.info(" ('tool_call()', 'TOOL'),") - logger.info(" ('Complete answer', 'ASSISTANT')") - logger.info(" ])") - logger.info(" Total: 7 lines for complex flow") - - logger.info("\n2. API Calls for Common Tasks") - logger.info(" Get conversation history from branch: 1 call - list_branch_events()") - logger.info(" Find all branches: 1 call - list_branches()") - logger.info(" Save complex interaction: 1 call - save_conversation()") - - logger.info("\n3. Key Improvements") - logger.info(" āœ… Natural message flow representation") - logger.info(" āœ… Complete branch navigation") - logger.info(" āœ… Flexible message combinations") - logger.info(" āœ… Type-safe strategy methods") - - features = [ - ("Save user question without response", "30 seconds"), - ("Handle tool-augmented response", "1 minute"), - ("A/B test responses with branches", "2 minutes"), - ("Get branch conversation", "30 seconds"), - ("Find all branches", "1 API call"), - ] - - logger.info("\n4. Feature Implementation Time") - logger.info("\nFeature Time to Implement ") - logger.info("-" * 55) - for feature, impl_time in features: - logger.info("%-35s %-20s", feature, impl_time) - - -def test_edge_cases_and_validation(client: MemoryClient, memory_id: str): - """Test edge cases and validation improvements.""" - - logger.info("\n%s", "=" * 80) - logger.info("EDGE CASES AND VALIDATION") - logger.info("=" * 80) - - actor_id = "edge-test" - session_id = "edge-session" - - # Test 1: Very long conversation - logger.info("\n1. Testing very long conversation...") - - # MODIFIED: Split long conversation into smaller batches - for i in range(20): - messages = [] - messages.append(("Question %d about the product" % i, "USER")) - messages.append(("Answer %d with detailed information" % i, "ASSISTANT")) - - try: - long_event = client.save_conversation( - memory_id=memory_id, actor_id=actor_id, session_id=session_id, messages=messages - ) - logger.info("āœ“ Saved messages %d: %s", i + 1, long_event["eventId"]) - time.sleep(0.5) # Small delay between batches - except Exception as e: - logger.error("āŒ Failed to save messages %d: %s", i + 1, e) - - logger.info("āœ“ Saved long conversation in batches") - - # Test 2: Rapid branch creation - logger.info("\n2. Testing rapid branch creation...") - - base_event = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id="rapid-branch-test", - messages=[("Start conversation", "USER")], - ) - - # MODIFIED: Added delays between branch creations - for i in range(5): - try: - time.sleep(1) # Delay before creating branch - _branch = client.fork_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id="rapid-branch-test", - root_event_id=base_event["eventId"], - branch_name="branch-%d" % i, - new_messages=[("Branch %d message" % i, "ASSISTANT")], - ) - logger.info("āœ“ Created branch-%d", i) - except Exception as e: - logger.error("āŒ Failed to create branch-%d: %s", i, e) - - # Test 3: Unicode and special characters - logger.info("\n3. Testing Unicode and special characters...") - - # MODIFIED: Split into smaller message groups - time.sleep(1) - _special_event = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("Hello! šŸ‘‹ How can I help? 你儽ļ¼", "ASSISTANT"), - ], - ) - - time.sleep(1) - _special_event2 = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[ - ("I need help with ā‚¬100 payment", "USER"), - ("I'll help with your ā‚¬100 payment šŸ’³", "ASSISTANT"), - ], - ) - - logger.info("āœ“ Handled Unicode and special characters") - - # Test 4: Empty messages - logger.info("\n4. Testing empty message content...") - - try: - time.sleep(1) - _empty_event = client.save_conversation( - memory_id=memory_id, - actor_id=actor_id, - session_id=session_id, - messages=[("", "USER"), ("I didn't catch that. Could you repeat?", "ASSISTANT")], - ) - logger.info("āœ“ Handled empty message content") - except Exception as e: - logger.error("āŒ Failed with empty message: %s", e) - - -def generate_developer_report(client: MemoryClient): - """Generate a final developer experience report.""" - - logger.info("\n%s", "=" * 80) - logger.info("DEVELOPER EXPERIENCE REPORT") - logger.info("=" * 80) - - logger.info("\nšŸŽÆ KEY IMPROVEMENTS") - - improvements = [ - {"area": "Conversation Flexibility", "impact": "90% reduction in code for complex flows"}, - {"area": "Branch Management", "impact": "New scenarios now possible"}, - {"area": "Developer Intuition", "impact": "Faster onboarding, fewer errors"}, - {"area": "Real-world Scenarios", "impact": "Better user experiences"}, - ] - - for imp in improvements: - logger.info("\n%s:", imp["area"]) - logger.info(" Impact: %s", imp["impact"]) - - logger.info("\nšŸ“Š METRICS SUMMARY") - logger.info(" ā€¢ Code reduction: 60-90% for complex scenarios") - logger.info(" ā€¢ New capabilities: 5+ previously impossible features") - logger.info(" ā€¢ API calls saved: 50-80% for multi-message flows") - logger.info(" ā€¢ Learning curve: Significantly reduced") - - logger.info("\nāœ… RECOMMENDATION") - logger.info("The SDK improvements successfully address developer pain points.") - logger.info("Developers can now build more sophisticated agents with less code.") - logger.info("Branch management enables new use cases like A/B testing.") - logger.info("The flexible conversation API matches real-world requirements.") - - -def main(): - """Run complete developer experience evaluation.""" - - print_developer_journey() - - role_arn = os.getenv("MEMORY_ROLE_ARN") - if not role_arn: - logger.error("Please set MEMORY_ROLE_ARN environment variable") - return - - # Get region and environment from environment variables with defaults - region = os.getenv("AWS_REGION", "us-west-2") - environment = os.getenv("MEMORY_ENVIRONMENT", "prod") - - logger.info("Using region: %s, environment: %s", region, environment) - - client = MemoryClient(region_name=region) - - logger.info("\nCreating test memory with strategies...") - memory = client.create_memory( - name="DXTest_%s" % datetime.now().strftime("%Y%m%d%H%M%S"), - description="Developer experience evaluation", - strategies=[ - { - "semanticMemoryStrategy": { - "name": "CustomerInfo", - "description": "Extract customer information and issues", - "namespaces": ["support/facts/{sessionId}/"], - # NO configuration block - } - }, - { - "userPreferenceMemoryStrategy": { - "name": "CustomerPreferences", - "description": "Track customer preferences and history", - "namespaces": ["customers/{actorId}/preferences/"], - # NO configuration block - } - }, - ], - event_expiry_days=7, - memory_execution_role_arn=role_arn, - ) - - memory_id = memory["memoryId"] - logger.info("Created memory: %s", memory_id) - - logger.info("Waiting for memory activation...") - for _ in range(30): - time.sleep(10) - status = client.get_memory_status(memory_id) - if status == "ACTIVE": - logger.info("Memory is active!") - logger.info("Waiting additional 120 seconds for vector store initialization...") - time.sleep(120) - break - elif status == "FAILED": - logger.error("Memory creation failed!") - return - - try: - test_complete_agent_workflow(client, memory_id) - test_bedrock_integration(client, memory_id) - test_developer_productivity_metrics(client, memory_id) - test_edge_cases_and_validation(client, memory_id) - generate_developer_report(client) - - logger.info("\n%s", "=" * 80) - logger.info("DEVELOPER EXPERIENCE EVALUATION COMPLETE") - logger.info("=" * 80) - - except Exception as e: - logger.exception("Test failed: %s", e) - finally: - logger.info("\nTest memory ID: %s", memory_id) - logger.info("You can delete it with: client.delete_memory('%s')", memory_id) - - -if __name__ == "__main__": - main()