feat(config,runtime): 上下文预算按模型窗口推导 + usage/缓存可观测 + sidecar 存全文(T1)
- 窗口声明:models.json 组级 contextWindow + env LLM_CONTEXT_WINDOW(per-group materialize 播种,0 不写盘、负数报错)。 - 预算三态:CONTEXT_MAX_TOKENS 未设置 ⇒ 按窗口推导;>0 显式覆盖;0|负 关闭压缩(不变)。推导 budget = min(窗口×比例, 窗口−输出保留),比例默认 2/3(CONTEXT_BUDGET_RATIO,1M ⇒ 699050 tokens ≈ 683Ki);输出保留 = max(32768, 该组 maxTokens) 且封顶窗口/2(防配置矛盾时退化成 1 token);窗口未声明 ⇒ 回落 131072 + 恰好一条 WARN。接线在组合根 wire_context.go(唯一决策点)。 - usage/缓存可观测(此前**完全不可观测**):Response 增 CacheHit/CacheMiss/Reasoning tokens + CacheHitRate;解析 OpenAI 兼容(prompt_cache_hit_tokens/prompt_cache_miss_tokens/cached_tokens/reasoning_tokens)与 anthropic(cache_read_input_tokens/cache_creation_input_tokens)两种形状;记录点在适配器 record(换入口不丢成本日志),日志无凭据;单次输入 ≥预算 90% 记 WARN(不阻断)。 - 根治清单「续读」假承诺:抽取期不再按注入预算截断(sidecar 存全文,只受 MaxSidecarBytes=8 MiB 安全上限,ExtractorVersion v1→v2 旧 sidecar 失效重抽);注入预算改在装配期生效 ⇒ ref.Chars/Truncated 是注入侧事实、游标真是继续读的起点;document-read 的 limit 上限改为独立具名常量。 - 顺手修越界:config.go 505 行(超 500)拆出 llm.go。 - 主会话独立验收:隔离 worktree(HEAD + 本切片 39 文件,排除在途 gateway)build/vet 绿、全树 59 包 0 FAIL、linux+darwin 交叉编译绿;变异独立复现(比例改 0.25 ⇒ 推导用例红 262144 vs 699050)。
This commit is contained in:
1 parent
5b2b385524
commit
72b95df22b
39 files changed
+2496
-377
No files matched your search
@@ -20,15 +20,26 @@ import (
|
||||
// 抽取器的常量。它们全是编译期常量或纯函数分支,使抽取路径里不出现魔法数字。
|
||||
const (
|
||||
// ExtractorVersion 是抽取器的版本号,写进每条 sidecar 的元数据。抽取逻辑
|
||||
// 一旦变化(清洗规则、摘要口径、预算)必须递增:版本不匹配时旧 sidecar
|
||||
// 一旦变化(清洗规则、摘要口径、截断口径)必须递增:版本不匹配时旧 sidecar
|
||||
// 被判为失效并重抽一次,从而不会让「升级前抽出来的正文」永远留在缓存里
|
||||
// (design/24 §4.3 要求 sidecar 记「抽取器版本」,这就是它的用处)。
|
||||
ExtractorVersion = "documents/v1"
|
||||
//
|
||||
// v2(切片 T1 §7.2):抽取期**不再**按注入预算截断,只在安全上限
|
||||
// (MaxSidecarBytes)处截断 —— 旧缓存里的正文可能少了一大截,必须重抽。
|
||||
ExtractorVersion = "documents/v2"
|
||||
|
||||
// MaxExtractChars 是**单份文档**注入上下文的字符预算(design/24 §3.4、
|
||||
// §4.3 的 64K 字符)。超出即按 rune 截断并置 Result.Truncated;正文绝不
|
||||
// 静默消失 —— 循环据 result 渲染 truncated 与续读游标。
|
||||
MaxExtractChars = 64 * 1024
|
||||
// MaxSidecarBytes 是 sidecar 正文的**安全上限**(字节,8 MiB 文本)。
|
||||
//
|
||||
// 它取代了旧的 MaxExtractChars(64K 字符):那个数曾经同时兼任「抽取预算」与
|
||||
// 「注入预算」,导致 sidecar 里存的就是被注入预算砍过的正文 —— 于是清单行给
|
||||
// 出的续读游标(= 已注入字符数)恒等于 sidecar 正文长度,模型照它去
|
||||
// `document-read` 必然读到空(切片 T1 §7.1 实测的断裂)。
|
||||
//
|
||||
// 现在两件事分开:sidecar 存**抽取得到的全文**,只受这个安全上限约束(防止
|
||||
// 一份畸形或超大文件把内存/磁盘拖垮);每回合注入多少由**装配期**的注入预算
|
||||
// (loop.Options.DocumentMaxInjectChars)决定。命中本上限时 Truncated=true,
|
||||
// 如实记进 sidecar 元数据 —— 「正文到此为止」绝不等于「原文件到此为止」。
|
||||
MaxSidecarBytes = 8 << 20
|
||||
|
||||
// csvSummaryMaxColumns 限制摘要里列出的列名个数:超宽表只列前若干列,
|
||||
// 免得摘要本身挤占注入预算。
|
||||
@@ -72,12 +83,14 @@ var (
|
||||
// Result 是一次抽取的产物。它是抽取器的全部对外事实:正文、字符数、是否
|
||||
// 截断、种类与抽取器版本,sidecar 元数据与注入渲染都只读它。
|
||||
type Result struct {
|
||||
// Text 是要注入模型可见上下文的文本(CSV/TSV 的摘要行 + 正文,已按 rune
|
||||
// 截断到预算内)。
|
||||
// Text 是要落进 sidecar(并在装配期按注入预算取用)的文本(CSV/TSV 的摘要行
|
||||
// + 正文,已按安全上限 MaxSidecarBytes 截断)。
|
||||
Text string
|
||||
// Chars 是 Text 的字符数(rune 数)。
|
||||
Chars int
|
||||
// Truncated 报告正文是否因预算被截断(design/24 §4.3:绝不静默截断)。
|
||||
// Truncated 报告正文是否撞上了**安全上限**(design/24 §4.3:绝不静默截断)。
|
||||
// 它与「注入被截断」是两件事:后者由装配期按注入预算判定并写进
|
||||
// history.DocumentRef(切片 T1 §7.2)。
|
||||
Truncated bool
|
||||
// Kind 是抽取出内容的种类(text / csv / tsv)。
|
||||
Kind string
|
||||
@@ -115,8 +128,8 @@ func Extract(name, mime string, data []byte) (Result, error) {
|
||||
text = summary + "\n" + body
|
||||
}
|
||||
res := Result{Kind: kind, ExtractorVersion: ExtractorVersion}
|
||||
if utf8.RuneCountInString(text) > MaxExtractChars {
|
||||
text = prefixRunes(text, MaxExtractChars)
|
||||
if len(text) > MaxSidecarBytes {
|
||||
text = prefixBytes(text, MaxSidecarBytes)
|
||||
res.Truncated = true
|
||||
}
|
||||
res.Text = text
|
||||
@@ -264,17 +277,19 @@ func splitColumns(line, delim string) []string {
|
||||
return out
|
||||
}
|
||||
|
||||
// prefixRunes 返回 s 的前 n 个 rune:rune 安全,绝不把多字节字符切成两半。
|
||||
func prefixRunes(s string, n int) string {
|
||||
if n <= 0 {
|
||||
// prefixBytes 返回 s 的前若干个**完整 rune**,使其字节数不超过 max:
|
||||
// 安全上限按字节设(MaxSidecarBytes),因此截断必须落在 rune 边界上,
|
||||
// 绝不把多字节字符切成两半(否则 sidecar 正文会是非法 UTF-8)。
|
||||
func prefixBytes(s string, max int) string {
|
||||
if max <= 0 {
|
||||
return ""
|
||||
}
|
||||
count := 0
|
||||
for i := range s {
|
||||
if count == n {
|
||||
return s[:i]
|
||||
}
|
||||
count++
|
||||
if len(s) <= max {
|
||||
return s
|
||||
}
|
||||
return s
|
||||
cut := max
|
||||
for cut > 0 && !utf8.RuneStart(s[cut]) {
|
||||
cut--
|
||||
}
|
||||
return s[:cut]
|
||||
}
|
||||
@@ -88,11 +88,15 @@ func TestExtract_TSVPrependsSummary(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestExtract_TruncatesHonestly 钉住「绝不静默截断」(design/24 §4.3):超预算时
|
||||
// 截到预算、置 Truncated,且截断是 rune 安全的(多字节字符不被切成两半)。
|
||||
func TestExtract_TruncatesHonestly(t *testing.T) {
|
||||
// TestExtract_TruncatesAtSafetyCap 钉住「绝不静默截断」(design/24 §4.3):
|
||||
// 撞上**安全上限**(MaxSidecarBytes,8 MiB 文本)时截到上限并置 Truncated,且截断
|
||||
// 是 rune 安全的(多字节字符不被切成两半)。
|
||||
//
|
||||
// 注意它钉的不再是「注入预算」:注入预算由装配期按上下文预算决定(切片 T1 §7.2),
|
||||
// 抽取器只保证 sidecar 里存得下全文、且撞上上限时如实记账。
|
||||
func TestExtract_TruncatesAtSafetyCap(t *testing.T) {
|
||||
t.Parallel()
|
||||
body := strings.Repeat("你好", MaxExtractChars)
|
||||
body := strings.Repeat("你好", MaxSidecarBytes/len("你好")+16)
|
||||
res, err := Extract(txtName, "text/plain", []byte(body))
|
||||
if err != nil {
|
||||
t.Fatalf("Extract: %v", err)
|
||||
@@ -100,15 +104,37 @@ func TestExtract_TruncatesHonestly(t *testing.T) {
|
||||
if !res.Truncated {
|
||||
t.Fatal("Truncated = false, want true")
|
||||
}
|
||||
if res.Chars != MaxExtractChars {
|
||||
t.Fatalf("Chars = %d, want %d", res.Chars, MaxExtractChars)
|
||||
if len(res.Text) > MaxSidecarBytes {
|
||||
t.Fatalf("截断后正文 %d 字节,超出安全上限 %d", len(res.Text), MaxSidecarBytes)
|
||||
}
|
||||
if want := strings.Repeat("你好", MaxExtractChars/2); res.Text != want {
|
||||
t.Fatalf("Text 不是预算内的 rune 前缀(len=%d)", len(res.Text))
|
||||
if res.Chars != utf8.RuneCountInString(res.Text) {
|
||||
t.Fatalf("Chars = %d, want 正文 rune 数 %d", res.Chars, utf8.RuneCountInString(res.Text))
|
||||
}
|
||||
if !utf8.ValidString(res.Text) {
|
||||
t.Fatal("截断后的正文不是合法 UTF-8")
|
||||
}
|
||||
// 上限内的正文逐字无损:截断只砍尾巴,不改写已有内容。
|
||||
if !strings.HasPrefix(body, res.Text) {
|
||||
t.Fatal("截断后的正文不是原文的前缀")
|
||||
}
|
||||
}
|
||||
|
||||
// TestExtract_KeepsFullTextBelowSafetyCap 钉住切片 T1 §7.2 的核心:抽取期**不再**
|
||||
// 按注入预算截断,因此一份远超旧 64K 字符、但仍在安全上限内的文档必须整份存下 ——
|
||||
// 否则 document-read 永远读不到注入预算之外的正文(S4 实测的断裂)。
|
||||
func TestExtract_KeepsFullTextBelowSafetyCap(t *testing.T) {
|
||||
t.Parallel()
|
||||
body := strings.Repeat("宽", 200*1024) // 20 万字符:远超旧的 64K 注入预算
|
||||
res, err := Extract(txtName, "text/plain", []byte(body))
|
||||
if err != nil {
|
||||
t.Fatalf("Extract: %v", err)
|
||||
}
|
||||
if res.Truncated {
|
||||
t.Fatal("Truncated = true, want false(安全上限内的正文不该被截断)")
|
||||
}
|
||||
if res.Chars != 200*1024 || res.Text != body {
|
||||
t.Fatalf("Chars = %d(want %d),正文是否逐字无损 = %v", res.Chars, 200*1024, res.Text == body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExtract_SanitizesInvalidUTF8 钉住「非法 UTF-8 要处理,不可 panic」:
|
||||
|
||||
@@ -46,10 +46,14 @@ const (
|
||||
sidecarDirMode = 0o700
|
||||
sidecarFileMode = 0o600
|
||||
|
||||
// maxSidecarBytes 是读 sidecar 时的硬上限(正文预算 64K 字符,UTF-8 最多
|
||||
// 4 字节/字符,再留元数据余量)。超过即判为损坏并重抽,绝不把一整份畸形
|
||||
// maxSidecarBytes 是读 sidecar 时的硬上限:正文安全上限(MaxSidecarBytes)
|
||||
// 再加元数据行与 JSON 编码的余量。超过即判为损坏并重抽,绝不把一整份畸形
|
||||
// 文件读进内存。
|
||||
maxSidecarBytes = MaxExtractChars*4 + 1<<10
|
||||
maxSidecarBytes = MaxSidecarBytes + sidecarReadAllowanceBytes
|
||||
|
||||
// sidecarReadAllowanceBytes 是给首行元数据(含 unreadable 的原因短句)与
|
||||
// JSON 编码留的余量。
|
||||
sidecarReadAllowanceBytes = 64 * 1024
|
||||
|
||||
// sidecarTempPattern 是原子写入用的临时文件前缀。
|
||||
sidecarTempPattern = ".sidecar-*"
|
||||
@@ -88,11 +92,12 @@ type cachedExtract struct {
|
||||
Reason string
|
||||
// Kind 是抽取所得内容的种类(text / csv / tsv)。
|
||||
Kind string
|
||||
// Chars 是注入正文的字符数。
|
||||
// Chars 是 sidecar 正文的字符数(= 抽取所得全文的字符数,除非撞上安全上限)。
|
||||
Chars int
|
||||
// Truncated 报告正文是否被预算截断。
|
||||
// Truncated 报告抽取期是否撞上了安全上限 MaxSidecarBytes(「正文到此为止」
|
||||
// 不等于「原文件到此为止」)。
|
||||
Truncated bool
|
||||
// Text 是 Status == extracted 时的注入正文。
|
||||
// Text 是 Status == extracted 时的全文正文。
|
||||
Text string
|
||||
}
|
||||
|
||||
@@ -393,6 +398,11 @@ func (r *Resolver) extract(part model.Content) cachedExtract {
|
||||
// documentRefFrom 把 sidecar 记录折进文档引用:身份(存储名、MIME、字节数)
|
||||
// 来自部件本身,status/chars/truncated 来自抽取结论。空状态归为 listed
|
||||
// (「清单已给出、正文未注入」是事实,而不是「已抽取」)。
|
||||
//
|
||||
// 注意这里给出的 chars/truncated 是**抽取侧**的事实(sidecar 正文的字符数、
|
||||
// 是否撞上安全上限)。装配期(loop.resolveDocuments)会按本回合的注入预算把
|
||||
// 它们改写成「实际注入了多少字符 / 注入是否被截断」—— 清单行的续读游标因此
|
||||
// 真的是「继续读的起点」(切片 T1 §7.2)。
|
||||
func documentRefFrom(rec cachedExtract, ref history.DocumentRef) history.DocumentRef {
|
||||
ref.Status = rec.Status
|
||||
ref.Chars = rec.Chars
|
||||
|
||||
@@ -33,7 +33,9 @@ type Sidecar struct {
|
||||
Kind string
|
||||
// Chars 是抽取期写盘时的正文字符数(rune)。
|
||||
Chars int
|
||||
// Truncated 报告抽取期的正文是否已按预算截断(design/24 §3.4 的 64K 字符)。
|
||||
// Truncated 报告 sidecar 正文在**抽取期**是否撞上了安全上限
|
||||
// (documents.MaxSidecarBytes,8 MiB 文本)—— 它与「本回合注入被截断」是两件事:
|
||||
// 后者由装配期按注入预算判定并写进 history.DocumentRef。
|
||||
Truncated bool
|
||||
// Text 是正文本身:sidecar 第一行元数据**之后**的部分(含 CSV/TSV 的摘要行)。
|
||||
Text string
|
||||
|
||||
@@ -95,27 +95,37 @@ func TestResolver_ExtractsOnceAndThenReadsSidecar(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestResolver_RecordsTruncationInSidecar 钉住截断统计随 sidecar 往返:第二次
|
||||
// 读 sidecar(原文件已删)仍然带着 truncated 与截断后的字符数。
|
||||
func TestResolver_RecordsTruncationInSidecar(t *testing.T) {
|
||||
// TestResolver_KeepsFullBodyAndRecordsSafetyTruncationInSidecar 钉住 sidecar 的
|
||||
// 存全文语义(切片 T1 §7.2 第 1 条)与安全上限的如实记账:
|
||||
//
|
||||
// - 正文远超旧的 64K 注入预算、但未撞上安全上限 ⇒ 整份存下,truncated=false,
|
||||
// 因此 document-read 按清单游标续读能真的读到预算之外的正文;
|
||||
// - 撞上安全上限(MaxSidecarBytes)⇒ truncated=true 且字符数随 sidecar 往返。
|
||||
func TestResolver_KeepsFullBodyAndRecordsSafetyTruncationInSidecar(t *testing.T) {
|
||||
t.Parallel()
|
||||
layout := newTestLayout(t)
|
||||
name := layout.writeUpload(t, txtName, []byte(strings.Repeat("宽", MaxExtractChars+10)))
|
||||
name := layout.writeUpload(t, txtName, []byte(strings.Repeat("宽", 100*1024)))
|
||||
r := NewResolver(Config{DataDir: layout.dataDir, UploadsDir: layout.uploads})
|
||||
|
||||
ref, _ := resolveOne(t, r, name)
|
||||
if !ref.Truncated || ref.Chars != MaxExtractChars {
|
||||
t.Fatalf("首次解析 = %+v, want truncated + Chars=%d", ref, MaxExtractChars)
|
||||
ref, body := resolveOne(t, r, name)
|
||||
if ref.Truncated || ref.Chars != 100*1024 {
|
||||
t.Fatalf("首次解析 = %+v, want 全文(%d 字符、未截断)", ref, 100*1024)
|
||||
}
|
||||
if err := os.Remove(filepath.Join(layout.uploads, name)); err != nil {
|
||||
t.Fatalf("删除原文件: %v", err)
|
||||
}
|
||||
ref2, body2 := resolveOne(t, r, name)
|
||||
if !ref2.Truncated || ref2.Chars != MaxExtractChars {
|
||||
t.Fatalf("sidecar 命中 = %+v, want 保留 truncated + Chars", ref2)
|
||||
if ref2.Truncated || ref2.Chars != 100*1024 || body2 != body {
|
||||
t.Fatalf("sidecar 命中 = %+v, want 从 sidecar 读回同一份全文", ref2)
|
||||
}
|
||||
if len([]rune(body2)) != MaxExtractChars {
|
||||
t.Fatalf("sidecar 正文 rune 数 = %d, want %d", len([]rune(body2)), MaxExtractChars)
|
||||
|
||||
capped := layout.writeUpload(t, "0123456789abcdef0123456789abcdee.txt", []byte(strings.Repeat("宽", MaxSidecarBytes)))
|
||||
ref3, body3 := resolveOne(t, r, capped)
|
||||
if !ref3.Truncated {
|
||||
t.Fatalf("撞上安全上限的文档 = %+v, want truncated=true", ref3)
|
||||
}
|
||||
if ref3.Chars != len([]rune(body3)) || len(body3) > MaxSidecarBytes {
|
||||
t.Fatalf("上限命中:Chars=%d / 正文字节=%d,want 一致且不超过 %d", ref3.Chars, len(body3), MaxSidecarBytes)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -68,34 +68,38 @@ type Compactor interface {
|
||||
Compact(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error)
|
||||
}
|
||||
|
||||
// defaultCompactor 是内置的结构化 compactor。它无状态,
|
||||
// 且并发安全。
|
||||
type defaultCompactor struct{}
|
||||
// defaultCompactor 是内置的结构化 compactor。它无状态(documentInjectChars 只在
|
||||
// 构造时给定),且并发安全。
|
||||
type defaultCompactor struct {
|
||||
// documentInjectChars 是单份附件注入上下文的字符预算,用于估算文档部件的
|
||||
// token 代价(documentPartTokens)。0 表示用 defaultDocumentInjectChars。
|
||||
documentInjectChars int
|
||||
}
|
||||
|
||||
// Compact 实现 Compactor。
|
||||
func (defaultCompactor) Compact(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error) {
|
||||
func (c defaultCompactor) Compact(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error) {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, fmt.Errorf("loop: compact: %w", err)
|
||||
}
|
||||
return compactMessages(ctx, messages, budget)
|
||||
return compactMessages(ctx, messages, budget, c.documentInjectChars)
|
||||
}
|
||||
|
||||
// compactMessages 是共享的压缩实现。它绝不改写
|
||||
// 输入切片;压缩结果始终是新分配的。
|
||||
func compactMessages(ctx context.Context, messages []model.Message, budget int) ([]model.Message, error) {
|
||||
func compactMessages(ctx context.Context, messages []model.Message, budget, docInjectChars int) ([]model.Message, error) {
|
||||
if budget <= 0 || len(messages) < minCompactMessages {
|
||||
return messages, nil
|
||||
}
|
||||
if estimateContextTokens(messages) <= budget {
|
||||
if estimateContextTokensWith(messages, docInjectChars) <= budget {
|
||||
return messages, nil
|
||||
}
|
||||
|
||||
prefixEnd := summaryPrefixEnd(messages)
|
||||
keepTokens := max(int(float64(budget)*keepRecentFraction), minKeepTokens)
|
||||
keepStart := findKeepStart(messages, prefixEnd, keepTokens)
|
||||
keepStart := findKeepStart(messages, prefixEnd, keepTokens, docInjectChars)
|
||||
if keepStart <= prefixEnd {
|
||||
// 全部内容都被保留;没有可安全摘要的前缀。
|
||||
return messages, nil
|
||||
@@ -144,7 +148,7 @@ func summaryPrefixEnd(messages []model.Message) int {
|
||||
// 给 turn 计分,因此最近的 turn 得以存活,尾部的工具结果
|
||||
// 绝不被切开。不存在用户 turn 时(少见),它不额外保留任何东西
|
||||
// 并从 from 返回,因此什么也不被摘要。
|
||||
func findKeepStart(messages []model.Message, from, keepTokens int) int {
|
||||
func findKeepStart(messages []model.Message, from, keepTokens, docInjectChars int) int {
|
||||
n := len(messages)
|
||||
turns := make([]int, 0, 8)
|
||||
for i := from; i < n; i++ {
|
||||
@@ -163,8 +167,9 @@ func findKeepStart(messages []model.Message, from, keepTokens int) int {
|
||||
if i+1 < len(turns) {
|
||||
to = turns[i+1]
|
||||
}
|
||||
if i == len(turns)-1 || acc+estimateContextTokens(messages[turns[i]:to]) <= keepTokens {
|
||||
acc += estimateContextTokens(messages[turns[i]:to])
|
||||
size := estimateContextTokensWith(messages[turns[i]:to], docInjectChars)
|
||||
if i == len(turns)-1 || acc+size <= keepTokens {
|
||||
acc += size
|
||||
keptStart = turns[i]
|
||||
} else {
|
||||
break
|
||||
@@ -173,11 +178,19 @@ func findKeepStart(messages []model.Message, from, keepTokens int) int {
|
||||
return keptStart
|
||||
}
|
||||
|
||||
// estimateContextTokens 返回 messages 的保守 token 估算。它是
|
||||
// 启发式(字符 ÷ charsPerToken,加上每条消息、每个图像与每个文档的开销),
|
||||
// 因此无依赖且离线,但单调性足够驱动
|
||||
// 压缩决策。
|
||||
// estimateContextTokens 返回 messages 的保守 token 估算,文档部件按**缺省**注入
|
||||
// 上限计入(defaultDocumentInjectChars)。需要按本回合真实注入预算估算的调用方
|
||||
// (compactor)请用 estimateContextTokensWith。
|
||||
func estimateContextTokens(messages []model.Message) int {
|
||||
return estimateContextTokensWith(messages, defaultDocumentInjectChars)
|
||||
}
|
||||
|
||||
// estimateContextTokensWith 返回 messages 的保守 token 估算。它是
|
||||
// 启发式(字符 ÷ charsPerToken,加上每条消息、每个图像与每个文档的开销),
|
||||
// 因此无依赖且离线,但单调性足够驱动压缩决策。
|
||||
// docInjectChars 是单份文档注入上下文的字符上限,决定文档部件在估算里最多
|
||||
// 算多少(0 = 缺省值)。
|
||||
func estimateContextTokensWith(messages []model.Message, docInjectChars int) int {
|
||||
total := 0
|
||||
for _, m := range messages {
|
||||
total += messageOverheadTokens
|
||||
@@ -188,7 +201,7 @@ func estimateContextTokens(messages []model.Message) int {
|
||||
case model.ContentTypeImage:
|
||||
total += imageTokens
|
||||
case model.ContentTypeDocument:
|
||||
total += documentPartTokens(part)
|
||||
total += documentPartTokens(part, docInjectChars)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -203,10 +216,13 @@ func estimateContextTokens(messages []model.Message) int {
|
||||
// 被估成 0 token,在压缩里等同于空内容(design/24 §9.5 移交的静默吞掉)。
|
||||
// 大文件(尤其在预算处截断注入、或压根读不到)会被这个估计偏高地计入,方向
|
||||
// 是「少放内容进上下文」——保守收紧,而不是溢出模型窗口。
|
||||
func documentPartTokens(part model.Content) int {
|
||||
func documentPartTokens(part model.Content, docInjectChars int) int {
|
||||
if docInjectChars <= 0 {
|
||||
docInjectChars = defaultDocumentInjectChars
|
||||
}
|
||||
chars := part.DocumentSize
|
||||
if chars > documentInjectedCharsCap {
|
||||
chars = documentInjectedCharsCap
|
||||
if chars > int64(docInjectChars) {
|
||||
chars = int64(docInjectChars)
|
||||
}
|
||||
if chars < 0 {
|
||||
chars = 0
|
||||
@@ -300,9 +316,20 @@ func capSummary(s string, max int) string {
|
||||
|
||||
// truncateRunes 把 s 缩短到至多 n 个 rune,Unicode 安全。
|
||||
func truncateRunes(s string, n int) string {
|
||||
out, _ := truncateRunesAt(s, n)
|
||||
return out
|
||||
}
|
||||
|
||||
// truncateRunesAt 与 truncateRunes 同义,另外报告是否发生了截断。
|
||||
// 「是否截断」是清单行里续读游标的唯一判据(截断才给游标),因此它必须来自
|
||||
// 截断动作本身,而不是「长度是否等于上限」这类再推断。
|
||||
func truncateRunesAt(s string, n int) (string, bool) {
|
||||
if n <= 0 {
|
||||
return "", s != ""
|
||||
}
|
||||
runes := []rune(s)
|
||||
if len(runes) <= n {
|
||||
return s
|
||||
return s, false
|
||||
}
|
||||
return string(runes[:n])
|
||||
return string(runes[:n]), true
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// TestReportContextPressureWarnsAtNinetyPercent 钉住输入逼近预算的 WARN
|
||||
// (design/04 附录 A.3 第 5 条):达到预算 90% 记一条 WARN、低于不记,且**不阻断**
|
||||
// 任何流程(它只是日志)。
|
||||
func TestReportContextPressureWarnsAtNinetyPercent(t *testing.T) {
|
||||
t.Parallel()
|
||||
const budget = 1000
|
||||
cases := []struct {
|
||||
name string
|
||||
promptTokens int
|
||||
wantWarn bool
|
||||
}{
|
||||
{"恰好 90%", 900, true},
|
||||
{"超过 90%", 990, true},
|
||||
{"低于 90%", 899, false},
|
||||
{"提供商未上报用量", 0, false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
log, buf := captureLoopLog()
|
||||
l := New(Options{MaxContextTokens: budget, Log: log})
|
||||
l.reportContextPressure(c.promptTokens)
|
||||
|
||||
got := strings.Contains(buf.String(), "near the context budget")
|
||||
if got != c.wantWarn {
|
||||
t.Fatalf("promptTokens=%d budget=%d: WARN=%v, want %v(日志:%q)",
|
||||
c.promptTokens, budget, got, c.wantWarn, buf.String())
|
||||
}
|
||||
if !c.wantWarn {
|
||||
return
|
||||
}
|
||||
records := loopJSONRecords(t, buf)
|
||||
if len(records) != 1 || records[0]["prompt_tokens"] != float64(c.promptTokens) {
|
||||
t.Fatalf("告警记录 = %+v, want 恰好一条且带上 prompt_tokens=%d", records, c.promptTokens)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestReportContextPressureSilentWhenCompactionOff 钉住「压缩关闭时不吵」:
|
||||
// 关闭压缩(预算 0/负)的部署没有预算可言,绝不能因为预算为 0 而把每条调用都
|
||||
// 判成「超 90%」。
|
||||
func TestReportContextPressureSilentWhenCompactionOff(t *testing.T) {
|
||||
t.Parallel()
|
||||
for _, budget := range []int{0, -1} {
|
||||
log, buf := captureLoopLog()
|
||||
l := New(Options{MaxContextTokens: budget, Log: log})
|
||||
l.reportContextPressure(1 << 20)
|
||||
if strings.Contains(buf.String(), "near the context budget") {
|
||||
t.Fatalf("预算 %d 时不该记 WARN:%q", budget, buf.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamStepCarriesPromptTokensForPressure 钉住接线:流式主路径把提供商上报的
|
||||
// 输入 token 带进 stepOutcome,逼近预算的判定因此真的会在这条路径上发生 ——
|
||||
// 而不是只在 reportContextPressure 这个函数里「理论上可行」。
|
||||
func TestStreamStepCarriesPromptTokensForPressure(t *testing.T) {
|
||||
t.Parallel()
|
||||
log, buf := captureLoopLog()
|
||||
ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Model: "m", Content: "好", PromptTokens: 950}}}}
|
||||
l := New(Options{Agent: newTestAgent(t, ad, nil), MaxContextTokens: 1000, Log: log,
|
||||
OnStream: func(model.StreamEvent) {}})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess-pressure", []model.Message{model.TextMessage(model.RoleUser, "在吗")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if !strings.Contains(buf.String(), "near the context budget") {
|
||||
t.Fatalf("流式路径没有触发预算告警:%q", buf.String())
|
||||
}
|
||||
records := loopJSONRecords(t, buf)
|
||||
found := false
|
||||
for _, rec := range records {
|
||||
if rec["msg"] == "model input is near the context budget" {
|
||||
found = true
|
||||
if rec["prompt_tokens"] != float64(950) || rec["budget"] != float64(1000) {
|
||||
t.Fatalf("告警字段 = %+v, want prompt_tokens=950 / budget=1000", rec)
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("日志里没有预算告警记录:%q", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
// captureLoopLog 返回输出 JSON 记录的 logger 与缓冲:断言结构化字段(而不是文本
|
||||
// 子串)时用它。nil logger 在循环里会被换成丢弃 logger,所以测试必须显式注入。
|
||||
func captureLoopLog() (*slog.Logger, *bytes.Buffer) {
|
||||
buf := &bytes.Buffer{}
|
||||
return slog.New(slog.NewJSONHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})), buf
|
||||
}
|
||||
|
||||
// loopJSONRecords 把缓冲里的 JSON 记录逐条解出来,便于按字段断言。
|
||||
func loopJSONRecords(t *testing.T, buf *bytes.Buffer) []map[string]any {
|
||||
t.Helper()
|
||||
out := []map[string]any{}
|
||||
for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {
|
||||
if strings.TrimSpace(line) == "" {
|
||||
continue
|
||||
}
|
||||
rec := map[string]any{}
|
||||
if err := json.Unmarshal([]byte(line), &rec); err != nil {
|
||||
t.Fatalf("日志行不是合法 JSON: %v(%q)", err, line)
|
||||
}
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
@@ -18,16 +19,24 @@ import (
|
||||
// 2. **清单恒在**:每份文档至少要留下「名、MIME、大小、结论」这一行,
|
||||
// 绝不出现「传上去了但模型一无所知」的空上下文(§4.2-3);
|
||||
// 3. **失败必须可见**:抽取失败解析为诚实的文本标记 + status=unreadable,
|
||||
// 回合照常继续(§4.4)。
|
||||
// 回合照常继续(§4.4);
|
||||
// 4. **注入预算在这里生效**(切片 T1 §7.2):正文按本回合的注入预算
|
||||
// (l.documentInjectChars)截取,ref.Chars = 实际注入的字符数、
|
||||
// ref.Truncated = 注入是否被截断。sidecar 存的是抽取全文,因此清单行给出的
|
||||
// 续读游标从此**真的是**「继续读的起点」,与 document-read 的 offset 对齐。
|
||||
//
|
||||
// 它作用于**装配后的模型可见视图**而不是消息窗口:窗口(checkpoint 快照、
|
||||
// 记忆提取输入)里只留文档引用(§2.4),正文每次装配时按引用读 sidecar 得到。
|
||||
const (
|
||||
// documentInjectedCharsCap 是单份文档注入上下文的字符上限(design/24 §3.4
|
||||
// 的 64K 字符)。它与抽取器的 documents.MaxExtractChars **同值同义**:这里
|
||||
// 不重复实现截断(截断由抽取器完成并写进 sidecar,供后续回合复用),只在
|
||||
// 压缩估算里按同一口径计入文档部件的代价。两个包各自独立,避免反向依赖。
|
||||
documentInjectedCharsCap = 64 * 1024
|
||||
// defaultDocumentInjectChars 是单份文档注入上下文的**字符**上限的缺省值
|
||||
// (64K 字符)。它的实际取值由组合根按推导出的上下文预算给出
|
||||
// (Options.DocumentMaxInjectChars / DocumentInjectCharsFor):64K 是
|
||||
// 「小窗口下不退化」的下限,不再兼任抽取期的截断预算 —— 抽取期只受
|
||||
// documents.MaxSidecarBytes 这个安全上限约束(切片 T1 §7.2)。
|
||||
//
|
||||
// 它同时是压缩估算里文档部件代价的封顶依据(compactor.go 的
|
||||
// documentPartTokens):注入多少,就按多少计入 token。
|
||||
defaultDocumentInjectChars = 64 * 1024
|
||||
|
||||
// documentMinTokens 是清单行(名/MIME/大小/结论)的粗略 token 下限:无论
|
||||
// 文档多小、甚至完全读不到,清单行都进了上下文,因此它不是免费的。
|
||||
@@ -40,8 +49,35 @@ const (
|
||||
// bytesPerKiB / bytesPerMiB 是清单行渲染大小用的单位(仅展示用)。
|
||||
bytesPerKiB = 1024
|
||||
bytesPerMiB = 1024 * bytesPerKiB
|
||||
|
||||
// documentInjectBudgetShare 是单份附件注入正文可占推导预算的份数(1/份数):
|
||||
// 一个回合最多 5 个附件(design/24 §3.4),取 8 意味着「每份最多占预算的
|
||||
// 1/8」—— 5 份合计不超过 5/8,其余留给系统提示词、历史、工具 schema 与本轮
|
||||
// 输出。字符与 token 的换算沿用循环的离线估算口径(charsPerToken)。
|
||||
documentInjectBudgetShare = 8
|
||||
)
|
||||
|
||||
// DocumentInjectCharsFor 由推导出的上下文预算(token)算出**单份附件**注入
|
||||
// 模型可见上下文的字符预算(design/24 §9.6 连带项;切片 T1 §7.2 第 2 条)。
|
||||
//
|
||||
// 这是「注入预算随上下文预算推导」这条要求的落点:1M 窗口 × 2/3 ⇒ 699050 token
|
||||
// 的预算 ⇒ 单份附件可注入 349525 字符,而不是旧的 64K 卡死。取值下限是历史默认
|
||||
// 64K(小窗口下附件行为不退化),并且绝不超过预算本身(预算被显式设得很小时,
|
||||
// 也不该让一份附件就吃掉整个窗口)。
|
||||
func DocumentInjectCharsFor(budgetTokens int) int {
|
||||
if budgetTokens <= 0 {
|
||||
return defaultDocumentInjectChars
|
||||
}
|
||||
chars := budgetTokens * charsPerToken / documentInjectBudgetShare
|
||||
if chars < defaultDocumentInjectChars {
|
||||
chars = defaultDocumentInjectChars
|
||||
}
|
||||
if chars > budgetTokens {
|
||||
chars = budgetTokens
|
||||
}
|
||||
return chars
|
||||
}
|
||||
|
||||
// 清单行与结论文案(design/24 §4.3、§4.4)。清单行恒在,结论文案随结论变化:
|
||||
// 已抽取(可含截断与续读游标)、仅清单、不可读。
|
||||
const (
|
||||
@@ -124,7 +160,7 @@ func (l *Loop) resolveDocumentMessage(ctx context.Context, m model.Message) (mod
|
||||
content = append(content, part)
|
||||
continue
|
||||
}
|
||||
ref, body := documentResolution(part, refs, bodies, failed)
|
||||
ref, body := documentResolution(part, refs, bodies, failed, l.documentInjectChars)
|
||||
content = append(content, model.Content{Type: model.ContentTypeText, Text: documentBlock(part, ref, body)})
|
||||
out = append(out, ref)
|
||||
}
|
||||
@@ -152,7 +188,12 @@ func (l *Loop) documentResolutions(ctx context.Context, m model.Message) (map[st
|
||||
// 状态口径:解析器给出的三态照用;缺失时 —— 解析器整个缺席(功能未接线)算
|
||||
// listed(「只给了清单」是事实),而解析器**报了错**(整条消息解析不了)算
|
||||
// unreadable(作者「读不到正文」比「只给了清单」更准确,也让失败对客户端可见)。
|
||||
func documentResolution(part model.Content, refs map[string]history.DocumentRef, bodies map[string]string, failed bool) (history.DocumentRef, string) {
|
||||
//
|
||||
// 注入预算(injectChars)在这一步生效:已抽取的正文按 rune 截到 injectChars,
|
||||
// ref.Chars 记**实际注入**的字符数、ref.Truncated 记**注入**是否被截断。抽取侧
|
||||
// 自己那套 chars/truncated(sidecar 正文长度、是否撞上安全上限)不是客户端关心
|
||||
// 的东西,读侧另有出口(document-read 的 totalChars / extractTruncated)。
|
||||
func documentResolution(part model.Content, refs map[string]history.DocumentRef, bodies map[string]string, failed bool, injectChars int) (history.DocumentRef, string) {
|
||||
ref := history.DocumentRef{Name: part.DocumentName, MIME: part.DocumentMIME, Size: part.DocumentSize}
|
||||
if got, ok := refs[part.DocumentName]; ok {
|
||||
ref.Status, ref.Chars, ref.Truncated = got.Status, got.Chars, got.Truncated
|
||||
@@ -168,7 +209,13 @@ func documentResolution(part model.Content, refs map[string]history.DocumentRef,
|
||||
if ref.Status == history.DocumentStatusUnreadable && body == "" {
|
||||
body = documentUnreadableBody
|
||||
}
|
||||
return ref, body
|
||||
if ref.Status != history.DocumentStatusExtracted {
|
||||
return ref, body
|
||||
}
|
||||
injected, truncated := truncateRunesAt(body, injectChars)
|
||||
ref.Chars = utf8.RuneCountInString(injected)
|
||||
ref.Truncated = truncated
|
||||
return ref, injected
|
||||
}
|
||||
|
||||
// documentStatusKnown 报告 status 是不是三个已知取值之一。unknown 一律按
|
||||
@@ -204,7 +251,8 @@ func documentBlock(part model.Content, ref history.DocumentRef, body string) str
|
||||
|
||||
// documentStatusNote 渲染清单行里的结论部分。截断时给出**续读游标**(已注入
|
||||
// 的字符数即抽取正文里的偏移量),使模型与后续的 document-read 工具知道从哪
|
||||
// 继续(design/24 §4.3:截断绝不静默)。
|
||||
// 继续(design/24 §4.3:截断绝不静默)。因为 sidecar 存的是抽取全文
|
||||
// (切片 T1 §7.2),这个游标**真的**是「继续读的起点」。
|
||||
func documentStatusNote(ref history.DocumentRef) string {
|
||||
switch ref.Status {
|
||||
case history.DocumentStatusExtracted:
|
||||
|
||||
@@ -59,11 +59,12 @@ func messageTextOf(m model.Message) string {
|
||||
}
|
||||
|
||||
// TestResolveDocuments_InjectsListAndBody 钉住主路径:清单行恒在,正文紧随其后,
|
||||
// 文档部件全部被替换掉(同一位置)。
|
||||
// 文档部件全部被替换掉(同一位置)。Chars 是**实际注入**的字符数(装配期按注入
|
||||
// 预算算出,切片 T1 §7.2),不是解析器自称的那个数。
|
||||
func TestResolveDocuments_InjectsListAndBody(t *testing.T) {
|
||||
t.Parallel()
|
||||
res := &fakeDocumentResolver{
|
||||
refs: map[string]history.DocumentRef{docA: {Status: history.DocumentStatusExtracted, Chars: 5}},
|
||||
refs: map[string]history.DocumentRef{docA: {Status: history.DocumentStatusExtracted, Chars: 4}},
|
||||
bodies: map[string]string{docA: "正文内容"},
|
||||
}
|
||||
l := New(Options{DocumentResolver: res})
|
||||
@@ -83,8 +84,8 @@ func TestResolveDocuments_InjectsListAndBody(t *testing.T) {
|
||||
if !strings.HasPrefix(text, "看看这个") {
|
||||
t.Fatalf("用户原文顺序被改变: %q", text)
|
||||
}
|
||||
if len(refs) != 1 || refs[0].Status != history.DocumentStatusExtracted || refs[0].Chars != 5 {
|
||||
t.Fatalf("结论 = %+v, want extracted/chars=5", refs)
|
||||
if len(refs) != 1 || refs[0].Status != history.DocumentStatusExtracted || refs[0].Chars != 4 {
|
||||
t.Fatalf("结论 = %+v, want extracted/chars=4", refs)
|
||||
}
|
||||
if refs[0].Name != docA || refs[0].MIME != "text/plain" || refs[0].Size != 2048 {
|
||||
t.Fatalf("结论的身份字段 = %+v, want 取自部件", refs[0])
|
||||
@@ -110,21 +111,25 @@ func TestResolveDocuments_KeepsListWithoutResolver(t *testing.T) {
|
||||
}
|
||||
|
||||
// TestResolveDocuments_TruncatedGivesCursor 钉住截断诚实(design/24 §4.3):
|
||||
// 截断时清单行给出续读游标(已注入字符数),绝不静默截断。
|
||||
// 注入被截断时清单行给出续读游标(已注入字符数 = 继续读的起点),绝不静默截断。
|
||||
func TestResolveDocuments_TruncatedGivesCursor(t *testing.T) {
|
||||
t.Parallel()
|
||||
res := &fakeDocumentResolver{
|
||||
refs: map[string]history.DocumentRef{docA: {Status: history.DocumentStatusExtracted, Chars: 700, Truncated: true}},
|
||||
bodies: map[string]string{docA: "前 700 字符"},
|
||||
refs: map[string]history.DocumentRef{docA: {Status: history.DocumentStatusExtracted, Chars: 700}},
|
||||
bodies: map[string]string{docA: strings.Repeat("正", 700)},
|
||||
}
|
||||
l := New(Options{DocumentResolver: res})
|
||||
l := New(Options{DocumentResolver: res, DocumentMaxInjectChars: 100})
|
||||
out, _ := l.resolveDocuments(context.Background(), []model.Message{documentMessage(docA, "text/plain", 99)})
|
||||
text := messageTextOf(out[0])
|
||||
for _, want := range []string{"已截断", "offset=700"} {
|
||||
for _, want := range []string{"已截断", "offset=100"} {
|
||||
if !strings.Contains(text, want) {
|
||||
t.Fatalf("注入块 %q 缺 %q", text, want)
|
||||
}
|
||||
}
|
||||
_, body, _ := strings.Cut(text, "\n")
|
||||
if body != strings.Repeat("正", 100) {
|
||||
t.Fatalf("注入正文不是 100 个字符的 rune 前缀: %q", body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestResolveDocuments_UnreadableIsHonest 钉住失败可见(design/24 §4.4):
|
||||
@@ -279,6 +284,89 @@ func TestUserMessageKey_IncludesDocumentRef(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestResolveDocuments_AppliesInjectionBudget 钉住切片 T1 §7.2 第 2 条:注入预算在
|
||||
// **装配期**生效 —— ref.Chars 是实际注入的字符数、ref.Truncated 是注入是否被截断,
|
||||
// 清单行的续读游标因此是「继续读的起点」(sidecar 存全文,游标之外还有正文)。
|
||||
func TestResolveDocuments_AppliesInjectionBudget(t *testing.T) {
|
||||
t.Parallel()
|
||||
const budget = 10
|
||||
body := "0123456789ABCDEFGHIJ"
|
||||
res := &fakeDocumentResolver{
|
||||
// 解析器给的是抽取侧的事实(sidecar 全文 20 字符、未撞安全上限)。
|
||||
refs: map[string]history.DocumentRef{docA: {Status: history.DocumentStatusExtracted, Chars: len(body)}},
|
||||
bodies: map[string]string{docA: body},
|
||||
}
|
||||
l := New(Options{DocumentResolver: res, DocumentMaxInjectChars: budget})
|
||||
out, refs := l.resolveDocuments(context.Background(), []model.Message{documentMessage(docA, "text/plain", 20)})
|
||||
|
||||
text := messageTextOf(out[0])
|
||||
if !strings.Contains(text, "已截断") || !strings.Contains(text, fmt.Sprintf("offset=%d", budget)) {
|
||||
t.Fatalf("注入块缺截断与续读游标: %q", text)
|
||||
}
|
||||
if !strings.Contains(text, body[:budget]) || strings.Contains(text, body[budget:]) {
|
||||
t.Fatalf("注入正文不是预算内的前缀: %q", text)
|
||||
}
|
||||
if len(refs) != 1 || refs[0].Chars != budget || !refs[0].Truncated {
|
||||
t.Fatalf("结论 = %+v, want chars=%d + truncated(注入侧事实)", refs, budget)
|
||||
}
|
||||
|
||||
// 正文在预算内时:不截断、不给游标,Chars 仍是实际注入的字符数。
|
||||
small := &fakeDocumentResolver{
|
||||
refs: map[string]history.DocumentRef{docA: {Status: history.DocumentStatusExtracted, Chars: 3}},
|
||||
bodies: map[string]string{docA: "短正文"},
|
||||
}
|
||||
l2 := New(Options{DocumentResolver: small, DocumentMaxInjectChars: budget})
|
||||
out2, refs2 := l2.resolveDocuments(context.Background(), []model.Message{documentMessage(docA, "text/plain", 9)})
|
||||
if text2 := messageTextOf(out2[0]); strings.Contains(text2, "已截断") {
|
||||
t.Fatalf("预算内却被标成截断: %q", text2)
|
||||
}
|
||||
if len(refs2) != 1 || refs2[0].Chars != 3 || refs2[0].Truncated {
|
||||
t.Fatalf("结论 = %+v, want chars=3 + 未截断", refs2)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUnexportedInjectBudgetDefaultsOnConstruction 钉住零值语义:Options 不设
|
||||
// DocumentMaxInjectChars 时用缺省 64K 字符(小窗口部署行为不退化),而不是 0
|
||||
// (0 会把正文整个截掉)。
|
||||
func TestUnexportedInjectBudgetDefaultsOnConstruction(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := New(Options{}).documentInjectChars; got != defaultDocumentInjectChars {
|
||||
t.Fatalf("缺省注入预算 = %d, want %d", got, defaultDocumentInjectChars)
|
||||
}
|
||||
if got := New(Options{DocumentMaxInjectChars: -5}).documentInjectChars; got != defaultDocumentInjectChars {
|
||||
t.Fatalf("非正注入预算 = %d, want 回落 %d", got, defaultDocumentInjectChars)
|
||||
}
|
||||
if got := New(Options{DocumentMaxInjectChars: 1234}).documentInjectChars; got != 1234 {
|
||||
t.Fatalf("显式注入预算 = %d, want 1234", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDocumentInjectCharsFor 钉住「注入预算随上下文预算推导」这条口径(表驱动):
|
||||
// 1M 窗口推导出的 699050 token ⇒ 单份附件可注入 300K+ 字符(远超旧的 64K),
|
||||
// 但小预算下不跌破历史默认,且绝不超过预算本身。
|
||||
func TestDocumentInjectCharsFor(t *testing.T) {
|
||||
t.Parallel()
|
||||
cases := []struct {
|
||||
name string
|
||||
budget int
|
||||
want int
|
||||
}{
|
||||
{"1M 窗口推导出的预算", 699050, 699050 * charsPerToken / documentInjectBudgetShare},
|
||||
{"fallback 窗口推导出的预算(推导值跌破历史默认 ⇒ 取默认)", 87381, defaultDocumentInjectChars},
|
||||
{"小预算不超过预算本身", 32768, 32768},
|
||||
{"预算关闭", 0, defaultDocumentInjectChars},
|
||||
{"负预算", -1, defaultDocumentInjectChars},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := DocumentInjectCharsFor(c.budget); got != c.want {
|
||||
t.Fatalf("%s: DocumentInjectCharsFor(%d) = %d, want %d", c.name, c.budget, got, c.want)
|
||||
}
|
||||
}
|
||||
if big := DocumentInjectCharsFor(699050); big <= defaultDocumentInjectChars {
|
||||
t.Fatalf("大窗口下的注入预算 = %d, want > 旧的 64K 卡死值", big)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEstimateContextTokens_CountsDocuments 是 design/24 §9.5 移交的回归:文档
|
||||
// 部件必须在 token 估算里占分量,否则纯文档回合被估成 0 token,在压缩里等同于
|
||||
// 空内容。
|
||||
@@ -292,12 +380,21 @@ func TestEstimateContextTokens_CountsDocuments(t *testing.T) {
|
||||
if estimateContextTokens(big) <= estimateContextTokens(small) {
|
||||
t.Fatalf("大附件估算未变大:%d vs %d", estimateContextTokens(big), estimateContextTokens(small))
|
||||
}
|
||||
if got := documentPartTokens(model.Content{Type: model.ContentTypeDocument}); got != documentMinTokens {
|
||||
if got := documentPartTokens(model.Content{Type: model.ContentTypeDocument}, 0); got != documentMinTokens {
|
||||
t.Fatalf("documentPartTokens(未知字节数) = %d, want %d", got, documentMinTokens)
|
||||
}
|
||||
if got := documentPartTokens(model.Content{Type: model.ContentTypeDocument, DocumentSize: documentInjectedCharsCap * 4}) /
|
||||
1; got != documentInjectedCharsCap/charsPerToken {
|
||||
t.Fatalf("documentPartTokens(超大附件) = %d, want 按注入预算封顶 %d", got, documentInjectedCharsCap/charsPerToken)
|
||||
if got := documentPartTokens(model.Content{Type: model.ContentTypeDocument, DocumentSize: defaultDocumentInjectChars * 4},
|
||||
defaultDocumentInjectChars); got != defaultDocumentInjectChars/charsPerToken {
|
||||
t.Fatalf("documentPartTokens(超大附件) = %d, want 按注入预算封顶 %d", got, defaultDocumentInjectChars/charsPerToken)
|
||||
}
|
||||
// 注入预算变大(1M 窗口部署)时,同一个文档部件必须按新预算计入 —— 否则
|
||||
// 注入 683K 字符的附件在估算里只按 64K 计,压缩永远迟一步。
|
||||
bigger := DocumentInjectCharsFor(699050)
|
||||
if bigger <= defaultDocumentInjectChars {
|
||||
t.Fatalf("DocumentInjectCharsFor(1M 窗口推导出的预算) = %d, want > %d", bigger, defaultDocumentInjectChars)
|
||||
}
|
||||
if got, want := documentPartTokens(model.Content{Type: model.ContentTypeDocument, DocumentSize: int64(bigger)}, bigger), bigger/charsPerToken; got != want {
|
||||
t.Fatalf("按更大注入预算的估算 = %d, want %d", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -132,6 +132,7 @@ type Loop struct {
|
||||
transformContext func(context.Context, []model.Message, int) []model.Message
|
||||
imageResolver ImageResolver
|
||||
documentResolver DocumentResolver
|
||||
documentInjectChars int
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
@@ -157,6 +158,16 @@ func New(opts Options) *Loop {
|
||||
if compactor == nil {
|
||||
compactor = defaultCompactor{}
|
||||
}
|
||||
docInjectChars := opts.DocumentMaxInjectChars
|
||||
if docInjectChars <= 0 {
|
||||
docInjectChars = defaultDocumentInjectChars
|
||||
}
|
||||
if d, ok := compactor.(defaultCompactor); ok {
|
||||
// 内置 compactor 的文档部件估值必须与注入预算同口径:否则注入 683K 字符的
|
||||
// 附件在压缩估算里只按 64K 计,压缩永远迟一步(切片 T1 §7.2)。
|
||||
d.documentInjectChars = docInjectChars
|
||||
compactor = d
|
||||
}
|
||||
timeLocation, timeZoneName := loadTimezone(opts.Timezone, sourcedLog(opts.Log))
|
||||
return &Loop{
|
||||
agent: opts.Agent,
|
||||
@@ -194,6 +205,7 @@ func New(opts Options) *Loop {
|
||||
transformContext: opts.TransformContext,
|
||||
imageResolver: opts.Vision,
|
||||
documentResolver: opts.DocumentResolver,
|
||||
documentInjectChars: docInjectChars,
|
||||
log: sourcedLog(opts.Log),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -199,6 +199,15 @@ type Options struct {
|
||||
// 只是没有任何正文被注入 —— 因此即使忘了接线,也不会出现「文档部件抵达
|
||||
// 编码器」这种 400,更不会出现空上下文。
|
||||
DocumentResolver DocumentResolver
|
||||
// DocumentMaxInjectChars 是**单份附件**注入模型可见上下文的字符上限
|
||||
// (切片 T1 §7.2)。0 选用 defaultDocumentInjectChars(64K 字符);组合根按
|
||||
// 推导出的上下文预算给出实际值(DocumentInjectCharsFor),使大窗口部署真的
|
||||
// 能吃到大附件。
|
||||
//
|
||||
// 两个后果:① 装配期按它截取注入正文,清单行的续读游标因此**真的是**继续读
|
||||
// 的起点(sidecar 存的是抽取全文);② 压缩的 token 估算按它给文档部件封顶
|
||||
// (注入多少就按多少计入),文档回合不会被估成「免费」。
|
||||
DocumentMaxInjectChars int
|
||||
// TransformContext 非 nil 时,在每个模型步之前立即被调用,
|
||||
// 传入组装好的消息列表(记忆注入与任何压缩
|
||||
// 之后)以及当前轮次号(从 1 开始)。它可以投影、过滤
|
||||
|
||||
@@ -25,6 +25,28 @@ type stepOutcome struct {
|
||||
//(该路径自行构建请求),在 agent 路径上为空。它
|
||||
// 记录在 llm_call span 的日志事件上。
|
||||
model string
|
||||
// promptTokens 是本次调用的输入侧 token 数(提供商上报时,否则 0)。它只用于
|
||||
// 「输入逼近上下文预算」的 WARN(design/04 附录 A.3 第 5 条,不阻断任何流程)。
|
||||
promptTokens int
|
||||
}
|
||||
|
||||
// contextPressureRatio 是「单次输入逼近上下文预算」的告警线:达到预算的 90%
|
||||
// 就记一条 WARN,**不阻断**(design/04 附录 A.3 第 5 条)。它只是让「快溢出了」
|
||||
// 这件事在日志里可见,压缩仍按自己的预算独立判定。
|
||||
const contextPressureRatio = 0.9
|
||||
|
||||
// reportContextPressure 在本次调用的输入 token 达到预算的 contextPressureRatio
|
||||
// 时记一条 WARN。预算关闭(<= 0)或提供商没上报用量时不记 —— 没有依据就不吵。
|
||||
func (l *Loop) reportContextPressure(promptTokens int) {
|
||||
if l.maxContextTokens <= 0 || promptTokens <= 0 {
|
||||
return
|
||||
}
|
||||
if float64(promptTokens) < float64(l.maxContextTokens)*contextPressureRatio {
|
||||
return
|
||||
}
|
||||
l.log.Warn("model input is near the context budget",
|
||||
"prompt_tokens", promptTokens, "budget", l.maxContextTokens,
|
||||
"ratio", contextPressureRatio, "action", "compaction triggers at the budget")
|
||||
}
|
||||
|
||||
// step 运行单次模型调用。配置了 OnStream 回调时,它
|
||||
@@ -57,6 +79,9 @@ func (l *Loop) step(ctx context.Context, sessionID string, messages []model.Mess
|
||||
l.spanProp(llmSpan, "model_stop_reason", out.modelStopReason)
|
||||
}
|
||||
l.spanProp(llmSpan, "tool_calls", len(out.toolCalls))
|
||||
// 输入逼近预算的 WARN(不阻断):覆盖流式路径(agent 路径不经本函数的上报,
|
||||
// promptTokens 为 0 时静默跳过)。
|
||||
l.reportContextPressure(out.promptTokens)
|
||||
l.recordLLMDone(llmSpan, out, l.agentFor(ctx))
|
||||
l.finishSpan(llmSpan, trace.SpanStatusOK)
|
||||
return out, nil
|
||||
|
||||
@@ -89,10 +89,10 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO
|
||||
|
||||
if calls, ok := decodeToolCalls(resp); ok {
|
||||
return &stepOutcome{final: false, toolCalls: calls, stopReason: agent.StopReasonToolCalls,
|
||||
modelStopReason: resp.StopReason, model: resp.Model}, nil
|
||||
modelStopReason: resp.StopReason, model: resp.Model, promptTokens: resp.PromptTokens}, nil
|
||||
}
|
||||
return &stepOutcome{final: true, text: resp.Content, stopReason: agent.StopReasonStop,
|
||||
modelStopReason: resp.StopReason, model: resp.Model}, nil
|
||||
modelStopReason: resp.StopReason, model: resp.Model, promptTokens: resp.PromptTokens}, nil
|
||||
}
|
||||
|
||||
// checkOutputAsync 在后台运行输出侧护栏检查(裁决 13.4.1:真并发、不阻塞
|
||||
|
||||
@@ -97,10 +97,29 @@ type anthropicBlock struct {
|
||||
}
|
||||
|
||||
// anthropicUsage 上报 token 消耗。Anthropic 上报的是 input_tokens
|
||||
// 与 output_tokens,而不是单一总量。
|
||||
// 与 output_tokens,而不是单一总量;提示缓存另有两个字段
|
||||
// (cache_creation_input_tokens = 本次写入缓存的输入 token,
|
||||
// cache_read_input_tokens = 本次从缓存读到的输入 token)。它们全都容忍缺失:
|
||||
// 未启用缓存的端点只会给前两个。
|
||||
type anthropicUsage struct {
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
|
||||
CacheReadInputTokens int `json:"cache_read_input_tokens"`
|
||||
}
|
||||
|
||||
// applyAnthropicUsage 把一个 usage 对象折进 Response(只填用量字段,
|
||||
// input_tokens 与 output_tokens 的既有口径不变)。缓存两档的对应关系:
|
||||
// 命中 = cache_read_input_tokens;未命中 = input_tokens + cache_creation_input_tokens
|
||||
// —— Anthropic 的 input_tokens **不含**缓存读到的部分,若只取它,命中率会被算成 0。
|
||||
func applyAnthropicUsage(out *Response, u *anthropicUsage) {
|
||||
if u == nil {
|
||||
return
|
||||
}
|
||||
out.PromptTokens = u.InputTokens
|
||||
out.CompletionTokens = u.OutputTokens
|
||||
out.CacheHitTokens = u.CacheReadInputTokens
|
||||
out.CacheMissTokens = u.InputTokens + u.CacheCreationInputTokens
|
||||
}
|
||||
|
||||
// AnthropicChatAdapter 调用 Anthropic Messages(/v1/messages)端点。
|
||||
@@ -206,7 +225,7 @@ func (a *AnthropicChatAdapter) Chat(ctx context.Context, req Request, opts Optio
|
||||
if err != nil {
|
||||
return Response{}, err
|
||||
}
|
||||
a.record(req, start, out.TotalTokens)
|
||||
a.record(req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -263,7 +282,7 @@ func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onEv
|
||||
return Response{}, err
|
||||
}
|
||||
out.Model = model
|
||||
a.record(req, start, out.TotalTokens)
|
||||
a.record(req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -381,14 +400,17 @@ func (a *AnthropicChatAdapter) upstreamError(resp *http.Response) error {
|
||||
return fmt.Errorf("model: %w: status %d: %s", ErrUpstream, resp.StatusCode, strings.TrimSpace(string(snippet)))
|
||||
}
|
||||
|
||||
func (a *AnthropicChatAdapter) record(req Request, start time.Time, tokens int) {
|
||||
// record 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
|
||||
// 所有调用点都覆盖 —— 见 usage.go),再喂给可选的 CallRecord 计账出口。
|
||||
func (a *AnthropicChatAdapter) record(req Request, start time.Time, resp Response) {
|
||||
logUsage(a.log, a.name, req, resp, time.Since(start))
|
||||
if a.recorder == nil {
|
||||
return
|
||||
}
|
||||
a.recorder(CallRecord{
|
||||
Model: req.Model,
|
||||
Purpose: req.Purpose,
|
||||
Tokens: tokens,
|
||||
Tokens: resp.TotalTokens,
|
||||
TookMs: time.Since(start).Milliseconds(),
|
||||
})
|
||||
}
|
||||
@@ -413,10 +435,9 @@ func responseFromAnthropicWire(wire *anthropicResponse) Response {
|
||||
out.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: out.Content}}}
|
||||
out.ToolCalls = toolCalls
|
||||
out.StopReason = wire.StopReason
|
||||
if u := wire.Usage; u != nil {
|
||||
out.PromptTokens = u.InputTokens
|
||||
out.CompletionTokens = u.OutputTokens
|
||||
out.TotalTokens = u.InputTokens + u.OutputTokens
|
||||
applyAnthropicUsage(&out, wire.Usage)
|
||||
if wire.Usage != nil {
|
||||
out.TotalTokens = wire.Usage.InputTokens + wire.Usage.OutputTokens
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -114,9 +114,10 @@ func applyAnthropicEvent(
|
||||
case "message_start":
|
||||
if evt.Message != nil {
|
||||
agg.Model = evt.Message.Model
|
||||
if evt.Message.Usage != nil {
|
||||
agg.PromptTokens = evt.Message.Usage.InputTokens
|
||||
}
|
||||
// 输入侧用量(含缓存读/写两档)在 message_start 就齐了,
|
||||
// 因此在这里一次折进去 —— 它与 message_delta 的输出侧用量
|
||||
// 合起来构成这次调用的完整成本面。
|
||||
applyAnthropicUsage(agg, evt.Message.Usage)
|
||||
}
|
||||
case "content_block_start":
|
||||
if evt.Index == nil || evt.Content == nil {
|
||||
|
||||
@@ -267,6 +267,33 @@ type Response struct {
|
||||
PromptTokens int
|
||||
CompletionTokens int
|
||||
TotalTokens int
|
||||
// CacheHitTokens / CacheMissTokens 是**输入侧**的两档 token 数(design/04
|
||||
// 附录 A.1:提供商按缓存命中/未命中两档计价,命中价约为未命中的 1/50)。
|
||||
// 两种线格式都解析进这里:
|
||||
//
|
||||
// - OpenAI 兼容:usage.prompt_cache_hit_tokens / prompt_cache_miss_tokens
|
||||
// (DeepSeek),缺失时用 prompt_tokens_details.cached_tokens 作命中数;
|
||||
// - Anthropic:usage.cache_read_input_tokens 作命中数,
|
||||
// input_tokens + cache_creation_input_tokens 作未命中数。
|
||||
//
|
||||
// 提供商未上报(缺字段)时为零 —— 缺字段是**容错**路径,绝不 panic、绝不
|
||||
// 影响正文解析。
|
||||
CacheHitTokens int
|
||||
CacheMissTokens int
|
||||
// ReasoningTokens 是完成侧里属于**思维链**的 token 数(OpenAI 兼容的
|
||||
// usage.completion_tokens_details.reasoning_tokens)。Anthropic 不上报它
|
||||
// (思考 token 已计入 output_tokens),因此那里恒为 0。
|
||||
ReasoningTokens int
|
||||
}
|
||||
|
||||
// CacheHitRate 返回输入侧的缓存命中率(hit / (hit+miss))。提供商未上报任何
|
||||
// 输入侧明细、或两档都是 0 时返回 0 —— 不编造一个 100% 或 0% 之外的假值。
|
||||
func (r Response) CacheHitRate() float64 {
|
||||
total := r.CacheHitTokens + r.CacheMissTokens
|
||||
if total <= 0 {
|
||||
return 0
|
||||
}
|
||||
return float64(r.CacheHitTokens) / float64(total)
|
||||
}
|
||||
|
||||
// CallRecord 描述一次成功完成的 LLM 调用,供 M9
|
||||
|
||||
@@ -143,7 +143,7 @@ func (a *OpenAIChatAdapter) Chat(ctx context.Context, req Request, opts Options)
|
||||
return Response{}, fmt.Errorf("model: %w: decode response: %v", ErrUpstream, err)
|
||||
}
|
||||
out := responseFromWire(&wire)
|
||||
a.record(req, start, out.TotalTokens)
|
||||
a.record(req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -180,7 +180,7 @@ func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onEvent
|
||||
return Response{}, err
|
||||
}
|
||||
out.Model = model
|
||||
a.record(req, start, out.TotalTokens)
|
||||
a.record(req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -230,14 +230,17 @@ func (a *OpenAIChatAdapter) upstreamError(resp *http.Response) error {
|
||||
return fmt.Errorf("model: %w: status %d: %s", ErrUpstream, resp.StatusCode, strings.TrimSpace(string(snippet)))
|
||||
}
|
||||
|
||||
func (a *OpenAIChatAdapter) record(req Request, start time.Time, tokens int) {
|
||||
// record 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
|
||||
// 所有调用点都覆盖 —— 见 usage.go),再喂给可选的 CallRecord 计账出口。
|
||||
func (a *OpenAIChatAdapter) record(req Request, start time.Time, resp Response) {
|
||||
logUsage(a.log, a.name, req, resp, time.Since(start))
|
||||
if a.recorder == nil {
|
||||
return
|
||||
}
|
||||
a.recorder(CallRecord{
|
||||
Model: req.Model,
|
||||
Purpose: req.Purpose,
|
||||
Tokens: tokens,
|
||||
Tokens: resp.TotalTokens,
|
||||
TookMs: time.Since(start).Milliseconds(),
|
||||
})
|
||||
}
|
||||
@@ -48,11 +48,53 @@ type openAIToolCallDelta struct {
|
||||
Function openAIToolCallFn `json:"function"`
|
||||
}
|
||||
|
||||
// openAIUsage 上报 token 消耗。
|
||||
// openAIUsage 上报 token 消耗。缓存两档(prompt_cache_hit_tokens /
|
||||
// prompt_cache_miss_tokens)由 DeepSeek 上报;OpenAI 兼容家族里另有
|
||||
// prompt_tokens_details.cached_tokens 表达同一件事,两者都解析、缺一个就用
|
||||
// 另一个(见 applyOpenAIUsage)。
|
||||
type openAIUsage struct {
|
||||
PromptTokens int `json:"prompt_tokens"`
|
||||
CompletionTokens int `json:"completion_tokens"`
|
||||
TotalTokens int `json:"total_tokens"`
|
||||
PromptTokens int `json:"prompt_tokens"`
|
||||
CompletionTokens int `json:"completion_tokens"`
|
||||
TotalTokens int `json:"total_tokens"`
|
||||
PromptCacheHitTokens int `json:"prompt_cache_hit_tokens"`
|
||||
PromptCacheMissTokens int `json:"prompt_cache_miss_tokens"`
|
||||
PromptTokensDetails openAIPromptTokenDetails `json:"prompt_tokens_details"`
|
||||
CompletionTokenDetails openAICompletionTokenDetails `json:"completion_tokens_details"`
|
||||
}
|
||||
|
||||
// openAIPromptTokenDetails 是输入侧明细(OpenAI / DashScope 形状)。
|
||||
type openAIPromptTokenDetails struct {
|
||||
// CachedTokens 是命中上下文缓存的 token 数,与 prompt_cache_hit_tokens 同义。
|
||||
CachedTokens int `json:"cached_tokens"`
|
||||
}
|
||||
|
||||
// openAICompletionTokenDetails 是完成侧明细:推理模型把思维链 token 报在这里。
|
||||
type openAICompletionTokenDetails struct {
|
||||
// ReasoningTokens 是思维链 token 数(design/04 附录 A.3 第 4 条要观测它)。
|
||||
ReasoningTokens int `json:"reasoning_tokens"`
|
||||
}
|
||||
|
||||
// applyOpenAIUsage 把一个 usage 对象折进 Response。它容忍任何字段缺失:
|
||||
// 缺失即零值,绝不 panic,也绝不因为用量字段的形态问题影响正文解析。
|
||||
//
|
||||
// 未命中的补齐用的是提供商自己的不变式「prompt_tokens = hit + miss」(见
|
||||
// DeepSeek 文档)—— 只在提供商没给 miss 时才补,免得把命中率算成 100%。
|
||||
func applyOpenAIUsage(out *Response, u *openAIUsage) {
|
||||
if u == nil {
|
||||
return
|
||||
}
|
||||
out.PromptTokens = u.PromptTokens
|
||||
out.CompletionTokens = u.CompletionTokens
|
||||
out.TotalTokens = u.TotalTokens
|
||||
out.CacheHitTokens = u.PromptCacheHitTokens
|
||||
if out.CacheHitTokens == 0 {
|
||||
out.CacheHitTokens = u.PromptTokensDetails.CachedTokens
|
||||
}
|
||||
out.CacheMissTokens = u.PromptCacheMissTokens
|
||||
if out.CacheMissTokens == 0 && out.PromptTokens > out.CacheHitTokens {
|
||||
out.CacheMissTokens = out.PromptTokens - out.CacheHitTokens
|
||||
}
|
||||
out.ReasoningTokens = u.CompletionTokenDetails.ReasoningTokens
|
||||
}
|
||||
|
||||
func responseFromWire(wire *openAIChatResponse) Response {
|
||||
@@ -70,10 +112,6 @@ func responseFromWire(wire *openAIChatResponse) Response {
|
||||
out.ToolCalls = append(out.ToolCalls, ToolCall{Name: tc.Function.Name, ID: tc.ID, Args: args})
|
||||
}
|
||||
}
|
||||
if u := wire.Usage; u != nil {
|
||||
out.PromptTokens = u.PromptTokens
|
||||
out.CompletionTokens = u.CompletionTokens
|
||||
out.TotalTokens = u.TotalTokens
|
||||
}
|
||||
applyOpenAIUsage(&out, wire.Usage)
|
||||
return out
|
||||
}
|
||||
@@ -88,9 +88,9 @@ func (st *openAIStreamState) emit(ev StreamEvent) {
|
||||
// 片段增量(若有)则被路由到其处理器。
|
||||
func (st *openAIStreamState) applyChunk(chunk openAIChatResponse) {
|
||||
if u := chunk.Usage; u != nil {
|
||||
st.agg.PromptTokens = u.PromptTokens
|
||||
st.agg.CompletionTokens = u.CompletionTokens
|
||||
st.agg.TotalTokens = u.TotalTokens
|
||||
// 用量块(含缓存的 hit/miss 与思维链 token)只在启用 include_usage 时
|
||||
// 随末块到达,且必须原样折进聚合结果 —— 这正是「成本可观测」的落点。
|
||||
applyOpenAIUsage(&st.agg, u)
|
||||
}
|
||||
if len(chunk.Choices) == 0 {
|
||||
return
|
||||
|
||||
@@ -170,8 +170,10 @@ func TestOpenAIChat_ReasoningEffortPassthrough(t *testing.T) {
|
||||
if got := (<-ch).body["reasoning_effort"]; got != "high" {
|
||||
t.Fatalf("body.reasoning_effort = %v, want high", got)
|
||||
}
|
||||
if buf.Len() != 0 {
|
||||
t.Errorf("unexpected log for a supported value: %q", buf.String())
|
||||
// 用量日志(每次调用一条 Info)不算「意外日志」:这里断言的是**没有 WARN**
|
||||
// —— 受支持的档位不该被上报成问题。
|
||||
if strings.Contains(buf.String(), "level=WARN") {
|
||||
t.Errorf("unexpected WARN for a supported value: %q", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 本文件是**用量与缓存的可观测出口**(design/04 附录 A.3 第 4 条)。
|
||||
//
|
||||
// 背景(附录 A.2 第 3 条):在此之前输入侧用量从未被记录 —— 全仓只消费了
|
||||
// PromptTokens / TotalTokens,`prompt_cache_hit_tokens` / `miss` 一律没解析、没落
|
||||
// 日志,于是「大上下文到底贵不贵」根本无法实测。而提供商按命中/未命中两档计价,
|
||||
// 命中价约为未命中的 1/50:看不见命中率,就等于看不见成本。
|
||||
//
|
||||
// 记录点选在**适配器内部**(每个适配器的 record 都调它一次),而不是循环里:
|
||||
// 这样流式、非流式、后台思考、子代理、管理面板试连等**所有**模型调用点都覆盖到,
|
||||
// 不会有「换个调用入口就没了成本日志」的缺口。
|
||||
//
|
||||
// 红线:日志里只出现模型名、用途、token 数与耗时 —— **凭据(apiKey / token)绝不
|
||||
// 出现在这里**,它连参数都进不来。
|
||||
|
||||
// usageLogEvent 是每条用量日志的事件名,便于按事件过滤与统计。
|
||||
const usageLogEvent = "model usage"
|
||||
|
||||
// logUsage 记一条结构化用量日志:prompt / completion / total / reasoning /
|
||||
// cacheHit / cacheMiss / hitRate,外加模型名、用途、适配器与耗时。
|
||||
//
|
||||
// log 可以为 nil(适配器在未接日志时用的是丢弃 logger),这里仍做一次防御:
|
||||
// 缺字段容错、绝不 panic。
|
||||
func logUsage(log *slog.Logger, adapter string, req Request, resp Response, took time.Duration) {
|
||||
if log == nil {
|
||||
return
|
||||
}
|
||||
log.Info(usageLogEvent,
|
||||
"adapter", adapter,
|
||||
"model", resp.Model,
|
||||
"purpose", req.Purpose,
|
||||
"prompt", resp.PromptTokens,
|
||||
"completion", resp.CompletionTokens,
|
||||
"total", resp.TotalTokens,
|
||||
"reasoning", resp.ReasoningTokens,
|
||||
"cacheHit", resp.CacheHitTokens,
|
||||
"cacheMiss", resp.CacheMissTokens,
|
||||
"hitRate", resp.CacheHitRate(),
|
||||
"tookMs", took.Milliseconds(),
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,247 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
)
|
||||
|
||||
// 本文件钉住 usage/缓存可观测(design/04 附录 A.3 第 4 条):
|
||||
// - 两种线格式的缓存字段都解析进 Response,**缺字段是容错路径**(不 panic、不影响正文);
|
||||
// - 每次模型调用后记一条结构化用量日志,字段齐全且**不含任何凭据**。
|
||||
|
||||
// TestOpenAIUsage_ParsesCacheAndReasoningTokens 钉住 OpenAI 兼容形状:DeepSeek 式的
|
||||
// prompt_cache_hit_tokens / prompt_cache_miss_tokens 与 completion_tokens_details
|
||||
// .reasoning_tokens 都进 Response。
|
||||
func TestOpenAIUsage_ParsesCacheAndReasoningTokens(t *testing.T) {
|
||||
t.Parallel()
|
||||
body := `{
|
||||
"id": "cmpl-1",
|
||||
"model": "deepseek-chat",
|
||||
"choices": [{"index":0,"message":{"role":"assistant","content":"你好"},"finish_reason":"stop"}],
|
||||
"usage": {
|
||||
"prompt_tokens": 1000, "completion_tokens": 200, "total_tokens": 1200,
|
||||
"prompt_cache_hit_tokens": 900, "prompt_cache_miss_tokens": 100,
|
||||
"completion_tokens_details": {"reasoning_tokens": 150}
|
||||
}
|
||||
}`
|
||||
srv := newCompletionServer(t, http.StatusOK, "application/json", body, nil)
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "deepseek-chat"})
|
||||
|
||||
resp, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, Options{})
|
||||
if err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
if resp.PromptTokens != 1000 || resp.CompletionTokens != 200 || resp.TotalTokens != 1200 {
|
||||
t.Fatalf("用量 = (%d,%d,%d), want (1000,200,1200)", resp.PromptTokens, resp.CompletionTokens, resp.TotalTokens)
|
||||
}
|
||||
if resp.CacheHitTokens != 900 || resp.CacheMissTokens != 100 {
|
||||
t.Fatalf("缓存两档 = (%d,%d), want (900,100)", resp.CacheHitTokens, resp.CacheMissTokens)
|
||||
}
|
||||
if resp.ReasoningTokens != 150 {
|
||||
t.Fatalf("ReasoningTokens = %d, want 150", resp.ReasoningTokens)
|
||||
}
|
||||
if got := resp.CacheHitRate(); got != 0.9 {
|
||||
t.Fatalf("命中率 = %v, want 0.9", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOpenAIUsage_FallsBackToCachedTokensDetails 钉住另一条 OpenAI 兼容形态:
|
||||
// 只有 prompt_tokens_details.cached_tokens(OpenAI / DashScope)时也能得到命中数,
|
||||
// 且未命中数按提供商自己的不变式(prompt = hit + miss)补齐,而不是把命中率算成 100%。
|
||||
func TestOpenAIUsage_FallsBackToCachedTokensDetails(t *testing.T) {
|
||||
t.Parallel()
|
||||
body := `{
|
||||
"id": "cmpl-2", "model": "gpt-test",
|
||||
"choices": [{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
|
||||
"usage": {"prompt_tokens": 400, "completion_tokens": 10, "total_tokens": 410,
|
||||
"prompt_tokens_details": {"cached_tokens": 300}}
|
||||
}`
|
||||
srv := newCompletionServer(t, http.StatusOK, "application/json", body, nil)
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "gpt-test"})
|
||||
|
||||
resp, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, Options{})
|
||||
if err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
if resp.CacheHitTokens != 300 || resp.CacheMissTokens != 100 {
|
||||
t.Fatalf("缓存两档 = (%d,%d), want (300,100)", resp.CacheHitTokens, resp.CacheMissTokens)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsage_MissingFieldsAreTolerated 钉住缺字段容错:用量块只有最基本的三个数时
|
||||
// 不 panic、不报错,缓存命中数为零,未命中数按提供商自己的不变式(prompt = hit +
|
||||
// miss)补齐 —— 也就是「这次输入全都没命中缓存」,命中率因此是 0 而不是编造值。
|
||||
func TestUsage_MissingFieldsAreTolerated(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := newCompletionServer(t, http.StatusOK, "application/json", okCompletion, nil)
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "gpt-test"})
|
||||
|
||||
resp, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, Options{})
|
||||
if err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
if resp.CacheHitTokens != 0 || resp.CacheMissTokens != 5 || resp.ReasoningTokens != 0 {
|
||||
t.Fatalf("缺失字段 = (%d,%d,%d), want (0,5,0)", resp.CacheHitTokens, resp.CacheMissTokens, resp.ReasoningTokens)
|
||||
}
|
||||
if resp.CacheHitRate() != 0 {
|
||||
t.Fatalf("无缓存命中时的命中率 = %v, want 0", resp.CacheHitRate())
|
||||
}
|
||||
if resp.Content != "Hello!" {
|
||||
t.Fatalf("正文受用量字段影响: %q", resp.Content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOpenAIStreamUsage_ParsesCacheFields 钉住流式末块的用量块:缓存的 hit/miss 与
|
||||
// 思维链 token 一样原样折进聚合结果(流式是最常用的路径,漏了它等于没做)。
|
||||
func TestOpenAIStreamUsage_ParsesCacheFields(t *testing.T) {
|
||||
t.Parallel()
|
||||
lines := []string{
|
||||
`data: {"model":"deepseek-chat","choices":[{"index":0,"delta":{"content":"你好"}}]}`,
|
||||
`data: {"model":"deepseek-chat","choices":[],"usage":{"prompt_tokens":500,"completion_tokens":50,"total_tokens":550,"prompt_cache_hit_tokens":480,"prompt_cache_miss_tokens":20,"completion_tokens_details":{"reasoning_tokens":30}}}`,
|
||||
`data: [DONE]`,
|
||||
}
|
||||
srv := newCompletionServer(t, http.StatusOK, "text/event-stream", strings.Join(lines, "\n")+"\n", nil)
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "deepseek-chat"})
|
||||
|
||||
resp, err := ad.ChatStream(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("ChatStream: %v", err)
|
||||
}
|
||||
if resp.PromptTokens != 500 || resp.CacheHitTokens != 480 || resp.CacheMissTokens != 20 || resp.ReasoningTokens != 30 {
|
||||
t.Fatalf("流式用量 = %+v, want prompt 500 / hit 480 / miss 20 / reasoning 30", resp)
|
||||
}
|
||||
if resp.Content != "你好" {
|
||||
t.Fatalf("流式正文 = %q, want 你好", resp.Content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicUsage_ParsesPromptCacheFields 钉住 Anthropic 形状:命中 =
|
||||
// cache_read_input_tokens,未命中 = input_tokens + cache_creation_input_tokens
|
||||
// (Anthropic 的 input_tokens 不含缓存读到的部分,只取它会把命中率算成 0)。
|
||||
func TestAnthropicUsage_ParsesPromptCacheFields(t *testing.T) {
|
||||
t.Parallel()
|
||||
body := `{
|
||||
"id": "msg_1", "model": "claude-test", "stop_reason": "end_turn",
|
||||
"content": [{"type":"text","text":"你好"}],
|
||||
"usage": {"input_tokens": 100, "output_tokens": 20,
|
||||
"cache_creation_input_tokens": 300, "cache_read_input_tokens": 700}
|
||||
}`
|
||||
srv := newAnthropicServer(t, http.StatusOK, "application/json", body, nil)
|
||||
ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"})
|
||||
|
||||
resp, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, Options{})
|
||||
if err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
if resp.CacheHitTokens != 700 || resp.CacheMissTokens != 400 {
|
||||
t.Fatalf("缓存两档 = (%d,%d), want (700,400)", resp.CacheHitTokens, resp.CacheMissTokens)
|
||||
}
|
||||
// 既有口径不变:input_tokens 仍然只算 PromptTokens。
|
||||
if resp.PromptTokens != 100 || resp.TotalTokens != 120 {
|
||||
t.Fatalf("用量 = (%d,...,%d), want prompt 100 / total 120", resp.PromptTokens, resp.TotalTokens)
|
||||
}
|
||||
if resp.ReasoningTokens != 0 {
|
||||
t.Fatalf("Anthropic 不单独上报思维链 token,ReasoningTokens = %d, want 0", resp.ReasoningTokens)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicUsage_MissingCacheFieldsAreTolerated 钉住 Anthropic 的缺字段容错:
|
||||
// 未启用提示缓存的端点只给 input/output 两个数,解析照常、缓存为零。
|
||||
func TestAnthropicUsage_MissingCacheFieldsAreTolerated(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := newAnthropicServer(t, http.StatusOK, "application/json", anthropicTextResponse, nil)
|
||||
ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"})
|
||||
|
||||
resp, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, Options{})
|
||||
if err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
if resp.CacheHitTokens != 0 || resp.CacheMissTokens != 5 {
|
||||
// anthropicTextResponse 的 usage 是 input_tokens=5 / output_tokens=2:
|
||||
// 未命中数 = input + cache_creation = 5(缓存字段缺失时即为输入本身)。
|
||||
t.Fatalf("缓存两档 = (%d,%d), want (0,5)", resp.CacheHitTokens, resp.CacheMissTokens)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageLog_HasAllFieldsAndNoCredentials 钉住每次模型调用后那条结构化日志:
|
||||
// prompt / completion / total / reasoning / cacheHit / cacheMiss / hitRate 一个不少,
|
||||
// 且**绝不出现凭据**(apiKey / token)。
|
||||
func TestUsageLog_HasAllFieldsAndNoCredentials(t *testing.T) {
|
||||
t.Parallel()
|
||||
const secret = "sk-super-secret-credential"
|
||||
body := `{
|
||||
"id": "cmpl-1", "model": "deepseek-chat",
|
||||
"choices": [{"index":0,"message":{"role":"assistant","content":"你好"},"finish_reason":"stop"}],
|
||||
"usage": {"prompt_tokens": 1000, "completion_tokens": 200, "total_tokens": 1200,
|
||||
"prompt_cache_hit_tokens": 900, "prompt_cache_miss_tokens": 100,
|
||||
"completion_tokens_details": {"reasoning_tokens": 150}}
|
||||
}`
|
||||
srv := newCompletionServer(t, http.StatusOK, "application/json", body, nil)
|
||||
log, buf := captureJSONLog()
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, APIKey: secret, Model: "deepseek-chat"},
|
||||
WithOpenAILogger(log))
|
||||
|
||||
if _, err := ad.Chat(context.Background(), Request{
|
||||
Purpose: "chat",
|
||||
Messages: []Message{TextMessage(RoleUser, "hi")},
|
||||
}, Options{}); err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
|
||||
rec := usageLogRecord(t, buf.String())
|
||||
if rec["msg"] != usageLogEvent {
|
||||
t.Fatalf("事件名 = %v, want %q", rec["msg"], usageLogEvent)
|
||||
}
|
||||
for field, want := range map[string]float64{
|
||||
"prompt": 1000, "completion": 200, "total": 1200, "reasoning": 150,
|
||||
"cacheHit": 900, "cacheMiss": 100, "hitRate": 0.9,
|
||||
} {
|
||||
if got, ok := rec[field].(float64); !ok || got != want {
|
||||
t.Fatalf("日志字段 %s = %v, want %v(完整记录:%v)", field, rec[field], want, rec)
|
||||
}
|
||||
}
|
||||
if rec["model"] != "deepseek-chat" || rec["purpose"] != "chat" {
|
||||
t.Fatalf("日志缺少模型/用途:%v", rec)
|
||||
}
|
||||
raw := buf.String()
|
||||
for _, credential := range []string{secret, "Authorization", "apiKey", "Bearer"} {
|
||||
if strings.Contains(raw, credential) {
|
||||
t.Fatalf("用量日志里出现了凭据相关内容 %q:%s", credential, raw)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// captureJSONLog 返回输出 JSON 记录的 logger 与缓冲:用量日志要按**字段**断言,
|
||||
// 因此不能像 reasoning 用例那样只看文本子串。
|
||||
func captureJSONLog() (*slog.Logger, *bytes.Buffer) {
|
||||
buf := &bytes.Buffer{}
|
||||
return slog.New(slog.NewJSONHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})), buf
|
||||
}
|
||||
|
||||
// usageLogRecord 从捕获的日志文本里取出唯一一条用量记录(找不到或不止一条就报错)。
|
||||
func usageLogRecord(t *testing.T, raw string) map[string]any {
|
||||
t.Helper()
|
||||
var found []map[string]any
|
||||
for _, line := range strings.Split(strings.TrimSpace(raw), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" || !strings.Contains(line, `"msg":"`+usageLogEvent+`"`) {
|
||||
continue
|
||||
}
|
||||
rec := map[string]any{}
|
||||
if err := json.Unmarshal([]byte(line), &rec); err != nil {
|
||||
t.Fatalf("用量日志不是合法 JSON 行: %v(%q)", err, line)
|
||||
}
|
||||
found = append(found, rec)
|
||||
}
|
||||
if len(found) != 1 {
|
||||
t.Fatalf("用量日志条数 = %d, want 1(每次调用恰好一条):%s", len(found), raw)
|
||||
}
|
||||
return found[0]
|
||||
}
|
||||
Reference in new issue
Block a user