Merge pull request #2380 from pipecat-ai/pk/deprecate-llm-messages-frame

Deprecate `LLMMessagesFrame`, `LLMUserResponseAggregator`, and `LLMAssistantResponseAggregator`
2025-08-07 15:13:01 -04:00
parent 315e45d41b 9ea06c33f7
commit 9ca79232c1
16 changed files with 143 additions and 84 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,20 @@ 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]
+
+### Deprecated
+
+- `LLMMessagesFrame` is deprecated, in favor of either:
+
+  - `LLMMessagesUpdateFrame` with `run_llm=True`
+  - `OpenAILLMContextFrame` with desired messages in a new context
+
+- `LLMUserResponseAggregator` and `LLMAssistantResponseAggregator` are
+  deprecated, as they depended on the now-deprecated `LLMMessagesFrame`. Use
+  `LLMUserContextAggregator` and `LLMAssistantResponseAggregator` (or
+  LLM-specific subclasses thereof) instead.
+
 ## [0.0.78] - 2025-08-07

 ### Added
--- a/env.example
+++ b/env.example
@@ -29,6 +29,9 @@ CARTESIA_API_KEY=...
 DAILY_API_KEY=...
 DAILY_SAMPLE_ROOM_URL=https://...

+# Deepgram
+DEEPGRAM_API_KEY=...
+
 # ElevenLabs
 ELEVENLABS_API_KEY=...
 ELEVENLABS_VOICE_ID=...
--- a/examples/foundational/02-llm-say-one-thing.py
+++ b/examples/foundational/02-llm-say-one-thing.py
@@ -9,10 +9,14 @@ import os
 from dotenv import load_dotenv
 from loguru import logger

-from pipecat.frames.frames import EndFrame, LLMMessagesFrame
+from pipecat.frames.frames import EndFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -59,7 +63,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    # Register an event handler so we can play the audio when the client joins
    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
-        await task.queue_frames([LLMMessagesFrame(messages), EndFrame()])
+        await task.queue_frames([OpenAILLMContextFrame(OpenAILLMContext(messages)), EndFrame()])

    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)

--- a/examples/foundational/05-sync-speech-and-image.py
+++ b/examples/foundational/05-sync-speech-and-image.py
@@ -15,13 +15,16 @@ from pipecat.frames.frames import (
    DataFrame,
    Frame,
    LLMFullResponseStartFrame,
-    LLMMessagesFrame,
    TextFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
 from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.processors.aggregators.sentence import SentenceAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.runner.types import RunnerArguments
@@ -153,7 +156,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
                }
            ]
            frames.append(MonthFrame(month=month))
-            frames.append(LLMMessagesFrame(messages))
+            frames.append(OpenAILLMContextFrame(OpenAILLMContext(messages)))

        task = PipelineTask(
            pipeline,
--- a/examples/foundational/05a-local-sync-speech-and-image.py
+++ b/examples/foundational/05a-local-sync-speech-and-image.py
@@ -15,7 +15,6 @@ from loguru import logger

 from pipecat.frames.frames import (
    Frame,
-    LLMMessagesFrame,
    OutputAudioRawFrame,
    TextFrame,
    TTSAudioRawFrame,
@@ -25,6 +24,10 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
 from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.processors.aggregators.sentence import SentenceAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia.tts import CartesiaHttpTTSService
@@ -137,7 +140,7 @@ async def main():
            )

            task = PipelineTask(pipeline)
-            await task.queue_frame(LLMMessagesFrame(messages))
+            await task.queue_frame(OpenAILLMContextFrame(OpenAILLMContext(messages)))
            await task.stop_when_done()

            await runner.run(task)
--- a/examples/foundational/07b-interruptible-langchain.py
+++ b/examples/foundational/07b-interruptible-langchain.py
@@ -16,13 +16,16 @@ from langchain_openai import ChatOpenAI
 from loguru import logger

 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.frames.frames import LLMMessagesUpdateFrame
 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,
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+)
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
 )
 from pipecat.processors.frameworks.langchain import LangchainProcessor
 from pipecat.runner.types import RunnerArguments
