diff --git a/backends/advanced/src/advanced_omi_backend/config.py b/backends/advanced/src/advanced_omi_backend/config.py new file mode 100644 index 000000000..08c777789 --- /dev/null +++ b/backends/advanced/src/advanced_omi_backend/config.py @@ -0,0 +1,111 @@ +""" +Configuration management for Friend-Lite backend. + +Currently contains diarization settings because they were used in multiple places +causing circular imports. Other configurations can be moved here as needed. +""" + +import json +import logging +import shutil +from pathlib import Path + +logger = logging.getLogger(__name__) + +# Default diarization settings +DEFAULT_DIARIZATION_SETTINGS = { + "diarization_source": "pyannote", + "similarity_threshold": 0.15, + "min_duration": 0.5, + "collar": 2.0, + "min_duration_off": 1.5, + "min_speakers": 2, + "max_speakers": 6 +} + +# Global cache for diarization settings +_diarization_settings = None + + +def get_diarization_config_path(): + """Get the path to the diarization config file.""" + # Try different locations in order of preference + # 1. Data directory (for persistence across container restarts) + data_path = Path("/app/data/diarization_config.json") + if data_path.parent.exists(): + return data_path + + # 2. App root directory + app_path = Path("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/app/diarization_config.json") + if app_path.parent.exists(): + return app_path + + # 3. Local development path + local_path = Path("diarization_config.json") + return local_path + + +def load_diarization_settings_from_file(): + """Load diarization settings from file or create from template.""" + global _diarization_settings + + config_path = get_diarization_config_path() + template_path = Path("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/app/diarization_config.json.template") + + # If no template, try local development path + if not template_path.exists(): + template_path = Path("diarization_config.json.template") + + # If config doesn't exist, try to copy from template + if not config_path.exists(): + if template_path.exists(): + try: + # Ensure parent directory exists + config_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy(template_path, config_path) + logger.info(f"Created diarization config from template at {config_path}") + except Exception as e: + logger.warning(f"Could not copy template to {config_path}: {e}") + + # Load from file if it exists + if config_path.exists(): + try: + with open(config_path, 'r') as f: + _diarization_settings = json.load(f) + logger.info(f"Loaded diarization settings from {config_path}") + return _diarization_settings + except Exception as e: + logger.error(f"Error loading diarization settings from {config_path}: {e}") + + # Fall back to defaults + _diarization_settings = DEFAULT_DIARIZATION_SETTINGS.copy() + logger.info("Using default diarization settings") + return _diarization_settings + + +def save_diarization_settings_to_file(settings): + """Save diarization settings to file.""" + global _diarization_settings + + config_path = get_diarization_config_path() + + try: + # Ensure parent directory exists + config_path.parent.mkdir(parents=True, exist_ok=True) + + # Write settings to file + with open(config_path, 'w') as f: + json.dump(settings, f, indent=2) + + # Update cache + _diarization_settings = settings + + logger.info(f"Saved diarization settings to {config_path}") + return True + except Exception as e: + logger.error(f"Error saving diarization settings to {config_path}: {e}") + return False + + +# Initialize settings on module load +_diarization_settings = load_diarization_settings_from_file() \ No newline at end of file diff --git a/backends/advanced/src/advanced_omi_backend/controllers/system_controller.py b/backends/advanced/src/advanced_omi_backend/controllers/system_controller.py index dab22ea74..282509dab 100644 --- a/backends/advanced/src/advanced_omi_backend/controllers/system_controller.py +++ b/backends/advanced/src/advanced_omi_backend/controllers/system_controller.py @@ -19,6 +19,7 @@ from wyoming.audio import AudioChunk from advanced_omi_backend.client_manager import generate_client_id +from advanced_omi_backend.config import load_diarization_settings_from_file, save_diarization_settings_to_file from advanced_omi_backend.database import chunks_col from advanced_omi_backend.job_tracker import FileStatus, JobStatus, get_job_tracker from advanced_omi_backend.processors import AudioProcessingItem, get_processor_manager @@ -263,6 +264,7 @@ async def process_audio_files( audio_item = AudioProcessingItem( client_id=client_id, user_id=user.user_id, + user_email=user.email, audio_chunk=chunk, timestamp=chunk.timestamp, ) @@ -636,6 +638,7 @@ async def process_files_with_content( audio_item = AudioProcessingItem( client_id=client_id, user_id=user.user_id, + user_email=user.email, audio_chunk=chunk, timestamp=chunk.timestamp, ) @@ -748,103 +751,7 @@ async def process_files_with_content( await job_tracker.update_job_status(job_id, JobStatus.FAILED, error_msg) -# Default diarization settings -DEFAULT_DIARIZATION_SETTINGS = { - "diarization_source": "pyannote", - "similarity_threshold": 0.15, - "min_duration": 0.5, - "collar": 2.0, - "min_duration_off": 1.5, - "min_speakers": 2, - "max_speakers": 6 -} - -# Global cache for diarization settings -_diarization_settings = None - - -def get_diarization_config_path(): - """Get the path to the diarization config file.""" - # Try different locations in order of preference - # 1. Data directory (for persistence across container restarts) - data_path = Path("/app/data/diarization_config.json") - if data_path.parent.exists(): - return data_path - - # 2. App root directory - app_path = Path("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/app/diarization_config.json") - if app_path.parent.exists(): - return app_path - - # 3. Local development path - local_path = Path("diarization_config.json") - return local_path - - -def load_diarization_settings_from_file(): - """Load diarization settings from file or create from template.""" - global _diarization_settings - - config_path = get_diarization_config_path() - template_path = Path("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/app/diarization_config.json.template") - - # If no template, try local development path - if not template_path.exists(): - template_path = Path("diarization_config.json.template") - - # If config doesn't exist, try to copy from template - if not config_path.exists(): - if template_path.exists(): - try: - # Ensure parent directory exists - config_path.parent.mkdir(parents=True, exist_ok=True) - shutil.copy(template_path, config_path) - logger.info(f"Created diarization config from template at {config_path}") - except Exception as e: - logger.warning(f"Could not copy template to {config_path}: {e}") - - # Load from file if it exists - if config_path.exists(): - try: - with open(config_path, 'r') as f: - _diarization_settings = json.load(f) - logger.info(f"Loaded diarization settings from {config_path}") - return _diarization_settings - except Exception as e: - logger.error(f"Error loading diarization settings from {config_path}: {e}") - - # Fall back to defaults - _diarization_settings = DEFAULT_DIARIZATION_SETTINGS.copy() - logger.info("Using default diarization settings") - return _diarization_settings - - -def save_diarization_settings_to_file(settings): - """Save diarization settings to file.""" - global _diarization_settings - - config_path = get_diarization_config_path() - - try: - # Ensure parent directory exists - config_path.parent.mkdir(parents=True, exist_ok=True) - - # Write settings to file - with open(config_path, 'w') as f: - json.dump(settings, f, indent=2) - - # Update cache - _diarization_settings = settings - - logger.info(f"Saved diarization settings to {config_path}") - return True - except Exception as e: - logger.error(f"Error saving diarization settings to {config_path}: {e}") - return False - - -# Initialize settings on module load -_diarization_settings = load_diarization_settings_from_file() +# Configuration functions moved to config.py to avoid circular imports async def get_diarization_settings(): @@ -884,26 +791,28 @@ async def save_diarization_settings(settings: dict): return JSONResponse( status_code=400, content={"error": f"Invalid value for {key}: must be integer 1-20"} ) + elif key == "diarization_source": + if not isinstance(value, str) or value not in ["pyannote", "deepgram"]: + return JSONResponse( + status_code=400, content={"error": f"Invalid value for {key}: must be 'pyannote' or 'deepgram'"} + ) else: if not isinstance(value, (int, float)) or value < 0: return JSONResponse( status_code=400, content={"error": f"Invalid value for {key}: must be positive number"} ) - # Update global settings with new values - global _diarization_settings - if _diarization_settings is None: - _diarization_settings = DEFAULT_DIARIZATION_SETTINGS.copy() - - _diarization_settings.update(settings) + # Get current settings and merge with new values + current_settings = load_diarization_settings_from_file() + current_settings.update(settings) # Save to file - if save_diarization_settings_to_file(_diarization_settings): + if save_diarization_settings_to_file(current_settings): logger.info(f"Updated and saved diarization settings: {settings}") return { "message": "Diarization settings saved successfully", - "settings": _diarization_settings, + "settings": current_settings, "status": "success" } else: @@ -911,7 +820,7 @@ async def save_diarization_settings(settings: dict): logger.warning("Settings updated in memory but file save failed") return { "message": "Settings updated (file save failed)", - "settings": _diarization_settings, + "settings": current_settings, "status": "partial" } @@ -1059,3 +968,187 @@ async def get_speaker_service_status(): "message": f"Health check failed: {str(e)}", "status": "error" } + + +# Memory Configuration Management Functions + +async def get_memory_config_raw(): + """Get current memory configuration YAML as plain text.""" + try: + from advanced_omi_backend.memory_config_loader import get_config_loader + + config_loader = get_config_loader() + config_path = config_loader.config_path + + if not os.path.exists(config_path): + return JSONResponse( + status_code=404, content={"error": f"Memory config file not found: {config_path}"} + ) + + with open(config_path, 'r') as file: + config_yaml = file.read() + + return { + "config_yaml": config_yaml, + "config_path": config_path, + "status": "success" + } + + except Exception as e: + logger.error(f"Error reading memory config: {e}") + return JSONResponse( + status_code=500, content={"error": f"Failed to read memory config: {str(e)}"} + ) + + +async def update_memory_config_raw(config_yaml: str): + """Update memory configuration YAML and hot reload.""" + try: + import yaml + from advanced_omi_backend.memory_config_loader import get_config_loader + + # First validate YAML syntax + try: + yaml.safe_load(config_yaml) + except yaml.YAMLError as e: + return JSONResponse( + status_code=400, content={"error": f"Invalid YAML syntax: {str(e)}"} + ) + + config_loader = get_config_loader() + config_path = config_loader.config_path + + # Create backup + backup_path = f"{config_path}.bak" + if os.path.exists(config_path): + shutil.copy2(config_path, backup_path) + logger.info(f"Created backup at {backup_path}") + + # Write new configuration + with open(config_path, 'w') as file: + file.write(config_yaml) + + # Hot reload configuration + reload_success = config_loader.reload_config() + + if reload_success: + logger.info("Memory configuration updated and reloaded successfully") + return { + "message": "Memory configuration updated and reloaded successfully", + "config_path": config_path, + "backup_created": os.path.exists(backup_path), + "status": "success" + } + else: + return JSONResponse( + status_code=500, content={"error": "Configuration saved but reload failed"} + ) + + except Exception as e: + logger.error(f"Error updating memory config: {e}") + return JSONResponse( + status_code=500, content={"error": f"Failed to update memory config: {str(e)}"} + ) + + +async def validate_memory_config(config_yaml: str): + """Validate memory configuration YAML syntax.""" + try: + import yaml + from advanced_omi_backend.memory_config_loader import MemoryConfigLoader + + # Parse YAML + try: + parsed_config = yaml.safe_load(config_yaml) + if not parsed_config: + return JSONResponse( + status_code=400, content={"error": "Configuration file is empty"} + ) + except yaml.YAMLError as e: + return JSONResponse( + status_code=400, content={"error": f"Invalid YAML syntax: {str(e)}"} + ) + + # Create a temporary config loader to validate structure + try: + # Create a temporary file for validation + import tempfile + with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as tmp_file: + tmp_file.write(config_yaml) + tmp_path = tmp_file.name + + # Try to load with MemoryConfigLoader to validate structure + temp_loader = MemoryConfigLoader(tmp_path) + temp_loader.validate_config() + + # Clean up temp file + os.unlink(tmp_path) + + return { + "message": "Configuration is valid", + "status": "success" + } + + except ValueError as e: + return JSONResponse( + status_code=400, content={"error": f"Configuration validation failed: {str(e)}"} + ) + + except Exception as e: + logger.error(f"Error validating memory config: {e}") + return JSONResponse( + status_code=500, content={"error": f"Failed to validate memory config: {str(e)}"} + ) + + +async def reload_memory_config(): + """Reload memory configuration from file.""" + try: + from advanced_omi_backend.memory_config_loader import get_config_loader + + config_loader = get_config_loader() + reload_success = config_loader.reload_config() + + if reload_success: + logger.info("Memory configuration reloaded successfully") + return { + "message": "Memory configuration reloaded successfully", + "config_path": config_loader.config_path, + "status": "success" + } + else: + return JSONResponse( + status_code=500, content={"error": "Failed to reload memory configuration"} + ) + + except Exception as e: + logger.error(f"Error reloading memory config: {e}") + return JSONResponse( + status_code=500, content={"error": f"Failed to reload memory config: {str(e)}"} + ) + + +async def delete_all_user_memories(user: User): + """Delete all memories for the current user.""" + try: + from advanced_omi_backend.memory import get_memory_service + + memory_service = get_memory_service() + + # Delete all memories for the user + deleted_count = await memory_service.delete_all_user_memories(user.user_id) + + logger.info(f"Deleted {deleted_count} memories for user {user.user_id}") + + return { + "message": f"Successfully deleted {deleted_count} memories", + "deleted_count": deleted_count, + "user_id": user.user_id, + "status": "success" + } + + except Exception as e: + logger.error(f"Error deleting all memories for user {user.user_id}: {e}") + return JSONResponse( + status_code=500, content={"error": f"Failed to delete memories: {str(e)}"} + ) diff --git a/backends/advanced/src/advanced_omi_backend/database.py b/backends/advanced/src/advanced_omi_backend/database.py index 7d2c303f7..907ce32e3 100644 --- a/backends/advanced/src/advanced_omi_backend/database.py +++ b/backends/advanced/src/advanced_omi_backend/database.py @@ -51,6 +51,8 @@ async def create_chunk( audio_path, client_id, timestamp, + user_id=None, + user_email=None, transcript=None, speakers_identified=None, memories=None, @@ -62,6 +64,8 @@ async def create_chunk( "audio_path": audio_path, "client_id": client_id, "timestamp": timestamp, + "user_id": user_id, + "user_email": user_email, "transcript": transcript or [], # List of conversation segments "speakers_identified": speakers_identified or [], # List of identified speakers "memories": memories or [], # List of memory references created from this audio diff --git a/backends/advanced/src/advanced_omi_backend/main.py b/backends/advanced/src/advanced_omi_backend/main.py index ae2433533..ac597f83c 100644 --- a/backends/advanced/src/advanced_omi_backend/main.py +++ b/backends/advanced/src/advanced_omi_backend/main.py @@ -602,6 +602,7 @@ async def ws_endpoint_omi( AudioProcessingItem( client_id=client_id, user_id=user.user_id, + user_email=user.email, audio_chunk=chunk, timestamp=chunk.timestamp, ) @@ -843,6 +844,7 @@ async def ws_endpoint_pcm( AudioProcessingItem( client_id=client_id, user_id=user.user_id, + user_email=user.email, audio_chunk=chunk, timestamp=chunk.timestamp, ) @@ -882,6 +884,7 @@ async def ws_endpoint_pcm( AudioProcessingItem( client_id=client_id, user_id=user.user_id, + user_email=user.email, audio_chunk=chunk, timestamp=chunk.timestamp, ) diff --git a/backends/advanced/src/advanced_omi_backend/processors.py b/backends/advanced/src/advanced_omi_backend/processors.py index cbedc117f..59c1fbd66 100644 --- a/backends/advanced/src/advanced_omi_backend/processors.py +++ b/backends/advanced/src/advanced_omi_backend/processors.py @@ -46,6 +46,7 @@ class AudioProcessingItem: client_id: str user_id: str + user_email: str audio_chunk: AudioChunk audio_uuid: Optional[str] = None timestamp: Optional[int] = None @@ -658,6 +659,8 @@ async def _audio_processor(self): audio_path=wav_filename, client_id=item.client_id, timestamp=timestamp, + user_id=item.user_id, + user_email=item.user_email, ) # Notify client state about new audio UUID diff --git a/backends/advanced/src/advanced_omi_backend/routers/modules/system_routes.py b/backends/advanced/src/advanced_omi_backend/routers/modules/system_routes.py index 782697a56..5e5d34d6f 100644 --- a/backends/advanced/src/advanced_omi_backend/routers/modules/system_routes.py +++ b/backends/advanced/src/advanced_omi_backend/routers/modules/system_routes.py @@ -128,3 +128,41 @@ async def get_enrolled_speakers(current_user: User = Depends(current_active_user async def get_speaker_service_status(current_user: User = Depends(current_superuser)): """Check speaker recognition service health status. Admin only.""" return await system_controller.get_speaker_service_status() + + +# Memory Configuration Management Endpoints + +@router.get("/admin/memory/config/raw") +async def get_memory_config_raw(current_user: User = Depends(current_superuser)): + """Get current memory configuration YAML as plain text. Admin only.""" + return await system_controller.get_memory_config_raw() + + +@router.post("/admin/memory/config/raw") +async def update_memory_config_raw( + config_yaml: str, + current_user: User = Depends(current_superuser) +): + """Update memory configuration YAML and hot reload. Admin only.""" + return await system_controller.update_memory_config_raw(config_yaml) + + +@router.post("/admin/memory/config/validate") +async def validate_memory_config( + config_yaml: str, + current_user: User = Depends(current_superuser) +): + """Validate memory configuration YAML syntax. Admin only.""" + return await system_controller.validate_memory_config(config_yaml) + + +@router.post("/admin/memory/config/reload") +async def reload_memory_config(current_user: User = Depends(current_superuser)): + """Reload memory configuration from file. Admin only.""" + return await system_controller.reload_memory_config() + + +@router.delete("/admin/memory/delete-all") +async def delete_all_user_memories(current_user: User = Depends(current_active_user)): + """Delete all memories for the current user.""" + return await system_controller.delete_all_user_memories(current_user) diff --git a/backends/advanced/src/advanced_omi_backend/speaker_recognition_client.py b/backends/advanced/src/advanced_omi_backend/speaker_recognition_client.py index e6427746b..0445b394c 100644 --- a/backends/advanced/src/advanced_omi_backend/speaker_recognition_client.py +++ b/backends/advanced/src/advanced_omi_backend/speaker_recognition_client.py @@ -63,7 +63,7 @@ async def diarize_identify_match( logger.info(f"Diarizing, identifying, and matching words for {audio_path}") # Read diarization source from existing config system - from advanced_omi_backend.controllers.system_controller import load_diarization_settings_from_file + from advanced_omi_backend.config import load_diarization_settings_from_file config = load_diarization_settings_from_file() diarization_source = config.get("diarization_source", "pyannote") @@ -78,54 +78,42 @@ async def diarize_identify_match( ) if diarization_source == "deepgram": - # DEEPGRAM PATH: Use /v1/listen with structured config - logger.info("Using Deepgram diarization path") + # DEEPGRAM DIARIZATION PATH: We EXPECT transcript has speaker info from Deepgram + # Only need speaker identification of existing segments + logger.info("Using Deepgram diarization path - transcript should have speaker segments, identifying speakers") - # Structure config for Deepgram diarization - diarization_config = {"diarization": {"provider": "deepgram"}} + # TODO: Implement proper speaker identification for Deepgram segments + # For now, use diarize-identify-match as fallback until we implement segment identification + logger.warning("Deepgram segment identification not yet implemented, using diarize-identify-match as fallback") - # Log warning if pyannote params provided - pyannote_params = ["min_speakers", "max_speakers", "collar", "min_duration_off"] - provided_params = [p for p in pyannote_params if config.get(p) is not None] - if provided_params: - logger.warning("Ignoring pyannote parameters for Deepgram diarization: %s", provided_params) - - # Add structured diarization config - form_data.add_field("diarization_config", json.dumps(diarization_config)) - - # Add speaker identification params - form_data.add_field("enhance_speakers", "true") + form_data.add_field("transcript_data", json.dumps(transcript_data)) form_data.add_field("user_id", "1") # TODO: Implement proper user mapping - form_data.add_field("speaker_confidence_threshold", str(config.get("similarity_threshold", 0.15))) + form_data.add_field("similarity_threshold", str(config.get("similarity_threshold", 0.15))) + form_data.add_field("min_duration", str(config.get("min_duration", 0.5))) - # Use /v1/listen endpoint - endpoint = "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/v1/listen" + # Use /v1/diarize-identify-match endpoint as fallback + endpoint = "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/v1/diarize-identify-match" else: # pyannote (default) - # PYANNOTE PATH: Use /v1/listen with structured config including pyannote parameters - logger.info("Using Pyannote diarization path") - - # Structure config for Pyannote diarization - diarization_config = { - "diarization": { - "provider": "pyannote", - "min_speakers": config.get("min_speakers"), - "max_speakers": config.get("max_speakers"), - "collar": config.get("collar", 2.0), - "min_duration_off": config.get("min_duration_off", 1.5) - } - } + # PYANNOTE PATH: Backend has transcript, need diarization + speaker identification + logger.info("Using Pyannote path - diarizing backend transcript and identifying speakers") - # Add structured diarization config - form_data.add_field("diarization_config", json.dumps(diarization_config)) - - # Add speaker identification params - form_data.add_field("enhance_speakers", "true") + # Send existing transcript for diarization and speaker matching + form_data.add_field("transcript_data", json.dumps(transcript_data)) form_data.add_field("user_id", "1") # TODO: Implement proper user mapping - form_data.add_field("speaker_confidence_threshold", str(config.get("similarity_threshold", 0.15))) + form_data.add_field("similarity_threshold", str(config.get("similarity_threshold", 0.15))) + + # Add pyannote diarization parameters + form_data.add_field("min_duration", str(config.get("min_duration", 0.5))) + form_data.add_field("collar", str(config.get("collar", 2.0))) + form_data.add_field("min_duration_off", str(config.get("min_duration_off", 1.5))) + if config.get("min_speakers"): + form_data.add_field("min_speakers", str(config.get("min_speakers"))) + if config.get("max_speakers"): + form_data.add_field("max_speakers", str(config.get("max_speakers"))) - # Use /v1/listen endpoint (now supports both providers) - endpoint = "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/v1/listen" + # Use /v1/diarize-identify-match endpoint for backend integration + endpoint = "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/v1/diarize-identify-match" # Make the request to the consolidated endpoint async with session.post( diff --git a/backends/advanced/src/advanced_omi_backend/transcription.py b/backends/advanced/src/advanced_omi_backend/transcription.py index 563a047d6..b92022c0a 100644 --- a/backends/advanced/src/advanced_omi_backend/transcription.py +++ b/backends/advanced/src/advanced_omi_backend/transcription.py @@ -7,12 +7,14 @@ from wyoming.audio import AudioChunk from advanced_omi_backend.client_manager import get_client_manager +from advanced_omi_backend.conversation_repository import get_conversation_repository from advanced_omi_backend.speaker_recognition_client import SpeakerRecognitionClient from advanced_omi_backend.transcript_coordinator import get_transcript_coordinator from advanced_omi_backend.transcription_providers import ( BaseTranscriptionProvider, get_transcription_provider, ) +from advanced_omi_backend.config import load_diarization_settings_from_file # ASR Configuration TRANSCRIPTION_PROVIDER = os.getenv("TRANSCRIPTION_PROVIDER") # Optional: 'deepgram' or 'parakeet' @@ -117,7 +119,18 @@ async def _get_transcript(self, audio_duration_seconds: Optional[float] = None): logger.warning("No audio data or sample rate available for transcription") return None - return await self.provider.transcribe(combined_audio, sample_rate) + # Check if we should request diarization based on configuration + config = load_diarization_settings_from_file() + diarization_source = config.get("diarization_source", "pyannote") + + # Request diarization if using Deepgram as diarization source + should_diarize = (diarization_source == "deepgram" and + self.provider.name in ["Deepgram", "Deepgram-Streaming"]) + + if should_diarize: + logger.info(f"Requesting diarization from {self.provider.name} (diarization_source=deepgram)") + + return await self.provider.transcribe(combined_audio, sample_rate, diarize=should_diarize) except Exception as e: logger.error(f"Error getting transcript from {self.provider.name}: {e}") @@ -337,27 +350,28 @@ async def _queue_memory_processing(self): This is data-driven - no event coordination needed since we know transcript is ready. """ try: - # Get current client for user info - current_client = self._get_current_client() - if not current_client: + # Get user info from persistent conversation data instead of ephemeral client state + conversation_repo = get_conversation_repository() + conversation = await conversation_repo.get_conversation(self._current_audio_uuid) + if not conversation: logger.warning( - f"No client state available for memory processing {self._current_audio_uuid}" + f"No conversation found for memory processing {self._current_audio_uuid}" ) return # Check if we have required data if not all( - [self._current_audio_uuid, current_client.user_id, current_client.user_email] + [self._current_audio_uuid, conversation.get("user_id"), conversation.get("user_email")] ): logger.warning( f"Memory processing skipped - missing required data for {self._current_audio_uuid}" ) logger.warning(f" - audio_uuid: {bool(self._current_audio_uuid)}") logger.warning( - f" - user_id: {bool(current_client.user_id) if current_client else False}" + f" - user_id: {bool(conversation.get('user_id'))}" ) logger.warning( - f" - user_email: {bool(current_client.user_email) if current_client else False}" + f" - user_email: {bool(conversation.get('user_email'))}" ) return @@ -376,8 +390,8 @@ async def _queue_memory_processing(self): await processor_manager.queue_memory( MemoryProcessingItem( client_id=self._client_id, - user_id=current_client.user_id, - user_email=current_client.user_email, + user_id=conversation["user_id"], + user_email=conversation["user_email"], audio_uuid=self._current_audio_uuid, ) ) diff --git a/backends/advanced/src/advanced_omi_backend/transcription_providers.py b/backends/advanced/src/advanced_omi_backend/transcription_providers.py index 662fc57c7..2a91f63e7 100644 --- a/backends/advanced/src/advanced_omi_backend/transcription_providers.py +++ b/backends/advanced/src/advanced_omi_backend/transcription_providers.py @@ -45,13 +45,14 @@ class BaseTranscriptionProvider(abc.ABC): """Abstract base class for all transcription providers.""" @abc.abstractmethod - async def transcribe(self, audio_data: bytes, sample_rate: int) -> dict: + async def transcribe(self, audio_data: bytes, sample_rate: int, **kwargs) -> dict: """ Transcribe audio data to text with word-level timestamps. Args: audio_data: Raw audio bytes (PCM format) sample_rate: Audio sample rate (Hz) + **kwargs: Additional parameters (e.g. diarize=True for speaker diarization) Returns: Dictionary containing: @@ -90,8 +91,14 @@ def mode(self) -> str: return "streaming" @abc.abstractmethod - async def start_stream(self, client_id: str, sample_rate: int = 16000): - """Start a transcription stream for a client.""" + async def start_stream(self, client_id: str, sample_rate: int = 16000, diarize: bool = False): + """Start a transcription stream for a client. + + Args: + client_id: Unique client identifier + sample_rate: Audio sample rate + diarize: Whether to enable speaker diarization (provider-dependent) + """ pass @abc.abstractmethod @@ -116,6 +123,17 @@ class BatchTranscriptionProvider(BaseTranscriptionProvider): @property def mode(self) -> str: return "batch" + + @abc.abstractmethod + async def transcribe(self, audio_data: bytes, sample_rate: int, diarize: bool = False) -> dict: + """Transcribe audio data. + + Args: + audio_data: Raw audio bytes + sample_rate: Audio sample rate + diarize: Whether to enable speaker diarization (provider-dependent) + """ + pass class DeepgramProvider(BatchTranscriptionProvider): @@ -129,15 +147,21 @@ def __init__(self, api_key: str): def name(self) -> str: return "Deepgram" - async def transcribe(self, audio_data: bytes, sample_rate: int) -> dict: - """Transcribe audio using Deepgram's REST API.""" + async def transcribe(self, audio_data: bytes, sample_rate: int, diarize: bool = False) -> dict: + """Transcribe audio using Deepgram's REST API. + + Args: + audio_data: Raw audio bytes + sample_rate: Audio sample rate + diarize: Whether to enable speaker diarization + """ try: params = { "model": "nova-3", "language": "multi", "smart_format": "true", "punctuate": "true", - "diarize": "false", + "diarize": "true" if diarize else "false", "encoding": "linear16", "sample_rate": str(sample_rate), "channels": "1", @@ -285,10 +309,16 @@ def __init__(self, api_key: str): def name(self) -> str: return "Deepgram-Streaming" - async def start_stream(self, client_id: str, sample_rate: int = 16000): - """Start a WebSocket connection for streaming transcription.""" + async def start_stream(self, client_id: str, sample_rate: int = 16000, diarize: bool = False): + """Start a WebSocket connection for streaming transcription. + + Args: + client_id: Unique client identifier + sample_rate: Audio sample rate + diarize: Whether to enable speaker diarization + """ try: - logger.info(f"Starting Deepgram streaming for client {client_id}") + logger.info(f"Starting Deepgram streaming for client {client_id} (diarize={diarize})") # WebSocket connection parameters params = { @@ -296,7 +326,7 @@ async def start_stream(self, client_id: str, sample_rate: int = 16000): "language": "multi", "smart_format": "true", "punctuate": "true", - "diarize": "false", + "diarize": "true" if diarize else "false", "encoding": "linear16", "sample_rate": str(sample_rate), "channels": "1", @@ -438,7 +468,7 @@ async def end_stream(self, client_id: str) -> dict: del self._streams[client_id] return {"text": "", "words": [], "segments": []} - async def transcribe(self, audio_data: bytes, sample_rate: int) -> dict: + async def transcribe(self, audio_data: bytes, sample_rate: int, **kwargs) -> dict: """For streaming provider, this method is not typically used.""" logger.warning("transcribe() called on streaming provider - use streaming methods instead") return {"text": "", "words": [], "segments": []} @@ -468,7 +498,7 @@ def __init__(self, service_url: str): def name(self) -> str: return "Parakeet" - async def transcribe(self, audio_data: bytes, sample_rate: int) -> dict: + async def transcribe(self, audio_data: bytes, sample_rate: int, **kwargs) -> dict: """Transcribe audio using Parakeet HTTP service.""" try: @@ -539,8 +569,16 @@ def __init__(self, service_url: str): def name(self) -> str: return "Parakeet-Streaming" - async def start_stream(self, client_id: str, sample_rate: int = 16000): - """Start a WebSocket connection for streaming transcription.""" + async def start_stream(self, client_id: str, sample_rate: int = 16000, diarize: bool = False): + """Start a WebSocket connection for streaming transcription. + + Args: + client_id: Unique client identifier + sample_rate: Audio sample rate + diarize: Whether to enable speaker diarization (ignored - Parakeet doesn't support diarization) + """ + if diarize: + logger.warning(f"Parakeet streaming provider does not support diarization, ignoring diarize=True for client {client_id}") try: logger.info(f"Starting Parakeet streaming for client {client_id}") @@ -707,7 +745,7 @@ async def end_stream(self, client_id: str) -> dict: del self._streams[client_id] return {"text": "", "words": [], "segments": []} - async def transcribe(self, audio_data: bytes, sample_rate: int) -> dict: + async def transcribe(self, audio_data: bytes, sample_rate: int, **kwargs) -> dict: """For streaming provider, this method is not typically used.""" logger.warning("transcribe() called on streaming provider - use streaming methods instead") return {"text": "", "words": [], "segments": []} diff --git a/backends/advanced/webui/src/components/MemorySettings.tsx b/backends/advanced/webui/src/components/MemorySettings.tsx new file mode 100644 index 000000000..d01a84eaf --- /dev/null +++ b/backends/advanced/webui/src/components/MemorySettings.tsx @@ -0,0 +1,235 @@ +import { useState, useEffect } from 'react' +import { Brain, RefreshCw, CheckCircle, Trash2, Save, RotateCcw, AlertCircle } from 'lucide-react' +import { systemApi, memoriesApi } from '../services/api' + +interface MemorySettingsProps { + className?: string +} + +export default function MemorySettings({ className }: MemorySettingsProps) { + const [configYaml, setConfigYaml] = useState('') + const [loading, setLoading] = useState(false) + const [validating, setValidating] = useState(false) + const [saving, setSaving] = useState(false) + const [deleting, setDeleting] = useState(false) + const [message, setMessage] = useState('') + const [error, setError] = useState('') + + useEffect(() => { + loadMemoryConfig() + }, []) + + const loadMemoryConfig = async () => { + setLoading(true) + setError('') + setMessage('') + + try { + const response = await systemApi.getMemoryConfigRaw() + setConfigYaml(response.data.config_yaml) + setMessage('Configuration loaded successfully') + setTimeout(() => setMessage(''), 3000) + } catch (err: any) { + setError(err.response?.data?.error || 'Failed to load memory configuration') + } finally { + setLoading(false) + } + } + + const validateConfig = async () => { + if (!configYaml.trim()) { + setError('Configuration cannot be empty') + return + } + + setValidating(true) + setError('') + setMessage('') + + try { + await systemApi.validateMemoryConfig(configYaml) + setMessage('✅ Configuration is valid') + setTimeout(() => setMessage(''), 3000) + } catch (err: any) { + setError(err.response?.data?.error || 'Validation failed') + } finally { + setValidating(false) + } + } + + const saveConfig = async () => { + if (!configYaml.trim()) { + setError('Configuration cannot be empty') + return + } + + setSaving(true) + setError('') + setMessage('') + + try { + await systemApi.updateMemoryConfigRaw(configYaml) + setMessage('✅ Configuration saved and reloaded successfully') + setTimeout(() => setMessage(''), 5000) + } catch (err: any) { + setError(err.response?.data?.error || 'Failed to save configuration') + } finally { + setSaving(false) + } + } + + const reloadConfig = async () => { + await loadMemoryConfig() + } + + const resetConfig = () => { + loadMemoryConfig() + setMessage('Configuration reset to file version') + setTimeout(() => setMessage(''), 3000) + } + + const deleteAllMemories = async () => { + const confirmed = window.confirm( + '⚠️ WARNING: This will permanently delete ALL your memories. This action cannot be undone.\n\nAre you sure you want to continue?' + ) + + if (!confirmed) return + + const doubleConfirmed = window.confirm( + '🚨 FINAL CONFIRMATION: You are about to delete ALL memories permanently.\n\nType "DELETE" in the next dialog to confirm.' + ) + + if (!doubleConfirmed) return + + const userInput = window.prompt('Type "DELETE" to confirm memory deletion:') + if (userInput !== 'DELETE') { + setMessage('Deletion cancelled - confirmation text did not match') + return + } + + setDeleting(true) + setError('') + setMessage('') + + try { + const response = await memoriesApi.deleteAll() + setMessage(`✅ Successfully deleted ${response.data.deleted_count || 'all'} memories`) + setTimeout(() => setMessage(''), 5000) + } catch (err: any) { + setError(err.response?.data?.error || 'Failed to delete memories') + } finally { + setDeleting(false) + } + } + + return ( +
+
+ {/* Header */} +
+
+ +

+ Memory Configuration +

+
+
+ + +
+
+ +

+ Edit the memory extraction configuration. This controls how memories are extracted from conversations, quality + control settings, and processing parameters. +

+ + {/* Status Messages */} + {message && ( +
+

{message}

+
+ )} + + {error && ( +
+
+ +

{error}

+
+
+ )} + + {/* YAML Editor */} +
+
+ +