diff --git a/.env.example b/.env.example index 08ca2b9..7a45996 100644 --- a/.env.example +++ b/.env.example @@ -66,6 +66,13 @@ PGW_TELEMETRY_BACKEND=none # sqlite | postgres | none(必填) # # sqlite 则是下游自己的本地文件(runs/*.db):没有 DBA、没有迁移工具、 # # 没有第二个系统碰它,ALTER 是毫秒级元数据操作,强加手工 SQL 步骤是净损失。 # PGW_TELEMETRY_PG_DSN=postgresql://user:pass@host:5432/polygateway # postgres 时必填;严禁指向在用业务库(实验室约定: 专用库 polygateway) +# PGW_TELEMETRY_TEXT_CAP=2000 # 遥测落库正文的字符上限,须 > 0;**不设 = 不截断**(缺省,逐字节留全文)。 +# # 作用于 messages 的每条文本 content、多模态 text part、response 与 thinking; +# # 超出部分头部保留、尾部换成 `…(略 N 字)`。多模态 image_url 的 sha256 摘要不受影响。 +# # 缺省为何是"不截断": 遥测被下游当**审计证据**用——出了问题要回答"当时到底发了什么", +# # 也要能拿原样的请求复现与重放;截断后这两件事都做不成,而既有下游正依赖这一行为。 +# # 反面同样要看清: 不截断意味着客户合同、标书全文无限期留在 llm_calls 里, +# # 多租户下还混在同一张表。真在意留存面的部署应显式设一个上限,并配保留期与访问控制。 # PGW_PRICING_PATH=config/prices.json # 可选: {"": {"input_per_1m": x, "output_per_1m": y}};缺省 cost 恒 None # # 可选第三档 "cached_input_per_1m": z —— 供应商 prompt cache 命中部分的单价; # # 不填即命中部分也按 input 全额计(库不猜折扣率),cost 会偏高 diff --git a/src/polygateway/client.py b/src/polygateway/client.py index 7ba53ab..1bfbd0b 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -317,6 +317,7 @@ class GatewayClient: pricing=PricingTable.from_file(settings.pricing_path) if settings.pricing_path is not None else None, + text_cap=settings.telemetry_text_cap, cache=cache if cache is not None else _build_cache(settings), cache_namespace=settings.cache_namespace, cache_ttl_s=settings.cache_ttl_s, diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 7ab8e8b..8dfe5b2 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -58,6 +58,8 @@ _TELEMETRY_BACKENDS = frozenset({"sqlite", "postgres", "none"}) # 遥测 schema 档位(issue #13): auto 允许 recorder 给旧表 ALTER 补列,manual 不发 DDL _SCHEMA_MODES = frozenset({"auto", "manual"}) _SCHEMA_MODE_KEY = "PGW_TELEMETRY_SCHEMA_MODE" +# 遥测正文字符上限(issue #12);二态键,未设 = 不截断 +_TEXT_CAP_KEY = "PGW_TELEMETRY_TEXT_CAP" _REDIS_DEPENDENT_BACKENDS = ("limiter_backend", "breaker_backend", "cache_backend") # 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源) _DEFAULT_STALL_WINDOW_S = 300.0 @@ -139,6 +141,11 @@ class GatewaySettings: # backend=none 时恒 False 这条跨字段不变量则由 `_validate_telemetry` # 把关,对直接构造与 `dataclasses.replace` 同样生效 telemetry_auto_migrate: bool + # 遥测落库正文的字符上限(issue #12);None = 不截断,与本字段出现之前逐字节相同。 + # 缺省不截断是人类决策: 截断后的遥测不再是审计证据、也无法用于复现与重放,而 + # 既有下游正依赖这一行为。值域(> 0)由 `_validate_telemetry` 把关,直接构造、 + # `dataclasses.replace` 与 env 三条路一并覆盖 + telemetry_text_cap: int | None redis_url: str | None pricing_path: str | None structured_max_retries: int @@ -224,7 +231,16 @@ class GatewaySettings: recorder 消费它,True 是个自相矛盾却无害的状态。`from_env` 那条路的派生 已经给出 False,归一化是为了直接构造与 `dataclasses.replace` 也一致—— 不变量挂在构造期,才不用每加一个装配工厂就多一处要同步。 + + `telemetry_text_cap` 的值域则是**报错**而非归一化: 0 与负数都不是"不截断" + 的写法(不截断写 None),把它们悄悄改成 None 等于用默认值掩盖调用方的错误。 + 报错文本同时点出字段名与 env 键名,两条装配路的调用方各看得懂自己那套。 """ + if self.telemetry_text_cap is not None and self.telemetry_text_cap <= 0: + raise ValueError( + f"telemetry_text_cap({_TEXT_CAP_KEY})必须 > 0: {self.telemetry_text_cap};" + "不截断请不设该键(None),0 只会让每条正文退化成一个省略标记" + ) if self.telemetry_backend == "none" and self.telemetry_auto_migrate: object.__setattr__(self, "telemetry_auto_migrate", False) if self.telemetry_backend == "sqlite" and not self.telemetry_sqlite_path: @@ -461,6 +477,7 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: else None, "telemetry_pg_dsn": _load_pg_dsn(env) if telemetry_backend == "postgres" else None, "telemetry_auto_migrate": auto_migrate, + "telemetry_text_cap": _load_text_cap(env), "redis_url": redis_url, "pricing_path": env.get("PGW_PRICING_PATH") or None, "structured_max_retries": _load_structured_retries(env), @@ -496,6 +513,28 @@ def _load_schema_mode(env: Mapping[str, str], telemetry_backend: str) -> bool: return _load_choice(env, _SCHEMA_MODE_KEY, _SCHEMA_MODES, "auto") == "auto" +def _load_text_cap(env: Mapping[str, str]) -> int | None: + """读 `PGW_TELEMETRY_TEXT_CAP`(issue #12);键未设即 None = 不截断。 + + 与相邻的 `PGW_TELEMETRY_SCHEMA_MODE` 不同,这个键是**二态**而非三态: + "未设"本身就是最终答案(不截断),没有需要按后端派生的第二种缺省,故不必像 + 那边一样先探"设没设"再分两条路取值,读到什么解什么即可。 + + 值域(> 0)刻意不在此处判: 构造期守卫那道同时覆盖直接构造与 + `dataclasses.replace`,而报错文本已点出本键名,env 路的调用方不会看丢。 + + Args: + env: 已合并的环境映射。 + + Returns: + 遥测正文的字符上限;键未设或为空串时返回 None(不截断)。 + """ + found = _first(env, _TEXT_CAP_KEY) + if found is None: + return None + return int(_cast(found[1], "int", found[0])) + + def _strip_dsn_driver(dsn: str) -> str: """剥 SQLAlchemy 风格的 `+driver` 后缀(asyncpg 不认);已干净的原样返回。""" scheme, sep, rest = dsn.partition("://") diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index 5b2dc56..2616f6c 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -563,6 +563,9 @@ class EmbeddingClient: pricing=PricingTable.from_file(gw.pricing_path) if gw.pricing_path is not None else None, + # embed 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 + # 一半不受控(issue #12) + text_cap=gw.telemetry_text_cap, batch_size=settings.batch_size, normalize=settings.normalize, expected_dim=settings.expected_dim, diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index da1e9f8..058637f 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -144,7 +144,12 @@ class TelemetryEmitter: """`text_cap` 无默认值是有意的: 它是关键行为参数,漏传即静默改变落库正文。 本类是库内部类,唯一构造者是三个公共 Client,必填能保证没有一处漏传。 + 同理,值域校验也放在这一处: 三个 Client 的 `text_cap` 全部汇流到这里, + `GatewaySettings` 那道只管 env 一条路,而直接构造 Client 是库承诺的另一 + 条公共装配路——`text_cap=0` 会让每条正文只剩一个省略标记(P5 不得静默)。 """ + if text_cap is not None and text_cap <= 0: + raise ValueError(f"text_cap 必须 > 0(不截断请传 None): {text_cap}") self._recorder = recorder self._pricing = pricing self._text_cap = text_cap diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 0c5ebe0..7f1b151 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -573,6 +573,9 @@ class OcrClient: backpressure=gw.backpressure, quota_full=gw.quota_full, telemetry=telemetry if telemetry is not None else _build_telemetry(gw), + # OCR 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 + # 一半不受控(issue #12) + text_cap=gw.telemetry_text_cap, ) @classmethod diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 6575071..302e644 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -351,6 +351,97 @@ class TestFactories: assert isinstance(client, GatewayClient) +class TestTelemetryTextCapWiring: + """`PGW_TELEMETRY_TEXT_CAP` 必须走通全部三条 `from_settings` 装配路(issue #12)。 + + 三条链路写的是**同一张** `llm_calls` 表:只接通 chat,embed 与 OCR 的行就 + 永远不受 cap 约束,同表内一半受控一半不受控——那正是本 issue 要消灭的状态。 + """ + + _CAP_ENV = dict(_ENV, PGW_TELEMETRY_TEXT_CAP="8") + _OCR_CAP_ENV = { + "OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866", + "OCR__MONKEY__1__API_KEY": "none", + "OCR__MONKEY__1__MODEL": "monkey-ocr", + "OCR__MONKEY__1__TIMEOUT_S": "120", + "LLM_MAX_RETRIES": "3", + "LLM_RETRY_BASE_DELAY": "2.0", + "LLM_RETRY_MAX_DELAY": "30.0", + "LLM_CIRCUIT_BREAKER_THRESHOLD": "5", + "LLM_CIRCUIT_BREAKER_COOLDOWN": "60", + "PGW_CACHE_BACKEND": "none", + "PGW_TELEMETRY_BACKEND": "none", + "PGW_TELEMETRY_TEXT_CAP": "8", + } + + def test_gateway_from_settings_wires_the_cap(self): + settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV) + client = GatewayClient.from_settings(settings, telemetry=_MemoryRecorder()) + assert client._terminal._emitter._text_cap == 8 + # 对照组: 不设该键时 emitter 拿到的必须是 None,否则 8 可能是硬编码来的 + unset = GatewayClient.from_settings( + GatewaySettings.from_env("LLM", env=_ENV), telemetry=_MemoryRecorder() + ) + assert unset._terminal._emitter._text_cap is None + + def test_embedding_from_settings_wires_the_cap(self): + from polygateway.config import EmbeddingSettings + from polygateway.embedding import EmbeddingClient + + gateway = GatewaySettings.from_env("LLM", env=self._CAP_ENV) + client = EmbeddingClient.from_settings( + EmbeddingSettings(gateway=gateway, batch_size=2), telemetry=_MemoryRecorder() + ) + assert client._emitter._text_cap == 8 + unset = EmbeddingClient.from_settings( + EmbeddingSettings(gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2), + telemetry=_MemoryRecorder(), + ) + assert unset._emitter._text_cap is None + + def test_ocr_from_settings_wires_the_cap(self): + from polygateway.config import OcrSettings + from polygateway.ocr import OcrClient + + settings = OcrSettings.from_env("OCR", env=dict(self._OCR_CAP_ENV)) + client = OcrClient.from_settings(settings, telemetry=_MemoryRecorder()) + assert client._emitter._text_cap == 8 + no_cap = dict(self._OCR_CAP_ENV) + no_cap.pop("PGW_TELEMETRY_TEXT_CAP") + unset = OcrClient.from_settings( + OcrSettings.from_env("OCR", env=no_cap), telemetry=_MemoryRecorder() + ) + assert unset._emitter._text_cap is None + + async def test_capped_body_reaches_the_recorder_end_to_end(self, monkeypatch): + """装配路通了还不够: 真跑一次 chat,落库的 messages 与 response 确已截断。 + + `from_settings` 自建 transport(没有 client_factory 入口),故在装配点 + 换掉该类以接上 MockTransport——洋葱其余各层仍是 `from_settings` 装的真件。 + """ + recorder = _MemoryRecorder() + long_text = "甲乙丙丁戊己庚辛壬癸" # 10 字,cap=8 → 略 2 字 + monkeypatch.setattr( + "polygateway.client.OpenAICompatTransport", + lambda **kwargs: OpenAICompatTransport( + client_factory=lambda source: httpx.AsyncClient( + transport=httpx.MockTransport(lambda request: _sse(content=long_text)) + ) + ), + ) + settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV) + async with GatewayClient.from_settings(settings, telemetry=recorder) as client: + await client.chat([{"role": "user", "content": long_text}]) + row = recorder.rows[-1] + assert json.loads(row["messages"])[0]["content"] == "甲乙丙丁戊己庚辛…(略 2 字)" + assert row["response"] == "甲乙丙丁戊己庚辛…(略 2 字)" + + def test_non_positive_cap_rejected_on_the_direct_construction_path(self): + """直接构造是库承诺的另一条公共装配路;cap=0 会让每条正文只剩省略标记。""" + with pytest.raises(ValueError, match="text_cap"): + _client(telemetry=_MemoryRecorder(), text_cap=0) + + class TestSharedBackend: async def test_two_clients_share_global_concurrency_gate(self): """VT R5: 两个逻辑角色显式注入同一 limiter → 共享全局并发闸。""" diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 0bf1d27..4f9772f 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -386,6 +386,33 @@ class TestTelemetrySchemaMode: ) +class TestTelemetryTextCap: + """`PGW_TELEMETRY_TEXT_CAP`(issue #12): 二态键,未设即不截断。 + + 与 `PGW_TELEMETRY_SCHEMA_MODE` 的三态不同,这里"未设"本身就是最终答案 + (不截断),没有需要按后端派生的第二种缺省,故不走 `_load_choice` 那套。 + """ + + def test_unset_key_means_no_truncation(self): + """缺省不截断是人类决策: 截断后的遥测不再是审计证据、无法复现重放。""" + assert GatewaySettings.from_env("LLM", env=_env()).telemetry_text_cap is None + + def test_positive_value_is_parsed_as_int(self): + s = GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_TEXT_CAP="2000")) + assert s.telemetry_text_cap == 2000 + + @pytest.mark.parametrize("raw", ["0", "-1"]) + def test_non_positive_rejected(self, raw): + """0 会把每条正文退化成一个省略标记,负数无意义;都不是"不截断"的写法。""" + with pytest.raises(ValueError, match="PGW_TELEMETRY_TEXT_CAP"): + GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_TEXT_CAP=raw)) + + def test_non_integer_rejected_naming_the_env_key(self): + """报错须点出 env 键名: 这条路的调用方看得懂的是键名,不是字段名。""" + with pytest.raises(ValueError, match="PGW_TELEMETRY_TEXT_CAP"): + GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_TEXT_CAP="2k")) + + class TestOcrSettings: """M3 OcrSettings(设计 §3.4): 复用 GatewaySettings,无 OCR 专用键。""" @@ -622,6 +649,12 @@ class TestCrossFieldInvariants: # —— 标量域 —— + def test_non_positive_text_cap_rejected(self): + """env 路只覆盖 from_env;直接构造与 replace 同样能把 0 传进来(issue #12)。""" + base = self._base() + with pytest.raises(ValueError, match="telemetry_text_cap"): + dataclasses.replace(base, telemetry_text_cap=0) + def test_negative_structured_retries_rejected(self): base = self._base() with pytest.raises(ValueError, match="structured_max_retries"): diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 6c1d03b..30d938b 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -1602,6 +1602,16 @@ class TestTelemetryTextCap: assert messages[1]["content"][0]["text"] == _LONG assert json.loads(row["messages"])[0]["content"] == _CAPPED # 落库那份确已截断 + def test_non_positive_cap_rejected_at_construction(self): + """emitter 是三个 Client 唯一的汇合点,值域校验放这一处即覆盖全部装配路。 + + settings 层那道只管 env;直接构造 `GatewayClient(..., text_cap=0)` 是库 + 承诺的另一条公共装配路,没有这道闸就会把每条正文写成一个光秃秃的省略标记。 + """ + for bad in (0, -1): + with pytest.raises(ValueError, match="text_cap"): + TelemetryEmitter(_MemoryRecorder(), text_cap=bad) + class _StubEmbedTransport: async def embed(self, *, texts, source, call_id):