#!/usr/bin/env python3 """ Скрипт для аналізу збережених логів MAI-DX """ import os import json import pandas as pd from datetime import datetime from collections import defaultdict import argparse def load_all_sessions(log_dir="mai_dx_logs"): """Завантажити всі сесії з директорії логів""" sessions = [] if not os.path.exists(log_dir): print(f"❌ Директорія {log_dir} не існує") return sessions for filename in os.listdir(log_dir): if filename.endswith('.json') and not filename.endswith('_analytics.json'): filepath = os.path.join(log_dir, filename) try: with open(filepath, 'r', encoding='utf-8') as f: session = json.load(f) sessions.append(session) except Exception as e: print(f"⚠️ Помилка читання {filename}: {e}") return sessions def analyze_session(session): """Детальний аналіз однієї сесії""" print(f"\n{'='*60}") print(f"📋 Аналіз сесії: {session['case_id']}") print(f"{'='*60}") print(f"\n📊 Основна інформація:") print(f" - Назва випадку: {session['case_name']}") print(f" - Час: {session['timestamp']}") print(f" - Режим: {session['mode']}") print(f" - Статус: {session['status']}") print(f"\n💊 Діагноз:") print(f" - Фінальний діагноз: {session['diagnosis']}") print(f" - Впевненість: {session['confidence']}/5.0") print(f" - Обґрунтування: {session['reasoning'][:100]}...") print(f"\n💰 Економіка:") print(f" - Витрачено: ${session['cost']}") print(f" - Бюджет: ${session['budget']}") print(f" - Ефективність: {((session['budget'] - session['cost']) / session['budget'] * 100):.1f}%") print(f"\n⏱️ Продуктивність:") print(f" - Тривалість: {session['duration']:.1f} секунд") print(f" - Ітерацій: {session['iterations']}") # Аналіз розмов conversations = session.get('conversations', []) if conversations: print(f"\n💬 Аналіз розмов:") print(f" - Всього раундів: {len(conversations)}") # Підрахунок активності агентів agent_activity = defaultdict(int) message_types = defaultdict(int) total_messages = 0 for conv in conversations: for msg in conv.get('messages', []): agent_activity[msg['agent_name']] += 1 message_types[msg['message_type']] += 1 total_messages += 1 print(f" - Всього повідомлень: {total_messages}") print(f"\n 🤖 Активність агентів:") for agent, count in sorted(agent_activity.items(), key=lambda x: x[1], reverse=True): print(f" - {agent}: {count} повідомлень ({count/total_messages*100:.1f}%)") print(f"\n 📝 Типи повідомлень:") for msg_type, count in sorted(message_types.items(), key=lambda x: x[1], reverse=True): print(f" - {msg_type}: {count} ({count/total_messages*100:.1f}%)") # Деталі раундів print(f"\n 🔄 Деталі раундів:") for i, conv in enumerate(conversations): msgs = len(conv.get('messages', [])) cost = conv.get('cost_incurred', 0) decision = conv.get('decision', 'N/A') print(f" Раунд {i+1}: {msgs} повідомлень, ${cost:.2f}, Рішення: {decision}") def generate_summary_report(sessions, output_file="mai_dx_summary_report.html"): """Генерація зведеного HTML звіту по всіх сесіях""" if not sessions: print("❌ Немає сесій для аналізу") return html_content = """ MAI-DX Summary Report

🏥 MAI-DX Summary Report

""" # Загальна статистика total_sessions = len(sessions) avg_accuracy = sum(s['confidence'] for s in sessions) / total_sessions avg_cost = sum(s['cost'] for s in sessions) / total_sessions avg_duration = sum(s['duration'] for s in sessions) / total_sessions success_rate = sum(1 for s in sessions if s['confidence'] >= 3.0) / total_sessions * 100 html_content += f"""

📊 Загальна статистика

{total_sessions}
Всього сесій
{avg_accuracy:.2f}/5.0
Середня точність
${avg_cost:.2f}
Середня вартість
{avg_duration:.1f}с
Середній час
{success_rate:.1f}%
Успішність

📋 Детальна таблиця сесій

""" # Сортуємо сесії за часом sorted_sessions = sorted(sessions, key=lambda x: x['timestamp'], reverse=True) for session in sorted_sessions: status_class = 'success' if session['confidence'] >= 3.0 else 'warning' html_content += f""" """ html_content += """
ID сесії Час Випадок Діагноз Точність Вартість Час Статус
{session['case_id']} {session['timestamp'][:19]} {session['case_name']} {session['diagnosis']} {session['confidence']:.1f}/5.0 ${session['cost']:.2f} {session['duration']:.1f}с {session['status']}

Згенеровано: """ + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + """

""" with open(output_file, 'w', encoding='utf-8') as f: f.write(html_content) print(f"\n✅ Зведений звіт збережено: {output_file}") def export_to_csv(sessions, output_file="mai_dx_sessions.csv"): """Експорт сесій в CSV""" if not sessions: print("❌ Немає сесій для експорту") return # Створюємо DataFrame data = [] for session in sessions: # Підраховуємо агентів та повідомлення total_messages = 0 unique_agents = set() for conv in session.get('conversations', []): for msg in conv.get('messages', []): total_messages += 1 unique_agents.add(msg['agent_name']) data.append({ 'case_id': session['case_id'], 'timestamp': session['timestamp'], 'case_name': session['case_name'], 'mode': session['mode'], 'diagnosis': session['diagnosis'], 'confidence': session['confidence'], 'cost': session['cost'], 'budget': session['budget'], 'duration': session['duration'], 'iterations': session['iterations'], 'status': session['status'], 'total_messages': total_messages, 'unique_agents': len(unique_agents) }) df = pd.DataFrame(data) df.to_csv(output_file, index=False, encoding='utf-8') print(f"✅ Дані експортовано в CSV: {output_file}") print(f" Всього записів: {len(df)}") def main(): parser = argparse.ArgumentParser(description='Аналіз логів MAI-DX') parser.add_argument('--dir', default='mai_dx_logs', help='Директорія з логами') parser.add_argument('--case', help='ID конкретної сесії для детального аналізу') parser.add_argument('--export-csv', action='store_true', help='Експортувати в CSV') parser.add_argument('--summary', action='store_true', help='Створити зведений HTML звіт') args = parser.parse_args() print("🔬 Аналіз логів MAI-DX") print("=" * 60) # Завантажуємо сесії sessions = load_all_sessions(args.dir) print(f"📁 Знайдено {len(sessions)} сесій в {args.dir}") if args.case: # Аналіз конкретної сесії session = next((s for s in sessions if s['case_id'] == args.case), None) if session: analyze_session(session) else: print(f"❌ Сесія {args.case} не знайдена") elif args.export_csv: # Експорт в CSV export_to_csv(sessions) elif args.summary: # Створення зведеного звіту generate_summary_report(sessions) else: # Загальна статистика if sessions: print(f"\n📊 Загальна статистика:") print(f" - Середня точність: {sum(s['confidence'] for s in sessions) / len(sessions):.2f}/5.0") print(f" - Середня вартість: ${sum(s['cost'] for s in sessions) / len(sessions):.2f}") print(f" - Середній час: {sum(s['duration'] for s in sessions) / len(sessions):.1f} секунд") print(f"\n💡 Для детального аналізу використовуйте:") print(f" python {__file__} --case ") print(f" python {__file__} --export-csv") print(f" python {__file__} --summary") if __name__ == "__main__": main()