From 236ac93ac60803279ebde30ea3a5a5b2d3e9495d Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 24 Sep 2025 15:54:01 -0400 Subject: [PATCH] Remove remaining usage of `OpenAILLMContext` throughout the codebase in favor of `LLMContext`, except for: - Usage in classes that are already deprecated - Usage related to realtime LLMs, which don't yet support `LLMContext` - Usage in (soon-to-be-deprecated) code paths related to `OpenAILLMContext` itself and associated machinery --- scripts/evals/eval.py | 7 +++-- .../aggregators/llm_response_universal.py | 22 +++++++++----- .../processors/aggregators/user_response.py | 12 ++++---- ...st_integration_unified_function_calling.py | 27 +++++++++-------- tests/test_langchain.py | 30 +++++++++---------- 5 files changed, 53 insertions(+), 45 deletions(-) diff --git a/scripts/evals/eval.py b/scripts/evals/eval.py index fbfef7de5..5a164ad84 100644 --- a/scripts/evals/eval.py +++ b/scripts/evals/eval.py @@ -34,7 +34,8 @@ from pipecat.frames.frames import EndTaskFrame, LLMRunFrame, OutputImageRawFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.processors.frame_processor import FrameDirection from pipecat.runner.types import RunnerArguments @@ -283,8 +284,8 @@ async def run_eval_pipeline( }, ] - context = OpenAILLMContext(messages, tools) - context_aggregator = llm.create_context_aggregator(context) + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) audio_buffer = AudioBufferProcessor() diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 9d7b7a47c..69a8dd280 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -13,6 +13,7 @@ LLM processing, and text-to-speech components in conversational AI pipelines. import asyncio import json +from abc import abstractmethod from typing import Any, Dict, List, Literal, Optional, Set from loguru import logger @@ -169,6 +170,11 @@ class LLMContextAggregator(FrameProcessor): """Reset the aggregation state.""" self._aggregation = "" + @abstractmethod + async def push_aggregation(self): + """Push the current aggregation downstream.""" + pass + class LLMUserAggregator(LLMContextAggregator): """User LLM aggregator that processes speech-to-text transcriptions. @@ -301,7 +307,7 @@ class LLMUserAggregator(LLMContextAggregator): frame = LLMContextFrame(self._context) await self.push_frame(frame) - async def _push_aggregation(self): + async def push_aggregation(self): """Push the current aggregation based on interruption strategies and conditions.""" if len(self._aggregation) > 0: if self.interruption_strategies and self._bot_speaking: @@ -392,7 +398,7 @@ class LLMUserAggregator(LLMContextAggregator): # pushing the aggregation as we will probably get a final transcription. if len(self._aggregation) > 0: if not self._seen_interim_results: - await self._push_aggregation() + await self.push_aggregation() # Handles the case where both the user and the bot are not speaking, # and the bot was previously speaking before the user interruption. # So in this case we are resetting the aggregation timer @@ -471,7 +477,7 @@ class LLMUserAggregator(LLMContextAggregator): await self._maybe_emulate_user_speaking() except asyncio.TimeoutError: if not self._user_speaking: - await self._push_aggregation() + await self.push_aggregation() # If we are emulating VAD we still need to send the user stopped # speaking frame. @@ -607,12 +613,12 @@ class LLMAssistantAggregator(LLMContextAggregator): elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id: await self._handle_user_image_frame(frame) elif isinstance(frame, BotStoppedSpeakingFrame): - await self._push_aggregation() + await self.push_aggregation() await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) - async def _push_aggregation(self): + async def push_aggregation(self): """Push the current assistant aggregation with timestamp.""" if not self._aggregation: return @@ -644,7 +650,7 @@ class LLMAssistantAggregator(LLMContextAggregator): await self.push_context_frame(FrameDirection.UPSTREAM) async def _handle_interruptions(self, frame: InterruptionFrame): - await self._push_aggregation() + await self.push_aggregation() self._started = 0 await self.reset() @@ -778,7 +784,7 @@ class LLMAssistantAggregator(LLMContextAggregator): text=frame.request.context, ) - await self._push_aggregation() + await self.push_aggregation() await self.push_context_frame(FrameDirection.UPSTREAM) async def _handle_llm_start(self, _: LLMFullResponseStartFrame): @@ -786,7 +792,7 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_llm_end(self, _: LLMFullResponseEndFrame): self._started -= 1 - await self._push_aggregation() + await self.push_aggregation() async def _handle_text(self, frame: TextFrame): if not self._started: diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index 958c6513f..274a31d52 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/src/pipecat/processors/aggregators/user_response.py @@ -12,14 +12,14 @@ in conversational pipelines. """ from pipecat.frames.frames import TextFrame -from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMUserAggregator -class UserResponseAggregator(LLMUserContextAggregator): +class UserResponseAggregator(LLMUserAggregator): """Aggregates user responses into TextFrame objects. - This aggregator extends LLMUserContextAggregator to specifically handle + This aggregator extends LLMUserAggregator to specifically handle user input by collecting text responses and outputting them as TextFrame objects when the aggregation is complete. """ @@ -28,9 +28,9 @@ class UserResponseAggregator(LLMUserContextAggregator): """Initialize the user response aggregator. Args: - **kwargs: Additional arguments passed to parent LLMUserContextAggregator. + **kwargs: Additional arguments passed to parent LLMUserAggregator. """ - super().__init__(context=OpenAILLMContext(), **kwargs) + super().__init__(context=LLMContext(), **kwargs) async def push_aggregation(self): """Push the aggregated user response as a TextFrame. diff --git a/tests/integration/test_integration_unified_function_calling.py b/tests/integration/test_integration_unified_function_calling.py index 09611fd3a..4f66ab3cf 100644 --- a/tests/integration/test_integration_unified_function_calling.py +++ b/tests/integration/test_integration_unified_function_calling.py @@ -12,14 +12,12 @@ from dotenv import load_dotenv from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.frames.frames import LLMContextFrame from pipecat.pipeline.pipeline import Pipeline -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.services.anthropic.llm import AnthropicLLMService from pipecat.services.google.llm import GoogleLLMService -from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallParams, LLMService from pipecat.services.openai.llm import OpenAILLMService from pipecat.tests.utils import run_test @@ -48,8 +46,13 @@ def standard_tools() -> ToolsSchema: async def _test_llm_function_calling(llm: LLMService): - # Create an AsyncMock for the function - mock_fetch_weather = AsyncMock() + # Create a mock weather function + call_count = 0 + + async def mock_fetch_weather(params: FunctionCallParams): + nonlocal call_count + call_count += 1 + pass llm.register_function(None, mock_fetch_weather) @@ -60,21 +63,19 @@ async def _test_llm_function_calling(llm: LLMService): }, {"role": "user", "content": " How is the weather today in San Francisco, California?"}, ] - context = OpenAILLMContext(messages, standard_tools()) - # This is done by default inside the create_context_aggregator - context.set_llm_adapter(llm.get_llm_adapter()) + context = LLMContext(messages, standard_tools()) pipeline = Pipeline([llm]) - frames_to_send = [OpenAILLMContextFrame(context)] + frames_to_send = [LLMContextFrame(context)] await run_test( pipeline, frames_to_send=frames_to_send, expected_down_frames=None, ) - # Assert that the mock function was called - mock_fetch_weather.assert_called_once() + # Assert that the weather function was called once + assert call_count == 1 @pytest.mark.skipif(os.getenv("OPENAI_API_KEY") is None, reason="OPENAI_API_KEY is not set") diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 366dfeb97..dd7f9ccef 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -10,24 +10,21 @@ from langchain.prompts import ChatPromptTemplate from langchain_core.language_models import FakeStreamingListLLM from pipecat.frames.frames import ( + LLMContextAssistantTimestampFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, - OpenAILLMContextAssistantTimestampFrame, TextFrame, TranscriptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) from pipecat.pipeline.pipeline import Pipeline +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, - LLMAssistantContextAggregator, - LLMUserContextAggregator, -) -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, ) +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.frame_processor import FrameProcessor from pipecat.processors.frameworks.langchain import LangchainProcessor from pipecat.tests.utils import SleepFrame, run_test @@ -67,13 +64,14 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): proc = LangchainProcessor(chain=chain) self.mock_proc = self.MockProcessor("token_collector") - context = OpenAILLMContext() - tma_in = LLMUserContextAggregator(context) - tma_out = LLMAssistantContextAggregator( - context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + context = LLMContext() + context_aggregator = LLMContextAggregatorPair( + context, assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False) ) - pipeline = Pipeline([tma_in, proc, self.mock_proc, tma_out]) + pipeline = Pipeline( + [context_aggregator.user(), proc, self.mock_proc, context_aggregator.assistant()] + ) frames_to_send = [ UserStartedSpeakingFrame(), @@ -84,8 +82,8 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ UserStartedSpeakingFrame, UserStoppedSpeakingFrame, - OpenAILLMContextFrame, - OpenAILLMContextAssistantTimestampFrame, + LLMContextFrame, + LLMContextAssistantTimestampFrame, ] await run_test( pipeline, @@ -94,4 +92,6 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): ) self.assertEqual("".join(self.mock_proc.token), self.expected_response) - self.assertEqual(tma_out.messages[-1]["content"], self.expected_response) + self.assertEqual( + context_aggregator.assistant().messages[-1]["content"], self.expected_response + )