examples: use OpenAILLMContext in all the examples

CHANGELOG.md

@@ -33,6 +33,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Renamed `OpenAILLMServiceRealtimeBeta` to `OpenAIRealtimeBetaLLMService` to
   match other services.
 
+### Deprecated
+
+- `LLMUserResponseAggregator` and `LLMAssistantResponseAggregator` are
+  mostly deprecated, use `OpenAILLMContext` instead.
+
 - The `vad` package is now deprecated and `audio.vad` should be used
   instead. The `avd` package will get removed in a future release.
 

examples/canonical-metrics/bot.py

@@ -19,10 +19,7 @@ from pipecat.frames.frames import EndFrame, LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.services.canonical import CanonicalMetricsService
 from pipecat.services.elevenlabs import ElevenLabsTTSService
@@ -92,8 +89,8 @@ async def main():
             },
         ]
 
-        user_response = LLMUserResponseAggregator()
+        context = OpenAILLMContext(messages)
-        assistant_response = LLMAssistantResponseAggregator()
+        context_aggregator = llm.create_context_aggregator(context)
 
         """
         CanonicalMetrics uses AudioBufferProcessor under the hood to buffer the audio. On
@@ -113,13 +110,13 @@ async def main():
         pipeline = Pipeline(
             [
                 transport.input(),  # microphone
-                user_response,
+                context_aggregator.user(),
                 llm,
                 tts,
                 transport.output(),
                 audio_buffer_processor,  # captures audio into a buffer
                 canonical,  # uploads audio buffer to Canonical AI for metrics
-                assistant_response,
+                context_aggregator.assistant(),
             ]
         )
 

examples/chatbot-audio-recording/bot.py

@@ -18,10 +18,7 @@ from pipecat.frames.frames import EndFrame, LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
@@ -90,19 +87,19 @@ async def main():
             },
         ]
 
-        user_response = LLMUserResponseAggregator()
+        context = OpenAILLMContext(messages)
-        assistant_response = LLMAssistantResponseAggregator()
+        context_aggregator = llm.create_context_aggregator(context)
 
         audiobuffer = AudioBufferProcessor()
         pipeline = Pipeline(
             [
                 transport.input(),  # microphone
-                user_response,
+                context_aggregator.user(),
                 llm,
                 tts,
                 transport.output(),
                 audiobuffer,  # used to buffer the audio in the pipeline
-                assistant_response,
+                context_aggregator.assistant(),
             ]
         )
 

examples/deployment/flyio-example/bot.py

@@ -7,11 +7,8 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.frames.frames import LLMMessagesFrame, EndFrame
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -60,17 +57,17 @@ async def main(room_url: str, token: str):
         },
     ]
 
-    tma_in = LLMUserResponseAggregator(messages)
+    context = OpenAILLMContext(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
+    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            tma_in,
+            context_aggregator.user(),
             llm,
             tts,
             transport.output(),
-            tma_out,
+            context_aggregator.assistant(),
         ]
     )
 

examples/dialin-chatbot/bot_daily.py

@@ -7,11 +7,8 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.frames.frames import LLMMessagesFrame, EndFrame
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyDialinSettings
@@ -66,17 +63,17 @@ async def main(room_url: str, token: str, callId: str, callDomain: str):
         },
     ]
 
-    tma_in = LLMUserResponseAggregator(messages)
+    context = OpenAILLMContext(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
+    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            tma_in,
+            context_aggregator.user(),
             llm,
             tts,
             transport.output(),
-            tma_out,
+            context_aggregator.assistant(),
         ]
     )
 

examples/dialin-chatbot/bot_twilio.py

@@ -7,11 +7,8 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.frames.frames import LLMMessagesFrame, EndFrame
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -69,17 +66,17 @@ async def main(room_url: str, token: str, callId: str, sipUri: str):
         },
     ]
 
-    tma_in = LLMUserResponseAggregator(messages)
+    context = OpenAILLMContext(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
+    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            tma_in,
+            context_aggregator.user(),
             llm,
             tts,
             transport.output(),
-            tma_out,
+            context_aggregator.assistant(),
         ]
     )
 

examples/foundational/06-listen-and-respond.py

@@ -20,10 +20,7 @@ from pipecat.metrics.metrics import (
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
@@ -92,18 +89,19 @@ async def main():
                 "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.",
             },
         ]
-        tma_in = LLMUserResponseAggregator(messages)
+
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                tma_in,
+                context_aggregator.user(),
                 llm,
                 tts,
                 ml,
                 transport.output(),
-                tma_out,
+                context_aggregator.assistant(),
             ]
         )
 

examples/foundational/06a-image-sync.py

@@ -16,10 +16,7 @@ from pipecat.frames.frames import Frame, OutputImageRawFrame, SystemFrame, TextF
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaHttpTTSService
 from pipecat.services.openai import OpenAILLMService
@@ -105,8 +102,8 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         image_sync_aggregator = ImageSyncAggregator(
             os.path.join(os.path.dirname(__file__), "assets", "speaking.png"),
@@ -117,11 +114,11 @@ async def main():
             [
                 transport.input(),
                 image_sync_aggregator,
-                tma_in,
+                context_aggregator.user(),
                 llm,
                 tts,
                 transport.output(),
-                tma_out,
+                context_aggregator.assistant(),
             ]
         )
 

examples/foundational/07-interruptible-vad.py

@@ -13,11 +13,8 @@ from pipecat.frames.frames import LLMMessagesFrame
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.processors.audio.vad.silero import SileroVAD
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -65,18 +62,18 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
-                transport.input(),  # Transport user input
+                transport.input(),
                 vad,
-                tma_in,  # User responses
+                context_aggregator.user(),
-                llm,  # LLM
+                llm,
-                tts,  # TTS
+                tts,
-                transport.output(),  # Transport bot output
+                transport.output(),
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),
             ]
         )
 

examples/foundational/07-interruptible.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -64,17 +61,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07c-interruptible-deepgram.py

@@ -18,10 +18,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -61,18 +58,18 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07d-interruptible-elevenlabs.py

@@ -11,6 +11,7 @@ import sys
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from runner import configure
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -18,10 +19,6 @@ from pipecat.frames.frames import LLMMessagesFrame
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -62,17 +59,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07e-interruptible-playht.py

@@ -18,10 +18,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.playht import PlayHTTTSService
 from pipecat.transcriptions.language import Language
@@ -66,17 +63,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07f-interruptible-azure.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.azure import AzureLLMService, AzureSTTService, AzureTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -74,18 +71,18 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07g-interruptible-openai-tts.py

@@ -11,6 +11,7 @@ import sys
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from runner import configure
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -18,10 +19,6 @@ from pipecat.frames.frames import LLMMessagesFrame
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.openai import OpenAILLMService, OpenAITTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -59,17 +56,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07h-interruptible-openpipe.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openpipe import OpenPipeLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -70,17 +67,18 @@ async def main():
                 "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.",
             },
         ]
-        tma_in = LLMUserResponseAggregator(messages)
+
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07i-interruptible-xtts.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.xtts import XTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -66,17 +63,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07j-interruptible-gladia.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.gladia import GladiaSTTService
 from pipecat.services.openai import OpenAILLMService
@@ -69,18 +66,18 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07k-interruptible-lmnt.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.lmnt import LmntTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -62,17 +59,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),  # User respones
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07l-interruptible-together.py

@@ -52,7 +52,7 @@ async def main():
 
         llm = TogetherLLMService(
             api_key=os.getenv("TOGETHER_API_KEY"),
-            model=os.getenv("TOGETHER_MODEL"),
+            model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
             params=TogetherLLMService.InputParams(
                 temperature=1.0,
                 top_p=0.9,

examples/foundational/07m-interruptible-aws.py

@@ -18,10 +18,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.aws import AWSTTSService
 from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.openai import OpenAILLMService
@@ -69,18 +66,18 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/07n-interruptible-google.py

@@ -18,10 +18,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.google import GoogleTTSService
 from pipecat.services.openai import OpenAILLMService
@@ -66,18 +63,18 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                tma_in,  # User responses
+                context_aggregator.user(),  # User respones
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/10-wake-phrase.py

@@ -10,14 +10,11 @@ import os
 import sys
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.processors.filters.wake_check_filter import WakeCheckFilter
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
+from pipecat.processors.filters.wake_check_filter import WakeCheckFilter
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -65,18 +62,19 @@ async def main():
         ]
 
         hey_robot_filter = WakeCheckFilter(["hey robot", "hey, robot"])
-        tma_in = LLMUserResponseAggregator(messages)
+
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 hey_robot_filter,  # Filter out speech not directed at the robot
-                tma_in,  # User responses
+                context_aggregator.user(),  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses
             ]
         )
 

examples/foundational/11-sound-effects.py

@@ -20,10 +20,7 @@ from pipecat.frames.frames import (
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMUserResponseAggregator,
-    LLMAssistantResponseAggregator,
-)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.logger import FrameLogger
 from pipecat.services.cartesia import CartesiaHttpTTSService
@@ -113,8 +110,8 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
         out_sound = OutboundSoundEffectWrapper()
         in_sound = InboundSoundEffectWrapper()
         fl = FrameLogger("LLM Out")
@@ -123,7 +120,7 @@ async def main():
         pipeline = Pipeline(
             [
                 transport.input(),
-                tma_in,
+                context_aggregator.user(),
                 in_sound,
                 fl2,
                 llm,
@@ -131,7 +128,7 @@ async def main():
                 tts,
                 out_sound,
                 transport.output(),
-                tma_out,
+                context_aggregator.assistant(),
             ]
         )
 

examples/foundational/16-gpu-container-local-bot.py

@@ -18,10 +18,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.deepgram import DeepgramTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import (
@@ -75,17 +72,17 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                tma_in,  # User responses
+                context_aggregator.user(),
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),
             ]
         )
 
@@ -123,7 +120,7 @@ async def main():
                         )
                     )
                     # And push to the pipeline for the Daily transport.output to send
-                    await tma_in.push_frame(
+                    await task.queue_frame(
                         DailyTransportMessageFrame(
                             message={"latency-pong-pipeline-delivery": {"ts": ts}},
                             participant_id=sender,

examples/foundational/17-detect-user-idle.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.processors.user_idle_processor import UserIdleProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
@@ -65,8 +62,8 @@ async def main():
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         async def user_idle_callback(user_idle: UserIdleProcessor):
             messages.append(
@@ -83,11 +80,11 @@ async def main():
             [
                 transport.input(),  # Transport user input
                 user_idle,  # Idle user check-in
-                tma_in,  # User responses
+                context_aggregator.user(),
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
+                context_aggregator.assistant(),
             ]
         )
 

examples/moondream-chatbot/bot.py

@@ -28,7 +28,7 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregator
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.aggregators.sentence import SentenceAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -182,17 +182,19 @@ async def main():
             },
         ]
 
-        ura = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                ura,
+                context_aggregator.user(),
                 llm,
                 ParallelPipeline([sa, ir, va, moondream], [tf, imgf]),
                 tts,
                 ta,
                 transport.output(),
+                context_aggregator.assistant(),
             ]
         )
 

examples/simple-chatbot/bot.py

@@ -15,10 +15,6 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
 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,
-    LLMUserResponseAggregator,
-)
 from pipecat.frames.frames import (
     OutputImageRawFrame,
     SpriteFrame,
@@ -27,6 +23,7 @@ from pipecat.frames.frames import (
     TTSAudioRawFrame,
     TTSStoppedFrame,
 )
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
@@ -143,20 +140,20 @@ async def main():
             },
         ]
 
-        user_response = LLMUserResponseAggregator()
+        context = OpenAILLMContext(messages)
-        assistant_response = LLMAssistantResponseAggregator()
+        context_aggregator = llm.create_context_aggregator(context)
 
         ta = TalkingAnimation()
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                user_response,
+                context_aggregator.user(),
                 llm,
                 tts,
                 ta,
                 transport.output(),
-                assistant_response,
+                context_aggregator.assistant(),
             ]
         )
 

examples/storytelling-chatbot/src/bot.py

@@ -14,10 +14,7 @@ from pipecat.frames.frames import EndFrame, LLMMessagesFrame, StopTaskFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.fal import FalImageGenService
 from pipecat.services.openai import OpenAILLMService
@@ -82,8 +79,8 @@ async def main(room_url, token=None):
         story_pages = []
 
         # We need aggregators to keep track of user and LLM responses
-        llm_responses = LLMAssistantResponseAggregator(message_history)
+        context = OpenAILLMContext(message_history)
-        user_responses = LLMUserResponseAggregator(message_history)
+        context_aggregator = llm_service.create_context_aggregator(context)
 
         # -------------- Processors ------------- #
 
@@ -126,13 +123,13 @@ async def main(room_url, token=None):
         main_pipeline = Pipeline(
             [
                 transport.input(),
-                user_responses,
+                context_aggregator.user(),
                 llm_service,
                 story_processor,
                 image_processor,
                 tts_service,
                 transport.output(),
-                llm_responses,
+                context_aggregator.assistant(),
             ]
         )
 

examples/studypal/studypal.py

@@ -13,10 +13,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -150,17 +147,17 @@ Your task is to help the user understand and learn from this article in 2 senten
             },
         ]
 
-        tma_in = LLMUserResponseAggregator(messages)
+        context = OpenAILLMContext(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                tma_in,
+                context_aggregator.user(),
                 llm,
                 tts,
                 transport.output(),
-                tma_out,
+                context_aggregator.assistant(),
             ]
         )
 

examples/twilio-chatbot/bot.py

@@ -6,10 +6,7 @@ from pipecat.frames.frames import EndFrame, LLMMessagesFrame
 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 (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.deepgram import DeepgramSTTService
@@ -58,18 +55,18 @@ async def run_bot(websocket_client, stream_sid):
         },
     ]
 
-    tma_in = LLMUserResponseAggregator(messages)
+    context = OpenAILLMContext(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
+    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Websocket input from client
             stt,  # Speech-To-Text
-            tma_in,  # User responses
+            context_aggregator.user(),
             llm,  # LLM
             tts,  # Text-To-Speech
             transport.output(),  # Websocket output to client
-            tma_out,  # LLM responses
+            context_aggregator.assistant(),
         ]
     )
 

examples/websocket-server/bot.py

@@ -13,10 +13,7 @@ from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
-)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.openai import OpenAILLMService
@@ -62,18 +59,18 @@ async def main():
         },
     ]
 
-    tma_in = LLMUserResponseAggregator(messages)
+    context = OpenAILLMContext(messages)
-    tma_out = LLMAssistantResponseAggregator(messages)
+    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Websocket input from client
             stt,  # Speech-To-Text
-            tma_in,  # User responses
+            context_aggregator.user(),
             llm,  # LLM
             tts,  # Text-To-Speech
             transport.output(),  # Websocket output to client
-            tma_out,  # LLM responses
+            context_aggregator.assistant(),
         ]
     )
 

src/pipecat/processors/frameworks/langchain.py

@@ -39,7 +39,7 @@ class LangchainProcessor(FrameProcessor):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, LLMMessagesFrame):
-            # Messages are accumulated by the `LLMUserResponseAggregator` in a list of messages.
+            # Messages are accumulated on the context as a list of messages.
             # The last one by the human is the one we want to send to the LLM.
             logger.debug(f"Got transcription frame {frame}")
             text: str = frame.messages[-1]["content"]