agent_conversation_logger.py

#!/usr/bin/env python3
"""
Модуль для логування та управління діагностичними сесіями MAI-DX.

Цей модуль забезпечує:
- Створення та відстеження окремих діагностичних сесій.
- Запис детальних повідомлень від кожного ШІ-агента.
- Збереження повної інформації про сесію у форматі JSON.
- Надання функцій для аналітики: перелік сесій, експорт у CSV та генерація звітів.
"""
import os
import json
import csv
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict, field
import threading
import uuid

@dataclass
class DiagnosisSession:
    """
    Датаклас для зберігання повної інформації про одну діагностичну сесію.
    """
    case_id: str
    timestamp: str
    case_name: str
    patient_info: str
    mode: str
    budget: int
    diagnosis: str = "N/A"
    confidence: float = 0.0
    cost: float = 0.0
    iterations: int = 0
    duration: float = 0.0
    status: str = "In Progress"
    reasoning: str = "N/A"
    messages: List[Dict[str, Any]] = field(default_factory=list)

class AgentConversationLogger:
    """
    Клас для управління логуванням діагностичних сесій.
    Зберігає кожну сесію в окремому JSON-файлі для легкого доступу та аналізу.
    """

    def __init__(self, log_dir: str = "mai_dx_logs"):
        """
        Ініціалізує логгер, створюючи директорію для логів, якщо її не існує.

        Args:
            log_dir (str): Директорія для збереження файлів логів.
        """
        self.log_dir = log_dir
        os.makedirs(self.log_dir, exist_ok=True)
        # Використовуємо локальне сховище потоку для ізоляції сесій
        self.thread_local = threading.local()
        print(f"✅ Справжній AgentConversationLogger ініціалізовано. Логи зберігаються в '{self.log_dir}'.")

    def start_conversation(self, case_name: str, patient_info: str, mode: str) -> str:
        """
        Розпочинає нову діагностичну сесію.

        Args:
            case_name (str): Назва клінічного випадку.
            patient_info (str): Вхідна інформація про пацієнта.
            mode (str): Режим діагностики.

        Returns:
            str: Унікальний ідентифікатор створеної сесії (case_id).
        """
        case_id = f"case_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
        session = DiagnosisSession(
            case_id=case_id,
            timestamp=datetime.now().isoformat(),
            case_name=case_name,
            patient_info=patient_info,
            mode=mode,
            budget=0 # Буде встановлено пізніше, якщо потрібно
        )
        self.thread_local.session = session
        return case_id

    def log_agent_message(self, agent_name: str, action: str, content: str, reasoning: str, confidence: float, cost: float, iteration: int):
        """
        Логує повідомлення від одного з ШІ-агентів.
        """
        if not hasattr(self.thread_local, 'session'):
            print("⚠️ Попередження: Спроба логування без активної сесії.")
            return

        message = {
            "timestamp": datetime.now().isoformat(),
            "agent": agent_name,
            "action": action,
            "content": content,
            "reasoning": reasoning,
            "confidence": confidence,
            "cost": cost,
            "iteration": iteration,
        }
        self.thread_local.session.messages.append(message)

    def end_conversation(self, final_diagnosis: str, confidence: float, cost: float, **kwargs) -> str:
        """
        Завершує сесію, оновлює фінальні дані та зберігає її у файл.

        Returns:
            str: Ідентифікатор збереженої сесії.
        """
        if not hasattr(self.thread_local, 'session'):
            print("⚠️ Попередження: Спроба завершити неіснуючу сесію.")
            return "no_active_session"

        session = self.thread_local.session
        
        # Оновлення фінальних даних сесії
        session.diagnosis = final_diagnosis
        session.confidence = confidence
        session.cost = cost
        session.status = "✅ Успішно" if confidence >= 3.0 else "⚠️ Потребує перегляду"
        session.reasoning = kwargs.get('reasoning', 'N/A')
        session.iterations = kwargs.get('iterations', len(session.messages))
        
        start_time = datetime.fromisoformat(session.timestamp)
        session.duration = (datetime.now() - start_time).total_seconds()

        # Збереження сесії у файл
        file_path = os.path.join(self.log_dir, f"{session.case_id}.json")
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(asdict(session), f, ensure_ascii=False, indent=4)

        del self.thread_local.session # Очищення сесії для поточного потоку
        return session.case_id

    def list_conversations(self) -> List[Dict[str, Any]]:
        """
        Зчитує всі збережені сесії з директорії логів.

        Returns:
            List[Dict[str, Any]]: Список словників, де кожен словник - це одна сесія.
        """
        all_sessions = []
        for filename in sorted(os.listdir(self.log_dir), reverse=True):
            if filename.endswith(".json"):
                try:
                    with open(os.path.join(self.log_dir, filename), 'r', encoding='utf-8') as f:
                        data = json.load(f)
                        # Видаляємо детальні повідомлення для короткого списку
                        data.pop('messages', None) 
                        all_sessions.append(data)
                except (json.JSONDecodeError, IOError) as e:
                    print(f"Помилка читання файлу логу {filename}: {e}")
        return all_sessions

    def export_analytics_csv(self) -> str:
        """
        Експортує зведену аналітику по всіх сесіях у CSV файл.

        Returns:
            str: Назва створеного CSV файлу.
        """
        sessions = self.list_conversations()
        if not sessions:
            return "Немає даних для експорту."

        filename = f"analytics_report_{datetime.now().strftime('%Y%m%d_%H%M')}.csv"
        file_path = os.path.join(self.log_dir, filename)

        # Визначаємо поля (колонки) на основі ключів першої сесії
        fieldnames = sessions[0].keys()

        with open(file_path, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(sessions)

        return file_path

    def export_conversation_report(self, case_id: str, format: str = 'html') -> str:
        """
        Створює детальний звіт для конкретного випадку.

        Args:
            case_id (str): Ідентифікатор сесії для звіту.
            format (str): Формат звіту ('html' або 'json').

        Returns:
            str: Назва створеного файлу звіту.
        """
        file_path = os.path.join(self.log_dir, f"{case_id}.json")
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
        except FileNotFoundError:
            return f"Помилка: Файл для випадку {case_id} не знайдено."

        if format == 'json':
            return file_path

        # Генерація HTML-звіту
        report_filename = f"report_{case_id}.html"
        report_path = os.path.join(self.log_dir, report_filename)

        html = f"""
        <html>
        <head>
            <title>Звіт по випадку: {data['case_name']}</title>
            <style>
                body {{ font-family: sans-serif; line-height: 1.6; padding: 20px; }}
                h1, h2 {{ color: #333; }}
                table {{ width: 100%; border-collapse: collapse; margin-top: 20px; }}
                th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
                th {{ background-color: #f2f2f2; }}
                .summary {{ background-color: #eef; padding: 15px; border-radius: 8px; }}
            </style>
        </head>
        <body>
            <h1>Клінічний звіт по випадку: {data['case_name']}</h1>
            <p><strong>ID:</strong> {data['case_id']}</p>
            <p><strong>Час:</strong> {data['timestamp']}</p>
            
            <div class="summary">
                <h2>Підсумки діагностики</h2>
                <p><strong>Фінальний діагноз:</strong> {data['diagnosis']}</p>
                <p><strong>Впевненість:</strong> {data['confidence']}/5.0</p>
                <p><strong>Вартість:</strong> ${data['cost']}</p>
                <p><strong>Тривалість:</strong> {data['duration']:.2f} сек.</p>
            </div>

            <h2>Детальний лог взаємодії агентів</h2>
            <table>
                <tr>
                    <th>Час</th><th>Агент</th><th>Дія</th><th>Результат/Контент</th>
                </tr>
        """
        for msg in data.get('messages', []):
            html += f"""
                <tr>
                    <td>{datetime.fromisoformat(msg['timestamp']).strftime('%H:%M:%S')}</td>
                    <td>{msg['agent']}</td>
                    <td>{msg['action']}</td>
                    <td>{msg['content']}</td>
                </tr>
            """
        html += """
            </table>
        </body>
        </html>
        """

        with open(report_path, 'w', encoding='utf-8') as f:
            f.write(html)

        return report_path