From f6112713e84ce89fab7b90f92882cef4542b3aca Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 1 Jul 2025 10:29:20 -0700 Subject: [PATCH 1/2] Add user_id to TranscriptionFrame and InterimTranscriptionFrame pushed by STTServices --- CHANGELOG.md | 6 ++++++ src/pipecat/services/assemblyai/stt.py | 4 ++-- src/pipecat/services/aws/stt.py | 4 ++-- src/pipecat/services/azure/stt.py | 2 +- src/pipecat/services/cartesia/stt.py | 14 ++++++++++++-- src/pipecat/services/deepgram/stt.py | 4 ++-- src/pipecat/services/fal/stt.py | 2 +- src/pipecat/services/gladia/stt.py | 4 ++-- src/pipecat/services/google/stt.py | 4 ++-- src/pipecat/services/riva/stt.py | 9 ++++++--- src/pipecat/services/stt_service.py | 22 ++++++++++++++++++++++ src/pipecat/services/whisper/base_stt.py | 6 +++++- src/pipecat/services/whisper/stt.py | 14 ++++++++++++-- 13 files changed, 75 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 834faaa94..f3c683b8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +<<<<<<< HEAD - Added support for providing "direct" functions, which don't need an accompanying `FunctionSchema` or function definition dict. Instead, metadata (i.e. `name`, `description`, `properties`, and `required`) are automatically @@ -39,6 +40,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 tools = ToolsSchema(standard_tools=[do_something]) ``` +======= +- `user_id` is now populated in the `TranscriptionFrame` and + `InterimTranscriptionFrame` when using a service that provides a `user_id`, + like `DailyTransport` or `LiveKitTransport`. +>>>>>>> 5f958226 (Add user_id to TranscriptionFrame and InterimTranscriptionFrame pushed by STTServices) - Added `watchdog_coroutine()`. This is a watchdog helper for couroutines. So, if you have a coroutine that is waiting for a result and that takes a long diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index 0e7103885..5d8a596ca 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -311,7 +311,7 @@ class AssemblyAISTTService(STTService): await self.push_frame( TranscriptionFrame( message.transcript, - "", # participant + self._user_id, time_now_iso8601(), self._language, message, @@ -323,7 +323,7 @@ class AssemblyAISTTService(STTService): await self.push_frame( InterimTranscriptionFrame( message.transcript, - "", # participant + self._user_id, time_now_iso8601(), self._language, message, diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index a7f8fea97..22e8c8e63 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -366,7 +366,7 @@ class AWSTranscribeSTTService(STTService): await self.push_frame( TranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), self._settings["language"], result=result, @@ -382,7 +382,7 @@ class AWSTranscribeSTTService(STTService): await self.push_frame( InterimTranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), self._settings["language"], result=result, diff --git a/src/pipecat/services/azure/stt.py b/src/pipecat/services/azure/stt.py index 415f91550..6965ba188 100644 --- a/src/pipecat/services/azure/stt.py +++ b/src/pipecat/services/azure/stt.py @@ -183,7 +183,7 @@ class AzureSTTService(STTService): language = getattr(event.result, "language", None) or self._settings.get("language") frame = TranscriptionFrame( event.result.text, - "", + self._user_id, time_now_iso8601(), language, result=event, diff --git a/src/pipecat/services/cartesia/stt.py b/src/pipecat/services/cartesia/stt.py index 5ca9cce0f..50d42e467 100644 --- a/src/pipecat/services/cartesia/stt.py +++ b/src/pipecat/services/cartesia/stt.py @@ -289,14 +289,24 @@ class CartesiaSTTService(STTService): await self.stop_ttfb_metrics() if is_final: await self.push_frame( - TranscriptionFrame(transcript, "", time_now_iso8601(), language) + TranscriptionFrame( + transcript, + self._user_id, + time_now_iso8601(), + language, + ) ) await self._handle_transcription(transcript, is_final, language) await self.stop_processing_metrics() else: # For interim transcriptions, just push the frame without tracing await self.push_frame( - InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language) + InterimTranscriptionFrame( + transcript, + self._user_id, + time_now_iso8601(), + language, + ) ) async def _disconnect(self): diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index d897d8c92..d30e4da39 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -278,7 +278,7 @@ class DeepgramSTTService(STTService): await self.push_frame( TranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), language, result=result, @@ -291,7 +291,7 @@ class DeepgramSTTService(STTService): await self.push_frame( InterimTranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), language, result=result, diff --git a/src/pipecat/services/fal/stt.py b/src/pipecat/services/fal/stt.py index 3485a7de1..202c03c1b 100644 --- a/src/pipecat/services/fal/stt.py +++ b/src/pipecat/services/fal/stt.py @@ -291,7 +291,7 @@ class FalSTTService(SegmentedSTTService): logger.debug(f"Transcription: [{text}]") yield TranscriptionFrame( text, - "", + self._user_id, time_now_iso8601(), Language(self._settings["language"]), result=response, diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index c436a7ea9..f931993b3 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -567,7 +567,7 @@ class GladiaSTTService(STTService): await self.push_frame( TranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), language, result=content, @@ -582,7 +582,7 @@ class GladiaSTTService(STTService): await self.push_frame( InterimTranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), language, result=content, diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index e94fbbb12..9cd2ac3ae 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -862,7 +862,7 @@ class GoogleSTTService(STTService): await self.push_frame( TranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), primary_language, result=result, @@ -880,7 +880,7 @@ class GoogleSTTService(STTService): await self.push_frame( InterimTranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), primary_language, result=result, diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index ba8750f91..a7d114a12 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -314,7 +314,7 @@ class RivaSTTService(STTService): await self.push_frame( TranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), self._language_code, result=result, @@ -329,7 +329,7 @@ class RivaSTTService(STTService): await self.push_frame( InterimTranscriptionFrame( transcript, - "", + self._user_id, time_now_iso8601(), self._language_code, result=result, @@ -636,7 +636,10 @@ class RivaSegmentedSTTService(SegmentedSTTService): if text: logger.debug(f"Transcription: [{text}]") yield TranscriptionFrame( - text, "", time_now_iso8601(), self._language_enum + text, + self._user_id, + time_now_iso8601(), + self._language_enum, ) transcription_found = True diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py index db777c77f..5aebe10f3 100644 --- a/src/pipecat/services/stt_service.py +++ b/src/pipecat/services/stt_service.py @@ -57,6 +57,7 @@ class STTService(AIService): self._sample_rate = 0 self._settings: Dict[str, Any] = {} self._muted: bool = False + self._user_id: str = "" @property def is_muted(self) -> bool: @@ -132,6 +133,11 @@ class STTService(AIService): async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection): """Process an audio frame for speech recognition. + If the service is muted, this method does nothing. Otherwise, it + processes the audio frame and runs speech-to-text on it, yielding + transcription results. If the frame has a user_id, it is stored + for later use in transcription. + Args: frame: The audio frame to process. direction: The direction of frame processing. @@ -139,6 +145,13 @@ class STTService(AIService): if self._muted: return + # UserAudioRawFrame contains a user_id (e.g. Daily, Livekit) + if hasattr(frame, "user_id"): + self._user_id = frame.user_id + # AudioRawFrame does not have a user_id (e.g. SmallWebRTCTransport, websockets) + else: + self._user_id = "" + await self.process_generator(self.run_stt(frame.audio)) async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -241,10 +254,19 @@ class SegmentedSTTService(STTService): Continuously buffers audio, growing the buffer while user is speaking and maintaining a small buffer when not speaking to account for VAD delay. + If the frame has a user_id, it is stored for later use in transcription. + Args: frame: The audio frame to process. direction: The direction of frame processing. """ + # UserAudioRawFrame contains a user_id (e.g. Daily, Livekit) + if hasattr(frame, "user_id"): + self._user_id = frame.user_id + # AudioRawFrame does not have a user_id (e.g. SmallWebRTCTransport, websockets) + else: + self._user_id = "" + # If the user is speaking the audio buffer will keep growing. self._audio_buffer += frame.audio diff --git a/src/pipecat/services/whisper/base_stt.py b/src/pipecat/services/whisper/base_stt.py index 6f8ac26f2..a7dc6244c 100644 --- a/src/pipecat/services/whisper/base_stt.py +++ b/src/pipecat/services/whisper/base_stt.py @@ -219,7 +219,11 @@ class BaseWhisperSTTService(SegmentedSTTService): if text: await self._handle_transcription(text, True, self._language) logger.debug(f"Transcription: [{text}]") - yield TranscriptionFrame(text, "", time_now_iso8601()) + yield TranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + ) else: logger.warning("Received empty transcription from API") diff --git a/src/pipecat/services/whisper/stt.py b/src/pipecat/services/whisper/stt.py index ace18ab56..559c0a1e1 100644 --- a/src/pipecat/services/whisper/stt.py +++ b/src/pipecat/services/whisper/stt.py @@ -395,7 +395,12 @@ class WhisperSTTService(SegmentedSTTService): if text: await self._handle_transcription(text, True, self._settings["language"]) logger.debug(f"Transcription: [{text}]") - yield TranscriptionFrame(text, "", time_now_iso8601(), self._settings["language"]) + yield TranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + self._settings["language"], + ) class WhisperSTTServiceMLX(WhisperSTTService): @@ -500,7 +505,12 @@ class WhisperSTTServiceMLX(WhisperSTTService): if text: await self._handle_transcription(text, True, self._settings["language"]) logger.debug(f"Transcription: [{text}]") - yield TranscriptionFrame(text, "", time_now_iso8601(), self._settings["language"]) + yield TranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + self._settings["language"], + ) except Exception as e: logger.exception(f"MLX Whisper transcription error: {e}") From 8cbce555e423e490753ff790febc3b99b789d63b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 1 Jul 2025 11:09:57 -0700 Subject: [PATCH 2/2] Add user_id to stt_traced decorator --- CHANGELOG.md | 10 ++++------ src/pipecat/utils/tracing/service_attributes.py | 5 +++++ src/pipecat/utils/tracing/service_decorators.py | 1 + 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f3c683b8b..f1e873569 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -<<<<<<< HEAD - Added support for providing "direct" functions, which don't need an accompanying `FunctionSchema` or function definition dict. Instead, metadata (i.e. `name`, `description`, `properties`, and `required`) are automatically @@ -40,11 +39,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 tools = ToolsSchema(standard_tools=[do_something]) ``` -======= -- `user_id` is now populated in the `TranscriptionFrame` and - `InterimTranscriptionFrame` when using a service that provides a `user_id`, - like `DailyTransport` or `LiveKitTransport`. ->>>>>>> 5f958226 (Add user_id to TranscriptionFrame and InterimTranscriptionFrame pushed by STTServices) + + - `user_id` is now populated in the `TranscriptionFrame` and + `InterimTranscriptionFrame` when using a transport that provides a + `user_id`, like `DailyTransport` or `LiveKitTransport`. - Added `watchdog_coroutine()`. This is a watchdog helper for couroutines. So, if you have a coroutine that is waiting for a result and that takes a long diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py index 3896bd028..fa9d21904 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -125,6 +125,7 @@ def add_stt_span_attributes( transcript: Optional[str] = None, is_final: Optional[bool] = None, language: Optional[str] = None, + user_id: Optional[str] = None, settings: Optional[Dict[str, Any]] = None, vad_enabled: bool = False, ttfb: Optional[float] = None, @@ -140,6 +141,7 @@ def add_stt_span_attributes( transcript: The transcribed text. is_final: Whether this is a final transcript. language: Detected or configured language. + user_id: User ID associated with the audio being transcribed. settings: Service configuration settings. vad_enabled: Whether voice activity detection is enabled. ttfb: Time to first byte in seconds. @@ -161,6 +163,9 @@ def add_stt_span_attributes( if language: span.set_attribute("language", language) + if user_id: + span.set_attribute("user_id", user_id) + if ttfb is not None: span.set_attribute("metrics.ttfb", ttfb) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 9078bba15..c5612d2b2 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -270,6 +270,7 @@ def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) - transcript=transcript, is_final=is_final, language=str(language) if language else None, + user_id=getattr(self, "_user_id", None), vad_enabled=getattr(self, "vad_enabled", False), settings=settings, ttfb=ttfb,