Merge pull request #1588 from pipecat-ai/aleix/llm-aggregator-params

LLM aggregator params

Author: aconchillo

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `DeepgramTTSService` accepts `base_url` argument again, allowing you to
   connect to an on-prem service.
 
+- Added `LLMUserAggregatorParams` and `LLMAssistantAggregatorParams` which allow
+  you to control aggregator settings. You can now pass these arguments when
+  creating aggregator pairs with `create_context_aggregator()`.
+
 - It is now possible to disable `SoundfileMixer` when created. You can then use
   `MixerEnableFrame` to dynamically enable it when necessary.
 
@@ -38,14 +42,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `DeepgramSTTService` parameter `url` is now deprecated, use `base_url`
   instead.
 
+### Removed
+
+- Parameters `user_kwargs` and `assistant_kwargs` when creating a context
+  aggregator pair using `create_context_aggregator()` have been removed. Use
+  `user_params` and `assistant_params` instead.
+
 ### Fixed
 
 - Fixed a `TavusVideoService` issue that was causing audio choppiness.
 
 - Fixed an issue in `SmallWebRTCTransport` where an error was thrown if the
   client did not create a video transceiver.
 
-- Fixed an  issue where LLM input parameters were not working and applied correctly in `GoogleVertexLLMService`, causing 
+- Fixed an  issue where LLM input parameters were not working and applied correctly in `GoogleVertexLLMService`, causing
   unexpected behavior during inference.
 
 ## [0.0.63] - 2025-04-11

examples/foundational/22d-natural-conversation-gemini-audio.py

@@ -33,7 +33,10 @@
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMAssistantResponseAggregator,
+)
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -478,7 +481,7 @@ class LLMAggregatorBuffer(LLMAssistantResponseAggregator):
     """Buffers the output of the transcription LLM. Used by the bot output gate."""
 
     def __init__(self, **kwargs):
-        super().__init__(expect_stripped_words=False)
+        super().__init__(params=LLMAssistantAggregatorParams(expect_stripped_words=False))
         self._transcription = ""
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):

pyproject.toml

@@ -54,7 +54,7 @@ fal = [ "fal-client~=0.5.9" ]
 fireworks = []
 fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
 gladia = [ "websockets~=13.1" ]
-google = [ "google-cloud-speech~=2.31.1", "google-cloud-texttospeech~=2.25.1", "google-genai~=1.7.0", "google-generativeai~=0.8.4" ]
+google = [ "google-cloud-speech~=2.31.1", "google-cloud-texttospeech~=2.25.1", "google-genai~=1.7.0", "google-generativeai~=0.8.4", "websockets~=13.1" ]
 grok = []
 groq = [ "groq~=0.20.0" ]
 gstreamer = [ "pygobject~=3.50.0" ]

src/pipecat/processors/aggregators/llm_response.py

@@ -6,6 +6,7 @@
 
 import asyncio
 from abc import abstractmethod
+from dataclasses import dataclass
 from typing import Dict, List, Literal, Set
 
 from loguru import logger
@@ -46,6 +47,16 @@
 from pipecat.utils.time import time_now_iso8601
 
 
