wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor for new on_context_updated callback and new frame properties

Browse Source
This commit is contained in:
Mark Backman
2025-01-13 17:20:41 -05:00
parent 1ca6ecc46e
commit 8c0ecb89de
7 changed files with 67 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
CHANGELOG.md
View File

@@ -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`.

13
src/pipecat/frames/frames.py
View File

@@ -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

16
src/pipecat/processors/aggregators/openai_llm_context.py
View File

@@ -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)

15
src/pipecat/services/anthropic.py
View File

@@ -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)

15
src/pipecat/services/google.py
View File

@@ -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)

17
src/pipecat/services/grok.py
View File

@@ -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)

17
src/pipecat/services/openai.py
View File

@@ -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)