Source code for pipecat.processors.aggregators.llm_context

#
# Copyright (c) 2024-2026, 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 asyncio
import base64
import copy
import io
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal, TypeAlias, cast, overload

from loguru import logger
from PIL import Image

from pipecat.adapters.schemas.direct_function import DirectFunction
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.utils import pcm_to_wav
from pipecat.frames.frames import AudioRawFrame

# The sentinel is part of LLMContext's public surface — tools and tool_choice
# default to it — so it is re-exported here for callers. The redundant aliases
# mark that intent; a plain import would look unused and be stripped.
from pipecat.utils.types import NOT_GIVEN as NOT_GIVEN
from pipecat.utils.types import NotGiven as NotGiven
from pipecat.utils.types import is_given as is_given

# "Re-export" types from OpenAI that we're using as universal context types.
# NOTE: these are aliased to OpenAI's for type checking, but callers should
# treat them as LLMContext's own types — independent definitions that happen to
# coincide with OpenAI's as an implementation detail. If universal context types
# need to someday diverge from OpenAI's, we should consider managing our own
# definitions (but with care, since OpenAI's types are somewhat of a standard
# and we want to continue supporting them). In the meantime, code at the
# LLMContext/OpenAI boundary should use explicit casts rather than rely on
# the aliasing.
#
# The aliases resolve under type checking only. Every LLM service reaches this
# module, so importing the OpenAI SDK here would put its load on the startup
# path of pipelines that never talk to OpenAI. At runtime both are structurally
# dicts or strings, which is all the annotations need them to be.
if TYPE_CHECKING:
    from openai.types.chat import (
        ChatCompletionMessageParam,
        ChatCompletionToolChoiceOptionParam,
    )

    LLMStandardMessage: TypeAlias = ChatCompletionMessageParam
    LLMContextToolChoice: TypeAlias = ChatCompletionToolChoiceOptionParam
else:
    LLMStandardMessage: TypeAlias = Any
    LLMContextToolChoice: TypeAlias = Any


