diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index 3125a37..efad32e 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -61,6 +61,8 @@ class TelemetryEmitter: max_inter_token_ms=response.max_inter_token_ms if response else None, cache_hit=False, error=error, + cached_prompt_tokens=response.cached_prompt_tokens if response else None, + model_reported=response.model_reported if response else None, ) async def emit_cache_hit(self, *, request: ChatRequest, response: LLMResponse) -> None: @@ -81,6 +83,10 @@ class TelemetryEmitter: max_inter_token_ms=None, cache_hit=True, error=None, + # 决策 B1: 与 model/prompt_tokens 同一口径,原样回放历史值。 + # 统计供应商缓存命中率必须带 WHERE cache_hit = false,否则重复计数。 + cached_prompt_tokens=response.cached_prompt_tokens, + model_reported=response.model_reported, ) async def emit_terminal_failure( @@ -103,6 +109,8 @@ class TelemetryEmitter: max_inter_token_ms=None, cache_hit=False, error=error, + cached_prompt_tokens=None, + model_reported=None, ) async def _record( @@ -123,6 +131,8 @@ class TelemetryEmitter: max_inter_token_ms: float | None, cache_hit: bool, error: str | None, + cached_prompt_tokens: int | None, + model_reported: str | None, ) -> None: try: # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用); @@ -134,7 +144,9 @@ class TelemetryEmitter: # 必须排在 cache_hit 之后——缓存命中未产生新调用,0.0 是事实而非未知 cost = 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, cached_prompt_tokens + ) else: cost = None # messages 落库前多模态摘要,与缓存 key 共用同一函数(VT R12) @@ -158,6 +170,8 @@ class TelemetryEmitter: cache_hit=cache_hit, error=error, cost=cost, + cached_prompt_tokens=cached_prompt_tokens, + model_reported=model_reported, ) except asyncio.CancelledError: raise diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index f0b0638..d94b8cf 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -245,7 +245,11 @@ class StructuredOutputStrategy(Protocol): @runtime_checkable class TelemetryRecorder(Protocol): - """遥测后端;18 字段冻结(M1 设计 §4.4),唯一调用点是 TelemetryEmitter。""" + """遥测后端;20 字段冻结(M1 设计 §4.4 + issue #3),唯一调用点是 TelemetryEmitter。 + + 新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名 + Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。 + """ async def record_llm_call( self, @@ -268,4 +272,6 @@ class TelemetryRecorder(Protocol): cache_hit: bool, error: str | None, cost: float | None, + cached_prompt_tokens: int | None, + model_reported: str | None, ) -> None: ... diff --git a/src/polygateway/telemetry/postgres.py b/src/polygateway/telemetry/postgres.py index 4e7fd81..efb6469 100644 --- a/src/polygateway/telemetry/postgres.py +++ b/src/polygateway/telemetry/postgres.py @@ -6,7 +6,7 @@ ① 结构性失败(建池/建表)→ warning 一次后永久降级(池置 None 短路); ② 运行时单条写失败 → 逐条 warning 丢弃,不降级不重试(连接抖动由 asyncpg 池自恢复;避免浸泡开头一次抖动导致后续全程失遥测)。 -构造不连库(lazy),18 列 schema 与 SQLite 版同名同序。 +构造不连库(lazy),20 列 schema 与 SQLite 版同名同序。 """ from __future__ import annotations @@ -39,10 +39,18 @@ CREATE TABLE IF NOT EXISTS llm_calls ( cache_hit BOOLEAN NOT NULL DEFAULT FALSE, error TEXT, cost DOUBLE PRECISION, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + cached_prompt_tokens INTEGER, + model_reported TEXT ); """ +# 新列排在 created_at 之后: 与旧表 ALTER 追加的位置一致(见 sqlite.py 同款注释) +_BACKFILL = ( + "ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS cached_prompt_tokens INTEGER", + "ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS model_reported TEXT", +) + _COLUMNS = ( "call_id", "parent_call_id", @@ -62,6 +70,8 @@ _COLUMNS = ( "cache_hit", "error", "cost", + "cached_prompt_tokens", + "model_reported", ) _INSERT = ( @@ -104,6 +114,9 @@ class PostgresRecorder: self._pool = await asyncpg.create_pool(self._dsn, timeout=10) async with self._pool.acquire() as conn: await conn.execute(_DDL) + for statement in _BACKFILL: + # 已存在的旧表补新列(issue #3);ADD COLUMN IF NOT EXISTS 原生幂等 + await conn.execute(statement) self._schema_ready = True return self._pool except asyncio.CancelledError: diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 0b5d58a..fb992fc 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -34,10 +34,16 @@ CREATE TABLE IF NOT EXISTS llm_calls ( cache_hit INTEGER NOT NULL DEFAULT 0, error TEXT, cost REAL, - created_at TEXT NOT NULL DEFAULT (datetime('now')) + created_at TEXT NOT NULL DEFAULT (datetime('now')), + cached_prompt_tokens INTEGER, + model_reported TEXT ); """ +# 新列必须排在 created_at 之后: 旧表只能经 ALTER 追加到末尾,新建库若把它们 +# 插在前面,两条路径的物理列序会分叉(列序断言测试无合规修法)。 +_BACKFILL_COLUMNS = (("cached_prompt_tokens", "INTEGER"), ("model_reported", "TEXT")) + _COLUMNS = ( "call_id", "parent_call_id", @@ -57,6 +63,8 @@ _COLUMNS = ( "cache_hit", "error", "cost", + "cached_prompt_tokens", + "model_reported", ) _INSERT = ( @@ -82,9 +90,32 @@ class SQLiteRecorder: self._conn = conn except (OSError, sqlite3.Error) as exc: logger.warning("SQLite 遥测初始化失败,后续记录降级为 no-op: {}", exc) + self._backfill_columns() + + def _backfill_columns(self) -> None: + """给已存在的旧表补新列(issue #3);独立 try,失败只降级为逐行丢弃。 + + 必须放在 `self._conn` 赋值**之后**并先判空: 初始化失败时连接为 None, + 无守卫的补列会抛 AttributeError 逃出 `__init__`,把"静默降级"变成崩溃。 + 补列失败也绝不清空 `self._conn`——那会让整个 recorder 永久 no-op, + 比逐行丢弃严重得多。 + """ + if self._conn is None: + return + try: + existing = {row[1] for row in self._conn.execute("PRAGMA table_info(llm_calls)")} + for column, decl in _BACKFILL_COLUMNS: + if column in existing: + continue + self._conn.execute(f"ALTER TABLE llm_calls ADD COLUMN {column} {decl}") + self._conn.commit() + except sqlite3.Error as exc: + # duplicate column: 多进程共库时后到者必然撞上,属预期竞态,视为成功 + if "duplicate column" not in str(exc).lower(): + logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc) async def record_llm_call(self, **fields: object) -> None: - """写一行遥测;字段集合即 18 字段冻结签名(ports.TelemetryRecorder)。""" + """写一行遥测;字段集合即 20 字段冻结签名(ports.TelemetryRecorder)。""" if self._conn is None: return row = tuple(fields[col] for col in _COLUMNS) diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 51e9c92..be2218a 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -39,6 +39,8 @@ _EXPECTED_COLUMNS = [ "error", "cost", "created_at", + "cached_prompt_tokens", + "model_reported", ] # run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见 @@ -100,6 +102,8 @@ async def _record_minimal( "cache_hit": False, "error": None, "cost": None, + "cached_prompt_tokens": None, + "model_reported": None, } fields.update(overrides) await recorder.record_llm_call(**fields) diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index 341d6d7..cb60efd 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -114,6 +114,8 @@ class _DummyRecorder: cache_hit, error, cost, + cached_prompt_tokens, + model_reported, ) -> None: ... diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 59d7e2a..55c82d7 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -35,6 +35,8 @@ _EXPECTED_COLUMNS = [ "error", "cost", "created_at", + "cached_prompt_tokens", + "model_reported", ] @@ -93,6 +95,8 @@ async def _record_minimal(recorder, call_id="c1", **overrides): "cache_hit": False, "error": None, "cost": None, + "cached_prompt_tokens": None, + "model_reported": None, } fields.update(overrides) await recorder.record_llm_call(**fields) @@ -134,6 +138,84 @@ class TestSQLiteRecorder: await _record_minimal(recorder) # 不抛 recorder.close() + async def test_observability_columns_round_trip(self, tmp_path): + recorder = SQLiteRecorder(tmp_path / "t.db") + await _record_minimal(recorder, call_id="c-hit", cached_prompt_tokens=64) + await _record_minimal(recorder, call_id="c-zero", cached_prompt_tokens=0) + await _record_minimal(recorder, call_id="c-none", model_reported="MiniMax-Text-01") + recorder.close() + rows = dict( + sqlite3.connect(tmp_path / "t.db") + .execute("SELECT call_id, cached_prompt_tokens FROM llm_calls") + .fetchall() + ) + assert rows["c-hit"] == 64 + assert rows["c-zero"] == 0 # 真实零命中,读回仍是 0 而非 NULL + assert rows["c-none"] is None + + +class TestSQLiteColumnBackfill: + """issue #3: 已存在的 18 列旧表必须自动补列,否则每行写入都被丢弃。""" + + _LEGACY_DDL = """ + CREATE TABLE 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 REAL, + max_inter_token_ms REAL, + cache_hit INTEGER NOT NULL DEFAULT 0, + error TEXT, + cost REAL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + """ + + async def test_legacy_table_is_upgraded_in_place(self, tmp_path): + db = tmp_path / "legacy.db" + legacy = sqlite3.connect(db) + legacy.execute(self._LEGACY_DDL) + legacy.commit() + legacy.close() + + recorder = SQLiteRecorder(db) + await _record_minimal(recorder, cached_prompt_tokens=7, model_reported="m-real") + recorder.close() + + conn = sqlite3.connect(db) + cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] + assert cols == _EXPECTED_COLUMNS # ALTER 追加到末尾,与新建库列序一致 + assert conn.execute( + "SELECT cached_prompt_tokens, model_reported FROM llm_calls" + ).fetchone() == (7, "m-real") + + 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 也无从谈起——这是最坏路径。 + """ + db = tmp_path / "view.db" + conn = sqlite3.connect(db) + conn.execute("CREATE TABLE real_rows (call_id TEXT)") + conn.execute("CREATE VIEW llm_calls AS SELECT call_id FROM real_rows") + conn.commit() + conn.close() + + recorder = SQLiteRecorder(db) # 不得抛 + await _record_minimal(recorder) # 不得抛 + recorder.close() + class _MemoryRecorder: def __init__(self): @@ -143,6 +225,155 @@ class _MemoryRecorder: self.rows.append(fields) +class TestEmitterRecorderContract: + """emitter 的实参键集合必须与两个后端的 _COLUMNS 完全一致(issue #3)。 + + 两个后端的 `row = tuple(fields[col] for col in _COLUMNS)` 都在 try **之外**, + emitter 漏传一个键就抛 KeyError,被 `_record` 的 except Exception 吞成 warning + → 遥测静默全丢。而 8 个 `**fields` 形态的 fake 一个都拦不住,故显式断言。 + """ + + async def test_emitter_supplies_exactly_the_backend_columns(self): + from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS + from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS + + rec = _MemoryRecorder() + await TelemetryEmitter(rec).emit_attempt( + request=_REQ, + source=_source(), + call_id="cid-1", + latency_ms=42, + response=_resp(), + error=None, + ) + assert set(rec.rows[0]) == set(SQLITE_COLUMNS) == set(PG_COLUMNS) + + @pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"]) + async def test_every_entry_point_supplies_the_same_keys(self, emit): + from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS + + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec) + if emit == "attempt": + await emitter.emit_attempt( + request=_REQ, + source=_source(), + call_id="c", + latency_ms=1, + response=None, + error="boom", + ) + elif emit == "cache_hit": + await emitter.emit_cache_hit(request=_REQ, response=_resp()) + else: + await emitter.emit_terminal_failure( + request=_REQ, call_id="c", latency_ms=1, error="dead" + ) + assert set(rec.rows[0]) == set(SQLITE_COLUMNS) + + +class TestEmitterObservabilityFields: + """issue #3: 三个入口各自的取值口径(设计 §5 表)。""" + + async def test_attempt_carries_the_response_values(self): + rec = _MemoryRecorder() + await TelemetryEmitter(rec).emit_attempt( + request=_REQ, + source=_source(), + call_id="cid-1", + latency_ms=42, + response=_resp(cached_prompt_tokens=64, model_reported="m-real"), + error=None, + ) + assert rec.rows[0]["cached_prompt_tokens"] == 64 + assert rec.rows[0]["model_reported"] == "m-real" + + async def test_failed_attempt_has_no_provider_facts(self): + rec = _MemoryRecorder() + await TelemetryEmitter(rec).emit_attempt( + request=_REQ, + source=_source(), + call_id="cid-2", + latency_ms=7, + response=None, + error="boom", + ) + assert rec.rows[0]["cached_prompt_tokens"] is None + assert rec.rows[0]["model_reported"] 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") + ) + row = rec.rows[0] + assert row["cache_hit"] is True + assert row["cached_prompt_tokens"] == 64 and row["model_reported"] == "m-real" + + async def test_terminal_failure_records_none(self): + rec = _MemoryRecorder() + await TelemetryEmitter(rec).emit_terminal_failure( + request=_REQ, call_id="c", latency_ms=1, error="dead" + ) + assert rec.rows[0]["cached_prompt_tokens"] is None + assert rec.rows[0]["model_reported"] is None + + +class TestCostWithCachedTier: + """issue #3: 命中部分按缓存单价计费,避免 cost 系统性高估。""" + + _TABLE = PricingTable( + {"m": ModelPrice(input_per_1m=10.0, output_per_1m=20.0, cached_input_per_1m=2.0)} + ) + + async def test_cached_hit_lowers_the_recorded_cost(self): + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec, pricing=self._TABLE) + full = _resp(prompt_tokens=1_000_000, completion_tokens=0) + await emitter.emit_attempt( + request=_REQ, + source=_source(), + call_id="c1", + latency_ms=1, + response=full, + error=None, + ) + await emitter.emit_attempt( + request=_REQ, + source=_source(), + call_id="c2", + latency_ms=1, + response=_resp( + prompt_tokens=1_000_000, completion_tokens=0, cached_prompt_tokens=600_000 + ), + error=None, + ) + assert rec.rows[0]["cost"] == pytest.approx(10.0) + assert rec.rows[1]["cost"] == pytest.approx(5.2) # 400k×10 + 600k×2 + + async def test_cache_hit_row_still_costs_zero(self): + """缓存命中未产生新调用 → cost 恒 0.0,该短路必须排在任何换算之前。""" + rec = _MemoryRecorder() + await TelemetryEmitter(rec, pricing=self._TABLE).emit_cache_hit( + request=_REQ, + response=_resp(prompt_tokens=1_000_000, cached_prompt_tokens=600_000), + ) + assert rec.rows[0]["cost"] == 0.0 + + async def test_unavailable_usage_still_costs_none(self): + rec = _MemoryRecorder() + await TelemetryEmitter(rec, pricing=self._TABLE).emit_attempt( + request=_REQ, + source=_source(), + call_id="c", + latency_ms=1, + response=_resp(usage_source="unavailable", cached_prompt_tokens=5), + error=None, + ) + assert rec.rows[0]["cost"] is None + + class TestEmitter: async def test_attempt_success_row(self): rec = _MemoryRecorder()