diff --git a/CHANGELOG.md b/CHANGELOG.md index f6f8754ee..d9768405c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,78 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added a new "universal" (LLM-agnostic) `LLMContext` and accompanying + `LLMContextAggregatorPair`, which will eventually replace `OpenAILLMContext` + (and the other under-the-hood contexts) and the other context aggregators. + The new universal `LLMContext` machinery allows a single context to be shared + between different LLMs, enabling runtime LLM switching and scenarios like + failover. + + From the developer's point of view, switching to using the new universal + context machinery will usually be a matter of going from this: + + ```python + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + ``` + + To this: + + ```python + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + ``` + + To start, the universal `LLMContext` is supported with the following LLM + services: + + - `OpenAILLMService` + - `GoogleLLMService` + +- Added a new `LLMSwitcher` class to enable runtime LLM switching, built atop a + new generic `ServiceSwitcher`. + + Switchers take a switching strategy. The first available strategy is + `ServiceSwitcherStrategyManual`. + + To switch LLMs at runtime, the LLMs must be sharing one instance of the new + universal `LLMContext` (see above bullet). + + ```python + # Instantiate your LLM services + llm_openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm_google = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + # Instantiate a switcher + # (ServiceSwitcherStrategyManual defaults to OpenAI, as it's first in the list) + llm_switcher = LLMSwitcher( + llms=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual + ) + + # Create your pipeline + pipeline = Pipeline( + [ + transport.input(), + stt, + context_aggregator.user(), + llm_switcher, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + + # ... + # Whenever is appropriate, switch LLMs! + await task.queue_frames([ManuallySwitchServiceFrame(service=llm_google)]) + ``` + +- Added an `LLMService.run_inference()` method to LLM services to enable + direct, out-of-band (i.e. out-of-pipeline) inference. + ### Fixed - Fixed a `CartesiaTTSService` issue that was causing the application to hang @@ -62,7 +134,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated - `FrameProcessor.wait_for_task()` is deprecated. Use `await task` or `await - asyncio.wait_for(task, timeout)` instead. +asyncio.wait_for(task, timeout)` instead. ### Removed diff --git a/env.example b/env.example index fae6bb2a4..93ebafeb2 100644 --- a/env.example +++ b/env.example @@ -59,6 +59,9 @@ GOOGLE_VERTEX_TEST_CREDENTIALS=... LMNT_API_KEY=... LMNT_VOICE_ID=... +# Perplexity +PERPLEXITY_API_KEY=... + # PlayHT PLAY_HT_USER_ID=... PLAY_HT_API_KEY=... diff --git a/examples/foundational/14x-function-calling-universal-context.py b/examples/foundational/14x-function-calling-universal-context.py new file mode 100644 index 000000000..890c145ab --- /dev/null +++ b/examples/foundational/14x-function-calling-universal-context.py @@ -0,0 +1,170 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import TTSSpeakFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + # You can also register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + transport.input(), + stt, + context_aggregator.user(), + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/foundational/14y-function-calling-google-universal-context.py b/examples/foundational/14y-function-calling-google-universal-context.py new file mode 100644 index 000000000..b22f2596c --- /dev/null +++ b/examples/foundational/14y-function-calling-google-universal-context.py @@ -0,0 +1,229 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import asyncio +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import TTSSpeakFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import ( + create_transport, + get_transport_client_id, + maybe_capture_participant_camera, +) +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + + +# Global variable to store the client ID +client_id = "" + + +async def get_weather(params: FunctionCallParams): + location = params.arguments["location"] + await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +async def get_image(params: FunctionCallParams): + question = params.arguments["question"] + logger.debug(f"Requesting image with user_id={client_id}, question={question}") + + # Request the image frame + await params.llm.request_image_frame( + user_id=client_id, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + text_content=question, + ) + + # Wait a short time for the frame to be processed + await asyncio.sleep(0.5) + + # Return a result to complete the function call + await params.result_callback( + f"I've captured an image from your camera and I'm analyzing what you asked about: {question}" + ) + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001") + llm.register_function("get_weather", get_weather) + llm.register_function("get_image", get_image) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + @llm.event_handler("on_function_calls_started") + async def on_function_calls_started(service, function_calls): + await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + + weather_function = FunctionSchema( + name="get_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + get_image_function = FunctionSchema( + name="get_image", + description="Get an image from the video stream.", + properties={ + "question": { + "type": "string", + "description": "The question that the user is asking about the image.", + } + }, + required=["question"], + ) + tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_function]) + + system_prompt = """\ +You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. + +Your response will be turned into speech so use only simple words and punctuation. + +You have access to three tools: get_weather, get_restaurant_recommendation, and get_image. + +You can respond to questions about the weather using the get_weather tool. + +You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ +indicate you should use the get_image tool are: +- What do you see? +- What's in the video? +- Can you describe the video? +- Tell me about what you see. +- Tell me something interesting about what you see. +- What's happening in the video? +""" + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": "Say hello."}, + ] + + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + transport.input(), + stt, + context_aggregator.user(), + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected: {client}") + + await maybe_capture_participant_camera(transport, client) + + global client_id + client_id = get_transport_client_id(transport, client) + + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py index 6a957c267..d24dd4b26 100644 --- a/src/pipecat/adapters/base_llm_adapter.py +++ b/src/pipecat/adapters/base_llm_adapter.py @@ -11,21 +11,45 @@ adapters that handle tool format conversion and standardization. """ from abc import ABC, abstractmethod -from typing import Any, List, Union, cast +from typing import Any, Generic, List, TypeVar, Union, cast from loguru import logger from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven + +# Should be a TypedDict +TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any]) -class BaseLLMAdapter(ABC): +class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]): """Abstract base class for LLM provider adapters. - Provides a standard interface for converting between Pipecat's standardized - tool schemas and provider-specific tool formats. Subclasses must implement - provider-specific conversion logic. + Provides a standard interface for converting to provider-specific formats. + + Handles: + + - Extracting provider-specific parameters for LLM invocation from a + universal LLM context + - Converting standardized tools schema to provider-specific tool formats. + - Extracting messages from the LLM context for the purposes of logging + about the specific provider. + + Subclasses must implement provider-specific conversion logic. """ + @abstractmethod + def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams: + """Get provider-specific LLM invocation parameters from a universal LLM context. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Provider-specific parameters for invoking the LLM. + """ + pass + @abstractmethod def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]: """Convert tools schema to the provider's specific format. @@ -38,7 +62,20 @@ class BaseLLMAdapter(ABC): """ pass - def from_standard_tools(self, tools: Any) -> List[Any]: + @abstractmethod + def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about this provider. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about this + provider. + """ + pass + + def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven: """Convert tools from standard format to provider format. Args: diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index fb5abe108..4ba73956b 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -6,20 +6,58 @@ """Anthropic LLM adapter for Pipecat.""" -from typing import Any, Dict, List +from typing import Any, Dict, List, TypedDict from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.processors.aggregators.llm_context import LLMContext -class AnthropicLLMAdapter(BaseLLMAdapter): +class AnthropicLLMInvocationParams(TypedDict): + """Context-based parameters for invoking Anthropic's LLM API. + + This is a placeholder until support for universal LLMContext machinery is added for Anthropic. + """ + + pass + + +class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): """Adapter for converting tool schemas to Anthropic's function-calling format. This adapter handles the conversion of Pipecat's standard function schemas to the specific format required by Anthropic's Claude models for function calling. """ + def get_llm_invocation_params(self, context: LLMContext) -> AnthropicLLMInvocationParams: + """Get Anthropic-specific LLM invocation parameters from a universal LLM context. + + This is a placeholder until support for universal LLMContext machinery is added for Anthropic. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Dictionary of parameters for invoking Anthropic's LLM API. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + + def get_messages_for_logging(self, context) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about Anthropic. + + Removes or truncates sensitive data like image content for safe logging. + + This is a placeholder until support for universal LLMContext machinery is added for Anthropic. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about Anthropic. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + @staticmethod def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]: """Convert a single function schema to Anthropic's format. diff --git a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py index 2875f8272..2627052eb 100644 --- a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py +++ b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py @@ -7,20 +7,58 @@ """AWS Nova Sonic LLM adapter for Pipecat.""" import json -from typing import Any, Dict, List +from typing import Any, Dict, List, TypedDict from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.processors.aggregators.llm_context import LLMContext -class AWSNovaSonicLLMAdapter(BaseLLMAdapter): +class AWSNovaSonicLLMInvocationParams(TypedDict): + """Context-based parameters for invoking AWS Nova Sonic LLM API. + + This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic. + """ + + pass + + +class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]): """Adapter for AWS Nova Sonic language models. Converts Pipecat's standard function schemas into AWS Nova Sonic's specific function-calling format, enabling tool use with Nova Sonic models. """ + def get_llm_invocation_params(self, context: LLMContext) -> AWSNovaSonicLLMInvocationParams: + """Get AWS Nova Sonic-specific LLM invocation parameters from a universal LLM context. + + This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Dictionary of parameters for invoking AWS Nova Sonic's LLM API. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for AWS Nova Sonic.") + + def get_messages_for_logging(self, context) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about AWS Nova Sonic. + + Removes or truncates sensitive data like image content for safe logging. + + This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about AWS Nova Sonic. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for AWS Nova Sonic.") + @staticmethod def _to_aws_nova_sonic_function_format(function: FunctionSchema) -> Dict[str, Any]: """Convert a function schema to AWS Nova Sonic format. diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index 364ad87d2..5f556ec9b 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/src/pipecat/adapters/services/bedrock_adapter.py @@ -6,20 +6,58 @@ """AWS Bedrock LLM adapter for Pipecat.""" -from typing import Any, Dict, List +from typing import Any, Dict, List, TypedDict from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.processors.aggregators.llm_context import LLMContext -class AWSBedrockLLMAdapter(BaseLLMAdapter): +class AWSBedrockLLMInvocationParams(TypedDict): + """Context-based parameters for invoking AWS Bedrock's LLM API. + + This is a placeholder until support for universal LLMContext machinery is added for Bedrock. + """ + + pass + + +class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): """Adapter for AWS Bedrock LLM integration with Pipecat. Provides conversion utilities for transforming Pipecat function schemas into AWS Bedrock's expected tool format for function calling capabilities. """ + def get_llm_invocation_params(self, context: LLMContext) -> AWSBedrockLLMInvocationParams: + """Get AWS Bedrock-specific LLM invocation parameters from a universal LLM context. + + This is a placeholder until support for universal LLMContext machinery is added for Bedrock. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Dictionary of parameters for invoking AWS Bedrock's LLM API. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.") + + def get_messages_for_logging(self, context) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about AWS Bedrock. + + Removes or truncates sensitive data like image content for safe logging. + + This is a placeholder until support for universal LLMContext machinery is added for Bedrock. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about AWS Bedrock. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.") + @staticmethod def _to_bedrock_function_format(function: FunctionSchema) -> Dict[str, Any]: """Convert a function schema to Bedrock's tool format. diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index 2139e0057..31345821f 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -6,20 +6,71 @@ """Gemini LLM adapter for Pipecat.""" -from typing import Any, Dict, List, Union +import base64 +import json +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, TypedDict + +from loguru import logger +from openai import NotGiven from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMContextMessage, + LLMSpecificMessage, + LLMStandardMessage, +) + +try: + from google.genai.types import ( + Blob, + Content, + FunctionCall, + FunctionResponse, + Part, + ) +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.") + raise Exception(f"Missing module: {e}") -class GeminiLLMAdapter(BaseLLMAdapter): - """LLM adapter for Google's Gemini service. +class GeminiLLMInvocationParams(TypedDict): + """Context-based parameters for invoking Gemini LLM.""" - Provides tool schema conversion functionality to transform standard tool - definitions into Gemini's specific function-calling format for use with - Gemini LLM models. + system_instruction: Optional[str] + messages: List[Content] + tools: List[Any] | NotGiven + + +class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): + """Gemini-specific adapter for Pipecat. + + Handles: + - Extracting parameters for Gemini's API from a universal LLM context + - Converting Pipecat's standardized tools schema to Gemini's function-calling format. + - Extracting and sanitizing messages from the LLM context for logging with Gemini. """ + def get_llm_invocation_params(self, context: LLMContext) -> GeminiLLMInvocationParams: + """Get Gemini-specific LLM invocation parameters from a universal LLM context. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Dictionary of parameters for Gemini's API. + """ + messages = self._from_universal_context_messages(self._get_messages(context)) + return { + "system_instruction": messages.system_instruction, + "messages": messages.messages, + # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) + "tools": self.from_standard_tools(context.tools), + } + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]: """Convert tool schemas to Gemini's function-calling format. @@ -39,3 +90,223 @@ class GeminiLLMAdapter(BaseLLMAdapter): custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, []) return formatted_standard_tools + custom_gemini_tools + + def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about Gemini. + + Removes or truncates sensitive data like image content for safe logging. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about Gemini. + """ + # Get messages in Gemini's format + messages = self._from_universal_context_messages(self._get_messages(context)).messages + + # Sanitize messages for logging + messages_for_logging = [] + for message in messages: + obj = message.to_json_dict() + try: + if "parts" in obj: + for part in obj["parts"]: + if "inline_data" in part: + part["inline_data"]["data"] = "..." + except Exception as e: + logger.debug(f"Error: {e}") + messages_for_logging.append(obj) + return messages_for_logging + + def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + return context.get_messages("google") + + @dataclass + class ConvertedMessages: + """Container for Google-formatted messages converted from universal context.""" + + messages: List[Content] + system_instruction: Optional[str] = None + + def _from_universal_context_messages( + self, universal_context_messages: List[LLMContextMessage] + ) -> ConvertedMessages: + """Restructures messages to ensure proper Google format and message ordering. + + This method handles conversion of OpenAI-formatted messages to Google format, + with special handling for function calls, function responses, and system messages. + System messages are added back to the context as user messages when needed. + + The final message order is preserved as: + + 1. Function calls (from model) + 2. Function responses (from user) + 3. Text messages (converted from system messages) + + Note:: + + System messages are only added back when there are no regular text + messages in the context, ensuring proper conversation continuity + after function calls. + """ + system_instruction = None + messages = [] + + # Process each message, preserving Google-formatted messages and converting others + for message in universal_context_messages: + if isinstance(message, LLMSpecificMessage): + # Assume that LLMSpecificMessage wraps a message in Google format + messages.append(message.message) + continue + + # Convert standard format to Google format + converted = self._from_standard_message( + message, already_have_system_instruction=bool(system_instruction) + ) + if isinstance(converted, Content): + # Regular (non-system) message + messages.append(converted) + else: + # System instruction + system_instruction = converted + + # Check if we only have function-related messages (no regular text) + has_regular_messages = any( + len(msg.parts) == 1 + and getattr(msg.parts[0], "text", None) + and not getattr(msg.parts[0], "function_call", None) + and not getattr(msg.parts[0], "function_response", None) + for msg in messages + ) + + # Add system instruction back as a user message if we only have function messages + if system_instruction and not has_regular_messages: + messages.append(Content(role="user", parts=[Part(text=system_instruction)])) + + # Remove any empty messages + messages = [m for m in messages if m.parts] + + return self.ConvertedMessages(messages=messages, system_instruction=system_instruction) + + def _from_standard_message( + self, message: LLMStandardMessage, already_have_system_instruction: bool + ) -> Content | str: + """Convert universal context message to Google Content object. + + Handles conversion of text, images, and function calls to Google's + format. + System instructions are returned as a plain string. + + Args: + message: Message in universal context format. + already_have_system_instruction: Whether we already have a system instruction + + Returns: + Content object with role and parts, or a plain string for system + messages. + + Examples: + Standard text message:: + + { + "role": "user", + "content": "Hello there" + } + + Converts to Google Content with:: + + Content( + role="user", + parts=[Part(text="Hello there")] + ) + + Standard function call message:: + + { + "role": "assistant", + "tool_calls": [ + { + "function": { + "name": "search", + "arguments": '{"query": "test"}' + } + } + ] + } + + Converts to Google Content with:: + + Content( + role="model", + parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))] + ) + """ + role = message["role"] + content = message.get("content", []) + if role == "system": + if already_have_system_instruction: + role = "user" # Convert system message to user role if we already have a system instruction + else: + # System instructions are returned as plain text + if isinstance(content, str): + return content + elif isinstance(content, list): + # If content is a list, we assume it's a list of text parts, per the standard + return " ".join(part["text"] for part in content if part.get("type") == "text") + elif role == "assistant": + role = "model" + + parts = [] + if message.get("tool_calls"): + for tc in message["tool_calls"]: + parts.append( + Part( + function_call=FunctionCall( + name=tc["function"]["name"], + args=json.loads(tc["function"]["arguments"]), + ) + ) + ) + elif role == "tool": + role = "model" + try: + response = json.loads(message["content"]) + if isinstance(response, dict): + response_dict = response + else: + response_dict = {"value": response} + except Exception as e: + # Response might not be JSON-deserializable. + # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string. + response_dict = {"value": message["content"]} + parts.append( + Part( + function_response=FunctionResponse( + name="tool_call_result", # seems to work to hard-code the same name every time + response=response_dict, + ) + ) + ) + elif isinstance(content, str): + parts.append(Part(text=content)) + elif isinstance(content, list): + for c in content: + if c["type"] == "text": + parts.append(Part(text=c["text"])) + elif c["type"] == "image_url": + parts.append( + Part( + inline_data=Blob( + mime_type="image/jpeg", + data=base64.b64decode(c["image_url"]["url"].split(",")[1]), + ) + ) + ) + elif c["type"] == "input_audio": + input_audio = c["input_audio"] + audio_bytes = base64.b64decode(input_audio["data"]) + parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes))) + + message = Content(role=role, parts=parts) + return message diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py index 59d70aa1e..2ba5b0319 100644 --- a/src/pipecat/adapters/services/open_ai_adapter.py +++ b/src/pipecat/adapters/services/open_ai_adapter.py @@ -6,22 +6,63 @@ """OpenAI LLM adapter for Pipecat.""" -from typing import List +import copy +import json +from typing import Any, List, TypedDict -from openai.types.chat import ChatCompletionToolParam +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 pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMContextMessage, + LLMContextToolChoice, + NotGiven, +) -class OpenAILLMAdapter(BaseLLMAdapter): - """Adapter for converting tool schemas to OpenAI's format. +class OpenAILLMInvocationParams(TypedDict): + """Context-based parameters for invoking OpenAI ChatCompletion API.""" - Provides conversion utilities for transforming Pipecat's standard tool - schemas into the format expected by OpenAI's ChatCompletion API for - function calling capabilities. + messages: List[ChatCompletionMessageParam] + tools: List[ChatCompletionToolParam] | OpenAINotGiven + tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven + + +class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]): + """OpenAI-specific adapter for Pipecat. + + Handles: + + - Extracting parameters for OpenAI's ChatCompletion API from a universal + LLM context + - Converting Pipecat's standardized tools schema to OpenAI's function-calling format. + - Extracting and sanitizing messages from the LLM context for logging about OpenAI. """ + def get_llm_invocation_params(self, context: LLMContext) -> OpenAILLMInvocationParams: + """Get OpenAI-specific LLM invocation parameters from a universal LLM context. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Dictionary of parameters for OpenAI's ChatCompletion API. + """ + return { + "messages": self._from_universal_context_messages(self._get_messages(context)), + # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) + "tools": self.from_standard_tools(context.tools), + "tool_choice": context.tool_choice, + } + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]: """Convert function schemas to OpenAI's function-calling format. @@ -37,3 +78,43 @@ class OpenAILLMAdapter(BaseLLMAdapter): ChatCompletionToolParam(type="function", function=func.to_default_dict()) for func in functions_schema ] + + def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about OpenAI. + + Removes or truncates sensitive data like image content for safe logging. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about OpenAI. + """ + msgs = [] + for message in self._get_messages(context): + msg = copy.deepcopy(message) + if "content" in msg: + if isinstance(msg["content"], list): + for item in msg["content"]: + if item["type"] == "image_url": + if item["image_url"]["url"].startswith("data:image/"): + item["image_url"]["url"] = "data:image/..." + if "mime_type" in msg and msg["mime_type"].startswith("image/"): + msg["data"] = "..." + msgs.append(msg) + return json.dumps(msgs, ensure_ascii=False) + + def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + return context.get_messages("openai") + + def _from_universal_context_messages( + self, messages: List[LLMContextMessage] + ) -> List[ChatCompletionMessageParam]: + # Just a pass-through: messages are already the right type + return messages + + def _from_standard_tool_choice( + self, tool_choice: LLMContextToolChoice | NotGiven + ) -> ChatCompletionToolChoiceOptionParam | OpenAINotGiven: + # Just a pass-through: tool_choice is already the right type + return tool_choice diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 58aea5a9a..705df525e 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -6,11 +6,21 @@ """OpenAI Realtime LLM adapter for Pipecat.""" -from typing import Any, Dict, List, Union +from typing import Any, Dict, List, TypedDict, Union from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.processors.aggregators.llm_context import LLMContext + + +class OpenAIRealtimeLLMInvocationParams(TypedDict): + """Context-based parameters for invoking OpenAI Realtime API. + + This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime. + """ + + pass class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): @@ -20,6 +30,34 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): OpenAI's Realtime API for function calling capabilities. """ + def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams: + """Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context. + + This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime. + + Args: + context: The LLM context containing messages, tools, etc. + + Returns: + Dictionary of parameters for invoking OpenAI Realtime's API. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.") + + def get_messages_for_logging(self, context) -> List[dict[str, Any]]: + """Get messages from a universal LLM context in a format ready for logging about OpenAI Realtime. + + Removes or truncates sensitive data like image content for safe logging. + + This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime. + + Args: + context: The LLM context containing messages. + + Returns: + List of messages in a format ready for logging about OpenAI Realtime. + """ + raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.") + @staticmethod def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]: """Convert a function schema to OpenAI Realtime format. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index d1d3806d5..01c840a71 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -36,6 +36,7 @@ from pipecat.utils.time import nanoseconds_to_str from pipecat.utils.utils import obj_count, obj_id if TYPE_CHECKING: + from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor @@ -403,6 +404,11 @@ class OpenAILLMContextAssistantTimestampFrame(DataFrame): timestamp: str +# A more universal (LLM-agnostic) name for +# OpenAILLMContextAssistantTimestampFrame, matching LLMContext +LLMContextAssistantTimestampFrame = OpenAILLMContextAssistantTimestampFrame + + @dataclass class TranscriptionMessage: """A message in a conversation transcript. @@ -474,6 +480,20 @@ class TranscriptionUpdateFrame(DataFrame): return f"{self.name}(pts: {pts}, messages: {len(self.messages)})" +@dataclass +class LLMContextFrame(Frame): + """Frame containing a universal 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" + + @dataclass class LLMMessagesFrame(DataFrame): """Frame containing LLM messages for chat completion. @@ -1445,3 +1465,20 @@ class MixerEnableFrame(MixerControlFrame): """ enable: bool + + +@dataclass +class ServiceSwitcherFrame(ControlFrame): + """A base class for frames that control ServiceSwitcher behavior.""" + + pass + + +@dataclass +class ManuallySwitchServiceFrame(ServiceSwitcherFrame): + """A frame to request a manual switch in the active service in a ServiceSwitcher. + + Handled by ServiceSwitcherStrategyManual to switch the active service. + """ + + service: "FrameProcessor" diff --git a/src/pipecat/pipeline/llm_switcher.py b/src/pipecat/pipeline/llm_switcher.py new file mode 100644 index 000000000..d1906119a --- /dev/null +++ b/src/pipecat/pipeline/llm_switcher.py @@ -0,0 +1,84 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""LLM switcher for switching between different LLMs at runtime, with different switching strategies.""" + +from typing import Any, List, Optional, Type + +from pipecat.pipeline.service_switcher import ServiceSwitcher, StrategyType +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.services.llm_service import LLMService + + +class LLMSwitcher(ServiceSwitcher[StrategyType]): + """A pipeline that switches between different LLMs at runtime.""" + + def __init__(self, llms: List[LLMService], strategy_type: Type[StrategyType]): + """Initialize the service switcher with a list of LLMs and a switching strategy.""" + super().__init__(llms, strategy_type) + + @property + def llms(self) -> List[LLMService]: + """Get the list of LLMs managed by this switcher.""" + return self.services + + @property + def active_llm(self) -> Optional[LLMService]: + """Get the currently active LLM, if any.""" + return self.strategy.active_service + + async def run_inference( + self, context: LLMContext, system_instruction: Optional[str] = None + ) -> Optional[str]: + """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM. + + Args: + context: The LLM context containing conversation history. + system_instruction: Optional system instruction to guide the LLM's + behavior. You could also (again, optionally) provide a system + instruction directly in the context. If both are provided, the + one in the context takes precedence. + + Returns: + The LLM's response as a string, or None if no response is generated. + """ + if self.active_llm: + return await self.active_llm.run_inference( + context=context, system_instruction=system_instruction + ) + return None + + def register_function( + self, + function_name: Optional[str], + handler: Any, + start_callback=None, + *, + cancel_on_interruption: bool = True, + ): + """Register a function handler for LLM function calls, on all LLMs, active or not. + + Args: + function_name: The name of the function to handle. Use None to handle + all function calls with a catch-all handler. + handler: The function handler. Should accept a single FunctionCallParams + parameter. + start_callback: Legacy callback function (deprecated). Put initialization + code at the top of your handler instead. + + .. deprecated:: 0.0.59 + The `start_callback` parameter is deprecated and will be removed in a future version. + + cancel_on_interruption: Whether to cancel this function call when an + interruption occurs. Defaults to True. + """ + for llm in self.llms: + llm.register_function( + function_name=function_name, + handler=handler, + start_callback=start_callback, + cancel_on_interruption=cancel_on_interruption, + ) diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py new file mode 100644 index 000000000..7dd36c503 --- /dev/null +++ b/src/pipecat/pipeline/service_switcher.py @@ -0,0 +1,144 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Service switcher for switching between different services at runtime, with different switching strategies.""" + +from typing import Any, Generic, List, Optional, Type, TypeVar + +from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, ServiceSwitcherFrame +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.processors.filters.function_filter import FunctionFilter +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class ServiceSwitcherStrategy: + """Base class for service switching strategies.""" + + def __init__(self, services: List[FrameProcessor]): + """Initialize the service switcher strategy with a list of services.""" + self.services = services + self.active_service: Optional[FrameProcessor] = None + + def is_active(self, service: FrameProcessor) -> bool: + """Determine if the given service is the currently active one. + + This method should be overridden by subclasses to implement specific logic. + + Args: + service: The service to check. + + Returns: + True if the given service is the active one, False otherwise. + """ + raise NotImplementedError("Subclasses must implement this method.") + + def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): + """Handle a frame that controls service switching. + + This method can be overridden by subclasses to implement specific logic + for handling frames that control service switching. + + Args: + frame: The frame to handle. + direction: The direction of the frame (upstream or downstream). + """ + raise NotImplementedError("Subclasses must implement this method.") + + +class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy): + """A strategy for switching between services manually. + + This strategy allows the user to manually select which service is active. + The initial active service is the first one in the list. + """ + + def __init__(self, services: List[FrameProcessor]): + """Initialize the manual service switcher strategy with a list of services.""" + super().__init__(services) + self.active_service = services[0] if services else None + + def is_active(self, service: FrameProcessor) -> bool: + """Check if the given service is the currently active one. + + Args: + service: The service to check. + + Returns: + True if the given service is the active one, False otherwise. + """ + return service == self.active_service + + def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection): + """Handle a frame that controls service switching. + + Args: + frame: The frame to handle. + direction: The direction of the frame (upstream or downstream). + """ + if isinstance(frame, ManuallySwitchServiceFrame): + self._set_active(frame.service) + else: + raise ValueError(f"Unsupported frame type: {type(frame)}") + + def _set_active(self, service: FrameProcessor): + """Set the active service to the given one. + + Args: + service: The service to set as active. + """ + if service in self.services: + self.active_service = service + else: + raise ValueError(f"Service {service} is not in the list of available services.") + + +StrategyType = TypeVar("StrategyType", bound=ServiceSwitcherStrategy) + + +class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]): + """A pipeline that switches between different services at runtime.""" + + def __init__(self, services: List[FrameProcessor], strategy_type: Type[StrategyType]): + """Initialize the service switcher with a list of services and a switching strategy.""" + strategy = strategy_type(services) + super().__init__(*self._make_pipeline_definitions(services, strategy)) + self.services = services + self.strategy = strategy + + @staticmethod + def _make_pipeline_definitions( + services: List[FrameProcessor], strategy: ServiceSwitcherStrategy + ) -> List[Any]: + pipelines = [] + for service in services: + pipelines.append(ServiceSwitcher._make_pipeline_definition(service, strategy)) + return pipelines + + @staticmethod + def _make_pipeline_definition( + service: FrameProcessor, strategy: ServiceSwitcherStrategy + ) -> Any: + async def filter(frame) -> bool: + _ = frame + return strategy.is_active(service) + + return [ + FunctionFilter(filter, direction=FrameDirection.DOWNSTREAM), + service, + FunctionFilter(filter, direction=FrameDirection.UPSTREAM), + ] + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process a frame, handling frames which affect service switching. + + Args: + frame: The frame to process. + direction: The direction of the frame (upstream or downstream). + """ + await super().process_frame(frame, direction) + + if isinstance(frame, ServiceSwitcherFrame): + self.strategy.handle_frame(frame, direction) diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py new file mode 100644 index 000000000..8b677cf02 --- /dev/null +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -0,0 +1,277 @@ +# +# 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 universal format (based on 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 Any, List, Optional, TypeAlias, Union + +from loguru import logger +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, +) +from PIL import Image + +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.frames.frames import AudioRawFrame + +# "Re-export" types from OpenAI that we're using as universal context types. +# NOTE: if universal message types need to someday diverge from OpenAI's, we +# should consider managing our own definitions. But we should do so carefully, +# as the OpenAI messages are somewhat of a standard and we want to continue +# supporting them. +LLMStandardMessage = ChatCompletionMessageParam +LLMContextToolChoice = ChatCompletionToolChoiceOptionParam +NOT_GIVEN = OPEN_AI_NOT_GIVEN +NotGiven = OpenAINotGiven + + +@dataclass +class LLMSpecificMessage: + """A container for a context message that is specific to a particular LLM service. + + Enables the use of service-specific message types while maintaining + compatibility with the universal LLM context format. + """ + + llm: str + message: Any + + +LLMContextMessage: TypeAlias = Union[LLMStandardMessage, LLMSpecificMessage] + + +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: ToolsSchema | NotGiven = 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: ToolsSchema | NotGiven = LLMContext._normalize_and_validate_tools(tools) + self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice + + def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]: + """Get the current messages list. + + Args: + llm_specific_filter: Optional filter to return LLM-specific + messages for the given LLM, in addition to the standard + messages. If messages end up being filtered, an error will be + logged. + + Returns: + List of conversation messages. + """ + if llm_specific_filter is None: + return self._messages + filtered_messages = [ + msg + for msg in self._messages + if not isinstance(msg, LLMSpecificMessage) or msg.llm == llm_specific_filter + ] + if len(filtered_messages) < len(self._messages): + logger.error( + f"Attempted to use incompatible LLMSpecificMessages with LLM '{llm_specific_filter}'." + ) + return filtered_messages + + @property + def tools(self) -> ToolsSchema | NotGiven: + """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: ToolsSchema | NotGiven = NOT_GIVEN): + """Set the available tools for the LLM. + + Args: + tools: A ToolsSchema or NOT_GIVEN to disable tools. + """ + self._tools = LLMContext._normalize_and_validate_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 = "Audio follows" + ): + """Add a message containing audio frames. + + Args: + audio_frames: List of audio frame objects to include. + text: Optional text to include with the audio. + """ + if not audio_frames: + return + + sample_rate = audio_frames[0].sample_rate + num_channels = audio_frames[0].num_channels + + content = [] + content.append({"type": "text", "text": text}) + data = b"".join(frame.audio for frame in audio_frames) + data = bytes( + self._create_wav_header( + sample_rate, + num_channels, + 16, + len(data), + ) + + data + ) + encoded_audio = base64.b64encode(data).decode("utf-8") + content.append( + { + "type": "input_audio", + "input_audio": {"data": encoded_audio, "format": "wav"}, + } + ) + self.add_message({"role": "user", "content": content}) + + def _create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size): + """Create a WAV file header for audio data. + + Args: + sample_rate: Audio sample rate in Hz. + num_channels: Number of audio channels. + bits_per_sample: Bits per audio sample. + data_size: Size of audio data in bytes. + + Returns: + WAV header as a bytearray. + """ + # RIFF chunk descriptor + header = bytearray() + header.extend(b"RIFF") # ChunkID + header.extend((data_size + 36).to_bytes(4, "little")) # ChunkSize: total size - 8 + header.extend(b"WAVE") # Format + # "fmt " sub-chunk + header.extend(b"fmt ") # Subchunk1ID + header.extend((16).to_bytes(4, "little")) # Subchunk1Size (16 for PCM) + header.extend((1).to_bytes(2, "little")) # AudioFormat (1 for PCM) + header.extend(num_channels.to_bytes(2, "little")) # NumChannels + header.extend(sample_rate.to_bytes(4, "little")) # SampleRate + # Calculate byte rate and block align + byte_rate = sample_rate * num_channels * (bits_per_sample // 8) + block_align = num_channels * (bits_per_sample // 8) + header.extend(byte_rate.to_bytes(4, "little")) # ByteRate + header.extend(block_align.to_bytes(2, "little")) # BlockAlign + header.extend(bits_per_sample.to_bytes(2, "little")) # BitsPerSample + # "data" sub-chunk + header.extend(b"data") # Subchunk2ID + header.extend(data_size.to_bytes(4, "little")) # Subchunk2Size + return header + + @staticmethod + def _normalize_and_validate_tools(tools: ToolsSchema | NotGiven) -> ToolsSchema | NotGiven: + """Normalize and validate the given tools. + + Raises: + TypeError: If tools are not a ToolsSchema or NotGiven. + """ + if isinstance(tools, ToolsSchema): + if not tools.standard_tools and not tools.custom_tools: + return NOT_GIVEN + return tools + elif tools is NOT_GIVEN: + return NOT_GIVEN + else: + raise TypeError( + f"In LLMContext, tools must be a ToolsSchema object or NOT_GIVEN. Got type: {type(tools)}", + ) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py new file mode 100644 index 000000000..e84194b78 --- /dev/null +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -0,0 +1,827 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""LLM response aggregators for handling conversation context and message aggregation. + +This module provides aggregators that process and accumulate LLM responses, user inputs, +and conversation context. These aggregators handle the flow between speech-to-text, +LLM processing, and text-to-speech components in conversational AI pipelines. +""" + +import asyncio +import json +from dataclasses import dataclass +from typing import Any, Dict, List, Literal, Optional, Set + +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, + LLMContextAssistantTimestampFrame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesAppendFrame, + LLMMessagesUpdateFrame, + LLMSetToolChoiceFrame, + LLMSetToolsFrame, + SpeechControlParamsFrame, + StartFrame, + StartInterruptionFrame, + TextFrame, + TranscriptionFrame, + UserImageRawFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMContextMessage, + LLMSpecificMessage, +) +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.time import time_now_iso8601 + + +class LLMContextAggregator(FrameProcessor): + """Base LLM aggregator that uses an LLMContext for conversation storage. + + 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: LLMContext, role: str, **kwargs): + """Initialize the context response aggregator. + + Args: + 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. + """ + super().__init__(**kwargs) + self._context = context + self._role = role + + self._aggregation: str = "" + + @property + def messages(self) -> List[LLMContextMessage]: + """Get messages from the LLM context. + + Returns: + List of message dictionaries from the context. + """ + return self._context.get_messages() + + @property + def role(self) -> str: + """Get the role for this aggregator. + + Returns: + The role string for this aggregator. + """ + return self._role + + @property + def context(self): + """Get the LLM context. + + Returns: + The LLMContext instance used by this aggregator. + """ + return self._context + + def get_context_frame(self) -> LLMContextFrame: + """Create a context frame with the current context. + + Returns: + LLMContextFrame containing the current context. + """ + return LLMContextFrame(context=self._context) + + async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a context frame in the specified direction. + + Args: + direction: The direction to push the frame (upstream or downstream). + """ + frame = self.get_context_frame() + await self.push_frame(frame, direction) + + def add_messages(self, messages): + """Add messages to the context. + + Args: + messages: Messages to add to the conversation context. + """ + self._context.add_messages(messages) + + def set_messages(self, messages): + """Set the context messages. + + Args: + messages: Messages to replace the current context messages. + """ + self._context.set_messages(messages) + + def set_tools(self, tools: List): + """Set tools in the context. + + Args: + tools: List of tool definitions to set in the context. + """ + self._context.set_tools(tools) + + def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict): + """Set tool choice in the context. + + Args: + tool_choice: Tool choice configuration for the context. + """ + self._context.set_tool_choice(tool_choice) + + async def reset(self): + """Reset the aggregation state.""" + self._aggregation = "" + + +class LLMUserAggregator(LLMContextAggregator): + """User LLM aggregator that processes speech-to-text transcriptions. + + This aggregator handles the complex logic of aggregating user speech transcriptions + from STT services. It manages multiple scenarios including: + + - Transcriptions received between VAD events + - Transcriptions received outside VAD events + - Interim vs final transcriptions + - User interruptions during bot speech + - Emulated VAD for whispered or short utterances + + The aggregator uses timeouts to handle cases where transcriptions arrive + after VAD events or when no VAD is available. + """ + + def __init__( + self, + context: LLMContext, + *, + params: Optional[LLMUserAggregatorParams] = None, + **kwargs, + ): + """Initialize the user context aggregator. + + Args: + 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._vad_params: Optional[VADParams] = None + self._turn_params: Optional[SmartTurnParams] = None + + if "aggregation_timeout" in kwargs: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter 'aggregation_timeout' is deprecated, use 'params' instead.", + DeprecationWarning, + ) + + self._params.aggregation_timeout = kwargs["aggregation_timeout"] + + self._user_speaking = False + self._bot_speaking = False + self._was_bot_speaking = False + self._emulating_vad = False + self._seen_interim_results = False + self._waiting_for_aggregation = False + + self._aggregation_event = asyncio.Event() + self._aggregation_task = None + + async def reset(self): + """Reset the aggregation state and interruption strategies.""" + await super().reset() + self._was_bot_speaking = False + self._seen_interim_results = False + self._waiting_for_aggregation = False + [await s.reset() for s in self._interruption_strategies] + + 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() + self._context.add_message({"role": self.role, "content": aggregation}) + frame = LLMContextFrame(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(LLMContextFrame(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=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._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. Behavior when bot is speaking depends on the + enable_emulated_vad_interruptions parameter. + """ + # 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 and not self._params.enable_emulated_vad_interruptions: + # If emulated VAD interruptions are disabled and bot is speaking, ignore + logger.debug("Ignoring user speaking emulation, bot is speaking.") + await self.reset() + else: + # Either bot is not speaking, or emulated VAD interruptions are enabled + # - trigger user speaking emulation. + await self.push_frame(EmulateUserStartedSpeakingFrame(), FrameDirection.UPSTREAM) + self._emulating_vad = True + + +class LLMAssistantAggregator(LLMContextAggregator): + """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: LLMContext, + *, + 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 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: + self._context.add_message({"role": "assistant", "content": aggregation}) + + # Push context frame + await self.push_context_frame() + + # Push timestamp frame with current time + timestamp_frame = LLMContextAssistantTimestampFrame(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}]" + ) + + # Update context with the in-progress function call + 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, + } + ) + + 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 + + # Update context with the function call result + if frame.result: + result = json.dumps(frame.result) + self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + else: + self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED") + + 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: + # Update context with the function call cancellation + self._update_function_call_result(frame.function_name, frame.tool_call_id, "CANCELLED") + del self._function_calls_in_progress[frame.tool_call_id] + + def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any): + for message in self._context.get_messages(): + if ( + not isinstance(message, LLMSpecificMessage) + and message["role"] == "tool" + and message["tool_call_id"] + and message["tool_call_id"] == tool_call_id + ): + message["content"] = result + + 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] + + # Update context with the image frame + self._update_function_call_result( + frame.request.function_name, frame.request.tool_call_id, "COMPLETED" + ) + self._context.add_image_frame_message( + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.request.context, + ) + + 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) + + +class LLMContextAggregatorPair: + """Pair of LLM context aggregators for updating context with user and assistant messages.""" + + def __init__( + self, + context: LLMContext, + *, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), + ): + """Initialize the LLM context aggregator pair. + + Args: + context: The context to be managed by the aggregators. + user_params: Parameters for the user context aggregator. + assistant_params: Parameters for the assistant context aggregator. + """ + self._user = LLMUserAggregator(context, params=user_params) + self._assistant = LLMAssistantAggregator(context, params=assistant_params) + + def user(self) -> LLMUserAggregator: + """Get the user context aggregator. + + Returns: + The user context aggregator instance. + """ + return self._user + + def assistant(self) -> LLMAssistantAggregator: + """Get the assistant context aggregator. + + Returns: + The assistant context aggregator instance. + """ + return self._assistant diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index cd65e27ab..2d91dd380 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -42,6 +42,7 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, InputAudioRawFrame, InterimTranscriptionFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, @@ -916,7 +917,10 @@ class RTVIObserver(BaseObserver): and self._params.user_transcription_enabled ): await self._handle_user_transcriptions(frame) - elif isinstance(frame, OpenAILLMContextFrame) and self._params.user_llm_enabled: + elif ( + isinstance(frame, (OpenAILLMContextFrame, LLMContextFrame)) + and self._params.user_llm_enabled + ): await self._handle_context(frame) elif isinstance(frame, LLMFullResponseStartFrame) and self._params.bot_llm_enabled: await self.push_transport_message_urgent(RTVIBotLLMStartedMessage()) @@ -1017,16 +1021,20 @@ class RTVIObserver(BaseObserver): if message: await self.push_transport_message_urgent(message) - async def _handle_context(self, frame: OpenAILLMContextFrame): + async def _handle_context(self, frame: OpenAILLMContextFrame | LLMContextFrame): """Process LLM context frames to extract user messages for the RTVI client.""" try: - messages = frame.context.messages + if isinstance(frame, OpenAILLMContextFrame): + messages = frame.context.messages + else: + messages = frame.context.get_messages() if not messages: return message = messages[-1] # Handle Google LLM format (protobuf objects with attributes) + # Note: not possible if frame is a universal LLMContextFrame if hasattr(message, "role") and message.role == "user" and hasattr(message, "parts"): text = "".join(part.text for part in message.parts if hasattr(part, "text")) if text: diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 3a26519f6..ee042aa1b 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -31,6 +31,7 @@ from pipecat.frames.frames import ( FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, + LLMContextFrame, LLMEnablePromptCachingFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -41,6 +42,7 @@ from pipecat.frames.frames import ( VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMAssistantContextAggregator, @@ -197,6 +199,46 @@ class AnthropicLLMService(LLMService): response = await api_call(**params) return response + async def run_inference( + self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None + ) -> Optional[str]: + """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context. + + Args: + context: The LLM context containing conversation history. + system_instruction: Optional system instruction to guide the LLM's + behavior. You could also (again, optionally) provide a system + instruction directly in the context. If both are provided, the + one in the context takes precedence. + + Returns: + The LLM's response as a string, or None if no response is generated. + """ + messages = [] + system = [] + if isinstance(context, LLMContext): + # Future code will be something like this: + # adapter = self.get_llm_adapter() + # params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(context) + # messages = params["messages"] + # system = params["system_instruction"] + raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") + else: + context = AnthropicLLMContext.upgrade_to_anthropic(context) + messages = context.messages + system = getattr(context, "system", None) or system_instruction + + # LLM completion + response = await self._client.messages.create( + model=self.model_name, + messages=messages, + system=system, + max_tokens=8192, + stream=False, + ) + + return response.content[0].text + @property def enable_prompt_caching_beta(self) -> bool: """Check if prompt caching beta feature is enabled. @@ -408,6 +450,8 @@ class AnthropicLLMService(LLMService): context = None if isinstance(frame, OpenAILLMContextFrame): context: "AnthropicLLMContext" = AnthropicLLMContext.upgrade_to_anthropic(frame.context) + elif isinstance(frame, LLMContextFrame): + raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.") elif isinstance(frame, LLMMessagesFrame): context = AnthropicLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index a4e1366c2..6e109c4c1 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -31,6 +31,7 @@ from pipecat.frames.frames import ( FunctionCallFromLLM, FunctionCallInProgressFrame, FunctionCallResultFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, @@ -40,6 +41,7 @@ from pipecat.frames.frames import ( VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMAssistantContextAggregator, @@ -789,6 +791,81 @@ class AWSBedrockLLMService(LLMService): """ return True + async def run_inference( + self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None + ) -> Optional[str]: + """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context. + + Args: + context: The LLM context containing conversation history. + system_instruction: Optional system instruction to guide the LLM's + behavior. You could also (again, optionally) provide a system + instruction directly in the context. If both are provided, the + one in the context takes precedence. + + Returns: + The LLM's response as a string, or None if no response is generated. + """ + try: + messages = [] + system = [] + if isinstance(context, LLMContext): + # Future code will be something like this: + # adapter = self.get_llm_adapter() + # params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(context) + # messages = params["messages"] + # system = params["system_instruction"] + raise NotImplementedError( + "Universal LLMContext is not yet supported for AWS Bedrock." + ) + else: + context = AWSBedrockLLMContext.upgrade_to_bedrock(context) + messages = context.messages + system = getattr(context, "system", None) or system_instruction + + # Determine if we're using Claude or Nova based on model ID + model_id = self.model_name + + # Prepare request parameters + request_params = { + "modelId": model_id, + "messages": messages, + "inferenceConfig": { + "maxTokens": 8192, + "temperature": 0.7, + "topP": 0.9, + }, + } + + if system: + request_params["system"] = [{"text": system}] + + async with self._aws_session.client( + service_name="bedrock-runtime", **self._aws_params + ) as client: + # Call Bedrock without streaming + response = await client.converse(**request_params) + + # Extract the response text + if ( + "output" in response + and "message" in response["output"] + and "content" in response["output"]["message"] + ): + content = response["output"]["message"]["content"] + if isinstance(content, list): + for item in content: + if item.get("text"): + return item["text"] + elif isinstance(content, str): + return content + + return None + + except Exception as e: + logger.error(f"Bedrock summary generation failed: {e}", exc_info=True) + return None + async def _create_converse_stream(self, client, request_params): """Create converse stream with optional timeout and retry. @@ -1044,6 +1121,8 @@ class AWSBedrockLLMService(LLMService): context = None if isinstance(frame, OpenAILLMContextFrame): context = AWSBedrockLLMContext.upgrade_to_bedrock(frame.context) + if isinstance(frame, LLMContextFrame): + raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.") elif isinstance(frame, LLMMessagesFrame): context = AWSBedrockLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py index 6ca6c9f61..acc76f1ce 100644 --- a/src/pipecat/services/aws_nova_sonic/aws.py +++ b/src/pipecat/services/aws_nova_sonic/aws.py @@ -34,6 +34,7 @@ from pipecat.frames.frames import ( FunctionCallFromLLM, InputAudioRawFrame, InterimTranscriptionFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, @@ -322,6 +323,10 @@ class AWSNovaSonicLLMService(LLMService): if isinstance(frame, OpenAILLMContextFrame): await self._handle_context(frame.context) + elif isinstance(frame, LLMContextFrame): + raise NotImplementedError( + "Universal LLMContext is not yet supported for AWS Nova Sonic." + ) elif isinstance(frame, InputAudioRawFrame): await self._handle_input_audio_frame(frame) elif isinstance(frame, BotStoppedSpeakingFrame): diff --git a/src/pipecat/services/azure/llm.py b/src/pipecat/services/azure/llm.py index a4b93f2a4..47a6ef280 100644 --- a/src/pipecat/services/azure/llm.py +++ b/src/pipecat/services/azure/llm.py @@ -60,3 +60,12 @@ class AzureLLMService(OpenAILLMService): azure_endpoint=self._endpoint, api_version=self._api_version, ) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as Azure service does yet not support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/cerebras/llm.py b/src/pipecat/services/cerebras/llm.py index c3577af82..9bdc5b963 100644 --- a/src/pipecat/services/cerebras/llm.py +++ b/src/pipecat/services/cerebras/llm.py @@ -9,9 +9,8 @@ from typing import List from loguru import logger -from openai.types.chat import ChatCompletionMessageParam -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.services.openai.llm import OpenAILLMService @@ -54,25 +53,40 @@ class CerebrasLLMService(OpenAILLMService): logger.debug(f"Creating Cerebras client with api {base_url}") return super().create_client(api_key, base_url, **kwargs) - def build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for Cerebras chat completion request. Cerebras supports a subset of OpenAI parameters, focusing on core completion settings without advanced features like frequency/presence penalties. + + Args: + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. + + Returns: + Dictionary of parameters for the chat completion request. """ params = { "model": self.model_name, "stream": True, - "messages": messages, - "tools": context.tools, - "tool_choice": context.tool_choice, "seed": self._settings["seed"], "temperature": self._settings["temperature"], "top_p": self._settings["top_p"], "max_completion_tokens": self._settings["max_completion_tokens"], } + # Messages, tools, tool_choice + params.update(params_from_context) + params.update(self._settings["extra"]) return params + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as Cerebras service does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/deepseek/llm.py b/src/pipecat/services/deepseek/llm.py index 1e6bfcc5c..7b616a293 100644 --- a/src/pipecat/services/deepseek/llm.py +++ b/src/pipecat/services/deepseek/llm.py @@ -9,9 +9,8 @@ from typing import List from loguru import logger -from openai.types.chat import ChatCompletionMessageParam -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.services.openai.llm import OpenAILLMService @@ -54,19 +53,22 @@ class DeepSeekLLMService(OpenAILLMService): logger.debug(f"Creating DeepSeek client with api {base_url}") return super().create_client(api_key, base_url, **kwargs) - def _build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def _build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for DeepSeek chat completion request. DeepSeek doesn't support some OpenAI parameters like seed and max_completion_tokens. + + Args: + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. + + Returns: + Dictionary of parameters for the chat completion request. """ params = { "model": self.model_name, "stream": True, - "messages": messages, - "tools": context.tools, - "tool_choice": context.tool_choice, "stream_options": {"include_usage": True}, "frequency_penalty": self._settings["frequency_penalty"], "presence_penalty": self._settings["presence_penalty"], @@ -75,5 +77,17 @@ class DeepSeekLLMService(OpenAILLMService): "max_tokens": self._settings["max_tokens"], } + # Messages, tools, tool_choice + params.update(params_from_context) + params.update(self._settings["extra"]) return params + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as DeepSeekLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/fireworks/llm.py b/src/pipecat/services/fireworks/llm.py index e28ff9759..194adfc51 100644 --- a/src/pipecat/services/fireworks/llm.py +++ b/src/pipecat/services/fireworks/llm.py @@ -9,9 +9,8 @@ from typing import List from loguru import logger -from openai.types.chat import ChatCompletionMessageParam -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.services.openai.llm import OpenAILLMService @@ -54,20 +53,23 @@ class FireworksLLMService(OpenAILLMService): logger.debug(f"Creating Fireworks client with api {base_url}") return super().create_client(api_key, base_url, **kwargs) - def build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for Fireworks chat completion request. Fireworks doesn't support some OpenAI parameters like seed, max_completion_tokens, and stream_options. + + Args: + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. + + Returns: + Dictionary of parameters for the chat completion request. """ params = { "model": self.model_name, "stream": True, - "messages": messages, - "tools": context.tools, - "tool_choice": context.tool_choice, "frequency_penalty": self._settings["frequency_penalty"], "presence_penalty": self._settings["presence_penalty"], "temperature": self._settings["temperature"], @@ -75,5 +77,17 @@ class FireworksLLMService(OpenAILLMService): "max_tokens": self._settings["max_tokens"], } + # Messages, tools, tool_choice + params.update(params_from_context) + params.update(self._settings["extra"]) return params + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as FireworksLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 88664c9d8..7a140c363 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -16,19 +16,20 @@ import json import os import uuid from dataclasses import dataclass -from typing import Any, Dict, List, Optional +from typing import Any, AsyncIterator, Dict, List, Optional from loguru import logger from PIL import Image from pydantic import BaseModel, Field -from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter +from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams from pipecat.frames.frames import ( AudioRawFrame, Frame, FunctionCallCancelFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, @@ -38,6 +39,7 @@ from pipecat.frames.frames import ( VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, @@ -67,6 +69,7 @@ try: FunctionCall, FunctionResponse, GenerateContentConfig, + GenerateContentResponse, HttpOptions, Part, ) @@ -418,7 +421,14 @@ class GoogleLLMContext(OpenAILLMContext): role = message["role"] content = message.get("content", []) if role == "system": - self.system_message = content + # System instructions are returned as plain text + if isinstance(content, str): + self.system_message = content + elif isinstance(content, list): + # If content is a list, we assume it's a list of text parts, per the standard + self.system_message = " ".join( + part["text"] for part in content if part.get("type") == "text" + ) return None elif role == "assistant": role = "model" @@ -436,11 +446,20 @@ class GoogleLLMContext(OpenAILLMContext): ) elif role == "tool": role = "model" + try: + response = json.loads(message["content"]) + if isinstance(response, dict): + response_dict = response + else: + response_dict = {"value": response} + except Exception as e: + # Response might not be JSON-deserializable (e.g. plain text). + response_dict = {"value": message["content"]} parts.append( Part( function_response=FunctionResponse( name="tool_call_result", # seems to work to hard-code the same name every time - response=json.loads(message["content"]), + response=response_dict, ) ) ) @@ -636,9 +655,8 @@ class GoogleLLMService(LLMService): """Google AI (Gemini) LLM service implementation. This class implements inference with Google's AI models, translating internally - from OpenAILLMContext to the messages format expected by the Google AI model. - We use OpenAILLMContext as a lingua franca for all LLM services to enable - easy switching between different LLMs. + from an OpenAILLMContext or a universal LLMContext to the messages format + expected by the Google AI model. """ # Overriding the default adapter to use the Gemini one. @@ -715,6 +733,50 @@ class GoogleLLMService(LLMService): def _create_client(self, api_key: str, http_options: Optional[HttpOptions] = None): self._client = genai.Client(api_key=api_key, http_options=http_options) + async def run_inference( + self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None + ) -> Optional[str]: + """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context. + + Args: + context: The LLM context containing conversation history. + system_instruction: Optional system instruction to guide the LLM's + behavior. You could also (again, optionally) provide a system + instruction directly in the context. If both are provided, the + one in the context takes precedence. + + Returns: + The LLM's response as a string, or None if no response is generated. + """ + messages = [] + system = [] + if isinstance(context, LLMContext): + adapter = self.get_llm_adapter() + params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context) + messages = params["messages"] + system = params["system_instruction"] + else: + context = GoogleLLMContext.upgrade_to_google(context) + messages = context.messages + system = getattr(context, "system_message", None) or system_instruction + + generation_config = GenerateContentConfig(system_instruction=system) + + # Use the new google-genai client's async method + response = await self._client.aio.models.generate_content( + model=self._model_name, + contents=messages, + config=generation_config, + ) + + # Extract text from response + if response.candidates and response.candidates[0].content: + for part in response.candidates[0].content.parts: + if part.text: + return part.text + + return None + def needs_mcp_alternate_schema(self) -> bool: """Check if this LLM service requires alternate MCP schema. @@ -740,8 +802,89 @@ class GoogleLLMService(LLMService): except Exception as e: logger.exception(f"Failed to unset thinking budget: {e}") + async def _stream_content( + self, params_from_context: GeminiLLMInvocationParams + ) -> AsyncIterator[GenerateContentResponse]: + messages = params_from_context["messages"] + if ( + params_from_context["system_instruction"] + and self._system_instruction != params_from_context["system_instruction"] + ): + logger.debug(f"System instruction changed: {params_from_context['system_instruction']}") + self._system_instruction = params_from_context["system_instruction"] + + tools = [] + if params_from_context["tools"]: + tools = params_from_context["tools"] + elif self._tools: + tools = self._tools + tool_config = None + if self._tool_config: + tool_config = self._tool_config + + # Filter out None values and create GenerationContentConfig + generation_params = { + k: v + for k, v in { + "system_instruction": self._system_instruction, + "temperature": self._settings["temperature"], + "top_p": self._settings["top_p"], + "top_k": self._settings["top_k"], + "max_output_tokens": self._settings["max_tokens"], + "tools": tools, + "tool_config": tool_config, + }.items() + if v is not None + } + + if self._settings["extra"]: + generation_params.update(self._settings["extra"]) + + # possibly modify generation_params (in place) to set thinking to off by default + self._maybe_unset_thinking_budget(generation_params) + + generation_config = ( + GenerateContentConfig(**generation_params) if generation_params else None + ) + + await self.start_ttfb_metrics() + return await self._client.aio.models.generate_content_stream( + model=self._model_name, + contents=messages, + config=generation_config, + ) + + async def _stream_content_specific_context( + self, context: OpenAILLMContext + ) -> AsyncIterator[GenerateContentResponse]: + logger.debug( + # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]" + f"{self}: Generating chat from OpenAI context [{context.get_messages_for_logging()}]" + ) + + params = GeminiLLMInvocationParams( + messages=context.messages, + system_instruction=context.system_message, + tools=context.tools, + ) + + return await self._stream_content(params) + + async def _stream_content_universal_context( + self, context: LLMContext + ) -> AsyncIterator[GenerateContentResponse]: + adapter = self.get_llm_adapter() + logger.debug( + # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]" + f"{self}: Generating chat from universal context [{adapter.get_messages_for_logging(context)}]" + ) + + params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context) + + return await self._stream_content(params) + @traced_llm - async def _process_context(self, context: OpenAILLMContext): + async def _process_context(self, context: OpenAILLMContext | LLMContext): await self.push_frame(LLMFullResponseStartFrame()) prompt_tokens = 0 @@ -754,55 +897,11 @@ class GoogleLLMService(LLMService): search_result = "" try: - logger.debug( - # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]" - f"{self}: Generating chat [{context.get_messages_for_logging()}]" - ) - - messages = context.messages - if context.system_message and self._system_instruction != context.system_message: - logger.debug(f"System instruction changed: {context.system_message}") - self._system_instruction = context.system_message - - tools = [] - if context.tools: - tools = context.tools - elif self._tools: - tools = self._tools - tool_config = None - if self._tool_config: - tool_config = self._tool_config - - # Filter out None values and create GenerationContentConfig - generation_params = { - k: v - for k, v in { - "system_instruction": self._system_instruction, - "temperature": self._settings["temperature"], - "top_p": self._settings["top_p"], - "top_k": self._settings["top_k"], - "max_output_tokens": self._settings["max_tokens"], - "tools": tools, - "tool_config": tool_config, - }.items() - if v is not None - } - - if self._settings["extra"]: - generation_params.update(self._settings["extra"]) - - # possibly modify generation_params (in place) to set thinking to off by default - self._maybe_unset_thinking_budget(generation_params) - - generation_config = ( - GenerateContentConfig(**generation_params) if generation_params else None - ) - - await self.start_ttfb_metrics() - response = await self._client.aio.models.generate_content_stream( - model=self._model_name, - contents=messages, - config=generation_config, + # Generate content using either OpenAILLMContext or universal LLMContext + response = await ( + self._stream_content_specific_context(context) + if isinstance(context, OpenAILLMContext) + else self._stream_content_universal_context(context) ) function_calls = [] @@ -915,9 +1014,18 @@ class GoogleLLMService(LLMService): if isinstance(frame, OpenAILLMContextFrame): context = GoogleLLMContext.upgrade_to_google(frame.context) + elif isinstance(frame, LLMContextFrame): + # Handle universal (LLM-agnostic) LLM context frames + context = frame.context elif isinstance(frame, LLMMessagesFrame): + # NOTE: LLMMessagesFrame is deprecated, so we don't support the newer universal + # LLMContext with it context = GoogleLLMContext(frame.messages) elif isinstance(frame, VisionImageRawFrame): + # This is only useful in very simple pipelines because it creates + # a new context. Generally we want a context manager to catch + # UserImageRawFrames coming through the pipeline and add them + # to the context. context = GoogleLLMContext() context.add_image_frame_message( format=frame.format, size=frame.size, image=frame.image, text=frame.text diff --git a/src/pipecat/services/google/llm_openai.py b/src/pipecat/services/google/llm_openai.py index bcd350380..2c64f050f 100644 --- a/src/pipecat/services/google/llm_openai.py +++ b/src/pipecat/services/google/llm_openai.py @@ -39,6 +39,10 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): Note: This service includes a workaround for a Google API bug where function call indices may be incorrectly set to None, resulting in empty function names. + .. deprecated:: 0.0.82 + GoogleLLMOpenAIBetaService is deprecated and will be removed in a future version. + Use GoogleLLMService instead for better integration with Google's native API. + Reference: https://ai.google.dev/gemini-api/docs/openai """ @@ -59,8 +63,26 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): model: Google model name to use (e.g., "gemini-2.0-flash"). **kwargs: Additional arguments passed to the parent OpenAILLMService. """ + import warnings + + warnings.warn( + "GoogleLLMOpenAIBetaService is deprecated and will be removed in a future version. " + "Use GoogleLLMService instead for better integration with Google's native API.", + DeprecationWarning, + stacklevel=2, + ) + super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs) + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as GoogleLLMOpenAIBetaService does not yet support universal LLMContext. + """ + return False + async def _process_context(self, context: OpenAILLMContext): functions_list = [] arguments_list = [] @@ -72,9 +94,9 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): await self.start_ttfb_metrics() - chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions( - context - ) + chunk_stream: AsyncStream[ + ChatCompletionChunk + ] = await self._stream_chat_completions_specific_context(context) async for chunk in chunk_stream: if chunk.usage: diff --git a/src/pipecat/services/google/llm_vertex.py b/src/pipecat/services/google/llm_vertex.py index 22b6258a5..bdbf2dda1 100644 --- a/src/pipecat/services/google/llm_vertex.py +++ b/src/pipecat/services/google/llm_vertex.py @@ -139,3 +139,12 @@ class GoogleVertexLLMService(OpenAILLMService): creds.refresh(Request()) # Ensure token is up-to-date, lifetime is 1 hour. return creds.token + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as GoogleVertexLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/grok/llm.py b/src/pipecat/services/grok/llm.py index 2a9704008..49fe2e802 100644 --- a/src/pipecat/services/grok/llm.py +++ b/src/pipecat/services/grok/llm.py @@ -190,3 +190,12 @@ class GrokLLMService(OpenAILLMService): user = OpenAIUserContextAggregator(context, params=user_params) assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) return GrokContextAggregatorPair(_user=user, _assistant=assistant) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as GrokLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/groq/llm.py b/src/pipecat/services/groq/llm.py index 57f2a533d..d3166ff8b 100644 --- a/src/pipecat/services/groq/llm.py +++ b/src/pipecat/services/groq/llm.py @@ -49,3 +49,12 @@ class GroqLLMService(OpenAILLMService): """ logger.debug(f"Creating Groq client with api {base_url}") return super().create_client(api_key, base_url, **kwargs) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as GroqLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 8f12a598b..3152a0083 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -14,6 +14,7 @@ from typing import ( Awaitable, Callable, Dict, + List, Mapping, Optional, Protocol, @@ -40,6 +41,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, UserImageRequestFrame, ) +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, @@ -88,7 +90,7 @@ class FunctionCallParams: tool_call_id: str arguments: Mapping[str, Any] llm: "LLMService" - context: OpenAILLMContext + context: OpenAILLMContext | LLMContext result_callback: FunctionCallResultCallback @@ -129,7 +131,7 @@ class FunctionCallRunnerItem: function_name: str tool_call_id: str arguments: Mapping[str, Any] - context: OpenAILLMContext + context: OpenAILLMContext | LLMContext run_llm: Optional[bool] = None @@ -189,6 +191,24 @@ class LLMService(AIService): """ return self._adapter + async def run_inference( + self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None + ) -> Optional[str]: + """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context. + + Must be implemented by subclasses. + + Args: + context: The LLM context containing conversation history. + system_instruction: Optional system instruction to guide the LLM's + behavior. You could also (again, optionally) provide a system + instruction directly in the context. + + Returns: + The LLM's response as a string, or None if no response is generated. + """ + raise NotImplementedError(f"run_inference() not supported by {self.__class__.__name__}") + def create_context_aggregator( self, context: OpenAILLMContext, @@ -432,7 +452,9 @@ class LLMService(AIService): else: await self._sequential_runner_queue.put(runner_item) - async def _call_start_function(self, context: OpenAILLMContext, function_name: str): + async def _call_start_function( + self, context: OpenAILLMContext | LLMContext, function_name: str + ): if function_name in self._start_callbacks.keys(): await self._start_callbacks[function_name](function_name, self, context) elif None in self._start_callbacks.keys(): diff --git a/src/pipecat/services/nim/llm.py b/src/pipecat/services/nim/llm.py index 052b94274..fdfb8bf6b 100644 --- a/src/pipecat/services/nim/llm.py +++ b/src/pipecat/services/nim/llm.py @@ -47,6 +47,15 @@ class NimLLMService(OpenAILLMService): self._has_reported_prompt_tokens = False self._is_processing = False + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as NimLLMService does not yet support universal LLMContext. + """ + return False + async def _process_context(self, context: OpenAILLMContext): """Process a context through the LLM and accumulate token usage metrics. diff --git a/src/pipecat/services/ollama/llm.py b/src/pipecat/services/ollama/llm.py index 2284a5070..aa6f58b59 100644 --- a/src/pipecat/services/ollama/llm.py +++ b/src/pipecat/services/ollama/llm.py @@ -43,3 +43,12 @@ class OLLamaLLMService(OpenAILLMService): """ logger.debug(f"Creating Ollama client with api {base_url}") return super().create_client(base_url=base_url, **kwargs) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as OLLamaLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 1dec6e91b..e51755cba 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Base OpenAI LLM service implementation.""" +"""Base LLM service implementation for services that use the AsyncOpenAI client.""" import asyncio import base64 @@ -23,8 +23,10 @@ from openai import ( from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam from pydantic import BaseModel, Field +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.frames.frames import ( Frame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, @@ -33,6 +35,7 @@ from pipecat.frames.frames import ( VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -45,10 +48,11 @@ from pipecat.utils.tracing.service_decorators import traced_llm class BaseOpenAILLMService(LLMService): """Base class for all services that use the AsyncOpenAI client. - This service consumes OpenAILLMContextFrame frames, which contain a reference - to an OpenAILLMContext object. The context defines what is sent to the LLM for - completion, including user, assistant, and system messages, as well as tool - choices and function call configurations. + This service consumes OpenAILLMContextFrame or LLMContextFrame frames, + which contain a reference to an OpenAILLMContext or LLMContext object. The + context defines what is sent to the LLM for completion, including user, + assistant, and system messages, as well as tool choices and function call + configurations. """ class InputParams(BaseModel): @@ -180,18 +184,19 @@ class BaseOpenAILLMService(LLMService): return True async def get_chat_completions( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] + self, params_from_context: OpenAILLMInvocationParams ) -> AsyncStream[ChatCompletionChunk]: """Get streaming chat completions from OpenAI API with optional timeout and retry. Args: - context: The LLM context containing tools and configuration. - messages: List of chat completion messages to send. + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. Returns: Async stream of chat completion chunks. """ - params = self.build_chat_completion_params(context, messages) + params = self.build_chat_completion_params(params_from_context) if self._retry_on_timeout: try: @@ -208,16 +213,15 @@ class BaseOpenAILLMService(LLMService): chunks = await self._client.chat.completions.create(**params) return chunks - def build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for chat completion request. Subclasses can override this to customize parameters for different providers. Args: - context: The LLM context containing tools and configuration. - messages: List of chat completion messages to send. + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. Returns: Dictionary of parameters for the chat completion request. @@ -225,9 +229,6 @@ class BaseOpenAILLMService(LLMService): params = { "model": self.model_name, "stream": True, - "messages": messages, - "tools": context.tools, - "tool_choice": context.tool_choice, "stream_options": {"include_usage": True}, "frequency_penalty": self._settings["frequency_penalty"], "presence_penalty": self._settings["presence_penalty"], @@ -238,13 +239,48 @@ class BaseOpenAILLMService(LLMService): "max_completion_tokens": self._settings["max_completion_tokens"], } + # Messages, tools, tool_choice + params.update(params_from_context) + params.update(self._settings["extra"]) return params - async def _stream_chat_completions( + async def run_inference( + self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None + ) -> Optional[str]: + """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context. + + Args: + context: The LLM context containing conversation history. + system_instruction: Optional system instruction to guide the LLM's + behavior. You could also (again, optionally) provide a system + instruction directly in the context. + + Returns: + The LLM's response as a string, or None if no response is generated. + """ + if isinstance(context, LLMContext): + adapter = self.get_llm_adapter() + params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context) + messages = params["messages"] + else: + messages = context.messages + + # LLM completion + response = await self._client.chat.completions.create( + model=self.model_name, + messages=messages, + stream=False, + ) + + return response.choices[0].message.content + + async def _stream_chat_completions_specific_context( self, context: OpenAILLMContext ) -> AsyncStream[ChatCompletionChunk]: - logger.debug(f"{self}: Generating chat [{context.get_messages_for_logging()}]") + logger.debug( + f"{self}: Generating chat from OpenAI context [{context.get_messages_for_logging()}]" + ) messages: List[ChatCompletionMessageParam] = context.get_messages() @@ -263,12 +299,28 @@ class BaseOpenAILLMService(LLMService): del message["data"] del message["mime_type"] - chunks = await self.get_chat_completions(context, messages) + params = OpenAILLMInvocationParams( + messages=messages, tools=context.tools, tool_choice=context.tool_choice + ) + chunks = await self.get_chat_completions(params) + + return chunks + + async def _stream_chat_completions_universal_context( + self, context: LLMContext + ) -> AsyncStream[ChatCompletionChunk]: + adapter = self.get_llm_adapter() + logger.debug( + f"{self}: Generating chat from universal context [{adapter.get_messages_for_logging(context)}]" + ) + + params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context) + chunks = await self.get_chat_completions(params) return chunks @traced_llm - async def _process_context(self, context: OpenAILLMContext): + async def _process_context(self, context: OpenAILLMContext | LLMContext): functions_list = [] arguments_list = [] tool_id_list = [] @@ -279,8 +331,11 @@ class BaseOpenAILLMService(LLMService): await self.start_ttfb_metrics() - chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions( - context + # Generate chat completions using either OpenAILLMContext or universal LLMContext + chunk_stream = await ( + self._stream_chat_completions_specific_context(context) + if isinstance(context, OpenAILLMContext) + else self._stream_chat_completions_universal_context(context) ) async for chunk in chunk_stream: @@ -364,11 +419,24 @@ class BaseOpenAILLMService(LLMService): await self.run_function_calls(function_calls) + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + Whether service supports universal LLMContext. + """ + # Return True in subclasses that support universal LLMContext + # This property lets us gradually roll out support for universal + # LLMContext to OpenAI-like services in a controlled manner. + return False + async def process_frame(self, frame: Frame, direction: FrameDirection): """Process frames for LLM completion requests. - Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame, - and LLMUpdateSettingsFrame to trigger LLM completions and manage settings. + Handles OpenAILLMContextFrame, LLMContextFrame, LLMMessagesFrame, + VisionImageRawFrame, and LLMUpdateSettingsFrame to trigger LLM + completions and manage settings. Args: frame: The frame to process. @@ -378,10 +446,26 @@ class BaseOpenAILLMService(LLMService): context = None if isinstance(frame, OpenAILLMContextFrame): - context: OpenAILLMContext = frame.context + # Handle OpenAI-specific context frames + context = frame.context + elif isinstance(frame, LLMContextFrame): + # Handle universal (LLM-agnostic) LLM context frames + if self.supports_universal_context: + context = frame.context + else: + raise NotImplementedError( + f"Universal LLMContext is not yet supported for {self.__class__.__name__}." + ) elif isinstance(frame, LLMMessagesFrame): + # NOTE: LLMMessagesFrame is deprecated, so we don't support the newer universal + # LLMContext with it context = OpenAILLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): + # This is only useful in very simple pipelines because it creates + # a new context. Generally we want a context manager to catch + # UserImageRawFrames coming through the pipeline and add them + # to the context. + # TODO: support the newer universal LLMContext with a VisionImageRawFrame equivalent? context = OpenAILLMContext() context.add_image_frame_message( format=frame.format, size=frame.size, image=frame.image, text=frame.text diff --git a/src/pipecat/services/openai/llm.py b/src/pipecat/services/openai/llm.py index 7919dd159..9f5d896b6 100644 --- a/src/pipecat/services/openai/llm.py +++ b/src/pipecat/services/openai/llm.py @@ -107,6 +107,15 @@ class OpenAILLMService(BaseOpenAILLMService): assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + True, as OpenAI service supports universal LLMContext. + """ + return True + class OpenAIUserContextAggregator(LLMUserContextAggregator): """OpenAI-specific user context aggregator. diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index bc7af9a46..dd22694f7 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -23,6 +23,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InterimTranscriptionFrame, + LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, @@ -343,6 +344,10 @@ class OpenAIRealtimeBetaLLMService(LLMService): await self.reset_conversation() # Run the LLM at next opportunity await self._create_response() + elif isinstance(frame, LLMContextFrame): + raise NotImplementedError( + "Universal LLMContext is not yet supported for OpenAI Realtime." + ) elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) diff --git a/src/pipecat/services/openpipe/llm.py b/src/pipecat/services/openpipe/llm.py index 581bb045f..2e491ea26 100644 --- a/src/pipecat/services/openpipe/llm.py +++ b/src/pipecat/services/openpipe/llm.py @@ -13,9 +13,8 @@ enabling integration with OpenPipe's fine-tuning and monitoring capabilities. from typing import Dict, List, Optional from loguru import logger -from openai.types.chat import ChatCompletionMessageParam -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.services.openai.llm import OpenAILLMService try: @@ -86,22 +85,21 @@ class OpenPipeLLMService(OpenAILLMService): ) return client - def build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for OpenPipe chat completion request. Adds OpenPipe-specific logging and tagging parameters. Args: - context: The LLM context containing tools and configuration. - messages: List of chat completion messages to send. + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. Returns: Dictionary of parameters for the chat completion request. """ # Start with base parameters - params = super().build_chat_completion_params(context, messages) + params = super().build_chat_completion_params(params_from_context) # Add OpenPipe-specific parameters params["openpipe"] = { @@ -110,3 +108,12 @@ class OpenPipeLLMService(OpenAILLMService): } return params + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as OpenPipeLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/openrouter/llm.py b/src/pipecat/services/openrouter/llm.py index 97a9d336a..3ba1ae6f6 100644 --- a/src/pipecat/services/openrouter/llm.py +++ b/src/pipecat/services/openrouter/llm.py @@ -61,3 +61,12 @@ class OpenRouterLLMService(OpenAILLMService): """ logger.debug(f"Creating OpenRouter client with api {base_url}") return super().create_client(api_key, base_url, **kwargs) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as OpenRouterLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/perplexity/llm.py b/src/pipecat/services/perplexity/llm.py index 59cbe520b..3e39206c6 100644 --- a/src/pipecat/services/perplexity/llm.py +++ b/src/pipecat/services/perplexity/llm.py @@ -11,11 +11,9 @@ an OpenAI-compatible interface. It handles Perplexity's unique token usage reporting patterns while maintaining compatibility with the Pipecat framework. """ -from typing import List - from openai import NOT_GIVEN -from openai.types.chat import ChatCompletionMessageParam +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.openai.llm import OpenAILLMService @@ -53,17 +51,23 @@ class PerplexityLLMService(OpenAILLMService): self._has_reported_prompt_tokens = False self._is_processing = False - def build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for Perplexity chat completion request. Perplexity uses a subset of OpenAI parameters and doesn't support tools. + + Args: + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. + + Returns: + Dictionary of parameters for the chat completion request. """ params = { "model": self.model_name, "stream": True, - "messages": messages, + "messages": params_from_context["messages"], } # Add OpenAI-compatible parameters if they're set @@ -80,6 +84,15 @@ class PerplexityLLMService(OpenAILLMService): return params + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as PerplexityLLMService does not yet support universal LLMContext. + """ + return False + async def _process_context(self, context: OpenAILLMContext): """Process a context through the LLM and accumulate token usage metrics. diff --git a/src/pipecat/services/qwen/llm.py b/src/pipecat/services/qwen/llm.py index 648cbd9e8..1c842ded6 100644 --- a/src/pipecat/services/qwen/llm.py +++ b/src/pipecat/services/qwen/llm.py @@ -50,3 +50,12 @@ class QwenLLMService(OpenAILLMService): """ logger.debug(f"Creating Qwen client with base URL: {base_url}") return super().create_client(api_key, base_url, **kwargs) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as QwenLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/sambanova/llm.py b/src/pipecat/services/sambanova/llm.py index f2ee082ed..d39eb51a2 100644 --- a/src/pipecat/services/sambanova/llm.py +++ b/src/pipecat/services/sambanova/llm.py @@ -7,12 +7,13 @@ """SambaNova LLM service implementation using OpenAI-compatible interface.""" import json -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Optional from loguru import logger from openai import AsyncStream -from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam +from openai.types.chat import ChatCompletionChunk +from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams from pipecat.frames.frames import ( LLMTextFrame, ) @@ -67,17 +68,16 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore logger.debug(f"Creating SambaNova client with API {base_url}") return super().create_client(api_key, base_url, **kwargs) - def build_chat_completion_params( - self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> dict: + def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict: """Build parameters for SambaNova chat completion request. SambaNova doesn't support some OpenAI parameters like frequency_penalty, presence_penalty, and seed. Args: - context: The LLM context containing tools and configuration. - messages: List of chat completion messages to send. + params_from_context: Parameters, derived from the LLM context, to + use for the chat completion. Contains messages, tools, and tool + choice. Returns: Dictionary of parameters for the chat completion request. @@ -85,9 +85,6 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore params = { "model": self.model_name, "stream": True, - "messages": messages, - "tools": context.tools, - "tool_choice": context.tool_choice, "stream_options": {"include_usage": True}, "temperature": self._settings["temperature"], "top_p": self._settings["top_p"], @@ -95,6 +92,9 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore "max_completion_tokens": self._settings["max_completion_tokens"], } + # Messages, tools, tool_choice + params.update(params_from_context) + params.update(self._settings["extra"]) return params @@ -122,9 +122,9 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore await self.start_ttfb_metrics() - chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions( - context - ) + chunk_stream: AsyncStream[ + ChatCompletionChunk + ] = await self._stream_chat_completions_specific_context(context) async for chunk in chunk_stream: if chunk.usage: @@ -210,3 +210,12 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore ) await self.run_function_calls(function_calls) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as SambaNovaLLMService does not yet support universal LLMContext. + """ + return False diff --git a/src/pipecat/services/together/llm.py b/src/pipecat/services/together/llm.py index 7a22c885a..2a004f1c9 100644 --- a/src/pipecat/services/together/llm.py +++ b/src/pipecat/services/together/llm.py @@ -49,3 +49,12 @@ class TogetherLLMService(OpenAILLMService): """ logger.debug(f"Creating Together.ai client with api {base_url}") return super().create_client(api_key, base_url, **kwargs) + + @property + def supports_universal_context(self) -> bool: + """Check if this service supports universal LLMContext. + + Returns: + False, as TogetherLLMService does not yet support universal LLMContext. + """ + return False