Spaces:
Sleeping
Sleeping
File size: 9,301 Bytes
953a0b1 0c6718b 953a0b1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 |
#!/usr/bin/env python3
"""
Conversation Logger for Medical Assistant.
Logs conversations with spiritual classification indicators for analysis.
"""
import json
import os
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from src.core.spiritual_state import SpiritualState, SpiritualAssessment
@dataclass
class ConversationEntry:
"""Single conversation entry with classification data."""
timestamp: str
user_message: str
assistant_response: str
spiritual_classification: str # GREEN, YELLOW, RED
classification_confidence: float
classification_indicators: List[str]
classification_reasoning: str
session_id: str
message_index: int
@dataclass
class ConversationSession:
"""Complete conversation session."""
session_id: str
start_time: str
end_time: Optional[str]
patient_name: str
total_messages: int
entries: List[ConversationEntry]
session_summary: Dict[str, Any]
class ConversationLogger:
"""Logger for conversation sessions with spiritual classification data."""
def __init__(self, session_id: str = None, patient_name: str = "Anonymous"):
"""Initialize conversation logger."""
self.session_id = session_id or self._generate_session_id()
self.patient_name = patient_name
self.start_time = datetime.now().isoformat()
self.entries: List[ConversationEntry] = []
self.message_counter = 0
# Create logs directory if it doesn't exist
self.logs_dir = "conversation_logs"
os.makedirs(self.logs_dir, exist_ok=True)
def _generate_session_id(self) -> str:
"""Generate unique session ID."""
return f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
def log_exchange(
self,
user_message: str,
assistant_response: str,
assessment: SpiritualAssessment
) -> None:
"""
Log a conversation exchange with spiritual classification.
Args:
user_message: User's message
assistant_response: Assistant's response
assessment: Spiritual assessment of the user message
"""
self.message_counter += 1
entry = ConversationEntry(
timestamp=datetime.now().isoformat(),
user_message=user_message,
assistant_response=assistant_response,
spiritual_classification=assessment.state.value.upper(),
classification_confidence=assessment.confidence,
classification_indicators=assessment.indicators,
classification_reasoning=assessment.reasoning,
session_id=self.session_id,
message_index=self.message_counter
)
self.entries.append(entry)
# Auto-save after each entry
self._save_session()
def get_classification_indicator(self, state: SpiritualState) -> str:
"""Get colored emoji indicator for spiritual state."""
indicators = {
SpiritualState.GREEN: "🟢",
SpiritualState.YELLOW: "🟡",
SpiritualState.RED: "🔴"
}
return indicators.get(state, "⚪")
def get_classification_text(self, assessment: SpiritualAssessment) -> str:
"""Get formatted classification text for display."""
indicator = self.get_classification_indicator(assessment.state)
confidence_percent = int(assessment.confidence * 100)
classification_text = f"{indicator} **{assessment.state.value.upper()}** ({confidence_percent}%)"
if assessment.indicators:
indicators_text = ", ".join(assessment.indicators[:3]) # Show max 3 indicators
if len(assessment.indicators) > 3:
indicators_text += f" +{len(assessment.indicators) - 3} more"
classification_text += f"\n*Indicators: {indicators_text}*"
# Add reasoning in italics
if assessment.reasoning:
classification_text += f"\n*{assessment.reasoning}*"
return classification_text
def _save_session(self) -> None:
"""Save current session to JSON file."""
session = ConversationSession(
session_id=self.session_id,
start_time=self.start_time,
end_time=None, # Will be set when session ends
patient_name=self.patient_name,
total_messages=self.message_counter,
entries=self.entries,
session_summary=self._generate_session_summary()
)
filename = f"{self.session_id}.json"
filepath = os.path.join(self.logs_dir, filename)
try:
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(asdict(session), f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"Error saving conversation log: {e}")
def _generate_session_summary(self) -> Dict[str, Any]:
"""Generate summary statistics for the session."""
if not self.entries:
return {}
# Count classifications
green_count = sum(1 for e in self.entries if e.spiritual_classification == "GREEN")
yellow_count = sum(1 for e in self.entries if e.spiritual_classification == "YELLOW")
red_count = sum(1 for e in self.entries if e.spiritual_classification == "RED")
# Calculate average confidence
avg_confidence = sum(e.classification_confidence for e in self.entries) / len(self.entries)
# Collect all indicators
all_indicators = []
for entry in self.entries:
all_indicators.extend(entry.classification_indicators)
# Count unique indicators
indicator_counts = {}
for indicator in all_indicators:
indicator_counts[indicator] = indicator_counts.get(indicator, 0) + 1
return {
"total_exchanges": len(self.entries),
"classification_counts": {
"green": green_count,
"yellow": yellow_count,
"red": red_count
},
"classification_percentages": {
"green": round(green_count / len(self.entries) * 100, 1),
"yellow": round(yellow_count / len(self.entries) * 100, 1),
"red": round(red_count / len(self.entries) * 100, 1)
},
"average_confidence": round(avg_confidence, 3),
"top_indicators": dict(sorted(indicator_counts.items(), key=lambda x: x[1], reverse=True)[:5]),
"session_duration_minutes": self._calculate_session_duration()
}
def _calculate_session_duration(self) -> float:
"""Calculate session duration in minutes."""
if not self.entries:
return 0.0
start = datetime.fromisoformat(self.start_time)
last_entry = datetime.fromisoformat(self.entries[-1].timestamp)
duration = (last_entry - start).total_seconds() / 60
return round(duration, 1)
def end_session(self) -> str:
"""End the session and return final log file path."""
# Update end time in the last save
if self.entries:
self.entries[-1].timestamp = datetime.now().isoformat()
self._save_session()
filename = f"{self.session_id}.json"
return os.path.join(self.logs_dir, filename)
def get_session_summary(self) -> Dict[str, Any]:
"""Get current session summary."""
return self._generate_session_summary()
def export_csv(self) -> str:
"""Export conversation to CSV format."""
import csv
csv_filename = f"{self.session_id}.csv"
csv_filepath = os.path.join(self.logs_dir, csv_filename)
try:
with open(csv_filepath, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = [
'timestamp', 'message_index', 'user_message', 'assistant_response',
'spiritual_classification', 'classification_confidence',
'classification_indicators', 'classification_reasoning'
]
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for entry in self.entries:
writer.writerow({
'timestamp': entry.timestamp,
'message_index': entry.message_index,
'user_message': entry.user_message,
'assistant_response': entry.assistant_response,
'spiritual_classification': entry.spiritual_classification,
'classification_confidence': entry.classification_confidence,
'classification_indicators': '; '.join(entry.classification_indicators),
'classification_reasoning': entry.classification_reasoning
})
return csv_filepath
except Exception as e:
print(f"Error exporting to CSV: {e}")
return "" |