Spaces:
Sleeping
Sleeping
File size: 8,995 Bytes
e8c7fad |
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 |
# simplified_chat_handlers.py
"""
Chat handlers for simplified interface with Provider Summary replacement.
Based on Or_4.txt requirement:
"Instead of showing the final user message (consent confirmation),
show the Provider Summary message for the spiritual care team."
"""
import gradio as gr
import html
from src.interface.session_manager import SimplifiedSessionData
from src.interface.stats_handlers import get_conversation_stats
def handle_message_simplified(message: str, history, session: SimplifiedSessionData):
"""
Handle user message with Provider Summary replacement for simplified interface.
Key difference from standard handler:
- When RED flag is triggered and consent is given, the last assistant message
is replaced with the Provider Summary (message for spiritual care team)
instead of the confirmation message to the user.
- All assistant messages are prefixed with "To patient:" or "To spiritual care team:"
"""
if session is None:
session = SimplifiedSessionData()
session.update_activity()
# Apply per-session model overrides (if configured)
custom_models = getattr(session, 'custom_models', None)
if custom_models:
session.app_instance.set_model_overrides(custom_models)
else:
session.app_instance.set_model_overrides({})
# Apply per-session prompt overrides (if configured)
custom_prompts = getattr(session, 'custom_prompts', None)
if custom_prompts:
session.app_instance.set_prompt_overrides(custom_prompts)
else:
session.app_instance.set_prompt_overrides({})
# Store previous spiritual state to detect transitions
previous_state = session.app_instance.spiritual_state.spiritual_state
# Process message
new_history, status = session.app_instance.process_message(message, history)
# Check if we just transitioned to RED state (consent was given)
current_state = session.app_instance.spiritual_state.spiritual_state
consent_just_given = (previous_state.value == "awaiting_consent" and
current_state.value == "red")
# SIMPLIFIED INTERFACE MODIFICATION:
# Replace last assistant message with Provider Summary if consent was just given
if consent_just_given:
last_summary = session.app_instance.get_last_provider_summary()
if last_summary and new_history:
# Get the COHERENT NARRATIVE summary (LLM-generated from spiritual_care_message.txt)
# This is the Medical Brain compatible format, not the structured format
try:
provider_summary_text = session.app_instance.provider_summary_generator.format_coherent_paragraph(last_summary)
if not provider_summary_text:
# Fallback to structured format if coherent generation fails
provider_summary_text = session.app_instance.provider_summary_generator.format_for_export(last_summary)
except Exception as e:
print(f"ERROR: Failed to generate coherent summary: {e}")
# Fallback to structured format
provider_summary_text = session.app_instance.provider_summary_generator.format_for_export(last_summary)
# Format with clear addressee label after classification
# Provider Summary should also have classification info, then "**To spiritual care team:**"
summary_display = f"""**To spiritual care team:**
{provider_summary_text}"""
# Replace the last assistant message with the provider summary
# History format: list of dicts with 'role' and 'content'
if len(new_history) > 0:
last_item = new_history[-1]
if isinstance(last_item, dict) and last_item.get('role') == 'assistant':
# Replace the content
new_history[-1] = {
'role': 'assistant',
'content': summary_display
}
elif isinstance(last_item, (list, tuple)) and len(last_item) > 1:
# Old format: (user_msg, assistant_msg)
new_history[-1] = (last_item[0], summary_display)
print(f"DEBUG: Replaced last message with Provider Summary (coherent narrative)")
print(f"DEBUG: Summary length: {len(provider_summary_text)} chars")
else:
# Add "To patient:" after classification status for all other assistant messages
if new_history and len(new_history) > 0:
last_item = new_history[-1]
if isinstance(last_item, dict) and last_item.get('role') == 'assistant':
content = last_item.get('content', '')
# Only add prefix if not already present
if 'To patient:' not in content and 'To spiritual care team:' not in content:
# Insert "To patient:" after the classification line
# Format: "🟢 GREEN (95%)\n*Indicators...*\n*Reasoning...*\n\nActual message"
# We need to insert "To patient:\n" after the classification block
lines = content.split('\n')
# Find where classification ends (after reasoning line or after first empty line)
insert_index = 0
for i, line in enumerate(lines):
if line.strip().startswith('*') or line.strip().startswith('🟢') or line.strip().startswith('🟡') or line.strip().startswith('🔴') or line.strip().startswith('⚪'):
insert_index = i + 1
elif insert_index > 0 and line.strip() == '':
# Found empty line after classification
insert_index = i
break
# Insert "**To patient:**" at the right position (after the empty line)
if insert_index > 0 and insert_index < len(lines):
lines.insert(insert_index, '**To patient:**')
lines.insert(insert_index + 1, '') # Add empty line after "**To patient:**"
new_content = '\n'.join(lines)
else:
# Fallback: add at the beginning
new_content = f"**To patient:**\n\n{content}"
new_history[-1] = {
'role': 'assistant',
'content': new_content
}
elif isinstance(last_item, (list, tuple)) and len(last_item) > 1:
content = last_item[1]
if 'To patient:' not in content and 'To spiritual care team:' not in content:
# Same logic for tuple format
lines = content.split('\n')
insert_index = 0
for i, line in enumerate(lines):
if line.strip().startswith('*') or line.strip().startswith('🟢') or line.strip().startswith('🟡') or line.strip().startswith('🔴') or line.strip().startswith('⚪'):
insert_index = i + 1
elif insert_index > 0 and line.strip() == '':
insert_index = i + 1
break
if insert_index > 0 and insert_index < len(lines):
lines.insert(insert_index, '**To patient:**')
lines.insert(insert_index + 1, '') # Add empty line after "**To patient:**"
new_content = '\n'.join(lines)
else:
new_content = f"**To patient:**\n\n{content}"
new_history[-1] = (last_item[0], new_content)
# Get updated conversation stats
stats = get_conversation_stats(session)
return (
new_history,
status,
session,
"", # Clear input
stats
)
def handle_clear_simplified(session: SimplifiedSessionData):
"""
Handle clear chat button for simplified interface.
Resets entire session including:
- Chat history
- Spiritual monitoring state
- Conversation statistics
"""
if session is None:
session = SimplifiedSessionData()
session.update_activity()
new_history, status = session.app_instance.reset_session()
return (
new_history, # Clear chat history
status, # Reset status
session # Updated session
)
def send_example_simplified(example_text: str, history, session: SimplifiedSessionData):
"""Send example message in simplified interface."""
return handle_message_simplified(example_text, history, session)
|