Spaces:
Sleeping
Sleeping
File size: 3,851 Bytes
60d4850 | 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 | """
Pydantic models for AI Voice Assistant conversation state and messaging.
"""
import uuid
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class ConversationState(str, Enum):
"""Dialogue state machine states."""
GREETING = "greeting"
CHIEF_COMPLAINT = "chief_complaint"
SYMPTOM_DETAILS = "symptom_details"
FOLLOW_UP = "follow_up"
SUMMARY = "summary"
EMERGENCY_ESCALATION = "emergency_escalation"
ENDED = "ended"
class ConversationMode(str, Enum):
"""Who is using the voice assistant."""
PATIENT = "patient"
CLINICIAN = "clinician"
AMBIENT = "ambient" # Phase 3: passive ambient listening mode
class ConversationTurn(BaseModel):
"""A single turn in the conversation."""
role: str # "assistant" or "user"
content: str
timestamp: datetime = Field(default_factory=datetime.utcnow)
state: Optional[ConversationState] = None
entities_extracted: Optional[Dict[str, Any]] = None
class AssistantResponse(BaseModel):
"""Response from the dialogue manager for a single turn."""
text: str
state: ConversationState
previous_state: Optional[ConversationState] = None
entities_update: Optional[Dict[str, Any]] = None
is_final: bool = False
is_emergency: bool = False
rag_grounded: bool = False
documentation: Optional[Dict[str, Any]] = None
class ConversationSessionData(BaseModel):
"""In-memory conversation session state."""
session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
mode: ConversationMode = ConversationMode.PATIENT
state: ConversationState = ConversationState.GREETING
turns: List[ConversationTurn] = Field(default_factory=list)
extracted_entities: Dict[str, Any] = Field(default_factory=lambda: {
"conditions": [],
"medications": [],
})
collected_symptoms: Dict[str, Any] = Field(default_factory=dict)
rag_context: Optional[Dict[str, Any]] = None
accumulated_transcript: str = ""
followup_round: int = 0
language: str = "en"
created_at: datetime = Field(default_factory=datetime.utcnow)
def add_turn(self, role: str, content: str, **kwargs):
"""Add a conversation turn."""
turn = ConversationTurn(
role=role,
content=content,
state=self.state,
**kwargs,
)
self.turns.append(turn)
if role == "user":
self.accumulated_transcript += f" {content}" if self.accumulated_transcript else content
def get_conversation_history(self) -> List[Dict[str, str]]:
"""Get conversation history formatted for LLM chat template."""
return [
{"role": t.role, "content": t.content}
for t in self.turns
]
# WebSocket protocol messages
class WSClientAction(BaseModel):
"""Message from client to server."""
action: str # "start", "stop", "text_input", "interrupt"
mode: Optional[str] = None
language: Optional[str] = None
text: Optional[str] = None
class WSServerMessage(BaseModel):
"""Message from server to client."""
type: str # "connected", "assistant_text", "assistant_audio", "user_transcript",
# "entities_update", "state_change", "summary", "error"
session_id: Optional[str] = None
text: Optional[str] = None
audio: Optional[str] = None # base64-encoded WAV
format: Optional[str] = None
sample_rate: Optional[int] = None
state: Optional[str] = None
from_state: Optional[str] = None
to_state: Optional[str] = None
entities: Optional[Dict[str, Any]] = None
documentation: Optional[Dict[str, Any]] = None
message: Optional[str] = None
is_final: bool = False
is_emergency: bool = False
rag_grounded: bool = False
|