@@ -97,8 +100,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    )
    lc = LangchainProcessor(history_chain)

-    tma_in = LLMUserResponseAggregator()
-    tma_out = LLMAssistantResponseAggregator()
+    context = OpenAILLMContext()
+    tma_in = LLMUserContextAggregator(context=context)
+    tma_out = LLMAssistantContextAggregator(context=context)

    pipeline = Pipeline(
        [
@@ -125,11 +129,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    async def on_client_connected(transport, client):
        logger.info(f"Client connected")
        # Kick off the conversation.
-        # the `LLMMessagesFrame` will be picked up by the LangchainProcessor using
+        # An `OpenAILLMContextFrame` will be picked up by the LangchainProcessor using
        # only the content of the last message to inject it in the prompt defined
        # above. So no role is required here.
        messages = [({"content": "Please briefly introduce yourself to the user."})]
-        await task.queue_frames([LLMMessagesFrame(messages)])
+        await task.queue_frames([LLMMessagesUpdateFrame(messages, run_llm=True)])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
--- a/examples/foundational/08-bots-arguing.py
+++ b/examples/foundational/08-bots-arguing.py
@@ -6,9 +6,13 @@ from typing import Tuple
 import aiohttp
 from dotenv import load_dotenv

-from pipecat.frames.frames import AudioFrame, EndFrame, ImageFrame, LLMMessagesFrame, TextFrame
+from pipecat.frames.frames import AudioFrame, EndFrame, ImageFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.aggregators import SentenceAggregator
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.runner.daily import configure
 from pipecat.services.azure import AzureLLMService, AzureTTSService
 from pipecat.services.elevenlabs import ElevenLabsTTSService
@@ -79,7 +83,7 @@ async def main():
            sentence_aggregator = SentenceAggregator()
            pipeline = Pipeline([llm, sentence_aggregator, tts1], source_queue, sink_queue)

-            await source_queue.put(LLMMessagesFrame(messages))
+            await source_queue.put(OpenAILLMContextFrame(OpenAILLMContext(messages)))
            await source_queue.put(EndFrame())
            await pipeline.run_pipeline()

--- a/examples/foundational/17-detect-user-idle.py
+++ b/examples/foundational/17-detect-user-idle.py
@@ -11,7 +11,7 @@ from dotenv import load_dotenv
 from loguru import logger

 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import EndFrame, LLMMessagesFrame, TTSSpeakFrame
+from pipecat.frames.frames import EndFrame, LLMMessagesAppendFrame, TTSSpeakFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -75,23 +75,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    async def handle_user_idle(user_idle: UserIdleProcessor, retry_count: int) -> bool:
        if retry_count == 1:
            # First attempt: Add a gentle prompt to the conversation
-            messages.append(
-                {
-                    "role": "system",
-                    "content": "The user has been quiet. Politely and briefly ask if they're still there.",
-                }
-            )
-            await user_idle.push_frame(LLMMessagesFrame(messages))
+            message = {
+                "role": "system",
+                "content": "The user has been quiet. Politely and briefly ask if they're still there.",
+            }
+            await user_idle.push_frame(LLMMessagesAppendFrame([message], run_llm=True))
            return True
        elif retry_count == 2:
            # Second attempt: More direct prompt
-            messages.append(
-                {
-                    "role": "system",
-                    "content": "The user is still inactive. Ask if they'd like to continue our conversation.",
-                }
-            )
-            await user_idle.push_frame(LLMMessagesFrame(messages))
+            message = {
+                "role": "system",
+                "content": "The user is still inactive. Ask if they'd like to continue our conversation.",
+            }
+            await user_idle.push_frame(LLMMessagesAppendFrame([message], run_llm=True))
            return True
        else:
            # Third attempt: End the conversation
--- a/examples/foundational/22b-natural-conversation-proposal.py
+++ b/examples/foundational/22b-natural-conversation-proposal.py
@@ -19,7 +19,6 @@ from pipecat.frames.frames import (
    Frame,
    FunctionCallInProgressFrame,
    FunctionCallResultFrame,
-    LLMMessagesFrame,
    StartFrame,
    StartInterruptionFrame,
    StopInterruptionFrame,
@@ -60,10 +59,6 @@ classifier_statement = "Determine if the user's statement ends with a complete t


 class StatementJudgeContextFilter(FrameProcessor):
-    def __init__(self, notifier: BaseNotifier, **kwargs):
-        super().__init__(**kwargs)
-        self._notifier = notifier
-
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        # We must not block system frames.
@@ -71,13 +66,8 @@ class StatementJudgeContextFilter(FrameProcessor):
            await self.push_frame(frame, direction)
            return

-        # Just treat an LLMMessagesFrame as complete, no matter what.
-        if isinstance(frame, LLMMessagesFrame):
-            await self._notifier.notify()
-            return
-
-        # Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple
-        # messages frame that contains a system prompt and the most recent user messages,
+        # We only want to handle OpenAILLMContextFrames, and only want to push through a simplified
+        # context frame that contains a system prompt and the most recent user messages,
        # concatenated.
        if isinstance(frame, OpenAILLMContextFrame):
            logger.debug(f"Context Frame: {frame}")
@@ -96,7 +86,7 @@ class StatementJudgeContextFilter(FrameProcessor):
                    for content in message["content"]:
                        if content["type"] == "text":
                            user_text_messages.insert(0, content["text"])
-            # If we have any user text content, push an LLMMessagesFrame
+            # If we have any user text content, push a context frame with the simplified context.
            if user_text_messages:
                logger.debug(f"User text messages: {user_text_messages}")
                user_message = " ".join(reversed(user_text_messages))
@@ -110,7 +100,7 @@ class StatementJudgeContextFilter(FrameProcessor):
                if last_assistant_message:
                    messages.append(last_assistant_message)
                messages.append({"role": "user", "content": user_message})
-                await self.push_frame(LLMMessagesFrame(messages))
+                await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages)))


 class CompletenessCheck(FrameProcessor):
