#
# 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)}",
)