From c4f9171fe106d1662b4d48efa23020d8f59a4366 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Mar 2025 14:37:36 -0700 Subject: [PATCH 1/3] frames: indicate if UserStartedSpeakingFrame/UserStoppedSpeakingFrame are emulated --- CHANGELOG.md | 12 ++++++++++-- src/pipecat/frames/frames.py | 4 ++-- src/pipecat/transports/base_input.py | 4 ++-- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a93901bd1..69b1ba743 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- It is now possible to tell whether `UserStartedSpeakingFrame` or + `UserStoppedSpeakingFrame` have been generated because of emulation frames. + ## [0.0.61] - 2025-03-26 ### Added @@ -21,9 +28,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - ElevenLabs TTS services now support a sample rate of 8000. -- Added support for `instructions` in `OpenAITTSService` +- Added support for `instructions` in `OpenAITTSService`. -- Added support for `base_url` in `OpenAIImageGenService` and `OpenAITTSService` +- Added support for `base_url` in `OpenAIImageGenService` and + `OpenAITTSService`. ### Fixed diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 11cd17801..30d8622d9 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -562,14 +562,14 @@ class UserStartedSpeakingFrame(SystemFrame): """ - pass + emulated: bool = False @dataclass class UserStoppedSpeakingFrame(SystemFrame): """Emitted by the VAD to indicate that a user stopped speaking.""" - pass + emulated: bool = False @dataclass diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 971dfe066..26f386576 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -117,10 +117,10 @@ class BaseInputTransport(FrameProcessor): await self._handle_bot_interruption(frame) elif isinstance(frame, EmulateUserStartedSpeakingFrame): logger.debug("Emulating user started speaking") - await self._handle_user_interruption(UserStartedSpeakingFrame()) + await self._handle_user_interruption(UserStartedSpeakingFrame(emulated=True)) elif isinstance(frame, EmulateUserStoppedSpeakingFrame): logger.debug("Emulating user stopped speaking") - await self._handle_user_interruption(UserStoppedSpeakingFrame()) + await self._handle_user_interruption(UserStoppedSpeakingFrame(emulated=True)) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) From 750bb8858632d3de914356a0eaa3766596054d9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Mar 2025 14:38:48 -0700 Subject: [PATCH 2/3] SegmentedSTTService: ignore emulated frames --- CHANGELOG.md | 6 ++++++ src/pipecat/services/ai_services.py | 7 ++++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 69b1ba743..1cf797076 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - It is now possible to tell whether `UserStartedSpeakingFrame` or `UserStoppedSpeakingFrame` have been generated because of emulation frames. +### Fixed + +- Fixed an issue that would cause `SegmentedSTTService` based services + (e.g. `OpenAISTTService`) to try to transcribe non-spoken audio, causing + invalid transcriptions. + ## [0.0.61] - 2025-03-26 ### Added diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index a78c268dd..97aad0d40 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -1048,9 +1048,14 @@ class SegmentedSTTService(STTService): await self._handle_user_stopped_speaking(frame) async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame): + if frame.emulated: + return self._user_speaking = True async def _handle_user_stopped_speaking(self, frame: UserStoppedSpeakingFrame): + if frame.emulated: + return + self._user_speaking = False content = io.BytesIO() @@ -1068,7 +1073,7 @@ class SegmentedSTTService(STTService): self._audio_buffer.clear() async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection): - # If the user is speaking the audio buffer will keep growin. + # If the user is speaking the audio buffer will keep growing. self._audio_buffer += frame.audio # If the user is not speaking we keep just a little bit of audio. From 055a3f1c53a2612afd26405993ca3ad2d249731f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Mar 2025 14:39:12 -0700 Subject: [PATCH 3/3] LLMAssistantContextAggregator: stop emulations if the user starts speaking --- src/pipecat/processors/aggregators/llm_response.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index af8bf1a2e..e40ad266b 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -249,7 +249,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): self._waiting_for_aggregation = False async def handle_aggregation(self, aggregation: str): - self._context.add_message({"role": self.role, "content": self._aggregation}) + self._context.add_message({"role": self.role, "content": aggregation}) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -290,12 +290,14 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): async def push_aggregation(self): if len(self._aggregation) > 0: - await self.handle_aggregation(self._aggregation) + aggregation = self._aggregation # Reset the aggregation. Reset it before pushing it down, otherwise # if the tasks gets cancelled we won't be able to clear things up. self.reset() + await self.handle_aggregation(aggregation) + frame = OpenAILLMContextFrame(self._context) await self.push_frame(frame) @@ -308,10 +310,16 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): async def _cancel(self, frame: CancelFrame): await self._cancel_aggregation_task() - async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame): + async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame): self._user_speaking = True self._waiting_for_aggregation = True + # If we get a non-emulated UserStartedSpeakingFrame but we are in the + # middle of emulating VAD, let's stop emulating VAD (i.e. don't send the + # EmulateUserStoppedSpeakingFrame). + if not frame.emulated and self._emulating_vad: + self._emulating_vad = False + async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame): self._user_speaking = False # We just stopped speaking. Let's see if there's some aggregation to