iomgaa/PolyGateway
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 9 Wiki Activity

feat: record the observability fields end to end through telemetry

Browse Source
This commit is contained in:
iomgaa
2026-07-31 08:03:43 -04:00
parent 0ed9dc107c
commit c2fcd5b1f8
7 changed files with 307 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/polygateway/middleware/telemetry.py
+15 -1
View File
@@ -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
src/polygateway/ports.py
+7 -1
View File
@@ -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: ...
src/polygateway/telemetry/postgres.py
+15 -2
View File
@@ -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:
src/polygateway/telemetry/sqlite.py
+33 -2
View File
@@ -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)
tests/integration/test_postgres_telemetry.py
+4
View File
@@ -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)
tests/unit/test_ports.py
+2
View File
@@ -114,6 +114,8 @@ class _DummyRecorder:
cache_hit,
error,
cost,
cached_prompt_tokens,
model_reported,
) -> None: ...
tests/unit/test_telemetry.py
+231
View File
@@ -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()