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] #