[docs] @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 = LLMStandardMessage | LLMSpecificMessage
[docs] 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. """
[docs] def __init__( self, messages: list[LLMContextMessage] | None = None, tools: ToolsSchema | list[FunctionSchema | DirectFunction] | 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. May be a ``ToolsSchema`` or a plain list of direct functions and/or ``FunctionSchema`` objects (normalized to a ``ToolsSchema`` internally). Any tool that carries a handler — a direct function, or a ``FunctionSchema`` with its ``handler`` set — is registered with the LLM service automatically, so no separate ``register_function`` call is needed. 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
[docs] @staticmethod def create_image_url_message( *, role: str = "user", url: str, text: str | None = None, ) -> LLMContextMessage: """Create a context message containing an image URL. Args: role: The role of this message (defaults to "user"). url: The URL of the image. text: Optional text to include with the image. """ content: list[dict[str, Any]] = [] if text: content.append({"type": "text", "text": text}) content.append({"type": "image_url", "image_url": {"url": url}}) return cast(LLMContextMessage, {"role": role, "content": content})
[docs] @staticmethod async def create_image_message( *, role: str = "user", format: str, size: tuple[int, int], image: bytes, text: str | None = None, ) -> LLMContextMessage: """Create a context message containing an image. Args: role: The role of this message (defaults to "user"). format: Image format (e.g., 'RGB', 'RGBA', or, if already encoded, the MIME type like 'image/jpeg'). size: Image dimensions as (width, height) tuple. image: Raw image bytes. text: Optional text to include with the image. """ # Format is a mime type: image is already encoded image_already_encoded = format.startswith("image/") def encode_image(): if image_already_encoded: bytes = image else: # Encode to JPEG buffer = io.BytesIO() Image.frombytes(format, size, image).save(buffer, format="JPEG") bytes = buffer.getvalue() encoded_image = base64.b64encode(bytes).decode("utf-8") return encoded_image encoded_image = await asyncio.to_thread(encode_image) url = f"data:{format if image_already_encoded else 'image/jpeg'};base64,{encoded_image}" return LLMContext.create_image_url_message(role=role, url=url, text=text)
[docs] @staticmethod async def create_audio_message( *, role: str = "user", audio_frames: list[AudioRawFrame], text: str = "Audio follows" ) -> LLMContextMessage: """Create a context message containing audio. Args: role: The role of this message (defaults to "user"). audio_frames: List of audio frame objects to include. text: Optional text to include with the audio. """ content: list[dict[str, Any]] = [{"type": "text", "text": text}] def encode_audio(): sample_rate = audio_frames[0].sample_rate num_channels = audio_frames[0].num_channels data = b"".join(frame.audio for frame in audio_frames) wav = pcm_to_wav(data, sample_rate, num_channels) return base64.b64encode(wav).decode("utf-8") encoded_audio = await asyncio.to_thread(encode_audio) content.append( { "type": "input_audio", "input_audio": {"data": encoded_audio, "format": "wav"}, } ) return cast(LLMContextMessage, {"role": role, "content": content})
@property def messages(self) -> list[LLMContextMessage]: """Get the current messages list. NOTE: This is equivalent to calling `get_messages()` with no filter. If you want to filter out LLM-specific messages that don't pertain to your LLM, use `get_messages()` directly. Returns: List of conversation messages. """ return self.get_messages()
[docs] def get_messages( self, llm_specific_filter: str | None = None, *, truncate_large_values: bool = False, ) -> 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; this is intended to catch accidental use of incompatible LLM-specific messages. truncate_large_values: If True, return deep copies of messages with large values shortened. For standard messages, known binary data (base64-encoded images, audio) is replaced with short placeholders. For LLM-specific messages, long string values are truncated. Returns: List of conversation messages. """ if llm_specific_filter is None: messages = self._messages else: messages = [ msg for msg in self._messages if not isinstance(msg, LLMSpecificMessage) or msg.llm == llm_specific_filter ] if len(messages) < len(self._messages): logger.error( f"Attempted to use incompatible LLMSpecificMessages with LLM '{llm_specific_filter}'." ) if truncate_large_values: messages = LLMContext._truncate_large_values_from_messages(messages) return messages
@staticmethod def _truncate_large_values_from_messages( messages: list[LLMContextMessage], ) -> list[LLMContextMessage]: """Return deep copies of messages with large values replaced by placeholders. For standard (universal-format) messages, the following known binary patterns are replaced with short placeholders: - ``image_url`` items with ``data:image/...`` base64 URLs - ``input_audio`` items with ``input_audio.data`` or ``audio`` fields - ``audio`` items with an ``audio`` field - Top-level messages with a ``mime_type`` starting with ``image/`` For ``LLMSpecificMessage`` instances, long string values are truncated since the internal structure is provider-specific. """ result = [] for message in messages: if isinstance(message, LLMSpecificMessage): msg_copy = copy.deepcopy(message) msg_copy.message = LLMContext._truncate_long_strings(msg_copy.message) result.append(msg_copy) continue # The standard message variant is a union of TypedDicts; the # mutations below operate on plain dicts at runtime. Treat as # such for the duration of the redaction loop. msg: dict[str, Any] = cast(dict[str, Any], copy.deepcopy(message)) content = msg.get("content") if isinstance(content, list): for item in content: item_type = item.get("type") if item_type == "image_url": url = item.get("image_url", {}).get("url", "") if url.startswith("data:image/"): item["image_url"]["url"] = "data:image/..." elif item_type == "input_audio": if "input_audio" in item: item["input_audio"]["data"] = "..." if "audio" in item: item["audio"] = "..." elif item_type == "audio": if "audio" in item: item["audio"] = "..." if msg.get("mime_type", "").startswith("image/"): msg["data"] = "..." result.append(msg) return result @staticmethod def _truncate_long_strings(value: Any, *, max_length: int = 100) -> Any: """Recursively truncate long strings in a nested structure. Preserves the structure of dicts and lists while truncating any string values that exceed ``max_length``. Args: value: The value to process (dict, list, str, or other). max_length: Strings longer than this are truncated. Returns: A copy of the structure with long strings truncated. """ if isinstance(value, str): if len(value) > max_length: return f"{value[:max_length]}...({len(value)} chars)" return value elif isinstance(value, dict): return { k: LLMContext._truncate_long_strings(v, max_length=max_length) for k, v in value.items() } elif isinstance(value, list): return [ LLMContext._truncate_long_strings(item, max_length=max_length) for item in value ] return value @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
[docs] 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)
[docs] 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)
[docs] 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
[docs] def transform_messages( self, transform: Callable[[list[LLMContextMessage]], list[LLMContextMessage]] ): """Transform the current messages using the provided function. Args: transform: A function that takes the current list of messages and returns a modified list of messages to set in the context. """ self.set_messages(transform(self._messages))
[docs] def set_tools( self, tools: ToolsSchema | list[FunctionSchema | DirectFunction] | NotGiven = NOT_GIVEN, ): """Set the available tools for the LLM. Args: tools: A ToolsSchema, a plain list of direct functions and/or ``FunctionSchema`` objects, or NOT_GIVEN to disable tools. """ self._tools = LLMContext._normalize_and_validate_tools(tools)
[docs] 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
[docs] async def add_image_frame_message( self, *, format: str, size: tuple[int, int], image: bytes, text: str | None = None, role: str = "user", ): """Add a message containing an image frame. Args: format: Image format (e.g., 'RGB', 'RGBA', or, if already encoded, the MIME type like 'image/jpeg'). size: Image dimensions as (width, height) tuple. image: Raw image bytes. text: Optional text to include with the image. role: The role of this message (defaults to "user"). """ message = await LLMContext.create_image_message( role=role, format=format, size=size, image=image, text=text ) self.add_message(message)
[docs] async 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. """ message = await LLMContext.create_audio_message(audio_frames=audio_frames, text=text) self.add_message(message)
@overload @staticmethod def _normalize_and_validate_tools( tools: ToolsSchema | list[FunctionSchema | DirectFunction] | NotGiven, *, allow_provider_tools: Literal[False] = False, ) -> ToolsSchema | NotGiven: ... @overload @staticmethod def _normalize_and_validate_tools( tools: ToolsSchema | list[Any] | NotGiven, *, allow_provider_tools: Literal[True], ) -> ToolsSchema | list[Any] | NotGiven: ... @staticmethod def _normalize_and_validate_tools( tools: ToolsSchema | list[Any] | NotGiven, *, allow_provider_tools: bool = False, ) -> ToolsSchema | list[Any] | NotGiven: """Normalize and validate the given tools. A plain list of direct functions and/or ``FunctionSchema`` objects is wrapped in a ``ToolsSchema``. Args: tools: The tools to normalize: a ``ToolsSchema``, a list of direct functions and/or ``FunctionSchema`` objects, or ``NOT_GIVEN``. allow_provider_tools: If True, a list that isn't entirely standard tools (direct functions / ``FunctionSchema`` objects) is taken to be already-formatted, provider-native tools and returned unchanged rather than raising. For callers whose tools parameter accepts provider-native tools alongside standard ones. Raises: TypeError: If tools aren't a ``ToolsSchema``, list, or ``NOT_GIVEN`` — or, unless ``allow_provider_tools`` is set, if a list contains anything other than standard tools. """ if isinstance(tools, list): if allow_provider_tools and not all( isinstance(t, FunctionSchema) or callable(t) for t in tools ): # Already-formatted, provider-native tools; pass through unchanged. return tools tools = ToolsSchema(standard_tools=tools) 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( "In LLMContext, tools must be a ToolsSchema, a list of direct functions / " f"FunctionSchema objects, or NOT_GIVEN. Got type: {type(tools)}", )