From d8257df100f3a4d3b10af50607a724c7d5f559c0 Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Fri, 28 Aug 2026 18:11:24 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20messages=20=E6=B5=81=E5=BC=8F=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E5=9C=BA=E6=99=AF=20token=20=E8=AE=B0=E8=B4=A6?= =?UTF-8?q?=E9=94=99=E4=B9=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit qwen/dashscope 等上游 messages 流式的 usage 语义： - message_start.usage.input_tokens = 总输入 - message_delta.usage.input_tokens = 非缓存输入（缓存部分单列 cache_read/cache_creation 字段），是最终计费口径 原 usageSink 字段级合并中 delta 的 input 覆盖 start 的 input， 总输入丢失（31790 → 8）；缓存写也未参与计费。 - push：带 cache_* 字段的 usage 视为最终口径，整体替换 sink - finishUsage：缓存写按 1.25× 输入价计费（Anthropic 5m 口径）； 落库 input_tokens 存总量（含缓存读/写）便于对账 - 估算兜底条件排除已有缓存计数的请求 - 回归测试：缓存写/缓存命中/chat 末块合并不回归 --- server/internal/proxy/passthrough.go | 20 ++++++++-- server/internal/proxy/sse_fix_test.go | 53 +++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 3 deletions(-) diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index c56cfc0..ba5fd65 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -497,6 +497,14 @@ func (u *usageSink) push(raw json.RawMessage) { if json.Unmarshal(raw, &t) != nil { return } + // messages 流式最终事件（message_delta 的 usage）带 cache_* 字段，是上游的最终计费口径， + // 其中 input_tokens 仅指"非缓存输入"（与 message_start 的"总输入"语义不同）。 + // 整体替换而非字段合并，避免 delta 的非缓存 input 覆盖 start 的总 input 后语义错乱 + // （实际消耗由 finishUsage 按 input + cache_read + cache_creation 汇总）。 + if t.CacheReadInputTokens > 0 || t.CacheCreationInputTokens > 0 { + u.us = t + return + } // 零值不覆盖：不同事件携带不同字段 if t.PromptTokens > 0 { u.us.PromptTokens = t.PromptTokens @@ -550,7 +558,7 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time // message_delta 不带 usage，只能按已收发内容估算，否则记账为 0 消耗 // 非流式上游必返回 usage，此处 in/out 非 0 不受影响。 if status == store.UsageStatusCanceled || status == store.UsageStatusSuccess { - if in == 0 { + if in == 0 && cacheRead == 0 && cacheCreate == 0 { if est, ok := c.Get("est_input_text"); ok { if v, ok2 := est.(string); ok2 && v != "" { in = int64(tokenizer.Count(v, mn)) @@ -568,12 +576,18 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time _ = g.db.Where("name = ?", mn).First(&model).Error if model.ID > 0 { modelID = model.ID + // 计价口径：in=非缓存输入、cacheRead=缓存读、cacheCreate=缓存写（Anthropic 语义， + // messages 流式 message_delta 的 input_tokens 即非缓存部分）。 + // 缓存写按 1.25× 输入价（Anthropic 5m 口径）。 cost = float64(in)/1e6*model.InputPrice + float64(out)/1e6*model.OutputPrice + - float64(cacheRead)/1e6*model.CacheReadPrice + float64(cacheRead)/1e6*model.CacheReadPrice + + float64(cacheCreate)/1e6*model.InputPrice*1.25 } else { - cost = float64(in)/1e6*0.15 + float64(out)/1e6*0.60 // 无定价模型时按示例价 + cost = float64(in+cacheRead+cacheCreate)/1e6*0.15 + float64(out)/1e6*0.60 // 无定价模型时按示例价 } + // 落库的 input_tokens 存输入总量（含缓存读/写），与上游 message_start 口径一致，便于对账展示。 + in += cacheRead + cacheCreate proto, _ := c.Get("protocol") p, _ := proto.(string) diff --git a/server/internal/proxy/sse_fix_test.go b/server/internal/proxy/sse_fix_test.go index 0135080..a973810 100644 --- a/server/internal/proxy/sse_fix_test.go +++ b/server/internal/proxy/sse_fix_test.go @@ -46,3 +46,56 @@ func TestSSEDataPayloadSpacedAndMultiLine(t *testing.T) { t.Fatalf("multiline block: got %q, want yo", got) } } + +// 复现线上 qwen（dashscope）messages 流式缓存场景： +// message_start.usage.input_tokens 是总输入，message_delta.usage.input_tokens 是非缓存输入 +// 且带 cache_read/cache_creation，是最终计费口径。合并后： +// in(落库)=input+cache_read+cache_creation，计价 in 只算非缓存部分。 +// 此前 delta 的 input 覆盖 start 的 input 导致总输入丢失（31790 → 8）。 +func TestUsageSinkMessageDeltaAuthoritative(t *testing.T) { + sink := &usageSink{} + // message_start：总输入 31790 + start := json.RawMessage(`{"input_tokens":31790,"output_tokens":0}`) + sink.push(start) + if got := sink.us.InputTokens; got != 31790 { + t.Fatalf("after start: input=%d, want 31790", got) + } + // message_delta：非缓存输入 8 + 缓存写 33229（最终口径，整体替换） + delta := json.RawMessage(`{"output_tokens":8,"cache_creation_input_tokens":33229,"input_tokens":8,"cache_read_input_tokens":0}`) + sink.push(delta) + s := sink.Shape() + if s.InputTokens != 8 || s.CacheCreationInputTokens != 33229 || s.OutputTokens != 8 { + t.Fatalf("after delta: %+v, want input=8 cache_create=33229 output=8", s) + } + // finishUsage 口径：落库 input = 8 + 0 + 33229 = 33237（总量），计价 in=8、cacheCreate=33229 + in := s.InputTokens + s.CacheReadInputTokens + s.CacheCreationInputTokens + if in != 33237 { + t.Fatalf("total input=%d, want 33237", in) + } +} + +// 缓存命中场景（id=55）：delta input=76 非缓存 + cache_read=33229 + cache_creation=17。 +func TestUsageSinkCacheHitMerge(t *testing.T) { + sink := &usageSink{} + sink.push(json.RawMessage(`{"input_tokens":31862,"output_tokens":0}`)) + sink.push(json.RawMessage(`{"output_tokens":32,"cache_creation_input_tokens":17,"input_tokens":76,"cache_read_input_tokens":33229}`)) + s := sink.Shape() + total := s.InputTokens + s.CacheReadInputTokens + s.CacheCreationInputTokens + if total != 33322 { + t.Fatalf("total input=%d, want 33322 (76+33229+17)", total) + } + if s.OutputTokens != 32 { + t.Fatalf("output=%d, want 32", s.OutputTokens) + } +} + +// OpenAI chat 末块（无 cache 字段）仍走零值不覆盖合并，不受整体替换影响。 +func TestUsageSinkChatLastChunkStillMerges(t *testing.T) { + sink := &usageSink{} + sink.push(json.RawMessage(`{"prompt_tokens":65,"completion_tokens":0}`)) + sink.push(json.RawMessage(`{"prompt_tokens":65,"completion_tokens":82}`)) + s := sink.Shape() + if s.PromptTokens != 65 || s.CompletionTokens != 82 { + t.Fatalf("chat merge broken: %+v", s) + } +}