From 32d7869043cf9dd14f16454835b4f9c6ad06ad08 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 31 Jul 2026 08:28:41 -0400 Subject: [PATCH] fix: harden the observability fields against the verifier findings --- research-wiki/index.md | 4 +- research-wiki/log.md | 1 + research-wiki/schemas/llm-calls.md | 19 +++- src/polygateway/pricing.py | 3 +- src/polygateway/transports/openai_compat.py | 9 +- tests/integration/test_postgres_telemetry.py | 99 ++++++++++++++++++++ tests/unit/test_openai_compat.py | 13 +++ tests/unit/test_pricing.py | 4 + tests/unit/test_telemetry.py | 6 +- 9 files changed, 148 insertions(+), 10 deletions(-) diff --git a/research-wiki/index.md b/research-wiki/index.md index 5be9bbf..f3ea093 100644 --- a/research-wiki/index.md +++ b/research-wiki/index.md @@ -1,6 +1,6 @@ # Research Wiki 索引 -> 自动生成，更新时间：2026-07-31 11:11 UTC +> 自动生成，更新时间：2026-07-31 12:25 UTC ## design (18) - [2026-07-20-m1-core-design](designs/2026-07-20-m1-core-design.md) `design:2026-07-20-m1-core-design` @@ -52,7 +52,7 @@ - [响应可观测字段扩展实现计划](plans/response-observability-fields.md) `plan:response-observability-fields` ## schema (1) -- [表结构: llm_calls(遥测 18 字段)](schemas/llm-calls.md) `schema:llm-calls` +- [表结构: llm_calls(遥测 20 字段)](schemas/llm-calls.md) `schema:llm-calls` ## metric (2) - [OCR 治理调用成功率与错误分类分布](metrics/ocr-call-success.md) `metric:ocr-call-success` diff --git a/research-wiki/log.md b/research-wiki/log.md index 21209e7..50cc40f 100644 --- a/research-wiki/log.md +++ b/research-wiki/log.md @@ -63,3 +63,4 @@ - [2026-07-31 11:10 UTC] 新增边: plan:response-observability-fields --implements--> design:response-observability-fields - [2026-07-31 11:10 UTC] 重建索引: 46 篇页面 - [2026-07-31 11:11 UTC] 重建索引: 46 篇页面 +- [2026-07-31 12:25 UTC] 重建索引: 46 篇页面 diff --git a/research-wiki/schemas/llm-calls.md b/research-wiki/schemas/llm-calls.md index 6bd2ad6..2c0c1ea 100644 --- a/research-wiki/schemas/llm-calls.md +++ b/research-wiki/schemas/llm-calls.md @@ -1,11 +1,11 @@ --- type: schema node_id: schema:llm-calls -title: "表结构: llm_calls(遥测 18 字段)" +title: "表结构: llm_calls(遥测 20 字段)" date: 2026-07-20 --- -# 表结构: llm_calls(遥测 18 字段) +# 表结构: llm_calls(遥测 20 字段) ## 列定义(冻结,M1 设计 §4.4 / ARCH §7.8) @@ -24,6 +24,8 @@ date: 2026-07-20 | error | TEXT | 异常信息;取消记 "cancelled" | | cost | REAL | M2 起 pricing 换算;`usage_source='unavailable'` 的真实调用行为 NULL(缓存命中行例外,仍为 0.0) | | created_at | TEXT NOT NULL DEFAULT (datetime('now')) | 落库时刻 | +| cached_prompt_tokens | INTEGER | 供应商 prompt cache 命中的输入 token(2026-07-31,issue #3);NULL = 该源未上报,`0` = 上报了真实零命中,两者不可混同 | +| model_reported | TEXT | API 响应体实际返回的 model;NULL = 未上报。与 `model`(配置别名)可能分叉 | ## usage/成本口径(2026-07-30,est_tokens 解耦) @@ -35,6 +37,19 @@ date: 2026-07-20 `SUM(cost)` 天然跳过 NULL,故账单汇总不再被虚构的估值污染;账目缺口的度量口径固定为 `WHERE usage_source = 'unavailable' AND cache_hit = false`。**`cache_hit` 限定不可省**:缓存命中行未产生新调用,cost 是事实上的 `0.0` 而非未知,本无账目缺口,漏掉该条件会让缺口度量偏高。 +## 供应商 prompt cache 口径(2026-07-31,issue #3) + +新增两列排在 `created_at` **之后**——旧表只能经 `ALTER TABLE ADD COLUMN` 追加到末尾,DDL 里若插在前面,新建库与升级库的物理列序会分叉(列序断言无合规修法)。两个后端在初始化期幂等补列:`CREATE TABLE IF NOT EXISTS` 不会给旧表加列,不补则每行写入被逐行 warning 丢弃、遥测静默全失;补列失败只降级为逐行丢弃,绝不让 recorder 整体失能。 + +`cache_hit` 指 **PolyGateway 自身响应缓存**,与供应商 prompt cache 是两回事。缓存命中行的这两列是**原样回放**的历史值(与 `model`/`prompt_tokens` 同一口径),故命中率度量口径固定为: + +```sql +SELECT SUM(cached_prompt_tokens)::float / NULLIF(SUM(prompt_tokens), 0) +FROM llm_calls WHERE cache_hit = false AND cached_prompt_tokens IS NOT NULL; +``` + +`WHERE cache_hit = false` 不可省,理由与上面 cost 缺口口径同源:回放行计入即重复计数。 + ## 埋点位置(单一 helper 铁律) - `middleware/telemetry.py::TelemetryEmitter` 是全库**唯一** `record_llm_call` 调用点; diff --git a/src/polygateway/pricing.py b/src/polygateway/pricing.py index 08d5568..0b4d249 100644 --- a/src/polygateway/pricing.py +++ b/src/polygateway/pricing.py @@ -99,7 +99,8 @@ class PricingTable: logger.warning("pricing 表无 model {!r} 的单价,cost 记 None", model) return None billed_input = prompt_tokens / 1_000_000 * price.input_per_1m - if price.cached_input_per_1m is not None and cached_prompt_tokens: + # 负数按"无命中"处理: cost() 是公共方法,不能假定调用方已过 transport 的校验 + if price.cached_input_per_1m is not None and (cached_prompt_tokens or 0) > 0: cached = self._clamp_cached(model, prompt_tokens, cached_prompt_tokens) billed_input = (prompt_tokens - cached) / 1_000_000 * price.input_per_1m + ( cached / 1_000_000 * price.cached_input_per_1m diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index 117a069..7703f54 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -45,9 +45,12 @@ def _sse_delta(chunk: dict[str, Any], usage_sink: dict[str, Any]) -> tuple[bool, """从 chunk 提取增量: (True, content) 或 (False, reasoning);usage 帧旁路进 sink。""" if chunk.get("usage"): usage_sink["usage"] = chunk["usage"] - if "model" not in usage_sink and chunk.get("model") is not None: - # 首次写入即固定: 末帧的异常值不得覆盖首帧报的真实版本(issue #3) - usage_sink["model"] = chunk["model"] + if "model" not in usage_sink: + # 首个**有效**值即固定: 末帧的异常值不得覆盖它;但首帧报空串也不能锁死 + # sink——否则后续真实版本会丢(issue #3) + reported = _coerce_model_reported(chunk.get("model")) + if reported is not None: + usage_sink["model"] = reported choices = chunk.get("choices") or [] if not choices: return None diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index be2218a..49d6981 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -119,6 +119,105 @@ async def _fetch(dsn: str, sql: str, *args): await conn.close() +_LEGACY_DDL = """ +CREATE TABLE {schema}.llm_calls ( + call_id TEXT PRIMARY KEY, + parent_call_id TEXT, + session_id TEXT, + model TEXT NOT NULL, + provider TEXT NOT NULL, + source_name TEXT NOT NULL, + messages TEXT NOT NULL, + response TEXT NOT NULL, + thinking TEXT NOT NULL DEFAULT '', + prompt_tokens INTEGER NOT NULL, + completion_tokens INTEGER NOT NULL, + usage_source TEXT NOT NULL, + latency_ms INTEGER NOT NULL, + ttft_ms DOUBLE PRECISION, + max_inter_token_ms DOUBLE PRECISION, + cache_hit BOOLEAN NOT NULL DEFAULT FALSE, + error TEXT, + cost DOUBLE PRECISION, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +) +""" + + +@pytest.fixture +async def legacy_schema(dsn): + """在**自建的临时 schema** 里造一张 18 列旧表,验证补列(issue #3)。 + + 绝不碰共享的 public.llm_calls: 用 search_path 把 recorder 指向临时 schema, + teardown 只 DROP 自己建的 schema。 + """ + import asyncpg + + name = f"pgwtest_{uuid4().hex[:8]}" + conn = await asyncpg.connect(dsn, timeout=10) + try: + await conn.execute(f"CREATE SCHEMA {name}") + await conn.execute(_LEGACY_DDL.format(schema=name)) + finally: + await conn.close() + sep = "&" if "?" in dsn else "?" + yield f"{dsn}{sep}options=-csearch_path%3D{name}", name + conn = await asyncpg.connect(dsn, timeout=10) + try: + await conn.execute(f"DROP SCHEMA {name} CASCADE") + finally: + await conn.close() + + +class TestObservabilityColumns: + """issue #3: 两列写入可回读,且已存在的 18 列旧表会被自动补列。""" + + async def test_values_round_trip(self, dsn): + recorder = PostgresRecorder(dsn) + try: + await _record_minimal(recorder, call_id=_cid("hit"), cached_prompt_tokens=64) + await _record_minimal(recorder, call_id=_cid("zero"), cached_prompt_tokens=0) + await _record_minimal(recorder, call_id=_cid("model"), model_reported="MiniMax-01") + rows = await _fetch( + dsn, + "SELECT call_id, cached_prompt_tokens, model_reported FROM llm_calls " + "WHERE call_id LIKE $1", + f"{_RUN_PREFIX}-%", + ) + by_id = {r["call_id"]: r for r in rows} + assert by_id[_cid("hit")]["cached_prompt_tokens"] == 64 + assert by_id[_cid("zero")]["cached_prompt_tokens"] == 0 # 真实零命中 ≠ NULL + assert by_id[_cid("model")]["cached_prompt_tokens"] is None + assert by_id[_cid("model")]["model_reported"] == "MiniMax-01" + finally: + await recorder.aclose() + + async def test_legacy_table_is_upgraded_in_place(self, legacy_schema): + """18 列旧表不补列的话,每行写入都会被逐行 warning 丢弃(遥测静默全失)。""" + schema_dsn, schema = legacy_schema + recorder = PostgresRecorder(schema_dsn) + try: + await _record_minimal( + recorder, call_id=_cid("legacy"), cached_prompt_tokens=7, model_reported="m-real" + ) + cols = await _fetch( + schema_dsn, + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", + schema, + ) + # ALTER 只能追加到末尾: 与新建库的列序一致才不会分叉 + assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS + rows = await _fetch( + schema_dsn, + "SELECT cached_prompt_tokens, model_reported FROM llm_calls WHERE call_id = $1", + _cid("legacy"), + ) + assert (rows[0]["cached_prompt_tokens"], rows[0]["model_reported"]) == (7, "m-real") + finally: + await recorder.aclose() + + class TestSchema: async def test_schema_has_frozen_columns_in_order(self, dsn): recorder = PostgresRecorder(dsn) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index f76b046..65c5f48 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -360,6 +360,19 @@ class TestObservabilityFields: result = await _complete(_transport_for(handler), _source()) assert result.model_reported == "MiniMax-Text-01-250321" + async def test_empty_first_model_does_not_block_a_later_real_one(self): + """首帧报空串不得锁死 sink: 守卫按"有效值"判断,否则真实版本会丢。""" + + def handler(request): + return _sse_stream( + _chunk(content="a", model=""), + _chunk(content="b", model="MiniMax-Text-01-250321"), + _chunk(usage=_USAGE), + ) + + result = await _complete(_transport_for(handler), _source()) + assert result.model_reported == "MiniMax-Text-01-250321" + @pytest.mark.parametrize("bad", [None, "", " ", 123, {}]) async def test_missing_or_malformed_model_is_none(self, bad): def handler(request): diff --git a/tests/unit/test_pricing.py b/tests/unit/test_pricing.py index 19a388e..b4bb24d 100644 --- a/tests/unit/test_pricing.py +++ b/tests/unit/test_pricing.py @@ -77,6 +77,10 @@ class TestCachedInputTier: def test_no_hit_is_billed_in_full(self, cached): assert self._CACHED.cost("m", 1_000_000, 0, cached) == pytest.approx(10.0) + def test_negative_cached_is_billed_in_full(self): + """负数命中数不得抬高成本: cost() 是公共方法,外部输入须校验后使用(P5)。""" + assert self._CACHED.cost("m", 1_000_000, 0, -500_000) == pytest.approx(10.0) + def test_cached_over_prompt_is_clamped_and_never_negative(self): """网关口径异常时按输入总数夹取: 全部按缓存价,不得算出负成本。""" clamped = self._CACHED.cost("m", 1_000_000, 0, 5_000_000) diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index eb98b3e..390c606 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -202,8 +202,9 @@ class TestSQLiteColumnBackfill: async def test_backfill_failure_keeps_the_recorder_usable(self, tmp_path): """补列失败只能逐行降级,绝不能把 recorder 整体变成 no-op(设计 D1 纪律)。 - 把 llm_calls 建成 view: 表不存在故 CREATE TABLE IF NOT EXISTS 会撞名失败, - ALTER 也无从谈起——这是最坏路径。 + 把 llm_calls 建成同名 view: `CREATE TABLE IF NOT EXISTS` 遇 view 静默 + no-op(不抛),随后的 ALTER 才抛 "Cannot add a column to a view"——正是 + 补列失败这条分支。`_conn` 必须保持非 None,否则整个 recorder 永久失能。 """ db = tmp_path / "view.db" conn = sqlite3.connect(db) @@ -213,6 +214,7 @@ class TestSQLiteColumnBackfill: conn.close() recorder = SQLiteRecorder(db) # 不得抛 + assert recorder._conn is not None # 补列失败 ≠ recorder 失能(D1 纪律) await _record_minimal(recorder) # 不得抛 recorder.close()