From 36fea8f9e8be90cba549fe47527a1f87190e1170 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 23 Jul 2025 14:34:17 -0400 Subject: [PATCH] Progress on LLM failover support --- .../processors/aggregators/llm_context.py | 185 ++++ .../processors/aggregators/llm_response.py | 883 +----------------- 2 files changed, 210 insertions(+), 858 deletions(-) create mode 100644 src/pipecat/processors/aggregators/llm_context.py diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py new file mode 100644 index 000000000..6cfdd3e9d --- /dev/null +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -0,0 +1,185 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Universal LLM context management for LLM services in Pipecat. + +Context contents are represented in a generic format (extended from OpenAI) +that supports a union of known Pipecat LLM service functionality. + +Whenever an LLM service needs to access context, it does a just-in-time +translation from this universal context into whatever format it needs, using a +service-specific adapter. +""" + +import base64 +import io +from dataclasses import dataclass +from typing import List, Optional + +from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN +from openai._types import NotGiven as OpenAINotGiven +from openai.types.chat import ( + ChatCompletionMessageParam, + ChatCompletionToolChoiceOptionParam, + ChatCompletionToolParam, +) +from PIL import Image + +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.frames.frames import AudioRawFrame, Frame + +# "Re-export" types from OpenAI that we're using as universal context types. +LLMContextMessage = ChatCompletionMessageParam +LLMContextTool = ChatCompletionToolParam +LLMContextToolChoice = ChatCompletionToolChoiceOptionParam +NOT_GIVEN = OPEN_AI_NOT_GIVEN +NotGiven = OpenAINotGiven + + +class LLMContext: + """Manages conversation context for LLM interactions. + + Handles message history, tool definitions, tool choices, and multimedia + content for LLM conversations. Provides methods for message manipulation, + and content formatting. + """ + + def __init__( + self, + messages: Optional[List[LLMContextMessage]] = None, + tools: List[LLMContextTool] | NotGiven | ToolsSchema = NOT_GIVEN, + tool_choice: LLMContextToolChoice | NotGiven = NOT_GIVEN, + ): + """Initialize the LLM context. + + Args: + messages: Initial list of conversation messages. + tools: Available tools for the LLM to use. + tool_choice: Tool selection strategy for the LLM. + """ + self._messages: List[LLMContextMessage] = messages if messages else [] + self._tools: List[LLMContextTool] | NotGiven | ToolsSchema = tools + self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice + + @property + def messages(self) -> List[LLMContextMessage]: + """Get the current messages list. + + Returns: + List of conversation messages. + """ + return self._messages + + @property + def tools(self) -> List[LLMContextTool] | NotGiven | List[Any]: + """Get the tools list. + + Returns: + Tools list. + """ + return self._tools + + @property + def tool_choice(self) -> LLMContextToolChoice | NotGiven: + """Get the current tool choice setting. + + Returns: + The tool choice configuration. + """ + return self._tool_choice + + def add_message(self, message: LLMContextMessage): + """Add a single message to the context. + + Args: + message: The message to add to the conversation history. + """ + self._messages.append(message) + + def add_messages(self, messages: List[LLMContextMessage]): + """Add multiple messages to the context. + + Args: + messages: List of messages to add to the conversation history. + """ + self._messages.extend(messages) + + def set_messages(self, messages: List[LLMContextMessage]): + """Replace all messages in the context. + + Args: + messages: New list of messages to replace the current history. + """ + self._messages[:] = messages + + def set_tools(self, tools: List[LLMContextTool] | NotGiven | ToolsSchema = NOT_GIVEN): + """Set the available tools for the LLM. + + Args: + tools: List of tools available to the LLM, a ToolsSchema, or NOT_GIVEN to disable tools. + """ + # TODO: convert empty ToolsSchema to NOT_GIVEN if needed + if isinstance(tools, list) and len(tools) == 0: + tools = NOT_GIVEN + self._tools = tools + + def set_tool_choice(self, tool_choice: LLMContextToolChoice | NotGiven): + """Set the tool choice configuration. + + Args: + tool_choice: Tool selection strategy for the LLM. + """ + self._tool_choice = tool_choice + + def add_image_frame_message( + self, *, format: str, size: tuple[int, int], image: bytes, text: str = None + ): + """Add a message containing an image frame. + + Args: + format: Image format (e.g., 'RGB', 'RGBA'). + size: Image dimensions as (width, height) tuple. + image: Raw image bytes. + text: Optional text to include with the image. + """ + buffer = io.BytesIO() + Image.frombytes(format, size, image).save(buffer, format="JPEG") + encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8") + + content = [] + if text: + content.append({"type": "text", "text": text}) + content.append( + {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}}, + ) + self.add_message({"role": "user", "content": content}) + + def add_audio_frames_message(self, *, audio_frames: list[AudioRawFrame], text: str = None): + """Add a message containing audio frames. + + Args: + audio_frames: List of audio frame objects to include. + text: Optional text to include with the audio. + + Note: + This method is currently a placeholder for future implementation. + """ + # TODO: implement storing universal representation of audio frames in context (only used by Google for now) + pass + + +@dataclass +class LLMContextFrame(Frame): + """Frame containing LLM context. + + Used as a signal to LLM services to ingest the provided context and + generate a response based on it. + + Parameters: + context: The LLM context containing messages, tools, and configuration. + """ + + context: LLMContext diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 568314f77..8ddeedcc8 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -12,59 +12,18 @@ LLM processing, and text-to-speech components in conversational AI pipelines. """ import asyncio -from abc import abstractmethod from dataclasses import dataclass -from typing import Dict, List, Literal, Optional, Set +from typing import List, Literal, Optional -from loguru import logger - -from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import ( - BotInterruptionFrame, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - CancelFrame, - EmulateUserStartedSpeakingFrame, - EmulateUserStoppedSpeakingFrame, - EndFrame, - Frame, - FunctionCallCancelFrame, - FunctionCallInProgressFrame, - FunctionCallResultFrame, - FunctionCallsStartedFrame, - InputAudioRawFrame, - InterimTranscriptionFrame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMMessagesAppendFrame, - LLMMessagesFrame, - LLMMessagesUpdateFrame, - LLMSetToolChoiceFrame, - LLMSetToolsFrame, - LLMTextFrame, - OpenAILLMContextAssistantTimestampFrame, - SpeechControlParamsFrame, - StartFrame, - StartInterruptionFrame, - TextFrame, - TranscriptionFrame, - UserImageRawFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, -) -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.time import time_now_iso8601 @dataclass -class LLMUserAggregatorParams: - """Parameters for configuring LLM user aggregation behavior. +class LLMUserContextAggregatorParams: + """Parameters for configuring LLM user context aggregation behavior. Parameters: aggregation_timeout: Maximum time in seconds to wait for additional @@ -80,8 +39,8 @@ class LLMUserAggregatorParams: @dataclass -class LLMAssistantAggregatorParams: - """Parameters for configuring LLM assistant aggregation behavior. +class LLMAssistantContextAggregatorParams: + """Parameters for configuring LLM assistant context aggregation behavior. Parameters: expect_stripped_words: Whether to expect and handle stripped words @@ -91,190 +50,19 @@ class LLMAssistantAggregatorParams: expect_stripped_words: bool = True -class LLMFullResponseAggregator(FrameProcessor): - """Aggregates complete LLM responses between start and end frames. +class LLMContextAggregator(FrameProcessor): + """Base LLM aggregator that uses an LLMContext for conversation storage. - This aggregator collects LLM text frames (tokens) received between - `LLMFullResponseStartFrame` and `LLMFullResponseEndFrame` and provides - the complete response via an event handler. - - The aggregator provides an "on_completion" event that fires when a full - completion is available:: - - @aggregator.event_handler("on_completion") - async def on_completion( - aggregator: LLMFullResponseAggregator, - completion: str, - completed: bool, - ): - # Handle the completion - pass - """ - - def __init__(self, **kwargs): - """Initialize the LLM full response aggregator. - - Args: - **kwargs: Additional arguments passed to parent FrameProcessor. - """ - super().__init__(**kwargs) - - self._aggregation = "" - self._started = False - - self._register_event_handler("on_completion") - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process incoming frames and aggregate LLM text content. - - Args: - frame: The frame to process. - direction: The direction of frame flow in the pipeline. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, StartInterruptionFrame): - await self._call_event_handler("on_completion", self._aggregation, False) - self._aggregation = "" - self._started = False - elif isinstance(frame, LLMFullResponseStartFrame): - await self._handle_llm_start(frame) - elif isinstance(frame, LLMFullResponseEndFrame): - await self._handle_llm_end(frame) - elif isinstance(frame, LLMTextFrame): - await self._handle_llm_text(frame) - - await self.push_frame(frame, direction) - - async def _handle_llm_start(self, _: LLMFullResponseStartFrame): - self._started = True - - async def _handle_llm_end(self, _: LLMFullResponseEndFrame): - await self._call_event_handler("on_completion", self._aggregation, True) - self._started = False - self._aggregation = "" - - async def _handle_llm_text(self, frame: TextFrame): - if not self._started: - return - self._aggregation += frame.text - - -class BaseLLMResponseAggregator(FrameProcessor): - """Base class for all LLM response aggregators. - - These aggregators process incoming frames and aggregate content until they are - ready to push the aggregation downstream. They maintain conversation state - and handle message flow between different components in the pipeline. - - The aggregators keep a store (e.g. message list or LLM context) of the current - conversation, storing messages from both users and the bot. - """ - - def __init__(self, **kwargs): - """Initialize the base LLM response aggregator. - - Args: - **kwargs: Additional arguments passed to parent FrameProcessor. - """ - super().__init__(**kwargs) - - @property - @abstractmethod - def messages(self) -> List[dict]: - """Get the messages from the current conversation. - - Returns: - List of message dictionaries representing the conversation history. - """ - pass - - @property - @abstractmethod - def role(self) -> str: - """Get the role for this aggregator. - - Returns: - The role string (e.g. "user", "assistant") for this aggregator. - """ - pass - - @abstractmethod - def add_messages(self, messages): - """Add the given messages to the conversation. - - Args: - messages: Messages to append to the conversation history. - """ - pass - - @abstractmethod - def set_messages(self, messages): - """Reset the conversation with the given messages. - - Args: - messages: Messages to replace the current conversation history. - """ - pass - - @abstractmethod - def set_tools(self, tools): - """Set LLM tools to be used in the current conversation. - - Args: - tools: List of tool definitions for the LLM to use. - """ - pass - - @abstractmethod - def set_tool_choice(self, tool_choice): - """Set the tool choice for the LLM. - - Args: - tool_choice: Tool choice configuration for the LLM context. - """ - pass - - @abstractmethod - async def reset(self): - """Reset the internal state of this aggregator. - - This should clear aggregation state but not modify the conversation messages. - """ - pass - - @abstractmethod - async def handle_aggregation(self, aggregation: str): - """Add the given aggregation to the conversation store. - - Args: - aggregation: The aggregated text content to add to the conversation. - """ - pass - - @abstractmethod - async def push_aggregation(self): - """Push the current aggregation downstream. - - The specific frame type pushed depends on the aggregator implementation - (e.g. context frame, messages frame). - """ - pass - - -class LLMContextResponseAggregator(BaseLLMResponseAggregator): - """Base LLM aggregator that uses an OpenAI LLM context for conversation storage. - - This aggregator maintains conversation state using an OpenAILLMContext and - pushes OpenAILLMContextFrame objects as aggregation frames. It provides + This aggregator maintains conversation state using an LLMContext and + pushes LLMContextFrame objects as aggregation frames. It provides common functionality for context-based conversation management. """ - def __init__(self, *, context: OpenAILLMContext, role: str, **kwargs): + def __init__(self, *, context: LLMContext, role: str, **kwargs): """Initialize the context response aggregator. Args: - context: The OpenAI LLM context to use for conversation storage. + context: The LLM context to use for conversation storage. role: The role this aggregator represents (e.g. "user", "assistant"). **kwargs: Additional arguments passed to parent class. """ @@ -291,7 +79,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator): Returns: List of message dictionaries from the context. """ - return self._context.get_messages() + return self._context.messages @property def role(self) -> str: @@ -304,20 +92,20 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator): @property def context(self): - """Get the OpenAI LLM context. + """Get the LLM context. Returns: - The OpenAILLMContext instance used by this aggregator. + The LLMContext instance used by this aggregator. """ return self._context - def get_context_frame(self) -> OpenAILLMContextFrame: + def get_context_frame(self) -> LLMContextFrame: """Create a context frame with the current context. Returns: - OpenAILLMContextFrame containing the current context. + LLMContextFrame containing the current context. """ - return OpenAILLMContextFrame(context=self._context) + return LLMContextFrame(context=self._context) async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM): """Push a context frame in the specified direction. @@ -352,6 +140,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator): """ self._context.set_tools(tools) + # TODO: should we be using LLMContextToolChoice here? def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict): """Set tool choice in the context. @@ -365,7 +154,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator): self._aggregation = "" -class LLMUserContextAggregator(LLMContextResponseAggregator): +class LLMUserContextAggregator(LLMContextAggregator): """User LLM aggregator that processes speech-to-text transcriptions. This aggregator handles the complex logic of aggregating user speech transcriptions @@ -383,20 +172,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): def __init__( self, - context: OpenAILLMContext, + context: LLMContext, *, - params: Optional[LLMUserAggregatorParams] = None, + params: Optional[LLMUserContextAggregatorParams] = None, **kwargs, ): """Initialize the user context aggregator. Args: - context: The OpenAI LLM context for conversation storage. + context: The LLM context for conversation storage. params: Configuration parameters for aggregation behavior. **kwargs: Additional arguments. Supports deprecated 'aggregation_timeout'. """ super().__init__(context=context, role="user", **kwargs) - self._params = params or LLMUserAggregatorParams() + self._params = params or LLMUserContextAggregatorParams() self._vad_params: Optional[VADParams] = None self._turn_params: Optional[SmartTurnParams] = None @@ -438,626 +227,4 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): """ self._context.add_message({"role": self.role, "content": aggregation}) - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames for user speech aggregation and context management. - - Args: - frame: The frame to process. - direction: The direction of frame flow in the pipeline. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, StartFrame): - # Push StartFrame before start(), because we want StartFrame to be - # processed by every processor before any other frame is processed. - await self.push_frame(frame, direction) - await self._start(frame) - elif isinstance(frame, EndFrame): - # Push EndFrame before stop(), because stop() waits on the task to - # finish and the task finishes when EndFrame is processed. - await self.push_frame(frame, direction) - await self._stop(frame) - elif isinstance(frame, CancelFrame): - await self._cancel(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, InputAudioRawFrame): - await self._handle_input_audio(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, UserStartedSpeakingFrame): - await self._handle_user_started_speaking(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, UserStoppedSpeakingFrame): - await self._handle_user_stopped_speaking(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, BotStartedSpeakingFrame): - await self._handle_bot_started_speaking(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, BotStoppedSpeakingFrame): - await self._handle_bot_stopped_speaking(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, TranscriptionFrame): - await self._handle_transcription(frame) - elif isinstance(frame, InterimTranscriptionFrame): - await self._handle_interim_transcription(frame) - elif isinstance(frame, LLMMessagesAppendFrame): - await self._handle_llm_messages_append(frame) - elif isinstance(frame, LLMMessagesUpdateFrame): - await self._handle_llm_messages_update(frame) - elif isinstance(frame, LLMSetToolsFrame): - self.set_tools(frame.tools) - elif isinstance(frame, LLMSetToolChoiceFrame): - self.set_tool_choice(frame.tool_choice) - elif isinstance(frame, SpeechControlParamsFrame): - self._vad_params = frame.vad_params - self._turn_params = frame.turn_params - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) - - async def _process_aggregation(self): - """Process the current aggregation and push it downstream.""" - aggregation = self._aggregation - await self.reset() - await self.handle_aggregation(aggregation) - frame = OpenAILLMContextFrame(self._context) - await self.push_frame(frame) - - async def push_aggregation(self): - """Push the current aggregation based on interruption strategies and conditions.""" - if len(self._aggregation) > 0: - if self.interruption_strategies and self._bot_speaking: - should_interrupt = await self._should_interrupt_based_on_strategies() - - if should_interrupt: - logger.debug( - "Interruption conditions met - pushing BotInterruptionFrame and aggregation" - ) - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) - await self._process_aggregation() - else: - logger.debug("Interruption conditions not met - not pushing aggregation") - # Don't process aggregation, just reset it - await self.reset() - else: - # No interruption config - normal behavior (always push aggregation) - await self._process_aggregation() - # Handles the case where both the user and the bot are not speaking, - # and the bot was previously speaking before the user interruption. - # Normally, when the user stops speaking, new text is expected, - # which triggers the bot to respond. However, if no new text - # is received, this safeguard ensures - # the bot doesn't hang indefinitely while waiting to speak again. - elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking: - logger.warning("User stopped speaking but no new aggregation received.") - # Resetting it so we don't trigger this twice - self._was_bot_speaking = False - # TODO: we are not enabling this for now, due to some STT services which can take as long as 2 seconds two return a transcription - # So we need more tests and probably make this feature configurable, disabled it by default. - # We are just pushing the same previous context to be processed again in this case - # await self.push_frame(OpenAILLMContextFrame(self._context)) - - async def _should_interrupt_based_on_strategies(self) -> bool: - """Check if interruption should occur based on configured strategies. - - Returns: - True if any interruption strategy indicates interruption should occur. - """ - - async def should_interrupt(strategy: BaseInterruptionStrategy): - await strategy.append_text(self._aggregation) - return await strategy.should_interrupt() - - return any([await should_interrupt(s) for s in self._interruption_strategies]) - - async def _start(self, frame: StartFrame): - self._create_aggregation_task() - - async def _stop(self, frame: EndFrame): - await self._cancel_aggregation_task() - - async def _cancel(self, frame: CancelFrame): - await self._cancel_aggregation_task() - - async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame): - self.add_messages(frame.messages) - if frame.run_llm: - await self.push_context_frame() - - async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame): - self.set_messages(frame.messages) - if frame.run_llm: - await self.push_context_frame() - - async def _handle_input_audio(self, frame: InputAudioRawFrame): - for s in self.interruption_strategies: - await s.append_audio(frame.audio, frame.sample_rate) - - async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame): - self._user_speaking = True - self._waiting_for_aggregation = True - self._was_bot_speaking = self._bot_speaking - - # If we get a non-emulated UserStartedSpeakingFrame but we are in the - # middle of emulating VAD, let's stop emulating VAD (i.e. don't send the - # EmulateUserStoppedSpeakingFrame). - if not frame.emulated and self._emulating_vad: - self._emulating_vad = False - - async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame): - self._user_speaking = False - # We just stopped speaking. Let's see if there's some aggregation to - # push. If the last thing we saw is an interim transcription, let's wait - # pushing the aggregation as we will probably get a final transcription. - if len(self._aggregation) > 0: - if not self._seen_interim_results: - await self.push_aggregation() - # Handles the case where both the user and the bot are not speaking, - # and the bot was previously speaking before the user interruption. - # So in this case we are resetting the aggregation timer - elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking: - # Reset aggregation timer. - self._aggregation_event.set() - - async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame): - self._bot_speaking = True - - async def _handle_bot_stopped_speaking(self, _: BotStoppedSpeakingFrame): - self._bot_speaking = False - - async def _handle_transcription(self, frame: TranscriptionFrame): - text = frame.text - - # Make sure we really have some text. - if not text.strip(): - return - - self._aggregation += f" {text}" if self._aggregation else text - # We just got a final result, so let's reset interim results. - self._seen_interim_results = False - # Reset aggregation timer. - self._aggregation_event.set() - - async def _handle_interim_transcription(self, _: InterimTranscriptionFrame): - self._seen_interim_results = True - - def _create_aggregation_task(self): - if not self._aggregation_task: - self._aggregation_task = self.create_task(self._aggregation_task_handler()) - - async def _cancel_aggregation_task(self): - if self._aggregation_task: - await self.cancel_task(self._aggregation_task) - self._aggregation_task = None - - async def _aggregation_task_handler(self): - while True: - try: - # The _aggregation_task_handler handles two distinct timeout scenarios: - # - # 1. When emulating_vad=True: Wait for emulated VAD timeout before - # pushing aggregation (simulating VAD behavior when no actual VAD - # detection occurred). - # - # 2. When emulating_vad=False: Use aggregation_timeout as a buffer - # to wait for potential late-arriving transcription frames after - # a real VAD event. - # - # For emulated VAD scenarios, the timeout strategy depends on whether - # a turn analyzer is configured: - # - # - WITH turn analyzer: Use turn_emulated_vad_timeout parameter because - # the VAD's stop_secs is set very low (e.g. 0.2s) for rapid speech - # chunking to feed the turn analyzer. This low value is too fast - # for emulated VAD scenarios where we need to allow users time to - # finish speaking (e.g. 0.8s). - # - # - WITHOUT turn analyzer: Use VAD's stop_secs directly to maintain - # consistent user experience between real VAD detection and - # emulated VAD scenarios. - if not self._emulating_vad: - timeout = self._params.aggregation_timeout - elif self._turn_params: - timeout = self._params.turn_emulated_vad_timeout - else: - # Use VAD stop_secs when no turn analyzer is present, fallback if no VAD params - timeout = ( - self._vad_params.stop_secs - if self._vad_params - else self._params.turn_emulated_vad_timeout - ) - await asyncio.wait_for(self._aggregation_event.wait(), timeout) - await self._maybe_emulate_user_speaking() - except asyncio.TimeoutError: - if not self._user_speaking: - await self.push_aggregation() - - # If we are emulating VAD we still need to send the user stopped - # speaking frame. - if self._emulating_vad: - await self.push_frame( - EmulateUserStoppedSpeakingFrame(), FrameDirection.UPSTREAM - ) - self._emulating_vad = False - finally: - self.reset_watchdog() - self._aggregation_event.clear() - - async def _maybe_emulate_user_speaking(self): - """Maybe emulate user speaking based on transcription. - - Emulate user speaking if we got a transcription but it was not - detected by VAD. Only do that if the bot is not speaking. - """ - # Check if we received a transcription but VAD was not able to detect - # voice (e.g. when you whisper a short utterance). In that case, we need - # to emulate VAD (i.e. user start/stopped speaking), but we do it only - # if the bot is not speaking. If the bot is speaking and we really have - # a short utterance we don't really want to interrupt the bot. - if ( - not self._user_speaking - and not self._waiting_for_aggregation - and len(self._aggregation) > 0 - ): - if self._bot_speaking: - # If we reached this case and the bot is speaking, let's ignore - # what the user said. - logger.debug("Ignoring user speaking emulation, bot is speaking.") - await self.reset() - else: - # The bot is not speaking so, let's trigger user speaking - # emulation. - await self.push_frame(EmulateUserStartedSpeakingFrame(), FrameDirection.UPSTREAM) - self._emulating_vad = True - - -class LLMAssistantContextAggregator(LLMContextResponseAggregator): - """Assistant LLM aggregator that processes bot responses and function calls. - - This aggregator handles the complex logic of processing assistant responses including: - - - Text frame aggregation between response start/end markers - - Function call lifecycle management - - Context updates with timestamps - - Tool execution and result handling - - Interruption handling during responses - - The aggregator manages function calls in progress and coordinates between - text generation and tool execution phases of LLM responses. - """ - - def __init__( - self, - context: OpenAILLMContext, - *, - params: Optional[LLMAssistantAggregatorParams] = None, - **kwargs, - ): - """Initialize the assistant context aggregator. - - Args: - context: The OpenAI LLM context for conversation storage. - params: Configuration parameters for aggregation behavior. - **kwargs: Additional arguments. Supports deprecated 'expect_stripped_words'. - """ - super().__init__(context=context, role="assistant", **kwargs) - self._params = params or LLMAssistantAggregatorParams() - - if "expect_stripped_words" in kwargs: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Parameter 'expect_stripped_words' is deprecated, use 'params' instead.", - DeprecationWarning, - ) - - self._params.expect_stripped_words = kwargs["expect_stripped_words"] - - self._started = 0 - self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} - self._context_updated_tasks: Set[asyncio.Task] = set() - - @property - def has_function_calls_in_progress(self) -> bool: - """Check if there are any function calls currently in progress. - - Returns: - True if function calls are in progress, False otherwise. - """ - return bool(self._function_calls_in_progress) - - async def handle_aggregation(self, aggregation: str): - """Add the aggregated assistant text to the context. - - Args: - aggregation: The aggregated assistant text to add as an assistant message. - """ - self._context.add_message({"role": "assistant", "content": aggregation}) - - async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): - """Handle a function call that is in progress. - - Args: - frame: The function call in progress frame to handle. - """ - pass - - async def handle_function_call_result(self, frame: FunctionCallResultFrame): - """Handle the result of a completed function call. - - Args: - frame: The function call result frame to handle. - """ - pass - - async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): - """Handle cancellation of a function call. - - Args: - frame: The function call cancel frame to handle. - """ - pass - - async def handle_user_image_frame(self, frame: UserImageRawFrame): - """Handle a user image frame associated with a function call. - - Args: - frame: The user image frame to handle. - """ - pass - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames for assistant response aggregation and function call management. - - Args: - frame: The frame to process. - direction: The direction of frame flow in the pipeline. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, StartInterruptionFrame): - await self._handle_interruptions(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, LLMFullResponseStartFrame): - await self._handle_llm_start(frame) - elif isinstance(frame, LLMFullResponseEndFrame): - await self._handle_llm_end(frame) - elif isinstance(frame, TextFrame): - await self._handle_text(frame) - elif isinstance(frame, LLMMessagesAppendFrame): - await self._handle_llm_messages_append(frame) - elif isinstance(frame, LLMMessagesUpdateFrame): - await self._handle_llm_messages_update(frame) - elif isinstance(frame, LLMSetToolsFrame): - self.set_tools(frame.tools) - elif isinstance(frame, LLMSetToolChoiceFrame): - self.set_tool_choice(frame.tool_choice) - elif isinstance(frame, FunctionCallsStartedFrame): - await self._handle_function_calls_started(frame) - 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, 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() - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) - - async def push_aggregation(self): - """Push the current assistant aggregation with timestamp.""" - if not self._aggregation: - return - - aggregation = self._aggregation.strip() - await 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_llm_messages_append(self, frame: LLMMessagesAppendFrame): - self.add_messages(frame.messages) - if frame.run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame): - self.set_messages(frame.messages) - if frame.run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - async def _handle_interruptions(self, frame: StartInterruptionFrame): - await self.push_aggregation() - self._started = 0 - await self.reset() - - async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame): - function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls] - logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}") - for function_call in frame.function_calls: - self._function_calls_in_progress[function_call.tool_call_id] = None - - 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_llm = False - - # Run inference if the function call result requires it. - if 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 - elif frame.run_llm is not None: - # If the frame is indicating we should run the LLM, do it. - run_llm = frame.run_llm - else: - # If this is the last function call in progress, run the LLM. - run_llm = not bool(self._function_calls_in_progress) - - if run_llm: - await self.push_context_frame(FrameDirection.UPSTREAM) - - # Call the `on_context_updated` callback once the function call result - # is added to the context. Also, run this in a separate task to make - # sure we don't block the pipeline. - if properties and properties.on_context_updated: - task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated" - task = self.create_task(properties.on_context_updated(), task_name) - self._context_updated_tasks.add(task) - task.add_done_callback(self._context_updated_task_finished) - - 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_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) - - async def _handle_llm_start(self, _: LLMFullResponseStartFrame): - self._started += 1 - - async def _handle_llm_end(self, _: LLMFullResponseEndFrame): - self._started -= 1 - await self.push_aggregation() - - async def _handle_text(self, frame: TextFrame): - if not self._started: - return - - if self._params.expect_stripped_words: - self._aggregation += f" {frame.text}" if self._aggregation else frame.text - else: - self._aggregation += frame.text - - def _context_updated_task_finished(self, task: asyncio.Task): - self._context_updated_tasks.discard(task) - # The task is finished so this should exit immediately. We need to do - # this because otherwise the task manager would report a dangling task - # if we don't remove it. - asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop()) - - -class LLMUserResponseAggregator(LLMUserContextAggregator): - """User response aggregator that outputs LLMMessagesFrame instead of context frames. - - This aggregator extends LLMUserContextAggregator but pushes LLMMessagesFrame - objects downstream instead of OpenAILLMContextFrame objects. This is useful - when you need message-based output rather than context-based output. - """ - - def __init__( - self, - messages: Optional[List[dict]] = None, - *, - params: Optional[LLMUserAggregatorParams] = None, - **kwargs, - ): - """Initialize the user response aggregator. - - Args: - messages: Initial messages for the conversation context. - params: Configuration parameters for aggregation behavior. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs) - - async def push_aggregation(self): - """Push the aggregated user input as an LLMMessagesFrame.""" - 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. - await self.reset() - - frame = LLMMessagesFrame(self._context.messages) - await self.push_frame(frame) - - -class LLMAssistantResponseAggregator(LLMAssistantContextAggregator): - """Assistant response aggregator that outputs LLMMessagesFrame instead of context frames. - - This aggregator extends LLMAssistantContextAggregator but pushes LLMMessagesFrame - objects downstream instead of OpenAILLMContextFrame objects. This is useful - when you need message-based output rather than context-based output. - """ - - def __init__( - self, - messages: Optional[List[dict]] = None, - *, - params: Optional[LLMAssistantAggregatorParams] = None, - **kwargs, - ): - """Initialize the assistant response aggregator. - - Args: - messages: Initial messages for the conversation context. - params: Configuration parameters for aggregation behavior. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs) - - async def push_aggregation(self): - """Push the aggregated assistant response as an LLMMessagesFrame.""" - 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. - await self.reset() - - frame = LLMMessagesFrame(self._context.messages) - await self.push_frame(frame) + # TODO: continue porting things over from LLMUserContextAggregator in backup file