Merge pull request #2714 from pipecat-ai/pk/gated-openai-llm-context-aggregator-support-univeral-context

Add support for universal `LLMContext` to `GatedOpenAILLMContextAggre…

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.gated_llm_context import GatedLLMContextAggregator
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+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 = LLMContext(statement_messages)
-        statement_context_aggregator = statement_llm.create_context_aggregator(statement_context)
+        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(
+        gated_context_aggregator = GatedLLMContextAggregator(notifier=notifier, start_open=True)
-            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 = LLMContext(messages)
-    context_aggregator = llm_main.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     # LLM + turn detection (with an extra LLM as a judge)
     llm = TurnDetectionLLM(llm_main, context_aggregator)

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):
+class GatedLLMContextAggregator(FrameProcessor):
-    """Aggregator that gates OpenAI LLM context frames until notified.
+    """Aggregator that gates LLM context frames until notified.
 
-    This aggregator captures OpenAI LLM context frames and holds them until
+    This aggregator captures LLM context frames and holds them until a notifier
-    a notifier signals that they can be released. This is useful for controlling
+    signals that they can be released. This is useful for controlling the flow
-    the flow of context frames based on external conditions or timing.
+    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)

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