diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py new file mode 100644 index 0000000..aeda520 --- /dev/null +++ b/src/polygateway/middleware/telemetry.py @@ -0,0 +1,182 @@ +"""TelemetryMW + TelemetryEmitter: 遥测调用点收敛为单一 helper(铁律)。 + +Emitter 是全库**唯一**调用 `record_llm_call` 的地方(三项目 4 处逐字复制 +15 参调用的教训)。分工: RetryMW 经 Emitter 逐次记录每次尝试;TelemetryMW +(最外层)只记尝试层看不见的事件——缓存命中、scope 级失败、取消; +RequestRejected/ResultInvalid 已被尝试层记录,最外层放行不重复记。 +""" + +from __future__ import annotations + +import asyncio +import json +import time +import uuid +from typing import TYPE_CHECKING + +from loguru import logger + +from polygateway.errors import GatewayUnavailableError, GovernanceBackendError +from polygateway.middleware.cache import digest_messages + +if TYPE_CHECKING: + from collections.abc import Callable + + from polygateway.ports import CallNext, TelemetryRecorder + from polygateway.types import ChatRequest, LLMResponse, SourceConfig + + +class TelemetryEmitter: + """从请求与结果组装 18 字段并写入 recorder;一切写失败降级 warning。""" + + def __init__(self, recorder: TelemetryRecorder) -> None: + self._recorder = recorder + + async def emit_attempt( + self, + *, + request: ChatRequest, + source: SourceConfig, + call_id: str, + latency_ms: int, + response: LLMResponse | None, + error: str | None, + ) -> None: + """逐次尝试记录(RetryMW 调用);失败尝试 usage 按 estimated 记 0。""" + await self._record( + request=request, + call_id=call_id, + model=source.model, + provider=source.provider, + source_name=source.name, + response_text=response.content if response else "", + thinking=response.thinking if response else "", + prompt_tokens=response.prompt_tokens if response else 0, + completion_tokens=response.completion_tokens if response else 0, + usage_source=response.usage_source if response else "estimated", + latency_ms=latency_ms, + ttft_ms=response.ttft_ms if response else None, + max_inter_token_ms=response.max_inter_token_ms if response else None, + cache_hit=False, + error=error, + ) + + async def emit_cache_hit(self, *, request: ChatRequest, response: LLMResponse) -> None: + """缓存命中记录: cache_hit=True、latency_ms=0(VT 同款)。""" + await self._record( + request=request, + call_id=response.call_id, + model=response.model, + provider=response.provider, + source_name=response.source_name, + response_text=response.content, + thinking=response.thinking, + prompt_tokens=response.prompt_tokens, + completion_tokens=response.completion_tokens, + usage_source=response.usage_source, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=True, + error=None, + ) + + async def emit_terminal_failure( + self, *, request: ChatRequest, call_id: str, latency_ms: int, error: str + ) -> None: + """scope 级失败/取消记录: 无具体源,溯源字段置空标记。""" + await self._record( + request=request, + call_id=call_id, + model="", + provider="", + source_name="", + response_text="", + thinking="", + prompt_tokens=0, + completion_tokens=0, + usage_source="estimated", + latency_ms=latency_ms, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + error=error, + ) + + async def _record( + self, + *, + request: ChatRequest, + call_id: str, + model: str, + provider: str, + source_name: str, + response_text: str, + thinking: str, + prompt_tokens: int, + completion_tokens: int, + usage_source: str, + latency_ms: int, + ttft_ms: float | None, + max_inter_token_ms: float | None, + cache_hit: bool, + error: str | None, + ) -> None: + try: + # messages 落库前多模态摘要,与缓存 key 共用同一函数(VT R12) + messages_json = json.dumps(digest_messages(request.messages), ensure_ascii=False) + await self._recorder.record_llm_call( + call_id=call_id, + parent_call_id=request.parent_call_id, + session_id=request.session_id, + model=model, + provider=provider, + source_name=source_name, + messages=messages_json, + response=response_text, + thinking=thinking, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + usage_source=usage_source, + latency_ms=latency_ms, + ttft_ms=ttft_ms, + max_inter_token_ms=max_inter_token_ms, + cache_hit=cache_hit, + error=error, + cost=None, # M1 无 pricing;M2 换算后填充 + ) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning("遥测记录失败(降级不冒泡): {}", exc) + + +class TelemetryMW: + """洋葱最外层: 观测尝试层看不见的路径,任何路径都留痕(遥测必录)。""" + + def __init__( + self, emitter: TelemetryEmitter, now: Callable[[], float] = time.monotonic + ) -> None: + self._emitter = emitter + self._now = now + + async def __call__(self, request: ChatRequest, call_next: CallNext) -> LLMResponse: + started = self._now() + try: + response = await call_next(request) + except (GatewayUnavailableError, GovernanceBackendError) as exc: + await self._emitter.emit_terminal_failure( + request=request, call_id=str(uuid.uuid4()), + latency_ms=int((self._now() - started) * 1000), error=str(exc), + ) + raise + except asyncio.CancelledError: + # 尽力而为: 取消也留痕(§5.1 约定④);随后立即重抛 + await self._emitter.emit_terminal_failure( + request=request, call_id=str(uuid.uuid4()), + latency_ms=int((self._now() - started) * 1000), error="cancelled", + ) + raise + if response.cache_hit: + await self._emitter.emit_cache_hit(request=request, response=response) + return response diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py new file mode 100644 index 0000000..802b7ab --- /dev/null +++ b/src/polygateway/telemetry/sqlite.py @@ -0,0 +1,93 @@ +"""SQLite 遥测后端(默认): WAL + 单持久连接 + to_thread 桥接。 + +蓝本 VT `adapters/telemetry.py`: 构造期建连接与表,失败降级为 no-op +(记录基础设施不得拖垮业务调用);`INSERT OR IGNORE` 幂等(call_id 主键); +写入经 threading.Lock 串行化后由 `asyncio.to_thread` 执行,不阻塞事件循环。 +""" + +from __future__ import annotations + +import asyncio +import sqlite3 +import threading +from pathlib import Path + +from loguru import logger + +_DDL = """ +CREATE TABLE IF NOT EXISTS 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')) +); +""" + +_COLUMNS = ( + "call_id", "parent_call_id", "session_id", "model", "provider", "source_name", + "messages", "response", "thinking", "prompt_tokens", "completion_tokens", + "usage_source", "latency_ms", "ttft_ms", "max_inter_token_ms", "cache_hit", + "error", "cost", +) + +_INSERT = ( + f"INSERT OR IGNORE INTO llm_calls ({', '.join(_COLUMNS)}) " + f"VALUES ({', '.join('?' for _ in _COLUMNS)})" +) + + +class SQLiteRecorder: + """TelemetryRecorder 端口的 SQLite 实现;初始化/写入失败全降级 warning。""" + + def __init__(self, db_path: Path | str) -> None: + self._lock = threading.Lock() + self._conn: sqlite3.Connection | None = None + try: + path = Path(db_path) + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path, check_same_thread=False, timeout=10.0) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + conn.execute(_DDL) + conn.commit() + self._conn = conn + except (OSError, sqlite3.Error) as exc: + logger.warning("SQLite 遥测初始化失败,后续记录降级为 no-op: {}", exc) + + async def record_llm_call(self, **fields: object) -> None: + """写一行遥测;字段集合即 18 字段冻结签名(ports.TelemetryRecorder)。""" + if self._conn is None: + return + row = tuple(fields[col] for col in _COLUMNS) + try: + await asyncio.to_thread(self._write, row) + except (OSError, sqlite3.Error) as exc: + logger.warning("SQLite 遥测写入失败(降级不冒泡): {}", exc) + + def _write(self, row: tuple) -> None: + assert self._conn is not None # 内部不变量: 调用方已判空 + with self._lock: + self._conn.execute(_INSERT, row) + self._conn.commit() + + def close(self) -> None: + """幂等关闭持久连接。""" + conn, self._conn = self._conn, None + if conn is not None: + with self._lock: + conn.close() diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py new file mode 100644 index 0000000..196394d --- /dev/null +++ b/tests/unit/test_telemetry.py @@ -0,0 +1,203 @@ +"""遥测子系统测试: SQLiteRecorder(18 列)+ TelemetryEmitter 单一 helper + TelemetryMW。""" + +import asyncio +import sqlite3 +import subprocess +from pathlib import Path + +import pytest + +from polygateway.errors import CircuitOpenError, RequestRejectedError +from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW +from polygateway.telemetry.sqlite import SQLiteRecorder +from polygateway.types import ChatRequest, LLMResponse, SourceConfig + +_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1") + +_EXPECTED_COLUMNS = [ + "call_id", "parent_call_id", "session_id", "model", "provider", "source_name", + "messages", "response", "thinking", "prompt_tokens", "completion_tokens", + "usage_source", "latency_ms", "ttft_ms", "max_inter_token_ms", "cache_hit", + "error", "cost", "created_at", +] + + +def _resp(**overrides): + base = dict( + content="ok", thinking="", model="m", provider="p", prompt_tokens=1, + completion_tokens=2, latency_ms=30, ttft_ms=None, max_inter_token_ms=None, + cache_hit=False, call_id="cid-1", source_name="s1", usage_source="measured", + ) + base.update(overrides) + return LLMResponse(**base) + + +def _source(): + return SourceConfig( + name="s1", provider="p", base_url="https://gw.example/v1", + api_key="sk", model="m", timeout_s=10.0, + ) + + +async def _record_minimal(recorder, call_id="c1", **overrides): + fields = dict( + call_id=call_id, parent_call_id=None, session_id="sess-1", model="m", + provider="p", source_name="s1", messages="[]", response="ok", thinking="", + prompt_tokens=1, completion_tokens=2, usage_source="measured", latency_ms=10, + ttft_ms=None, max_inter_token_ms=None, cache_hit=False, error=None, cost=None, + ) + fields.update(overrides) + await recorder.record_llm_call(**fields) + + +class TestSQLiteRecorder: + async def test_schema_has_frozen_columns(self, tmp_path): + recorder = SQLiteRecorder(tmp_path / "t.db") + await _record_minimal(recorder) + recorder.close() + cols = [r[1] for r in sqlite3.connect(tmp_path / "t.db").execute( + "PRAGMA table_info(llm_calls)" + )] + assert cols == _EXPECTED_COLUMNS + + async def test_call_id_idempotent(self, tmp_path): + recorder = SQLiteRecorder(tmp_path / "t.db") + await _record_minimal(recorder, call_id="dup") + await _record_minimal(recorder, call_id="dup", response="second") + recorder.close() + rows = sqlite3.connect(tmp_path / "t.db").execute( + "SELECT response FROM llm_calls WHERE call_id='dup'" + ).fetchall() + assert rows == [("ok",)] # INSERT OR IGNORE: 第二次静默忽略 + + async def test_concurrent_writes_all_land(self, tmp_path): + recorder = SQLiteRecorder(tmp_path / "t.db") + await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50))) + recorder.close() + (count,) = sqlite3.connect(tmp_path / "t.db").execute( + "SELECT COUNT(*) FROM llm_calls" + ).fetchone() + assert count == 50 + + async def test_unwritable_path_degrades_silently(self): + recorder = SQLiteRecorder(Path("/nonexistent-root/deep/t.db")) + await _record_minimal(recorder) # 不抛 + recorder.close() + + +class _MemoryRecorder: + def __init__(self): + self.rows = [] + + async def record_llm_call(self, **fields): + self.rows.append(fields) + + +class TestEmitter: + async def test_attempt_success_row(self): + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec) + await emitter.emit_attempt( + request=_REQ, source=_source(), call_id="cid-1", latency_ms=42, + response=_resp(), error=None, + ) + row = rec.rows[0] + assert row["call_id"] == "cid-1" and row["error"] is None + assert row["session_id"] == "sess-1" and row["source_name"] == "s1" + assert row["response"] == "ok" and row["cost"] is None + + async def test_attempt_failure_row(self): + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec) + await emitter.emit_attempt( + request=_REQ, source=_source(), call_id="cid-2", latency_ms=7, + response=None, error="TransientError: boom", + ) + row = rec.rows[0] + assert row["error"].startswith("TransientError") + assert row["response"] == "" and row["usage_source"] == "estimated" + + async def test_multimodal_messages_digested_before_storage(self): + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec) + big = "data:image/png;base64," + "A" * 100_000 + req = ChatRequest(messages=[{"role": "user", "content": [ + {"type": "image_url", "image_url": {"url": big}}, + ]}]) + await emitter.emit_attempt( + request=req, source=_source(), call_id="c", latency_ms=1, + response=None, error="x", + ) + assert len(rec.rows[0]["messages"]) < 500 # base64 不整段进库(VT R12) + + async def test_recorder_failure_swallowed(self): + class Broken: + async def record_llm_call(self, **fields): + raise OSError("disk full") + + emitter = TelemetryEmitter(Broken()) + await emitter.emit_attempt( + request=_REQ, source=_source(), call_id="c", latency_ms=1, + response=_resp(), error=None, + ) # 不抛(降级不冒泡) + + +class TestTelemetryMW: + async def test_cache_hit_recorded(self): + rec = _MemoryRecorder() + mw = TelemetryMW(TelemetryEmitter(rec)) + + async def terminal(request): + return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid") + + resp = await mw(_REQ, terminal) + assert resp.cache_hit + assert len(rec.rows) == 1 + assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["latency_ms"] == 0 + + async def test_normal_success_not_double_recorded(self): + """成功尝试由 RetryMW 逐次记录;最外层不得重复记。""" + rec = _MemoryRecorder() + mw = TelemetryMW(TelemetryEmitter(rec)) + + async def terminal(request): + return _resp(cache_hit=False) + + await mw(_REQ, terminal) + assert rec.rows == [] + + async def test_scope_level_failure_recorded(self): + rec = _MemoryRecorder() + mw = TelemetryMW(TelemetryEmitter(rec)) + + async def terminal(request): + raise CircuitOpenError(scope="llm", retry_after_s=30.0) + + with pytest.raises(CircuitOpenError): + await mw(_REQ, terminal) + assert len(rec.rows) == 1 and "circuit_open" in rec.rows[0]["error"] + + async def test_attempt_level_failure_not_double_recorded(self): + """RequestRejected 已被 RetryMW 逐次记录 → 最外层跳过。""" + rec = _MemoryRecorder() + mw = TelemetryMW(TelemetryEmitter(rec)) + + async def terminal(request): + raise RequestRejectedError("400") + + with pytest.raises(RequestRejectedError): + await mw(_REQ, terminal) + assert rec.rows == [] + + +def test_single_emitter_discipline(): + """铁律执法: record_llm_call 在 src/ 的调用点只允许出现在 telemetry emitter。""" + out = subprocess.run( + ["grep", "-rln", "record_llm_call(", "src/polygateway"], + capture_output=True, text=True, cwd=Path(__file__).resolve().parents[2], + ).stdout.splitlines() + callers = [ + p for p in out + if not p.endswith(("ports.py", "telemetry/sqlite.py")) + ] + assert callers == ["src/polygateway/middleware/telemetry.py"]