diff --git a/python/semantic_kernel/connectors/ai/anthropic/services/utils.py b/python/semantic_kernel/connectors/ai/anthropic/services/utils.py index e41905e1cc91..7c74d13c7ce8 100644 --- a/python/semantic_kernel/connectors/ai/anthropic/services/utils.py +++ b/python/semantic_kernel/connectors/ai/anthropic/services/utils.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import base64 import json import logging from collections.abc import Callable, Mapping @@ -9,8 +10,10 @@ from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.function_call_content import FunctionCallContent from semantic_kernel.contents.function_result_content import FunctionResultContent +from semantic_kernel.contents.image_content import ImageContent from semantic_kernel.contents.text_content import TextContent from semantic_kernel.contents.utils.author_role import AuthorRole +from semantic_kernel.exceptions.service_exceptions import ServiceInvalidRequestError from semantic_kernel.functions.kernel_function_metadata import KernelFunctionMetadata logger: logging.Logger = logging.getLogger(__name__) @@ -21,6 +24,102 @@ from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings +ANTHROPIC_SUPPORTED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} + + +def _infer_image_mime_type(raw_bytes: bytes | bytearray | None, mime_type: str | None) -> str: + """Infer or validate a supported image MIME type for Anthropic vision.""" + if mime_type and mime_type in ANTHROPIC_SUPPORTED_IMAGE_TYPES: + return mime_type + if raw_bytes: + prefix = bytes(raw_bytes[:16]) + if prefix.startswith(b"\x89PNG\r\n\x1a\n"): + return "image/png" + if prefix.startswith(b"\xff\xd8\xff"): + return "image/jpeg" + if prefix.startswith((b"GIF87a", b"GIF89a")): + return "image/gif" + if prefix.startswith(b"RIFF") and b"WEBP" in prefix: + return "image/webp" + if mime_type and mime_type in ANTHROPIC_SUPPORTED_IMAGE_TYPES: + return mime_type + + if mime_type == "text/plain" or not mime_type: + return "image/jpeg" + + raise ServiceInvalidRequestError( + f"Unsupported image format: '{mime_type}'. Anthropic Claude vision only supports " + f"{', '.join(sorted(ANTHROPIC_SUPPORTED_IMAGE_TYPES))}." + ) + + +def _has_inline_image_data(image_content: ImageContent) -> bool: + """Check whether the ImageContent carries inline image data (not just a remote reference).""" + data = image_content.data + if isinstance(data, (bytes, bytearray)): + # BinaryContent defaults data to b"" when only a URI is given. + return len(data) > 0 + return bool(image_content.data_string) or data is not None + + +def _create_base64_image_content(image_content: ImageContent) -> dict[str, Any]: + """Create an Anthropic base64 image content block from inline image data.""" + raw_bytes = image_content.data if isinstance(image_content.data, (bytes, bytearray)) else None + mime_type = _infer_image_mime_type(raw_bytes, image_content.mime_type) + + if raw_bytes is not None: + data = base64.b64encode(raw_bytes).decode("utf-8") + elif image_content.data_string: + data = image_content.data_string + elif image_content.data is not None: + data = str(image_content.data) + else: + raise ServiceInvalidRequestError( + "ImageContent without data, data_uri, or valid http(s) uri while formatting message for Anthropic." + ) + + if not data: + raise ServiceInvalidRequestError( + "ImageContent without data, data_uri, or valid http(s) uri while formatting message for Anthropic." + ) + + return { + "type": "image", + "source": { + "type": "base64", + "media_type": mime_type, + "data": data, + }, + } + + +def _create_image_content(image_content: ImageContent) -> dict[str, Any]: + """Create an Anthropic image content block from an ImageContent object. + + Inline image data takes precedence over a remote URL: per ImageContent's + contract, its URI may refer to different content than its data, so sending + the URL when bytes are available can send the wrong image (or fail for + URLs requiring authentication). + """ + if _has_inline_image_data(image_content): + return _create_base64_image_content(image_content) + + if image_content.uri: + uri_str = str(image_content.uri) + if uri_str.startswith(("http://", "https://")): + return { + "type": "image", + "source": { + "type": "url", + "url": uri_str, + }, + } + + raise ServiceInvalidRequestError( + "ImageContent without data, data_uri, or valid http(s) uri while formatting message for Anthropic." + ) + + def _format_user_message(message: ChatMessageContent) -> dict[str, Any]: """Format a user message to the expected object for the Anthropic client. @@ -30,9 +129,30 @@ def _format_user_message(message: ChatMessageContent) -> dict[str, Any]: Returns: The formatted user message. """ + if not message.items or (len(message.items) == 1 and isinstance(message.items[0], TextContent)): + return { + "role": "user", + "content": message.content, + } + + contents: list[dict[str, Any]] = [] + for item in message.items: + if isinstance(item, TextContent): + if item.text: + contents.append({ + "type": "text", + "text": item.text, + }) + elif isinstance(item, ImageContent): + contents.append(_create_image_content(item)) + else: + logger.warning( + f"Unsupported item type in User message while formatting chat history for Anthropic: {type(item)}" + ) + return { "role": "user", - "content": message.content, + "content": contents, } diff --git a/python/tests/unit/connectors/ai/anthropic/services/test_anthropic_chat_completion.py b/python/tests/unit/connectors/ai/anthropic/services/test_anthropic_chat_completion.py index bff83bfe89d6..c5077eb8258d 100644 --- a/python/tests/unit/connectors/ai/anthropic/services/test_anthropic_chat_completion.py +++ b/python/tests/unit/connectors/ai/anthropic/services/test_anthropic_chat_completion.py @@ -16,8 +16,10 @@ ) from semantic_kernel.contents.chat_history import ChatHistory from semantic_kernel.contents.chat_message_content import ChatMessageContent, FunctionCallContent, TextContent +from semantic_kernel.contents.image_content import ImageContent from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent from semantic_kernel.contents.utils.author_role import AuthorRole +from semantic_kernel.connectors.ai.anthropic.services.utils import _format_user_message from semantic_kernel.exceptions import ( ServiceInitializationError, ServiceInvalidExecutionSettingsError, @@ -293,6 +295,181 @@ def test_prompt_execution_settings_class(anthropic_unit_test_env): assert prompt_execution_settings == AnthropicChatPromptExecutionSettings +def test_format_user_message_text_only(): + message = ChatMessageContent(role=AuthorRole.USER, content="Hello Anthropic") + formatted = _format_user_message(message) + assert formatted == {"role": "user", "content": "Hello Anthropic"} + + +def test_format_user_message_with_image(): + dummy_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + TextContent(text="Describe this image"), + ImageContent(data_uri=f"data:image/png;base64,{dummy_b64}"), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 2 + assert formatted["content"][0] == {"type": "text", "text": "Describe this image"} + assert formatted["content"][1] == { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": dummy_b64, + }, + } + + +def test_format_user_message_image_only(): + dummy_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(data_uri=f"data:image/png;base64,{dummy_b64}"), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["media_type"] == "image/png" + assert formatted["content"][0]["source"]["data"] == dummy_b64 + + +def test_format_user_message_raw_bytes(): + raw_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15c4" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(data=raw_png, mime_type="image/png"), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["media_type"] == "image/png" + import base64 + assert formatted["content"][0]["source"]["data"] == base64.b64encode(raw_png).decode("utf-8") + + +def test_format_user_message_image_url_only(): + url = "https://example.com/sample_image.png" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(uri=url), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["type"] == "url" + assert formatted["content"][0]["source"]["url"] == url + + +def test_format_user_message_dual_source_prefers_inline_bytes(): + # ImageContent documents that its URI may refer to different content than + # its data, so inline bytes must win over the remote URL. + raw_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15c4" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent( + uri="https://example.com/remote.png", + data=raw_png, + mime_type="image/png", + ), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["type"] == "base64" + assert formatted["content"][0]["source"]["media_type"] == "image/png" + import base64 + assert formatted["content"][0]["source"]["data"] == base64.b64encode(raw_png).decode("utf-8") + + +def test_format_user_message_dual_source_prefers_data_uri(): + dummy_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent( + uri="https://example.com/remote.png", + data_uri=f"data:image/png;base64,{dummy_b64}", + ), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["type"] == "base64" + assert formatted["content"][0]["source"]["media_type"] == "image/png" + assert formatted["content"][0]["source"]["data"] == dummy_b64 + + +def test_format_user_message_raw_bytes_default_mime_type(): + raw_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15c4" + # When mime_type is omitted, BinaryContent defaults to text/plain + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(data=raw_png), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["media_type"] == "image/png" + + +def test_format_user_message_jpeg_inferred_mime_type(): + raw_jpeg = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x01\x00`\x00`\x00\x00" + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(data=raw_jpeg), + ], + ) + formatted = _format_user_message(message) + assert formatted["role"] == "user" + assert len(formatted["content"]) == 1 + assert formatted["content"][0]["type"] == "image" + assert formatted["content"][0]["source"]["media_type"] == "image/jpeg" + + +def test_format_user_message_unsupported_svg_image_raises(): + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(data=b"", mime_type="image/svg+xml"), + ], + ) + with pytest.raises(ServiceInvalidRequestError, match="Unsupported image format: 'image/svg\\+xml'"): + _format_user_message(message) + + +def test_format_user_message_unsupported_bmp_data_uri_raises(): + message = ChatMessageContent( + role=AuthorRole.USER, + items=[ + ImageContent(data_uri="data:image/bmp;base64,Qk0="), + ], + ) + with pytest.raises(ServiceInvalidRequestError, match="Unsupported image format: 'image/bmp'"): + _format_user_message(message) + + async def test_with_different_execution_settings(kernel: Kernel, mock_anthropic_client_completion: MagicMock): chat_history = MagicMock() settings = OpenAIChatPromptExecutionSettings(temperature=0.2)