mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
fix(gateway): 流式请求从未记录 reasoning_tokens
## 症状
接了一个 reasoning-heavy 的源之后,读生产审计文件发现:它的**流式**请求全部
只有 completion、没有 reasoning。实测对照:
流式 compl=8166 reasoning=— ← 网关记的
非流式 compl=751 reasoning=167 ← 同一个 adapter,非流式正常
客户端两种情况的响应都正确,所以**没有任何报错**。只有审计记录更安静。
## 根因
两条记账路径,只有一条读了它:
recordChatUsage(非流式) 读 resp.TokenUsage.CompletionTokensDetails
流式的 usage 合并(内联在 handler) 只读 Prompt 和 Completion
cache 记账当时被镜像过去了,**reasoning 单纯没加**。所以一个花了几百 token
思考的模型,在流式下看起来像什么都没想。
这不是 qwen38 的问题 —— 是任何流式 reasoning 模型的问题,之前没有源暴露它。
## 为什么是「读审计文件」发现的,而不是失败的测试
现有判据 `TestRecordChatUsageKeepsReasoningTokens` 只覆盖非流式那条路径。
流式那段内联在 handler 里,测不到。抽出 `applyStreamUsageDetails` 之后
才有地方挂判据 —— 抽取不是为了好看,是因为不抽就没法测。
## 判据(四条)
- 流式记录 reasoning,且精确值(512)
- **reasoning 绝不被加进 completion** —— 关系式断言,不是「字段存在」检查。
写成 `rec.Compl += reasoning` 会让每个思考 token 被收两次费。
- 上游没报 reasoning 时不写字段(0 在审计里表示「未上报」,与「没思考」不同)
- cache 记账未被顺带破坏 —— 三种输入形态各测一遍
变异自证三次变红:删掉 reasoning、加到 completion 上、破坏 cache 块。
第四次变异(`d.ReasoningTokens > 0` 改成无条件赋值)没变红 —— 那一版与原版
**行为完全相同**:rec.Reasoning 初值就是 0,且 json tag 带 omitempty,两种情况下
客户端都看不到这个字段。这是无行为差异的变异,不是判据漏放。
## 生产实测
部署前(13 条):compl=204/277/1822/1264/740/… reasoning 全部未记录
部署后( 1 条):compl=213 reasoning=42 status=200
Co-Authored-By: ModelRouter <noreply@modelrouter.dev>
This commit is contained in:
@ -256,17 +256,17 @@ func (g *Gateway) checkQuota(ctx context.Context, model string) *quotaRejection
|
||||
}
|
||||
return nil
|
||||
}
|
||||
// A scope that simply does not list the model is a different problem from
|
||||
// a quota being exhausted, and the message has to say which: telling an
|
||||
// operator "not allowed for this key" when they configured a quota sends
|
||||
// them to the wrong setting. AUTO is checked here before the chain is even
|
||||
// consulted, so this is the message that surfaces for AUTO.
|
||||
if isAuto(model) {
|
||||
return "aRejection{msg: fmt.Sprintf(
|
||||
"model %q is not in this key's model scope, so it cannot use AUTO; "+
|
||||
"add %q to the key's models or remove the scope restriction", model, model)}
|
||||
}
|
||||
return "aRejection{msg: fmt.Sprintf("model %q is not allowed for this key", model)}
|
||||
// A scope that simply does not list the model is a different problem from
|
||||
// a quota being exhausted, and the message has to say which: telling an
|
||||
// operator "not allowed for this key" when they configured a quota sends
|
||||
// them to the wrong setting. AUTO is checked here before the chain is even
|
||||
// consulted, so this is the message that surfaces for AUTO.
|
||||
if isAuto(model) {
|
||||
return "aRejection{msg: fmt.Sprintf(
|
||||
"model %q is not in this key's model scope, so it cannot use AUTO; "+
|
||||
"add %q to the key's models or remove the scope restriction", model, model)}
|
||||
}
|
||||
return "aRejection{msg: fmt.Sprintf("model %q is not allowed for this key", model)}
|
||||
}
|
||||
|
||||
// quotaWindowSuffix describes a quota's reset window for an error message, so
|
||||
@ -1059,9 +1059,9 @@ func (g *Gateway) fireEnd(rec *Req) {
|
||||
// A SUBSET of completion_tokens, not an extra. Billing plugins that
|
||||
// add it to completion would double-charge every thinking token.
|
||||
"reasoning_tokens": rec.Reasoning,
|
||||
"image_count": rec.ImageCount,
|
||||
"error": rec.Err,
|
||||
"time": rec.Time,
|
||||
"image_count": rec.ImageCount,
|
||||
"error": rec.Err,
|
||||
"time": rec.Time,
|
||||
// chain_walk: the AUTO tier-by-tier trace, when the request went
|
||||
// through the chain. Empty for a direct request and for a gateway with
|
||||
// no plugins loaded. Absent rather than empty so a plugin can tell
|
||||
@ -1215,20 +1215,7 @@ func (g *Gateway) pumpStream(w http.ResponseWriter, rec *Req, chunks <-chan type
|
||||
var tut *types.TokenUsage
|
||||
if lastUsage != nil {
|
||||
tut = lastUsage
|
||||
// Write the upstream's cache accounting back onto the request record
|
||||
// so the audit trail carries the hit/miss split for streaming too.
|
||||
if d := tut.PromptTokensDetails; d != nil {
|
||||
rec.CacheReported = true
|
||||
if d.CachedTokens > 0 {
|
||||
rec.CacheHit = int64(d.CachedTokens)
|
||||
}
|
||||
} else if tut.PromptCacheHit > 0 {
|
||||
rec.CacheHit = int64(tut.PromptCacheHit)
|
||||
rec.CacheReported = true
|
||||
}
|
||||
if tut.PromptCacheMiss > 0 {
|
||||
rec.CacheMiss = int64(tut.PromptCacheMiss)
|
||||
}
|
||||
applyStreamUsageDetails(rec, lastUsage)
|
||||
} else if rec.Prompt+rec.Compl > 0 {
|
||||
u := types.TokenUsage{
|
||||
Prompt: int(rec.Prompt),
|
||||
@ -1459,3 +1446,39 @@ func (g *Gateway) handleImage(w http.ResponseWriter, r *http.Request) {
|
||||
Data: resp.ImageData,
|
||||
})
|
||||
}
|
||||
|
||||
// applyStreamUsageDetails mirrors the upstream's reasoning and cache numbers
|
||||
// onto a STREAMED request's audit record.
|
||||
//
|
||||
// It exists as a named function because this was inline in the streaming handler,
|
||||
// where it was untestable — and untested it stayed wrong. The non-streaming
|
||||
// recorder read completion_tokens_details.reasoning_tokens; this one did not. So
|
||||
// a source's streamed requests recorded completion and no reasoning at all while
|
||||
// its non-streamed calls recorded both. Nothing failed and no test was red: the
|
||||
// audit trail was simply quieter than it should have been.
|
||||
//
|
||||
// Found by reading the production audit file after wiring a reasoning-heavy
|
||||
// source: completion=8166 with reasoning absent, on every streamed request.
|
||||
//
|
||||
// Reasoning is a SUBSET of completion, never an addition to it. See
|
||||
// recordChatUsage for why that matters to billing.
|
||||
func applyStreamUsageDetails(rec *Req, u *types.TokenUsage) {
|
||||
if d := u.CompletionTokensDetails; d != nil && d.ReasoningTokens > 0 {
|
||||
rec.Reasoning = int64(d.ReasoningTokens)
|
||||
}
|
||||
// Cache accounting is here for the same reason it is kept apart: "zero hits
|
||||
// because nothing was cached" and "this provider never reports caching" are
|
||||
// the same number and mean opposite things.
|
||||
if d := u.PromptTokensDetails; d != nil {
|
||||
rec.CacheReported = true
|
||||
if d.CachedTokens > 0 {
|
||||
rec.CacheHit = int64(d.CachedTokens)
|
||||
}
|
||||
} else if u.PromptCacheHit > 0 {
|
||||
rec.CacheHit = int64(u.PromptCacheHit)
|
||||
rec.CacheReported = true
|
||||
}
|
||||
if u.PromptCacheMiss > 0 {
|
||||
rec.CacheMiss = int64(u.PromptCacheMiss)
|
||||
}
|
||||
}
|
||||
|
||||
@ -69,3 +69,92 @@ func TestRecordChatUsageReasoningAbsentStaysZero(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamedReasoningIsRecordedOnTheRequest is the streaming counterpart.
|
||||
//
|
||||
// TestRecordChatUsageKeepsReasoningTokens covers the non-streaming path only,
|
||||
// and for a long time the streamed path dropped the number: the non-streaming
|
||||
// recorder read TokenUsage.CompletionTokensDetails, while the streaming one read
|
||||
// only Prompt and Completion from the terminal usage chunk. (Cache accounting was
|
||||
// mirrored; reasoning simply never was.)
|
||||
//
|
||||
// The symptom is invisible in the response — the client sees a correct answer
|
||||
// either way. It only shows in the audit trail, where a reasoning-heavy source's
|
||||
// streamed requests recorded completion=8166 and no reasoning at all. Found by
|
||||
// reading the production audit file, not by a failing test, which is exactly why
|
||||
// the property is pinned here.
|
||||
func TestStreamedReasoningIsRecordedOnTheRequest(t *testing.T) {
|
||||
rec := &Req{}
|
||||
applyStreamUsageDetails(rec, &types.TokenUsage{
|
||||
Prompt: 159579,
|
||||
Completion: 751,
|
||||
Total: 160330,
|
||||
CompletionTokensDetails: &types.CompletionTokensDetails{
|
||||
ReasoningTokens: 512,
|
||||
},
|
||||
})
|
||||
if rec.Reasoning != 512 {
|
||||
t.Errorf("streamed Reasoning = %d, want 512 — a streamed request records no "+
|
||||
"reasoning, so a pricing rule that splits thinking from output silently "+
|
||||
"bills it as ordinary output", rec.Reasoning)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamedReasoningIsNeverAddedToCompletion pins the SUBSET relation in the
|
||||
// streaming path specifically. A presence check would pass if someone wrote
|
||||
// rec.Compl += reasoning, and billing would then charge every thinking token
|
||||
// twice.
|
||||
func TestStreamedReasoningIsNeverAddedToCompletion(t *testing.T) {
|
||||
rec := &Req{}
|
||||
rec.Compl = 751 // set upstream of the call, as the streaming handler does
|
||||
applyStreamUsageDetails(rec, &types.TokenUsage{
|
||||
Completion: 751,
|
||||
CompletionTokensDetails: &types.CompletionTokensDetails{ReasoningTokens: 512},
|
||||
})
|
||||
if rec.Compl != 751 {
|
||||
t.Errorf("Compl = %d, want 751 unchanged — reasoning is a subset, not an addition", rec.Compl)
|
||||
}
|
||||
if rec.Reasoning >= rec.Compl {
|
||||
t.Errorf("Reasoning (%d) must be a SUBSET of Compl (%d)", rec.Reasoning, rec.Compl)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamedUsageWithoutReasoningLeavesItUnset: an upstream that reports no
|
||||
// reasoning must not be turned into a reported zero. In the audit trail 0 means
|
||||
// "not reported", and a plugin cannot tell that from "no reasoning happened".
|
||||
func TestStreamedUsageWithoutReasoningLeavesItUnset(t *testing.T) {
|
||||
rec := &Req{}
|
||||
applyStreamUsageDetails(rec, &types.TokenUsage{Prompt: 10, Completion: 4, Total: 14})
|
||||
if rec.Reasoning != 0 {
|
||||
t.Errorf("Reasoning = %d for an upstream that reported none", rec.Reasoning)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamedCacheAccountingStillWorks is a regression guard for the block the
|
||||
// reasoning lines were added to. They share a function, so a future edit that
|
||||
// short-circuits one would break the other — silently, since a missing cache
|
||||
// number reads as "nothing was cached".
|
||||
func TestStreamedCacheAccountingStillWorks(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
u types.TokenUsage
|
||||
hit int64
|
||||
miss int64
|
||||
rep bool
|
||||
}{
|
||||
{"details object", types.TokenUsage{
|
||||
PromptTokensDetails: &types.PromptTokensDetails{CachedTokens: 4096},
|
||||
}, 4096, 0, true},
|
||||
{"legacy hit/miss", types.TokenUsage{
|
||||
PromptCacheHit: 2048, PromptCacheMiss: 512,
|
||||
}, 2048, 512, true},
|
||||
{"nothing reported", types.TokenUsage{}, 0, 0, false},
|
||||
} {
|
||||
rec := &Req{}
|
||||
applyStreamUsageDetails(rec, &c.u)
|
||||
if rec.CacheHit != c.hit || rec.CacheMiss != c.miss || rec.CacheReported != c.rep {
|
||||
t.Errorf("%s: hit=%d miss=%d reported=%v; want %d/%d/%v",
|
||||
c.name, rec.CacheHit, rec.CacheMiss, rec.CacheReported, c.hit, c.miss, c.rep)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user