diff --git a/CHANGELOG.md b/CHANGELOG.md index db73f92f5..d9a1a74f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,10 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Added support in `FunctionCallResultFrame` for controlling LLM completions - via `override_run_llm` flag. When set to `True`, the `run_llm` parameter - determines whether a completion is triggered, allowing finer control over LLM - behavior in function calls. +- 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`. @@ -35,12 +36,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Added `aws_session_token` to the `PollyTTSService`. +- Modified `OpenAIAssistantContextAggregator` to support controlled completions + and to emit context update callbacks via `FunctionCallResultProperties`. -- Modified `OpenAIAssistantContextAggregator` to respect `override_run_llm` - flag when processing function call results. This allows external control over - whether function calls trigger LLM completions while maintaining backward - compatibility with existing code. +- Added `aws_session_token` to the `PollyTTSService`. - Changed the default model for `PlayHTHttpTTSService` to `Play3.0-mini-http`. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e60f23f96..321f8514b 100644 --- a/src/pipecat/frames/frames.py +++ b/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,8 +337,7 @@ class FunctionCallResultFrame(DataFrame): tool_call_id: str arguments: str result: Any - run_llm: bool = True - override_run_llm: bool = False + properties: Optional[FunctionCallResultProperties] = None @dataclass diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 35412f365..ae11a94d3 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -218,30 +218,20 @@ 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): - # Extract result and frame parameters if provided - if isinstance(result, dict) and "result" in result: - frame_run_llm = result.get("run_llm", run_llm) - frame_override = result.get("override_run_llm", False) - else: - frame_run_llm = run_llm - frame_override = False - + 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=frame_run_llm, - override_run_llm=frame_override, + properties=properties, ) result_frame_upstream = FunctionCallResultFrame( function_name=function_name, tool_call_id=tool_call_id, arguments=arguments, result=result, - run_llm=frame_run_llm, - override_run_llm=frame_override, + properties=properties, ) await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 5189a47d5..08bde31b7 100644 --- a/src/pipecat/services/anthropic.py +++ b/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,13 +778,11 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): ], } ) - if frame.override_run_llm: - # Explicit override - print("Explicit override") - run_llm = frame.run_llm + 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 - print("Default behavior") run_llm = True elif aggregation: self._context.add_message({"role": "assistant", "content": aggregation}) @@ -800,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) diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index c25f14a6f..341c51636 100644 --- a/src/pipecat/services/google.py +++ b/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,11 +285,11 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator): ], ) ) - if frame.override_run_llm: - # Explicit override - run_llm = frame.run_llm + 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 + # 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(): @@ -308,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) diff --git a/src/pipecat/services/grok.py b/src/pipecat/services/grok.py index e3f378a0a..7221cc09e 100644 --- a/src/pipecat/services/grok.py +++ b/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,12 +69,13 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator): "tool_call_id": frame.tool_call_id, } ) - if frame.override_run_llm: - # Explicit override - run_llm = frame.run_llm + 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 + # 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}) @@ -88,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) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index d57ea545b..4a70838f4 100644 --- a/src/pipecat/services/openai.py +++ b/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,13 +583,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): "tool_call_id": frame.tool_call_id, } ) - - if frame.override_run_llm: - # Explicit override - run_llm = frame.run_llm + 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 + # 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}) @@ -604,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)