From 6cb55ec2cb8254a668760f83cbad76b279e50309 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 15:05:22 -0500 Subject: [PATCH 01/11] Add GoogleSTTService --- .../foundational/07n-interruptible-google.py | 6 +- src/pipecat/services/google/google.py | 243 +++++++++++++++++- 2 files changed, 246 insertions(+), 3 deletions(-) diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index cbeea755d..a53c20a03 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.deepgram import DeepgramSTTService from pipecat.services.google import GoogleTTSService +from pipecat.services.google.google import GoogleSTTService from pipecat.services.openai import OpenAILLMService from pipecat.transcriptions.language import Language from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -46,11 +47,14 @@ async def main(): ), ) - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = GoogleSTTService( + credentials=os.getenv("GOOGLE_TEST_CREDENTIALS"), + ) tts = GoogleTTSService( voice_id="en-US-Journey-F", params=GoogleTTSService.InputParams(language=Language.EN_US), + credentials=os.getenv("GOOGLE_TEST_CREDENTIALS"), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index dea2f5ab7..bd0214a0f 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -8,6 +8,11 @@ import asyncio import base64 import io import json +import os + +# Suppress gRPC fork warnings +os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" + from dataclasses import dataclass from typing import Any, AsyncGenerator, Dict, List, Literal, Optional @@ -17,15 +22,20 @@ from pydantic import BaseModel, Field from pipecat.frames.frames import ( AudioRawFrame, + CancelFrame, + EndFrame, ErrorFrame, Frame, FunctionCallResultProperties, + InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, LLMTextFrame, LLMUpdateSettingsFrame, OpenAILLMContextAssistantTimestampFrame, + StartFrame, + TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -38,7 +48,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import ImageGenService, LLMService, TTSService +from pipecat.services.ai_services import ImageGenService, LLMService, STTService, TTSService from pipecat.services.google.frames import LLMSearchResponseFrame from pipecat.services.openai import ( OpenAIAssistantContextAggregator, @@ -51,10 +61,12 @@ try: import google.ai.generativelanguage as glm import google.generativeai as gai from google import genai - from google.cloud import texttospeech_v1 + from google.cloud import speech_v2, texttospeech_v1 + from google.cloud.speech_v2.types import cloud_speech from google.genai import types from google.generativeai.types import GenerationConfig from google.oauth2 import service_account + except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -1097,3 +1109,230 @@ class GoogleImageGenService(ImageGenService): except Exception as e: logger.error(f"{self} error generating image: {e}") yield ErrorFrame(f"Image generation error: {str(e)}") + + +class GoogleSTTService(STTService): + class InputParams(BaseModel): + language: Optional[Language] = Language.EN_US + model: Optional[str] = "latest_long" + use_separate_recognition_per_channel: Optional[bool] = False + enable_automatic_punctuation: Optional[bool] = True + enable_spoken_punctuation: Optional[bool] = False + enable_spoken_emojis: Optional[bool] = False + profanity_filter: Optional[bool] = False + enable_word_time_offsets: Optional[bool] = False + enable_word_confidence: Optional[bool] = False + + def __init__( + self, + *, + credentials: Optional[str] = None, + credentials_path: Optional[str] = None, + location: str = "global", + recognition_config: Optional[dict] = None, + sample_rate: Optional[int] = None, + params: InputParams = InputParams(), + **kwargs, + ): + super().__init__(sample_rate=sample_rate, **kwargs) + + self._location = location + self._stream = None + self._config = None + self._request_queue = asyncio.Queue() + self._streaming_task = None + + # Extract project ID and create client + if credentials: + json_account_info = json.loads(credentials) + self._project_id = json_account_info.get("project_id") + creds = service_account.Credentials.from_service_account_info(json_account_info) + elif credentials_path: + with open(credentials_path) as f: + json_account_info = json.load(f) + self._project_id = json_account_info.get("project_id") + creds = service_account.Credentials.from_service_account_file(credentials_path) + else: + raise ValueError("Either credentials or credentials_path must be provided") + + if not self._project_id: + raise ValueError("Project ID not found in credentials") + + logger.debug(f"Using project ID from credentials: {self._project_id}") + + self._client = speech_v2.SpeechAsyncClient(credentials=creds) + + self._settings = { + "language_code": self.language_to_service_language(params.language or Language.EN_US), + "model": params.model, + "use_separate_recognition_per_channel": params.use_separate_recognition_per_channel, + "enable_automatic_punctuation": params.enable_automatic_punctuation, + "enable_spoken_punctuation": params.enable_spoken_punctuation, + "enable_spoken_emojis": params.enable_spoken_emojis, + "profanity_filter": params.profanity_filter, + "enable_word_time_offsets": params.enable_word_time_offsets, + "enable_word_confidence": params.enable_word_confidence, + } + + if recognition_config: + self._settings.update(recognition_config) + + def language_to_service_language(self, language: Language) -> str: + return str(language.value) + + async def set_language(self, language: Language): + logger.info(f"Switching STT language to: [{language}]") + self._settings["language_code"] = self.language_to_service_language(language) + # Recreate stream with new language + if self._streaming_task: + await self._disconnect() + await self._connect() + + async def set_model(self, model: str): + await super().set_model(model) + self._settings["model"] = model + # Recreate stream with new model + if self._streaming_task: + await self._disconnect() + await self._connect() + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + async def _connect(self): + """Initialize streaming recognition config and stream""" + logger.debug("Connecting to Google Speech-to-Text") + + # Create recognition config with explicit audio format + self._config = cloud_speech.StreamingRecognitionConfig( + config=cloud_speech.RecognitionConfig( + explicit_decoding_config=cloud_speech.ExplicitDecodingConfig( + encoding=cloud_speech.ExplicitDecodingConfig.AudioEncoding.LINEAR16, + sample_rate_hertz=self.sample_rate, + audio_channel_count=1, + ), + language_codes=[self._settings["language_code"]], + model=self._settings["model"], + features=cloud_speech.RecognitionFeatures( + enable_automatic_punctuation=self._settings["enable_automatic_punctuation"], + enable_spoken_punctuation=self._settings["enable_spoken_punctuation"], + enable_spoken_emojis=self._settings["enable_spoken_emojis"], + profanity_filter=self._settings["profanity_filter"], + enable_word_time_offsets=self._settings["enable_word_time_offsets"], + enable_word_confidence=self._settings["enable_word_confidence"], + ), + ) + ) + + # Start the streaming task using task manager + self._streaming_task = self.create_task(self._stream_audio()) + + async def _disconnect(self): + """Clean up streaming recognition resources""" + if self._streaming_task: + logger.debug("Disconnecting from Google Speech-to-Text") + # Send sentinel value to stop request generator + await self._request_queue.put(None) + await self.cancel_task(self._streaming_task) + self._streaming_task = None + # Clear any remaining items in the queue + while not self._request_queue.empty(): + try: + self._request_queue.get_nowait() + self._request_queue.task_done() + except asyncio.QueueEmpty: + break + + async def _request_generator(self): + """Generates requests for the streaming recognize method.""" + recognizer_path = f"projects/{self._project_id}/locations/{self._location}/recognizers/_" + logger.debug(f"Using recognizer path: {recognizer_path}") + + try: + # First, send the recognition config + config_request = cloud_speech.StreamingRecognizeRequest( + recognizer=recognizer_path, + streaming_config=self._config, + ) + yield config_request + + # Then send all audio data requests + while True: + try: + audio_data = await self._request_queue.get() + if audio_data is None: # Sentinel value to stop + break + yield cloud_speech.StreamingRecognizeRequest(audio=audio_data) + except asyncio.CancelledError: + break + finally: + self._request_queue.task_done() + except Exception as e: + logger.error(f"Error in request generator: {e}") + raise + + async def _stream_audio(self): + """Handle bi-directional streaming with Google STT""" + try: + # Start bi-directional streaming + streaming_recognize = await self._client.streaming_recognize( + requests=self._request_generator() + ) + + # Process responses using task manager + response_task = self.create_task(self._process_responses(streaming_recognize)) + + # Wait for the response processing to complete + await self.wait_for_task(response_task) + + except Exception as e: + logger.error(f"Error in streaming task: {e}") + await self.push_frame(ErrorFrame(str(e))) + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Process an audio chunk for STT transcription""" + if self._streaming_task: + # Queue the audio data + await self._request_queue.put(audio) + yield None + + async def _process_responses(self, streaming_recognize): + """Process streaming recognition responses""" + try: + async for response in streaming_recognize: + if not response.results: + continue + + for result in response.results: + if not result.alternatives: + continue + + transcript = result.alternatives[0].transcript + if not transcript: + continue + + if result.is_final: + await self.push_frame( + TranscriptionFrame( + transcript, "", time_now_iso8601(), self._settings["language_code"] + ) + ) + else: + await self.push_frame( + InterimTranscriptionFrame( + transcript, "", time_now_iso8601(), self._settings["language_code"] + ) + ) + + except Exception as e: + logger.error(f"Error processing Google STT responses: {e}") + await self.push_frame(ErrorFrame(str(e))) From 17420f4d0cc701cc02fccff0383e23800a25aa78 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 16:11:33 -0500 Subject: [PATCH 02/11] Update language support --- src/pipecat/services/google/google.py | 321 ++++++++++++++++++++++++- src/pipecat/transcriptions/language.py | 2 + 2 files changed, 313 insertions(+), 10 deletions(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index bd0214a0f..bfe22a0f4 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -75,7 +75,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -def language_to_google_language(language: Language) -> Optional[str]: +def language_to_google_tts_language(language: Language) -> Optional[str]: language_map = { # Afrikaans Language.AF: "af-ZA", @@ -235,6 +235,305 @@ def language_to_google_language(language: Language) -> Optional[str]: return language_map.get(language) +def language_to_google_stt_language(language: Language) -> Optional[str]: + """Maps Language enum to Google Speech-to-Text V2 language codes. + + Args: + language: Language enum value. + + Returns: + Optional[str]: Google STT language code or None if not supported. + """ + language_map = { + # Afrikaans + Language.AF: "af-ZA", + Language.AF_ZA: "af-ZA", + # Albanian + Language.SQ: "sq-AL", + Language.SQ_AL: "sq-AL", + # Amharic + Language.AM: "am-ET", + Language.AM_ET: "am-ET", + # Arabic + Language.AR: "ar-EG", # Default to Egypt + Language.AR_AE: "ar-AE", + Language.AR_BH: "ar-BH", + Language.AR_DZ: "ar-DZ", + Language.AR_EG: "ar-EG", + Language.AR_IQ: "ar-IQ", + Language.AR_JO: "ar-JO", + Language.AR_KW: "ar-KW", + Language.AR_LB: "ar-LB", + Language.AR_MA: "ar-MA", + Language.AR_OM: "ar-OM", + Language.AR_QA: "ar-QA", + Language.AR_SA: "ar-SA", + Language.AR_SY: "ar-SY", + Language.AR_TN: "ar-TN", + Language.AR_YE: "ar-YE", + # Armenian + Language.HY: "hy-AM", + Language.HY_AM: "hy-AM", + # Azerbaijani + Language.AZ: "az-AZ", + Language.AZ_AZ: "az-AZ", + # Basque + Language.EU: "eu-ES", + Language.EU_ES: "eu-ES", + # Bengali + Language.BN: "bn-IN", # Default to India + Language.BN_BD: "bn-BD", + Language.BN_IN: "bn-IN", + # Bosnian + Language.BS: "bs-BA", + Language.BS_BA: "bs-BA", + # Bulgarian + Language.BG: "bg-BG", + Language.BG_BG: "bg-BG", + # Burmese + Language.MY: "my-MM", + Language.MY_MM: "my-MM", + # Catalan + Language.CA: "ca-ES", + Language.CA_ES: "ca-ES", + # Chinese + Language.ZH: "cmn-Hans-CN", # Default to Simplified Chinese + Language.ZH_CN: "cmn-Hans-CN", + Language.ZH_HK: "cmn-Hans-HK", + Language.ZH_TW: "cmn-Hant-TW", + Language.YUE: "yue-Hant-HK", # Cantonese + Language.YUE_CN: "yue-Hant-HK", + # Croatian + Language.HR: "hr-HR", + Language.HR_HR: "hr-HR", + # Czech + Language.CS: "cs-CZ", + Language.CS_CZ: "cs-CZ", + # Danish + Language.DA: "da-DK", + Language.DA_DK: "da-DK", + # Dutch + Language.NL: "nl-NL", # Default to Netherlands + Language.NL_BE: "nl-BE", + Language.NL_NL: "nl-NL", + # English + Language.EN: "en-US", # Default to US + Language.EN_AU: "en-AU", + Language.EN_CA: "en-CA", + Language.EN_GB: "en-GB", + Language.EN_GH: "en-GH", + Language.EN_HK: "en-HK", + Language.EN_IN: "en-IN", + Language.EN_IE: "en-IE", + Language.EN_KE: "en-KE", + Language.EN_NG: "en-NG", + Language.EN_NZ: "en-NZ", + Language.EN_PH: "en-PH", + Language.EN_SG: "en-SG", + Language.EN_TZ: "en-TZ", + Language.EN_US: "en-US", + Language.EN_ZA: "en-ZA", + # Estonian + Language.ET: "et-EE", + Language.ET_EE: "et-EE", + # Filipino + Language.FIL: "fil-PH", + Language.FIL_PH: "fil-PH", + # Finnish + Language.FI: "fi-FI", + Language.FI_FI: "fi-FI", + # French + Language.FR: "fr-FR", # Default to France + Language.FR_BE: "fr-BE", + Language.FR_CA: "fr-CA", + Language.FR_CH: "fr-CH", + Language.FR_FR: "fr-FR", + # Galician + Language.GL: "gl-ES", + Language.GL_ES: "gl-ES", + # Georgian + Language.KA: "ka-GE", + Language.KA_GE: "ka-GE", + # German + Language.DE: "de-DE", # Default to Germany + Language.DE_AT: "de-AT", + Language.DE_CH: "de-CH", + Language.DE_DE: "de-DE", + # Greek + Language.EL: "el-GR", + Language.EL_GR: "el-GR", + # Gujarati + Language.GU: "gu-IN", + Language.GU_IN: "gu-IN", + # Hebrew + Language.HE: "iw-IL", + Language.HE_IL: "iw-IL", + # Hindi + Language.HI: "hi-IN", + Language.HI_IN: "hi-IN", + # Hungarian + Language.HU: "hu-HU", + Language.HU_HU: "hu-HU", + # Icelandic + Language.IS: "is-IS", + Language.IS_IS: "is-IS", + # Indonesian + Language.ID: "id-ID", + Language.ID_ID: "id-ID", + # Italian + Language.IT: "it-IT", + Language.IT_IT: "it-IT", + Language.IT_CH: "it-CH", + # Japanese + Language.JA: "ja-JP", + Language.JA_JP: "ja-JP", + # Javanese + Language.JV: "jv-ID", + Language.JV_ID: "jv-ID", + # Kannada + Language.KN: "kn-IN", + Language.KN_IN: "kn-IN", + # Kazakh + Language.KK: "kk-KZ", + Language.KK_KZ: "kk-KZ", + # Khmer + Language.KM: "km-KH", + Language.KM_KH: "km-KH", + # Korean + Language.KO: "ko-KR", + Language.KO_KR: "ko-KR", + # Lao + Language.LO: "lo-LA", + Language.LO_LA: "lo-LA", + # Latvian + Language.LV: "lv-LV", + Language.LV_LV: "lv-LV", + # Lithuanian + Language.LT: "lt-LT", + Language.LT_LT: "lt-LT", + # Macedonian + Language.MK: "mk-MK", + Language.MK_MK: "mk-MK", + # Malay + Language.MS: "ms-MY", + Language.MS_MY: "ms-MY", + # Malayalam + Language.ML: "ml-IN", + Language.ML_IN: "ml-IN", + # Marathi + Language.MR: "mr-IN", + Language.MR_IN: "mr-IN", + # Mongolian + Language.MN: "mn-MN", + Language.MN_MN: "mn-MN", + # Nepali + Language.NE: "ne-NP", + Language.NE_NP: "ne-NP", + # Norwegian + Language.NO: "no-NO", + Language.NB: "no-NO", + Language.NB_NO: "no-NO", + # Persian + Language.FA: "fa-IR", + Language.FA_IR: "fa-IR", + # Polish + Language.PL: "pl-PL", + Language.PL_PL: "pl-PL", + # Portuguese + Language.PT: "pt-PT", # Default to Portugal + Language.PT_BR: "pt-BR", + Language.PT_PT: "pt-PT", + # Punjabi + Language.PA: "pa-Guru-IN", + Language.PA_IN: "pa-Guru-IN", + # Romanian + Language.RO: "ro-RO", + Language.RO_RO: "ro-RO", + # Russian + Language.RU: "ru-RU", + Language.RU_RU: "ru-RU", + # Serbian + Language.SR: "sr-RS", + Language.SR_RS: "sr-RS", + # Sinhala + Language.SI: "si-LK", + Language.SI_LK: "si-LK", + # Slovak + Language.SK: "sk-SK", + Language.SK_SK: "sk-SK", + # Slovenian + Language.SL: "sl-SI", + Language.SL_SI: "sl-SI", + # Spanish + Language.ES: "es-ES", # Default to Spain + Language.ES_AR: "es-AR", + Language.ES_BO: "es-BO", + Language.ES_CL: "es-CL", + Language.ES_CO: "es-CO", + Language.ES_CR: "es-CR", + Language.ES_DO: "es-DO", + Language.ES_EC: "es-EC", + Language.ES_ES: "es-ES", + Language.ES_GT: "es-GT", + Language.ES_HN: "es-HN", + Language.ES_MX: "es-MX", + Language.ES_NI: "es-NI", + Language.ES_PA: "es-PA", + Language.ES_PE: "es-PE", + Language.ES_PR: "es-PR", + Language.ES_PY: "es-PY", + Language.ES_SV: "es-SV", + Language.ES_US: "es-US", + Language.ES_UY: "es-UY", + Language.ES_VE: "es-VE", + # Sundanese + Language.SU: "su-ID", + Language.SU_ID: "su-ID", + # Swahili + Language.SW: "sw-TZ", # Default to Tanzania + Language.SW_KE: "sw-KE", + Language.SW_TZ: "sw-TZ", + # Swedish + Language.SV: "sv-SE", + Language.SV_SE: "sv-SE", + # Tamil + Language.TA: "ta-IN", # Default to India + Language.TA_IN: "ta-IN", + Language.TA_MY: "ta-MY", + Language.TA_SG: "ta-SG", + Language.TA_LK: "ta-LK", + # Telugu + Language.TE: "te-IN", + Language.TE_IN: "te-IN", + # Thai + Language.TH: "th-TH", + Language.TH_TH: "th-TH", + # Turkish + Language.TR: "tr-TR", + Language.TR_TR: "tr-TR", + # Ukrainian + Language.UK: "uk-UA", + Language.UK_UA: "uk-UA", + # Urdu + Language.UR: "ur-IN", # Default to India + Language.UR_IN: "ur-IN", + Language.UR_PK: "ur-PK", + # Uzbek + Language.UZ: "uz-UZ", + Language.UZ_UZ: "uz-UZ", + # Vietnamese + Language.VI: "vi-VN", + Language.VI_VN: "vi-VN", + # Xhosa + Language.XH: "xh-ZA", + # Zulu + Language.ZU: "zu-ZA", + Language.ZU_ZA: "zu-ZA", + } + + return language_map.get(language) + + class GoogleUserContextAggregator(OpenAIUserContextAggregator): async def _push_aggregation(self): if len(self._aggregation) > 0: @@ -939,7 +1238,7 @@ class GoogleTTSService(TTSService): return True def language_to_service_language(self, language: Language) -> Optional[str]: - return language_to_google_language(language) + return language_to_google_tts_language(language) def _construct_ssml(self, text: str) -> str: ssml = "" @@ -1163,7 +1462,9 @@ class GoogleSTTService(STTService): self._client = speech_v2.SpeechAsyncClient(credentials=creds) self._settings = { - "language_code": self.language_to_service_language(params.language or Language.EN_US), + "language_code": self.language_to_service_language(params.language) + if params.language + else "en-US", "model": params.model, "use_separate_recognition_per_channel": params.use_separate_recognition_per_channel, "enable_automatic_punctuation": params.enable_automatic_punctuation, @@ -1177,8 +1478,8 @@ class GoogleSTTService(STTService): if recognition_config: self._settings.update(recognition_config) - def language_to_service_language(self, language: Language) -> str: - return str(language.value) + def language_to_service_language(self, language: Language) -> Optional[str]: + return language_to_google_stt_language(language) async def set_language(self, language: Language): logger.info(f"Switching STT language to: [{language}]") @@ -1209,7 +1510,7 @@ class GoogleSTTService(STTService): await self._disconnect() async def _connect(self): - """Initialize streaming recognition config and stream""" + """Initialize streaming recognition config and stream.""" logger.debug("Connecting to Google Speech-to-Text") # Create recognition config with explicit audio format @@ -1237,7 +1538,7 @@ class GoogleSTTService(STTService): self._streaming_task = self.create_task(self._stream_audio()) async def _disconnect(self): - """Clean up streaming recognition resources""" + """Clean up streaming recognition resources.""" if self._streaming_task: logger.debug("Disconnecting from Google Speech-to-Text") # Send sentinel value to stop request generator @@ -1281,7 +1582,7 @@ class GoogleSTTService(STTService): raise async def _stream_audio(self): - """Handle bi-directional streaming with Google STT""" + """Handle bi-directional streaming with Google STT.""" try: # Start bi-directional streaming streaming_recognize = await self._client.streaming_recognize( @@ -1299,14 +1600,14 @@ class GoogleSTTService(STTService): await self.push_frame(ErrorFrame(str(e))) async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: - """Process an audio chunk for STT transcription""" + """Process an audio chunk for STT transcription.""" if self._streaming_task: # Queue the audio data await self._request_queue.put(audio) yield None async def _process_responses(self, streaming_recognize): - """Process streaming recognition responses""" + """Process streaming recognition responses.""" try: async for response in streaming_recognize: if not response.results: diff --git a/src/pipecat/transcriptions/language.py b/src/pipecat/transcriptions/language.py index f788885bf..b8b9fafe9 100644 --- a/src/pipecat/transcriptions/language.py +++ b/src/pipecat/transcriptions/language.py @@ -101,6 +101,7 @@ class Language(StrEnum): EN_AU = "en-AU" EN_CA = "en-CA" EN_GB = "en-GB" + EN_GH = "en-GH" EN_HK = "en-HK" EN_IE = "en-IE" EN_IN = "en-IN" @@ -208,6 +209,7 @@ class Language(StrEnum): # Italian IT = "it" IT_IT = "it-IT" + IT_CH = "it-CH" # Inuktitut IU_CANS = "iu-Cans" From b44ddf2456531461248295253b9535f7af7fb4cf Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 16:17:26 -0500 Subject: [PATCH 03/11] 07n uses all Google services --- examples/foundational/07n-interruptible-google.py | 10 +++------- src/pipecat/services/google/google.py | 2 +- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index a53c20a03..dd33c24cb 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -18,10 +18,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.services.deepgram import DeepgramSTTService -from pipecat.services.google import GoogleTTSService -from pipecat.services.google.google import GoogleSTTService -from pipecat.services.openai import OpenAILLMService +from pipecat.services.google import GoogleLLMService, GoogleSTTService, GoogleTTSService from pipecat.transcriptions.language import Language from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -48,16 +45,15 @@ async def main(): ) stt = GoogleSTTService( - credentials=os.getenv("GOOGLE_TEST_CREDENTIALS"), + params=GoogleSTTService.InputParams(language=Language.EN_US), ) tts = GoogleTTSService( voice_id="en-US-Journey-F", params=GoogleTTSService.InputParams(language=Language.EN_US), - credentials=os.getenv("GOOGLE_TEST_CREDENTIALS"), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) messages = [ { diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index bfe22a0f4..ebcde726d 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -932,7 +932,7 @@ class GoogleLLMContext(OpenAILLMContext): class GoogleLLMService(LLMService): - """This class implements inference with Google's AI models + """This class implements inference with Google's AI models. This service translates internally from OpenAILLMContext to the messages format expected by the Google AI model. We are using the OpenAILLMContext as a lingua From 9f1732c390cebb88fa0c0653a3a25886ca95b3bc Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 16:21:09 -0500 Subject: [PATCH 04/11] Update CHANGELOG and README --- CHANGELOG.md | 20 +++++++++++++++----- README.md | 2 +- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 848123fd8..37bf29b32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for Google Cloud Speech-to-Text V2 through `GoogleSTTService`. + - Added `RimeTTSService`, a new `WordTTSService`. Updated the foundational example `07q-interruptible-rime.py` to use `RimeTTSService`. @@ -21,10 +23,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 OpenAI-compatible interface. Also, added foundational example `14n-function-calling-perplexity.py`. -- Added `DailyTransport.update_remote_participants()`. This allows you to update remote participant's settings, like their permissions or which of their devices are enabled. Requires that the local participant have participant admin permission. +- Added `DailyTransport.update_remote_participants()`. This allows you to + update remote participant's settings, like their permissions or which of + their devices are enabled. Requires that the local participant have + participant admin permission. ### Changed +- Updated foundational example `07n-interruptible-google.py` to use all Google + services. + - `RimeHttpTTSService` now uses the `mistv2` model by default. - Improved error handling in `AzureTTSService` to properly detect and log @@ -40,7 +48,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 model. - `RTVIObserver` doesn't handle `LLMSearchResponseFrame` frames anymore. For - now, to handle those frames you need to create a `GoogleRTVIObserver` instead. + now, to handle those frames you need to create a `GoogleRTVIObserver` + instead. ### Deprecated @@ -63,8 +72,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an `RTVI` issue that was causing `bot-tts-text` messages to be sent before being processed by the output transport. -- Fixed an issue[#1192] in 11labs where we are trying to reconnect/disconnect the - websocket connection even when the connection is already closed. +- Fixed an issue[#1192] in 11labs where we are trying to reconnect/disconnect + the websocket connection even when the connection is already closed. ## [0.0.56] - 2025-02-06 @@ -85,7 +94,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 logging setup. - Fixed a `SentryMetrics` issue that was preventing any metrics to be sent to - Sentry and also was preventing from metrics frames to be pushed to the pipeline. + Sentry and also was preventing from metrics frames to be pushed to the + pipeline. - Fixed an issue in `BaseOutputTransport` where incoming audio would not be resampled to the desired output sample rate. diff --git a/README.md b/README.md index 99fe37f51..f992764ab 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,7 @@ pip install "pipecat-ai[option,...]" | Category | Services | Install Command Example | | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- | -| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | +| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Google](https://docs.pipecat.ai/server/services/stt/google), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | | LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | `pip install "pipecat-ai[openai]"` | | Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"` | | Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | `pip install "pipecat-ai[google]"` | From 66a6a6a2954fc3e92589fc4f73c039e3bbe31418 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 16:33:30 -0500 Subject: [PATCH 05/11] Enable interim transcriptions, add VAD events option --- src/pipecat/services/google/google.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index ebcde726d..91e698820 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -1418,9 +1418,11 @@ class GoogleSTTService(STTService): enable_automatic_punctuation: Optional[bool] = True enable_spoken_punctuation: Optional[bool] = False enable_spoken_emojis: Optional[bool] = False - profanity_filter: Optional[bool] = False + profanity_filter: Optional[bool] = True enable_word_time_offsets: Optional[bool] = False enable_word_confidence: Optional[bool] = False + enable_interim_results: Optional[bool] = True + enable_voice_activity_events: Optional[bool] = False def __init__( self, @@ -1473,6 +1475,8 @@ class GoogleSTTService(STTService): "profanity_filter": params.profanity_filter, "enable_word_time_offsets": params.enable_word_time_offsets, "enable_word_confidence": params.enable_word_confidence, + "enable_interim_results": params.enable_interim_results, + "enable_voice_activity_events": params.enable_voice_activity_events, } if recognition_config: @@ -1531,7 +1535,11 @@ class GoogleSTTService(STTService): enable_word_time_offsets=self._settings["enable_word_time_offsets"], enable_word_confidence=self._settings["enable_word_confidence"], ), - ) + ), + streaming_features=cloud_speech.StreamingRecognitionFeatures( + enable_voice_activity_events=self._settings["enable_voice_activity_events"], + interim_results=self._settings["enable_interim_results"], + ), ) # Start the streaming task using task manager From ce0358804b7a121ddb78968f21e8a7d8fa0724b4 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 16:45:49 -0500 Subject: [PATCH 06/11] Docstrings and cleanup --- src/pipecat/services/google/google.py | 55 +++++++++++++++++++++++---- 1 file changed, 48 insertions(+), 7 deletions(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 91e698820..7b2bae7c6 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -1411,7 +1411,32 @@ class GoogleImageGenService(ImageGenService): class GoogleSTTService(STTService): + """Google Cloud Speech-to-Text V2 service implementation. + + Provides real-time speech recognition using Google Cloud's Speech-to-Text V2 API + with streaming support. Handles audio transcription and optional voice activity detection. + + Attributes: + InputParams: Configuration parameters for the STT service. + """ + class InputParams(BaseModel): + """Configuration parameters for Google Speech-to-Text. + + Attributes: + language: Recognition language (defaults to US English). + model: Speech recognition model to use. + use_separate_recognition_per_channel: Process each audio channel separately. + enable_automatic_punctuation: Add punctuation to transcripts. + enable_spoken_punctuation: Include spoken punctuation in transcript. + enable_spoken_emojis: Include spoken emojis in transcript. + profanity_filter: Filter profanity from transcript. + enable_word_time_offsets: Include timing information for each word. + enable_word_confidence: Include confidence scores for each word. + enable_interim_results: Stream partial recognition results. + enable_voice_activity_events: Detect voice activity in audio. + """ + language: Optional[Language] = Language.EN_US model: Optional[str] = "latest_long" use_separate_recognition_per_channel: Optional[bool] = False @@ -1430,11 +1455,24 @@ class GoogleSTTService(STTService): credentials: Optional[str] = None, credentials_path: Optional[str] = None, location: str = "global", - recognition_config: Optional[dict] = None, sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): + """Initialize the Google STT service. + + Args: + credentials: JSON string containing Google Cloud service account credentials. + credentials_path: Path to service account credentials JSON file. + location: Google Cloud location (e.g., "global", "us-central1"). + sample_rate: Audio sample rate in Hertz. + params: Configuration parameters for the service. + **kwargs: Additional arguments passed to STTService. + + Raises: + ValueError: If neither credentials nor credentials_path is provided. + ValueError: If project ID is not found in credentials. + """ super().__init__(sample_rate=sample_rate, **kwargs) self._location = location @@ -1459,8 +1497,6 @@ class GoogleSTTService(STTService): if not self._project_id: raise ValueError("Project ID not found in credentials") - logger.debug(f"Using project ID from credentials: {self._project_id}") - self._client = speech_v2.SpeechAsyncClient(credentials=creds) self._settings = { @@ -1479,13 +1515,19 @@ class GoogleSTTService(STTService): "enable_voice_activity_events": params.enable_voice_activity_events, } - if recognition_config: - self._settings.update(recognition_config) - def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert Language enum to Google STT language code. + + Args: + language: Language enum value. + + Returns: + str: Google STT language code. + """ return language_to_google_stt_language(language) async def set_language(self, language: Language): + """Update the service's recognition language.""" logger.info(f"Switching STT language to: [{language}]") self._settings["language_code"] = self.language_to_service_language(language) # Recreate stream with new language @@ -1542,7 +1584,6 @@ class GoogleSTTService(STTService): ), ) - # Start the streaming task using task manager self._streaming_task = self.create_task(self._stream_audio()) async def _disconnect(self): From a9c2197dc6966b576b3c2334b43eb32ed5999a7d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 10 Feb 2025 16:57:44 -0500 Subject: [PATCH 07/11] Add ability to update options --- README.md | 2 +- src/pipecat/services/google/google.py | 101 +++++++++++++++++++++++++- 2 files changed, 101 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f992764ab..ec09c8f72 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,7 @@ pip install "pipecat-ai[option,...]" | Category | Services | Install Command Example | | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- | -| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Google](https://docs.pipecat.ai/server/services/stt/google), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | +| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | | LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | `pip install "pipecat-ai[openai]"` | | Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"` | | Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | `pip install "pipecat-ai[google]"` | diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 7b2bae7c6..437fa2763 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -1536,6 +1536,7 @@ class GoogleSTTService(STTService): await self._connect() async def set_model(self, model: str): + """Update the service's recognition model.""" await super().set_model(model) self._settings["model"] = model # Recreate stream with new model @@ -1555,6 +1556,104 @@ class GoogleSTTService(STTService): await super().cancel(frame) await self._disconnect() + async def update_options( + self, + *, + language: Optional[Language] = None, + model: Optional[str] = None, + enable_automatic_punctuation: Optional[bool] = None, + enable_spoken_punctuation: Optional[bool] = None, + enable_spoken_emojis: Optional[bool] = None, + profanity_filter: Optional[bool] = None, + enable_word_time_offsets: Optional[bool] = None, + enable_word_confidence: Optional[bool] = None, + enable_interim_results: Optional[bool] = None, + enable_voice_activity_events: Optional[bool] = None, + location: Optional[str] = None, + ) -> None: + """Update service options dynamically. + + Args: + language: New recognition language. + model: New recognition model. + enable_automatic_punctuation: Enable/disable automatic punctuation. + enable_spoken_punctuation: Enable/disable spoken punctuation. + enable_spoken_emojis: Enable/disable spoken emojis. + profanity_filter: Enable/disable profanity filter. + enable_word_time_offsets: Enable/disable word timing info. + enable_word_confidence: Enable/disable word confidence scores. + enable_interim_results: Enable/disable interim results. + enable_voice_activity_events: Enable/disable voice activity detection. + location: New Google Cloud location. + + Note: + Changes that affect the streaming configuration will cause + the stream to be reconnected. + """ + needs_reconnect = False + + # Update settings with new values + if language is not None: + logger.debug(f"Updating language to: {language}") + self._settings["language_code"] = self.language_to_service_language(language) + needs_reconnect = True + + if model is not None: + logger.debug(f"Updating model to: {model}") + self._settings["model"] = model + needs_reconnect = True + + if enable_automatic_punctuation is not None: + logger.debug(f"Updating automatic punctuation to: {enable_automatic_punctuation}") + self._settings["enable_automatic_punctuation"] = enable_automatic_punctuation + needs_reconnect = True + + if enable_spoken_punctuation is not None: + logger.debug(f"Updating spoken punctuation to: {enable_spoken_punctuation}") + self._settings["enable_spoken_punctuation"] = enable_spoken_punctuation + needs_reconnect = True + + if enable_spoken_emojis is not None: + logger.debug(f"Updating spoken emojis to: {enable_spoken_emojis}") + self._settings["enable_spoken_emojis"] = enable_spoken_emojis + needs_reconnect = True + + if profanity_filter is not None: + logger.debug(f"Updating profanity filter to: {profanity_filter}") + self._settings["profanity_filter"] = profanity_filter + needs_reconnect = True + + if enable_word_time_offsets is not None: + logger.debug(f"Updating word time offsets to: {enable_word_time_offsets}") + self._settings["enable_word_time_offsets"] = enable_word_time_offsets + needs_reconnect = True + + if enable_word_confidence is not None: + logger.debug(f"Updating word confidence to: {enable_word_confidence}") + self._settings["enable_word_confidence"] = enable_word_confidence + needs_reconnect = True + + if enable_interim_results is not None: + logger.debug(f"Updating interim results to: {enable_interim_results}") + self._settings["enable_interim_results"] = enable_interim_results + needs_reconnect = True + + if enable_voice_activity_events is not None: + logger.debug(f"Updating voice activity events to: {enable_voice_activity_events}") + self._settings["enable_voice_activity_events"] = enable_voice_activity_events + needs_reconnect = True + + if location is not None: + logger.debug(f"Updating location to: {location}") + self._location = location + needs_reconnect = True + + # Reconnect the stream if necessary + if needs_reconnect and self._streaming_task: + logger.debug("Reconnecting stream due to configuration changes") + await self._disconnect() + await self._connect() + async def _connect(self): """Initialize streaming recognition config and stream.""" logger.debug("Connecting to Google Speech-to-Text") @@ -1605,7 +1704,7 @@ class GoogleSTTService(STTService): async def _request_generator(self): """Generates requests for the streaming recognize method.""" recognizer_path = f"projects/{self._project_id}/locations/{self._location}/recognizers/_" - logger.debug(f"Using recognizer path: {recognizer_path}") + logger.trace(f"Using recognizer path: {recognizer_path}") try: # First, send the recognition config From 8c2071f248f1f1bbedf7ca63b983d664a90321e0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 11 Feb 2025 15:44:37 -0500 Subject: [PATCH 08/11] Add ClientOptions for region selection --- src/pipecat/services/google/google.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 437fa2763..13489263a 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -61,6 +61,7 @@ try: import google.ai.generativelanguage as glm import google.generativeai as gai from google import genai + from google.api_core.client_options import ClientOptions from google.cloud import speech_v2, texttospeech_v1 from google.cloud.speech_v2.types import cloud_speech from google.genai import types @@ -1443,7 +1444,7 @@ class GoogleSTTService(STTService): enable_automatic_punctuation: Optional[bool] = True enable_spoken_punctuation: Optional[bool] = False enable_spoken_emojis: Optional[bool] = False - profanity_filter: Optional[bool] = True + profanity_filter: Optional[bool] = False enable_word_time_offsets: Optional[bool] = False enable_word_confidence: Optional[bool] = False enable_interim_results: Optional[bool] = True @@ -1481,6 +1482,11 @@ class GoogleSTTService(STTService): self._request_queue = asyncio.Queue() self._streaming_task = None + # Configure client options based on location + client_options = None + if self._location != "global": + client_options = ClientOptions(api_endpoint=f"{self._location}-speech.googleapis.com") + # Extract project ID and create client if credentials: json_account_info = json.loads(credentials) @@ -1497,7 +1503,7 @@ class GoogleSTTService(STTService): if not self._project_id: raise ValueError("Project ID not found in credentials") - self._client = speech_v2.SpeechAsyncClient(credentials=creds) + self._client = speech_v2.SpeechAsyncClient(credentials=creds, client_options=client_options) self._settings = { "language_code": self.language_to_service_language(params.language) From 32d8f6153f2f6a619563271795c01f2b546e732f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 11 Feb 2025 16:21:58 -0500 Subject: [PATCH 09/11] Update InputParams to languages: support str or List of Languages --- .../foundational/07n-interruptible-google.py | 2 +- src/pipecat/services/google/google.py | 78 ++++++++++++------- 2 files changed, 52 insertions(+), 28 deletions(-) diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index dd33c24cb..f1cf9712e 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -45,7 +45,7 @@ async def main(): ) stt = GoogleSTTService( - params=GoogleSTTService.InputParams(language=Language.EN_US), + params=GoogleSTTService.InputParams(languages=Language.EN_US), ) tts = GoogleTTSService( diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 13489263a..736c9b506 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -14,11 +14,11 @@ import os os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" from dataclasses import dataclass -from typing import Any, AsyncGenerator, Dict, List, Literal, Optional +from typing import Any, AsyncGenerator, Dict, List, Literal, Optional, Union from loguru import logger from PIL import Image -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from pipecat.frames.frames import ( AudioRawFrame, @@ -1425,7 +1425,7 @@ class GoogleSTTService(STTService): """Configuration parameters for Google Speech-to-Text. Attributes: - language: Recognition language (defaults to US English). + languages: Single language or list of recognition languages. First language is primary. model: Speech recognition model to use. use_separate_recognition_per_channel: Process each audio channel separately. enable_automatic_punctuation: Add punctuation to transcripts. @@ -1438,7 +1438,7 @@ class GoogleSTTService(STTService): enable_voice_activity_events: Detect voice activity in audio. """ - language: Optional[Language] = Language.EN_US + languages: Union[Language, List[Language]] = Field(default_factory=lambda: [Language.EN_US]) model: Optional[str] = "latest_long" use_separate_recognition_per_channel: Optional[bool] = False enable_automatic_punctuation: Optional[bool] = True @@ -1450,6 +1450,19 @@ class GoogleSTTService(STTService): enable_interim_results: Optional[bool] = True enable_voice_activity_events: Optional[bool] = False + @field_validator("languages", mode="before") + @classmethod + def validate_languages(cls, v) -> List[Language]: + if isinstance(v, Language): + return [v] + return v + + @property + def language_list(self) -> List[Language]: + """Get languages as a guaranteed list.""" + assert isinstance(self.languages, list) + return self.languages + def __init__( self, *, @@ -1506,9 +1519,9 @@ class GoogleSTTService(STTService): self._client = speech_v2.SpeechAsyncClient(credentials=creds, client_options=client_options) self._settings = { - "language_code": self.language_to_service_language(params.language) - if params.language - else "en-US", + "language_codes": [ + self.language_to_service_language(lang) for lang in params.language_list + ], "model": params.model, "use_separate_recognition_per_channel": params.use_separate_recognition_per_channel, "enable_automatic_punctuation": params.enable_automatic_punctuation, @@ -1521,22 +1534,30 @@ class GoogleSTTService(STTService): "enable_voice_activity_events": params.enable_voice_activity_events, } - def language_to_service_language(self, language: Language) -> Optional[str]: - """Convert Language enum to Google STT language code. + def language_to_service_language(self, language: Language | List[Language]) -> str | List[str]: + """Convert Language enum(s) to Google STT language code(s). Args: - language: Language enum value. + language: Single Language enum or list of Language enums. Returns: - str: Google STT language code. + str | List[str]: Google STT language code(s). """ - return language_to_google_stt_language(language) + if isinstance(language, list): + return [language_to_google_stt_language(lang) or "en-US" for lang in language] + return language_to_google_stt_language(language) or "en-US" - async def set_language(self, language: Language): - """Update the service's recognition language.""" - logger.info(f"Switching STT language to: [{language}]") - self._settings["language_code"] = self.language_to_service_language(language) - # Recreate stream with new language + async def set_languages(self, languages: List[Language]): + """Update the service's recognition languages. + + Args: + languages: List of languages for recognition. First language is primary. + """ + logger.info(f"Switching STT languages to: {languages}") + self._settings["language_codes"] = [ + self.language_to_service_language(lang) for lang in languages + ] + # Recreate stream with new languages if self._streaming_task: await self._disconnect() await self._connect() @@ -1565,7 +1586,7 @@ class GoogleSTTService(STTService): async def update_options( self, *, - language: Optional[Language] = None, + languages: Optional[List[Language]] = None, model: Optional[str] = None, enable_automatic_punctuation: Optional[bool] = None, enable_spoken_punctuation: Optional[bool] = None, @@ -1580,7 +1601,7 @@ class GoogleSTTService(STTService): """Update service options dynamically. Args: - language: New recognition language. + languages: New list of recongition languages. model: New recognition model. enable_automatic_punctuation: Enable/disable automatic punctuation. enable_spoken_punctuation: Enable/disable spoken punctuation. @@ -1599,9 +1620,11 @@ class GoogleSTTService(STTService): needs_reconnect = False # Update settings with new values - if language is not None: - logger.debug(f"Updating language to: {language}") - self._settings["language_code"] = self.language_to_service_language(language) + if languages is not None: + logger.debug(f"Updating language to: {languages}") + self._settings["language_codes"] = [ + self.language_to_service_language(lang) for lang in languages + ] needs_reconnect = True if model is not None: @@ -1672,7 +1695,7 @@ class GoogleSTTService(STTService): sample_rate_hertz=self.sample_rate, audio_channel_count=1, ), - language_codes=[self._settings["language_code"]], + language_codes=self._settings["language_codes"], model=self._settings["model"], features=cloud_speech.RecognitionFeatures( enable_automatic_punctuation=self._settings["enable_automatic_punctuation"], @@ -1775,16 +1798,17 @@ class GoogleSTTService(STTService): if not transcript: continue + # Use the primary language (first in the list) + primary_language = self._settings["language_codes"][0] + if result.is_final: await self.push_frame( - TranscriptionFrame( - transcript, "", time_now_iso8601(), self._settings["language_code"] - ) + TranscriptionFrame(transcript, "", time_now_iso8601(), primary_language) ) else: await self.push_frame( InterimTranscriptionFrame( - transcript, "", time_now_iso8601(), self._settings["language_code"] + transcript, "", time_now_iso8601(), primary_language ) ) From 9f728aa623eabff29c27ebab7c9904567c036f3e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 11 Feb 2025 17:07:48 -0500 Subject: [PATCH 10/11] Add reconnect logic to handle Google's 5 min time limit --- src/pipecat/services/google/google.py | 85 ++++++++++++++++++++++----- 1 file changed, 70 insertions(+), 15 deletions(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 736c9b506..b5bf4adb8 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -9,6 +9,7 @@ import base64 import io import json import os +import time # Suppress gRPC fork warnings os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" @@ -1421,6 +1422,12 @@ class GoogleSTTService(STTService): InputParams: Configuration parameters for the STT service. """ + # Google Cloud's STT service has a connection time limit of 5 minutes per stream. + # They've shared an "endless streaming" example that guided this implementation: + # https://cloud.google.com/speech-to-text/docs/transcribe-streaming-audio#endless-streaming + + STREAMING_LIMIT = 240000 # 4 minutes in milliseconds + class InputParams(BaseModel): """Configuration parameters for Google Speech-to-Text. @@ -1495,6 +1502,18 @@ class GoogleSTTService(STTService): self._request_queue = asyncio.Queue() self._streaming_task = None + # Used for keep-alive logic + self._stream_start_time = 0 + self._last_audio_input = [] + self._audio_input = [] + self._result_end_time = 0 + self._is_final_end_time = 0 + self._final_request_end_time = 0 + self._bridging_offset = 0 + self._last_transcript_was_final = False + self._new_stream = True + self._restart_counter = 0 + # Configure client options based on location client_options = None if self._location != "global": @@ -1687,7 +1706,10 @@ class GoogleSTTService(STTService): """Initialize streaming recognition config and stream.""" logger.debug("Connecting to Google Speech-to-Text") - # Create recognition config with explicit audio format + # Set stream start time + self._stream_start_time = int(time.time() * 1000) + self._new_stream = True + self._config = cloud_speech.StreamingRecognitionConfig( config=cloud_speech.RecognitionConfig( explicit_decoding_config=cloud_speech.ExplicitDecodingConfig( @@ -1736,24 +1758,37 @@ class GoogleSTTService(STTService): logger.trace(f"Using recognizer path: {recognizer_path}") try: - # First, send the recognition config - config_request = cloud_speech.StreamingRecognizeRequest( + # Send initial config + yield cloud_speech.StreamingRecognizeRequest( recognizer=recognizer_path, streaming_config=self._config, ) - yield config_request - # Then send all audio data requests while True: try: audio_data = await self._request_queue.get() if audio_data is None: # Sentinel value to stop break + + # Check streaming limit + if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT: + logger.debug("Streaming limit reached, initiating graceful reconnection") + # Instead of immediate reconnection, we'll break and let the stream close naturally + self._last_audio_input = self._audio_input + self._audio_input = [] + self._restart_counter += 1 + # Put the current audio chunk back in the queue + await self._request_queue.put(audio_data) + break + + self._audio_input.append(audio_data) yield cloud_speech.StreamingRecognizeRequest(audio=audio_data) + except asyncio.CancelledError: break finally: self._request_queue.task_done() + except Exception as e: logger.error(f"Error in request generator: {e}") raise @@ -1761,16 +1796,31 @@ class GoogleSTTService(STTService): async def _stream_audio(self): """Handle bi-directional streaming with Google STT.""" try: - # Start bi-directional streaming - streaming_recognize = await self._client.streaming_recognize( - requests=self._request_generator() - ) + while True: + try: + # Start bi-directional streaming + streaming_recognize = await self._client.streaming_recognize( + requests=self._request_generator() + ) - # Process responses using task manager - response_task = self.create_task(self._process_responses(streaming_recognize)) + # Process responses + await self._process_responses(streaming_recognize) - # Wait for the response processing to complete - await self.wait_for_task(response_task) + # If we're here, check if we need to reconnect + if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT: + logger.debug("Reconnecting stream after timeout") + # Reset stream start time + self._stream_start_time = int(time.time() * 1000) + continue + else: + # Normal stream end + break + + except Exception as e: + logger.error(f"Stream error, attempting to reconnect: {e}") + await asyncio.sleep(1) # Brief delay before reconnecting + self._stream_start_time = int(time.time() * 1000) + continue except Exception as e: logger.error(f"Error in streaming task: {e}") @@ -1787,6 +1837,11 @@ class GoogleSTTService(STTService): """Process streaming recognition responses.""" try: async for response in streaming_recognize: + # Check streaming limit + if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT: + logger.debug("Stream timeout reached in response processing") + break + if not response.results: continue @@ -1798,14 +1853,15 @@ class GoogleSTTService(STTService): if not transcript: continue - # Use the primary language (first in the list) primary_language = self._settings["language_codes"][0] if result.is_final: + self._last_transcript_was_final = True await self.push_frame( TranscriptionFrame(transcript, "", time_now_iso8601(), primary_language) ) else: + self._last_transcript_was_final = False await self.push_frame( InterimTranscriptionFrame( transcript, "", time_now_iso8601(), primary_language @@ -1814,4 +1870,3 @@ class GoogleSTTService(STTService): except Exception as e: logger.error(f"Error processing Google STT responses: {e}") - await self.push_frame(ErrorFrame(str(e))) From 5d6370690c832e0257331815a740f81d82f46346 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 12 Feb 2025 10:00:48 -0500 Subject: [PATCH 11/11] Add _reconnect_if_needed to simplify reconnect logic --- src/pipecat/services/google/google.py | 38 +++++++++------------------ 1 file changed, 13 insertions(+), 25 deletions(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index b5bf4adb8..28cd0d421 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -1566,29 +1566,33 @@ class GoogleSTTService(STTService): return [language_to_google_stt_language(lang) or "en-US" for lang in language] return language_to_google_stt_language(language) or "en-US" + async def _reconnect_if_needed(self): + """Reconnect the stream if it's currently active.""" + if self._streaming_task: + logger.debug("Reconnecting stream due to configuration changes") + await self._disconnect() + await self._connect() + async def set_languages(self, languages: List[Language]): """Update the service's recognition languages. Args: languages: List of languages for recognition. First language is primary. """ - logger.info(f"Switching STT languages to: {languages}") + logger.debug(f"Switching STT languages to: {languages}") self._settings["language_codes"] = [ self.language_to_service_language(lang) for lang in languages ] # Recreate stream with new languages - if self._streaming_task: - await self._disconnect() - await self._connect() + await self._reconnect_if_needed() async def set_model(self, model: str): """Update the service's recognition model.""" + logger.debug(f"Switching STT model to: {model}") await super().set_model(model) self._settings["model"] = model # Recreate stream with new model - if self._streaming_task: - await self._disconnect() - await self._connect() + await self._reconnect_if_needed() async def start(self, frame: StartFrame): await super().start(frame) @@ -1636,71 +1640,55 @@ class GoogleSTTService(STTService): Changes that affect the streaming configuration will cause the stream to be reconnected. """ - needs_reconnect = False - # Update settings with new values if languages is not None: logger.debug(f"Updating language to: {languages}") self._settings["language_codes"] = [ self.language_to_service_language(lang) for lang in languages ] - needs_reconnect = True if model is not None: logger.debug(f"Updating model to: {model}") self._settings["model"] = model - needs_reconnect = True if enable_automatic_punctuation is not None: logger.debug(f"Updating automatic punctuation to: {enable_automatic_punctuation}") self._settings["enable_automatic_punctuation"] = enable_automatic_punctuation - needs_reconnect = True if enable_spoken_punctuation is not None: logger.debug(f"Updating spoken punctuation to: {enable_spoken_punctuation}") self._settings["enable_spoken_punctuation"] = enable_spoken_punctuation - needs_reconnect = True if enable_spoken_emojis is not None: logger.debug(f"Updating spoken emojis to: {enable_spoken_emojis}") self._settings["enable_spoken_emojis"] = enable_spoken_emojis - needs_reconnect = True if profanity_filter is not None: logger.debug(f"Updating profanity filter to: {profanity_filter}") self._settings["profanity_filter"] = profanity_filter - needs_reconnect = True if enable_word_time_offsets is not None: logger.debug(f"Updating word time offsets to: {enable_word_time_offsets}") self._settings["enable_word_time_offsets"] = enable_word_time_offsets - needs_reconnect = True if enable_word_confidence is not None: logger.debug(f"Updating word confidence to: {enable_word_confidence}") self._settings["enable_word_confidence"] = enable_word_confidence - needs_reconnect = True if enable_interim_results is not None: logger.debug(f"Updating interim results to: {enable_interim_results}") self._settings["enable_interim_results"] = enable_interim_results - needs_reconnect = True if enable_voice_activity_events is not None: logger.debug(f"Updating voice activity events to: {enable_voice_activity_events}") self._settings["enable_voice_activity_events"] = enable_voice_activity_events - needs_reconnect = True if location is not None: logger.debug(f"Updating location to: {location}") self._location = location - needs_reconnect = True - # Reconnect the stream if necessary - if needs_reconnect and self._streaming_task: - logger.debug("Reconnecting stream due to configuration changes") - await self._disconnect() - await self._connect() + # Reconnect the stream for updates + await self._reconnect_if_needed() async def _connect(self): """Initialize streaming recognition config and stream."""