PolyGateway/research-wiki/plans/2026-07-30-est-tokens-decoupling-plan.md

est_tokens 解耦实施计划

• 目标: 把 SourceConfig.est_tokens 的两个职责(TPM 入场预扣 / usage 缺失时的遥测用量兜底)拆开,并解绑 tpm > 0 ⇒ est_tokens > 0 装配约束。
• 方案概述: usage 不可得时遥测记 0/0 并标新值 unavailable、cost 落 NULL(不再拿限流押金编计费数字);est_tokens 未填时由库按 tpm // 60 派生预扣量,字段降为可选调优覆盖(保留不删不改名)。全部依据已批准设计 research-wiki/designs/2026-07-30-est-tokens-decoupling-design.md,其 §3.2 的 11 条改动项已钉死行号。
• 涉及技术: Python 3.11 frozen dataclass、pytest(含 tests/contracts 双后端契约测试)、真实 Redis(integration/contracts)、pydantic-settings 不涉及改动。
• 溯源: Gitea issue #2;wiki 实体 design:est-tokens-decoupling。

1. 任务排序的硬约束(先读这节再动手)

三处改动互相牵制,顺序错了会引入押金泄漏或计费造假,且中间态不报错、只静默偏差:

若单独先做	后果
先改 usage 兜底为 (0, 0),结算点还没切派生值	成功调用 actual = 0 而入场押了 est_tokens,delta 为负 → 押金整笔退回,TPM 闸退化成进门即放行
先解绑约束(允许 est_tokens=0),结算点还没切派生值	同上,est_tokens=0 的源入场押派生值、结算退 0
先改 openai_compat.py:176 而 _merge 还是二值逻辑	embedding 的 unavailable 批被 any(== "estimated") 判 False 从而误标 measured,cost 照算

因此排序为:先加能力(零行为变更)→ 再把所有调用点切到新能力(此时等价,因显式值优先)→ 再让三态生效 → 最后解绑约束。Task 1-2 完成后行为逐字不变,Task 3 才是行为变更主体,Task 4 才让派生值真正启用。不要合并或调换 Task 2 / 3 / 4 的顺序。

2. 文件结构

文件	职责	涉及任务
src/polygateway/types.py	新增 SourceConfig.effective_est_tokens() 派生方法与 usage_source 三态值域常量;删除 _validate_gates 里的条件必填;修 EmbeddingTransportResult 行内注释	T1、T3、T4
src/polygateway/middleware/ratelimit.py	QuotaGate.try_acquire 入场预扣改用派生值	T2
src/polygateway/middleware/retry.py	成功侧(:338)与失败侧(:370)结算改用派生值	T2
src/polygateway/embedding.py	同上(:271/:294);_merge 三态合并;_total_cost 存在不可得批时整体 NULL	T2、T3
src/polygateway/transports/openai_compat.py	两处 usage 兜底改 unavailable;打捞覆盖加前置条件	T3
src/polygateway/middleware/telemetry.py	cost 短路(插在 cache_hit 之后);失败尝试与终态失败改标 unavailable	T3
src/polygateway/ocr.py	不改(设计 §3.3 已剔出),仅加防回归测试	T3
research-wiki/ARCHITECTURE.md	§7.7 行 428、§5.1 行 331、§4.4 行 305、§7.1 行 384	T5
research-wiki/migrations/chsanalyzer.md	行 151 保留项改判为有意放弃;G2(行 185)标注已解决	T5
.env.example	行 11 删除"TPM > 0 时 EST_TOKENS 必填 > 0"	T5
CHANGELOG.md	行为变更小节(值域新增 + cost 口径 + 约束解绑)	T5

测试文件:tests/unit/test_types.py、test_retry.py、test_openai_compat.py、test_telemetry.py、test_embedding.py、test_ocr_client.py、tests/contracts/test_limiter_contract.py。

3. 保真校验

本计划不新增任何 reference/ 移植代码,但触及 ARCHITECTURE §1.4 关键资产索引中的"遥测口径"与"限流结算",且有意推翻一条已声明保留的迁移行为(CHS invokers.py:241-254 的"usage 缺失按 est 估算",见设计 §4)。故设以下检查点,每个任务完成时逐条确认:

1. Permit.settle() 的"多退少补 + 幂等 flag"语义不得改变;release() 的 finally 必然执行不得改变。
2. 不得触碰 backends/redis/limiter.py 的 Lua 脚本与 backends/memory/limiter.py 的窗口/租约算法——本计划只改传入 try_acquire 的数值来源,不改闸门算法。
3. 不得改变 errors.py 四分类归属,不得新增运行时异常类型;值域违反不走异常路径(设计 §3.1 已裁决)。
4. ocr.py 的 settle 恒 0 与 usage_source="measured" 保持不变。
5. 遥测 18 字段冻结不变、无 DDL 变更(两 schema 的 cost 列已可空)。

4. 任务清单

T1 — 加派生能力与值域常量(零行为变更)

• 改 src/polygateway/types.py

新增模块级值域常量与 SourceConfig 方法。派生按源自身 tpm,全局 tpm 不参与(设计 §7 已声明为既有限制、本次不修):

USAGE_SOURCES = frozenset({"measured", "estimated", "unavailable"})
"""usage_source 值域;仅约束库内生产侧取值,不在 frozen dataclass 上做运行时校验。"""

_EST_TOKENS_QUOTA_DIVISOR = 60
"""未显式配置时的预扣量除数: 假定一次调用约占一秒钟的 TPM 配额份额。"""

