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..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), [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/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index cbeea755d..f1cf9712e 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -18,9 +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.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 @@ -46,14 +44,16 @@ async def main(): ), ) - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = GoogleSTTService( + params=GoogleSTTService.InputParams(languages=Language.EN_US), + ) tts = GoogleTTSService( voice_id="en-US-Journey-F", params=GoogleTTSService.InputParams(language=Language.EN_US), ) - 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 dea2f5ab7..28cd0d421 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -8,24 +8,35 @@ import asyncio import base64 import io import json +import os +import time + +# Suppress gRPC fork warnings +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, + CancelFrame, + EndFrame, ErrorFrame, Frame, FunctionCallResultProperties, + InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, LLMTextFrame, LLMUpdateSettingsFrame, OpenAILLMContextAssistantTimestampFrame, + StartFrame, + TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -38,7 +49,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 +62,13 @@ try: import google.ai.generativelanguage as glm import google.generativeai as gai from google import genai - from google.cloud import texttospeech_v1 + 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 from google.generativeai.types import GenerationConfig from google.oauth2 import service_account + except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -63,7 +77,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", @@ -223,6 +237,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: @@ -621,7 +934,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 @@ -927,7 +1240,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 = "" @@ -1097,3 +1410,451 @@ 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): + """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. + """ + + # 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. + + Attributes: + 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. + 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. + """ + + 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 + 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 + 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, + *, + credentials: Optional[str] = None, + credentials_path: Optional[str] = None, + location: str = "global", + 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 + self._stream = None + self._config = None + 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": + 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) + 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") + + self._client = speech_v2.SpeechAsyncClient(credentials=creds, client_options=client_options) + + self._settings = { + "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, + "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, + "enable_interim_results": params.enable_interim_results, + "enable_voice_activity_events": params.enable_voice_activity_events, + } + + 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: Single Language enum or list of Language enums. + + Returns: + str | List[str]: Google STT language code(s). + """ + 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 _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.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 + 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 + await self._reconnect_if_needed() + + 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 update_options( + self, + *, + languages: Optional[List[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: + languages: New list of recongition languages. + 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. + """ + # 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 + ] + + if model is not None: + logger.debug(f"Updating model to: {model}") + self._settings["model"] = model + + 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 + + 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 + + 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 + + if profanity_filter is not None: + logger.debug(f"Updating profanity filter to: {profanity_filter}") + self._settings["profanity_filter"] = profanity_filter + + 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 + + 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 + + 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 + + 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 + + if location is not None: + logger.debug(f"Updating location to: {location}") + self._location = location + + # Reconnect the stream for updates + await self._reconnect_if_needed() + + async def _connect(self): + """Initialize streaming recognition config and stream.""" + logger.debug("Connecting to Google Speech-to-Text") + + # 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( + encoding=cloud_speech.ExplicitDecodingConfig.AudioEncoding.LINEAR16, + sample_rate_hertz=self.sample_rate, + audio_channel_count=1, + ), + language_codes=self._settings["language_codes"], + 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"], + ), + ), + streaming_features=cloud_speech.StreamingRecognitionFeatures( + enable_voice_activity_events=self._settings["enable_voice_activity_events"], + interim_results=self._settings["enable_interim_results"], + ), + ) + + 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.trace(f"Using recognizer path: {recognizer_path}") + + try: + # Send initial config + yield cloud_speech.StreamingRecognizeRequest( + recognizer=recognizer_path, + streaming_config=self._config, + ) + + 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 + + async def _stream_audio(self): + """Handle bi-directional streaming with Google STT.""" + try: + while True: + try: + # Start bi-directional streaming + streaming_recognize = await self._client.streaming_recognize( + requests=self._request_generator() + ) + + # Process responses + await self._process_responses(streaming_recognize) + + # 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}") + 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: + # 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 + + for result in response.results: + if not result.alternatives: + continue + + transcript = result.alternatives[0].transcript + if not transcript: + continue + + 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 + ) + ) + + except Exception as e: + logger.error(f"Error processing Google STT responses: {e}") 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"