production_mai_dx.py

#!/usr/bin/env python3
"""
Production-ready MAI-DX Orchestrator
Оптимізовано для реального використання
"""
import os
import sys
import warnings
from dotenv import load_dotenv

# Завантажуємо змінні середовища
load_dotenv()

# Виробничі налаштування для стабільності
os.environ["SWARMS_VERBOSITY"] = "SILENT"
os.environ["RICH_TRACEBACK"] = "0" 
os.environ["SWARMS_SHOW_PANEL"] = "false"
os.environ["SWARMS_AUTO_PRINT"] = "false"
os.environ["LOGURU_LEVEL"] = "WARNING"  # Менше спаму в логах
warnings.filterwarnings("ignore")

# Rich formatter патч (довелося ефективність!)
def patch_rich_formatter():
    try:
        import swarms.utils.formatter
        def dummy_print_panel(*args, **kwargs):
            pass
        if hasattr(swarms.utils.formatter, 'Formatter'):
            swarms.utils.formatter.Formatter._print_panel = dummy_print_panel
        return True
    except:
        return False

patch_rich_formatter()

from mai_dx import MaiDxOrchestrator
from loguru import logger

class ProductionMAIDX:
    """Production-wrapper для MAI-DX з поліпшеннями"""
    
    def __init__(self, model="gemini/gemini-2.5-flash"):
        self.model = model
        self.available_modes = ["question_only", "budgeted", "no_budget"]
        
    def create_optimized_orchestrator(self, mode="question_only", budget=2000, max_iterations=5):
        """Створює оптимізований оркестратор для production"""
        
        # Рекомендовані налаштування для кожного режиму
        configs = {
            "question_only": {"budget": 1000, "iterations": 3},
            "budgeted": {"budget": 3000, "iterations": 5}, 
            "no_budget": {"budget": 10000, "iterations": 7}
        }
        
        config = configs.get(mode, {"budget": budget, "iterations": max_iterations})
        
        try:
            orchestrator = MaiDxOrchestrator(
                model_name=self.model,
                max_iterations=config["iterations"],
                initial_budget=config["budget"],
                mode=mode
            )
            
            print(f"✅ MAI-DX готовий: {mode} режим, бюджет ${config['budget']}")
            return orchestrator
            
        except Exception as e:
            print(f"❌ Помилка створення: {e}")
            return None
    
    def diagnose_case(self, case_description, expected_diagnosis=None, mode="question_only"):
        """Основний метод для діагностики"""
        
        print(f"\n🩺 Запуск MAI-DX діагностики...")
        print(f"📋 Режим: {mode}")
        
        # Створюємо оркестратор
        orchestrator = self.create_optimized_orchestrator(mode)
        if not orchestrator:
            return None
            
        try:
            # Якщо немає expected_diagnosis, використовуємо загальний
            ground_truth = expected_diagnosis or "Unknown diagnosis"
            
            print("🔄 Аналіз випадку... (це може зайняти 2-5 хвилин)")
            
            # Запускаємо діагностику
            result = orchestrator.run(
                initial_case_info=case_description,
                full_case_details=case_description,
                ground_truth_diagnosis=ground_truth
            )
            
            return self._format_result(result, mode)
            
        except Exception as e:
            print(f"❌ Помилка діагностики: {e}")
            return None
    
    def _format_result(self, result, mode):
        """Форматуємо результат для зручного відображення"""
        
        # Оцінюємо якість результату
        if result.accuracy_score >= 4.0:
            quality = "🎉 ВІДМІННО"
        elif result.accuracy_score >= 3.0:
            quality = "👍 ДОБРЕ"  
        elif result.accuracy_score >= 2.0:
            quality = "⚠️ ЗАДОВІЛЬНО"
        elif "not reached" in result.final_diagnosis.lower():
            quality = "🔄 ПОТРІБНО БІЛЬШЕ ЧАСУ"
        else:
            quality = "🔍 ПОТРЕБУЄ ПЕРЕВІРКИ"
            
        formatted = {
            "diagnosis": result.final_diagnosis,
            "confidence": result.accuracy_score,
            "quality": quality,
            "cost": result.total_cost,
            "iterations": result.iterations,
            "mode": mode,
            "reasoning": getattr(result, 'accuracy_reasoning', 'Недоступно')[:200] + "..."
        }
        
        return formatted
    
    def batch_diagnose(self, cases, mode="question_only"):
        """Діагностика кількох випадків"""
        results = []
        
        for i, case in enumerate(cases, 1):
            print(f"\n📋 Випадок {i}/{len(cases)}")
            
            case_text = case.get("description", "")
            expected = case.get("expected", None)
            
            result = self.diagnose_case(case_text, expected, mode)
            if result:
                results.append({"case_id": i, **result})
        
        return results