+@dataclass
+class LLMUserAggregatorParams:
+    aggregation_timeout: float = 1.0
+
+
+@dataclass
+class LLMAssistantAggregatorParams:
+    expect_stripped_words: bool = True
+
+
 class LLMFullResponseAggregator(FrameProcessor):
     """This is an LLM aggregator that aggregates a full LLM completion. It
     aggregates LLM text frames (tokens) received between
@@ -230,11 +241,23 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
     def __init__(
         self,
         context: OpenAILLMContext,
-        aggregation_timeout: float = 1.0,
+        *,
+        params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
         **kwargs,
     ):
         super().__init__(context=context, role="user", **kwargs)
-        self._aggregation_timeout = aggregation_timeout
+        self._params = params
+        if "aggregation_timeout" in kwargs:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter 'aggregation_timeout' is deprecated, use 'params' instead.",
+                    DeprecationWarning,
+                )
+
+            self._params.aggregation_timeout = kwargs["aggregation_timeout"]
 
         self._seen_interim_results = False
         self._user_speaking = False
@@ -357,7 +380,9 @@ async def _cancel_aggregation_task(self):
     async def _aggregation_task_handler(self):
         while True:
             try:
-                await asyncio.wait_for(self._aggregation_event.wait(), self._aggregation_timeout)
+                await asyncio.wait_for(
+                    self._aggregation_event.wait(), self._params.aggregation_timeout
+                )
                 await self._maybe_push_bot_interruption()
             except asyncio.TimeoutError:
                 if not self._user_speaking:
@@ -394,9 +419,27 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
 
     """
 
-    def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True, **kwargs):
+    def __init__(
+        self,
+        context: OpenAILLMContext,
+        *,
+        params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+        **kwargs,
+    ):
         super().__init__(context=context, role="assistant", **kwargs)
-        self._expect_stripped_words = expect_stripped_words
+        self._params = params
+
+        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.",
+                    DeprecationWarning,
+                )
+
+            self._params.expect_stripped_words = kwargs["expect_stripped_words"]
 
         self._started = 0
         self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {}
@@ -558,7 +601,7 @@ async def _handle_text(self, frame: TextFrame):
         if not self._started:
             return
 
-        if self._expect_stripped_words:
+        if self._params.expect_stripped_words:
             self._aggregation += f" {frame.text}" if self._aggregation else frame.text
         else:
             self._aggregation += frame.text
@@ -572,8 +615,14 @@ def _context_updated_task_finished(self, task: asyncio.Task):
 
 
 class LLMUserResponseAggregator(LLMUserContextAggregator):
-    def __init__(self, messages: List[dict] = [], **kwargs):
-        super().__init__(context=OpenAILLMContext(messages), **kwargs)
+    def __init__(
+        self,
+        messages: List[dict] = [],
+        *,
+        params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        **kwargs,
+    ):
+        super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)
 
     async def push_aggregation(self):
         if len(self._aggregation) > 0:
@@ -588,8 +637,14 @@ async def push_aggregation(self):
 
 
 class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
-    def __init__(self, messages: List[dict] = [], **kwargs):
-        super().__init__(context=OpenAILLMContext(messages), **kwargs)
+    def __init__(
+        self,
+        messages: List[dict] = [],
+        *,
+        params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+        **kwargs,
+    ):
+        super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)
 
     async def push_aggregation(self):
         if len(self._aggregation) > 0:

src/pipecat/services/anthropic/llm.py

@@ -11,7 +11,7 @@
 import json
 import re
 from dataclasses import dataclass
-from typing import Any, Dict, List, Mapping, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import httpx
 from loguru import logger
@@ -35,7 +35,9 @@
 )
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
     LLMAssistantContextAggregator,
