From 05d53bc66fcde440c5c24574a4788e1c157e1ca1 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 28 Mar 2025 13:28:00 -0400 Subject: [PATCH] Refactor GladiaSTTService; add support for additional params --- CHANGELOG.md | 37 +++-- src/pipecat/services/gladia.py | 290 +++++++++++++++++++++++++++++---- 2 files changed, 282 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 695586744..1d6cb6cbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **voice-agent**: A minimal example of creating a voice agent with `SmallWebRTCTransport`. +- `GladiaSTTService` now have comprehensive support for the latest API config + options, including model, language detection, preprocessing, custom + vocabulary, custom spelling, translation, and message filtering options. + - Added `SmallWebRTCTransport`, a new P2P WebRTC transport. - Created two examples in `p2p-webrtc`: @@ -52,20 +56,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Pipecat services have been reorganized into packages. Each package can have one or more of the following modules (in the future new module names might be needed) depending on the services implemented: - - image: for image generation services - - llm: for LLM services - - memory: for memory services - - stt: for Speech-To-Text services - - tts: for Text-To-Speech services - - video: for video generation services - - vision: for video recognition services -### Deprecated + - image: for image generation services + - llm: for LLM services + - memory: for memory services + - stt: for Speech-To-Text services + - tts: for Text-To-Speech services + - video: for video generation services + - vision: for video recognition services -- All Pipecat services imports have been deprecated and a warning will be shown - when using the old import. The new import should be - `pipecat.services.[service].[image,llm,memory,stt,tts,video,vision]`. For - example, `from pipecat.services.openai.llm import OpenAILLMService`. +- `GladiaSTTService` now uses Gladia's default values. ### Fixed @@ -75,6 +75,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where `GoogleTTSService` was emitting two `TTSStoppedFrames`. +### Deprecated + +- All Pipecat services imports have been deprecated and a warning will be shown + when using the old import. The new import should be + `pipecat.services.[service].[image,llm,memory,stt,tts,video,vision]`. For + example, `from pipecat.services.openai.llm import OpenAILLMService`. + +- Deprecated the `language` parameter in `GladiaSTTService.InputParams` in + favor of `language_config`, which better aligns with Gladia's API. + +- Deprecated using `GladiaSTTService.InputParams` directly. Use the new + `GladiaInputParams` class instead. + ### Other - Added foundational example `37-mem0.py` demonstrating how to use the diff --git a/src/pipecat/services/gladia.py b/src/pipecat/services/gladia.py index 09bcd7aa3..73c36b74b 100644 --- a/src/pipecat/services/gladia.py +++ b/src/pipecat/services/gladia.py @@ -6,7 +6,8 @@ import base64 import json -from typing import AsyncGenerator, Optional +import warnings +from typing import Any, AsyncGenerator, Dict, List, Optional, Union import aiohttp from loguru import logger @@ -35,6 +36,14 @@ except ModuleNotFoundError as e: def language_to_gladia_language(language: Language) -> Optional[str]: + """Convert a Language enum to Gladia's language code format. + + Args: + language: The Language enum value to convert + + Returns: + The Gladia language code string or None if not supported + """ BASE_LANGUAGES = { Language.AF: "af", Language.AM: "am", @@ -129,14 +138,172 @@ def language_to_gladia_language(language: Language) -> Optional[str]: return result +# Configurations supported by Gladia +# Refer to the docs for more information: +# https://docs.gladia.io/api-reference/v2/live/init + + +class LanguageConfig(BaseModel): + """Configuration for language detection and handling. + + Attributes: + languages: List of language codes to use for transcription + code_switching: Whether to auto-detect language changes during transcription + """ + + languages: Optional[List[str]] = None + code_switching: Optional[bool] = None + + +class PreProcessingConfig(BaseModel): + """Configuration for audio pre-processing options. + + Attributes: + audio_enhancer: Whether to apply audio enhancement + speech_threshold: Sensitivity for speech detection (0-1) + """ + + audio_enhancer: Optional[bool] = None + speech_threshold: Optional[float] = None + + +class CustomVocabularyItem(BaseModel): + """Represents a custom vocabulary item with an intensity value. + + Attributes: + value: The vocabulary word or phrase + intensity: The bias intensity for this vocabulary item (0-1) + """ + + value: str + intensity: float + + +class CustomVocabularyConfig(BaseModel): + """Configuration for custom vocabulary. + + Attributes: + vocabulary: List of words/phrases or CustomVocabularyItem objects + default_intensity: Default intensity for simple string vocabulary items + """ + + vocabulary: Optional[List[Union[str, CustomVocabularyItem]]] = None + default_intensity: Optional[float] = None + + +class CustomSpellingConfig(BaseModel): + """Configuration for custom spelling rules. + + Attributes: + spelling_dictionary: Mapping of correct spellings to phonetic variations + """ + + spelling_dictionary: Optional[Dict[str, List[str]]] = None + + +class TranslationConfig(BaseModel): + """Configuration for real-time translation. + + Attributes: + target_languages: List of target language codes for translation + model: Translation model to use ("base" or "enhanced") + match_original_utterances: Whether to align translations with original utterances + """ + + target_languages: Optional[List[str]] = None + model: Optional[str] = None + match_original_utterances: Optional[bool] = None + + +class RealtimeProcessingConfig(BaseModel): + """Configuration for real-time processing features. + + Attributes: + words_accurate_timestamps: Whether to provide per-word timestamps + custom_vocabulary: Whether to enable custom vocabulary + custom_vocabulary_config: Custom vocabulary configuration + custom_spelling: Whether to enable custom spelling + custom_spelling_config: Custom spelling configuration + translation: Whether to enable translation + translation_config: Translation configuration + named_entity_recognition: Whether to enable named entity recognition + sentiment_analysis: Whether to enable sentiment analysis + """ + + words_accurate_timestamps: Optional[bool] = None + custom_vocabulary: Optional[bool] = None + custom_vocabulary_config: Optional[CustomVocabularyConfig] = None + custom_spelling: Optional[bool] = None + custom_spelling_config: Optional[CustomSpellingConfig] = None + translation: Optional[bool] = None + translation_config: Optional[TranslationConfig] = None + named_entity_recognition: Optional[bool] = None + sentiment_analysis: Optional[bool] = None + + +class MessagesConfig(BaseModel): + """Configuration for controlling which message types are sent via WebSocket. + + Attributes: + receive_partial_transcripts: Whether to receive intermediate transcription results + receive_final_transcripts: Whether to receive final transcription results + receive_speech_events: Whether to receive speech begin/end events + receive_pre_processing_events: Whether to receive pre-processing events + receive_realtime_processing_events: Whether to receive real-time processing events + receive_post_processing_events: Whether to receive post-processing events + receive_acknowledgments: Whether to receive acknowledgment messages + receive_errors: Whether to receive error messages + receive_lifecycle_events: Whether to receive lifecycle events + """ + + receive_partial_transcripts: Optional[bool] = None + receive_final_transcripts: Optional[bool] = None + receive_speech_events: Optional[bool] = None + receive_pre_processing_events: Optional[bool] = None + receive_realtime_processing_events: Optional[bool] = None + receive_post_processing_events: Optional[bool] = None + receive_acknowledgments: Optional[bool] = None + receive_errors: Optional[bool] = None + receive_lifecycle_events: Optional[bool] = None + + class GladiaSTTService(STTService): + """Speech-to-Text service using Gladia's API. + + This service connects to Gladia's WebSocket API for real-time transcription + with support for multiple languages, custom vocabulary, and various processing options. + + For complete API documentation, see: https://docs.gladia.io/api-reference/v2/live/init + """ + class InputParams(BaseModel): - language: Optional[Language] = Language.EN - endpointing: Optional[float] = 0.2 - maximum_duration_without_endpointing: Optional[int] = 10 - audio_enhancer: Optional[bool] = None - words_accurate_timestamps: Optional[bool] = None - speech_threshold: Optional[float] = 0.99 + """Configuration parameters for the Gladia STT service. + + Attributes: + encoding: Audio encoding format + bit_depth: Audio bit depth + channels: Number of audio channels + custom_metadata: Additional metadata to include with requests + endpointing: Silence duration in seconds to mark end of speech + maximum_duration_without_endpointing: Maximum utterance duration without silence + language: DEPRECATED - Use language_config instead + language_config: Detailed language configuration + pre_processing: Audio pre-processing options + realtime_processing: Real-time processing features + messages_config: WebSocket message filtering options + """ + + encoding: Optional[str] = "wav/pcm" + bit_depth: Optional[int] = 16 + channels: Optional[int] = 1 + custom_metadata: Optional[Dict[str, Any]] = None + endpointing: Optional[float] = None + maximum_duration_without_endpointing: Optional[int] = None + language: Optional[Language] = None # Deprecated + language_config: Optional[LanguageConfig] = None + pre_processing: Optional[PreProcessingConfig] = None + realtime_processing: Optional[RealtimeProcessingConfig] = None + messages_config: Optional[MessagesConfig] = None def __init__( self, @@ -145,51 +312,105 @@ class GladiaSTTService(STTService): url: str = "https://api.gladia.io/v2/live", confidence: float = 0.5, sample_rate: Optional[int] = None, + model: str = "fast", params: InputParams = InputParams(), **kwargs, ): + """Initialize the Gladia STT service. + + Args: + api_key: Gladia API key + url: Gladia API URL + confidence: Minimum confidence threshold for transcriptions + sample_rate: Audio sample rate in Hz + model: Model to use ("fast" or "accurate") + params: Additional configuration parameters + **kwargs: Additional arguments passed to the STTService + """ super().__init__(sample_rate=sample_rate, **kwargs) + + # Warn about deprecated language parameter if it's used + if params.language is not None: + warnings.warn( + "The 'language' parameter is deprecated and will be removed in a future version. " + "Use 'language_config' instead.", + DeprecationWarning, + stacklevel=2, + ) + self._api_key = api_key self._url = url - self._settings = { - "encoding": "wav/pcm", - "bit_depth": 16, - "sample_rate": 0, - "channels": 1, - "language_config": { - "languages": [self.language_to_service_language(params.language)] - if params.language - else [], - "code_switching": False, - }, - "endpointing": params.endpointing, - "maximum_duration_without_endpointing": params.maximum_duration_without_endpointing, - "pre_processing": { - "audio_enhancer": params.audio_enhancer, - "speech_threshold": params.speech_threshold, - }, - "realtime_processing": { - "words_accurate_timestamps": params.words_accurate_timestamps, - }, - } + self.set_model_name(model) self._confidence = confidence + self._params = params self._websocket = None self._receive_task = None def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert pipecat Language enum to Gladia's language code.""" return language_to_gladia_language(language) + def _prepare_settings(self) -> Dict[str, Any]: + settings = { + "encoding": self._params.encoding or "wav/pcm", + "bit_depth": self._params.bit_depth or 16, + "sample_rate": self.sample_rate, + "channels": self._params.channels or 1, + "model": self._model_name, + } + + # Add custom_metadata if provided + if self._params.custom_metadata: + settings["custom_metadata"] = self._params.custom_metadata + + # Add endpointing parameters if provided + if self._params.endpointing is not None: + settings["endpointing"] = self._params.endpointing + if self._params.maximum_duration_without_endpointing is not None: + settings["maximum_duration_without_endpointing"] = ( + self._params.maximum_duration_without_endpointing + ) + + # Add language configuration (prioritize language_config over deprecated language) + if self._params.language_config: + settings["language_config"] = self._params.language_config.dict(exclude_none=True) + elif self._params.language: # Backward compatibility for deprecated parameter + language_code = self.language_to_service_language(self._params.language) + if language_code: + settings["language_config"] = { + "languages": [language_code], + "code_switching": False, + } + + # Add pre_processing configuration if provided + if self._params.pre_processing: + settings["pre_processing"] = self._params.pre_processing.dict(exclude_none=True) + + # Add realtime_processing configuration if provided + if self._params.realtime_processing: + settings["realtime_processing"] = self._params.realtime_processing.dict( + exclude_none=True + ) + + # Add messages_config if provided + if self._params.messages_config: + settings["messages_config"] = self._params.messages_config.dict(exclude_none=True) + + return settings + async def start(self, frame: StartFrame): + """Start the Gladia STT websocket connection.""" await super().start(frame) if self._websocket: return - self._settings["sample_rate"] = self.sample_rate - response = await self._setup_gladia() + settings = self._prepare_settings() + response = await self._setup_gladia(settings) self._websocket = await websockets.connect(response["url"]) if not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler()) async def stop(self, frame: EndFrame): + """Stop the Gladia STT websocket connection.""" await super().stop(frame) await self._send_stop_recording() if self._websocket: @@ -200,6 +421,7 @@ class GladiaSTTService(STTService): self._receive_task = None async def cancel(self, frame: CancelFrame): + """Cancel the Gladia STT websocket connection.""" await super().cancel(frame) await self._websocket.close() if self._receive_task: @@ -207,17 +429,18 @@ class GladiaSTTService(STTService): self._receive_task = None async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Run speech-to-text on audio data.""" await self.start_processing_metrics() await self._send_audio(audio) await self.stop_processing_metrics() yield None - async def _setup_gladia(self): + async def _setup_gladia(self, settings: Dict[str, Any]): async with aiohttp.ClientSession() as session: async with session.post( self._url, headers={"X-Gladia-Key": self._api_key, "Content-Type": "application/json"}, - json=self._settings, + json=settings, ) as response: if response.ok: return await response.json() @@ -233,7 +456,8 @@ class GladiaSTTService(STTService): await self._websocket.send(json.dumps(message)) async def _send_stop_recording(self): - await self._websocket.send(json.dumps({"type": "stop_recording"})) + if self._websocket and not self._websocket.closed: + await self._websocket.send(json.dumps({"type": "stop_recording"})) async def _receive_task_handler(self): async for message in self._websocket: