REDROOM

Navigation

Breadcrumbs

/ / opt / hc_python / lib / python3.12 / site-packages / sentry_sdk / ai

Create

Create File | Create Folder | Upload Files | Run Command

Edit File Size: 25.42 KB

//opt/hc_python/lib/python3.12/site-packages/sentry_sdk/ai/utils.py

Text Base64

import inspect
import json
from copy import deepcopy
from typing import TYPE_CHECKING

from sentry_sdk._types import BLOB_DATA_SUBSTITUTE
from sentry_sdk.ai.consts import DATA_URL_BASE64_REGEX

if TYPE_CHECKING:
    from typing import Any, Callable, Dict, List, Optional, Tuple, Union

from sentry_sdk.tracing import Span

import sentry_sdk
from sentry_sdk.traces import StreamedSpan
from sentry_sdk.tracing_utils import has_span_streaming_enabled
from sentry_sdk.utils import logger

MAX_GEN_AI_MESSAGE_BYTES = 20_000  # 20KB
# Maximum characters when only a single message is left after bytes truncation
MAX_SINGLE_MESSAGE_CONTENT_CHARS = 10_000

class GEN_AI_ALLOWED_MESSAGE_ROLES:
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"

GEN_AI_MESSAGE_ROLE_REVERSE_MAPPING = {
    GEN_AI_ALLOWED_MESSAGE_ROLES.SYSTEM: ["system"],
    GEN_AI_ALLOWED_MESSAGE_ROLES.USER: ["user", "human"],
    GEN_AI_ALLOWED_MESSAGE_ROLES.ASSISTANT: ["assistant", "ai"],
    GEN_AI_ALLOWED_MESSAGE_ROLES.TOOL: ["tool", "tool_call"],
}

GEN_AI_MESSAGE_ROLE_MAPPING = {}
for target_role, source_roles in GEN_AI_MESSAGE_ROLE_REVERSE_MAPPING.items():
    for source_role in source_roles:
        GEN_AI_MESSAGE_ROLE_MAPPING[source_role] = target_role

def parse_data_uri(url: str) -> "Tuple[str, str]":
    """
    Parse a data URI and return (mime_type, content).

Data URI format (RFC 2397): data:[<mediatype>][;base64],<data>

Examples:
        data:image/jpeg;base64,/9j/4AAQ... → ("image/jpeg", "/9j/4AAQ...")
        data:text/plain,Hello → ("text/plain", "Hello")
        data:;base64,SGVsbG8= → ("", "SGVsbG8=")

Raises:
        ValueError: If the URL is not a valid data URI (missing comma separator)
    """
    if "," not in url:
        raise ValueError("Invalid data URI: missing comma separator")

header, content = url.split(",", 1)

# Extract mime type from header
    # Format: "data:<mime>[;param1][;param2]..." e.g. "data:image/jpeg;base64"
    # Remove "data:" prefix, then take everything before the first semicolon
    if header.startswith("data:"):
        mime_part = header[5:]  # Remove "data:" prefix
    else:
        mime_part = header

mime_type = mime_part.split(";")[0]

return mime_type, content

def get_modality_from_mime_type(mime_type: str) -> str:
    """
    Infer the content modality from a MIME type string.

Args:
        mime_type: A MIME type string (e.g., "image/jpeg", "audio/mp3")

Returns:
        One of: "image", "audio", "video", or "document"
        Defaults to "image" for unknown or empty MIME types.

Examples:
        "image/jpeg" -> "image"
        "audio/mp3" -> "audio"
        "video/mp4" -> "video"
        "application/pdf" -> "document"
        "text/plain" -> "document"
    """
    if not mime_type:
        return "image"  # Default fallback

mime_lower = mime_type.lower()
    if mime_lower.startswith("image/"):
        return "image"
    elif mime_lower.startswith("audio/"):
        return "audio"
    elif mime_lower.startswith("video/"):
        return "video"
    elif mime_lower.startswith("application/") or mime_lower.startswith("text/"):
        return "document"
    else:
        return "image"  # Default fallback for unknown types

def transform_openai_content_part(
    content_part: "Dict[str, Any]",
) -> "Optional[Dict[str, Any]]":
    """
    Transform an OpenAI/LiteLLM content part to Sentry's standardized format.

This handles the OpenAI image_url format used by OpenAI and LiteLLM SDKs.

Input format:
    - {"type": "image_url", "image_url": {"url": "..."}}
    - {"type": "image_url", "image_url": "..."} (string shorthand)

Output format (one of):
    - {"type": "blob", "modality": "image", "mime_type": "...", "content": "..."}
    - {"type": "uri", "modality": "image", "mime_type": "", "uri": "..."}

Args:
        content_part: A dictionary representing a content part from OpenAI/LiteLLM

Returns:
        A transformed dictionary in standardized format, or None if the format
        is not OpenAI image_url format or transformation fails.
    """
    if not isinstance(content_part, dict):
        return None

block_type = content_part.get("type")

if block_type != "image_url":
        return None

image_url_data = content_part.get("image_url")
    if isinstance(image_url_data, str):
        url = image_url_data
    elif isinstance(image_url_data, dict):
        url = image_url_data.get("url", "")
    else:
        return None

if not url:
        return None

# Check if it's a data URI (base64 encoded)
    if url.startswith("data:"):
        try:
            mime_type, content = parse_data_uri(url)
            return {
                "type": "blob",
                "modality": get_modality_from_mime_type(mime_type),
                "mime_type": mime_type,
                "content": content,
            }
        except ValueError:
            # If parsing fails, return as URI
            return {
                "type": "uri",
                "modality": "image",
                "mime_type": "",
                "uri": url,
            }
    else:
        # Regular URL
        return {
            "type": "uri",
            "modality": "image",
            "mime_type": "",
            "uri": url,
        }

def transform_anthropic_content_part(
    content_part: "Dict[str, Any]",
) -> "Optional[Dict[str, Any]]":
    """
    Transform an Anthropic content part to Sentry's standardized format.

This handles the Anthropic image and document formats with source dictionaries.

Input format:
    - {"type": "image", "source": {"type": "base64", "media_type": "...", "data": "..."}}
    - {"type": "image", "source": {"type": "url", "media_type": "...", "url": "..."}}
    - {"type": "image", "source": {"type": "file", "media_type": "...", "file_id": "..."}}
    - {"type": "document", "source": {...}} (same source formats)

Output format (one of):
    - {"type": "blob", "modality": "...", "mime_type": "...", "content": "..."}
    - {"type": "uri", "modality": "...", "mime_type": "...", "uri": "..."}
    - {"type": "file", "modality": "...", "mime_type": "...", "file_id": "..."}

Args:
        content_part: A dictionary representing a content part from Anthropic

Returns:
        A transformed dictionary in standardized format, or None if the format
        is not Anthropic format or transformation fails.
    """
    if not isinstance(content_part, dict):
        return None

block_type = content_part.get("type")

if block_type not in ("image", "document") or "source" not in content_part:
        return None

source = content_part.get("source")
    if not isinstance(source, dict):
        return None

source_type = source.get("type")
    media_type = source.get("media_type", "")
    modality = (
        "document"
        if block_type == "document"
        else get_modality_from_mime_type(media_type)
    )

if source_type == "base64":
        return {
            "type": "blob",
            "modality": modality,
            "mime_type": media_type,
            "content": source.get("data", ""),
        }
    elif source_type == "url":
        return {
            "type": "uri",
            "modality": modality,
            "mime_type": media_type,
            "uri": source.get("url", ""),
        }
    elif source_type == "file":
        return {
            "type": "file",
            "modality": modality,
            "mime_type": media_type,
            "file_id": source.get("file_id", ""),
        }

return None

