From 99731ca40ae124a408277314e7c6a7cb48798836 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 23 Sep 2025 09:52:13 -0400 Subject: [PATCH] Add support for universal `LLMContext` to `GatedOpenAILLMContextAggregator`, renaming it to `GatedLLMContextAggregator` in the process --- .../foundational/22-natural-conversation.py | 17 ++++++++--------- ...nai_llm_context.py => gated_llm_context.py} | 18 +++++++++--------- .../aggregators/gated_open_ai_llm_context.py | 12 ++++++++++++ 3 files changed, 29 insertions(+), 18 deletions(-) rename src/pipecat/processors/aggregators/{gated_openai_llm_context.py => gated_llm_context.py} (81%) create mode 100644 src/pipecat/processors/aggregators/gated_open_ai_llm_context.py diff --git a/examples/foundational/22-natural-conversation.py b/examples/foundational/22-natural-conversation.py index 4907aa137..98ad23c0b 100644 --- a/examples/foundational/22-natural-conversation.py +++ b/examples/foundational/22-natural-conversation.py @@ -16,8 +16,9 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.gated_openai_llm_context import GatedOpenAILLMContextAggregator -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.gated_llm_context import GatedLLMContextAggregator +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.filters.null_filter import NullFilter from pipecat.processors.filters.wake_notifier_filter import WakeNotifierFilter from pipecat.processors.user_idle_processor import UserIdleProcessor @@ -50,8 +51,8 @@ class TurnDetectionLLM(Pipeline): }, ] - statement_context = OpenAILLMContext(statement_messages) - statement_context_aggregator = statement_llm.create_context_aggregator(statement_context) + statement_context = LLMContext(statement_messages) + statement_context_aggregator = LLMContextAggregatorPair(statement_context) # We have instructed the LLM to return 'YES' if it thinks the user # completed a sentence. So, if it's 'YES' we will return true in this @@ -72,9 +73,7 @@ class TurnDetectionLLM(Pipeline): # This processor keeps the last context and will let it through once the # notifier is woken up. We start with the gate open because we send an # initial context frame to start the conversation. - gated_context_aggregator = GatedOpenAILLMContextAggregator( - notifier=notifier, start_open=True - ) + gated_context_aggregator = GatedLLMContextAggregator(notifier=notifier, start_open=True) # Notify if the user hasn't said anything. async def user_idle_notifier(frame): @@ -147,8 +146,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = OpenAILLMContext(messages) - context_aggregator = llm_main.create_context_aggregator(context) + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair(context) # LLM + turn detection (with an extra LLM as a judge) llm = TurnDetectionLLM(llm_main, context_aggregator) diff --git a/src/pipecat/processors/aggregators/gated_openai_llm_context.py b/src/pipecat/processors/aggregators/gated_llm_context.py similarity index 81% rename from src/pipecat/processors/aggregators/gated_openai_llm_context.py rename to src/pipecat/processors/aggregators/gated_llm_context.py index 56423403d..6535bffbb 100644 --- a/src/pipecat/processors/aggregators/gated_openai_llm_context.py +++ b/src/pipecat/processors/aggregators/gated_llm_context.py @@ -4,20 +4,20 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Gated OpenAI LLM context aggregator for controlled message flow.""" +"""Gated LLM context aggregator for controlled message flow.""" -from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame +from pipecat.frames.frames import CancelFrame, EndFrame, Frame, LLMContextFrame, StartFrame from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.sync.base_notifier import BaseNotifier -class GatedOpenAILLMContextAggregator(FrameProcessor): - """Aggregator that gates OpenAI LLM context frames until notified. +class GatedLLMContextAggregator(FrameProcessor): + """Aggregator that gates LLM context frames until notified. - This aggregator captures OpenAI LLM context frames and holds them until - a notifier signals that they can be released. This is useful for controlling - the flow of context frames based on external conditions or timing. + This aggregator captures LLM context frames and holds them until a notifier + signals that they can be released. This is useful for controlling the flow + of context frames based on external conditions or timing. """ def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs): @@ -35,7 +35,7 @@ class GatedOpenAILLMContextAggregator(FrameProcessor): self._gate_task = None async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process incoming frames, gating OpenAI LLM context frames. + """Process incoming frames, gating LLM context frames. Args: frame: The frame to process. @@ -49,7 +49,7 @@ class GatedOpenAILLMContextAggregator(FrameProcessor): if isinstance(frame, (EndFrame, CancelFrame)): await self._stop() await self.push_frame(frame) - elif isinstance(frame, OpenAILLMContextFrame): + elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)): if self._start_open: self._start_open = False await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/aggregators/gated_open_ai_llm_context.py b/src/pipecat/processors/aggregators/gated_open_ai_llm_context.py new file mode 100644 index 000000000..42e38928f --- /dev/null +++ b/src/pipecat/processors/aggregators/gated_open_ai_llm_context.py @@ -0,0 +1,12 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Gated OpenAI LLM context aggregator for controlled message flow.""" + +from pipecat.processors.aggregators.gated_llm_context import GatedLLMContextAggregator + +# Alias for backward compatibility with the previous name +GatedOpenAILLMContextAggregator = GatedLLMContextAggregator