From 14a1907c7bef3c516fb2697e87cfb2936f234959 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Tue, 9 Sep 2025 09:18:38 +0000 Subject: [PATCH 1/5] speaker service networking fix --- extras/speaker-recognition/docker-compose.yml | 4 ++ .../speaker-recognition/nginx.conf.template | 18 +++++++ services.py | 52 +++++++++++++++---- setup-requirements.txt | 3 +- 4 files changed, 66 insertions(+), 11 deletions(-) diff --git a/extras/speaker-recognition/docker-compose.yml b/extras/speaker-recognition/docker-compose.yml index 906ac01ce..0d6cdfb5b 100644 --- a/extras/speaker-recognition/docker-compose.yml +++ b/extras/speaker-recognition/docker-compose.yml @@ -44,6 +44,10 @@ services: speaker-service-gpu: <<: *base-speaker-service profiles: ["gpu"] + networks: + default: + aliases: + - speaker-service build: context: . dockerfile: Dockerfile diff --git a/extras/speaker-recognition/nginx.conf.template b/extras/speaker-recognition/nginx.conf.template index cbfbe6032..91626e23b 100644 --- a/extras/speaker-recognition/nginx.conf.template +++ b/extras/speaker-recognition/nginx.conf.template @@ -94,6 +94,24 @@ http { proxy_request_buffering off; } + # V1 WebSocket endpoints (Deepgram-compatible) + location /v1/ { + proxy_pass http://speaker_service/v1/; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection $connection_upgrade; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_cache_bypass $http_upgrade; + proxy_read_timeout 86400; + proxy_send_timeout 86400; + proxy_connect_timeout 60s; + proxy_buffering off; + proxy_request_buffering off; + } + # Health check endpoint (proxy to speaker service) location /health { proxy_pass http://speaker_service/health; diff --git a/services.py b/services.py index 95966c003..bc52b6ca4 100755 --- a/services.py +++ b/services.py @@ -8,10 +8,12 @@ import subprocess import sys from pathlib import Path +import os from rich import print as rprint from rich.console import Console from rich.table import Table +from dotenv import dotenv_values console = Console() @@ -68,16 +70,46 @@ def run_compose_command(service_name, command, build=False): return False cmd = ['docker', 'compose'] - if command == 'up': - cmd.extend(['up', '-d']) - if build: - cmd.append('--build') - elif command == 'down': - cmd.extend(['down']) - elif command == 'restart': - cmd.extend(['restart']) - elif command == 'status': - cmd.extend(['ps']) + + # Handle speaker-recognition service specially + if service_name == 'speaker-recognition' and command == 'up': + # Read configuration to determine how to start + env_file = service_path / '.env' + if env_file.exists(): + env_values = dotenv_values(env_file) + compute_mode = env_values.get('COMPUTE_MODE', 'cpu') + https_enabled = env_values.get('REACT_UI_HTTPS', 'false') + + if https_enabled.lower() == 'true': + # HTTPS mode: start with profile for all services (includes nginx) + if compute_mode == 'gpu': + cmd.extend(['--profile', 'gpu']) + else: + cmd.extend(['--profile', 'cpu']) + cmd.extend(['up', '-d']) + else: + # HTTP mode: start specific services with profile (no nginx) + if compute_mode == 'gpu': + cmd.extend(['--profile', 'gpu']) + else: + cmd.extend(['--profile', 'cpu']) + cmd.extend(['up', '-d', 'speaker-service-gpu' if compute_mode == 'gpu' else 'speaker-service-cpu', 'web-ui']) + else: + # Fallback: just start base service + cmd.extend(['up', '-d']) + else: + # Standard compose commands for other services + if command == 'up': + cmd.extend(['up', '-d']) + elif command == 'down': + cmd.extend(['down']) + elif command == 'restart': + cmd.extend(['restart']) + elif command == 'status': + cmd.extend(['ps']) + + if command == 'up' and build: + cmd.append('--build') try: # For commands that need real-time output (build), stream to console diff --git a/setup-requirements.txt b/setup-requirements.txt index 78ba74206..869e79fcc 100644 --- a/setup-requirements.txt +++ b/setup-requirements.txt @@ -1,2 +1,3 @@ # Dependencies for Friend-Lite root setup orchestrator -rich>=13.0.0 \ No newline at end of file +rich>=13.0.0 +dotenv \ No newline at end of file From eb05aeb7bd7dc26f2a0bfc899edbb5134015fea9 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Tue, 9 Sep 2025 10:54:08 +0000 Subject: [PATCH 2/5] please god please let tests pass --- .../src/advanced_omi_backend/config.py | 111 ++++++++++++++++ .../controllers/system_controller.py | 119 ++---------------- .../speaker_recognition_client.py | 68 +++++----- .../src/advanced_omi_backend/transcription.py | 14 ++- .../transcription_providers.py | 68 +++++++--- 5 files changed, 218 insertions(+), 162 deletions(-) create mode 100644 backends/advanced/src/advanced_omi_backend/config.py 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..929b5a58d 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 @@ -748,103 +749,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 +789,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 +818,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" } 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..2868dcd5e 100644 --- a/backends/advanced/src/advanced_omi_backend/transcription.py +++ b/backends/advanced/src/advanced_omi_backend/transcription.py @@ -13,6 +13,7 @@ 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 +118,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}") 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": []} From e680bb73eaf35600cbeb74361e461b454ad74f8f Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Tue, 9 Sep 2025 11:43:31 +0000 Subject: [PATCH 3/5] memory config from UI --- .../controllers/system_controller.py | 184 ++++++++++++++ .../routers/modules/system_routes.py | 38 +++ .../webui/src/components/MemorySettings.tsx | 232 ++++++++++++++++++ backends/advanced/webui/src/pages/System.tsx | 4 + backends/advanced/webui/src/services/api.ts | 13 + 5 files changed, 471 insertions(+) create mode 100644 backends/advanced/webui/src/components/MemorySettings.tsx 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 929b5a58d..5b7cec52d 100644 --- a/backends/advanced/src/advanced_omi_backend/controllers/system_controller.py +++ b/backends/advanced/src/advanced_omi_backend/controllers/system_controller.py @@ -966,3 +966,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/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/webui/src/components/MemorySettings.tsx b/backends/advanced/webui/src/components/MemorySettings.tsx new file mode 100644 index 000000000..61a8cc44d --- /dev/null +++ b/backends/advanced/webui/src/components/MemorySettings.tsx @@ -0,0 +1,232 @@ +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 { + const response = 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 { + const response = 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 */} +
+
+ +