Spaces:
Running
Running
File size: 11,642 Bytes
be1b5d2 |
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 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 |
#!/usr/bin/env python3
"""
Enhanced Verification System Demo.
Demonstrates the enhanced verification system with new display formats,
provider summaries, and CSV export capabilities.
Requirements: 8.2, 8.3, 8.4, 8.5
"""
import tempfile
import os
from datetime import datetime
from src.core.conversation_logger import ConversationLogger
from src.core.conversation_verification import (
EnhancedConversationVerificationManager,
VerificationFeedback
)
from src.core.verification_exporter import EnhancedVerificationExporter
from src.core.spiritual_state import SpiritualState, SpiritualAssessment
from src.core.provider_summary_generator import ProviderSummary
def create_sample_conversation():
"""Create a sample conversation with various classification levels."""
logger = ConversationLogger(patient_name="Demo Patient")
# Sample conversation exchanges with different classifications
exchanges = [
{
"user": "I've been struggling with the meaning of my illness lately",
"assistant": "I understand this is a very difficult time for you. Many people in your situation experience these feelings.",
"assessment": SpiritualAssessment(
state=SpiritualState.RED,
confidence=0.85,
indicators=["loss of meaning", "spiritual distress", "existential questioning"],
reasoning="Patient expressing significant spiritual concerns about meaning and purpose in context of illness"
)
},
{
"user": "Sometimes I wonder if there's any point to all this suffering",
"assistant": "Your feelings are completely valid. It's natural to question these things when facing challenges.",
"assessment": SpiritualAssessment(
state=SpiritualState.RED,
confidence=0.90,
indicators=["questioning purpose", "suffering distress", "existential crisis"],
reasoning="Patient expressing deep existential concerns about suffering and purpose"
)
},
{
"user": "I'm feeling a bit better today, but still uncertain",
"assistant": "I'm glad to hear you're feeling better. Uncertainty is normal during recovery.",
"assessment": SpiritualAssessment(
state=SpiritualState.YELLOW,
confidence=0.65,
indicators=["uncertainty", "mild improvement"],
reasoning="Patient showing some improvement but still experiencing uncertainty"
)
},
{
"user": "Thank you for listening. I feel more hopeful now",
"assistant": "You're very welcome. I'm here whenever you need support.",
"assessment": SpiritualAssessment(
state=SpiritualState.GREEN,
confidence=0.80,
indicators=["gratitude", "hope", "positive outlook"],
reasoning="Patient expressing gratitude and hope, indicating improved spiritual state"
)
}
]
# Log all exchanges
for exchange in exchanges:
logger.log_exchange(
exchange["user"],
exchange["assistant"],
exchange["assessment"]
)
return logger
def demonstrate_enhanced_verification():
"""Demonstrate the enhanced verification system."""
print("=" * 80)
print("ENHANCED VERIFICATION SYSTEM DEMONSTRATION")
print("=" * 80)
print()
# Create temporary directory for demo
temp_dir = tempfile.mkdtemp()
print(f"📁 Using temporary directory: {temp_dir}")
print()
try:
# Step 1: Create sample conversation
print("🗣️ STEP 1: Creating sample conversation...")
logger = create_sample_conversation()
print(f" ✅ Created conversation with {len(logger.entries)} exchanges")
print()
# Step 2: Create enhanced verification session
print("🔍 STEP 2: Creating enhanced verification session...")
verification_manager = EnhancedConversationVerificationManager(temp_dir)
session = verification_manager.create_verification_session(
logger,
verifier_name="Demo Verifier",
enable_enhanced_formats=True
)
print(f" ✅ Created enhanced verification session: {session.session_id}")
print(f" 📊 Enhanced format enabled: {session.enhanced_format_enabled}")
print(f" 📝 Total records: {len(session.verification_records)}")
print()
# Step 3: Show enhanced display formats
print("🎨 STEP 3: Demonstrating enhanced display formats...")
for i, record in enumerate(session.verification_records[:2], 1): # Show first 2
print(f" 📋 Record {i} Enhanced Display:")
if record.enhanced_display_format:
# Show first 200 chars of HTML
preview = record.enhanced_display_format[:200].replace('\n', ' ')
print(f" HTML Preview: {preview}...")
if record.visual_sections:
print(f" Visual Sections: {len(record.visual_sections)} sections")
for section in record.visual_sections:
print(f" - {section.get('type', 'unknown')}")
print()
# Step 4: Add provider summary to RED flag cases
print("📋 STEP 4: Adding provider summaries to RED flag cases...")
red_records = [r for r in session.verification_records if r.original_classification == "RED"]
for i, record in enumerate(red_records):
provider_summary = ProviderSummary(
patient_name="Demo Patient",
patient_phone="555-DEMO",
classification="RED",
confidence=record.original_confidence,
indicators=record.original_indicators,
reasoning=record.original_reasoning,
urgency_level="IMMEDIATE" if record.original_confidence > 0.8 else "URGENT",
severity_level="HIGH",
situation_description=f"Patient expressing {', '.join(record.original_indicators[:2])} requiring immediate spiritual care attention"
)
record.set_enhanced_formats(provider_summary=provider_summary)
print(f" ✅ Added provider summary to RED flag record {i+1}")
print(f" 📊 Provider summaries added to {len(red_records)} RED flag cases")
print()
# Step 5: Add verification feedback
print("✅ STEP 5: Adding verification feedback...")
feedbacks = [
{"correct": True, "notes": "Correctly identified spiritual distress"},
{"correct": True, "notes": "Appropriate RED classification"},
{"correct": False, "classification": "GREEN", "reason": "Patient showing improvement, should be GREEN"},
{"correct": True, "notes": "Correct GREEN classification"}
]
for i, (record, feedback_data) in enumerate(zip(session.verification_records, feedbacks)):
feedback = VerificationFeedback(
exchange_id=record.exchange_id,
is_correct=feedback_data["correct"],
correct_classification=feedback_data.get("classification"),
correction_reason=feedback_data.get("reason"),
notes=feedback_data.get("notes")
)
success = verification_manager.submit_exchange_verification(
session.session_id,
record.exchange_id,
feedback
)
if success:
print(f" ✅ Added feedback for exchange {i+1}")
else:
print(f" ❌ Failed to add feedback for exchange {i+1}")
print()
# Step 6: Export enhanced CSV
print("📊 STEP 6: Exporting enhanced CSV with new format data...")
exporter = EnhancedVerificationExporter(temp_dir)
csv_path = exporter.export_session_to_csv(session, include_enhanced_data=True)
print(f" ✅ Exported enhanced CSV: {os.path.basename(csv_path)}")
# Show CSV preview
with open(csv_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
print(f" 📄 CSV contains {len(lines)} lines")
print(" 📋 Enhanced columns included:")
# Find header line
for line in lines:
if 'has_enhanced_display' in line:
enhanced_columns = [col for col in line.split(',') if 'enhanced' in col.lower() or 'provider' in col.lower()]
for col in enhanced_columns[:5]: # Show first 5 enhanced columns
print(f" - {col.strip()}")
break
print()
# Step 7: Generate enhanced summary report
print("📈 STEP 7: Generating enhanced summary report...")
report_path = exporter.export_enhanced_summary_report(session)
print(f" ✅ Generated summary report: {os.path.basename(report_path)}")
# Show report preview
with open(report_path, 'r', encoding='utf-8') as f:
content = f.read()
# Extract key statistics
lines = content.split('\n')
for line in lines:
if 'Enhanced Display Coverage:' in line or 'Records with Enhanced Display:' in line:
print(f" 📊 {line.strip()}")
print()
# Step 8: Show session statistics
print("📈 STEP 8: Enhanced session statistics...")
stats = verification_manager.get_session_statistics(session.session_id)
if stats:
print(f" 📊 Total Exchanges: {stats.get('total_exchanges', 0)}")
print(f" ✅ Verified Exchanges: {stats.get('verified_exchanges', 0)}")
print(f" 🎯 Overall Accuracy: {stats.get('progress', {}).get('accuracy_overall', 0):.1%}")
print(f" 🎨 Enhanced Format Enabled: {stats.get('enhanced_format_enabled', False)}")
print(f" 📋 Records with Enhanced Display: {stats.get('records_with_enhanced_display', 0)}")
print(f" 📄 Records with Provider Summary: {stats.get('records_with_provider_summary', 0)}")
print()
print("🎉 DEMONSTRATION COMPLETE!")
print("=" * 80)
print()
print("📁 Generated Files:")
print(f" 📊 Enhanced CSV: {csv_path}")
print(f" 📈 Summary Report: {report_path}")
print(f" 💾 Session Data: {temp_dir}")
print()
print("✨ Enhanced verification system successfully demonstrated!")
print(" - Enhanced display formats with visual sections")
print(" - Provider summaries for RED flag cases")
print(" - Enhanced CSV export with new data fields")
print(" - Comprehensive summary reports")
print(" - Backward compatibility with existing system")
except Exception as e:
print(f"❌ Error during demonstration: {e}")
import traceback
traceback.print_exc()
finally:
# Clean up (optional - comment out to keep files for inspection)
# import shutil
# shutil.rmtree(temp_dir, ignore_errors=True)
# print(f"🧹 Cleaned up temporary directory")
pass
if __name__ == "__main__":
demonstrate_enhanced_verification() |