From ca4dc4b3b74b9f07bdc8a177b8a164e18f973f7f Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Wed, 2 Sep 2026 02:36:28 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=BC=93=E5=AD=98=20token=20=E8=AE=A1?= =?UTF-8?q?=E8=B4=B9=E6=8C=89=E4=B8=8A=E6=B8=B8=E5=8D=8F=E8=AE=AE=E5=8C=BA?= =?UTF-8?q?=E5=88=86=E5=8F=A3=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 ComputeCost：OpenAI 系 prompt 含缓存读需扣减（异常数据钳制为 0）； Anthropic input_tokens 不含缓存，按原值计费，缓存写按输入价 ×1.25（原实现误用输出价） - recordUsage 传入上游协议（用量语义跟随解析它的上游响应，而非客户端协议） - 补充单元测试覆盖两种协议口径与边界情况 --- backend/internal/proxy/cost.go | 29 +++++++++++ backend/internal/proxy/cost_test.go | 75 +++++++++++++++++++++++++++++ backend/internal/proxy/gateway.go | 24 ++++----- 3 files changed, 116 insertions(+), 12 deletions(-) create mode 100644 backend/internal/proxy/cost.go create mode 100644 backend/internal/proxy/cost_test.go diff --git a/backend/internal/proxy/cost.go b/backend/internal/proxy/cost.go new file mode 100644 index 0000000..94c1d20 --- /dev/null +++ b/backend/internal/proxy/cost.go @@ -0,0 +1,29 @@ +package proxy + +import ( + "opencatd-open/internal/proxy/convert" +) + +// cacheWriteInputMultiplier 缓存写（cache creation）相对输入价的倍数。 +// Anthropic 官方口径：缓存写按基础输入价的 1.25 倍计费（5m TTL）；OpenAI 系无缓存写概念。 +const cacheWriteInputMultiplier = 1.25 + +// ComputeCost 按上游协议的 token 语义计算一次请求的费用（USD）。 +// 价格均为每百万 token 的 USD 单价。tok 的 token 语义由解析它的上游协议决定： +// - chat / responses（OpenAI 系）：prompt_tokens 包含缓存读， +// 非缓存输入 = input − cacheRead；该协议没有缓存写，cacheCreation 恒为 0。 +// - messages（Anthropic）：input_tokens 不含缓存读/写（三个字段相互独立）， +// 非缓存输入 = input 原值，不得再扣减；缓存写按输入价 ×1.25。 +func ComputeCost(upstreamProto string, input, output, cacheRead, cacheCreation int, inputPrice, outputPrice, cacheReadPrice float64) float64 { + uncached := input + if upstreamProto != convert.ProtoMessages { + uncached -= cacheRead + if uncached < 0 { + uncached = 0 + } + } + return (float64(uncached)*inputPrice + + float64(cacheRead)*cacheReadPrice + + float64(cacheCreation)*inputPrice*cacheWriteInputMultiplier + + float64(output)*outputPrice) / 1e6 +} diff --git a/backend/internal/proxy/cost_test.go b/backend/internal/proxy/cost_test.go new file mode 100644 index 0000000..985b17c --- /dev/null +++ b/backend/internal/proxy/cost_test.go @@ -0,0 +1,75 @@ +package proxy + +import "testing" + +func TestComputeCost(t *testing.T) { + // 标准三价：输入 0.5 / 输出 1.5 / 缓存读 0.05（$/M） + const ( + inPrice = 0.5 + outPrice = 1.5 + cchPrice = 0.05 + ) + + tests := []struct { + name string + upstreamProto string + input int + output int + cacheRead int + cacheCreation int + inP float64 + outP float64 + cchP float64 + want float64 + }{ + { + // OpenAI：prompt 含缓存读，需扣减：(11000−10000)×0.5 + 10000×0.05 + 500×1.5 + name: "openai prompt includes cache read", + upstreamProto: "chat", + input: 11000, output: 500, cacheRead: 10000, + inP: inPrice, outP: outPrice, cchP: cchPrice, + want: (1000*inPrice + 10000*cchPrice + 500*outPrice) / 1e6, + }, + { + // Anthropic：input 不含缓存；缓存写按输入价 ×1.25 + name: "anthropic cache write at 1.25x input price", + upstreamProto: "messages", + input: 1000, output: 500, cacheRead: 10000, cacheCreation: 2000, + inP: inPrice, outP: outPrice, cchP: cchPrice, + want: (1000*inPrice + 10000*cchPrice + 2000*inPrice*1.25 + 500*outPrice) / 1e6, + }, + { + // Anthropic 口径不得扣减缓存读（否则这里非缓存输入会算成负数） + name: "anthropic does not subtract cache read", + upstreamProto: "messages", + input: 100, output: 0, cacheRead: 1000, + inP: inPrice, outP: outPrice, cchP: cchPrice, + want: (100*inPrice + 1000*cchPrice) / 1e6, + }, + { + // OpenAI 异常数据：cached > prompt 时非缓存输入钳制为 0，不出现负费用 + name: "openai clamps negative uncached input", + upstreamProto: "responses", + input: 100, output: 0, cacheRead: 5000, + inP: inPrice, outP: outPrice, cchP: cchPrice, + want: (5000 * cchPrice) / 1e6, + }, + { + // 未配置价格时费用为 0 + name: "no prices no cost", + upstreamProto: "messages", + input: 1000, output: 1000, cacheRead: 1000, cacheCreation: 1000, + want: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ComputeCost(tt.upstreamProto, tt.input, tt.output, tt.cacheRead, tt.cacheCreation, tt.inP, tt.outP, tt.cchP) + if diff := got - tt.want; diff > 1e-12 || diff < -1e-12 { + t.Fatalf("ComputeCost(%q, in=%d, out=%d, cr=%d, cw=%d) = %v, want %v", + tt.upstreamProto, tt.input, tt.output, tt.cacheRead, tt.cacheCreation, got, tt.want) + } + }) + } +} diff --git a/backend/internal/proxy/gateway.go b/backend/internal/proxy/gateway.go index b825e6e..2b663ab 100644 --- a/backend/internal/proxy/gateway.go +++ b/backend/internal/proxy/gateway.go @@ -204,7 +204,7 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { g.writeError(c, http.StatusServiceUnavailable, "no enabled channels for model: "+req.Model) g.recordUsage(req, nil, nil, usage.Event{ IsError: true, ErrorCode: "no_channel", - }, convert.TokenUsage{}) + }, convert.TokenUsage{}, "") return } @@ -273,7 +273,7 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { IsError: true, ErrorCode: "upstream_error", LatencyMS: int(time.Since(start).Milliseconds()), - }, convert.TokenUsage{}) + }, convert.TokenUsage{}, targetFormat) continue // 可重试：换下一个渠道 } @@ -289,7 +289,7 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { IsError: true, ErrorCode: fmt.Sprintf("upstream_%d", resp.StatusCode), LatencyMS: int(time.Since(start).Milliseconds()), - }, convert.TokenUsage{}) + }, convert.TokenUsage{}, targetFormat) // 429/5xx 可换渠道重试；4xx 直接透传 if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { lastErrStatus, lastErrBody = resp.StatusCode, string(body) @@ -318,13 +318,13 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { IsError: true, ErrorCode: errCode, LatencyMS: int(time.Since(start).Milliseconds()), - }, tok) + }, tok, targetFormat) return } // 成功记录：用量 + 定价计费。 g.recordUsage(req, cand, ch, usage.Event{ LatencyMS: int(time.Since(start).Milliseconds()), - }, tok) + }, tok, targetFormat) return } @@ -349,9 +349,10 @@ func rewriteModel(body []byte, upstreamModel string) []byte { return out } -// recordUsage 汇总一次请求的用量事件并异步落库。tok 为从上游响应提取的用量。 +// recordUsage 汇总一次请求的用量事件并异步落库。tok 为从上游响应提取的用量， +// 其 token 语义由 upstreamProto（渠道实际使用的上游协议）决定。 // cand/ch 可为 nil（无可用渠道的失败场景）。 -func (g *Gateway) recordUsage(req *Request, cand *channel.Candidate, ch *store.Channel, ev usage.Event, tok convert.TokenUsage) { +func (g *Gateway) recordUsage(req *Request, cand *channel.Candidate, ch *store.Channel, ev usage.Event, tok convert.TokenUsage, upstreamProto string) { if g.usageRec == nil { return } @@ -380,7 +381,8 @@ func (g *Gateway) recordUsage(req *Request, cand *channel.Candidate, ch *store.C } } // 定价与成本（价格按每百万 token 的 USD 单价）。 - // 成本口径：非缓存输入 × 输入价 + 缓存读 × 缓存价 + 缓存写与输出 × 输出价。 + // 成本口径按上游协议区分（详见 ComputeCost）：OpenAI 系 prompt 含缓存读需扣减； + // Anthropic 的 input_tokens 不含缓存，缓存写按输入价 ×1.25。 if ev.ModelID != 0 { if m, err := g.modelDAO.GetByID(ev.ModelID); err == nil { ev.InputPrice = m.InputPrice @@ -389,10 +391,8 @@ func (g *Gateway) recordUsage(req *Request, cand *channel.Candidate, ch *store.C } } if !ev.IsError { - ev.Cost = (float64(ev.PromptTokens-ev.CacheReadTokens)*ev.InputPrice + - float64(ev.CacheReadTokens)*ev.CacheReadPrice + - float64(ev.CacheCreationTokens)*ev.OutputPrice + - float64(ev.CompletionTokens)*ev.OutputPrice) / 1e6 + ev.Cost = ComputeCost(upstreamProto, tok.InputTokens, tok.OutputTokens, tok.CacheReadTokens, tok.CacheCreationTokens, + ev.InputPrice, ev.OutputPrice, ev.CacheReadPrice) } g.usageRec.Record(ev) }