diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index acb7da7..60c746b 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -438,6 +438,7 @@ class RetryMW: usage_source=result.usage_source, cached_prompt_tokens=result.cached_prompt_tokens, model_reported=result.model_reported, + reasoning_tokens=result.reasoning_tokens, ) async def _settle_and_release(self, permit: Permit, actual: int) -> None: diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index b2093c7..66b7cfb 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -64,6 +64,7 @@ class TelemetryEmitter: error=error, cached_prompt_tokens=response.cached_prompt_tokens if response else None, model_reported=response.model_reported if response else None, + reasoning_tokens=response.reasoning_tokens if response else None, # 唯一有"生效源"的入口,故是唯一能并上 extra_body 的(设计决策 D) sampling=canonical_sampling_json(merge_sampling(source.extra_body, request.sampling)), ) @@ -90,6 +91,7 @@ class TelemetryEmitter: # 统计供应商缓存命中率必须带 WHERE cache_hit = false,否则重复计数。 cached_prompt_tokens=response.cached_prompt_tokens, model_reported=response.model_reported, + reasoning_tokens=response.reasoning_tokens, # 由最外层 TelemetryMW 调用,手上没有 source。缓存命中行无损: # sampling 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同 sampling=canonical_sampling_json(request.sampling), @@ -117,6 +119,7 @@ class TelemetryEmitter: error=error, cached_prompt_tokens=None, model_reported=None, + reasoning_tokens=None, # 无具体源,与 model/provider/source_name 置空同一先例(设计决策 D) sampling=canonical_sampling_json(request.sampling), ) @@ -142,6 +145,7 @@ class TelemetryEmitter: cached_prompt_tokens: int | None, model_reported: str | None, sampling: str | None, + reasoning_tokens: int | None, ) -> None: try: # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用); @@ -182,6 +186,7 @@ class TelemetryEmitter: cached_prompt_tokens=cached_prompt_tokens, model_reported=model_reported, sampling=sampling, + reasoning_tokens=reasoning_tokens, ) except asyncio.CancelledError: raise diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index 5425dfe..1765d85 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -275,4 +275,5 @@ class TelemetryRecorder(Protocol): cached_prompt_tokens: int | None, model_reported: str | None, sampling: str | None, + reasoning_tokens: int | None, ) -> None: ... diff --git a/src/polygateway/telemetry/postgres.py b/src/polygateway/telemetry/postgres.py index f3fac3c..a195aa5 100644 --- a/src/polygateway/telemetry/postgres.py +++ b/src/polygateway/telemetry/postgres.py @@ -42,7 +42,8 @@ CREATE TABLE IF NOT EXISTS llm_calls ( created_at TIMESTAMPTZ NOT NULL DEFAULT now(), cached_prompt_tokens INTEGER, model_reported TEXT, - sampling TEXT + sampling TEXT, + reasoning_tokens INTEGER ); """ @@ -51,6 +52,7 @@ _BACKFILL = ( ("cached_prompt_tokens", "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER"), ("model_reported", "ALTER TABLE llm_calls ADD COLUMN model_reported TEXT"), ("sampling", "ALTER TABLE llm_calls ADD COLUMN sampling TEXT"), + ("reasoning_tokens", "ALTER TABLE llm_calls ADD COLUMN reasoning_tokens INTEGER"), ) # 探测现有列;尊重 search_path(to_regclass 按当前 search_path 解析) @@ -81,6 +83,7 @@ _COLUMNS = ( "cached_prompt_tokens", "model_reported", "sampling", + "reasoning_tokens", ) _INSERT = ( diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index a53b7d9..b8622c6 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -37,7 +37,8 @@ CREATE TABLE IF NOT EXISTS llm_calls ( created_at TEXT NOT NULL DEFAULT (datetime('now')), cached_prompt_tokens INTEGER, model_reported TEXT, - sampling TEXT + sampling TEXT, + reasoning_tokens INTEGER ); """ @@ -47,6 +48,7 @@ _BACKFILL_COLUMNS = ( ("cached_prompt_tokens", "INTEGER"), ("model_reported", "TEXT"), ("sampling", "TEXT"), + ("reasoning_tokens", "INTEGER"), ) _COLUMNS = ( @@ -71,6 +73,7 @@ _COLUMNS = ( "cached_prompt_tokens", "model_reported", "sampling", + "reasoning_tokens", ) _INSERT = ( diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index c8111df..054b494 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -177,6 +177,25 @@ def _coerce_cached_tokens(usage: Any) -> int | None: return cached +def _coerce_reasoning_tokens(usage: Any) -> int | None: + """取 usage.completion_tokens_details.reasoning_tokens(issue #6);形态异常一律 None。 + + 与 `_coerce_cached_tokens` 逐条同构(两者是 OpenAI 兼容 usage 里对称的一对): + `0` 如实保留、负数与非整数归 None、`bool` 显式排除。差别只在语义——本字段 + 的 None 是"**本次调用**未上报"而非"该源不上报": 中转在上游不返回 usage 时 + 会本地补算并整体替换 usage 对象,把 details 一并吃掉(findings §4c)。 + """ + if not isinstance(usage, dict): + return None + details = usage.get("completion_tokens_details") + if not isinstance(details, dict): + return None + reasoning = details.get("reasoning_tokens") + if isinstance(reasoning, bool) or not isinstance(reasoning, int) or reasoning < 0: + return None + return reasoning + + def _coerce_model_reported(value: Any) -> str | None: """取响应体的 model 字段(issue #3);非 str 或空白串一律 None,收口时去空白。 @@ -400,6 +419,7 @@ class OpenAICompatTransport: raw={"usage": sink.get("usage")}, cached_prompt_tokens=_coerce_cached_tokens(sink.get("usage")), model_reported=_coerce_model_reported(sink.get("model")), + reasoning_tokens=_coerce_reasoning_tokens(sink.get("usage")), ) def _check_done( @@ -484,6 +504,7 @@ class OpenAICompatTransport: raw={"usage": body.get("usage")}, cached_prompt_tokens=_coerce_cached_tokens(body.get("usage")), model_reported=_coerce_model_reported(body.get("model")), + reasoning_tokens=_coerce_reasoning_tokens(body.get("usage")), ) async def aclose(self) -> None: diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 723131f..7a21924 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -99,6 +99,15 @@ class LLMResponse: model_reported: str | None = None """API 响应体里的 model 字段;None = 未上报。与 `model`(配置别名)可能 分叉——供应商把别名指向新权重时,实验复现必须认这个串。""" + reasoning_tokens: int | None = None + """推理消耗的输出 token 数(含在 `completion_tokens` 内,故不影响成本总额, + 只补归因;issue #6)。 + + `None` = **本次调用**未上报,**不是**"该源不上报"——中转网关在上游不返回 + usage 时会用本地 tokenizer 补算并整体替换 usage 对象,把 + `completion_tokens_details` 一并吃掉(findings §4c 实测同一请求 10 轮呈 + 6:4 双峰)。实测三家供应商在未推理时都是整个 details 缺失、无人上报 `0`, + 故下游判据须为 `in (None, 0)`,写 `== 0` 的条件永远不成立。""" @dataclass(frozen=True) @@ -151,9 +160,10 @@ class TransportResult: ttft_ms: float | None max_inter_token_ms: float | None raw: dict[str, Any] - # —— 可观测字段(issue #3;带默认值,非 OpenAI 兼容的 transport 可不填)—— + # —— 可观测字段(issue #3/#6;带默认值,非 OpenAI 兼容的 transport 可不填)—— cached_prompt_tokens: int | None = None model_reported: str | None = None + reasoning_tokens: int | None = None @dataclass(frozen=True) diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 3b1203c..11382ad 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -43,6 +43,7 @@ _EXPECTED_COLUMNS = [ "cached_prompt_tokens", "model_reported", "sampling", + "reasoning_tokens", ] # run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见 @@ -107,6 +108,7 @@ async def _record_minimal( "cached_prompt_tokens": None, "model_reported": None, "sampling": None, + "reasoning_tokens": None, } fields.update(overrides) await recorder.record_llm_call(**fields) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 5bce879..973cc59 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -394,6 +394,81 @@ class TestObservabilityFields: assert set(result.raw) == {"usage"} +class TestReasoningTokens: + """issue #6: 推理消耗的输出 token,与 issue #3 的 cached_tokens 对称。 + + 实测三家供应商在"未推理"时是整个 completion_tokens_details 缺失,无人上报 + 0;且中转在上游不返回 usage 时会本地补算并吃掉该对象。故 None 的语义是 + "本次调用未上报",不是"该源不上报"(findings §4c)。 + """ + + def _reasoning_usage(self, reasoning): + return {**_USAGE, "completion_tokens_details": {"reasoning_tokens": reasoning}} + + async def test_stream_reads_reasoning_tokens(self): + def handler(request): + return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(7))) + + result = await _complete(_transport_for(handler), _source()) + assert result.reasoning_tokens == 7 + + async def test_non_stream_reads_reasoning_tokens(self): + def handler(request): + return httpx.Response( + 200, + json={ + "choices": [{"message": {"content": "42"}}], + "usage": self._reasoning_usage(7), + }, + ) + + result = await _complete(_transport_for(handler), _source(), stream=False) + assert result.reasoning_tokens == 7 + + async def test_zero_reasoning_tokens_is_a_real_zero(self): + """0(上报了且确实没推理)与 None(本次未上报)必须可区分。""" + + def handler(request): + return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(0))) + + result = await _complete(_transport_for(handler), _source()) + assert result.reasoning_tokens == 0 + + async def test_usage_without_details_is_none(self): + def handler(request): + return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE)) + + result = await _complete(_transport_for(handler), _source()) + assert result.reasoning_tokens is None + + @pytest.mark.parametrize("bad", ["abc", -1, True, 1.5, None, [], {"x": 1}]) + async def test_malformed_reasoning_tokens_degrade_to_none(self, bad): + """`True` 必须排除: Python 里 isinstance(True, int) 为真。""" + + def handler(request): + return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(bad))) + + result = await _complete(_transport_for(handler), _source()) + assert result.reasoning_tokens is None + + async def test_details_not_a_dict_is_none(self): + def handler(request): + usage = {**_USAGE, "completion_tokens_details": "oops"} + return _sse_stream(_chunk(content="ok"), _chunk(usage=usage)) + + result = await _complete(_transport_for(handler), _source()) + assert result.reasoning_tokens is None + + async def test_salvage_path_records_none_not_zero(self): + """打捞路径拿不到 usage 帧: 记 None(未知)而非 0(确定没推理)。""" + + def handler(request): + return _sse_stream(_chunk(content="ok"), done=False) + + result = await _complete(_transport_for(handler), _source(missing_done="salvage")) + assert result.reasoning_tokens is None + + class TestNonStreamFastPath: async def test_non_stream_parses_message(self): def handler(request): diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index 74bb5f9..fe722eb 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -117,6 +117,7 @@ class _DummyRecorder: cached_prompt_tokens, model_reported, sampling, + reasoning_tokens, ) -> None: ... diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index bb54ae9..affce44 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -209,11 +209,13 @@ class TestObservabilityPassthrough: raw={}, cached_prompt_tokens=64, model_reported="MiniMax-Text-01-250321", + reasoning_tokens=7, ) mw, *_ = _harness([_src("a")], [result]) resp = await mw(_REQ) assert resp.cached_prompt_tokens == 64 assert resp.model_reported == "MiniMax-Text-01-250321" + assert resp.reasoning_tokens == 7 # model 仍是配置别名: 真实版本是旁证,不顶替溯源主字段 assert resp.model == "m" @@ -221,6 +223,7 @@ class TestObservabilityPassthrough: mw, *_ = _harness([_src("a")], [_ok()]) resp = await mw(_REQ) assert resp.cached_prompt_tokens is None and resp.model_reported is None + assert resp.reasoning_tokens is None class TestRetryAndFailover: diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 25128af..5a2df3c 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -1,4 +1,4 @@ -"""遥测子系统测试: SQLiteRecorder(21 列)+ TelemetryEmitter 单一 helper + TelemetryMW。""" +"""遥测子系统测试: SQLiteRecorder(22 列)+ TelemetryEmitter 单一 helper + TelemetryMW。""" import asyncio import json @@ -39,6 +39,7 @@ _EXPECTED_COLUMNS = [ "cached_prompt_tokens", "model_reported", "sampling", + "reasoning_tokens", ] @@ -102,6 +103,7 @@ async def _record_minimal(recorder, call_id="c1", **overrides): "cached_prompt_tokens": None, "model_reported": None, "sampling": None, + "reasoning_tokens": None, } fields.update(overrides) await recorder.record_llm_call(**fields) @@ -158,6 +160,22 @@ class TestSQLiteRecorder: assert rows["c-zero"] == 0 # 真实零命中,读回仍是 0 而非 NULL assert rows["c-none"] is None + async def test_reasoning_tokens_column_round_trip(self, tmp_path): + """issue #6: 7 / 0 / None 三种值各自如实落库,0 与 NULL 不得混同。""" + recorder = SQLiteRecorder(tmp_path / "t.db") + await _record_minimal(recorder, call_id="r-some", reasoning_tokens=7) + await _record_minimal(recorder, call_id="r-zero", reasoning_tokens=0) + await _record_minimal(recorder, call_id="r-none", reasoning_tokens=None) + recorder.close() + rows = dict( + sqlite3.connect(tmp_path / "t.db") + .execute("SELECT call_id, reasoning_tokens FROM llm_calls") + .fetchall() + ) + assert rows["r-some"] == 7 + assert rows["r-zero"] == 0 # 上报了且确实没推理 + assert rows["r-none"] is None # 本次调用未上报 + async def test_sampling_column_round_trips(self, tmp_path): """issue #4: 采样参数落库,否则事后无法证明某批数据跑在什么温度下。""" recorder = SQLiteRecorder(tmp_path / "t.db") @@ -284,6 +302,7 @@ class TestPostgresBackfillDiscipline: "cached_prompt_tokens", "model_reported", "sampling", + "reasoning_tokens", ] def _recorder(self, conn): @@ -384,11 +403,12 @@ class TestEmitterObservabilityFields: source=_source(), call_id="cid-1", latency_ms=42, - response=_resp(cached_prompt_tokens=64, model_reported="m-real"), + response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7), error=None, ) assert rec.rows[0]["cached_prompt_tokens"] == 64 assert rec.rows[0]["model_reported"] == "m-real" + assert rec.rows[0]["reasoning_tokens"] == 7 async def test_failed_attempt_has_no_provider_facts(self): rec = _MemoryRecorder() @@ -402,16 +422,19 @@ class TestEmitterObservabilityFields: ) assert rec.rows[0]["cached_prompt_tokens"] is None assert rec.rows[0]["model_reported"] is None + assert rec.rows[0]["reasoning_tokens"] is None async def test_cache_hit_replays_the_recorded_values(self): """决策 B1: 命中行原样回放,故命中率统计必须带 WHERE cache_hit = false。""" rec = _MemoryRecorder() await TelemetryEmitter(rec).emit_cache_hit( - request=_REQ, response=_resp(cached_prompt_tokens=64, model_reported="m-real") + request=_REQ, + response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7), ) row = rec.rows[0] assert row["cache_hit"] is True assert row["cached_prompt_tokens"] == 64 and row["model_reported"] == "m-real" + assert row["reasoning_tokens"] == 7 # 与 cached 同口径原样回放 async def test_terminal_failure_records_none(self): rec = _MemoryRecorder() @@ -420,6 +443,7 @@ class TestEmitterObservabilityFields: ) assert rec.rows[0]["cached_prompt_tokens"] is None assert rec.rows[0]["model_reported"] is None + assert rec.rows[0]["reasoning_tokens"] is None class TestEmitterSamplingColumn: diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index 1eae3b8..de288c1 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -54,6 +54,7 @@ class TestLLMResponse: resp = LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid") assert resp.cached_prompt_tokens is None assert resp.model_reported is None + assert resp.reasoning_tokens is None # issue #6: 本次调用未上报 filled = LLMResponse( "c", "t", @@ -68,9 +69,11 @@ class TestLLMResponse: "cid", cached_prompt_tokens=0, model_reported="MiniMax-Text-01-250321", + reasoning_tokens=0, ) assert filled.cached_prompt_tokens == 0 # 真实零命中,不得与 None 混同 assert filled.model_reported == "MiniMax-Text-01-250321" + assert filled.reasoning_tokens == 0 # 上报了且确实没推理,不得与 None 混同 def test_frozen(self): resp = LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid") @@ -247,6 +250,7 @@ class TestAuxTypes: assert s.raw["id"] == "x" # issue #3: 新字段带默认值,不填也能构造(OCR 等其他 transport 零改动) assert s.cached_prompt_tokens is None and s.model_reported is None + assert s.reasoning_tokens is None class TestOcrTypes: