fix: 缓存 token 计费按上游协议区分口径
- 新增 ComputeCost：OpenAI 系 prompt 含缓存读需扣减（异常数据钳制为 0）； Anthropic input_tokens 不含缓存，按原值计费，缓存写按输入价 ×1.25（原实现误用输出价） - recordUsage 传入上游协议（用量语义跟随解析它的上游响应，而非客户端协议） - 补充单元测试覆盖两种协议口径与边界情况
This commit is contained in:
Sakurasan
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user