@@ -296,7 +286,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):

    # This turns the LLM context into an inference request to classify the user's speech
    # as complete or incomplete.
-    statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier)
+    statement_judge_context_filter = StatementJudgeContextFilter()

    # This sends a UserStoppedSpeakingFrame and triggers the notifier event
    completeness_check = CompletenessCheck(notifier=notifier)
@@ -316,7 +306,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    async def pass_only_llm_trigger_frames(frame):
        return (
            isinstance(frame, OpenAILLMContextFrame)
-            or isinstance(frame, LLMMessagesFrame)
            or isinstance(frame, StartInterruptionFrame)
            or isinstance(frame, StopInterruptionFrame)
            or isinstance(frame, FunctionCallInProgressFrame)
@@ -331,14 +320,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
            ParallelPipeline(
                [
                    # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
-                    # LLMMessagesFrame to the statement classifier LLM. The only frame this
+                    # simplified context frame to the statement classifier LLM. The only frame this
                    # sub-pipeline will output is a UserStoppedSpeakingFrame.
                    statement_judge_context_filter,
                    statement_llm,
                    completeness_check,
                ],
                [
-                    # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
+                    # Block everything except frames that trigger LLM inference.
                    FunctionFilter(filter=pass_only_llm_trigger_frames),
                    llm,
                    bot_output_gate,  # Buffer all llm/tts output until notified.
--- a/examples/foundational/22c-natural-conversation-mixed-llms.py
+++ b/examples/foundational/22c-natural-conversation-mixed-llms.py
@@ -19,7 +19,6 @@ from pipecat.frames.frames import (
    Frame,
    FunctionCallInProgressFrame,
    FunctionCallResultFrame,
-    LLMMessagesFrame,
    StartFrame,
    StartInterruptionFrame,
    StopInterruptionFrame,
@@ -266,10 +265,6 @@ Please be very concise in your responses. Unless you are explicitly asked to do


 class StatementJudgeContextFilter(FrameProcessor):
-    def __init__(self, notifier: BaseNotifier, **kwargs):
-        super().__init__(**kwargs)
-        self._notifier = notifier
-
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        # We must not block system frames.
@@ -277,14 +272,8 @@ class StatementJudgeContextFilter(FrameProcessor):
            await self.push_frame(frame, direction)
            return

-        # Just treat an LLMMessagesFrame as complete, no matter what.
-        if isinstance(frame, LLMMessagesFrame):
-            await self._notifier.notify()
-            return
-
-        # Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple
-        # messages frame that contains a system prompt and the most recent user messages,
-        # concatenated.
+        # We only want to handle OpenAILLMContextFrames, and only want to push through a simplified
+        # context frame that contains a system prompt and the most recent user messages,
        if isinstance(frame, OpenAILLMContextFrame):
            # Take text content from the most recent user messages.
            messages = frame.context.messages
@@ -301,7 +290,7 @@ class StatementJudgeContextFilter(FrameProcessor):
                    for content in message["content"]:
                        if content["type"] == "text":
                            user_text_messages.insert(0, content["text"])
-            # If we have any user text content, push an LLMMessagesFrame
+            # If we have any user text content, push a context frame with the simplified context.
            if user_text_messages:
                user_message = " ".join(reversed(user_text_messages))
                logger.debug(f"!!! {user_message}")
@@ -314,7 +303,7 @@ class StatementJudgeContextFilter(FrameProcessor):
                if last_assistant_message:
                    messages.append(last_assistant_message)
                messages.append({"role": "user", "content": user_message})
-                await self.push_frame(LLMMessagesFrame(messages))
+                await self.push_frame(OpenAILLMContextFrame(OpenAILLMContext(messages)))


 class CompletenessCheck(FrameProcessor):
@@ -499,7 +488,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):

    # This turns the LLM context into an inference request to classify the user's speech
    # as complete or incomplete.
-    statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier)
+    statement_judge_context_filter = StatementJudgeContextFilter()

    # This sends a UserStoppedSpeakingFrame and triggers the notifier event
    completeness_check = CompletenessCheck(notifier=notifier)
@@ -522,7 +511,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    async def pass_only_llm_trigger_frames(frame):
        return (
            isinstance(frame, OpenAILLMContextFrame)
-            or isinstance(frame, LLMMessagesFrame)
            or isinstance(frame, StartInterruptionFrame)
            or isinstance(frame, StopInterruptionFrame)
            or isinstance(frame, FunctionCallInProgressFrame)
@@ -542,14 +530,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
                ],
                [
                    # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
-                    # LLMMessagesFrame to the statement classifier LLM. The only frame this
+                    # simplified context frame to the statement classifier LLM. The only frame this
                    # sub-pipeline will output is a UserStoppedSpeakingFrame.
                    statement_judge_context_filter,
                    statement_llm,
                    completeness_check,
                ],
                [
-                    # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
+                    # Block everything except frames that trigger LLM inference.
                    FunctionFilter(filter=pass_only_llm_trigger_frames),
                    llm,
                    bot_output_gate,  # Buffer all llm/tts output until notified.
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -478,6 +478,12 @@ class TranscriptionUpdateFrame(DataFrame):
 class LLMMessagesFrame(DataFrame):
    """Frame containing LLM messages for chat completion.

+    .. deprecated:: 0.0.79
+        This class is deprecated and will be removed in a future version.
+        Instead, use either:
+        - `LLMMessagesUpdateFrame` with `run_llm=True`
+        - `OpenAILLMContextFrame` with desired messages in a new context
+
    A frame containing a list of LLM messages. Used to signal that an LLM
    service should run a chat completion and emit an LLMFullResponseStartFrame,
    TextFrames and an LLMFullResponseEndFrame. Note that the `messages`
@@ -490,6 +496,20 @@ class LLMMessagesFrame(DataFrame):

    messages: List[dict]

+    def __post_init__(self):
+        super().__post_init__()
+        import warnings
+
+        warnings.simplefilter("always")
+        warnings.warn(
+            "LLMMessagesFrame is deprecated and will be removed in a future version. "
+            "Instead, use either "
+            "`LLMMessagesUpdateFrame` with `run_llm=True`, or "
+            "`OpenAILLMContextFrame` with desired messages in a new context",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+

@dataclass
 class LLMMessagesAppendFrame(DataFrame):
--- a/src/pipecat/processors/aggregators/llm_response.py
+++ b/src/pipecat/processors/aggregators/llm_response.py
@@ -12,6 +12,7 @@ LLM processing, and text-to-speech components in conversational AI pipelines.
 """

 import asyncio
+import warnings
 from abc import abstractmethod
 from dataclasses import dataclass
 from typing import Dict, List, Literal, Optional, Set
@@ -994,6 +995,10 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
 class LLMUserResponseAggregator(LLMUserContextAggregator):
    """User response aggregator that outputs LLMMessagesFrame instead of context frames.

+    .. deprecated:: 0.0.79
+        This class is deprecated and will be removed in a future version.
+        Use `LLMUserContextAggregator` or another LLM-specific subclass instead.
+
    This aggregator extends LLMUserContextAggregator but pushes LLMMessagesFrame
    objects downstream instead of OpenAILLMContextFrame objects. This is useful
    when you need message-based output rather than context-based output.
@@ -1013,6 +1018,12 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
            params: Configuration parameters for aggregation behavior.
            **kwargs: Additional arguments passed to parent class.
        """
+        warnings.warn(
+            "LLMUserResponseAggregator is deprecated and will be removed in a future version. "
+            "Use LLMUserContextAggregator or another LLM-specific subclass instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
        super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)

    async def _process_aggregation(self):
@@ -1027,6 +1038,10 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
 class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
    """Assistant response aggregator that outputs LLMMessagesFrame instead of context frames.

+    .. deprecated:: 0.0.79
+        This class is deprecated and will be removed in a future version.
+        Use `LLMAssistantContextAggregator` or another LLM-specific subclass instead.
+
    This aggregator extends LLMAssistantContextAggregator but pushes LLMMessagesFrame
    objects downstream instead of OpenAILLMContextFrame objects. This is useful
    when you need message-based output rather than context-based output.
@@ -1046,6 +1061,12 @@ class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
            params: Configuration parameters for aggregation behavior.
            **kwargs: Additional arguments passed to parent class.
        """
+        warnings.warn(
+            "LLMAssistantResponseAggregator is deprecated and will be removed in a future version. "
+            "Use LLMAssistantContextAggregator or another LLM-specific subclass instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
        super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)

    async def push_aggregation(self):
--- a/src/pipecat/processors/aggregators/user_response.py
+++ b/src/pipecat/processors/aggregators/user_response.py
@@ -12,13 +12,14 @@ in conversational pipelines.
 """

 from pipecat.frames.frames import TextFrame
-from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregator
+from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext


-class UserResponseAggregator(LLMUserResponseAggregator):
+class UserResponseAggregator(LLMUserContextAggregator):
    """Aggregates user responses into TextFrame objects.

-    This aggregator extends LLMUserResponseAggregator to specifically handle
+    This aggregator extends LLMUserContextAggregator to specifically handle
    user input by collecting text responses and outputting them as TextFrame
    objects when the aggregation is complete.
    """
@@ -27,9 +28,9 @@ class UserResponseAggregator(LLMUserResponseAggregator):
        """Initialize the user response aggregator.

        Args:
-            **kwargs: Additional arguments passed to parent LLMUserResponseAggregator.
+            **kwargs: Additional arguments passed to parent LLMUserContextAggregator.
        """
-        super().__init__(**kwargs)
+        super().__init__(context=OpenAILLMContext(), **kwargs)

    async def push_aggregation(self):
        """Push the aggregated user response as a TextFrame.
--- a/src/pipecat/processors/frameworks/langchain.py
+++ b/src/pipecat/processors/frameworks/langchain.py
@@ -14,9 +14,9 @@ from pipecat.frames.frames import (
    Frame,
    LLMFullResponseEndFrame,
    LLMFullResponseStartFrame,
-    LLMMessagesFrame,
    TextFrame,
 )
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor

 try:
@@ -64,11 +64,11 @@ class LangchainProcessor(FrameProcessor):
        """
        await super().process_frame(frame, direction)

-        if isinstance(frame, LLMMessagesFrame):
+        if isinstance(frame, OpenAILLMContextFrame):
            # 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"]
+            text: str = frame.context.messages[-1]["content"]

            await self._ainvoke(text.strip())
        else:
--- a/src/pipecat/services/anthropic/llm.py
+++ b/src/pipecat/services/anthropic/llm.py
@@ -448,14 +448,16 @@ class AnthropicLLMContext(OpenAILLMContext):
            system: System message content.
        """
        super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
+        self.__setup_local()
+        self.system = system

+    def __setup_local(self):
        # For beta prompt caching. This is a counter that tracks the number of turns
        # we've seen above the cache threshold. We reset this when we reset the
        # messages list. We only care about this number being 0, 1, or 2. But
        # it's easiest just to treat it as a counter.
        self.turns_above_cache_threshold = 0
-
-        self.system = system
+        return

    @staticmethod
    def upgrade_to_anthropic(obj: OpenAILLMContext) -> "AnthropicLLMContext":
@@ -472,6 +474,7 @@ class AnthropicLLMContext(OpenAILLMContext):
        logger.debug(f"Upgrading to Anthropic: {obj}")
        if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AnthropicLLMContext):
            obj.__class__ = AnthropicLLMContext
+            obj.__setup_local()
            obj._restructure_from_openai_messages()
        return obj

--- a/tests/test_langchain.py
+++ b/tests/test_langchain.py
@@ -12,7 +12,7 @@ from langchain_core.language_models import FakeStreamingListLLM
 from pipecat.frames.frames import (
    LLMFullResponseEndFrame,
    LLMFullResponseStartFrame,
-    LLMMessagesFrame,
+    OpenAILLMContextAssistantTimestampFrame,
    TextFrame,
    TranscriptionFrame,
    UserStartedSpeakingFrame,
@@ -21,8 +21,12 @@ from pipecat.frames.frames import (
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.aggregators.llm_response import (
    LLMAssistantAggregatorParams,
-    LLMAssistantResponseAggregator,
-    LLMUserResponseAggregator,
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+)
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.processors.frameworks.langchain import LangchainProcessor
@@ -63,9 +67,10 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase):
        proc = LangchainProcessor(chain=chain)
        self.mock_proc = self.MockProcessor("token_collector")

-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(
-            messages, params=LLMAssistantAggregatorParams(expect_stripped_words=False)
+        context = OpenAILLMContext()
+        tma_in = LLMUserContextAggregator(context)
+        tma_out = LLMAssistantContextAggregator(
+            context, params=LLMAssistantAggregatorParams(expect_stripped_words=False)
        )

        pipeline = Pipeline([tma_in, proc, self.mock_proc, tma_out])
@@ -79,7 +84,8 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase):
        expected_down_frames = [
            UserStartedSpeakingFrame,
            UserStoppedSpeakingFrame,
-            LLMMessagesFrame,
+            OpenAILLMContextFrame,
+            OpenAILLMContextAssistantTimestampFrame,
        ]
        await run_test(
            pipeline,