def effective_est_tokens(self) -> int:
    """TPM 入场预扣量: 显式配置优先,否则按 tpm 派生(设计 §2.2)。"""
    if self.est_tokens > 0:
        return self.est_tokens
    if self.tpm > 0:
        return max(1, self.tpm // _EST_TOKENS_QUOTA_DIVISOR)
    return 0

验收标准: 方法存在且为纯函数(不读全局、不 await);此任务不修改任何调用点,全库行为逐字不变。

测试要求(先失败后通过:方法不存在时 AttributeError)——tests/unit/test_types.py:

• tpm=6000, est_tokens=0 → 100;tpm=600000, est_tokens=0 → 10000(尺度无关:两者在途上限同为 60)
• tpm=30, est_tokens=0 → 1(下界不塌到 0)
• tpm=0, est_tokens=0 → 0(TPM 闸未启用,不预扣)
• tpm=6000, est_tokens=4000 → 4000(显式值优先于派生)
• USAGE_SOURCES 恰为三元集合

验证: conda run --no-capture-output -n PolyGateway pytest tests/unit/test_types.py -v → 全 PASS

T2 — 五个调用点切到派生值(零行为变更)

此时 est_tokens > 0 仍是必填(约束未解绑),故 effective_est_tokens() 恒返回显式值,行为与改前逐字相同。这一步只是把数值来源换掉,为 T3 铺路。

• 改 src/polygateway/middleware/ratelimit.py:26 — source.est_tokens → source.effective_est_tokens()
• 改 src/polygateway/middleware/retry.py:370(失败侧,if not dead 分支内)→ source.effective_est_tokens()
• 改 src/polygateway/embedding.py:294(失败侧)→ 同上
• 改 src/polygateway/middleware/retry.py:338(成功侧)— 加不可得分支:

if result.usage_source == "unavailable":
    actual = source.effective_est_tokens()
else:
    actual = result.prompt_tokens + result.completion_tokens

• 改 src/polygateway/embedding.py:271(成功侧)— 同构,actual = result.prompt_tokens 落在 else 分支。

TransportResult.usage_source(types.py:75)与 EmbeddingTransportResult.usage_source(types.py:273)均为必填字段,在这两处的 result 局部变量上直接可读。

验收标准: 五处均不再直接读 source.est_tokens;新分支在本任务中永不触发(尚无 unavailable 生产者),现有测试全绿即证明零行为变更。不要在本任务修改 retry.py:329 的 actual = 0 初值,也不要动 RequestRejectedError/ResultInvalidError/SourceDeadError 三条失败分支(设计 §3.3:它们的 actual 停在 0 属既有行为)。

测试要求(回归保护,先失败后通过不适用于零行为变更任务,故以"现有测试不得回归 + 新增等价性断言"为门):

• 新增 tests/unit/test_retry.py:tpm=1000, est_tokens=400 的源在 usage 正常返回时,settle 收到的 actual 等于实测 token 之和(锁定 else 分支);现有 test_settle_uses_actual_usage 必须继续通过
• tests/contracts/test_limiter_contract.py 全绿(双后端)

验证:

conda run --no-capture-output -n PolyGateway pytest tests/unit tests/contracts -v

→ 全 PASS。Redis 契约测试用真实 Redis db3,不得与其他 Redis 测试并跑(时序隔离)。

T3 — 值域三态生效(行为变更主体)

• 改 src/polygateway/transports/openai_compat.py:146 — 兜底不再读 est_tokens:

return 0, 0, "unavailable"

• 改 src/polygateway/transports/openai_compat.py:176 — embedding 兜底同理 return 0, "unavailable"
• 改 src/polygateway/transports/openai_compat.py:336 — 打捞覆盖加前置条件(否则 0/0 会被标 estimated 而算出假的 0.0):

if salvaged and usage_source == "measured":
    usage_source = "estimated"  # 收到 usage 帧但流被截断: 数字真实、可信度降级

• 改 src/polygateway/middleware/telemetry.py:130-135 — cost 短路,插在 cache_hit 分支之后(缓存命中未产生新调用,0.0 是事实):

if cache_hit:
    cost: float | None = 0.0
elif usage_source == "unavailable":
    cost = None          # 用量不可得: 宁可算不出成本,也不算错成本
elif error is None and model and self._pricing is not None:
    cost = self._pricing.cost(model, prompt_tokens, completion_tokens)
else:
    cost = None

• 改 src/polygateway/middleware/telemetry.py:58 与 :100 — 失败尝试与终态失败的 usage_source 由 "estimated" 改 "unavailable"(cost 本已是 None,不改金额)
• 改 src/polygateway/embedding.py:383,390 — 二值合并扩三态(优先级:任一不可得 → 整体不可得):

sources = {o.result.usage_source for o in outcomes}
if "unavailable" in sources:
    merged_source = "unavailable"
elif "estimated" in sources:
    merged_source = "estimated"
else:
    merged_source = "measured"

• 改 src/polygateway/embedding.py:397 _total_cost — 存在 unavailable 批时整体返回 None(逐批求和会给出偏低却看似有效的金额)
• 改 src/polygateway/types.py:273 — 行内注释 # measured | estimated → 三态(内核里不留矛盾注释)

验收标准: 全库不再有任何位置把 est_tokens 写进遥测用量;ocr.py 一字未动。

测试要求(先失败后通过):

• test_openai_compat.py:est_tokens=4000 + usage 帧缺失 → (0, 0, "unavailable")(改前返回 (0, 4000, "estimated"),故先失败)
• test_openai_compat.py:打捞 + usage 帧存在 → estimated;打捞 + usage 缺失 → unavailable(改前后者为 estimated)
• test_telemetry.py:成功行 usage_source="unavailable" → record_llm_call 收到 cost=None(改前按 4000×输出单价算出 0.032)
• test_telemetry.py:cache_hit=True 且 unavailable → cost 仍为 0.0(锁定分支次序)
• test_telemetry.py:失败尝试与终态失败行 usage_source == "unavailable"
• test_embedding.py:混合批 measured + unavailable → 整体 unavailable 且 cost is None(改前误标 measured)
• test_ocr_client.py:OCR 成功行仍为 measured 且 settle 恒 0(防回归,锁定设计 §3.3 的剔出决定)

验证: conda run --no-capture-output -n PolyGateway pytest tests/unit -v → 全 PASS;conda run --no-capture-output -n PolyGateway pytest tests/contracts -v → 全 PASS(T2 的结算分支此时首次被激活,契约测试须复跑)

T4 — 解绑装配约束(派生值真正启用)

• 改 src/polygateway/types.py:125-126 — 删除:

if self.tpm > 0 and self.est_tokens <= 0:
    raise ValueError("启用 TPM 闸时 est_tokens 必须 > 0(入场预扣依据)")

_validate_gates 的其余部分(timeout_s > 0、四个限额非负)保留不动。est_tokens 字段本身与 {SCOPE}__{PROVIDER}__{N}__EST_TOKENS 环境键保留不删不改名(迁移兼容硬约束);config.py:40 的键映射无需改动。

验收标准: tpm=6000, est_tokens=0 可构造;该源入场预扣 100 且成功/失败/取消三侧结算均按 100 结算,TPM 窗口不出现负向偏差。

测试要求(先失败后通过:改前构造即抛 ValueError):

• test_types.py:tpm=6000, est_tokens=0 构造成功
• test_retry.py:成功侧——未填 est_tokens、tpm>0、usage 帧缺失的成功调用后,TPM 窗口残留量等于派生预扣量而非 0(这是设计中最易漏的一条,回归 §3.2 #9;在 test_retry.py:149 的 _src("a", tpm=1000, est_tokens=400) 旁加 est_tokens=0 用例)
• test_retry.py:失败侧——同配置的瞬时失败调用后,窗口残留量同为派生预扣量(回归 §3.2 #8)
• tests/contracts/test_limiter_contract.py:上述两条在内存与 Redis 双后端均成立

验证: conda run --no-capture-output -n PolyGateway make ci(即 check + test,含 import-linter 契约)→ 全 PASS

T5 — 权威文档与发布物同步

• 改 research-wiki/ARCHITECTURE.md 四处:§7.7 行 428(est_tokens 描述:可选调优覆盖 + 派生规则,删去"亦作 usage 缺失时的保守兜底")、§5.1 行 331(usage_source 三态 + cost NULL 口径)、§4.4 行 305("token 按 est_tokens 预扣" → 按有效预扣量)、§7.1 行 384(打捞路径强制 estimated → 仅在收到 usage 帧时降级)
• 改 research-wiki/migrations/chsanalyzer.md:行 151 由"保留"改判"有意放弃"并写入设计 §4 的理由(CHS 只记单个 total_tokens 不存在分配问题;保守在计费语境无安全方向);G2(行 185)标注已由本设计解决
• 改 .env.example 行 11:删除"TPM > 0 时 EST_TOKENS 必填 > 0",改注为"可选;未填则库按 tpm 派生"
• 改 CHANGELOG.md:新增"行为收紧/变更"小节三条——usage_source 新增 unavailable、用量不可得行 cost 由数值变 NULL、est_tokens 降为可选
• 改 wiki 用户文档站(按 docs-convention.md §2):usage/成本口径说明须写明缺口查询为 WHERE usage_source='unavailable' AND cache_hit = false(必须带 cache_hit 限定:缓存命中行按裁决 cost 为 0.0 且标 unavailable,本无账目缺口,不加限定则度量偏高)
• 回帖 Gitea issue #2:结论与下游可删绕行校验的时点

验收标准: 全库 grep EST_TOKENS 必填、est_tokens 兜底相关表述无残留;ARCHITECTURE.md 无自相矛盾表述。

测试要求: 纯文档,无行为测试。以 grep 输出为验收证据。

验证: conda run --no-capture-output -n PolyGateway make ci → PASS;grep -rn "EST_TOKENS 必填" . --exclude-dir=.git → 无输出

5. 完成判定

• T1-T5 全部 checkbox 勾选,每个任务一次语义化提交(commit skill)
• make ci 全绿(含 ruff、import-linter 洋葱契约、pytest 覆盖率)
• 设计 §6 测试表的 10 行断言全部有对应测试且可出示"先失败后通过"证据(T2 的零行为变更任务以"现有测试不回归 + 等价性断言"替代)
• 派新上下文 verifier subagent 独立验证(verification-before-completion,里程碑级/跨多文件硬门)
• 版本 bump 与 CHANGELOG 同步发布(不得裸发)

6. 明确不做

派生值取全局与单源 tpm 较紧者(需改三处 QuotaGate 装配,修的是既有缺口,设计 §7 已声明另开 issue);遥测驱动的 p90 自适应预估(设计 §5 已否决,待实测证据);ocr.py 的 usage 标记(设计 §3.3 已剔出);retry.py 另外三条失败分支的 actual 初值。