diff --git a/.sampo/changesets/cumulative-delta-output-tokens.md b/.sampo/changesets/cumulative-delta-output-tokens.md new file mode 100644 index 000000000..cbc8bc720 --- /dev/null +++ b/.sampo/changesets/cumulative-delta-output-tokens.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Streamed Anthropic generations no longer stack output tokens when a stream emits more than one `message_delta` (server tools can send several). Each delta's usage is cumulative, so the accumulator now keeps the latest total instead of adding every delta together. diff --git a/posthog/ai/anthropic/_anthropic_stream.py b/posthog/ai/anthropic/_anthropic_stream.py index ff3fbe0b8..2df160ef8 100644 --- a/posthog/ai/anthropic/_anthropic_stream.py +++ b/posthog/ai/anthropic/_anthropic_stream.py @@ -53,7 +53,15 @@ def consume(self, event: Any) -> None: event_usage = extract_anthropic_usage_from_event(event) merge_usage_stats(self.usage_stats, event_usage) if getattr(event, "type", None) == "message_delta": - self._apply_cumulative_input_usage(getattr(event, "usage", None)) + usage = getattr(event, "usage", None) + self._apply_cumulative_input_usage(usage) + output_tokens = getattr(usage, "output_tokens", None) + if isinstance(output_tokens, int) and not isinstance(output_tokens, bool): + # output_tokens on a message_delta is cumulative too, so the + # incremental merge above must not stack successive deltas + # (server tools can emit several in one stream). Keep the + # latest total; a field the event omits leaves the prior one. + self.usage_stats["output_tokens"] = output_tokens if getattr(event, "type", None) == "content_block_start": block, tool = handle_anthropic_content_block_start(event) diff --git a/posthog/test/ai/anthropic/test_anthropic.py b/posthog/test/ai/anthropic/test_anthropic.py index 08c66e7e2..ffc6feeed 100644 --- a/posthog/test/ai/anthropic/test_anthropic.py +++ b/posthog/test/ai/anthropic/test_anthropic.py @@ -907,6 +907,87 @@ async def mock_async_create(**kwargs): assert_cumulative_delta_usage(mock_client.capture.call_args.kwargs["properties"]) +def anthropic_stream_with_two_cumulative_deltas(): + # Server tools can emit more than one message_delta in a single stream, + # and each delta's usage is cumulative: later values replace earlier ones + # rather than adding to them. + return [ + RawMessageStartEvent( + type="message_start", + message=Message( + id="msg_1", + type="message", + role="assistant", + model="claude-haiku-4-5", + content=[], + stop_reason=None, + stop_sequence=None, + usage=Usage( + input_tokens=100, + output_tokens=1, + cache_creation_input_tokens=0, + cache_read_input_tokens=0, + ), + ), + ), + RawMessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason="pause_turn", stop_sequence=None), + usage=MessageDeltaUsage(output_tokens=50), + ), + RawMessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason="end_turn", stop_sequence=None), + usage=MessageDeltaUsage(output_tokens=110), + ), + RawMessageStopEvent(type="message_stop"), + ] + + +def test_streaming_multiple_message_deltas_do_not_stack_output_tokens(mock_client): + with patch( + "anthropic.resources.Messages.create", + return_value=iter(anthropic_stream_with_two_cumulative_deltas()), + ): + client = Anthropic(api_key="test-key", posthog_client=mock_client) + response = client.messages.create( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Hello"}], + stream=True, + ) + list(response) + + props = mock_client.capture.call_args.kwargs["properties"] + assert props["$ai_output_tokens"] == 110 + + +@pytest.mark.asyncio +async def test_streaming_multiple_message_deltas_do_not_stack_output_tokens_async( + mock_client, +): + async def mock_async_stream(): + for event in anthropic_stream_with_two_cumulative_deltas(): + yield event + + async def mock_async_create(**kwargs): + return mock_async_stream() + + with patch( + "anthropic.resources.messages.AsyncMessages.create", + side_effect=mock_async_create, + ): + client = AsyncAnthropic(api_key="test-key", posthog_client=mock_client) + response = await client.messages.create( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Hello"}], + stream=True, + ) + [event async for event in response] + + props = mock_client.capture.call_args.kwargs["properties"] + assert props["$ai_output_tokens"] == 110 + + def test_anthropic_bedrock_preserves_cache_creation_ttl_breakdown( mock_client, mock_anthropic_response_with_cache_ttl ):