def transform_google_content_part(
    content_part: "Dict[str, Any]",
) -> "Optional[Dict[str, Any]]":
    """
    Transform a Google GenAI content part to Sentry's standardized format.

This handles the Google GenAI inline_data and file_data formats.

Input format:
    - {"inline_data": {"mime_type": "...", "data": "..."}}
    - {"file_data": {"mime_type": "...", "file_uri": "..."}}

Output format (one of):
    - {"type": "blob", "modality": "...", "mime_type": "...", "content": "..."}
    - {"type": "uri", "modality": "...", "mime_type": "...", "uri": "..."}

Args:
        content_part: A dictionary representing a content part from Google GenAI

Returns:
        A transformed dictionary in standardized format, or None if the format
        is not Google format or transformation fails.
    """
    if not isinstance(content_part, dict):
        return None

# Handle Google inline_data format
    if "inline_data" in content_part:
        inline_data = content_part.get("inline_data")
        if isinstance(inline_data, dict):
            mime_type = inline_data.get("mime_type", "")
            return {
                "type": "blob",
                "modality": get_modality_from_mime_type(mime_type),
                "mime_type": mime_type,
                "content": inline_data.get("data", ""),
            }
        return None

# Handle Google file_data format
    if "file_data" in content_part:
        file_data = content_part.get("file_data")
        if isinstance(file_data, dict):
            mime_type = file_data.get("mime_type", "")
            return {
                "type": "uri",
                "modality": get_modality_from_mime_type(mime_type),
                "mime_type": mime_type,
                "uri": file_data.get("file_uri", ""),
            }
        return None

return None

def transform_generic_content_part(
    content_part: "Dict[str, Any]",
) -> "Optional[Dict[str, Any]]":
    """
    Transform a generic/LangChain-style content part to Sentry's standardized format.

This handles generic formats where the type indicates the modality and
    the data is provided via direct base64, url, or file_id fields.

Input format:
    - {"type": "image", "base64": "...", "mime_type": "..."}
    - {"type": "audio", "url": "...", "mime_type": "..."}
    - {"type": "video", "base64": "...", "mime_type": "..."}
    - {"type": "file", "file_id": "...", "mime_type": "..."}

Args:
        content_part: A dictionary representing a content part in generic format

Returns:
        A transformed dictionary in standardized format, or None if the format
        is not generic format or transformation fails.
    """
    if not isinstance(content_part, dict):
        return None

block_type = content_part.get("type")

if block_type not in ("image", "audio", "video", "file"):
        return None

# Ensure it's not Anthropic format (which also uses type: "image")
    if "source" in content_part:
        return None

mime_type = content_part.get("mime_type", "")
    modality = block_type if block_type != "file" else "document"

# Check for base64 encoded content
    if "base64" in content_part:
        return {
            "type": "blob",
            "modality": modality,
            "mime_type": mime_type,
            "content": content_part.get("base64", ""),
        }
    # Check for URL reference
    elif "url" in content_part:
        return {
            "type": "uri",
            "modality": modality,
            "mime_type": mime_type,
            "uri": content_part.get("url", ""),
        }
    # Check for file_id reference
    elif "file_id" in content_part:
        return {
            "type": "file",
            "modality": modality,
            "mime_type": mime_type,
            "file_id": content_part.get("file_id", ""),
        }

return None

def transform_content_part(
    content_part: "Dict[str, Any]",
) -> "Optional[Dict[str, Any]]":
    """
    Transform a content part from various AI SDK formats to Sentry's standardized format.

This is a heuristic dispatcher that detects the format and delegates to the
    appropriate SDK-specific transformer. For direct SDK integration, prefer using
    the specific transformers directly:
    - transform_openai_content_part() for OpenAI/LiteLLM
    - transform_anthropic_content_part() for Anthropic
    - transform_google_content_part() for Google GenAI
    - transform_generic_content_part() for LangChain and other generic formats

Detection order:
    1. OpenAI: type == "image_url"
    2. Google: "inline_data" or "file_data" keys present
    3. Anthropic: type in ("image", "document") with "source" key
    4. Generic: type in ("image", "audio", "video", "file") with base64/url/file_id

Args:
        content_part: A dictionary representing a content part from an AI SDK

Returns:
        A transformed dictionary in standardized format, or None if the format
        is unrecognized or transformation fails.
    """
    if not isinstance(content_part, dict):
        return None

