fix: void the cost of rows whose usage is unavailable

失败尝试与终态失败行的 usage_source 由 estimated 改 unavailable(用量确实
不可得),并在 TelemetryEmitter 的成本换算里为 unavailable 短路记 NULL。
短路刻意插在 cache_hit 分支之后: 缓存命中未产生新调用,0.0 是事实而非未知。
附 OCR 成功行的防回归钉(仍为 measured、settle 恒 0,设计 §3.3 剔出决定)。
This commit is contained in:
2026-07-30 10:37:48 -04:00
parent 76e7d9594c
commit 42e429eb58
3 changed files with 79 additions and 4 deletions
+7 -3
View File
@@ -44,7 +44,7 @@ class TelemetryEmitter:
response: LLMResponse | None, response: LLMResponse | None,
error: str | None, error: str | None,
) -> None: ) -> None:
"""逐次尝试记录(RetryMW 调用);失败尝试 usage 按 estimated 记 0""" """逐次尝试记录(RetryMW 调用);失败尝试无用量可言,记 0 并标 unavailable"""
await self._record( await self._record(
request=request, request=request,
call_id=call_id, call_id=call_id,
@@ -55,7 +55,7 @@ class TelemetryEmitter:
thinking=response.thinking if response else "", thinking=response.thinking if response else "",
prompt_tokens=response.prompt_tokens if response else 0, prompt_tokens=response.prompt_tokens if response else 0,
completion_tokens=response.completion_tokens if response else 0, completion_tokens=response.completion_tokens if response else 0,
usage_source=response.usage_source if response else "estimated", usage_source=response.usage_source if response else "unavailable",
latency_ms=latency_ms, latency_ms=latency_ms,
ttft_ms=response.ttft_ms if response else None, ttft_ms=response.ttft_ms if response else None,
max_inter_token_ms=response.max_inter_token_ms if response else None, max_inter_token_ms=response.max_inter_token_ms if response else None,
@@ -97,7 +97,7 @@ class TelemetryEmitter:
thinking="", thinking="",
prompt_tokens=0, prompt_tokens=0,
completion_tokens=0, completion_tokens=0,
usage_source="estimated", usage_source="unavailable",
latency_ms=latency_ms, latency_ms=latency_ms,
ttft_ms=None, ttft_ms=None,
max_inter_token_ms=None, max_inter_token_ms=None,
@@ -129,6 +129,10 @@ class TelemetryEmitter:
# 失败/终态行 None;未注入价格表 = 恒 None(M1 现状) # 失败/终态行 None;未注入价格表 = 恒 None(M1 现状)
if cache_hit: if cache_hit:
cost: float | None = 0.0 cost: float | None = 0.0
elif usage_source == "unavailable":
# 用量不可得: 宁可算不出成本,也不算错成本(解耦设计 §3.1 不变式)。
# 必须排在 cache_hit 之后——缓存命中未产生新调用,0.0 是事实而非未知
cost = None
elif error is None and model and self._pricing is not None: elif error is None and model and self._pricing is not None:
cost = self._pricing.cost(model, prompt_tokens, completion_tokens) cost = self._pricing.cost(model, prompt_tokens, completion_tokens)
else: else:
+15
View File
@@ -394,6 +394,21 @@ class TestTelemetry:
assert recorder.rows[1]["error"] is None assert recorder.rows[1]["error"] is None
assert recorder.rows[1]["prompt_tokens"] == 0 assert recorder.rows[1]["prompt_tokens"] == 0
async def test_success_row_stays_measured_and_settles_zero(self):
"""OCR 的 0 token 是**事实**而非未知(est_tokens 解耦设计 §3.3 剔出决定)。
三态化不得把 OCR 成功行改成 `unavailable`——那会灌水缺口度量
`COUNT(*) WHERE usage_source='unavailable'`;settle 恒 0 的差异①同样不动。
"""
recorder = _MemoryRecorder()
client, limiter, _ = _client([_src(tpm=1000, est_tokens=400)], ["text"], telemetry=recorder)
await client.recognize_text(b"jpg")
row = recorder.rows[0]
assert row["usage_source"] == "measured"
assert row["prompt_tokens"] == 0 and row["completion_tokens"] == 0
assert row["error"] is None
assert (await limiter.source_stats("m1")).tpm_used == 0 # settle(0) 全额退回预扣
class TestAssembly: class TestAssembly:
_ENV = { _ENV = {
+57 -1
View File
@@ -9,6 +9,7 @@ import pytest
from polygateway.errors import CircuitOpenError, RequestRejectedError from polygateway.errors import CircuitOpenError, RequestRejectedError
from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW
from polygateway.pricing import ModelPrice, PricingTable
from polygateway.telemetry.sqlite import SQLiteRecorder from polygateway.telemetry.sqlite import SQLiteRecorder
from polygateway.types import ChatRequest, LLMResponse, SourceConfig from polygateway.types import ChatRequest, LLMResponse, SourceConfig
@@ -68,6 +69,10 @@ def _source():
) )
# 输出单价 8 元/百万: 改前 `unavailable` 行按兜底的 0/4000 换算恰好是 0.032
_PRICING = PricingTable({"m": ModelPrice(input_per_1m=1.0, output_per_1m=8.0)})
async def _record_minimal(recorder, call_id="c1", **overrides): async def _record_minimal(recorder, call_id="c1", **overrides):
fields = { fields = {
"call_id": call_id, "call_id": call_id,
@@ -168,7 +173,58 @@ class TestEmitter:
) )
row = rec.rows[0] row = rec.rows[0]
assert row["error"].startswith("TransientError") assert row["error"].startswith("TransientError")
assert row["response"] == "" and row["usage_source"] == "estimated" # 失败尝试没有任何用量信息可言 → unavailable(设计 §3.2 #6)
assert row["response"] == "" and row["usage_source"] == "unavailable"
assert row["cost"] is None
async def test_terminal_failure_row_is_unavailable(self):
rec = _MemoryRecorder()
await TelemetryEmitter(rec, pricing=_PRICING).emit_terminal_failure(
request=_REQ, call_id="cid-t", latency_ms=5, error="cancelled"
)
row = rec.rows[0]
assert row["usage_source"] == "unavailable" and row["cost"] is None
@pytest.mark.parametrize(("prompt", "completion"), [(0, 0), (0, 4000)])
async def test_unavailable_success_row_has_null_cost(self, prompt, completion):
"""产生了真实调用但用量不可得 → cost 记 NULL(设计 §3.1 不变式)。
参数第二组是改前兜底写出的 `0/4000` 形态: 那时换算出 0.032 的假金额。
"""
rec = _MemoryRecorder()
await TelemetryEmitter(rec, pricing=_PRICING).emit_attempt(
request=_REQ,
source=_source(),
call_id="cid-u",
latency_ms=42,
response=_resp(
usage_source="unavailable", prompt_tokens=prompt, completion_tokens=completion
),
error=None,
)
assert rec.rows[0]["cost"] is None
async def test_measured_row_still_priced(self):
"""对照组: 同一价格表下 measured 行照常换算,证明 None 不是价格表没接上。"""
rec = _MemoryRecorder()
await TelemetryEmitter(rec, pricing=_PRICING).emit_attempt(
request=_REQ,
source=_source(),
call_id="cid-m",
latency_ms=42,
response=_resp(prompt_tokens=0, completion_tokens=4000),
error=None,
)
assert rec.rows[0]["cost"] == pytest.approx(0.032)
async def test_cache_hit_keeps_zero_cost_even_when_unavailable(self):
"""缓存命中未产生新调用,0.0 是事实而非未知 → 短路必须排在 cache_hit 之后。"""
rec = _MemoryRecorder()
await TelemetryEmitter(rec, pricing=_PRICING).emit_cache_hit(
request=_REQ,
response=_resp(cache_hit=True, usage_source="unavailable", completion_tokens=4000),
)
assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["cost"] == 0.0
async def test_multimodal_messages_digested_before_storage(self): async def test_multimodal_messages_digested_before_storage(self):
rec = _MemoryRecorder() rec = _MemoryRecorder()