diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f0f1598c..3f4d47273 100644 --- 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 diff --git a/env.example b/env.example index ea99c567e..fae6bb2a4 100644 --- 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=... diff --git a/examples/foundational/02-llm-say-one-thing.py b/examples/foundational/02-llm-say-one-thing.py index 3dc739059..1282f3788 100644 --- 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) diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index 6adff936d..384ef617c 100644 --- 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, diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index a01f07c0e..f40d2a163 100644 --- 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) diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index 37ec755bb..4afe73315 100644 --- 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): diff --git a/examples/foundational/08-bots-arguing.py b/examples/foundational/08-bots-arguing.py index b052ef08a..a38bc273a 100644 --- 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() diff --git a/examples/foundational/17-detect-user-idle.py b/examples/foundational/17-detect-user-idle.py index d40a9699c..2d97f10e2 100644 --- 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 diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 72c5ba310..90594cf09 100644 --- 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. diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index c9ea66afb..c44034f1c 100644 --- 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. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e587e563d..a8acd35bc 100644 --- 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): diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 3b1215473..38848ac1d 100644 --- 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): diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index 87c9d9f58..958c6513f 100644 --- 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. diff --git a/src/pipecat/processors/frameworks/langchain.py b/src/pipecat/processors/frameworks/langchain.py index 65bf56b70..b67e25105 100644 --- 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: diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 15c4cebee..fadf17f16 100644 --- 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 diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 3d907e084..366dfeb97 100644 --- 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,