IRIS Legacy

#!/usr/bin/env python3 """ 🚀 TEST E2E COMPLETO - MILESTONE 2: PROMPT SYSTEM Simula flussi utente reali end-to-end con tutti i componenti integrati """ import sys import os import asyncio from pathlib import Path from datetime import datetime, timezone # Aggiungi il path src per gli import sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from llm_core import IRISClient, get_iris_client, quick_chat from llm_core.prompts import PromptType from redis_client.client import RedisClient async def scenario_1_business_ceo_daily_brief(): """ SCENARIO 1: CEO che inizia la giornata con IRIS - Conversazione business assistant - Analisi email importante - Richiesta insights strategici """ print("🏢 SCENARIO 1: CEO Daily Brief") print("-" * 50) try: # Setup Redis redis_client = RedisClient() await redis_client.connect() # Inizializza IRIS Client client = IRISClient( default_provider="anthropic", conversation_manager=None # Userà quello di default ) # Profilo CEO user_profile = { "role": "CEO", "company": "TechCorp Italia", "communication_style": "professional", "expertise_level": "expert", "preferred_response_length": "medium" } # 1. Inizia conversazione business session = await client.start_conversation( user_id="marco_rossi_ceo", user_name="Marco Rossi", prompt_type=PromptType.BUSINESS_ASSISTANT, **user_profile ) print(f"✅ Session created: {session.session_id}") # 2. Saluto iniziale e richiesta brief response1 = await client.chat( session.session_id, "Buongiorno IRIS! Sono appena arrivato in ufficio. Puoi darmi un brief della situazione e delle priorità per oggi?", max_tokens=300 ) print("📤 CEO: Buongiorno IRIS! Brief della situazione...") print(f"🤖 IRIS: {response1.content[:200]}...") # 3. Richiesta analisi specifica response2 = await client.chat( session.session_id, "Perfetto. Ora ho ricevuto un'email importante da un investitore. Puoi aiutarmi ad analizzarla e preparare una strategia di risposta?", max_tokens=250 ) print("📤 CEO: Email da investitore, analisi e strategia...") print(f"🤖 IRIS: {response2.content[:200]}...") # 4. Verifica storico conversazione history = await client.get_conversation_history(session.session_id) assert len(history) >= 4 # System + 2 user + 2 assistant # 5. Statistiche sessione stats = await client.get_session_stats(session.session_id) print(f"📊 Session stats: {stats['message_count']} messages, {stats['total_tokens']} tokens") # 6. Chiudi conversazione await client.end_conversation(session.session_id) await redis_client.disconnect() print("✅ SCENARIO 1 COMPLETED") return True except Exception as e: print(f"❌ SCENARIO 1 FAILED: {e}") return False async def scenario_2_multi_user_personalization(): """ SCENARIO 2: Personalizzazione per utenti diversi - Stesso argomento, utenti con profili diversi - Verifica personalizzazione prompt """ print("\n👥 SCENARIO 2: Multi-User Personalization") print("-" * 50) try: # Utente 1: Principiante, stile casual response_beginner = await quick_chat( user_id="mario_intern", user_name="Mario Verdi", message="Spiegami cos'è l'intelligenza artificiale e come può aiutare la nostra azienda", prompt_type=PromptType.GENERAL_CHAT, role="Intern", expertise_level="beginner", communication_style="casual", preferred_response_length="short" ) # Utente 2: Esperto, stile formale response_expert = await quick_chat( user_id="anna_cto", user_name="Dr. Anna Rossi", message="Spiegami cos'è l'intelligenza artificiale e come può aiutare la nostra azienda", prompt_type=PromptType.GENERAL_CHAT, role="CTO", expertise_level="expert", communication_style="formal", preferred_response_length="long" ) print("📤 Question: Spiegami cos'è l'AI...") print(f"👶 Beginner (Mario): {response_beginner[:150]}...") print(f"🎓 Expert (Anna): {response_expert[:150]}...") # Verifica che le risposte siano diverse assert response_beginner != response_expert assert len(response_expert) > len(response_beginner) # Expert dovrebbe essere più lungo print("✅ SCENARIO 2 COMPLETED") return True except Exception as e: print(f"❌ SCENARIO 2 FAILED: {e}") return False async def scenario_3_email_analysis_workflow(): """ SCENARIO 3: Workflow completo analisi email - Ricezione email complessa - Analisi dettagliata - Raccomandazioni strategiche """ print("\n📧 SCENARIO 3: Email Analysis Workflow") print("-" * 50) try: # Setup redis_client = RedisClient() await redis_client.connect() client = IRISClient(default_provider="anthropic") # Email complessa da analizzare complex_email = { "sender": "sarah.johnson@vcfund.com", "recipient": "marco.rossi@techcorp.it", "subject": "Re: Series A Investment - Due Diligence Requirements", "content": """Dear Marco, Following our meeting last week, our investment committee has approved moving forward with the due diligence process for TechCorp's Series A round. We're prepared to lead a €3M investment at a €15M pre-money valuation, subject to satisfactory completion of due diligence. Required documents by March 15th: 1. Audited financial statements (2022-2023) 2. Customer contracts and revenue recognition details 3. IP portfolio and patent filings 4. Technical architecture documentation 5. Employee equity and option pool details 6. Competitive analysis and market positioning We'd also like to schedule calls with: - Your head of engineering - Two key customers - Your legal counsel Timeline: Complete due diligence by March 30th, with term sheet signing by April 15th. Please confirm your availability for a follow-up call this Thursday at 2 PM CET. Best regards, Sarah Johnson Partner, VCFund""", "date": datetime.now(timezone.utc), "priority": "urgent" } # Crea sessione per analisi email session = await client.start_conversation( user_id="marco_ceo", user_name="Marco Rossi", prompt_type=PromptType.EMAIL_ANALYSIS, role="CEO", company="TechCorp Italia", expertise_level="expert" ) # Simula invio email per analisi # (In realtà dovremmo passare l'email nel contesto, ma per il test usiamo il messaggio) analysis_request = f"""Ho ricevuto questa email importante da un investitore: From: {complex_email['sender']} Subject: {complex_email['subject']} {complex_email['content']} Puoi analizzarla e darmi raccomandazioni strategiche?""" response = await client.chat( session.session_id, analysis_request, max_tokens=500 ) print("📧 Email: Series A Investment Due Diligence...") print(f"🤖 IRIS Analysis: {response.content[:300]}...") # Follow-up per azioni specifiche followup = await client.chat( session.session_id, "Perfetto. Ora aiutami a prioritizzare le azioni e creare un piano di lavoro per rispettare la timeline.", max_tokens=300 ) print("📤 CEO: Aiutami a prioritizzare le azioni...") print(f"🤖 IRIS: {followup.content[:200]}...") # Verifica che l'analisi contenga elementi chiave full_response = response.content + followup.content assert "due diligence" in full_response.lower() assert "marzo" in full_response.lower() or "march" in full_response.lower() assert "€3M" in full_response or "3M" in full_response await client.end_conversation(session.session_id) await redis_client.disconnect() print("✅ SCENARIO 3 COMPLETED") return True except Exception as e: print(f"❌ SCENARIO 3 FAILED: {e}") return False async def scenario_4_document_analysis_simulation(): """ SCENARIO 4: Simulazione analisi documenti - Caricamento documenti multipli - Richiesta comparazione - Insights e raccomandazioni """ print("\n📄 SCENARIO 4: Document Analysis Simulation") print("-" * 50) try: # Documenti da analizzare documents = [ { "name": "Business_Plan_2024.pdf", "type": "PDF", "size": "2.5 MB", "date": "2024-01-15", "description": "Piano strategico 2024 con proiezioni finanziarie" }, { "name": "Q4_2023_Financial_Report.xlsx", "type": "Excel", "size": "1.8 MB", "date": "2024-01-10", "description": "Report finanziario Q4 2023 con KPI dettagliati" }, { "name": "Market_Analysis_2024.docx", "type": "Word", "size": "3.2 MB", "date": "2024-01-20", "description": "Analisi di mercato e competitive intelligence" } ] # Usa quick_chat per simulazione rapida analysis_request = f"""Ho caricato questi documenti per l'analisi: {chr(10).join([f"• {doc['name']} ({doc['type']}, {doc['size']}) - {doc['description']}" for doc in documents])} Puoi analizzarli e trovare correlazioni tra i dati finanziari del Q4, le proiezioni del business plan e le opportunità di mercato identificate?""" response = await quick_chat( user_id="laura_cfo", user_name="Laura Bianchi", message=analysis_request, prompt_type=PromptType.DOCUMENT_ANALYSIS, role="CFO", expertise_level="expert", communication_style="professional" ) print("📄 Documents: Business Plan, Financial Report, Market Analysis") print(f"🤖 IRIS: {response[:300]}...") # Verifica che la risposta menzioni i documenti assert "business plan" in response.lower() or "piano" in response.lower() assert "finanziario" in response.lower() or "financial" in response.lower() assert "mercato" in response.lower() or "market" in response.lower() print("✅ SCENARIO 4 COMPLETED") return True except Exception as e: print(f"❌ SCENARIO 4 FAILED: {e}") return False async def scenario_5_streaming_conversation(): """ SCENARIO 5: Conversazione con streaming - Test streaming response - Conversazione lunga e complessa """ print("\n🌊 SCENARIO 5: Streaming Conversation") print("-" * 50) try: redis_client = RedisClient() await redis_client.connect() client = IRISClient(default_provider="anthropic") session = await client.start_conversation( user_id="streaming_user", user_name="Test User", prompt_type=PromptType.BUSINESS_ASSISTANT, role="Manager", communication_style="professional" ) print("🌊 Starting streaming response...") # Test streaming full_response = "" chunk_count = 0 async for chunk in client.chat_stream( session.session_id, "Spiegami una strategia completa per il lancio di un nuovo prodotto AI, includendo analisi di mercato, go-to-market, pricing e metriche di successo." ): full_response += chunk chunk_count += 1 if chunk_count <= 5: # Mostra solo i primi chunk print(f"📦 Chunk {chunk_count}: {chunk[:50]}...") print(f"✅ Streaming completed: {chunk_count} chunks, {len(full_response)} chars") # Verifica che la risposta sia completa assert len(full_response) > 500 # Dovrebbe essere una risposta sostanziale assert "strategia" in full_response.lower() or "strategy" in full_response.lower() await client.end_conversation(session.session_id) await redis_client.disconnect() print("✅ SCENARIO 5 COMPLETED") return True except Exception as e: print(f"❌ SCENARIO 5 FAILED: {e}") return False async def scenario_6_system_health_and_recovery(): """ SCENARIO 6: Test salute sistema e recovery - Health check completo - Test resilienza """ print("\n🏥 SCENARIO 6: System Health & Recovery") print("-" * 50) try: client = get_iris_client() # Health check health = await client.health_check() print(f"🏥 System Health: {health['status']}") for component, status in health['components'].items(): status_emoji = "✅" if "healthy" in status else "⚠️" if "error" in status else "❓" print(f" {status_emoji} {component}: {status}") # Test multiple sessions per stesso utente sessions = [] for i in range(3): session = await client.start_conversation( user_id="test_user_multi", user_name="Multi Session User", prompt_type=PromptType.GENERAL_CHAT ) sessions.append(session) # Verifica sessioni attive user_sessions = await client.list_user_sessions("test_user_multi") assert len(user_sessions) >= 3 print(f"✅ Created {len(sessions)} sessions for same user") # Cleanup for session in sessions: await client.end_conversation(session.session_id) print("✅ SCENARIO 6 COMPLETED") return True except Exception as e: print(f"❌ SCENARIO 6 FAILED: {e}") return False async def main(): """Esegue tutti gli scenari E2E""" print("🚀 IRIS MILESTONE 2 - TEST E2E COMPLETO") print("=" * 70) print("Simulazione flussi utente reali end-to-end") print() scenarios = [ ("Business CEO Daily Brief", scenario_1_business_ceo_daily_brief), ("Multi-User Personalization", scenario_2_multi_user_personalization), ("Email Analysis Workflow", scenario_3_email_analysis_workflow), ("Document Analysis Simulation", scenario_4_document_analysis_simulation), ("Streaming Conversation", scenario_5_streaming_conversation), ("System Health & Recovery", scenario_6_system_health_and_recovery) ] passed = 0 total = len(scenarios) for scenario_name, scenario_func in scenarios: print(f"\n{'='*20} {scenario_name} {'='*20}") try: if await scenario_func(): passed += 1 print(f"✅ {scenario_name}: PASSED") else: print(f"❌ {scenario_name}: FAILED") except Exception as e: print(f"❌ {scenario_name}: ERROR - {e}") print(f"\n{'='*70}") print(f"🎯 E2E TEST RESULTS: {passed}/{total} scenarios passed") if passed == total: print("🎉 ALL E2E TESTS PASSED - MILESTONE 2 PRODUCTION READY!") print("\n🚀 READY FOR MILESTONE 3: TELEGRAM BOT FOUNDATION") return True else: print("❌ Some E2E tests failed - review required") return False if __name__ == "__main__": success = asyncio.run(main()) sys.exit(0 if success else 1)