diff --git a/python/semantic_kernel/connectors/ai/anthropic/services/anthropic_chat_completion.py b/python/semantic_kernel/connectors/ai/anthropic/services/anthropic_chat_completion.py index a2323a577a69..ba1fb679b9aa 100644 --- a/python/semantic_kernel/connectors/ai/anthropic/services/anthropic_chat_completion.py +++ b/python/semantic_kernel/connectors/ai/anthropic/services/anthropic_chat_completion.py @@ -16,11 +16,14 @@ from anthropic.types import ( ContentBlockStopEvent, Message, + MessageDeltaUsage, RawMessageDeltaEvent, RawMessageStartEvent, TextBlock, ToolUseBlock, + Usage, ) +from openai.types.completion_usage import PromptTokensDetails from pydantic import ValidationError from semantic_kernel.connectors.ai.anthropic.prompt_execution_settings.anthropic_prompt_execution_settings import ( @@ -32,6 +35,7 @@ ) from semantic_kernel.connectors.ai.anthropic.settings.anthropic_settings import AnthropicSettings from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase +from semantic_kernel.connectors.ai.completion_usage import CompletionUsage from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceType from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings from semantic_kernel.contents.chat_history import ChatHistory @@ -309,11 +313,12 @@ def _create_streaming_chat_message_content( ) elif isinstance(stream_event, RawMessageDeltaEvent): finish_reason = ANTHROPIC_TO_SEMANTIC_KERNEL_FINISH_REASON_MAP[str(stream_event.delta.stop_reason)] - output_tokens = stream_event.usage.output_tokens - if metadata is None: - metadata = {"usage": {"output_tokens": output_tokens}} - else: - metadata = metadata | {"usage": metadata.get("usage", {}) | {"output_tokens": output_tokens}} + usage = _get_completion_usage(stream_event.usage) + start_usage = metadata.get("usage") if metadata else None + if getattr(stream_event.usage, "input_tokens", None) is None and isinstance(start_usage, CompletionUsage): + # The input token counts are only reported at message start. + usage = start_usage.model_copy(update={"completion_tokens": usage.completion_tokens}) + metadata = (metadata or {}) | {"usage": usage} return StreamingChatMessageContent( choice_index=0, @@ -338,7 +343,7 @@ async def _send_chat_request(self, settings: AnthropicChatPromptExecutionSetting response_metadata: dict[str, Any] = {"id": response.id} if hasattr(response, "usage") and response.usage is not None: - response_metadata["usage"] = response.usage + response_metadata["usage"] = _get_completion_usage(response.usage) return [self._create_chat_message_content(response, response_metadata)] @@ -357,10 +362,10 @@ async def _send_chat_stream_request( """ try: async with self.async_client.messages.stream(**settings.prepare_settings_dict()) as stream: - metadata: dict[str, Any] = {"usage": {}, "id": None} + metadata: dict[str, Any] = {"usage": None, "id": None} async for stream_event in stream: if isinstance(stream_event, RawMessageStartEvent): - metadata["usage"]["input_tokens"] = stream_event.message.usage.input_tokens + metadata["usage"] = _get_completion_usage(stream_event.message.usage) metadata["id"] = stream_event.message.id elif isinstance(stream_event, (TextEvent, RawMessageDeltaEvent)) or ( isinstance(stream_event, ContentBlockStopEvent) @@ -391,3 +396,18 @@ def _get_tool_calls_from_message(self, message: Message) -> list[FunctionCallCon ) return tool_calls + + +def _get_completion_usage(usage: Usage | MessageDeltaUsage) -> CompletionUsage: + """Convert Anthropic usage to CompletionUsage. + + Anthropic reports the tokens read from and written to the prompt cache separately from `input_tokens`, + so they are added back to get the full prompt token count. Older SDK versions lack some of these fields. + """ + cache_read_tokens = getattr(usage, "cache_read_input_tokens", None) or 0 + cache_creation_tokens = getattr(usage, "cache_creation_input_tokens", None) or 0 + return CompletionUsage( + prompt_tokens=(getattr(usage, "input_tokens", None) or 0) + cache_read_tokens + cache_creation_tokens, + prompt_tokens_details=PromptTokensDetails(cached_tokens=cache_read_tokens) if cache_read_tokens else None, + completion_tokens=usage.output_tokens, + ) 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..14576c778eea 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 @@ -1,15 +1,19 @@ # Copyright (c) Microsoft. All rights reserved. +from functools import reduce from unittest.mock import AsyncMock, MagicMock, patch import pytest from anthropic import AsyncAnthropic -from anthropic.types import Message +from anthropic.lib.streaming import TextEvent +from anthropic.types import Message, MessageDeltaUsage, RawMessageDeltaEvent, RawMessageStartEvent, TextBlock, Usage +from anthropic.types.raw_message_delta_event import Delta from semantic_kernel.connectors.ai.anthropic.prompt_execution_settings.anthropic_prompt_execution_settings import ( AnthropicChatPromptExecutionSettings, ) from semantic_kernel.connectors.ai.anthropic.services.anthropic_chat_completion import AnthropicChatCompletion from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase +from semantic_kernel.connectors.ai.completion_usage import CompletionUsage from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import ( OpenAIChatPromptExecutionSettings, @@ -27,6 +31,8 @@ from semantic_kernel.functions.kernel_arguments import KernelArguments from semantic_kernel.functions.kernel_function_decorator import kernel_function from semantic_kernel.kernel import Kernel +from semantic_kernel.utils.telemetry.model_diagnostics import gen_ai_attributes +from semantic_kernel.utils.telemetry.model_diagnostics.model_diagnostics_settings import ModelDiagnosticSettings async def test_complete_chat_contents( @@ -547,3 +553,116 @@ def test_chat_completion_reset_settings( assert settings.tools is None assert settings.tool_choice is None + + +@pytest.fixture +def model_diagnostics_enabled(monkeypatch): + monkeypatch.setattr( + "semantic_kernel.utils.telemetry.model_diagnostics.decorators.MODEL_DIAGNOSTICS_SETTINGS", + ModelDiagnosticSettings(enable_otel_diagnostics=True), + ) + + +def _usage_span_attributes(mock_tracer: MagicMock) -> dict: + mock_span = mock_tracer.start_span.return_value + return { + args[0]: args[1] + for args, _ in mock_span.set_attribute.call_args_list + if args[0] in (gen_ai_attributes.INPUT_TOKENS, gen_ai_attributes.OUTPUT_TOKENS) + } + + +@patch("semantic_kernel.utils.telemetry.model_diagnostics.decorators.tracer") +async def test_usage_is_reported_as_completion_usage( + mock_tracer: MagicMock, + model_diagnostics_enabled, + mock_settings: AnthropicChatPromptExecutionSettings, +): + response = Message( + id="test_message_id", + content=[TextBlock(text="Hello", type="text")], + model="claude-sonnet-4-5", + role="assistant", + stop_reason="end_turn", + stop_sequence=None, + type="message", + usage=Usage(input_tokens=10, output_tokens=7, cache_read_input_tokens=100, cache_creation_input_tokens=20), + ) + client = MagicMock(spec=AsyncAnthropic) + client.messages = MagicMock() + client.messages.create = AsyncMock(return_value=response) + chat_completion = AnthropicChatCompletion(ai_model_id="test_model_id", api_key="", async_client=client) + chat_history = ChatHistory() + chat_history.add_user_message("test_user_message") + + result = await chat_completion.get_chat_message_content(chat_history, mock_settings) + + usage = result.metadata["usage"] + assert isinstance(usage, CompletionUsage) + assert usage.prompt_tokens == 130 + assert usage.prompt_tokens_details.cached_tokens == 100 + assert usage.completion_tokens == 7 + assert _usage_span_attributes(mock_tracer) == { + gen_ai_attributes.INPUT_TOKENS: 130, + gen_ai_attributes.OUTPUT_TOKENS: 7, + } + + +@patch("semantic_kernel.utils.telemetry.model_diagnostics.decorators.tracer") +async def test_streaming_usage_is_reported_as_completion_usage( + mock_tracer: MagicMock, + model_diagnostics_enabled, + mock_settings: AnthropicChatPromptExecutionSettings, +): + stream_events = [ + RawMessageStartEvent( + message=Message( + id="test_message_id", + content=[], + model="claude-sonnet-4-5", + role="assistant", + stop_reason=None, + stop_sequence=None, + type="message", + usage=Usage( + input_tokens=10, output_tokens=1, cache_read_input_tokens=100, cache_creation_input_tokens=20 + ), + ), + type="message_start", + ), + TextEvent(type="text", text="Hello", snapshot="Hello"), + RawMessageDeltaEvent( + delta=Delta(stop_reason="end_turn", stop_sequence=None), + type="message_delta", + usage=MessageDeltaUsage(output_tokens=7), + ), + ] + + async def async_generator(): + for event in stream_events: + yield event + + stream_mock = AsyncMock() + stream_mock.__aenter__.return_value = async_generator() + client = MagicMock(spec=AsyncAnthropic) + client.messages = MagicMock() + client.messages.stream.return_value = stream_mock + chat_completion = AnthropicChatCompletion(ai_model_id="test_model_id", api_key="", async_client=client) + chat_history = ChatHistory() + chat_history.add_user_message("test_user_message") + + chunks = [ + chunk + async for chunks in chat_completion.get_streaming_chat_message_contents(chat_history, mock_settings) + for chunk in chunks + ] + + usage = reduce(lambda x, y: x + y, chunks).metadata["usage"] + assert isinstance(usage, CompletionUsage) + assert usage.prompt_tokens == 130 + assert usage.prompt_tokens_details.cached_tokens == 100 + assert usage.completion_tokens == 7 + assert _usage_span_attributes(mock_tracer) == { + gen_ai_attributes.INPUT_TOKENS: 130, + gen_ai_attributes.OUTPUT_TOKENS: 7, + }