Healthcare-Expert-ai

class EnhancedRAGVoiceAgent: def __init__(self, verbose_logging=False): # AssemblyAI Universal-Streaming API v3 configuration self.assemblyai_api_key = os.getenv("ASSEMBLYAI_API_KEY") self.CONNECTION_PARAMS = { "sample_rate": 16000, "format_turns": True, # Critical for conversation-based healthcare interactions } self.API_ENDPOINT_BASE_URL = "wss://streaming.assemblyai.com/v3/ws" self.streaming_endpoint = f"{self.API_ENDPOINT_BASE_URL}?{urlencode(self.CONNECTION_PARAMS)}" # Enhanced thread management for real-time healthcare consultations self.stop_event = threading.Event() self.audio_queue = queue.Queue() self.transcription_queue = queue.Queue() 

async def setup_streaming_transcription(self): """Setup AssemblyAI Universal-Streaming WebSocket connection with enhanced thread management""" try: print("🔗 Setting up AssemblyAI Universal-Streaming API v3...") # Reset stop events for new session (healthcare session management) self.stop_event.clear() self.stop_streaming.clear() # Create WebSocketApp with v3 endpoint and healthcare-optimized headers self.ws_app = websocket.WebSocketApp( self.streaming_endpoint, header={"Authorization": self.assemblyai_api_key}, on_open=self.on_ws_open, on_message=self.on_ws_message, on_error=self.on_ws_error, on_close=self.on_ws_close, ) # Enhanced WebSocket thread management for healthcare reliability self.ws_thread = threading.Thread( target=self.ws_app.run_forever, name="HealthcareWebSocketThread" ) self.ws_thread.daemon = True self.ws_thread.start() # Wait for connection with healthcare timeout requirements for i in range(50): # 5-second healthcare connection timeout if self.ws_connected: print("✅ AssemblyAI streaming connection established for healthcare") return True await asyncio.sleep(0.1) return False except Exception as e: print(f"❌ Error setting up healthcare streaming transcription: {e}") return False 

def on_ws_message(self, ws, message): """Enhanced message handling following AssemblyAI Universal-Streaming API v3 pattern""" try: data = json.loads(message) msg_type = data.get('type') if msg_type == "Begin": session_id = data.get('id') expires_at = data.get('expires_at') if expires_at: expiry_time = datetime.fromtimestamp(expires_at) print(f"🚀 Healthcare Session began: ID={session_id}, ExpiresAt={expiry_time}") else: print(f"🚀 Healthcare Session began: ID={session_id}") elif msg_type == "Turn": transcript = data.get('transcript', '') formatted = data.get('turn_is_formatted', False) # Healthcare-optimized transcript handling  if formatted: # Clear partial transcript and show final (critical for medical accuracy)  print('\r' + ' ' * 80 + '\r', end='') print(f"🏥 Patient: {transcript}") # Enhanced healthcare transcript processing with latency tracking  if transcript.strip(): self.latency_tracker.end_timing("speech_end_to_final", "- Speech end to final transcript") self.latency_tracker.start_timing("transcript_to_ai_response") self.latency_tracker.log_integration_point("📝", "Healthcare query received") # Queue for healthcare AI processing  self.transcription_queue.put(transcript) else: # Real-time partial transcripts for healthcare interaction feedback  if self.verbose_logging: print(f"\r🎧 Listening: {transcript}", end='') elif msg_type == "Termination": audio_duration = data.get('audio_duration_seconds', 0) session_duration = data.get('session_duration_seconds', 0) print(f"🔚 Healthcare Session Terminated: Audio={audio_duration}s, Session={session_duration}s") except json.JSONDecodeError as e: print(f"❌ Error decoding healthcare message: {e}") except Exception as e: print(f"❌ Error handling healthcare message: {e}") 

