From 19e521b75afc4059d45babf26e818ca04c6c8385 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 31 Mar 2026 18:35:48 -0400 Subject: [PATCH] Simplify LLMContextFrame handling in process_frame methods Now that LLMContextFrame is the only frame that provides a context, remove the intermediate `context = None` / `if context:` pattern and handle context processing directly in the isinstance branch. --- src/pipecat/services/anthropic/llm.py | 6 +----- src/pipecat/services/aws/llm.py | 6 +----- src/pipecat/services/aws/nova_sonic/llm.py | 3 +-- src/pipecat/services/google/llm.py | 7 +------ src/pipecat/services/mem0/memory.py | 5 ----- src/pipecat/services/openai/base_llm.py | 10 +++------- src/pipecat/services/openai/responses/llm.py | 20 ++++++-------------- 7 files changed, 13 insertions(+), 44 deletions(-) diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 58dc1da07..9c07a8444 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -552,18 +552,14 @@ class AnthropicLLMService(LLMService): """ await super().process_frame(frame, direction) - context = None if isinstance(frame, LLMContextFrame): - context = frame.context + await self._process_context(frame.context) elif isinstance(frame, LLMEnablePromptCachingFrame): logger.debug(f"Setting enable prompt caching to: [{frame.enable}]") self._settings.enable_prompt_caching = frame.enable else: await self.push_frame(frame, direction) - if context: - await self._process_context(context) - def _estimate_tokens(self, text: str) -> int: return int(len(re.split(r"[^\w]+", text)) * 1.3) diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 2e2bd3c59..722072592 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -564,15 +564,11 @@ class AWSBedrockLLMService(LLMService): """ await super().process_frame(frame, direction) - context = None if isinstance(frame, LLMContextFrame): - context = frame.context + await self._process_context(frame.context) else: await self.push_frame(frame, direction) - if context: - await self._process_context(context) - def _estimate_tokens(self, text: str) -> int: return int(len(re.split(r"[^\w]+", text)) * 1.3) diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py index 545c42b1c..9aa36c5db 100644 --- a/src/pipecat/services/aws/nova_sonic/llm.py +++ b/src/pipecat/services/aws/nova_sonic/llm.py @@ -524,8 +524,7 @@ class AWSNovaSonicLLMService(LLMService): await super().process_frame(frame, direction) if isinstance(frame, LLMContextFrame): - context = frame.context - await self._handle_context(context) + await self._handle_context(frame.context) elif isinstance(frame, InputAudioRawFrame): await self._handle_input_audio_frame(frame) elif isinstance(frame, InterruptionFrame): diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 7af1754a6..b336fcef5 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -637,16 +637,11 @@ class GoogleLLMService(LLMService): """ await super().process_frame(frame, direction) - context = None - if isinstance(frame, LLMContextFrame): - context = frame.context + await self._process_context(frame.context) else: await self.push_frame(frame, direction) - if context: - await self._process_context(context) - async def stop(self, frame): """Override stop to gracefully close the client.""" await super().stop(frame) diff --git a/src/pipecat/services/mem0/memory.py b/src/pipecat/services/mem0/memory.py index 86f37be41..91396cab4 100644 --- a/src/pipecat/services/mem0/memory.py +++ b/src/pipecat/services/mem0/memory.py @@ -265,12 +265,8 @@ class Mem0MemoryService(FrameProcessor): """ await super().process_frame(frame, direction) - context = None - if isinstance(frame, LLMContextFrame): context = frame.context - - if context: try: # Get the latest user message to use as a query for memory retrieval context_messages = context.get_messages() @@ -300,5 +296,4 @@ class Mem0MemoryService(FrameProcessor): ) await self.push_frame(frame) # Still pass the original frame through else: - # For non-context frames, just pass them through await self.push_frame(frame, direction) diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index d007615ba..63f4d7328 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -532,17 +532,11 @@ class BaseOpenAILLMService(LLMService): """ await super().process_frame(frame, direction) - context = None if isinstance(frame, LLMContextFrame): - context = frame.context - else: - await self.push_frame(frame, direction) - - if context: try: await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() - await self._process_context(context) + await self._process_context(frame.context) except httpx.TimeoutException as e: await self._call_event_handler("on_completion_timeout") await self.push_error(error_msg="LLM completion timeout", exception=e) @@ -551,3 +545,5 @@ class BaseOpenAILLMService(LLMService): finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) + else: + await self.push_frame(frame, direction) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index fce6b46d8..e1b4ace78 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -674,17 +674,11 @@ class OpenAIResponsesLLMService(_BaseOpenAIResponsesLLMService, WebsocketLLMServ """ await super().process_frame(frame, direction) - context = None if isinstance(frame, LLMContextFrame): - context = frame.context - else: - await self.push_frame(frame, direction) - - if context: try: await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() - await self._process_context(context) + await self._process_context(frame.context) except asyncio.CancelledError: # The pipeline cancelled us (e.g. due to an interruption). # Ask the server to stop generating and flag that we need @@ -717,6 +711,8 @@ class OpenAIResponsesLLMService(_BaseOpenAIResponsesLLMService, WebsocketLLMServ finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) + else: + await self.push_frame(frame, direction) # -- core inference ------------------------------------------------------- @@ -960,17 +956,11 @@ class OpenAIResponsesHttpLLMService(_BaseOpenAIResponsesLLMService): """ await super().process_frame(frame, direction) - context = None if isinstance(frame, LLMContextFrame): - context = frame.context - else: - await self.push_frame(frame, direction) - - if context: try: await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() - await self._process_context(context) + await self._process_context(frame.context) except httpx.TimeoutException as e: await self._call_event_handler("on_completion_timeout") await self.push_error(error_msg="LLM completion timeout", exception=e) @@ -979,6 +969,8 @@ class OpenAIResponsesHttpLLMService(_BaseOpenAIResponsesLLMService): finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) + else: + await self.push_frame(frame, direction) @traced_llm async def _process_context(self, context: LLMContext):