From a98000fd1dcf3932f4c85a54e747d3e1d1445775 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 14:57:27 -0700 Subject: [PATCH 1/8] function calling now run in tasks --- CHANGELOG.md | 19 +- src/pipecat/frames/frames.py | 21 ++ src/pipecat/pipeline/task.py | 2 +- .../processors/aggregators/llm_response.py | 232 +++++++++++------- .../aggregators/openai_llm_context.py | 65 +---- src/pipecat/services/ai_services.py | 200 ++++++++++++--- src/pipecat/services/anthropic.py | 164 ++++--------- .../services/gemini_multimodal_live/gemini.py | 9 +- src/pipecat/services/google/google.py | 143 +++++------ src/pipecat/services/grok.py | 85 +------ src/pipecat/services/openai.py | 161 ++++-------- tests/test_context_aggregators.py | 3 +- 12 files changed, 537 insertions(+), 567 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b931c083f..4762186cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- When registering a function call it is now possible to indicate if you want + the function call to be cancelled if there's a user interruption via + `cancel_on_interruption` (defaults to False). This is now possible because + function calls are executed concurrently. + - Added support for detecting idle pipelines. By default, if no activity has been detected during 5 minutes, the `PipelineTask` will be automatically cancelled. It is possible to override this behavior by passing @@ -120,6 +125,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Function calls are now executed in tasks. This means that the pipeline will + not be blocked while the function call is being executed. + - ⚠️ `PipelineTask` will now be automatically cancelled if no bot activity is happening in the pipeline. There are a few settings to configure this behavior, see `PipelineTask` documentation for more details. @@ -140,6 +148,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- Passing a `start_callback` to `LLMService.register_function()` is now + deprecated, simply move the code from the start callback to the function call. + - `TTSService` parameter `text_filter` is now deprecated, use `text_filters` instead which is now a list. This allows passing multiple filters that will be executed in order. @@ -162,6 +173,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an assistant aggregator issue that could cause assistant text to be + split into multiple chunks during function calls. + +- Fixed an assistant aggregator issue that was causing assistant text to not be + added to the context during function calls. This could lead to duplications. + - Fixed a `SegmentedSTTService` issue that was causing audio to be sent prematurely to the STT service. Instead of analyzing the volume in this service we rely on VAD events which use both VAD and volume. @@ -1978,7 +1995,7 @@ async def on_connected(processor): completed. If a task is never ran `has_finished()` will return False. - `PipelineRunner` now supports SIGTERM. If received, the runner will be - canceled. + cancelled. ### Fixed diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 74dd2accb..e589e9480 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -634,6 +634,15 @@ class FunctionCallInProgressFrame(SystemFrame): function_name: str tool_call_id: str arguments: str + cancel_on_interruption: bool + + +@dataclass +class FunctionCallCancelFrame(SystemFrame): + """A frame to signal a function call has been cancelled.""" + + function_name: str + tool_call_id: str @dataclass @@ -706,6 +715,18 @@ class VisionImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" +@dataclass +class UserImageMessageFrame(SystemFrame): + """An image associated to a user.""" + + user_image_raw_frame: UserImageRawFrame + text: Optional[str] = None + + def __str__(self): + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, image: {self.user_image_raw_frame}, text: {self.text})" + + # # Control frames # diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index cf62be64f..8279373cb 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -409,7 +409,7 @@ class PipelineTask(BaseTask): async def _process_push_queue(self): """This is the task that runs the pipeline for the first time by sending a StartFrame and by pushing any other frames queued by the user. It runs - until the tasks is canceled or stopped (e.g. with an EndFrame). + until the tasks is cancelled or stopped (e.g. with an EndFrame). """ self._clock.start() diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index aed12db9e..494d1de38 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -7,14 +7,20 @@ import asyncio import time from abc import abstractmethod -from typing import List +from typing import Dict, List + +from loguru import logger from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame, EndFrame, Frame, + FunctionCallCancelFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -23,10 +29,12 @@ from pipecat.frames.frames import ( LLMMessagesUpdateFrame, LLMSetToolsFrame, LLMTextFrame, + OpenAILLMContextAssistantTimestampFrame, StartFrame, StartInterruptionFrame, TextFrame, TranscriptionFrame, + UserImageMessageFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -35,6 +43,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.time import time_now_iso8601 class LLMFullResponseAggregator(FrameProcessor): @@ -139,68 +148,20 @@ class BaseLLMResponseAggregator(FrameProcessor): pass @abstractmethod - async def push_aggregation(self): + async def handle_aggregation(self, aggregation: str): + """Adds the given aggregation to the aggregator. The aggregator can use + a simple list of message or a context. It doesn't not push any frames. + + """ pass - -class LLMResponseAggregator(BaseLLMResponseAggregator): - """This is a base LLM aggregator that uses a simple list of messages to - store the conversation. It pushes `LLMMessagesFrame` as an aggregation - frame. - - """ - - def __init__( - self, - *, - messages: List[dict], - role: str = "user", - **kwargs, - ): - super().__init__(**kwargs) - - self._messages = messages - self._role = role - - self._aggregation = "" - - self.reset() - - @property - def messages(self) -> List[dict]: - return self._messages - - @property - def role(self) -> str: - return self._role - - def add_messages(self, messages): - self._messages.extend(messages) - - def set_messages(self, messages): - self.reset() - self._messages.clear() - self._messages.extend(messages) - - def set_tools(self, tools): - pass - - def reset(self): - self._aggregation = "" - + @abstractmethod async def push_aggregation(self): - if len(self._aggregation) > 0: - self._messages.append({"role": self._role, "content": self._aggregation}) + """Pushes the current aggregation. For example, iN the case of context + aggregation this might push a new context frame. - # Reset the aggregation. Reset it before pushing it down, otherwise - # if the tasks gets cancelled we won't be able to clear things up. - self._aggregation = "" - - frame = LLMMessagesFrame(self._messages) - await self.push_frame(frame) - - # Reset our accumulator state. - self.reset() + """ + pass class LLMContextResponseAggregator(BaseLLMResponseAggregator): @@ -247,20 +208,6 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator): def reset(self): self._aggregation = "" - async def push_aggregation(self): - if len(self._aggregation) > 0: - self._context.add_message({"role": self.role, "content": self._aggregation}) - - # Reset the aggregation. Reset it before pushing it down, otherwise - # if the tasks gets cancelled we won't be able to clear things up. - self._aggregation = "" - - frame = OpenAILLMContextFrame(self._context) - await self.push_frame(frame) - - # Reset our accumulator state. - self.reset() - class LLMUserContextAggregator(LLMContextResponseAggregator): """This is a user LLM aggregator that uses an LLM context to store the @@ -290,12 +237,13 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): self._aggregation_event = asyncio.Event() self._aggregation_task = None - self.reset() - def reset(self): super().reset() self._seen_interim_results = False + async def handle_aggregation(self, aggregation: str): + self._context.add_message({"role": self.role, "content": self._aggregation}) + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -331,6 +279,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): else: await self.push_frame(frame, direction) + async def push_aggregation(self): + if len(self._aggregation) > 0: + await self.handle_aggregation(self._aggregation) + + # Reset the aggregation. Reset it before pushing it down, otherwise + # if the tasks gets cancelled we won't be able to clear things up. + self._aggregation = "" + + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + + # Reset our accumulator state. + self.reset() + async def _start(self, frame: StartFrame): self._create_aggregation_task() @@ -424,17 +386,29 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): super().__init__(context=context, role="assistant", **kwargs) self._expect_stripped_words = expect_stripped_words - self._started = False + self._started = 0 + self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {} - self.reset() + async def handle_aggregation(self, aggregation: str): + self._context.add_message({"role": "assistant", "content": aggregation}) + + async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): + pass + + async def handle_function_call_result(self, frame: FunctionCallResultFrame): + pass + + async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): + pass + + async def handle_image_frame_message(self, frame: UserImageMessageFrame): + pass async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartInterruptionFrame): - await self.push_aggregation() - # Reset anyways - self.reset() + await self._handle_interruptions(frame) await self.push_frame(frame, direction) elif isinstance(frame, LLMFullResponseStartFrame): await self._handle_llm_start(frame) @@ -448,14 +422,104 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self.set_messages(frame.messages) elif isinstance(frame, LLMSetToolsFrame): self.set_tools(frame.tools) + elif isinstance(frame, FunctionCallInProgressFrame): + await self._handle_function_call_in_progress(frame) + elif isinstance(frame, FunctionCallResultFrame): + await self._handle_function_call_result(frame) + elif isinstance(frame, FunctionCallCancelFrame): + await self._handle_function_call_cancel(frame) + elif isinstance(frame, UserImageMessageFrame): + await self._handle_image_frame_message(frame) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.push_aggregation() else: await self.push_frame(frame, direction) + async def push_aggregation(self): + if not self._aggregation: + return + + aggregation = self._aggregation.strip() + self.reset() + + if aggregation: + await self.handle_aggregation(aggregation) + + # Push context frame + await self.push_context_frame() + + # Push timestamp frame with current time + timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) + await self.push_frame(timestamp_frame) + + async def _handle_interruptions(self, frame: StartInterruptionFrame): + await self.push_aggregation() + self._started = 0 + self.reset() + + async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): + logger.debug( + f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]" + ) + await self.handle_function_call_in_progress(frame) + self._function_calls_in_progress[frame.tool_call_id] = frame + + async def _handle_function_call_result(self, frame: FunctionCallResultFrame): + logger.debug( + f"{self} FunctionCallResultFrame: [{frame.function_name}:{frame.tool_call_id}]" + ) + if frame.tool_call_id not in self._function_calls_in_progress: + logger.warning( + f"FunctionCallResultFrame tool_call_id [{frame.tool_call_id}] is not running" + ) + return + + del self._function_calls_in_progress[frame.tool_call_id] + + properties = frame.properties + + await self.handle_function_call_result(frame) + + # Run inference if the function call result requires it. + if frame.result: + run_llm = False + + 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) + + if run_llm: + await self.push_context_frame(FrameDirection.UPSTREAM) + + # Emit the on_context_updated callback once the function call + # result is added to the context + if properties and properties.on_context_updated: + await properties.on_context_updated() + + async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame): + logger.debug( + f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]" + ) + if frame.tool_call_id not in self._function_calls_in_progress: + return + + if self._function_calls_in_progress[frame.tool_call_id].cancel_on_interruption: + await self.handle_function_call_cancel(frame) + del self._function_calls_in_progress[frame.tool_call_id] + + async def _handle_image_frame_message(self, frame: UserImageMessageFrame): + await self.handle_image_frame_message(frame) + await self.push_aggregation() + await self.push_context_frame(FrameDirection.UPSTREAM) + async def _handle_llm_start(self, _: LLMFullResponseStartFrame): - self._started = True + self._started += 1 async def _handle_llm_end(self, _: LLMFullResponseEndFrame): - self._started = False + self._started -= 1 await self.push_aggregation() async def _handle_text(self, frame: TextFrame): @@ -474,7 +538,7 @@ class LLMUserResponseAggregator(LLMUserContextAggregator): async def push_aggregation(self): if len(self._aggregation) > 0: - self._context.add_message({"role": self.role, "content": self._aggregation}) + await self.handle_aggregation(self._aggregation) # Reset the aggregation. Reset it before pushing it down, otherwise # if the tasks gets cancelled we won't be able to clear things up. @@ -493,7 +557,7 @@ class LLMAssistantResponseAggregator(LLMAssistantContextAggregator): async def push_aggregation(self): if len(self._aggregation) > 0: - self._context.add_message({"role": self.role, "content": self._aggregation}) + await self.handle_aggregation(self._aggregation) # Reset the aggregation. Reset it before pushing it down, otherwise # if the tasks gets cancelled we won't be able to clear things up. diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index e8391d62b..93c2875be 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -9,9 +9,8 @@ import copy import io import json from dataclasses import dataclass -from typing import Any, Awaitable, Callable, List, Optional +from typing import Any, List, Optional -from loguru import logger from openai._types import NOT_GIVEN, NotGiven from openai.types.chat import ( ChatCompletionMessageParam, @@ -22,12 +21,7 @@ from PIL import Image from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.frames.frames import ( - AudioRawFrame, - Frame, - FunctionCallInProgressFrame, - FunctionCallResultFrame, -) +from pipecat.frames.frames import AudioRawFrame, Frame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor # JSON custom encoder to handle bytes arrays so that we can log contexts @@ -187,61 +181,6 @@ class OpenAILLMContext: # todo: implement for OpenAI models and others pass - async def call_function( - self, - f: Callable[ - [str, str, Any, FrameProcessor, "OpenAILLMContext", Callable[[Any], Awaitable[None]]], - Awaitable[None], - ], - *, - function_name: str, - tool_call_id: str, - arguments: str, - llm: FrameProcessor, - run_llm: bool = True, - ) -> None: - logger.info(f"Calling function {function_name} with arguments {arguments}") - # Push a SystemFrame downstream. This frame will let our assistant context aggregator - # know that we are in the middle of a function call. Some contexts/aggregators may - # not need this. But some definitely do (Anthropic, for example). - # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter. - progress_frame_downstream = FunctionCallInProgressFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - ) - progress_frame_upstream = FunctionCallInProgressFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - ) - - # Push frame both downstream and upstream - await llm.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM) - 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, *, properties=None): - result_frame_downstream = FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - result=result, - properties=properties, - ) - result_frame_upstream = FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - result=result, - properties=properties, - ) - - await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM) - await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM) - - await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback) - def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size): # RIFF chunk descriptor header = bytearray() diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index e00970433..2f30f505f 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -8,7 +8,8 @@ import asyncio import io import wave from abc import abstractmethod -from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Tuple, Type +from dataclasses import dataclass +from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Set, Tuple, Type from loguru import logger @@ -22,6 +23,9 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + FunctionCallCancelFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterimTranscriptionFrame, LLMFullResponseEndFrame, StartFrame, @@ -138,6 +142,13 @@ class AIService(FrameProcessor): await self.push_frame(f) +@dataclass +class FunctionEntry: + function_name: Optional[str] + callback: Any # TODO(aleix): add proper typing. + cancel_on_interruption: bool + + class LLMService(AIService): """This class is a no-op but serves as a base class for LLM services.""" @@ -147,38 +158,74 @@ class LLMService(AIService): def __init__(self, **kwargs): super().__init__(**kwargs) - self._callbacks = {} + self._functions = {} self._start_callbacks = {} self._adapter = self.adapter_class() + self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set() + + self._register_event_handler("on_completion_timeout") def get_llm_adapter(self) -> BaseLLMAdapter: return self._adapter def create_context_aggregator( - self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True + self, + context: OpenAILLMContext, + *, + user_kwargs: Mapping[str, Any] = {}, + assistant_kwargs: Mapping[str, Any] = {}, ) -> Any: pass - self._register_event_handler("on_completion_timeout") + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) - # TODO-CB: callback function type - def register_function(self, function_name: Optional[str], callback, start_callback=None): + if isinstance(frame, StartInterruptionFrame): + await self._handle_interruptions(frame) + + async def _handle_interruptions(self, frame: StartInterruptionFrame): + for function_name, entry in self._functions.items(): + if entry.cancel_on_interruption: + await self._cancel_function_call(function_name) + + def register_function( + self, + function_name: Optional[str], + callback: Any, + start_callback=None, + *, + cancel_on_interruption: bool = False, + ): # Registering a function with the function_name set to None will run that callback # for all functions - self._callbacks[function_name] = callback - # QUESTION FOR CB: maybe this isn't needed anymore? + self._functions[function_name] = FunctionEntry( + function_name=function_name, + callback=callback, + cancel_on_interruption=cancel_on_interruption, + ) + + # Start callbacks are now deprecated. if start_callback: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter 'start_callback' is deprecated, just put your code on top of the actual function call instead.", + DeprecationWarning, + ) + self._start_callbacks[function_name] = start_callback def unregister_function(self, function_name: Optional[str]): - del self._callbacks[function_name] + del self._functions[function_name] if self._start_callbacks[function_name]: del self._start_callbacks[function_name] def has_function(self, function_name: str): - if None in self._callbacks.keys(): + if None in self._functions.keys(): return True - return function_name in self._callbacks.keys() + return function_name in self._functions.keys() async def call_function( self, @@ -188,25 +235,18 @@ class LLMService(AIService): function_name: str, arguments: str, run_llm: bool = True, - ) -> None: - f = None - if function_name in self._callbacks.keys(): - f = self._callbacks[function_name] - elif None in self._callbacks.keys(): - f = self._callbacks[None] - else: - return None - await self.call_start_function(context, function_name) - await context.call_function( - f, - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - llm=self, - run_llm=run_llm, + ): + if not function_name in self._functions.keys() and not None in self._functions.keys(): + return + + task = self.create_task( + self._run_function_call(context, tool_call_id, function_name, arguments, run_llm) ) - # QUESTION FOR CB: maybe this isn't needed anymore? + self._function_call_tasks.add((task, tool_call_id, function_name)) + + task.add_done_callback(self._function_call_task_finished) + async def call_start_function(self, context: OpenAILLMContext, function_name: str): if function_name in self._start_callbacks.keys(): await self._start_callbacks[function_name](function_name, self, context) @@ -218,6 +258,106 @@ class LLMService(AIService): UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM ) + async def _run_function_call( + self, + context: OpenAILLMContext, + tool_call_id: str, + function_name: str, + arguments: str, + run_llm: bool = True, + ): + if function_name in self._functions.keys(): + entry = self._functions[function_name] + elif None in self._functions.keys(): + entry = self._functions[None] + else: + return + + logger.info(f"Calling function {function_name} with arguments {arguments}") + + # NOTE(aleix): This needs to be removed after we remove the deprecation. + await self.call_start_function(context, function_name) + + # Push a SystemFrame downstream. This frame will let our assistant context aggregator + # know that we are in the middle of a function call. Some contexts/aggregators may + # not need this. But some definitely do (Anthropic, for example). + # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter. + progress_frame_downstream = FunctionCallInProgressFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + cancel_on_interruption=entry.cancel_on_interruption, + ) + progress_frame_upstream = FunctionCallInProgressFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + cancel_on_interruption=entry.cancel_on_interruption, + ) + + # Push frame both downstream and upstream + await self.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM) + await self.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM) + + # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream. + 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, + properties=properties, + ) + result_frame_upstream = FunctionCallResultFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + result=result, + properties=properties, + ) + + await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM) + await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM) + + await entry.callback( + function_name, tool_call_id, arguments, self, context, function_call_result_callback + ) + + async def _cancel_function_call(self, function_name: str): + cancelled_tasks = set() + for task, tool_call_id, name in self._function_call_tasks: + if name == function_name: + # We remove the callback because we are going to cancel the task + # now, otherwise we will be removing it from the set while we + # are iterating. + task.remove_done_callback(self._function_call_task_finished) + + logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...") + + await self.cancel_task(task) + + frame = FunctionCallCancelFrame( + function_name=function_name, tool_call_id=tool_call_id + ) + await self.push_frame(frame) + + logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled") + + cancelled_tasks.add(task) + + # Remove all cancelled tasks from our set. + for task in cancelled_tasks: + self._function_call_task_finished(task) + + def _function_call_task_finished(self, task: asyncio.Task): + tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None) + if tuple_to_remove: + self._function_call_tasks.discard(tuple_to_remove) + # The task is finished so this should exit immediately. We need to + # do this because otherwise the task manager would have a dangling + # task if we don't remove it. + asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop()) + class TTSService(AIService): def __init__( @@ -366,12 +506,14 @@ class TTSService(AIService): else: await self.push_frame(frame, direction) elif isinstance(frame, TTSSpeakFrame): + # Store if we were processing text or not so we can set it back. + processing_text = self._processing_text await self._push_tts_frames(frame.text) # We pause processing incoming frames because we are sending data to # the TTS. We pause to avoid audio overlapping. await self._maybe_pause_frame_processing() await self.flush_audio() - self._processing_text = False + self._processing_text = processing_text elif isinstance(frame, TTSUpdateSettingsFrame): await self._update_settings(frame.settings) elif isinstance(frame, BotStoppedSpeakingFrame): diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 10a2ab7b7..8e06f4558 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -21,17 +21,16 @@ from pydantic import BaseModel, Field from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter from pipecat.frames.frames import ( Frame, + FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, - FunctionCallResultProperties, LLMEnablePromptCachingFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, LLMTextFrame, LLMUpdateSettingsFrame, - OpenAILLMContextAssistantTimestampFrame, - StartInterruptionFrame, + UserImageMessageFrame, UserImageRawFrame, UserImageRequestFrame, VisionImageRawFrame, @@ -47,7 +46,6 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import LLMService -from pipecat.utils.time import time_now_iso8601 try: from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven @@ -60,13 +58,6 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -# internal use only -- todo: refactor -@dataclass -class AnthropicImageMessageFrame(Frame): - user_image_raw_frame: UserImageRawFrame - text: Optional[str] = None - - @dataclass class AnthropicContextAggregatorPair: _user: "AnthropicUserContextAggregator" @@ -715,7 +706,7 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator): text = self._context._user_image_request_context.get(frame.user_id) or "" if text: del self._context._user_image_request_context[frame.user_id] - frame = AnthropicImageMessageFrame(user_image_raw_frame=frame, text=text) + frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text) await self.push_frame(frame) except Exception as e: logger.error(f"Error processing frame: {e}") @@ -734,110 +725,61 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator): class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs): super().__init__(context=context, **kwargs) - self._function_call_in_progress = None - self._function_call_result = None - self._pending_image_frame_message = None - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - # See note above about not calling push_frame() here. - if isinstance(frame, StartInterruptionFrame): - self._function_call_in_progress = None - self._function_call_finished = None - elif isinstance(frame, FunctionCallInProgressFrame): - self._function_call_in_progress = frame - elif isinstance(frame, FunctionCallResultFrame): - if ( - self._function_call_in_progress - and self._function_call_in_progress.tool_call_id == frame.tool_call_id - ): - self._function_call_in_progress = None - self._function_call_result = frame - await self.push_aggregation() - else: - logger.warning( - "FunctionCallResultFrame tool_call_id != InProgressFrame tool_call_id" - ) - self._function_call_in_progress = None - self._function_call_result = None - elif isinstance(frame, AnthropicImageMessageFrame): - self._pending_image_frame_message = frame - await self.push_aggregation() + async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): + assistant_message = {"role": "assistant", "content": []} + assistant_message["content"].append( + { + "type": "tool_use", + "id": frame.tool_call_id, + "name": frame.function_name, + "input": frame.arguments, + } + ) + self._context.add_message(assistant_message) + self._context.add_message( + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": frame.tool_call_id, + "content": "IN_PROGRESS", + } + ], + } + ) - async def push_aggregation(self): - if not ( - self._aggregation or self._function_call_result or self._pending_image_frame_message - ): + async def handle_function_call_result(self, frame: FunctionCallResultFrame): + if not frame.result: return - run_llm = False - properties: Optional[FunctionCallResultProperties] = None + result = json.dumps(frame.result) - aggregation = self._aggregation.strip() - self.reset() + await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) - try: - if aggregation: - self._context.add_message({"role": "assistant", "content": aggregation}) + async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, "CANCELLED" + ) - 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": []} - assistant_message["content"].append( - { - "type": "tool_use", - "id": frame.tool_call_id, - "name": frame.function_name, - "input": frame.arguments, - } - ) - self._context.add_message(assistant_message) - self._context.add_message( - { - "role": "user", - "content": [ - { - "type": "tool_result", - "tool_use_id": frame.tool_call_id, - "content": json.dumps(frame.result), - } - ], - } - ) - 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 + async def _update_function_call_result( + self, function_name: str, tool_call_id: str, result: str + ): + for message in self._context.messages: + if message["role"] == "user": + for content in message["content"]: + if ( + isinstance(content, dict) + and content["type"] == "tool_result" + and content["tool_use_id"] == tool_call_id + ): + content["content"] = result - if self._pending_image_frame_message: - frame = self._pending_image_frame_message - self._pending_image_frame_message = None - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, - ) - run_llm = True - - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - # 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 - await self.push_context_frame() - - # Push timestamp frame with current time - timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) - await self.push_frame(timestamp_frame) - - except Exception as e: - logger.error(f"Error processing frame: {e}") + async def handle_image_frame_message(self, frame: UserImageMessageFrame): + self._context.add_image_frame_message( + format=frame.user_image_raw_frame.format, + size=frame.user_image_raw_frame.size, + image=frame.user_image_raw_frame.image, + text=frame.text, + ) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 9484fc27d..da1697da6 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -39,6 +39,7 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, TTSTextFrame, + UserImageMessageFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -118,10 +119,10 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator): class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator): - async def push_aggregation(self): - # We don't want to store any images in the context. Revisit this later when the API evolves. - self._pending_image_frame_message = None - await super().push_aggregation() + async def handle_image_frame_message(self, frame: UserImageMessageFrame): + # We don't want to store any images in the context. Revisit this later + # when the API evolves. + pass @dataclass diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index c10e67531..c07707899 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -10,6 +10,7 @@ import io import json import os import time +import uuid from google.api_core.exceptions import DeadlineExceeded from openai import AsyncStream @@ -33,20 +34,22 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, - FunctionCallResultProperties, + FunctionCallCancelFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, LLMTextFrame, LLMUpdateSettingsFrame, - OpenAILLMContextAssistantTimestampFrame, StartFrame, TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, URLImageRawFrame, + UserImageMessageFrame, VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage @@ -565,91 +568,69 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator): class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator): - async def push_aggregation(self): - if not ( - self._aggregation or self._function_call_result or self._pending_image_frame_message - ): + async def handle_aggregation(self, aggregation: str): + self._context.add_message(glm.Content(role="model", parts=[glm.Part(text=aggregation)])) + + async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): + self._context.add_message( + glm.Content( + role="model", + parts=[ + glm.Part( + function_call=glm.FunctionCall( + id=frame.tool_call_id, name=frame.function_name, args=frame.arguments + ) + ) + ], + ) + ) + self._context.add_message( + glm.Content( + role="user", + parts=[ + glm.Part( + function_response=glm.FunctionResponse( + id=frame.tool_call_id, + name=frame.function_name, + response={"response": "IN_PROGRESS"}, + ) + ) + ], + ) + ) + + async def handle_function_call_result(self, frame: FunctionCallResultFrame): + if not frame.result: return - run_llm = False - properties: Optional[FunctionCallResultProperties] = None + if not isinstance(frame.result, str): + return - aggregation = self._aggregation.strip() - self.reset() + response = {"response": frame.result} - try: - if aggregation: - self._context.add_message( - glm.Content(role="model", parts=[glm.Part(text=aggregation)]) - ) + await self._update_function_call_result(frame.function_name, frame.tool_call_id, response) - 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}") - self._context.add_message( - glm.Content( - role="model", - parts=[ - glm.Part( - function_call=glm.FunctionCall( - name=frame.function_name, args=frame.arguments - ) - ) - ], - ) - ) - response = frame.result - if isinstance(response, str): - response = {"response": response} - self._context.add_message( - glm.Content( - role="user", - parts=[ - glm.Part( - function_response=glm.FunctionResponse( - name=frame.function_name, response=response - ) - ) - ], - ) - ) - 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) + async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, "CANCELLED" + ) - if self._pending_image_frame_message: - frame = self._pending_image_frame_message - self._pending_image_frame_message = None - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, - ) - run_llm = True + async def _update_function_call_result( + self, function_name: str, tool_call_id: str, result: Any + ): + for message in self._context.messages: + if message.role == "user": + for part in message.parts: + if part.function_response and part.function_response.id == tool_call_id: + part.function_response.response = result - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - # 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 - await self.push_context_frame() - - # Push timestamp frame with current time - timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) - await self.push_frame(timestamp_frame) - - except Exception as e: - logger.exception(f"Error processing frame: {e}") + async def handle_image_frame_message(self, frame: UserImageMessageFrame): + self._context.add_image_frame_message( + format=frame.user_image_raw_frame.format, + size=frame.user_image_raw_frame.size, + image=frame.user_image_raw_frame.image, + text=frame.text, + ) @dataclass @@ -1071,7 +1052,7 @@ class GoogleLLMService(LLMService): args = type(c.function_call).to_dict(c.function_call).get("args", {}) await self.call_function( context=context, - tool_call_id="what_should_this_be", + tool_call_id=str(uuid.uuid4()), function_name=c.function_call.name, arguments=args, ) diff --git a/src/pipecat/services/grok.py b/src/pipecat/services/grok.py index cf7d74f59..faed13050 100644 --- a/src/pipecat/services/grok.py +++ b/src/pipecat/services/grok.py @@ -25,94 +25,15 @@ from pipecat.services.openai import ( ) -class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator): - """Custom assistant context aggregator for Grok that handles empty content requirement.""" - - async def push_aggregation(self): - if not ( - self._aggregation or self._function_call_result or self._pending_image_frame_message - ): - return - - run_llm = False - properties: Optional[FunctionCallResultProperties] = None - - aggregation = self._aggregation.strip() - self.reset() - - try: - if aggregation: - self._context.add_message({"role": "assistant", "content": aggregation}) - - 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 - self._context.add_message( - { - "role": "assistant", - "content": "", # Required by Grok - "tool_calls": [ - { - "id": frame.tool_call_id, - "function": { - "name": frame.function_name, - "arguments": json.dumps(frame.arguments), - }, - "type": "function", - } - ], - } - ) - self._context.add_message( - { - "role": "tool", - "content": json.dumps(frame.result), - "tool_call_id": frame.tool_call_id, - } - ) - 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) - - if self._pending_image_frame_message: - frame = self._pending_image_frame_message - self._pending_image_frame_message = None - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, - ) - run_llm = True - - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - # 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() - - await self.push_context_frame() - - except Exception as e: - logger.error(f"Error processing frame: {e}") - - @dataclass class GrokContextAggregatorPair: _user: "OpenAIUserContextAggregator" - _assistant: "GrokAssistantContextAggregator" + _assistant: "OpenAIAssistantContextAggregator" def user(self) -> "OpenAIUserContextAggregator": return self._user - def assistant(self) -> "GrokAssistantContextAggregator": + def assistant(self) -> "OpenAIAssistantContextAggregator": return self._assistant @@ -235,5 +156,5 @@ class GrokLLMService(OpenAILLMService): context.set_llm_adapter(self.get_llm_adapter()) user = OpenAIUserContextAggregator(context, **user_kwargs) - assistant = GrokAssistantContextAggregator(context, **assistant_kwargs) + assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs) return GrokContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 89bf4a04b..700f1c0b5 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -27,21 +27,20 @@ from pydantic import BaseModel, Field from pipecat.frames.frames import ( ErrorFrame, Frame, + FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, - FunctionCallResultProperties, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, LLMTextFrame, LLMUpdateSettingsFrame, - OpenAILLMContextAssistantTimestampFrame, StartFrame, - StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, URLImageRawFrame, + UserImageMessageFrame, UserImageRawFrame, UserImageRequestFrame, VisionImageRawFrame, @@ -63,7 +62,6 @@ from pipecat.services.ai_services import ( ) from pipecat.services.base_whisper import BaseWhisperSTTService, Transcription from pipecat.transcriptions.language import Language -from pipecat.utils.time import time_now_iso8601 ValidVoice = Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"] @@ -558,13 +556,6 @@ class OpenAITTSService(TTSService): logger.exception(f"{self} error generating TTS: {e}") -# internal use only -- todo: refactor -@dataclass -class OpenAIImageMessageFrame(Frame): - user_image_raw_frame: UserImageRawFrame - text: Optional[str] = None - - class OpenAIUserContextAggregator(LLMUserContextAggregator): def __init__(self, context: OpenAILLMContext, **kwargs): super().__init__(context=context, **kwargs) @@ -596,7 +587,7 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator): text = self._context._user_image_request_context.get(frame.user_id) or "" if text: del self._context._user_image_request_context[frame.user_id] - frame = OpenAIImageMessageFrame(user_image_raw_frame=frame, text=text) + frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text) await self.push_frame(frame) except Exception as e: logger.error(f"Error processing frame: {e}") @@ -605,109 +596,59 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator): class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): def __init__(self, context: OpenAILLMContext, **kwargs): super().__init__(context=context, **kwargs) - self._function_calls_in_progress = {} - self._function_call_result = None - self._pending_image_frame_message = None - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - # See note above about not calling push_frame() here. - if isinstance(frame, StartInterruptionFrame): - self._function_calls_in_progress.clear() - self._function_call_finished = None - elif isinstance(frame, FunctionCallInProgressFrame): - logger.debug(f"FunctionCallInProgressFrame: {frame}") - self._function_calls_in_progress[frame.tool_call_id] = frame - elif isinstance(frame, FunctionCallResultFrame): - logger.debug(f"FunctionCallResultFrame: {frame}") - if frame.tool_call_id in self._function_calls_in_progress: - del self._function_calls_in_progress[frame.tool_call_id] - self._function_call_result = frame - # TODO-CB: Kwin wants us to refactor this out of here but I REFUSE - await self.push_aggregation() - else: - logger.warning( - "FunctionCallResultFrame tool_call_id does not match any function call in progress" - ) - self._function_call_result = None - elif isinstance(frame, OpenAIImageMessageFrame): - self._pending_image_frame_message = frame - await self.push_aggregation() + async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): + self._context.add_message( + { + "role": "assistant", + "tool_calls": [ + { + "id": frame.tool_call_id, + "function": { + "name": frame.function_name, + "arguments": json.dumps(frame.arguments), + }, + "type": "function", + } + ], + } + ) + self._context.add_message( + { + "role": "tool", + "content": "IN_PROGRESS", + "tool_call_id": frame.tool_call_id, + } + ) - async def push_aggregation(self): - if not ( - self._aggregation or self._function_call_result or self._pending_image_frame_message - ): + async def handle_function_call_result(self, frame: FunctionCallResultFrame): + if not frame.result: return - run_llm = False - properties: Optional[FunctionCallResultProperties] = None + result = json.dumps(frame.result) - aggregation = self._aggregation.strip() - self.reset() + await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) - try: - if aggregation: - self._context.add_message({"role": "assistant", "content": aggregation}) + async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, "CANCELLED" + ) - 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( - { - "role": "assistant", - "tool_calls": [ - { - "id": frame.tool_call_id, - "function": { - "name": frame.function_name, - "arguments": json.dumps(frame.arguments), - }, - "type": "function", - } - ], - } - ) - self._context.add_message( - { - "role": "tool", - "content": json.dumps(frame.result), - "tool_call_id": frame.tool_call_id, - } - ) - 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) + async def _update_function_call_result( + self, function_name: str, tool_call_id: str, result: str + ): + for message in self._context.messages: + if ( + message["role"] == "tool" + and message["tool_call_id"] + and message["tool_call_id"] == tool_call_id + ): + message["content"] = result - if self._pending_image_frame_message: - frame = self._pending_image_frame_message - self._pending_image_frame_message = None - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, - ) - run_llm = True - - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - # 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 - await self.push_context_frame() - - # Push timestamp frame with current time - timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) - await self.push_frame(timestamp_frame) - - except Exception as e: - logger.error(f"Error processing frame: {e}") + async def handle_image_frame_message(self, frame: UserImageMessageFrame): + self._context.add_image_frame_message( + format=frame.user_image_raw_frame.format, + size=frame.user_image_raw_frame.size, + image=frame.user_image_raw_frame.image, + text=frame.text, + ) diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index d4b8c35ce..0c9b6d5e4 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -418,7 +418,7 @@ class BaseTestUserContextAggregator: class BaseTestAssistantContextAggreagator: CONTEXT_CLASS = None # To be set in subclasses AGGREGATOR_CLASS = None # To be set in subclasses - EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame] + EXPECTED_CONTEXT_FRAMES = None # To be set in subclasses def check_message_content(self, context: OpenAILLMContext, index: int, content: str): assert context.messages[index]["content"] == content @@ -577,6 +577,7 @@ class TestLLMAssistantContextAggregator( ): CONTEXT_CLASS = OpenAILLMContext AGGREGATOR_CLASS = LLMAssistantContextAggregator + EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame] # From c15286b14836e35fbbf94eb5e2f39ce38761f79a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 14:57:49 -0700 Subject: [PATCH 2/8] examples: deprecate start_callback from LLMService.register_function() --- examples/foundational/14-function-calling.py | 12 ++---- .../14c-function-calling-together.py | 11 ++--- .../14e-function-calling-gemini.py | 9 +---- .../foundational/14f-function-calling-groq.py | 11 ++--- .../foundational/14g-function-calling-grok.py | 11 +---- .../14h-function-calling-azure.py | 11 ++--- .../14i-function-calling-fireworks.py | 13 ++---- .../foundational/14j-function-calling-nim.py | 11 ++--- .../14k-function-calling-cerebras.py | 11 ++--- .../14l-function-calling-deepseek.py | 11 ++--- .../14m-function-calling-openrouter.py | 11 ++--- ...o-function-calling-gemini-openai-format.py | 13 ++---- .../14p-function-calling-gemini-vertex-ai.py | 13 ++---- .../22b-natural-conversation-proposal.py | 11 ++--- .../22c-natural-conversation-mixed-llms.py | 9 +---- examples/foundational/24-stt-mute-filter.py | 6 +-- examples/patient-intake/bot.py | 40 ++++++++++--------- 17 files changed, 67 insertions(+), 147 deletions(-) diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index 1d4e37344..dddfed1d2 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -30,13 +30,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -62,9 +57,10 @@ async def main(): ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - # Register a function_name of None to get all functions + + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14c-function-calling-together.py b/examples/foundational/14c-function-calling-together.py index dd3eb714f..94d587799 100644 --- a/examples/foundational/14c-function-calling-together.py +++ b/examples/foundational/14c-function-calling-together.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -66,9 +61,9 @@ async def main(): api_key=os.getenv("TOGETHER_API_KEY"), model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo", ) - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index bf022a65a..f0003fbab 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/examples/foundational/14e-function-calling-gemini.py @@ -33,13 +33,8 @@ logger.add(sys.stderr, level="DEBUG") video_participant_id = None -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) location = arguments["location"] await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") @@ -72,7 +67,7 @@ async def main(): ) llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001") - llm.register_function("get_weather", get_weather, start_fetch_weather) + llm.register_function("get_weather", get_weather) llm.register_function("get_image", get_image) weather_function = FunctionSchema( diff --git a/examples/foundational/14f-function-calling-groq.py b/examples/foundational/14f-function-calling-groq.py index c402ac91a..9818d185f 100644 --- a/examples/foundational/14f-function-calling-groq.py +++ b/examples/foundational/14f-function-calling-groq.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -65,9 +60,9 @@ async def main(): ) llm = GroqLLMService(api_key=os.getenv("GROQ_API_KEY"), model="llama-3.3-70b-versatile") - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14g-function-calling-grok.py b/examples/foundational/14g-function-calling-grok.py index e6b822e39..e5bb2f9b3 100644 --- a/examples/foundational/14g-function-calling-grok.py +++ b/examples/foundational/14g-function-calling-grok.py @@ -16,7 +16,6 @@ from runner import configure from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -31,12 +30,6 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): await result_callback({"conditions": "nice", "temperature": "75"}) @@ -63,9 +56,9 @@ async def main(): ) llm = GrokLLMService(api_key=os.getenv("GROK_API_KEY")) - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14h-function-calling-azure.py b/examples/foundational/14h-function-calling-azure.py index aefa1a474..6c8465832 100644 --- a/examples/foundational/14h-function-calling-azure.py +++ b/examples/foundational/14h-function-calling-azure.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -67,9 +62,9 @@ async def main(): endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL"), ) - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14i-function-calling-fireworks.py b/examples/foundational/14i-function-calling-fireworks.py index 50291615b..887e734dd 100644 --- a/examples/foundational/14i-function-calling-fireworks.py +++ b/examples/foundational/14i-function-calling-fireworks.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -64,11 +59,11 @@ async def main(): llm = FireworksLLMService( api_key=os.getenv("FIREWORKS_API_KEY"), - model="accounts/fireworks/models/firefunction-v2", + model="accounts/fireworks/models/llama-v3p1-405b-instruct", ) - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14j-function-calling-nim.py b/examples/foundational/14j-function-calling-nim.py index ea8e25cf6..c628bedf0 100644 --- a/examples/foundational/14j-function-calling-nim.py +++ b/examples/foundational/14j-function-calling-nim.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -66,9 +61,9 @@ async def main(): llm = NimLLMService( api_key=os.getenv("NVIDIA_API_KEY"), model="meta/llama-3.3-70b-instruct" ) - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14k-function-calling-cerebras.py b/examples/foundational/14k-function-calling-cerebras.py index 9b3641307..541e6f250 100644 --- a/examples/foundational/14k-function-calling-cerebras.py +++ b/examples/foundational/14k-function-calling-cerebras.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -63,9 +58,9 @@ async def main(): ) llm = CerebrasLLMService(api_key=os.getenv("CEREBRAS_API_KEY"), model="llama-3.3-70b") - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14l-function-calling-deepseek.py b/examples/foundational/14l-function-calling-deepseek.py index 0360f0b84..812a6716e 100644 --- a/examples/foundational/14l-function-calling-deepseek.py +++ b/examples/foundational/14l-function-calling-deepseek.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -63,9 +58,9 @@ async def main(): ) llm = DeepSeekLLMService(api_key=os.getenv("DEEPSEEK_API_KEY"), model="deepseek-chat") - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14m-function-calling-openrouter.py b/examples/foundational/14m-function-calling-openrouter.py index e816f6322..a675c8ac4 100644 --- a/examples/foundational/14m-function-calling-openrouter.py +++ b/examples/foundational/14m-function-calling-openrouter.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -67,9 +62,9 @@ async def main(): llm = OpenRouterLLMService( api_key=os.getenv("OPENROUTER_API_KEY"), model="openai/gpt-4o-2024-11-20" ) - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14o-function-calling-gemini-openai-format.py b/examples/foundational/14o-function-calling-gemini-openai-format.py index a1bb53b6b..78100f0d3 100644 --- a/examples/foundational/14o-function-calling-gemini-openai-format.py +++ b/examples/foundational/14o-function-calling-gemini-openai-format.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -63,11 +58,9 @@ async def main(): ) llm = GoogleLLMOpenAIBetaService(api_key=os.getenv("GEMINI_API_KEY")) - # Register a function_name of None to get all functions + # You can aslo register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function( - "get_current_weather", fetch_weather_from_api, start_callback=start_fetch_weather - ) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/14p-function-calling-gemini-vertex-ai.py b/examples/foundational/14p-function-calling-gemini-vertex-ai.py index d64bae89d..8888a2922 100644 --- a/examples/foundational/14p-function-calling-gemini-vertex-ai.py +++ b/examples/foundational/14p-function-calling-gemini-vertex-ai.py @@ -31,13 +31,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -68,11 +63,9 @@ async def main(): project_id="", ) ) - # Register a function_name of None to get all functions + # You can aslo register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function( - "get_current_weather", fetch_weather_from_api, start_callback=start_fetch_weather - ) + llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( name="get_current_weather", diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index c97289217..d7f7eb597 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -199,13 +199,8 @@ class OutputGate(FrameProcessor): break -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -239,9 +234,9 @@ async def main(): # This is the regular LLM. llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - # Register a function_name of None to get all functions + # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) tools = [ ChatCompletionToolParam( diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index d342a2765..7fa1bb5fa 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -403,13 +403,8 @@ class OutputGate(FrameProcessor): break -async def start_fetch_weather(function_name, llm, context): - """Push a frame to the LLM; this is handy when the LLM response might take a while.""" - await llm.push_frame(TTSSpeakFrame("Let me check on that.")) - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + await llm.push_frame(TTSSpeakFrame("Let me check on that.")) await result_callback({"conditions": "nice", "temperature": "75"}) @@ -451,7 +446,7 @@ async def main(): ) # Register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) tools = [ ChatCompletionToolParam( diff --git a/examples/foundational/24-stt-mute-filter.py b/examples/foundational/24-stt-mute-filter.py index 7ae6c73fa..d2b7174d1 100644 --- a/examples/foundational/24-stt-mute-filter.py +++ b/examples/foundational/24-stt-mute-filter.py @@ -30,10 +30,6 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(function_name, llm, context): - logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}") - - async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): # Add a delay to test interruption during function calls logger.info("Weather API call starting...") @@ -72,7 +68,7 @@ async def main(): tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather) + llm.register_function("get_current_weather", fetch_weather_from_api) tools = [ ChatCompletionToolParam( diff --git a/examples/patient-intake/bot.py b/examples/patient-intake/bot.py index b6f7fb941..de6ad2051 100644 --- a/examples/patient-intake/bot.py +++ b/examples/patient-intake/bot.py @@ -142,7 +142,9 @@ class IntakeProcessor: ] ) - async def start_prescriptions(self, function_name, llm, context): + async def list_prescriptions( + self, function_name, tool_call_id, args, llm, context, result_callback + ): print(f"!!! doing start prescriptions") # Move on to allergies context.set_tools( @@ -182,9 +184,12 @@ class IntakeProcessor: print(f"!!! about to await llm process frame in start prescrpitions") await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) print(f"!!! past await process frame in start prescriptions") + await self.save_data(args, result_callback) - async def start_allergies(self, function_name, llm, context): - print("!!! doing start allergies") + async def list_allergies( + self, function_name, tool_call_id, args, llm, context, result_callback + ): + print("!!! doing list allergies") # Move on to conditions context.set_tools( [ @@ -221,8 +226,11 @@ class IntakeProcessor: } ) await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await self.save_data(args, result_callback) - async def start_conditions(self, function_name, llm, context): + async def list_conditions( + self, function_name, tool_call_id, args, llm, context, result_callback + ): print("!!! doing start conditions") # Move on to visit reasons context.set_tools( @@ -260,8 +268,11 @@ class IntakeProcessor: } ) await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await self.save_data(args, result_callback) - async def start_visit_reasons(self, function_name, llm, context): + async def list_visit_reasons( + self, function_name, tool_call_id, args, llm, context, result_callback + ): print("!!! doing start visit reasons") # move to finish call context.set_tools([]) @@ -269,8 +280,9 @@ class IntakeProcessor: {"role": "system", "content": "Now, thank the user and end the conversation."} ) await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await self.save_data(args, result_callback) - async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback): + async def save_data(self, args, result_callback): logger.info(f"!!! Saving data: {args}") # Since this is supposed to be "async", returning None from the callback # will prevent adding anything to context or re-prompting @@ -319,18 +331,10 @@ async def main(): intake = IntakeProcessor(context) llm.register_function("verify_birthday", intake.verify_birthday) - llm.register_function( - "list_prescriptions", intake.save_data, start_callback=intake.start_prescriptions - ) - llm.register_function( - "list_allergies", intake.save_data, start_callback=intake.start_allergies - ) - llm.register_function( - "list_conditions", intake.save_data, start_callback=intake.start_conditions - ) - llm.register_function( - "list_visit_reasons", intake.save_data, start_callback=intake.start_visit_reasons - ) + llm.register_function("list_prescriptions", intake.list_prescriptions) + llm.register_function("list_allergies", intake.list_allergies) + llm.register_function("list_conditions", intake.list_conditions) + llm.register_function("list_visit_reasons", intake.list_visit_reasons) fl = FrameLogger("LLM Output") From d1550d5a854ca1617068a67d368e28bfedb00b5d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 15:52:37 -0700 Subject: [PATCH 3/8] tests: remove TestFrameProcessor, reimplement with run_test() --- src/pipecat/tests/utils.py | 42 +++--- src/pipecat/utils/test_frame_processor.py | 48 ------- tests/integration/integration_azure_llm.py | 33 ----- tests/integration/integration_ollama_llm.py | 28 ---- tests/integration/integration_openai_llm.py | 128 ------------------ ...st_integration_unified_function_calling.py | 27 ++-- 6 files changed, 37 insertions(+), 269 deletions(-) delete mode 100644 src/pipecat/utils/test_frame_processor.py delete mode 100644 tests/integration/integration_azure_llm.py delete mode 100644 tests/integration/integration_ollama_llm.py delete mode 100644 tests/integration/integration_openai_llm.py diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index d68c87d88..e2368ba09 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -6,7 +6,7 @@ import asyncio from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Dict, Sequence, Tuple +from typing import Any, Awaitable, Callable, Dict, Optional, Sequence, Tuple from pipecat.frames.frames import ( EndFrame, @@ -80,8 +80,8 @@ async def run_test( processor: FrameProcessor, *, frames_to_send: Sequence[Frame], - expected_down_frames: Sequence[type], - expected_up_frames: Sequence[type] = [], + expected_down_frames: Optional[Sequence[type]] = None, + expected_up_frames: Optional[Sequence[type]] = None, ignore_start: bool = True, start_metadata: Dict[str, Any] = {}, send_end_frame: bool = True, @@ -126,33 +126,35 @@ async def run_test( # Down frames # received_down_frames: Sequence[Frame] = [] - while not received_down.empty(): - frame = await received_down.get() - if not isinstance(frame, EndFrame) or not send_end_frame: - received_down_frames.append(frame) + if expected_down_frames is not None: + while not received_down.empty(): + frame = await received_down.get() + if not isinstance(frame, EndFrame) or not send_end_frame: + received_down_frames.append(frame) - print("received DOWN frames =", received_down_frames) - print("expected DOWN frames =", expected_down_frames) + print("received DOWN frames =", received_down_frames) + print("expected DOWN frames =", expected_down_frames) - assert len(received_down_frames) == len(expected_down_frames) + assert len(received_down_frames) == len(expected_down_frames) - for real, expected in zip(received_down_frames, expected_down_frames): - assert isinstance(real, expected) + for real, expected in zip(received_down_frames, expected_down_frames): + assert isinstance(real, expected) # # Up frames # received_up_frames: Sequence[Frame] = [] - while not received_up.empty(): - frame = await received_up.get() - received_up_frames.append(frame) + if expected_up_frames is not None: + while not received_up.empty(): + frame = await received_up.get() + received_up_frames.append(frame) - print("received UP frames =", received_up_frames) - print("expected UP frames =", expected_up_frames) + print("received UP frames =", received_up_frames) + print("expected UP frames =", expected_up_frames) - assert len(received_up_frames) == len(expected_up_frames) + assert len(received_up_frames) == len(expected_up_frames) - for real, expected in zip(received_up_frames, expected_up_frames): - assert isinstance(real, expected) + for real, expected in zip(received_up_frames, expected_up_frames): + assert isinstance(real, expected) return (received_down_frames, received_up_frames) diff --git a/src/pipecat/utils/test_frame_processor.py b/src/pipecat/utils/test_frame_processor.py deleted file mode 100644 index b35864497..000000000 --- a/src/pipecat/utils/test_frame_processor.py +++ /dev/null @@ -1,48 +0,0 @@ -from typing import List - -from pipecat.processors.frame_processor import FrameProcessor - - -class TestException(Exception): - pass - - -class TestFrameProcessor(FrameProcessor): - __test__ = False # Prevents pytest from collecting this class as a test - - def __init__(self, test_frames): - self.test_frames = test_frames - self._list_counter = 0 - super().__init__() - - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - - if not self.test_frames[ - 0 - ]: # then we've run out of required frames but the generator is still going? - raise TestException(f"Oops, got an extra frame, {frame}") - if isinstance(self.test_frames[0], List): - # We need to consume frames until we see the next frame type after this - next_frame = self.test_frames[1] - if isinstance(frame, next_frame): - # we're done iterating the list I guess - print(f"TestFrameProcessor got expected list exit frame: {frame}") - # pop twice to get rid of the list, as well as the next frame - self.test_frames.pop(0) - self.test_frames.pop(0) - self.list_counter = 0 - else: - fl = self.test_frames[0] - fl_el = fl[self._list_counter % len(fl)] - if isinstance(frame, fl_el): - print(f"TestFrameProcessor got expected list frame: {frame}") - self._list_counter += 1 - else: - raise TestException(f"Inside a list, expected {fl_el} but got {frame}") - - else: - if not isinstance(frame, self.test_frames[0]): - raise TestException(f"Expected {self.test_frames[0]}, but got {frame}") - print(f"TestFrameProcessor got expected frame: {frame}") - self.test_frames.pop(0) diff --git a/tests/integration/integration_azure_llm.py b/tests/integration/integration_azure_llm.py deleted file mode 100644 index 8e49e9d04..000000000 --- a/tests/integration/integration_azure_llm.py +++ /dev/null @@ -1,33 +0,0 @@ -import asyncio -import os -import unittest - -from openai.types.chat import ( - ChatCompletionSystemMessageParam, -) - -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) -from pipecat.services.azure import AzureLLMService - -if __name__ == "__main__": - - @unittest.skip("Skip azure integration test") - async def test_chat(): - llm = AzureLLMService( - api_key=os.getenv("AZURE_CHATGPT_API_KEY"), - endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), - model=os.getenv("AZURE_CHATGPT_MODEL"), - ) - context = OpenAILLMContext() - message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Please tell the world hello.", name="system", role="system" - ) - context.add_message(message) - frame = OpenAILLMContextFrame(context) - async for s in llm.process_frame(frame): - print(s) - - asyncio.run(test_chat()) diff --git a/tests/integration/integration_ollama_llm.py b/tests/integration/integration_ollama_llm.py deleted file mode 100644 index 085500cb8..000000000 --- a/tests/integration/integration_ollama_llm.py +++ /dev/null @@ -1,28 +0,0 @@ -import asyncio -import unittest - -from openai.types.chat import ( - ChatCompletionSystemMessageParam, -) - -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) -from pipecat.services.ollama import OLLamaLLMService - -if __name__ == "__main__": - - @unittest.skip("Skip azure integration test") - async def test_chat(): - llm = OLLamaLLMService() - context = OpenAILLMContext() - message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Please tell the world hello.", name="system", role="system" - ) - context.add_message(message) - frame = OpenAILLMContextFrame(context) - async for s in llm.process_frame(frame): - print(s) - - asyncio.run(test_chat()) diff --git a/tests/integration/integration_openai_llm.py b/tests/integration/integration_openai_llm.py deleted file mode 100644 index 2c84c8882..000000000 --- a/tests/integration/integration_openai_llm.py +++ /dev/null @@ -1,128 +0,0 @@ -import asyncio -import json -import os -from typing import List - -from openai.types.chat import ( - ChatCompletionSystemMessageParam, - ChatCompletionToolParam, - ChatCompletionUserMessageParam, -) - -from pipecat.frames.frames import LLMFullResponseEndFrame, LLMFullResponseStartFrame, TextFrame -from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService -from pipecat.utils.test_frame_processor import TestFrameProcessor - -tools = [ - ChatCompletionToolParam( - type="function", - function={ - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", - }, - }, - "required": ["location", "format"], - }, - }, - ) -] - -if __name__ == "__main__": - - async def test_simple_functions(): - async def get_weather_from_api(llm, args): - return json.dumps({"conditions": "nice", "temperature": "75"}) - - api_key = os.getenv("OPENAI_API_KEY") - - llm = OpenAILLMService( - api_key=api_key or "", - model="gpt-4-1106-preview", - ) - - llm.register_function("get_current_weather", get_weather_from_api) - t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame]) - llm.link(t) - - context = OpenAILLMContext(tools=tools) - system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Ask the user to ask for a weather report", name="system", role="system" - ) - user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam( - content="Could you tell me the weather for Boulder, Colorado", - name="user", - role="user", - ) - context.add_message(system_message) - context.add_message(user_message) - frame = OpenAILLMContextFrame(context) - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - async def test_advanced_functions(): - async def get_weather_from_api(llm, args): - return [ - { - "role": "system", - "content": "The user has asked for live weather. Respond by telling them we don't currently support live weather for that area, but it's coming soon.", - } - ] - - api_key = os.getenv("OPENAI_API_KEY") - - llm = OpenAILLMService( - api_key=api_key or "", - model="gpt-4-1106-preview", - ) - - llm.register_function("get_current_weather", get_weather_from_api) - t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame]) - llm.link(t) - - context = OpenAILLMContext(tools=tools) - system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Ask the user to ask for a weather report", name="system", role="system" - ) - user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam( - content="Could you tell me the weather for Boulder, Colorado", - name="user", - role="user", - ) - context.add_message(system_message) - context.add_message(user_message) - frame = OpenAILLMContextFrame(context) - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - async def test_chat(): - api_key = os.getenv("OPENAI_API_KEY") - t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame]) - llm = OpenAILLMService( - api_key=api_key or "", - model="gpt-4o", - ) - llm.link(t) - context = OpenAILLMContext() - message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Please tell the world hello.", name="system", role="system" - ) - context.add_message(message) - frame = OpenAILLMContextFrame(context) - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - async def run_tests(): - await test_simple_functions() - await test_advanced_functions() - await test_chat() - - asyncio.run(run_tests()) diff --git a/tests/integration/test_integration_unified_function_calling.py b/tests/integration/test_integration_unified_function_calling.py index 88407d703..0696c531a 100644 --- a/tests/integration/test_integration_unified_function_calling.py +++ b/tests/integration/test_integration_unified_function_calling.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import os from unittest.mock import AsyncMock @@ -6,17 +12,12 @@ from dotenv import load_dotenv from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.frames.frames import ( - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMTextFrame, -) -from pipecat.processors.frame_processor import FrameDirection +from pipecat.pipeline.pipeline import Pipeline from pipecat.services.ai_services import LLMService from pipecat.services.anthropic import AnthropicLLMService from pipecat.services.google import GoogleLLMService from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService -from pipecat.utils.test_frame_processor import TestFrameProcessor +from pipecat.tests.utils import run_test load_dotenv(override=True) @@ -47,8 +48,6 @@ async def _test_llm_function_calling(llm: LLMService): mock_fetch_weather = AsyncMock() llm.register_function(None, mock_fetch_weather) - t = TestFrameProcessor([LLMFullResponseStartFrame, LLMTextFrame, LLMFullResponseEndFrame]) - llm.link(t) messages = [ { @@ -61,10 +60,14 @@ async def _test_llm_function_calling(llm: LLMService): # This is done by default inside the create_context_aggregator context.set_llm_adapter(llm.get_llm_adapter()) - frame = OpenAILLMContextFrame(context) + pipeline = Pipeline([llm]) - # This will fail if an exception is raised - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) + frames_to_send = [OpenAILLMContextFrame(context)] + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=None, + ) # Assert that the mock function was called mock_fetch_weather.assert_called_once() From d455fd070eebedcbab3fbf932b09d724dd39df55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 16:43:43 -0700 Subject: [PATCH 4/8] update CHANGELOG --- CHANGELOG.md | 3 +++ src/pipecat/services/google/google.py | 5 ++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4762186cd..247c3280d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 being aggregated. A text aggregator can be passed via `text_aggregator` to the TTS service. +- Added new `sample_rate` constructor parameter to `TavusVideoService` to allow + changing the output sample rate. + - Added new `UltravoxSTTService`. (see https://github.com/fixie-ai/ultravox) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index c07707899..3db2044db 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -1316,9 +1316,12 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): class GoogleVertexLLMService(OpenAILLMService): - """Implements inference with Google's AI models via Vertex AI while maintaining OpenAI API compatibility. + """Implements inference with Google's AI models via Vertex AI while + maintaining OpenAI API compatibility. + Reference: https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/call-vertex-using-openai-library + """ class InputParams(OpenAILLMService.InputParams): From 3e9678db8430101a7964d9f3e246a6db48355c9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 19 Mar 2025 13:12:42 -0700 Subject: [PATCH 5/8] user image requests can now be related to function calls --- .../14b-function-calling-anthropic-video.py | 7 ++- .../14d-function-calling-video.py | 9 ++- .../14e-function-calling-gemini.py | 7 ++- .../20d-persistent-context-gemini.py | 7 ++- examples/moondream-chatbot/bot.py | 1 - src/pipecat/frames/frames.py | 25 ++++---- .../processors/aggregators/llm_response.py | 12 ++-- .../aggregators/openai_llm_context.py | 1 - src/pipecat/services/ai_services.py | 17 +++++- src/pipecat/services/anthropic.py | 57 ++++--------------- .../services/gemini_multimodal_live/gemini.py | 4 +- src/pipecat/services/google/google.py | 15 +++-- src/pipecat/services/openai.py | 55 ++++-------------- src/pipecat/transports/services/daily.py | 21 ++++--- 14 files changed, 101 insertions(+), 137 deletions(-) diff --git a/examples/foundational/14b-function-calling-anthropic-video.py b/examples/foundational/14b-function-calling-anthropic-video.py index 302722be2..cd9777b72 100644 --- a/examples/foundational/14b-function-calling-anthropic-video.py +++ b/examples/foundational/14b-function-calling-anthropic-video.py @@ -39,7 +39,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback): question = arguments["question"] - await llm.request_image_frame(user_id=video_participant_id, text_content=question) + await llm.request_image_frame( + user_id=video_participant_id, + function_name=function_name, + tool_call_id=tool_call_id, + text_content=question, + ) async def main(): diff --git a/examples/foundational/14d-function-calling-video.py b/examples/foundational/14d-function-calling-video.py index 6e290d55f..a7e997f9b 100644 --- a/examples/foundational/14d-function-calling-video.py +++ b/examples/foundational/14d-function-calling-video.py @@ -39,7 +39,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback): logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}") question = arguments["question"] - await llm.request_image_frame(user_id=video_participant_id, text_content=question) + await llm.request_image_frame( + user_id=video_participant_id, + function_name=function_name, + tool_call_id=tool_call_id, + text_content=question, + ) async def main(): @@ -141,7 +146,7 @@ indicate you should use the get_image tool are: await transport.capture_participant_transcription(participant["id"]) await transport.capture_participant_video(video_participant_id, framerate=0) # Kick off the conversation. - await tts.say("Hi! Ask me about the weather in San Francisco.") + await task.queue_frames([context_aggregator.user().get_context_frame()]) runner = PipelineRunner() diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index f0003fbab..8448643b4 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/examples/foundational/14e-function-calling-gemini.py @@ -42,7 +42,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback): logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}") question = arguments["question"] - await llm.request_image_frame(user_id=video_participant_id, text_content=question) + await llm.request_image_frame( + user_id=video_participant_id, + function_name=function_name, + tool_call_id=tool_call_id, + text_content=question, + ) async def main(): diff --git a/examples/foundational/20d-persistent-context-gemini.py b/examples/foundational/20d-persistent-context-gemini.py index 740bdc68f..2381ea131 100644 --- a/examples/foundational/20d-persistent-context-gemini.py +++ b/examples/foundational/20d-persistent-context-gemini.py @@ -54,7 +54,12 @@ async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback): question = arguments["question"] - await llm.request_image_frame(user_id=video_participant_id, text_content=question) + await llm.request_image_frame( + user_id=video_participant_id, + function_name=function_name, + tool_call_id=tool_call_id, + text_content=question, + ) async def get_saved_conversation_filenames( diff --git a/examples/moondream-chatbot/bot.py b/examples/moondream-chatbot/bot.py index 5dd88f4d1..6639c08c1 100644 --- a/examples/moondream-chatbot/bot.py +++ b/examples/moondream-chatbot/bot.py @@ -23,7 +23,6 @@ from pipecat.frames.frames import ( OutputImageRawFrame, SpriteFrame, TextFrame, - UserImageRawFrame, UserImageRequestFrame, ) from pipecat.pipeline.parallel_pipeline import ParallelPipeline diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e589e9480..c2a79461f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -662,13 +662,19 @@ class TransportMessageUrgentFrame(SystemFrame): @dataclass class UserImageRequestFrame(SystemFrame): - """A frame user to request an image from the given user.""" + """A frame to request an image from the given user. The frame might be + generated by a function call in which case the corresponding fields will be + properly set. + + """ user_id: str context: Optional[Any] = None + function_name: Optional[str] = None + tool_call_id: Optional[str] = None def __str__(self): - return f"{self.name}, user: {self.user_id}" + return f"{self.name}(user: {self.user_id}, function: {self.function_name}, request: {self.tool_call_id})" @dataclass @@ -698,10 +704,11 @@ class UserImageRawFrame(InputImageRawFrame): """An image associated to a user.""" user_id: str + request: Optional[UserImageRequestFrame] = None def __str__(self): pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})" + return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format}, request: {self.request})" @dataclass @@ -715,18 +722,6 @@ class VisionImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" -@dataclass -class UserImageMessageFrame(SystemFrame): - """An image associated to a user.""" - - user_image_raw_frame: UserImageRawFrame - text: Optional[str] = None - - def __str__(self): - pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, image: {self.user_image_raw_frame}, text: {self.text})" - - # # Control frames # diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 494d1de38..ff1b6e1be 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -34,7 +34,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, TextFrame, TranscriptionFrame, - UserImageMessageFrame, + UserImageRawFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -401,7 +401,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): pass - async def handle_image_frame_message(self, frame: UserImageMessageFrame): + async def handle_user_image_frame(self, frame: UserImageRawFrame): pass async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -428,8 +428,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self._handle_function_call_result(frame) elif isinstance(frame, FunctionCallCancelFrame): await self._handle_function_call_cancel(frame) - elif isinstance(frame, UserImageMessageFrame): - await self._handle_image_frame_message(frame) + elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id: + await self._handle_user_image_frame(frame) elif isinstance(frame, BotStoppedSpeakingFrame): await self.push_aggregation() else: @@ -510,8 +510,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self.handle_function_call_cancel(frame) del self._function_calls_in_progress[frame.tool_call_id] - async def _handle_image_frame_message(self, frame: UserImageMessageFrame): - await self.handle_image_frame_message(frame) + async def _handle_user_image_frame(self, frame: UserImageRawFrame): + await self.handle_user_image_frame(frame) await self.push_aggregation() await self.push_context_frame(FrameDirection.UPSTREAM) diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 93c2875be..2e5ade0a0 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -46,7 +46,6 @@ class OpenAILLMContext: self._messages: List[ChatCompletionMessageParam] = messages if messages else [] self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice self._tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = tools - self._user_image_request_context = {} self._llm_adapter: Optional[BaseLLMAdapter] = None def get_llm_adapter(self) -> Optional[BaseLLMAdapter]: diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 2f30f505f..678132cb6 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -253,9 +253,22 @@ class LLMService(AIService): elif None in self._start_callbacks.keys(): return await self._start_callbacks[None](function_name, self, context) - async def request_image_frame(self, user_id: str, *, text_content: Optional[str] = None): + async def request_image_frame( + self, + user_id: str, + *, + function_name: Optional[str] = None, + tool_call_id: Optional[str] = None, + text_content: Optional[str] = None, + ): await self.push_frame( - UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM + UserImageRequestFrame( + user_id=user_id, + function_name=function_name, + tool_call_id=tool_call_id, + context=text_content, + ), + FrameDirection.UPSTREAM, ) async def _run_function_call( diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 8e06f4558..a80e2154c 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -30,9 +30,7 @@ from pipecat.frames.frames import ( LLMMessagesFrame, LLMTextFrame, LLMUpdateSettingsFrame, - UserImageMessageFrame, UserImageRawFrame, - UserImageRequestFrame, VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage @@ -674,42 +672,7 @@ class AnthropicLLMContext(OpenAILLMContext): class AnthropicUserContextAggregator(LLMUserContextAggregator): - def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs): - super().__init__(context=context, **kwargs) - - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - # Our parent method has already called push_frame(). So we can't interrupt the - # flow here and we don't need to call push_frame() ourselves. Possibly something - # to talk through (tagging @aleix). At some point we might need to refactor these - # context aggregators. - try: - if isinstance(frame, UserImageRequestFrame): - # The LLM sends a UserImageRequestFrame upstream. Cache any context provided with - # that frame so we can use it when we assemble the image message in the assistant - # context aggregator. - if frame.context: - if isinstance(frame.context, str): - self._context._user_image_request_context[frame.user_id] = frame.context - else: - logger.error( - f"Unexpected UserImageRequestFrame context type: {type(frame.context)}" - ) - del self._context._user_image_request_context[frame.user_id] - else: - if frame.user_id in self._context._user_image_request_context: - del self._context._user_image_request_context[frame.user_id] - elif isinstance(frame, UserImageRawFrame): - # Push a new AnthropicImageMessageFrame with the text context we cached - # downstream to be handled by our assistant context aggregator. This is - # necessary so that we add the message to the context in the right order. - text = self._context._user_image_request_context.get(frame.user_id) or "" - if text: - del self._context._user_image_request_context[frame.user_id] - frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text) - await self.push_frame(frame) - except Exception as e: - logger.error(f"Error processing frame: {e}") + pass # @@ -723,9 +686,6 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator): class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): - def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs): - super().__init__(context=context, **kwargs) - async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): assistant_message = {"role": "assistant", "content": []} assistant_message["content"].append( @@ -776,10 +736,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): ): content["content"] = result - async def handle_image_frame_message(self, frame: UserImageMessageFrame): - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, + async def handle_user_image_frame(self, frame: UserImageRawFrame): + await self._update_function_call_result( + frame.request.function_name, frame.request.tool_call_id, "COMPLETED" + ) + self._context.add_image_frame_message( + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.request.context, ) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index da1697da6..801af46b4 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -39,7 +39,7 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, TTSTextFrame, - UserImageMessageFrame, + UserImageRawFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -119,7 +119,7 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator): class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator): - async def handle_image_frame_message(self, frame: UserImageMessageFrame): + async def handle_user_image_frame(self, frame: UserImageRawFrame): # We don't want to store any images in the context. Revisit this later # when the API evolves. pass diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 3db2044db..128eb43b6 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -49,7 +49,7 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, URLImageRawFrame, - UserImageMessageFrame, + UserImageRawFrame, VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage @@ -624,12 +624,15 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator): if part.function_response and part.function_response.id == tool_call_id: part.function_response.response = result - async def handle_image_frame_message(self, frame: UserImageMessageFrame): + async def handle_user_image_frame(self, frame: UserImageRawFrame): + await self._update_function_call_result( + frame.request.function_name, frame.request.tool_call_id, "COMPLETED" + ) self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.request.context, ) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 700f1c0b5..8318739de 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -40,9 +40,7 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, URLImageRawFrame, - UserImageMessageFrame, UserImageRawFrame, - UserImageRequestFrame, VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage @@ -557,46 +555,10 @@ class OpenAITTSService(TTSService): class OpenAIUserContextAggregator(LLMUserContextAggregator): - def __init__(self, context: OpenAILLMContext, **kwargs): - super().__init__(context=context, **kwargs) - - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - # Our parent method has already called push_frame(). So we can't interrupt the - # flow here and we don't need to call push_frame() ourselves. - try: - if isinstance(frame, UserImageRequestFrame): - # The LLM sends a UserImageRequestFrame upstream. Cache any context provided with - # that frame so we can use it when we assemble the image message in the assistant - # context aggregator. - if frame.context: - if isinstance(frame.context, str): - self._context._user_image_request_context[frame.user_id] = frame.context - else: - logger.error( - f"Unexpected UserImageRequestFrame context type: {type(frame.context)}" - ) - del self._context._user_image_request_context[frame.user_id] - else: - if frame.user_id in self._context._user_image_request_context: - del self._context._user_image_request_context[frame.user_id] - elif isinstance(frame, UserImageRawFrame): - # Push a new OpenAIImageMessageFrame with the text context we cached - # downstream to be handled by our assistant context aggregator. This is - # necessary so that we add the message to the context in the right order. - text = self._context._user_image_request_context.get(frame.user_id) or "" - if text: - del self._context._user_image_request_context[frame.user_id] - frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text) - await self.push_frame(frame) - except Exception as e: - logger.error(f"Error processing frame: {e}") + pass class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): - def __init__(self, context: OpenAILLMContext, **kwargs): - super().__init__(context=context, **kwargs) - async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): self._context.add_message( { @@ -645,10 +607,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): ): message["content"] = result - async def handle_image_frame_message(self, frame: UserImageMessageFrame): - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, + async def handle_user_image_frame(self, frame: UserImageRawFrame): + await self._update_function_call_result( + frame.request.function_name, frame.request.tool_call_id, "COMPLETED" + ) + self._context.add_image_frame_message( + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.request.context, ) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index f4b83dfa7..af7d2308c 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -898,7 +898,7 @@ class DailyInputTransport(BaseInputTransport): await super().process_frame(frame, direction) if isinstance(frame, UserImageRequestFrame): - await self.request_participant_image(frame.user_id) + await self.request_participant_image(frame) # # Frames @@ -935,16 +935,16 @@ class DailyInputTransport(BaseInputTransport): self._video_renderers[participant_id] = { "framerate": framerate, "timestamp": 0, - "render_next_frame": False, + "render_next_frame": [], } await self._client.capture_participant_video( participant_id, self._on_participant_video_frame, framerate, video_source, color_format ) - async def request_participant_image(self, participant_id: str): - if participant_id in self._video_renderers: - self._video_renderers[participant_id]["render_next_frame"] = True + async def request_participant_image(self, frame: UserImageRequestFrame): + if frame.user_id in self._video_renderers: + self._video_renderers[frame.user_id]["render_next_frame"].append(frame) async def _on_participant_video_frame(self, participant_id: str, buffer, size, format): render_frame = False @@ -953,17 +953,24 @@ class DailyInputTransport(BaseInputTransport): prev_time = self._video_renderers[participant_id]["timestamp"] framerate = self._video_renderers[participant_id]["framerate"] + # Some times we render frames because of a request. + request_frame = None + if framerate > 0: next_time = prev_time + 1 / framerate render_frame = (next_time - curr_time) < 0.1 elif self._video_renderers[participant_id]["render_next_frame"]: - self._video_renderers[participant_id]["render_next_frame"] = False + request_frame = self._video_renderers[participant_id]["render_next_frame"].pop(0) render_frame = True if render_frame: frame = UserImageRawFrame( - user_id=participant_id, image=buffer, size=size, format=format + user_id=participant_id, + request=request_frame, + image=buffer, + size=size, + format=format, ) await self.push_frame(frame) self._video_renderers[participant_id]["timestamp"] = curr_time From 1f6ed01ba6b910a65cf8237bd69aff18800a43d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 19 Mar 2025 14:06:22 -0700 Subject: [PATCH 6/8] LLMAssistantContextAggregator: remove tool call id with image requests --- src/pipecat/processors/aggregators/llm_response.py | 12 ++++++++++++ src/pipecat/services/ai_services.py | 4 +++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index ff1b6e1be..9267af364 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -511,6 +511,18 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): del self._function_calls_in_progress[frame.tool_call_id] async def _handle_user_image_frame(self, frame: UserImageRawFrame): + logger.debug( + f"{self} UserImageRawFrame: [{frame.request.function_name}:{frame.request.tool_call_id}]" + ) + + if frame.request.tool_call_id not in self._function_calls_in_progress: + logger.warning( + f"UserImageRawFrame tool_call_id [{frame.request.tool_call_id}] is not running" + ) + return + + del self._function_calls_in_progress[frame.request.tool_call_id] + await self.handle_user_image_frame(frame) await self.push_aggregation() await self.push_context_frame(FrameDirection.UPSTREAM) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 678132cb6..9f9804e65 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -286,7 +286,9 @@ class LLMService(AIService): else: return - logger.info(f"Calling function {function_name} with arguments {arguments}") + logger.debug( + f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}" + ) # NOTE(aleix): This needs to be removed after we remove the deprecation. await self.call_start_function(context, function_name) From b1d506c137cec966f9aca0fbb835e20ce5822183 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 19 Mar 2025 14:09:13 -0700 Subject: [PATCH 7/8] GoogleAssistantContextAggregator: properly update function response --- src/pipecat/services/google/google.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 128eb43b6..7aa9b7b7b 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -622,7 +622,7 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator): if message.role == "user": for part in message.parts: if part.function_response and part.function_response.id == tool_call_id: - part.function_response.response = result + part.function_response.response = {"response": result} async def handle_user_image_frame(self, frame: UserImageRawFrame): await self._update_function_call_result( From e0c3f6ad832b8e583b188c34546dbec0edc6a021 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 19 Mar 2025 14:34:31 -0700 Subject: [PATCH 8/8] services: mark function calls as completed even the result is None --- src/pipecat/services/anthropic.py | 13 +++++++------ src/pipecat/services/google/google.py | 18 +++++++++++------- src/pipecat/services/openai.py | 13 +++++++------ 3 files changed, 25 insertions(+), 19 deletions(-) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index a80e2154c..6a95d04e2 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -711,12 +711,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): ) async def handle_function_call_result(self, frame: FunctionCallResultFrame): - if not frame.result: - return - - result = json.dumps(frame.result) - - await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + if frame.result: + result = json.dumps(frame.result) + await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + else: + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, "COMPLETED" + ) async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): await self._update_function_call_result( diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 7aa9b7b7b..bfddce46d 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -600,15 +600,19 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator): ) async def handle_function_call_result(self, frame: FunctionCallResultFrame): - if not frame.result: - return + if frame.result: + if not isinstance(frame.result, str): + return - if not isinstance(frame.result, str): - return + response = {"response": frame.result} - response = {"response": frame.result} - - await self._update_function_call_result(frame.function_name, frame.tool_call_id, response) + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, response + ) + else: + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, "COMPLETED" + ) async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): await self._update_function_call_result( diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 8318739de..b5f5d9d54 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -584,12 +584,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): ) async def handle_function_call_result(self, frame: FunctionCallResultFrame): - if not frame.result: - return - - result = json.dumps(frame.result) - - await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + if frame.result: + result = json.dumps(frame.result) + await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + else: + await self._update_function_call_result( + frame.function_name, frame.tool_call_id, "COMPLETED" + ) async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): await self._update_function_call_result(