diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ad18922f..d9a1a74f1 100644 --- a/CHANGELOG.md +++ b/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`. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 44b40504e..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,7 +337,7 @@ class FunctionCallResultFrame(DataFrame): tool_call_id: str arguments: str result: Any - run_llm: bool = True + 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 039d0e6d5..ae11a94d3 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/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) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 108e5f854..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,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) diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 643d19332..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,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) diff --git a/src/pipecat/services/grok.py b/src/pipecat/services/grok.py index ba3833e7a..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,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. - 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: 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) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 84c4c0560..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,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. - 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: 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)