# Try OpenAI format first (most common, clear indicator)
    result = transform_openai_content_part(content_part)
    if result is not None:
        return result

# Try Google format (unique keys make it easy to detect)
    result = transform_google_content_part(content_part)
    if result is not None:
        return result

# Try Anthropic format (has "source" key)
    result = transform_anthropic_content_part(content_part)
    if result is not None:
        return result

# Try generic format as fallback
    result = transform_generic_content_part(content_part)
    if result is not None:
        return result

# Unrecognized format
    return None

def transform_message_content(content: "Any") -> "Any":
    """
    Transform message content, handling both string content and list of content blocks.

For list content, each item is transformed using transform_content_part().
    Items that cannot be transformed (return None) are kept as-is.

Args:
        content: Message content - can be a string, list of content blocks, or other

Returns:
        - String content: returned as-is
        - List content: list with each transformable item converted to standardized format
        - Other: returned as-is
    """
    if isinstance(content, str):
        return content

if isinstance(content, (list, tuple)):
        transformed = []
        for item in content:
            if isinstance(item, dict):
                result = transform_content_part(item)
                # If transformation succeeded, use the result; otherwise keep original
                transformed.append(result if result is not None else item)
            else:
                transformed.append(item)
        return transformed

return content

def _normalize_data(data: "Any", unpack: bool = True) -> "Any":
    # convert pydantic data (e.g. OpenAI v1+) to json compatible format
    if hasattr(data, "model_dump"):
        # Check if it's a class (type) rather than an instance
        # Model classes can be passed as arguments (e.g., for schema definitions)
        if inspect.isclass(data):
            return f"<ClassType: {data.__name__}>"

try:
            return _normalize_data(data.model_dump(), unpack=unpack)
        except Exception as e:
            logger.warning("Could not convert pydantic data to JSON: %s", e)
            return data if isinstance(data, (int, float, bool, str)) else str(data)

if isinstance(data, list):
        if unpack and len(data) == 1:
            return _normalize_data(data[0], unpack=unpack)  # remove empty dimensions
        return list(_normalize_data(x, unpack=unpack) for x in data)

if isinstance(data, dict):
        return {k: _normalize_data(v, unpack=unpack) for (k, v) in data.items()}

return data if isinstance(data, (int, float, bool, str)) else str(data)

def set_data_normalized(
    span: "Union[Span, StreamedSpan]",
    key: str,
    value: "Any",
    unpack: bool = True,
) -> None:
    normalized = _normalize_data(value, unpack=unpack)
    if isinstance(normalized, (int, float, bool, str)):
        _set_span_data_attribute(span, key, normalized)
    else:
        _set_span_data_attribute(span, key, json.dumps(normalized))

def _set_span_data_attribute(
    span: "Union[Span, StreamedSpan]", key: str, value: "Any"
) -> None:
    if isinstance(span, StreamedSpan):
        span.set_attribute(key, value)
    else:
        span.set_data(key, value)

def normalize_message_role(role: str) -> str:
    """
    Normalize a message role to one of the 4 allowed gen_ai role values.
    Maps "ai" -> "assistant" and keeps other standard roles unchanged.
    """
    return GEN_AI_MESSAGE_ROLE_MAPPING.get(role, role)

def normalize_message_roles(messages: "list[dict[str, Any]]") -> "list[dict[str, Any]]":
    """
    Normalize roles in a list of messages to use standard gen_ai role values.
    Creates a deep copy to avoid modifying the original messages.
    """
    normalized_messages = []
    for message in messages:
        if not isinstance(message, dict):
            normalized_messages.append(message)
            continue
        normalized_message = message.copy()
        if "role" in message:
            normalized_message["role"] = normalize_message_role(message["role"])
        normalized_messages.append(normalized_message)

return normalized_messages

def get_start_span_function() -> "Callable[..., Any]":
    if has_span_streaming_enabled(sentry_sdk.get_client().options):
        return sentry_sdk.traces.start_span

