Merge branch 'main' into sarvam/stt

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_aggregator = LLMContextAggregatorPair(context)
-    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),
-  )
   ```
 
   (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_aggregator = LLMContextAggregatorPair(context)
-    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),
-  )
   ```
 
   (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
+- The development runner's `/start` endpoint now supports passing
-  `add_to_context` field indicates if this image and text should be added to the
+  `dailyRoomProperties` and `dailyMeetingTokenProperties` in the request body
-  LLM context (by the LLM assistant aggregator). The `text` field, if set, might
+  when `createDailyRoom` is true. Properties are validated against the
-  also guide the LLM or the vision service on how to analyze the image.
+  `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.

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()

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,
     )
 

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,
     )
 

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,
     )
 

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,
     )
 

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,
     )
 

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when OpenAI Realtime used with
-        # "audio" modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when OpenAI Realtime used with
-        # "audio" modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

examples/foundational/26a-gemini-live-transcription.py

@@ -92,12 +92,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             #     },
         ],
     )
-    context_aggregator = LLMContextAggregatorPair(
+    context_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     transcript = TranscriptProcessor()
 

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

examples/foundational/26c-gemini-live-video.py

@@ -75,12 +75,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             },
         ],
     )
-    context_aggregator = LLMContextAggregatorPair(
+    context_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     # Build the pipeline
     pipeline = Pipeline(

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

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_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     pipeline = Pipeline(
         [

examples/foundational/46-video-processing.py

@@ -111,12 +111,7 @@ async def run_bot(pipecat_transport):
     ]
 
     context = LLMContext(messages)
-    context_aggregator = LLMContextAggregatorPair(
+    context_aggregator = LLMContextAggregatorPair(context)
-        context,
-        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
-        #  modality (the default)
-        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
-    )
 
     # RTVI events for Pipecat client UI
     rtvi = RTVIProcessor()

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),

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

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
+        # This is the idle event. When selected frames are pushed from any
-        # idle.
+        # processor we consider the pipeline is not idle. We use an observer
-        self._idle_queue = asyncio.Queue()
+        # 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(
+                await asyncio.wait_for(self._idle_event.wait(), timeout=self._idle_timeout_secs)
-                    self._idle_queue.get(), timeout=self._idle_timeout_secs
+                self._idle_event.clear()
-                )
-
-                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()
-
             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")

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:

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

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:
+        # Make sure we really have text (spaces count, too!)
-            self._aggregation += f" {frame.text}" if self._aggregation else frame.text
+        if len(frame.text) == 0:
-        else:
+            return
-            self._aggregation += frame.text
+
+        self._aggregation.append(frame.text)
 
     def _context_updated_task_finished(self, task: asyncio.Task):
         self._context_updated_tasks.discard(task)

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
+            content = concatenate_aggregated_text(self._current_text_parts)
-            # 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()
-
             if content:
                 logger.trace(f"Emitting aggregated assistant message: {content}")
                 message = TranscriptionMessage(

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

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,

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)}")

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,

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)

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,
+                                text=request_text,
-                                add_to_context=request_frame.add_to_context
+                                append_to_context=add_to_context,
-                                if request_frame
-                                else None,
                             )
                             image_frame.transport_source = video_source
                             # Push the frame to the pipeline

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

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

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_aggregator = LLMContextAggregatorPair(context)
-            context, assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False)
-        )
 
         pipeline = Pipeline(
             [context_aggregator.user(), proc, self.mock_proc, context_aggregator.assistant()]

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):