From 82624b479a83aaf19427fc036b6327066e189306 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Fri, 16 Oct 2026 20:39:14 -0400 Subject: [PATCH] fix(ai): take cumulative Anthropic stream usage from message_delta Anthropic's message_delta usage is cumulative. With server tools (web search, web fetch, code execution) its input and cache counts exceed message_start's, but the stream accumulator only recorded message_start's input and cache counts, so streamed generations under-reported input tokens and cache writes. Keep the larger of the two for each count. Co-Authored-By: Claude Opus 5.5 --- .../cumulative-anthropic-stream-usage.md | 5 ++ posthog/ai/anthropic/_anthropic_stream.py | 25 ++++++ posthog/test/ai/anthropic/test_anthropic.py | 83 +++++++++++++++++++ 3 files changed, 113 insertions(+) create mode 100644 .sampo/changesets/cumulative-anthropic-stream-usage.md diff --git a/.sampo/changesets/cumulative-anthropic-stream-usage.md b/.sampo/changesets/cumulative-anthropic-stream-usage.md new file mode 100644 index 000000000..6e06e0513 --- /dev/null +++ b/.sampo/changesets/cumulative-anthropic-stream-usage.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Streamed Anthropic generations now take the input and cache token counts from the final `message_delta` usage, which is cumulative. With server tools such as web search, those counts grow past the `message_start` values, so streamed calls under-reported input tokens and cache writes. diff --git a/posthog/ai/anthropic/_anthropic_stream.py b/posthog/ai/anthropic/_anthropic_stream.py index e08f6f65e..ff3fbe0b8 100644 --- a/posthog/ai/anthropic/_anthropic_stream.py +++ b/posthog/ai/anthropic/_anthropic_stream.py @@ -26,9 +26,34 @@ def __init__(self) -> None: # web_search_tool_result, ...), so keep a list that lines up with it. self._blocks_by_index: List[StreamingContentBlock] = [] + def _apply_cumulative_input_usage(self, usage: Any) -> None: + # Usage on message_delta is cumulative. With server tools (web search, + # web fetch, code execution) its input and cache counts grow past + # message_start's, which is all the per-event merge records for them. + # The totals never shrink, so keep the larger value; a field the event + # omits leaves message_start's count in place. + input_tokens = getattr(usage, "input_tokens", None) + if isinstance(input_tokens, int): + self.usage_stats["input_tokens"] = max( + self.usage_stats.get("input_tokens") or 0, input_tokens + ) + cache_creation = getattr(usage, "cache_creation_input_tokens", None) + if isinstance(cache_creation, int): + self.usage_stats["cache_creation_input_tokens"] = max( + self.usage_stats.get("cache_creation_input_tokens") or 0, + cache_creation, + ) + cache_read = getattr(usage, "cache_read_input_tokens", None) + if isinstance(cache_read, int): + self.usage_stats["cache_read_input_tokens"] = max( + self.usage_stats.get("cache_read_input_tokens") or 0, cache_read + ) + 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)) 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 84275e9ad..08c66e7e2 100644 --- a/posthog/test/ai/anthropic/test_anthropic.py +++ b/posthog/test/ai/anthropic/test_anthropic.py @@ -824,6 +824,89 @@ async def mock_async_create(**kwargs): ) +def anthropic_stream_with_server_tool_usage(): + # With server tools, message_delta reports cumulative usage that exceeds + # message_start's (values from a recorded web search stream). + 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=2247, + output_tokens=26, + cache_creation_input_tokens=0, + cache_read_input_tokens=0, + ), + ), + ), + RawMessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason="end_turn", stop_sequence=None), + usage=MessageDeltaUsage( + input_tokens=2254, + output_tokens=110, + cache_creation_input_tokens=17269, + cache_read_input_tokens=0, + ), + ), + RawMessageStopEvent(type="message_stop"), + ] + + +def assert_cumulative_delta_usage(props): + assert props["$ai_input_tokens"] == 2254 + assert props["$ai_output_tokens"] == 110 + assert props["$ai_cache_creation_input_tokens"] == 17269 + assert props["$ai_cache_read_input_tokens"] == 0 + + +def test_streaming_takes_cumulative_message_delta_usage(mock_client): + with patch( + "anthropic.resources.Messages.create", + return_value=iter(anthropic_stream_with_server_tool_usage()), + ): + 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) + + assert_cumulative_delta_usage(mock_client.capture.call_args.kwargs["properties"]) + + +@pytest.mark.asyncio +async def test_streaming_takes_cumulative_message_delta_usage_async(mock_client): + async def mock_async_stream(): + for event in anthropic_stream_with_server_tool_usage(): + 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] + + assert_cumulative_delta_usage(mock_client.capture.call_args.kwargs["properties"]) + + def test_anthropic_bedrock_preserves_cache_creation_ttl_breakdown( mock_client, mock_anthropic_response_with_cache_ttl ):