diff --git a/CHANGELOG.md b/CHANGELOG.md index 25b57c20b..75dcc1b31 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,13 +5,16 @@ 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] +## [0.0.92] - 2025-10-31 🎃 "The Haunted Edition" 👻 ### Added - Added supprt for Sarvam Speech-to-Text service (`SarvamSTTService`) with streaming WebSocket support for `saarika` (STT) and `saaras` (STT-translate) models. +- Added a new `DeepgramHttpTTSService`, which delivers a meaningful reduction + in latency when compared to the `DeepgramTTSService`. + - Add support for `speaking_rate` input parameter in `GoogleHttpTTSService`. - Added `enable_speaker_diarization` and `enable_language_identification` to @@ -44,13 +47,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ```python context = LLMContext(messages, tools) - context_aggregator = LLMContextAggregatorPair( - context, - # This part is `OpenAIRealtimeLLMService`-specific. - # `expect_stripped_words=False` needed when OpenAI Realtime used with - # "audio" modality (the default). - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) ``` (Note that even though `OpenAIRealtimeLLMService` now supports the universal @@ -127,13 +124,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ```python context = LLMContext(messages, tools) - context_aggregator = LLMContextAggregatorPair( - context, - # This part is `GeminiLiveLLMService`-specific. - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default). - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) ``` (Note that even though `GeminiLiveLLMService` now supports the universal @@ -175,12 +166,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- `UserImageRawFrame` new fields `add_to_context` and `text`. The - `add_to_context` field indicates if this image and text should be added to the - LLM context (by the LLM assistant aggregator). The `text` field, if set, might - also guide the LLM or the vision service on how to analyze the image. +- The development runner's `/start` endpoint now supports passing + `dailyRoomProperties` and `dailyMeetingTokenProperties` in the request body + when `createDailyRoom` is true. Properties are validated against the + `DailyRoomProperties` and `DailyMeetingTokenProperties` types respectively + and passed to Daily's room and token creation APIs. -- `UserImageRequestFrame` new fiels `add_to_context` and `text`. Both fields +- `UserImageRawFrame` new fields `append_to_context` and `text`. The + `append_to_context` field indicates if this image and text should be added to + the LLM context (by the LLM assistant aggregator). The `text` field, if set, + might also guide the LLM or the vision service on how to analyze the image. + +- `UserImageRequestFrame` new fiels `append_to_context` and `text`. Both fields will be used to set the same fields on the captured `UserImageRawFrame`. - `UserImageRequestFrame` don't require function call name and ID anymore. @@ -213,6 +210,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- The `expect_stripped_words` parameter of `LLMAssistantAggregatorParams` is + ignored when used with the newer `LLMAssistantAggregator`, which now handles + word spacing automatically. + - `LLMService.request_image_frame()` is deprecated, push a `UserImageRequestFrame` instead. @@ -239,6 +240,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `PipelineTask` issue that was causing an idle timeout for frames that + were being generated but not reaching the end of the pipeline. Since the exact + point when frames are discarded is unknown, we now monitor pipeline frames + using an observer. If the observer detects frames are being generated, it will + prevent the pipeline from being considered idle. + - Fixed an issue in `HumeTTSService` that was only using Octave 2, which does not support the `description` field. Now, if a description is provided, it switches to Octave 1. diff --git a/examples/foundational/07c-interruptible-deepgram-http.py b/examples/foundational/07c-interruptible-deepgram-http.py new file mode 100644 index 000000000..c444b5638 --- /dev/null +++ b/examples/foundational/07c-interruptible-deepgram-http.py @@ -0,0 +1,132 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import LLMRunFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.deepgram.tts import DeepgramHttpTTSService +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + async with aiohttp.ClientSession() as session: + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = DeepgramHttpTTSService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + voice="aura-2-andromeda-en", + aiohttp_session=session, + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/foundational/14d-function-calling-anthropic-video.py b/examples/foundational/14d-function-calling-anthropic-video.py index b21b9fdba..c933779bb 100644 --- a/examples/foundational/14d-function-calling-anthropic-video.py +++ b/examples/foundational/14d-function-calling-anthropic-video.py @@ -53,7 +53,7 @@ async def fetch_user_image(params: FunctionCallParams): # Request a user image frame and indicate that it should be added to the # context. await params.llm.push_frame( - UserImageRequestFrame(user_id=user_id, text=question, add_to_context=True), + UserImageRequestFrame(user_id=user_id, text=question, append_to_context=True), FrameDirection.UPSTREAM, ) diff --git a/examples/foundational/14d-function-calling-aws-video.py b/examples/foundational/14d-function-calling-aws-video.py index dffee193c..392aefca7 100644 --- a/examples/foundational/14d-function-calling-aws-video.py +++ b/examples/foundational/14d-function-calling-aws-video.py @@ -53,7 +53,7 @@ async def fetch_user_image(params: FunctionCallParams): # Request a user image frame and indicate that it should be added to the # context. await params.llm.push_frame( - UserImageRequestFrame(user_id=user_id, text=question, add_to_context=True), + UserImageRequestFrame(user_id=user_id, text=question, append_to_context=True), FrameDirection.UPSTREAM, ) diff --git a/examples/foundational/14d-function-calling-gemini-flash-video.py b/examples/foundational/14d-function-calling-gemini-flash-video.py index acb977c68..c11a4de2e 100644 --- a/examples/foundational/14d-function-calling-gemini-flash-video.py +++ b/examples/foundational/14d-function-calling-gemini-flash-video.py @@ -53,7 +53,7 @@ async def fetch_user_image(params: FunctionCallParams): # Request a user image frame and indicate that it should be added to the # context. await params.llm.push_frame( - UserImageRequestFrame(user_id=user_id, text=question, add_to_context=True), + UserImageRequestFrame(user_id=user_id, text=question, append_to_context=True), FrameDirection.UPSTREAM, ) diff --git a/examples/foundational/14d-function-calling-moondream-video.py b/examples/foundational/14d-function-calling-moondream-video.py index 3a7889c00..83d6ffd66 100644 --- a/examples/foundational/14d-function-calling-moondream-video.py +++ b/examples/foundational/14d-function-calling-moondream-video.py @@ -55,7 +55,7 @@ async def fetch_user_image(params: FunctionCallParams): # image to be added to the context because we will process it with # Moondream. await params.llm.push_frame( - UserImageRequestFrame(user_id=user_id, text=question, add_to_context=False), + UserImageRequestFrame(user_id=user_id, text=question, append_to_context=False), FrameDirection.UPSTREAM, ) diff --git a/examples/foundational/14d-function-calling-openai-video.py b/examples/foundational/14d-function-calling-openai-video.py index ec6fb008b..d68ba7f7c 100644 --- a/examples/foundational/14d-function-calling-openai-video.py +++ b/examples/foundational/14d-function-calling-openai-video.py @@ -54,7 +54,7 @@ async def fetch_user_image(params: FunctionCallParams): # Request a user image frame and indicate that it should be added to the # context. await params.llm.push_frame( - UserImageRequestFrame(user_id=user_id, text=question, add_to_context=True), + UserImageRequestFrame(user_id=user_id, text=question, append_to_context=True), FrameDirection.UPSTREAM, ) diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 6907ec196..b5edc0ff2 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -187,12 +187,7 @@ Remember, your responses should be short. Just one or two sentences, usually. Re tools, ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when OpenAI Realtime used with - # "audio" modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py index 7d9cf1b4b..b56c05af9 100644 --- a/examples/foundational/19a-azure-realtime.py +++ b/examples/foundational/19a-azure-realtime.py @@ -175,12 +175,7 @@ Remember, your responses should be short. Just one or two sentences, usually. Re tools, ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when OpenAI Realtime used with - # "audio" modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/26a-gemini-live-transcription.py b/examples/foundational/26a-gemini-live-transcription.py index 9ac22a814..2aed3e650 100644 --- a/examples/foundational/26a-gemini-live-transcription.py +++ b/examples/foundational/26a-gemini-live-transcription.py @@ -92,12 +92,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # }, ], ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) transcript = TranscriptProcessor() diff --git a/examples/foundational/26b-gemini-live-function-calling.py b/examples/foundational/26b-gemini-live-function-calling.py index fd2b36ab1..19a23e798 100644 --- a/examples/foundational/26b-gemini-live-function-calling.py +++ b/examples/foundational/26b-gemini-live-function-calling.py @@ -144,12 +144,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): context = LLMContext( [{"role": "user", "content": "Say hello."}], ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/26c-gemini-live-video.py b/examples/foundational/26c-gemini-live-video.py index be036a557..3b68a650a 100644 --- a/examples/foundational/26c-gemini-live-video.py +++ b/examples/foundational/26c-gemini-live-video.py @@ -75,12 +75,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ], ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/26e-gemini-live-google-search.py b/examples/foundational/26e-gemini-live-google-search.py index e80ed4536..f5a3fd675 100644 --- a/examples/foundational/26e-gemini-live-google-search.py +++ b/examples/foundational/26e-gemini-live-google-search.py @@ -100,12 +100,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): } ], ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/26f-gemini-live-files-api.py b/examples/foundational/26f-gemini-live-files-api.py index 0091c01e6..e8b38ba6d 100644 --- a/examples/foundational/26f-gemini-live-files-api.py +++ b/examples/foundational/26f-gemini-live-files-api.py @@ -164,12 +164,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ) # Create context aggregator - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) # Build the pipeline pipeline = Pipeline( diff --git a/examples/foundational/26g-gemini-live-groundingMetadata.py b/examples/foundational/26g-gemini-live-groundingMetadata.py index df86094da..553539ab2 100644 --- a/examples/foundational/26g-gemini-live-groundingMetadata.py +++ b/examples/foundational/26g-gemini-live-groundingMetadata.py @@ -127,12 +127,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # Set up conversation context and management context = LLMContext(messages) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/26h-gemini-live-vertex-function-calling.py b/examples/foundational/26h-gemini-live-vertex-function-calling.py index 126d85ad7..4d1534829 100644 --- a/examples/foundational/26h-gemini-live-vertex-function-calling.py +++ b/examples/foundational/26h-gemini-live-vertex-function-calling.py @@ -140,12 +140,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) context = LLMContext([{"role": "user", "content": "Say hello."}]) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/26i-gemini-live-graceful-end.py b/examples/foundational/26i-gemini-live-graceful-end.py index 2865dbed4..9d3628777 100644 --- a/examples/foundational/26i-gemini-live-graceful-end.py +++ b/examples/foundational/26i-gemini-live-graceful-end.py @@ -157,12 +157,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): context = LLMContext( [{"role": "user", "content": "Say hello."}], ) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/examples/foundational/46-video-processing.py b/examples/foundational/46-video-processing.py index 5f92139bc..159da270a 100644 --- a/examples/foundational/46-video-processing.py +++ b/examples/foundational/46-video-processing.py @@ -111,12 +111,7 @@ async def run_bot(pipecat_transport): ] context = LLMContext(messages) - context_aggregator = LLMContextAggregatorPair( - context, - # `expect_stripped_words=False` needed when Gemini Live used with AUDIO - # modality (the default) - assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), - ) + context_aggregator = LLMContextAggregatorPair(context) # RTVI events for Pipecat client UI rtvi = RTVIProcessor() diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index ce0a32dd6..5c66dd75d 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -87,6 +87,7 @@ TESTS_07 = [ ("07b-interruptible-langchain.py", EVAL_SIMPLE_MATH), ("07c-interruptible-deepgram.py", EVAL_SIMPLE_MATH), ("07c-interruptible-deepgram-flux.py", EVAL_SIMPLE_MATH), + ("07c-interruptible-deepgram-http.py", EVAL_SIMPLE_MATH), ("07d-interruptible-elevenlabs.py", EVAL_SIMPLE_MATH), ("07d-interruptible-elevenlabs-http.py", EVAL_SIMPLE_MATH), ("07f-interruptible-azure.py", EVAL_SIMPLE_MATH), diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 70975d262..6577ae7a4 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1207,17 +1207,17 @@ class UserImageRequestFrame(SystemFrame): Parameters: user_id: Identifier of the user to request image from. text: An optional text associated to the image request. - add_to_context: Whether the requested image should be added to an LLM context. + append_to_context: Whether the requested image should be appended to the LLM context. video_source: Specific video source to capture from. """ user_id: str text: Optional[str] = None - add_to_context: Optional[bool] = None + append_to_context: Optional[bool] = None video_source: Optional[str] = None def __str__(self): - return f"{self.name}(user: {self.user_id}, text: {self.text}, add_to_context: {self.add_to_context}, {self.video_source})" + return f"{self.name}(user: {self.user_id}, text: {self.text}, append_to_context: {self.append_to_context}, {self.video_source})" @dataclass @@ -1292,16 +1292,16 @@ class UserImageRawFrame(InputImageRawFrame): Parameters: user_id: Identifier of the user who provided this image. text: An optional text associated to this image. - add_to_context: Whether this image should be added to an LLM context. + append_to_context: Whether the requested image should be appended to the LLM context. """ user_id: str = "" text: Optional[str] = None - add_to_context: Optional[bool] = None + append_to_context: Optional[bool] = None def __str__(self): pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.transport_source}, size: {self.size}, format: {self.format}, text: {self.text}, add_to_context: {self.add_to_context})" + return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.transport_source}, size: {self.size}, format: {self.format}, text: {self.text}, append_to_context: {self.append_to_context})" @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index a511db12a..1f9cbf63a 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -12,7 +12,6 @@ including heartbeats, idle detection, and observer integration. """ import asyncio -import time from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type from loguru import logger @@ -39,7 +38,7 @@ from pipecat.frames.frames import ( UserSpeakingFrame, ) from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData -from pipecat.observers.base_observer import BaseObserver +from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.observers.turn_tracking_observer import TurnTrackingObserver from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource @@ -57,6 +56,43 @@ IDLE_TIMEOUT_SECS = 300 CANCEL_TIMEOUT_SECS = 20.0 +class IdleFrameObserver(BaseObserver): + """Idle timeout observer. + + This observer waits for specific frames being generated in the pipeline. If + the frames are generated the given asyncio event is set. If the event is not + set it means the pipeline is probably idle. + + """ + + def __init__(self, *, idle_event: asyncio.Event, idle_timeout_frames: Tuple[Type[Frame], ...]): + """Initialize the observer. + + Args: + idle_event: The event to set if the idle timeout frames are being pushed. + idle_timeout_frames: A tuple with the frames that should set the event when received + """ + super().__init__() + self._idle_event = idle_event + self._idle_timeout_frames = idle_timeout_frames + self._processed_frames = set() + + async def on_push_frame(self, data: FramePushed): + """Callback executed when a frame is pushed in the pipeline. + + Args: + data: The frame push event data. + """ + # Skip already processed frames + if data.frame.id in self._processed_frames: + return + + self._processed_frames.add(data.frame.id) + + if isinstance(data.frame, StartFrame) or isinstance(data.frame, self._idle_timeout_frames): + self._idle_event.set() + + class PipelineParams(BaseModel): """Configuration parameters for pipeline execution. @@ -215,7 +251,6 @@ class PipelineTask(BasePipelineTask): self._conversation_id = conversation_id self._enable_tracing = enable_tracing and is_tracing_available() self._enable_turn_tracking = enable_turn_tracking - self._idle_timeout_frames = idle_timeout_frames self._idle_timeout_secs = idle_timeout_secs if self._params.observers: import warnings @@ -250,16 +285,24 @@ class PipelineTask(BasePipelineTask): # This queue is the queue used to push frames to the pipeline. self._push_queue = asyncio.Queue() self._process_push_task: Optional[asyncio.Task] = None + # This is the heartbeat queue. When a heartbeat frame is received in the # down queue we add it to the heartbeat queue for processing. self._heartbeat_queue = asyncio.Queue() self._heartbeat_push_task: Optional[asyncio.Task] = None self._heartbeat_monitor_task: Optional[asyncio.Task] = None - # This is the idle queue. When frames are received downstream they are - # put in the queue. If no frame is received the pipeline is considered - # idle. - self._idle_queue = asyncio.Queue() + + # This is the idle event. When selected frames are pushed from any + # processor we consider the pipeline is not idle. We use an observer + # which will be listening any part of the pipeline. + self._idle_event = asyncio.Event() self._idle_monitor_task: Optional[asyncio.Task] = None + if self._idle_timeout_secs: + idle_frame_observer = IdleFrameObserver( + idle_event=self._idle_event, + idle_timeout_frames=idle_timeout_frames, + ) + observers.append(idle_frame_observer) # This event is used to indicate the StartFrame has been received at the # end of the pipeline. @@ -530,7 +573,7 @@ class PipelineTask(BasePipelineTask): async def _maybe_cancel_idle_task(self): """Cancel idle monitoring task if it is running.""" - if self._idle_timeout_secs and self._idle_monitor_task: + if self._idle_monitor_task: await self._task_manager.cancel_task(self._idle_monitor_task) self._idle_monitor_task = None @@ -706,10 +749,6 @@ class PipelineTask(BasePipelineTask): processors have handled the EndFrame and therefore we can exit the task cleanly. """ - # Queue received frame to the idle queue so we can monitor idle - # pipelines. - await self._idle_queue.put(frame) - if isinstance(frame, self._reached_downstream_types): await self._call_event_handler("on_frame_reached_downstream", frame) @@ -772,33 +811,10 @@ class PipelineTask(BasePipelineTask): Note: Heartbeats are excluded from idle detection. """ running = True - last_frame_time = 0 - while running: try: - frame = await asyncio.wait_for( - self._idle_queue.get(), timeout=self._idle_timeout_secs - ) - - if isinstance(frame, StartFrame) or isinstance(frame, self._idle_timeout_frames): - # If we find a StartFrame or one of the frames that prevents a - # time out we update the time. - last_frame_time = time.time() - else: - # If we find any other frame we check if the pipeline is - # idle by checking the last time we received one of the - # valid frames. - diff_time = time.time() - last_frame_time - if diff_time >= self._idle_timeout_secs: - running = await self._idle_timeout_detected() - # Reset `last_frame_time` so we don't trigger another - # immediate idle timeout if we are not cancelling. For - # example, we might want to force the bot to say goodbye - # and then clean nicely with an `EndFrame`. - last_frame_time = time.time() - - self._idle_queue.task_done() - + await asyncio.wait_for(self._idle_event.wait(), timeout=self._idle_timeout_secs) + self._idle_event.clear() except asyncio.TimeoutError: running = await self._idle_timeout_detected() @@ -810,7 +826,7 @@ class PipelineTask(BasePipelineTask): """ # If we are cancelling, just exit the task. if self._cancelled: - return True + return False logger.warning("Idle timeout detected.") await self._call_event_handler("on_idle_timeout") diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 98ff7c91e..11aef58cb 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -129,7 +129,7 @@ class TaskObserver(BaseObserver): for proxy in self._proxies: await proxy.cleanup() - async def on_process_frame(self, data: FramePushed): + async def on_process_frame(self, data: FrameProcessed): """Queue frame data for all managed observers. Args: diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index ace7b94fd..afc091d5a 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -89,7 +89,9 @@ class LLMAssistantAggregatorParams: Parameters: expect_stripped_words: Whether to expect and handle stripped words - in text frames by adding spaces between tokens. + in text frames by adding spaces between tokens. This parameter is + ignored when used with the newer LLMAssistantAggregator, which + handles word spacing automatically. """ expect_stripped_words: bool = True diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 44c534b9b..882428a6e 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -13,6 +13,7 @@ LLM processing, and text-to-speech components in conversational AI pipelines. import asyncio import json +import warnings from abc import abstractmethod from typing import Any, Dict, List, Literal, Optional, Set @@ -65,6 +66,7 @@ from pipecat.processors.aggregators.llm_response import ( LLMUserAggregatorParams, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.string import concatenate_aggregated_text from pipecat.utils.time import time_now_iso8601 @@ -88,7 +90,7 @@ class LLMContextAggregator(FrameProcessor): self._context = context self._role = role - self._aggregation: str = "" + self._aggregation: List[str] = [] @property def messages(self) -> List[LLMContextMessage]: @@ -168,13 +170,21 @@ class LLMContextAggregator(FrameProcessor): async def reset(self): """Reset the aggregation state.""" - self._aggregation = "" + self._aggregation = [] @abstractmethod async def push_aggregation(self): """Push the current aggregation downstream.""" pass + def aggregation_string(self) -> str: + """Get the current aggregation as a string. + + Returns: + The concatenated aggregation string. + """ + return concatenate_aggregated_text(self._aggregation) + class LLMUserAggregator(LLMContextAggregator): """User LLM aggregator that processes speech-to-text transcriptions. @@ -212,8 +222,6 @@ class LLMUserAggregator(LLMContextAggregator): self._turn_params: Optional[SmartTurnParams] = None if "aggregation_timeout" in kwargs: - import warnings - with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( @@ -307,7 +315,7 @@ class LLMUserAggregator(LLMContextAggregator): async def _process_aggregation(self): """Process the current aggregation and push it downstream.""" - aggregation = self._aggregation + aggregation = self.aggregation_string() await self.reset() self._context.add_message({"role": self.role, "content": aggregation}) frame = LLMContextFrame(self._context) @@ -355,7 +363,7 @@ class LLMUserAggregator(LLMContextAggregator): """ async def should_interrupt(strategy: BaseInterruptionStrategy): - await strategy.append_text(self._aggregation) + await strategy.append_text(self.aggregation_string()) return await strategy.should_interrupt() return any([await should_interrupt(s) for s in self._interruption_strategies]) @@ -425,7 +433,7 @@ class LLMUserAggregator(LLMContextAggregator): if not text.strip(): return - self._aggregation += f" {text}" if self._aggregation else text + self._aggregation.append(text) # We just got a final result, so let's reset interim results. self._seen_interim_results = False # Reset aggregation timer. @@ -550,23 +558,31 @@ class LLMAssistantAggregator(LLMContextAggregator): Args: context: The OpenAI LLM context for conversation storage. params: Configuration parameters for aggregation behavior. - **kwargs: Additional arguments. Supports deprecated 'expect_stripped_words'. + **kwargs: Additional arguments. """ super().__init__(context=context, role="assistant", **kwargs) self._params = params or LLMAssistantAggregatorParams() if "expect_stripped_words" in kwargs: - import warnings - with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( - "Parameter 'expect_stripped_words' is deprecated, use 'params' instead.", + "Parameter 'expect_stripped_words' is deprecated. " + "LLMAssistantAggregator now handles word spacing automatically.", DeprecationWarning, ) self._params.expect_stripped_words = kwargs["expect_stripped_words"] + if params and not params.expect_stripped_words: + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "params.expect_stripped_words is deprecated. " + "LLMAssistantAggregator now handles word spacing automatically.", + DeprecationWarning, + ) + self._started = 0 self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} self._context_updated_tasks: Set[asyncio.Task] = set() @@ -629,7 +645,7 @@ class LLMAssistantAggregator(LLMContextAggregator): if not self._aggregation: return - aggregation = self._aggregation.strip() + aggregation = self.aggregation_string() await self.reset() if aggregation: @@ -767,10 +783,10 @@ class LLMAssistantAggregator(LLMContextAggregator): message["content"] = result async def _handle_user_image_frame(self, frame: UserImageRawFrame): - if not frame.add_to_context: + if not frame.append_to_context: return - logger.debug(f"{self} Adding UserImageRawFrame to LLM context (size: {frame.size})") + logger.debug(f"{self} Appending UserImageRawFrame to LLM context (size: {frame.size})") self._context.add_image_frame_message( format=frame.format, @@ -793,10 +809,11 @@ class LLMAssistantAggregator(LLMContextAggregator): if not self._started: return - if self._params.expect_stripped_words: - self._aggregation += f" {frame.text}" if self._aggregation else frame.text - else: - self._aggregation += frame.text + # Make sure we really have text (spaces count, too!) + if len(frame.text) == 0: + return + + self._aggregation.append(frame.text) def _context_updated_task_finished(self, task: asyncio.Task): self._context_updated_tasks.discard(task) diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py index 5900ffc7f..13b2bb97f 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/src/pipecat/processors/transcript_processor.py @@ -26,6 +26,7 @@ from pipecat.frames.frames import ( TTSTextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.string import concatenate_aggregated_text from pipecat.utils.time import time_now_iso8601 @@ -140,29 +141,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): Result: "Hello there how are you" """ if self._current_text_parts and self._aggregation_start_time: - # Check specifically for space characters, previously isspace() was used - # but that includes all whitespace characters (e.g. \n), not just spaces. - has_leading_spaces = any( - part and part[0] == " " for part in self._current_text_parts[1:] - ) - has_trailing_spaces = any( - part and part[-1] == " " for part in self._current_text_parts[:-1] - ) - - # If there are embedded spaces in the fragments, use direct concatenation - contains_spacing_between_fragments = has_leading_spaces or has_trailing_spaces - - # Apply corresponding joining method - if contains_spacing_between_fragments: - # Fragments already have spacing - just concatenate - content = "".join(self._current_text_parts) - else: - # Word-by-word fragments - join with spaces - content = " ".join(self._current_text_parts) - - # Clean up any excessive whitespace - content = content.strip() - + content = concatenate_aggregated_text(self._current_text_parts) if content: logger.trace(f"Emitting aggregated assistant message: {content}") message = TranscriptionMessage( diff --git a/src/pipecat/runner/daily.py b/src/pipecat/runner/daily.py index 568d54381..87f6829bc 100644 --- a/src/pipecat/runner/daily.py +++ b/src/pipecat/runner/daily.py @@ -44,6 +44,8 @@ from loguru import logger from pydantic import BaseModel from pipecat.transports.daily.utils import ( + DailyMeetingTokenParams, + DailyMeetingTokenProperties, DailyRESTHelper, DailyRoomParams, DailyRoomProperties, @@ -84,6 +86,7 @@ async def configure( sip_num_endpoints: Optional[int] = 1, sip_codecs: Optional[Dict[str, List[str]]] = None, room_properties: Optional[DailyRoomProperties] = None, + token_properties: Optional["DailyMeetingTokenProperties"] = None, ) -> DailyRoomConfig: """Configure Daily room URL and token with optional SIP capabilities. @@ -106,6 +109,9 @@ async def configure( individual parameters. When provided, this overrides room_exp_duration and SIP-related parameters. If not provided, properties are built from the individual parameters as before. + token_properties: Optional DailyMeetingTokenProperties to customize the meeting + token. When provided, these properties are passed to the token creation API. + Note that room_name, exp, and is_owner will be set automatically. Returns: DailyRoomConfig: Object with room_url, token, and optional sip_endpoint. @@ -179,7 +185,10 @@ async def configure( # Create token and return standard format expiry_time: float = token_exp_duration * 60 * 60 - token = await daily_rest_helper.get_token(room_url, expiry_time) + token_params = None + if token_properties: + token_params = DailyMeetingTokenParams(properties=token_properties) + token = await daily_rest_helper.get_token(room_url, expiry_time, params=token_params) return DailyRoomConfig(room_url=room_url, token=token) # Create a new room @@ -221,7 +230,12 @@ async def configure( # Create meeting token token_expiry_seconds = token_exp_duration * 60 * 60 - token = await daily_rest_helper.get_token(room_url, token_expiry_seconds) + token_params = None + if token_properties: + token_params = DailyMeetingTokenParams(properties=token_properties) + token = await daily_rest_helper.get_token( + room_url, token_expiry_seconds, params=token_params + ) if sip_enabled: # Return SIP configuration object diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 28ca81bb9..897d40fdc 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -555,6 +555,7 @@ def _setup_daily_routes(app: FastAPI): { "createDailyRoom": true, "dailyRoomProperties": { "start_video_off": true }, + "dailyMeetingTokenProperties": { "is_owner": true, "user_name": "Bot" }, "body": { "custom_data": "value" } } """ @@ -570,6 +571,8 @@ def _setup_daily_routes(app: FastAPI): create_daily_room = request_data.get("createDailyRoom", False) body = request_data.get("body", {}) + daily_room_properties_dict = request_data.get("dailyRoomProperties", None) + daily_token_properties_dict = request_data.get("dailyMeetingTokenProperties", None) bot_module = _get_bot_module() @@ -584,9 +587,37 @@ def _setup_daily_routes(app: FastAPI): import aiohttp from pipecat.runner.daily import configure + from pipecat.transports.daily.utils import ( + DailyMeetingTokenProperties, + DailyRoomProperties, + ) async with aiohttp.ClientSession() as session: - room_url, token = await configure(session) + # Parse dailyRoomProperties if provided + room_properties = None + if daily_room_properties_dict: + try: + room_properties = DailyRoomProperties(**daily_room_properties_dict) + logger.debug(f"Using custom room properties: {room_properties}") + except Exception as e: + logger.error(f"Failed to parse dailyRoomProperties: {e}") + # Continue without custom properties + + # Parse dailyMeetingTokenProperties if provided + token_properties = None + if daily_token_properties_dict: + try: + token_properties = DailyMeetingTokenProperties( + **daily_token_properties_dict + ) + logger.debug(f"Using custom token properties: {token_properties}") + except Exception as e: + logger.error(f"Failed to parse dailyMeetingTokenProperties: {e}") + # Continue without custom properties + + room_url, token = await configure( + session, room_properties=room_properties, token_properties=token_properties + ) runner_args = DailyRunnerArguments(room_url=room_url, token=token, body=body) result = { "dailyRoom": room_url, diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 5819e4123..f3869c0ba 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -12,6 +12,7 @@ for generating speech from text using various voice models. from typing import AsyncGenerator, Optional +import aiohttp from loguru import logger from pipecat.frames.frames import ( @@ -117,3 +118,114 @@ class DeepgramTTSService(TTSService): except Exception as e: logger.exception(f"{self} exception: {e}") yield ErrorFrame(f"Error getting audio: {str(e)}") + + +class DeepgramHttpTTSService(TTSService): + """Deepgram HTTP text-to-speech service. + + Provides text-to-speech synthesis using Deepgram's HTTP TTS API. + Supports various voice models and audio encoding formats with + configurable sample rates and quality settings. + """ + + def __init__( + self, + *, + api_key: str, + voice: str = "aura-2-helena-en", + aiohttp_session: aiohttp.ClientSession, + base_url: str = "https://api.deepgram.com", + sample_rate: Optional[int] = None, + encoding: str = "linear16", + **kwargs, + ): + """Initialize the Deepgram TTS service. + + Args: + api_key: Deepgram API key for authentication. + voice: Voice model to use for synthesis. Defaults to "aura-2-helena-en". + aiohttp_session: Shared aiohttp session for HTTP requests with connection pooling. + base_url: Custom base URL for Deepgram API. Defaults to "https://api.deepgram.com". + sample_rate: Audio sample rate in Hz. If None, uses service default. + encoding: Audio encoding format. Defaults to "linear16". + **kwargs: Additional arguments passed to parent TTSService class. + """ + super().__init__(sample_rate=sample_rate, **kwargs) + + self._api_key = api_key + self._session = aiohttp_session + self._base_url = base_url + self._settings = { + "encoding": encoding, + } + self.set_voice(voice) + + def can_generate_metrics(self) -> bool: + """Check if the service can generate metrics. + + Returns: + True, as Deepgram TTS service supports metrics generation. + """ + return True + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Deepgram's TTS API. + + Args: + text: The text to synthesize into speech. + + Yields: + Frame: Audio frames containing the synthesized speech, plus start/stop frames. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + # Build URL with parameters + url = f"{self._base_url}/v1/speak" + + headers = {"Authorization": f"Token {self._api_key}", "Content-Type": "application/json"} + + params = { + "model": self._voice_id, + "encoding": self._settings["encoding"], + "sample_rate": self.sample_rate, + "container": "none", + } + + payload = { + "text": text, + } + + try: + await self.start_ttfb_metrics() + + async with self._session.post( + url, headers=headers, json=payload, params=params + ) as response: + if response.status != 200: + error_text = await response.text() + raise Exception(f"HTTP {response.status}: {error_text}") + + await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() + + CHUNK_SIZE = self.chunk_size + + first_chunk = True + async for chunk in response.content.iter_chunked(CHUNK_SIZE): + if first_chunk: + await self.stop_ttfb_metrics() + first_chunk = False + + if chunk: + yield TTSAudioRawFrame( + audio=chunk, + sample_rate=self.sample_rate, + num_channels=1, + ) + + yield TTSStoppedFrame() + + except Exception as e: + logger.exception(f"{self} exception: {e}") + yield ErrorFrame(f"Error getting audio: {str(e)}") diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index c92eb6309..6ccce4b31 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -8,7 +8,7 @@ import asyncio from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Tuple +from typing import Awaitable, Callable, List, Optional, Sequence, Tuple from pipecat.frames.frames import ( EndFrame, diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index c6dd06986..ce97eb4dc 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -1843,7 +1843,7 @@ class DailyInputTransport(BaseInputTransport): size=(video_frame.width, video_frame.height), format=video_frame.color_format, text=request_frame.text if request_frame else None, - add_to_context=request_frame.add_to_context if request_frame else None, + append_to_context=request_frame.append_to_context if request_frame else None, ) frame.transport_source = video_source await self.push_video_frame(frame) diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 4bd18c9ef..6c2854e9a 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -661,6 +661,8 @@ class SmallWebRTCInputTransport(BaseInputTransport): # UserImageRawFrame. Use a shallow copy so we can remove # elements. for request_frame in self._image_requests[:]: + request_text = request_frame.text if request_frame else None + add_to_context = request_frame.append_to_context if request_frame else None if request_frame.video_source == video_source: # Create UserImageRawFrame using the current video frame image_frame = UserImageRawFrame( @@ -668,10 +670,8 @@ class SmallWebRTCInputTransport(BaseInputTransport): image=video_frame.image, size=video_frame.size, format=video_frame.format, - text=request_frame.text if request_frame else None, - add_to_context=request_frame.add_to_context - if request_frame - else None, + text=request_text, + append_to_context=add_to_context, ) image_frame.transport_source = video_source # Push the frame to the pipeline diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index 11964d23f..25ce6afd5 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -18,7 +18,7 @@ Dependencies: """ import re -from typing import FrozenSet, Optional, Sequence, Tuple +from typing import FrozenSet, List, Optional, Sequence, Tuple import nltk from loguru import logger @@ -196,3 +196,40 @@ def parse_start_end_tags( return (None, len(text)) return (None, current_tag_index) + + +def concatenate_aggregated_text(text_parts: List[str]) -> str: + """Concatenate a list of text parts into a single string. + + This function joins the provided list of text parts into a single string, + taking into account whether or not the parts already contain spacing. + + This function is useful for aggregating text segments received from LLMs or + transcription services. + + Args: + text_parts: A list of strings representing parts of text to concatenate. + + Returns: + A single concatenated string. + """ + # Check specifically for space characters, previously isspace() was used + # but that includes all whitespace characters (e.g. \n), not just spaces. + has_leading_spaces = any(part and part[0] == " " for part in text_parts[1:]) + has_trailing_spaces = any(part and part[-1] == " " for part in text_parts[:-1]) + + # If there are embedded spaces in the fragments, use direct concatenation + contains_spacing_between_fragments = has_leading_spaces or has_trailing_spaces + + # Apply corresponding joining method + if contains_spacing_between_fragments: + # Fragments already have spacing - just concatenate + result = "".join(text_parts) + else: + # Word-by-word fragments - join with spaces + result = " ".join(text_parts) + + # Clean up any excessive whitespace + result = result.strip() + + return result diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index 77b6acc87..6196032a3 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -6,7 +6,7 @@ import json import unittest -from typing import Any +from typing import Any, Optional from pipecat.audio.interruptions.min_words_interruption_strategy import MinWordsInterruptionStrategy from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams @@ -22,6 +22,8 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, InterruptionFrame, InterruptionTaskFrame, + LLMContextAssistantTimestampFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, OpenAILLMContextAssistantTimestampFrame, @@ -38,6 +40,7 @@ from pipecat.processors.aggregators.llm_response import ( LLMUserAggregatorParams, LLMUserContextAggregator, ) +from pipecat.processors.aggregators.llm_response_universal import LLMAssistantAggregator from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -586,11 +589,16 @@ class BaseTestUserContextAggregator: assert context_processor.context_received -class BaseTestAssistantContextAggreagator: +class BaseTestAssistantContextAggregator: CONTEXT_CLASS = None # To be set in subclasses AGGREGATOR_CLASS = None # To be set in subclasses EXPECTED_CONTEXT_FRAMES = None # To be set in subclasses + def create_assistant_aggregator_params( + self, **kwargs + ) -> Optional[LLMAssistantAggregatorParams]: + return LLMAssistantAggregatorParams(**kwargs) + def check_message_content(self, context: OpenAILLMContext, index: int, content: str): assert context.messages[index]["content"] == content @@ -641,7 +649,7 @@ class BaseTestAssistantContextAggreagator: context = self.CONTEXT_CLASS() aggregator = self.AGGREGATOR_CLASS( - context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + context, params=self.create_assistant_aggregator_params(expect_stripped_words=False) ) frames_to_send = [ LLMFullResponseStartFrame(), @@ -687,7 +695,7 @@ class BaseTestAssistantContextAggreagator: context = self.CONTEXT_CLASS() aggregator = self.AGGREGATOR_CLASS( - context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + context, params=self.create_assistant_aggregator_params(expect_stripped_words=False) ) frames_to_send = [ LLMFullResponseStartFrame(), @@ -714,7 +722,7 @@ class BaseTestAssistantContextAggreagator: context = self.CONTEXT_CLASS() aggregator = self.AGGREGATOR_CLASS( - context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + context, params=self.create_assistant_aggregator_params(expect_stripped_words=False) ) frames_to_send = [ LLMFullResponseStartFrame(), @@ -838,7 +846,7 @@ class TestAnthropicUserContextAggregator( class TestAnthropicAssistantContextAggregator( - BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase + BaseTestAssistantContextAggregator, unittest.IsolatedAsyncioTestCase ): CONTEXT_CLASS = AnthropicLLMContext AGGREGATOR_CLASS = AnthropicAssistantContextAggregator @@ -873,7 +881,7 @@ class TestAWSBedrockUserContextAggregator( class TestAWSBedrockAssistantContextAggregator( - BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase + BaseTestAssistantContextAggregator, unittest.IsolatedAsyncioTestCase ): CONTEXT_CLASS = AWSBedrockLLMContext AGGREGATOR_CLASS = AWSBedrockAssistantContextAggregator @@ -914,7 +922,7 @@ class TestGoogleUserContextAggregator( class TestGoogleAssistantContextAggregator( - BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase + BaseTestAssistantContextAggregator, unittest.IsolatedAsyncioTestCase ): CONTEXT_CLASS = GoogleLLMContext AGGREGATOR_CLASS = GoogleAssistantContextAggregator @@ -948,8 +956,27 @@ class TestOpenAIUserContextAggregator( class TestOpenAIAssistantContextAggregator( - BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase + BaseTestAssistantContextAggregator, unittest.IsolatedAsyncioTestCase ): CONTEXT_CLASS = OpenAILLMContext AGGREGATOR_CLASS = OpenAIAssistantContextAggregator EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame] + + +# +# Universal +# +class TestLLMAssistantAggregator( + BaseTestAssistantContextAggregator, unittest.IsolatedAsyncioTestCase +): + CONTEXT_CLASS = OpenAILLMContext + AGGREGATOR_CLASS = LLMAssistantAggregator + EXPECTED_CONTEXT_FRAMES = [LLMContextFrame, LLMContextAssistantTimestampFrame] + + # Override to remove 'expect_stripped_words' parameter, which is deprecated + # for LLMAssistantAggregator + def create_assistant_aggregator_params( + self, **kwargs + ) -> Optional[LLMAssistantAggregatorParams]: + kwargs.pop("expect_stripped_words", None) + return LLMAssistantAggregatorParams(**kwargs) if kwargs else None diff --git a/tests/test_langchain.py b/tests/test_langchain.py index dd7f9ccef..4e197b2aa 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -65,9 +65,7 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): self.mock_proc = self.MockProcessor("token_collector") context = LLMContext() - context_aggregator = LLMContextAggregatorPair( - context, assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False) - ) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [context_aggregator.user(), proc, self.mock_proc, context_aggregator.assistant()] diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 7d9ebec6f..3c7f50453 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -24,6 +24,7 @@ from pipecat.pipeline.base_task import PipelineTaskParams from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.filters.frame_filter import FrameFilter from pipecat.processors.filters.identity_filter import IdentityFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import HeartbeatsObserver, run_test @@ -383,6 +384,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): idle_timeout_secs = 0.2 sleep_time_secs = idle_timeout_secs / 2 + # Use the identify filter so the frames just reach the end of the pipeline. identity = IdentityFilter() pipeline = Pipeline([identity]) task = PipelineTask( @@ -392,6 +394,12 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): ) async def delayed_frames(): + """Sending multiple text frames. + + The total amount of elapsed time in this function should be greater + than the task idle timeout. If an idle timeout event is triggered it + means we haven't detected that the TextFrames have been pushed. + """ await asyncio.sleep(sleep_time_secs) await task.queue_frame(TextFrame("Hello Pipecat!")) await asyncio.sleep(sleep_time_secs) @@ -415,6 +423,51 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): # Wait for the pending tasks to complete. await asyncio.gather(*pending) + async def test_idle_task_swallowed_frames(self): + idle_timeout_secs = 0.2 + sleep_time_secs = idle_timeout_secs / 2 + + # Block all frames (except system frames). Here, we are testing that + # generated frames don't trigger an idle timeout (they don't need to + # reach the end of the pipeline). + filter = FrameFilter(types=()) + pipeline = Pipeline([filter]) + task = PipelineTask( + pipeline, + idle_timeout_secs=idle_timeout_secs, + idle_timeout_frames=(TextFrame,), + ) + + start_time = time.time() + + async def delayed_frames(): + """Sending multiple text frames. + + The total amount of elapsed time in this function should be greater + than the task idle timeout. If an idle timeout event is triggered it + means we haven't detected that the TextFrames have been pushed. + """ + await asyncio.sleep(sleep_time_secs) + await task.queue_frame(TextFrame("Hello Pipecat!")) + await asyncio.sleep(sleep_time_secs) + await task.queue_frame(TextFrame("Hello Pipecat!")) + await asyncio.sleep(sleep_time_secs) + await task.queue_frame(TextFrame("Hello Pipecat!")) + + tasks = [ + asyncio.create_task(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + asyncio.create_task(delayed_frames()), + ] + + _, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + + diff_time = time.time() - start_time + + self.assertGreater(diff_time, sleep_time_secs * 3) + + # Wait for the pending tasks to complete. + await asyncio.gather(*pending) + async def test_task_cancel_timeout(self): class CancelFilter(FrameProcessor): def __init__(self, **kwargs):