diff --git a/examples/foundational/14w-function-calling-universal-context.py b/examples/foundational/14w-function-calling-universal-context.py new file mode 100644 index 000000000..7087a70d6 --- /dev/null +++ b/examples/foundational/14w-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.create(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/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py index 6a957c267..c60182d0d 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 + +# Should be a TypedDict +TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any]) -class BaseLLMAdapter(ABC): +# TODO: fix everywhere we subclass BaseLLMAdapter... +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,6 +62,20 @@ class BaseLLMAdapter(ABC): """ pass + @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 + + # TODO: should this also be able to return NotGiven? def from_standard_tools(self, tools: Any) -> List[Any]: """Convert tools from standard format to provider format. @@ -54,4 +92,38 @@ class BaseLLMAdapter(ABC): # Fallback to return the same tools in case they are not in a standard format return tools + 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 + # TODO: we can move the logic to also handle the Messages here diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py index 59d70aa1e..9a0494e55 100644 --- a/src/pipecat/adapters/services/open_ai_adapter.py +++ b/src/pipecat/adapters/services/open_ai_adapter.py @@ -6,22 +6,62 @@ """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 OpenAILLMInvocationParams(TypedDict): + """Context-based parameters for invoking OpenAI ChatCompletion API.""" + + messages: List[ChatCompletionMessageParam] + tools: List[ChatCompletionToolParam] | OpenAINotGiven + tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven class OpenAILLMAdapter(BaseLLMAdapter): - """Adapter for converting tool schemas to OpenAI's format. + """OpenAI-specific adapter for Pipecat. - Provides conversion utilities for transforming Pipecat's standard tool - schemas into the format expected by OpenAI's ChatCompletion API for - function calling capabilities. + 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_standard_messages(context.messages), + # TODO: doesn't seem quite right that we may or may not need to convert tools here; they should already be guaranteed to exist in a universal format in the universal LLMContext, right? + "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 +77,40 @@ 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 context.messages: + 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 _from_standard_messages( + self, messages: List[LLMContextMessage] + ) -> List[ChatCompletionMessageParam]: + # Just a pass-through: messages is 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/services/google/llm.py b/src/pipecat/services/google/llm.py index 88664c9d8..fd4bdca9a 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -918,6 +918,10 @@ class GoogleLLMService(LLMService): elif isinstance(frame, LLMMessagesFrame): 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/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 1dec6e91b..2e04445d4 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,13 +184,13 @@ 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. @@ -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,18 @@ 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( 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 +269,29 @@ 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,9 +302,16 @@ class BaseOpenAILLMService(LLMService): await self.start_ttfb_metrics() - chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions( - context - ) + if isinstance(context, OpenAILLMContext): + # Use OpenAI-specific context + chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions( + context + ) + else: + # Use universal (LLM-agnostic) context + chunk_stream: AsyncStream[ + ChatCompletionChunk + ] = await self._stream_chat_completions_universal_context(context) async for chunk in chunk_stream: if chunk.usage: @@ -367,8 +397,9 @@ class BaseOpenAILLMService(LLMService): 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 +409,21 @@ 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 + context = frame.context 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