[WIP] Universal (LLM-agnostic) context machinery to support runtime LLM switching.

- Add to OpenAI LLM service support for universal LLM context

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()

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
+    Provides a standard interface for converting to provider-specific formats.
-    tool schemas and provider-specific tool formats. Subclasses must implement
+
-    provider-specific conversion logic.
+    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

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
+    Handles:
-    schemas into the format expected by OpenAI's ChatCompletion API for
+    - Extracting parameters for OpenAI's ChatCompletion API from a universal
-    function calling capabilities.
+      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

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

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
+    This service consumes OpenAILLMContextFrame or LLMContextFrame frames,
-    to an OpenAILLMContext object. The context defines what is sent to the LLM for
+    which contain a reference to an OpenAILLMContext or LLMContext object. The
-    completion, including user, assistant, and system messages, as well as tool
+    context defines what is sent to the LLM for completion, including user,
-    choices and function call configurations.
+    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.
+            params_from_context: Parameters, derived from the LLM context, to
-            messages: List of chat completion messages to send.
+                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(
+        if isinstance(context, OpenAILLMContext):
-            context
+            # 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,
+        Handles OpenAILLMContextFrame, LLMContextFrame, LLMMessagesFrame,
-        and LLMUpdateSettingsFrame to trigger LLM completions and manage settings.
+        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