+    LLMUserAggregatorParams,
     LLMUserContextAggregator,
 )
 from pipecat.processors.aggregators.openai_llm_context import (
@@ -49,10 +51,7 @@
     from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
-    logger.error(
-        "In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. "
-        + "Also, set `ANTHROPIC_API_KEY` environment variable."
-    )
+    logger.error("In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`.")
     raise Exception(f"Missing module: {e}")
 
 
@@ -120,21 +119,19 @@ def create_context_aggregator(
         self,
         context: OpenAILLMContext,
         *,
-        user_kwargs: Mapping[str, Any] = {},
-        assistant_kwargs: Mapping[str, Any] = {},
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
     ) -> AnthropicContextAggregatorPair:
         """Create an instance of AnthropicContextAggregatorPair from an
         OpenAILLMContext. Constructor keyword arguments for both the user and
         assistant aggregators can be provided.
 
         Args:
             context (OpenAILLMContext): The LLM context.
-            user_kwargs (Mapping[str, Any], optional): Additional keyword
-                arguments for the user context aggregator constructor. Defaults
-                to an empty mapping.
-            assistant_kwargs (Mapping[str, Any], optional): Additional keyword
-                arguments for the assistant context aggregator
-                constructor. Defaults to an empty mapping.
+            user_params (LLMUserAggregatorParams, optional): User aggregator
+                parameters.
+            assistant_params (LLMAssistantAggregatorParams, optional): User
+                aggregator parameters.
 
         Returns:
             AnthropicContextAggregatorPair: A pair of context aggregators, one
@@ -146,8 +143,8 @@ def create_context_aggregator(
 
         if isinstance(context, OpenAILLMContext):
             context = AnthropicLLMContext.from_openai_context(context)
-        user = AnthropicUserContextAggregator(context, **user_kwargs)
-        assistant = AnthropicAssistantContextAggregator(context, **assistant_kwargs)
+        user = AnthropicUserContextAggregator(context, params=user_params)
+        assistant = AnthropicAssistantContextAggregator(context, params=assistant_params)
         return AnthropicContextAggregatorPair(_user=user, _assistant=assistant)
 
     async def _process_context(self, context: OpenAILLMContext):

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -10,9 +10,8 @@
 import time
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Dict, List, Mapping, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
-import websockets
 from loguru import logger
 from pydantic import BaseModel, Field
 
@@ -45,6 +44,10 @@
     UserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMUserAggregatorParams,
+)
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -61,6 +64,13 @@
 from . import events
 from .audio_transcriber import AudioTranscriber
 
+try:
+    import websockets
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
+    raise Exception(f"Missing module: {e}")
+
 
 def language_to_gemini_language(language: Language) -> Optional[str]:
     """Maps a Language enum value to a Gemini Live supported language code.
@@ -871,21 +881,19 @@ def create_context_aggregator(
         self,
         context: OpenAILLMContext,
         *,
-        user_kwargs: Mapping[str, Any] = {},
-        assistant_kwargs: Mapping[str, Any] = {},
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
     ) -> GeminiMultimodalLiveContextAggregatorPair:
         """Create an instance of GeminiMultimodalLiveContextAggregatorPair from
         an OpenAILLMContext. Constructor keyword arguments for both the user and
         assistant aggregators can be provided.
 
         Args:
             context (OpenAILLMContext): The LLM context.
-            user_kwargs (Mapping[str, Any], optional): Additional keyword
-                arguments for the user context aggregator constructor. Defaults
-                to an empty mapping.
-            assistant_kwargs (Mapping[str, Any], optional): Additional keyword
-                arguments for the assistant context aggregator
-                constructor. Defaults to an empty mapping.
+            user_params (LLMUserAggregatorParams, optional): User aggregator
+                parameters.
+            assistant_params (LLMAssistantAggregatorParams, optional): User
+                aggregator parameters.
 
         Returns:
             GeminiMultimodalLiveContextAggregatorPair: A pair of context
@@ -896,11 +904,8 @@ def create_context_aggregator(
         context.set_llm_adapter(self.get_llm_adapter())
 
         GeminiMultimodalLiveContext.upgrade(context)
-        user = GeminiMultimodalLiveUserContextAggregator(context, **user_kwargs)
+        user = GeminiMultimodalLiveUserContextAggregator(context, params=user_params)
 
-        default_assistant_kwargs = {"expect_stripped_words": True}
-        default_assistant_kwargs.update(assistant_kwargs)
-        assistant = GeminiMultimodalLiveAssistantContextAggregator(
-            context, **default_assistant_kwargs
-        )
+        assistant_params.expect_stripped_words = True
+        assistant = GeminiMultimodalLiveAssistantContextAggregator(context, params=assistant_params)
         return GeminiMultimodalLiveContextAggregatorPair(_user=user, _assistant=assistant)