diff --git a/changelog/4272.added.md b/changelog/4272.added.md new file mode 100644 index 000000000..69bfcf8bb --- /dev/null +++ b/changelog/4272.added.md @@ -0,0 +1 @@ +- Added `truncate_large_values` parameter to `LLMContext.get_messages()`. When `True`, returns compact deep copies of messages with binary data (base64 images, audio) replaced by short placeholders and long string values in LLM-specific messages recursively truncated. Useful for serialization, logging, and debugging tools. diff --git a/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py index b1080d197..9c6747766 100644 --- a/src/pipecat/adapters/base_llm_adapter.py +++ b/src/pipecat/adapters/base_llm_adapter.py @@ -125,16 +125,22 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]): """ return LLMSpecificMessage(llm=self.id_for_llm_specific_messages, message=message) - def get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + def get_messages( + self, context: LLMContext, *, truncate_large_values: bool = False + ) -> List[LLMContextMessage]: """Get messages from the LLM context, including standard and LLM-specific messages. Args: context: The LLM context containing messages. + truncate_large_values: If True, return deep copies of messages with + large values replaced by short placeholders. Returns: List of messages including standard and LLM-specific messages. """ - return context.get_messages(self.id_for_llm_specific_messages) + return context.get_messages( + self.id_for_llm_specific_messages, truncate_large_values=truncate_large_values + ) def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven: """Convert tools from standard format to provider format. diff --git a/src/pipecat/adapters/services/grok_realtime_adapter.py b/src/pipecat/adapters/services/grok_realtime_adapter.py index 2017013be..cc98887f8 100644 --- a/src/pipecat/adapters/services/grok_realtime_adapter.py +++ b/src/pipecat/adapters/services/grok_realtime_adapter.py @@ -77,7 +77,7 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter): def get_messages_for_logging(self, context) -> List[Dict[str, Any]]: """Get messages from context in a format safe for logging. - Removes or truncates sensitive data like audio content. + Binary data (images, audio) is replaced with short placeholders. Args: context: The LLM context containing messages. @@ -85,18 +85,7 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages with sensitive data redacted. """ - 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.get("type") == "input_audio": - item["audio"] = "..." - if item.get("type") == "audio": - item["audio"] = "..." - msgs.append(msg) - return msgs + return self.get_messages(context, truncate_large_values=True) @dataclass class ConvertedMessages: diff --git a/src/pipecat/adapters/services/inworld_realtime_adapter.py b/src/pipecat/adapters/services/inworld_realtime_adapter.py index 3504c86c5..b022afe6b 100644 --- a/src/pipecat/adapters/services/inworld_realtime_adapter.py +++ b/src/pipecat/adapters/services/inworld_realtime_adapter.py @@ -77,7 +77,7 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter): def get_messages_for_logging(self, context) -> List[Dict[str, Any]]: """Get messages from context in a format safe for logging. - Removes or truncates sensitive data like audio content. + Binary data (images, audio) is replaced with short placeholders. Args: context: The LLM context containing messages. @@ -85,18 +85,7 @@ class InworldRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages with sensitive data redacted. """ - 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.get("type") == "input_audio": - item["audio"] = "..." - if item.get("type") == "audio": - item["audio"] = "..." - msgs.append(msg) - return msgs + return self.get_messages(context, truncate_large_values=True) @dataclass class ConvertedMessages: diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py index db9f3ee2f..a52fb84a6 100644 --- a/src/pipecat/adapters/services/open_ai_adapter.py +++ b/src/pipecat/adapters/services/open_ai_adapter.py @@ -6,7 +6,6 @@ """OpenAI LLM adapter for Pipecat.""" -import copy from typing import Any, Dict, List, Optional, TypedDict from openai._types import NotGiven as OpenAINotGiven @@ -119,7 +118,7 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]): 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. + Binary data (images, audio) is replaced with short placeholders. Args: context: The LLM context containing messages. @@ -127,21 +126,7 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]): 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 item["type"] == "input_audio": - item["input_audio"]["data"] = "..." - if "mime_type" in msg and msg["mime_type"].startswith("image/"): - msg["data"] = "..." - msgs.append(msg) - return msgs + return self.get_messages(context, truncate_large_values=True) def _from_universal_context_messages( self, diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 0481d1bd4..41f3ce89d 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -71,7 +71,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): 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. + Binary data (images, audio) is replaced with short placeholders. This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime. @@ -81,25 +81,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages in a format ready for logging about OpenAI Realtime. """ - # NOTE: this is the same as in OpenAIAdapter, as that's what it was - # prior to a refactor. Worth noting that for OpenAI Realtime - # specifically, not everything handled here is necessarily supported - # (or supported yet). - 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 item["type"] == "input_audio": - item["input_audio"]["data"] = "..." - if "mime_type" in msg and msg["mime_type"].startswith("image/"): - msg["data"] = "..." - msgs.append(msg) - return msgs + return self.get_messages(context, truncate_large_values=True) @dataclass class ConvertedMessages: diff --git a/src/pipecat/adapters/services/open_ai_responses_adapter.py b/src/pipecat/adapters/services/open_ai_responses_adapter.py index a58b26c91..f3dd67e03 100644 --- a/src/pipecat/adapters/services/open_ai_responses_adapter.py +++ b/src/pipecat/adapters/services/open_ai_responses_adapter.py @@ -6,7 +6,6 @@ """OpenAI Responses API adapter for Pipecat.""" -import copy from typing import Any, Dict, List, Optional, TypedDict from openai._types import NotGiven as OpenAINotGiven @@ -136,7 +135,7 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam def get_messages_for_logging(self, context: LLMContext) -> List[Dict[str, Any]]: """Get messages from context in a format ready for logging. - Removes or truncates sensitive data like image content for safe logging. + Binary data (images, audio) is replaced with short placeholders. Args: context: The LLM context containing messages. @@ -144,19 +143,7 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam Returns: List of messages in a format ready for logging. """ - 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.get("type") == "image_url": - if item["image_url"]["url"].startswith("data:image/"): - item["image_url"]["url"] = "data:image/..." - if item.get("type") == "input_audio": - item["input_audio"]["data"] = "..." - msgs.append(msg) - return msgs + return self.get_messages(context, truncate_large_values=True) def _convert_messages_to_input( self, messages: List[LLMContextMessage] diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index a5d96c1d0..7bcb68c86 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -16,6 +16,7 @@ service-specific adapter. import asyncio import base64 +import copy import io import wave from dataclasses import dataclass @@ -198,7 +199,12 @@ class LLMContext: """ return self.get_messages() - def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]: + def get_messages( + self, + llm_specific_filter: Optional[str] = None, + *, + truncate_large_values: bool = False, + ) -> List[LLMContextMessage]: """Get the current messages list. Args: @@ -207,22 +213,110 @@ class LLMContext: 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: - 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 + 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 + + msg = 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: diff --git a/tests/test_llm_context.py b/tests/test_llm_context.py new file mode 100644 index 000000000..e9b0c4d8a --- /dev/null +++ b/tests/test_llm_context.py @@ -0,0 +1,346 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Unit tests for LLMContext core functionality.""" + +import unittest + +from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMSpecificMessage, +) + + +class TestGetMessagesTruncateLargeValues(unittest.TestCase): + """Tests for LLMContext.get_messages(truncate_large_values=True).""" + + # -- Standard messages: binary elision ----------------------------------- + + def test_default_preserves_all_data(self): + """truncate_large_values defaults to False, preserving all data.""" + messages = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe this image"}, + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,/9j/4AAQSkZJRg=="}, + }, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages() + + self.assertEqual( + result[0]["content"][1]["image_url"]["url"], + "data:image/jpeg;base64,/9j/4AAQSkZJRg==", + ) + + def test_elides_base64_image_url(self): + """Base64 data:image/ URLs are replaced with a placeholder.""" + messages = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe this image"}, + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,/9j/4AAQSkZJRg=="}, + }, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["content"][0]["text"], "Describe this image") + self.assertEqual(result[0]["content"][1]["image_url"]["url"], "data:image/...") + + def test_preserves_http_image_url(self): + """HTTP image URLs are not elided (they aren't binary data).""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/image.jpg"}, + }, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual( + result[0]["content"][0]["image_url"]["url"], + "https://example.com/image.jpg", + ) + + def test_elides_input_audio_data(self): + """input_audio items have their data field elided.""" + messages = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Audio follows"}, + { + "type": "input_audio", + "input_audio": {"data": "UklGRiQA" * 1000, "format": "wav"}, + }, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["content"][1]["input_audio"]["data"], "...") + self.assertEqual(result[0]["content"][1]["input_audio"]["format"], "wav") + + def test_elides_audio_field(self): + """Items with an 'audio' field are elided (used by some realtime adapters).""" + messages = [ + { + "role": "user", + "content": [ + {"type": "input_audio", "audio": "UklGRiQA" * 1000}, + {"type": "audio", "audio": "UklGRiQA" * 1000}, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["content"][0]["audio"], "...") + self.assertEqual(result[0]["content"][1]["audio"], "...") + + def test_elides_top_level_mime_type_image(self): + """Messages with top-level mime_type image/ have their data elided.""" + messages = [ + { + "role": "user", + "mime_type": "image/png", + "data": "iVBORw0KGgoAAAANSU" * 1000, + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["data"], "...") + self.assertEqual(result[0]["mime_type"], "image/png") + + def test_mixed_content_elides_only_binary(self): + """In a message with text, image, and audio, only binary parts are elided.""" + messages = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Here is an image and audio"}, + { + "type": "image_url", + "image_url": {"url": "data:image/png;base64,iVBORw=="}, + }, + { + "type": "input_audio", + "input_audio": {"data": "UklGRiQA", "format": "wav"}, + }, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["content"][0]["text"], "Here is an image and audio") + self.assertEqual(result[0]["content"][1]["image_url"]["url"], "data:image/...") + self.assertEqual(result[0]["content"][2]["input_audio"]["data"], "...") + + def test_text_only_messages_unchanged(self): + """Plain text messages are completely unaffected.""" + messages = [ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "Hello!"}, + {"role": "assistant", "content": "Hi there!"}, + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result, messages) + + def test_does_not_mutate_original(self): + """Returns copies; originals are untouched.""" + original_url = "data:image/jpeg;base64,/9j/4AAQSkZJRg==" + messages = [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": original_url}, + }, + ], + } + ] + context = LLMContext(messages=messages) + _ = context.get_messages(truncate_large_values=True) + + self.assertEqual( + context.get_messages()[0]["content"][0]["image_url"]["url"], + original_url, + ) + + def test_multiple_images_all_elided(self): + """Multiple image_url items in the same message are all elided.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,AAAA"}, + }, + { + "type": "image_url", + "image_url": {"url": "data:image/png;base64,BBBB"}, + }, + { + "type": "image_url", + "image_url": {"url": "https://example.com/photo.jpg"}, + }, + ], + } + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["content"][0]["image_url"]["url"], "data:image/...") + self.assertEqual(result[0]["content"][1]["image_url"]["url"], "data:image/...") + self.assertEqual( + result[0]["content"][2]["image_url"]["url"], + "https://example.com/photo.jpg", + ) + + def test_works_with_llm_specific_filter(self): + """truncate_large_values works together with llm_specific_filter.""" + adapter = OpenAILLMAdapter() + std_msg = { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,/9j/4AAQ"}, + }, + ], + } + specific_msg = adapter.create_llm_specific_message( + {"role": "assistant", "content": "response"} + ) + context = LLMContext(messages=[std_msg, specific_msg]) + + result = context.get_messages("openai", truncate_large_values=True) + + self.assertEqual(len(result), 2) + self.assertEqual(result[0]["content"][0]["image_url"]["url"], "data:image/...") + + def test_string_content_with_no_binary(self): + """Messages with string content (not list) pass through fine.""" + messages = [ + {"role": "user", "content": "Just a string"}, + ] + context = LLMContext(messages=messages) + result = context.get_messages(truncate_large_values=True) + + self.assertEqual(result[0]["content"], "Just a string") + + # -- LLMSpecificMessage: long-string truncation -------------------------- + + def test_llm_specific_short_values_preserved(self): + """Short string values in LLMSpecificMessage are kept as-is.""" + inner = {"type": "thought", "text": "brief thought"} + specific_msg = LLMSpecificMessage(llm="anthropic", message=inner) + context = LLMContext(messages=[specific_msg]) + + result = context.get_messages(truncate_large_values=True) + + self.assertIsInstance(result[0], LLMSpecificMessage) + self.assertEqual(result[0].message["type"], "thought") + self.assertEqual(result[0].message["text"], "brief thought") + + def test_llm_specific_long_string_truncated(self): + """Long string values in LLMSpecificMessage are truncated.""" + long_signature = "a" * 500 + inner = {"type": "thought", "text": "short", "signature": long_signature} + specific_msg = LLMSpecificMessage(llm="anthropic", message=inner) + context = LLMContext(messages=[specific_msg]) + + result = context.get_messages(truncate_large_values=True) + + msg = result[0].message + self.assertEqual(msg["type"], "thought") + self.assertEqual(msg["text"], "short") + # Signature should be truncated + self.assertIn("...", msg["signature"]) + self.assertIn("500 chars", msg["signature"]) + self.assertTrue(len(msg["signature"]) < len(long_signature)) + + def test_llm_specific_nested_dict_truncated(self): + """Long strings nested in dicts within LLMSpecificMessage are truncated.""" + inner = { + "type": "thought_signature", + "signature": "x" * 200, + "bookmark": {"text": "y" * 200}, + } + specific_msg = LLMSpecificMessage(llm="google", message=inner) + context = LLMContext(messages=[specific_msg]) + + result = context.get_messages(truncate_large_values=True) + + msg = result[0].message + self.assertEqual(msg["type"], "thought_signature") + self.assertIn("...", msg["signature"]) + self.assertIn("...", msg["bookmark"]["text"]) + + def test_llm_specific_list_values_truncated(self): + """Long strings inside lists within LLMSpecificMessage are truncated.""" + inner = {"items": ["short", "a" * 200]} + specific_msg = LLMSpecificMessage(llm="test", message=inner) + context = LLMContext(messages=[specific_msg]) + + result = context.get_messages(truncate_large_values=True) + + msg = result[0].message + self.assertEqual(msg["items"][0], "short") + self.assertIn("...", msg["items"][1]) + + def test_llm_specific_non_string_values_preserved(self): + """Non-string values (ints, bools, None) in LLMSpecificMessage are untouched.""" + inner = {"type": "test", "count": 42, "active": True, "extra": None} + specific_msg = LLMSpecificMessage(llm="test", message=inner) + context = LLMContext(messages=[specific_msg]) + + result = context.get_messages(truncate_large_values=True) + + msg = result[0].message + self.assertEqual(msg["count"], 42) + self.assertEqual(msg["active"], True) + self.assertIsNone(msg["extra"]) + + def test_llm_specific_does_not_mutate_original(self): + """Truncation returns a copy; original LLMSpecificMessage is untouched.""" + long_sig = "a" * 500 + inner = {"signature": long_sig} + specific_msg = LLMSpecificMessage(llm="anthropic", message=inner) + context = LLMContext(messages=[specific_msg]) + + _ = context.get_messages(truncate_large_values=True) + + self.assertEqual(specific_msg.message["signature"], long_sig) + + +if __name__ == "__main__": + unittest.main()