def demo_production_usage():
    """Демонстрація production використання"""
    
    print("=" * 70)
    print("🏥 MAI-DX PRODUCTION DEMO")
    print("=" * 70)
    
    # Створюємо production instance
    mai_dx = ProductionMAIDX()
    
    # Тестові випадки
    test_cases = [
        {
            "description": "35-year-old man with sudden severe chest pain radiating to left arm, sweating, nausea. Pain started 2 hours ago during exercise.",
            "expected": "Myocardial infarction"
        },
        {
            "description": "22-year-old woman with severe headache, neck stiffness, fever 39°C, photophobia. Symptoms started this morning.",
            "expected": "Meningitis"
        },
        {
            "description": "45-year-old smoker with persistent cough for 3 months, weight loss 10kg, bloody sputum occasionally.",
            "expected": "Lung cancer"
        }
    ]
    
    print(f"📋 Тестування {len(test_cases)} випадків...")
    
    # Запускаємо діагностику
    results = mai_dx.batch_diagnose(test_cases, mode="question_only")
    
    # Виводимо результати
    print(f"\n{'='*70}")
    print("📊 РЕЗУЛЬТАТИ ТЕСТУВАННЯ")
    print("="*70)
    
    for result in results:
        print(f"\n🔍 Випадок {result['case_id']}:")
        print(f"   Діагноз: {result['diagnosis']}")
        print(f"   Якість: {result['quality']}")
        print(f"   Оцінка: {result['confidence']}/5.0")
        print(f"   Вартість: ${result['cost']}")
        print(f"   Ітерації: {result['iterations']}")
    
    # Статистика
    successful = sum(1 for r in results if r['confidence'] >= 2.0)
    print(f"\n📈 Статистика: {successful}/{len(results)} успішних діагнозів")
    
    if successful >= len(results) * 0.7:
        print("🎉 MAI-DX готовий для production!")
    else:
        print("⚠️ Потрібне додаткове налаштування")

def simple_diagnostic_session():
    """Простий інтерактивний сеанс діагностики"""
    
    print("\n🩺 Простий сеанс діагностики")
    print("-" * 40)
    
    # Простий випадок для демонстрації
    case = """
    29-year-old woman with sore throat and peritonsillar swelling and bleeding.
    Symptoms did not abate with antimicrobial therapy.
    
    History: Onset 7 weeks ago, worsening right-sided pain.
    Physical Exam: Right peritonsillar mass, displaces uvula.
    Biopsy: Round-cell neoplasm, positive for desmin and MyoD1.
    """
    
    expected = "Embryonal rhabdomyosarcoma of the pharynx"
    
    mai_dx = ProductionMAIDX()
    result = mai_dx.diagnose_case(case, expected, "question_only")
    
    if result:
        print(f"\n🎯 Діагноз: {result['diagnosis']}")
        print(f"⭐ Якість: {result['quality']}")
        print(f"💰 Вартість: ${result['cost']}")
        
        # Аналіз близькості діагнозу
        if "rhabdomyosarcoma" in result['diagnosis'].lower():
            print("✅ Правильний тип пухлини ідентифіковано!")
        
        return True
    else:
        print("❌ Діагностика не вдалася")
        return False

if __name__ == "__main__":
    try:
        # Запускаємо простий тест
        simple_diagnostic_session()
        
        # Опціонально - повний demo
        print("\n" + "="*50)
        print("Запустити повний production demo? (y/n): ", end="")
        
        # Для автоматичного запуску в тестах - пропускаємо input
        try:
            choice = input().lower()
            if choice == 'y':
                demo_production_usage()
        except:
            print("Пропускаємо інтерактивний demo")
            
    except KeyboardInterrupt:
        print("\n⏹️ Зупинено користувачем")
    except Exception as e:
        print(f"\n💥 Помилка: {e}")