Merge pull request #970 from pipecat-ai/mb/user-controlled-run-llm

Add an override_run_llm option to optionally defer function call completion

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `FunctionCallResultProperties` dataclass to provide a structured way to
+  control function call behavior, including:
+
+  - `run_llm`: Controls whether to trigger LLM completion
+  - `on_context_updated`: Optional callback triggered after context update
+
 - Added a new foundational example `07e-interruptible-playht-http.py` for easy
   testing of `PlayHTHttpTTSService`.
 
@@ -30,6 +36,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Modified `OpenAIAssistantContextAggregator` to support controlled completions
+  and to emit context update callbacks via `FunctionCallResultProperties`.
+
 - Added `aws_session_token` to the `PollyTTSService`.
 
 - Changed the default model for `PlayHTHttpTTSService` to `Play3.0-mini-http`.

src/pipecat/frames/frames.py

@@ -5,7 +5,7 @@
 #
 
 from dataclasses import dataclass, field
-from typing import Any, List, Literal, Mapping, Optional, Tuple
+from typing import Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple
 
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.clocks.base_clock import BaseClock
@@ -321,6 +321,14 @@ class LLMEnablePromptCachingFrame(DataFrame):
     enable: bool
 
 
+@dataclass
+class FunctionCallResultProperties:
+    """Properties for a function call result frame."""
+
+    run_llm: Optional[bool] = None
+    on_context_updated: Optional[Callable[[], Awaitable[None]]] = None
+
+
 @dataclass
 class FunctionCallResultFrame(DataFrame):
     """A frame containing the result of an LLM function (tool) call."""
@@ -329,7 +337,7 @@ class FunctionCallResultFrame(DataFrame):
     tool_call_id: str
     arguments: str
     result: Any
-    run_llm: bool = True
+    properties: Optional[FunctionCallResultProperties] = None
 
 
 @dataclass

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -9,7 +9,7 @@ import copy
 import io
 import json
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, List
+from typing import Any, Awaitable, Callable, List, Optional
 
 from loguru import logger
 from PIL import Image
@@ -218,23 +218,22 @@ class OpenAILLMContext:
         await llm.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
 
         # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
-        async def function_call_result_callback(result):
+        async def function_call_result_callback(result, *, properties=None):
             result_frame_downstream = FunctionCallResultFrame(
                 function_name=function_name,
                 tool_call_id=tool_call_id,
                 arguments=arguments,
                 result=result,
-                run_llm=run_llm,
+                properties=properties,
             )
             result_frame_upstream = FunctionCallResultFrame(
                 function_name=function_name,
                 tool_call_id=tool_call_id,
                 arguments=arguments,
                 result=result,
-                run_llm=run_llm,
+                properties=properties,
             )
 
-            # Push frame both downstream and upstream
             await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
             await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
 

src/pipecat/services/anthropic.py

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
     Frame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    FunctionCallResultProperties,
     LLMEnablePromptCachingFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -742,6 +743,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -749,6 +751,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     assistant_message = {"role": "assistant", "content": []}
@@ -775,7 +778,12 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
                             ],
                         }
                     )
-                    run_llm = True
+                    if properties and properties.run_llm is not None:
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior
+                        run_llm = True
             elif aggregation:
                 self._context.add_message({"role": "assistant", "content": aggregation})
 
@@ -793,6 +801,10 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)

src/pipecat/services/google.py

@@ -19,6 +19,7 @@ from pipecat.frames.frames import (
     AudioRawFrame,
     ErrorFrame,
     Frame,
+    FunctionCallResultProperties,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -245,6 +246,7 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -252,6 +254,7 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     logger.debug(f"FunctionCallResultFrame result: {frame.arguments}")
@@ -282,7 +285,12 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
                             ],
                         )
                     )
-                    run_llm = not bool(self._function_calls_in_progress)
+                    if properties and properties.run_llm is not None:
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior is to run the LLM if there are no function calls in progress
+                        run_llm = not bool(self._function_calls_in_progress)
             else:
                 if aggregation.strip():
                     self._context.add_message(
@@ -303,6 +311,10 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)

src/pipecat/services/grok.py

@@ -7,9 +7,11 @@
 
 import json
 from dataclasses import dataclass
+from typing import Optional
 
 from loguru import logger
 
+from pipecat.frames.frames import FunctionCallResultProperties
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
@@ -32,6 +34,7 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -39,6 +42,7 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     # Grok requires an empty content field for function calls
@@ -65,8 +69,13 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
                             "tool_call_id": frame.tool_call_id,
                         }
                     )
-                    # Only run the LLM if there are no more function calls in progress.
+                    if properties and properties.run_llm is not None:
-                    run_llm = not bool(self._function_calls_in_progress)
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior is to run the LLM if there are no function calls in progress
+                        run_llm = not bool(self._function_calls_in_progress)
+
             else:
                 self._context.add_message({"role": "assistant", "content": aggregation})
 
@@ -84,6 +93,10 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 

src/pipecat/services/openai.py

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
     Frame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    FunctionCallResultProperties,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -549,6 +550,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             return
 
         run_llm = False
+        properties: Optional[FunctionCallResultProperties] = None
 
         aggregation = self._aggregation
         self._reset()
@@ -556,6 +558,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
         try:
             if self._function_call_result:
                 frame = self._function_call_result
+                properties = frame.properties
                 self._function_call_result = None
                 if frame.result:
                     self._context.add_message(
@@ -580,8 +583,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
                             "tool_call_id": frame.tool_call_id,
                         }
                     )
-                    # Only run the LLM if there are no more function calls in progress.
+                    if properties and properties.run_llm is not None:
-                    run_llm = not bool(self._function_calls_in_progress)
+                        # If the tool call result has a run_llm property, use it
+                        run_llm = properties.run_llm
+                    else:
+                        # Default behavior is to run the LLM if there are no function calls in progress
+                        run_llm = not bool(self._function_calls_in_progress)
+
             else:
                 self._context.add_message({"role": "assistant", "content": aggregation})
 
@@ -599,6 +607,10 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Emit the on_context_updated callback once the function call result is added to the context
+            if properties and properties.on_context_updated is not None:
+                await properties.on_context_updated()
+
             # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)