diff --git a/providers/anthropic/anthropic.go b/providers/anthropic/anthropic.go index f8845e01f..509939628 100644 --- a/providers/anthropic/anthropic.go +++ b/providers/anthropic/anthropic.go @@ -1725,11 +1725,15 @@ func (a languageModel) Stream(ctx context.Context, call fantasy.Call) (fantasy.S sawMessageStop := false var refusalMeta *RefusalMetadata + var messageStartUsage *anthropic.Usage for stream.Next() { chunk := stream.Current() _ = acc.Accumulate(chunk) switch chunk.Type { + case "message_start": + usage := chunk.AsMessageStart().Message.Usage + messageStartUsage = &usage case "content_block_start": contentBlockType := chunk.ContentBlock.Type switch contentBlockType { @@ -1965,16 +1969,25 @@ func (a languageModel) Stream(ctx context.Context, call fantasy.Call) (fantasy.S return } + usage := acc.Usage + if messageStartUsage != nil { + // message_delta can sum input usage across internal iterations. Use + // message_start for per-request context occupancy. + usage.InputTokens = messageStartUsage.InputTokens + usage.CacheCreationInputTokens = messageStartUsage.CacheCreationInputTokens + usage.CacheReadInputTokens = messageStartUsage.CacheReadInputTokens + } + yield(fantasy.StreamPart{ Type: fantasy.StreamPartTypeFinish, ID: acc.ID, FinishReason: mapFinishReason(string(acc.StopReason)), Usage: fantasy.Usage{ - InputTokens: acc.Usage.InputTokens, - OutputTokens: acc.Usage.OutputTokens, - TotalTokens: acc.Usage.InputTokens + acc.Usage.OutputTokens, - CacheCreationTokens: acc.Usage.CacheCreationInputTokens, - CacheReadTokens: acc.Usage.CacheReadInputTokens, + InputTokens: usage.InputTokens, + OutputTokens: usage.OutputTokens, + TotalTokens: usage.InputTokens + usage.OutputTokens, + CacheCreationTokens: usage.CacheCreationInputTokens, + CacheReadTokens: usage.CacheReadInputTokens, }, ProviderMetadata: refusalProviderMetadata(refusalMeta), }) diff --git a/providers/anthropic/anthropic_test.go b/providers/anthropic/anthropic_test.go index 60596c9d1..1becece23 100644 --- a/providers/anthropic/anthropic_test.go +++ b/providers/anthropic/anthropic_test.go @@ -1237,6 +1237,27 @@ func TestStream_SendsOutputConfigEffort(t *testing.T) { requireAnthropicEffort(t, call.body, EffortHigh) } +func TestStream_FinishUsesMessageStartInputUsage(t *testing.T) { + t.Parallel() + + parts := streamAnthropicParts(t, []string{ + anthropicSSEEvent("message_start", `{"type":"message_start","message":{"id":"msg_multi_iteration","type":"message","role":"assistant","model":"claude-fable-5","content":[],"stop_reason":null,"usage":{"input_tokens":2,"cache_read_input_tokens":139956,"cache_creation_input_tokens":7770,"output_tokens":0}}}`), + anthropicSSEEvent("content_block_start", `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`), + anthropicSSEEvent("content_block_delta", `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}}`), + anthropicSSEEvent("content_block_stop", `{"type":"content_block_stop","index":0}`), + anthropicSSEEvent("message_delta", `{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"input_tokens":4,"cache_read_input_tokens":287682,"cache_creation_input_tokens":15556,"output_tokens":4996}}`), + anthropicSSEEvent("message_stop", `{"type":"message_stop"}`), + }) + + finishes := streamPartsByType(parts, fantasy.StreamPartTypeFinish) + require.Len(t, finishes, 1) + require.Equal(t, int64(2), finishes[0].Usage.InputTokens) + require.Equal(t, int64(139956), finishes[0].Usage.CacheReadTokens) + require.Equal(t, int64(7770), finishes[0].Usage.CacheCreationTokens) + require.Equal(t, int64(4996), finishes[0].Usage.OutputTokens) + require.Equal(t, int64(4998), finishes[0].Usage.TotalTokens) +} + func TestStream_RequiresMessageStopBeforeFinish(t *testing.T) { t.Parallel()