current_span = sentry_sdk.get_current_span()
    if isinstance(current_span, StreamedSpan):
        # mypy
        return sentry_sdk.traces.start_span

transaction_exists = (
        current_span is not None and current_span.containing_transaction is not None
    )
    return sentry_sdk.start_span if transaction_exists else sentry_sdk.start_transaction

def _truncate_single_message_content_if_present(
    message: "Dict[str, Any]", max_chars: int
) -> "Dict[str, Any]":
    """
    Truncate a message's content to at most `max_chars` characters and append an
    ellipsis if truncation occurs.
    """
    if not isinstance(message, dict) or "content" not in message:
        return message
    content = message["content"]

if isinstance(content, str):
        if len(content) <= max_chars:
            return message
        message["content"] = content[:max_chars] + "..."
        return message

if isinstance(content, list):
        remaining = max_chars
        for item in content:
            if isinstance(item, dict) and "text" in item:
                text = item["text"]
                if isinstance(text, str):
                    if len(text) > remaining:
                        item["text"] = text[:remaining] + "..."
                        remaining = 0
                    else:
                        remaining -= len(text)
        return message

return message

def _find_truncation_index(messages: "List[Dict[str, Any]]", max_bytes: int) -> int:
    """
    Find the index of the first message that would exceed the max bytes limit.
    Compute the individual message sizes, and return the index of the first message from the back
    of the list that would exceed the max bytes limit.
    """
    running_sum = 0
    for idx in range(len(messages) - 1, -1, -1):
        size = len(json.dumps(messages[idx], separators=(",", ":")).encode("utf-8"))
        running_sum += size
        if running_sum > max_bytes:
            return idx + 1

return 0

def _is_image_type_with_blob_content(item: "Dict[str, Any]") -> bool:
    """
    Some content blocks contain an image_url property with base64 content as its value.
    This is used to identify those while not leading to unnecessary copying of data when the image URL does not contain base64 content.
    """
    if item.get("type") != "image_url":
        return False

image_url_val = item.get("image_url")
    image_url = (
        image_url_val.get("url", "")
        if isinstance(image_url_val, dict)
        else (image_url_val or "")
    )
    data_url_match = DATA_URL_BASE64_REGEX.match(image_url)

return bool(data_url_match)

def redact_blob_message_parts(
    messages: "List[Dict[str, Any]]",
) -> "List[Dict[str, Any]]":
    """
    Redact blob message parts from the messages by replacing blob content with "[Filtered]".

This function creates a deep copy of messages that contain blob content to avoid
    mutating the original message dictionaries. Messages without blob content are
    returned as-is to minimize copying overhead.

e.g:
    {
        "role": "user",
        "content": [
            {
                "text": "How many ponies do you see in the image?",
                "type": "text"
            },
            {
                "type": "blob",
                "modality": "image",
                "mime_type": "image/jpeg",
                "content": "data:image/jpeg;base64,..."
            }
        ]
    }
    becomes:
    {
        "role": "user",
        "content": [
            {
                "text": "How many ponies do you see in the image?",
                "type": "text"
            },
            {
                "type": "blob",
                "modality": "image",
                "mime_type": "image/jpeg",
                "content": "[Filtered]"
            }
        ]
    }
    """

# First pass: check if any message contains blob content
    has_blobs = False
    for message in messages:
        if not isinstance(message, dict):
            continue
        content = message.get("content")
        if isinstance(content, list):
            for item in content:
                if isinstance(item, dict) and (
                    item.get("type") == "blob" or _is_image_type_with_blob_content(item)
                ):
                    has_blobs = True
                    break
        if has_blobs:
            break

# If no blobs found, return original messages to avoid unnecessary copying
    if not has_blobs:
        return messages

# Deep copy messages to avoid mutating the original
    messages_copy = deepcopy(messages)

# Second pass: redact blob content in the copy
    for message in messages_copy:
        if not isinstance(message, dict):
            continue

