fix: messages 流式缓存场景 token 记账错乱
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 末块合并不回归
This commit is contained in:
@@ -497,6 +497,14 @@ func (u *usageSink) push(raw json.RawMessage) {
|
|||||||
if json.Unmarshal(raw, &t) != nil {
|
if json.Unmarshal(raw, &t) != nil {
|
||||||
return
|
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 {
|
if t.PromptTokens > 0 {
|
||||||
u.us.PromptTokens = t.PromptTokens
|
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 消耗
|
// message_delta 不带 usage,只能按已收发内容估算,否则记账为 0 消耗
|
||||||
// 非流式上游必返回 usage,此处 in/out 非 0 不受影响。
|
// 非流式上游必返回 usage,此处 in/out 非 0 不受影响。
|
||||||
if status == store.UsageStatusCanceled || status == store.UsageStatusSuccess {
|
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 est, ok := c.Get("est_input_text"); ok {
|
||||||
if v, ok2 := est.(string); ok2 && v != "" {
|
if v, ok2 := est.(string); ok2 && v != "" {
|
||||||
in = int64(tokenizer.Count(v, mn))
|
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
|
_ = g.db.Where("name = ?", mn).First(&model).Error
|
||||||
if model.ID > 0 {
|
if model.ID > 0 {
|
||||||
modelID = model.ID
|
modelID = model.ID
|
||||||
|
// 计价口径:in=非缓存输入、cacheRead=缓存读、cacheCreate=缓存写(Anthropic 语义,
|
||||||
|
// messages 流式 message_delta 的 input_tokens 即非缓存部分)。
|
||||||
|
// 缓存写按 1.25× 输入价(Anthropic 5m 口径)。
|
||||||
cost = float64(in)/1e6*model.InputPrice +
|
cost = float64(in)/1e6*model.InputPrice +
|
||||||
float64(out)/1e6*model.OutputPrice +
|
float64(out)/1e6*model.OutputPrice +
|
||||||
float64(cacheRead)/1e6*model.CacheReadPrice
|
float64(cacheRead)/1e6*model.CacheReadPrice +
|
||||||
|
float64(cacheCreate)/1e6*model.InputPrice*1.25
|
||||||
} else {
|
} 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")
|
proto, _ := c.Get("protocol")
|
||||||
p, _ := proto.(string)
|
p, _ := proto.(string)
|
||||||
|
|||||||
@@ -46,3 +46,56 @@ func TestSSEDataPayloadSpacedAndMultiLine(t *testing.T) {
|
|||||||
t.Fatalf("multiline block: got %q, want yo", got)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user