def on_ws_open(self, ws): """Enhanced WebSocket open handler optimized for healthcare audio streaming""" print("🔗 AssemblyAI WebSocket connection opened for healthcare") print(f"📡 Connected to: {self.streaming_endpoint}") self.ws_connected = True self.stream_active = True def stream_healthcare_audio(): """Healthcare-optimized audio streaming with medical terminology focus""" print("🎤 Starting real-time healthcare audio streaming...") while not self.stop_event.is_set(): try: if not self.audio_queue.empty(): audio_chunk = self.audio_queue.get_nowait() if audio_chunk is not None: # Convert to int16 for AssemblyAI Universal-Streaming  audio_int16 = (audio_chunk * 32767).astype(np.int16) audio_bytes = audio_int16.tobytes() # Send 50ms chunks optimized for medical speech patterns  ws.send(audio_bytes, websocket.ABNF.OPCODE_BINARY) else: time.sleep(0.01) # Minimal delay for healthcare real-time requirements  except websocket.WebSocketConnectionClosedException: print("🔌 Healthcare WebSocket connection closed during streaming") break except Exception as e: print(f"❌ Error streaming healthcare audio: {e}") break print("🔇 Healthcare audio streaming stopped") # Start healthcare audio streaming thread  self.audio_thread = threading.Thread( target=stream_healthcare_audio, name="HealthcareAudioStreamThread" ) self.audio_thread.daemon = True self.audio_thread.start() 

def audio_callback(self, indata, frames, time, status): """Healthcare-optimized audio callback with medical environment VAD""" if status: print(f"⚠️ Healthcare audio input error: {status}") try: # Convert to mono for healthcare VAD processing  audio_mono = indata[:, 0] audio_data = (audio_mono * 32767).astype(np.int16).tobytes() # Healthcare-optimized Voice Activity Detection  try: is_speech = self.vad.is_speech(audio_data, self.sample_rate) except Exception as vad_error: # Healthcare fallback: assume speech for patient safety  is_speech = True if is_speech: # Stream immediately for healthcare real-time requirements  if self.ws_connected and not self.is_speaking: self.audio_queue.put(audio_mono) self.silence_count = 0 if not self.is_recording: self.is_recording = True self.latency_tracker.start_timing("speech_to_partial") self.latency_tracker.log_integration_point("🎤", "Patient speech detected") print("\n🎤 Patient speaking - streaming to AssemblyAI...") else: if self.is_recording: self.silence_count += 1 # Healthcare-optimized silence handling (shorter timeout for medical urgency)  if self.silence_count < self.silence_threshold: if self.ws_connected and not self.is_speaking: self.audio_queue.put(audio_mono) else: self.is_recording = False self.latency_tracker.start_timing("speech_end_to_final") print("\n🔇 Patient speech ended - processing medical query...") except Exception as e: print(f"❌ Healthcare audio callback error: {e}") 

def on_ws_error(self, ws, error): """Healthcare-focused error handling with patient safety priorities""" print(f"\n🚨 Healthcare WebSocket Error: {error}") # Healthcare emergency protocols  self.stop_event.set() # Immediate stop for patient safety  self.stop_streaming.set() # Healthcare state management  self.ws_connected = False self.stream_active = False # Log for healthcare compliance and debugging  try: self.conversation_active = False print("🏥 Healthcare session safely terminated due to connection error") except Exception as cleanup_error: print(f"⚠️ Error during healthcare emergency cleanup: {cleanup_error}") def on_ws_close(self, ws, close_status_code, close_msg): """Healthcare-compliant resource cleanup with audit trail""" print(f"\n🔚 Healthcare WebSocket Disconnected: Status={close_status_code}") # Healthcare session cleanup protocols  self.stop_event.set() self.ws_connected = False self.stream_active = False # Healthcare thread cleanup with timeout for patient safety  cleanup_threads = [ ("Healthcare Stream Thread", self.stream_thread), ("Healthcare Audio Thread", self.audio_thread) ] for thread_name, thread in cleanup_threads: if thread and thread.is_alive(): print(f"🔄 Cleaning up {thread_name} for healthcare compliance...") thread.join(timeout=1.0) # Quick timeout for healthcare responsiveness  print("✅ Healthcare session cleanup completed")