content = message.get("content")
        if isinstance(content, list):
            for item in content:
                if isinstance(item, dict):
                    if item.get("type") == "blob":
                        item["content"] = BLOB_DATA_SUBSTITUTE
                    elif _is_image_type_with_blob_content(item):
                        if isinstance(item["image_url"], dict):
                            item["image_url"]["url"] = BLOB_DATA_SUBSTITUTE
                        else:
                            item["image_url"] = BLOB_DATA_SUBSTITUTE

return messages_copy

def truncate_messages_by_size(
    messages: "List[Dict[str, Any]]",
    max_bytes: int = MAX_GEN_AI_MESSAGE_BYTES,
    max_single_message_chars: int = MAX_SINGLE_MESSAGE_CONTENT_CHARS,
) -> "Tuple[List[Dict[str, Any]], int]":
    """
    Returns a truncated messages list, consisting of
    - the last message, with its content truncated to `max_single_message_chars` characters,
      if the last message's size exceeds `max_bytes` bytes; otherwise,
    - the maximum number of messages, starting from the end of the `messages` list, whose total
      serialized size does not exceed `max_bytes` bytes.

In the single message case, the serialized message size may exceed `max_bytes`, because
    truncation is based only on character count in that case.
    """
    serialized_json = json.dumps(messages, separators=(",", ":"))
    current_size = len(serialized_json.encode("utf-8"))

if current_size <= max_bytes:
        return messages, 0

truncation_index = _find_truncation_index(messages, max_bytes)
    if truncation_index < len(messages):
        truncated_messages = messages[truncation_index:]
    else:
        truncation_index = len(messages) - 1
        truncated_messages = messages[-1:]

if len(truncated_messages) == 1:
        truncated_messages[0] = _truncate_single_message_content_if_present(
            deepcopy(truncated_messages[0]), max_chars=max_single_message_chars
        )

return truncated_messages, truncation_index

def truncate_and_annotate_messages(
    messages: "Optional[List[Dict[str, Any]]]",
    span: "Any",
    scope: "Any",
    max_single_message_chars: int = MAX_SINGLE_MESSAGE_CONTENT_CHARS,
) -> "Optional[List[Dict[str, Any]]]":
    if not messages:
        return None

messages = redact_blob_message_parts(messages)

truncated_message = _truncate_single_message_content_if_present(
        deepcopy(messages[-1]), max_chars=max_single_message_chars
    )
    if len(messages) > 1:
        scope._gen_ai_original_message_count[span.span_id] = len(messages)

return [truncated_message]

def truncate_and_annotate_embedding_inputs(
    messages: "Optional[List[Dict[str, Any]]]",
    span: "Any",
    scope: "Any",
    max_bytes: int = MAX_GEN_AI_MESSAGE_BYTES,
) -> "Optional[List[Dict[str, Any]]]":
    if not messages:
        return None

messages = redact_blob_message_parts(messages)

truncated_messages, removed_count = truncate_messages_by_size(messages, max_bytes)
    if removed_count > 0:
        scope._gen_ai_original_message_count[span.span_id] = len(messages)

return truncated_messages

def set_conversation_id(conversation_id: str) -> None:
    """
    Set the conversation_id in the scope.
    """
    scope = sentry_sdk.get_current_scope()
    scope.set_conversation_id(conversation_id)

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Directory Contents

Dirs: 1 Ã— Files: 6

Sort:

Enable drag-to-move

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-06-11 06:30:31	Edit Download recursive recursive
consts.py	275 B	lrw-r--r--	2026-06-11 06:30:30	Edit Download recursive recursive
monitoring.py	5.70 KB	lrw-r--r--	2026-06-11 06:30:30	Edit Download recursive recursive
utils.py	25.42 KB	lrw-r--r--	2026-06-11 06:30:30	Edit Download recursive recursive
_openai_completions_api.py	1.78 KB	lrw-r--r--	2026-06-11 06:30:30	Edit Download recursive recursive
_openai_responses_api.py	656 B	lrw-r--r--	2026-06-11 06:30:30	Edit Download recursive recursive
__init__.py	288 B	lrw-r--r--	2026-06-11 06:30:30	Edit Download recursive recursive