feat: strip extra_body on the embedding and OCR paths with a warning

Stripping is load-bearing, not tidying: those transports never send the
value, so leaving it would make telemetry record a parameter never sent.
This commit is contained in:
2026-07-31 21:36:50 -04:00
parent 4516761dbe
commit a5ebf72f17
5 changed files with 112 additions and 3 deletions
+9 -2
View File
@@ -41,7 +41,12 @@ from polygateway.middleware.ratelimit import QuotaGate
from polygateway.middleware.retry import _failure_reason, backoff_delay from polygateway.middleware.retry import _failure_reason, backoff_delay
from polygateway.middleware.telemetry import TelemetryEmitter from polygateway.middleware.telemetry import TelemetryEmitter
from polygateway.sources import SourceCooldownMemo from polygateway.sources import SourceCooldownMemo
from polygateway.types import ChatRequest, EmbeddingResponse, LLMResponse from polygateway.types import (
ChatRequest,
EmbeddingResponse,
LLMResponse,
strip_unsupported_extra_body,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Mapping from collections.abc import Awaitable, Callable, Mapping
@@ -111,7 +116,9 @@ class EmbeddingClient:
if expected_dim is not None and expected_dim < 1: if expected_dim is not None and expected_dim < 1:
raise ValueError("expected_dim 必须 ≥ 1") raise ValueError("expected_dim 必须 ≥ 1")
self._scope = scope self._scope = scope
self._sources = list(sources) # embed payload 硬编码 {model, input},带 extra_body 的源必须先剥离,
# 否则遥测会记录一个从未发出的采样参数(issue #4 决策 G)
self._sources = strip_unsupported_extra_body(list(sources), path="embedding")
self._selector = selector self._selector = selector
self._quota = QuotaGate(limiter) self._quota = QuotaGate(limiter)
self._breaker = BreakerGate(breaker) self._breaker = BreakerGate(breaker)
+4 -1
View File
@@ -44,6 +44,7 @@ from polygateway.types import (
OcrLayoutResult, OcrLayoutResult,
OcrTextResult, OcrTextResult,
Usage, Usage,
strip_unsupported_extra_body,
) )
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -113,7 +114,9 @@ class OcrClient:
if quota_full not in ("wait", "fail_fast"): if quota_full not in ("wait", "fail_fast"):
raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}") raise ValueError(f"quota_full 必须是 wait|fail_fast: {quota_full!r}")
self._scope = scope self._scope = scope
self._sources = list(sources) # MonkeyOCR 只发 multipart 表单,带 extra_body 的源必须先剥离,否则
# 遥测会记录一个从未发出的采样参数(issue #4 决策 G)
self._sources = strip_unsupported_extra_body(list(sources), path="OCR")
self._selector = selector self._selector = selector
self._feed_health = isinstance(selector, OutcomeAwareSelector) self._feed_health = isinstance(selector, OutcomeAwareSelector)
self._quota = QuotaGate(limiter) self._quota = QuotaGate(limiter)
+31
View File
@@ -4,12 +4,15 @@
fake,字段顺序即公共承诺;新增字段只增不删且必带默认值。 fake,字段顺序即公共承诺;新增字段只增不删且必带默认值。
""" """
import dataclasses
import json import json
from collections.abc import Mapping from collections.abc import Mapping
from dataclasses import dataclass, field from dataclasses import dataclass, field
from types import MappingProxyType from types import MappingProxyType
from typing import Any from typing import Any
from loguru import logger
_MISSING_DONE_DOMAIN = frozenset({"retry", "salvage"}) _MISSING_DONE_DOMAIN = frozenset({"retry", "salvage"})
_PROTECTED_OVERLAY_KEYS: Mapping[str, str] = MappingProxyType( _PROTECTED_OVERLAY_KEYS: Mapping[str, str] = MappingProxyType(
@@ -234,6 +237,34 @@ class SourceConfig:
object.__setattr__(self, "extra_body", MappingProxyType(validated)) object.__setattr__(self, "extra_body", MappingProxyType(validated))
def strip_unsupported_extra_body(
sources: list[SourceConfig], *, path: str
) -> list[SourceConfig]:
"""剥离非 chat 路径不消费的 `extra_body` 并 warning(issue #4 决策 G)。
剥离是必需的而非顺手清理: embedding 的 payload 硬编码 `{model, input}`、
MonkeyOCR 只发 multipart 表单,两者都不会把 `extra_body` 发出去;但遥测的
`sampling` 列会并上 `source.extra_body`,不剥离就等于**记录一个从未发出的
参数**——那是数据造假,污染的恰是事后复现的唯一依据。
选择 warning 放行而非报错: 这两条路径本无采样语义,配错的后果远轻于 chat
路径,不值得让下游整个装配起不来(2026-07-31 人类拍板)。
"""
stripped = []
for source in sources:
if source.extra_body:
logger.warning(
"{} 路径暂不支持 extra_body,源 {} 的该配置已被忽略"
"(需要 dimensions 等参数请提 issue): {}",
path,
source.name,
dict(source.extra_body),
)
source = dataclasses.replace(source, extra_body={})
stripped.append(source)
return stripped
@dataclass(frozen=True) @dataclass(frozen=True)
class RetryPolicy: class RetryPolicy:
"""重试策略;max_attempts = 总尝试次数(含首次,M1 设计 §2.3 统一语义)。""" """重试策略;max_attempts = 总尝试次数(含首次,M1 设计 §2.3 统一语义)。"""
+44
View File
@@ -4,11 +4,13 @@
VT adapters/embedding.py(归一化);库裁决见设计 §7.3 表。 VT adapters/embedding.py(归一化);库裁决见设计 §7.3 表。
""" """
import contextlib
import dataclasses import dataclasses
import json import json
import httpx import httpx
import pytest import pytest
from loguru import logger
from polygateway.errors import ( from polygateway.errors import (
RequestRejectedError, RequestRejectedError,
@@ -350,6 +352,48 @@ class TestEmbedTelemetry:
assert len(rec.rows[1]["messages"]) < 1000 # 长文本截断后入库 assert len(rec.rows[1]["messages"]) < 1000 # 长文本截断后入库
@contextlib.contextmanager
def _captured_warnings():
"""捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。"""
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
yield messages
finally:
logger.remove(sink_id)
class TestExtraBodyStripped:
"""issue #4 决策 G: embedding 路径不消费 extra_body,剥离并 warning。"""
async def test_stripped_with_warning_but_assembly_succeeds(self):
"""报错会让下游整个装配起不来,而这条路径本无采样语义(人类拍板)。"""
with _captured_warnings() as warnings:
client, _ = _embed_client([_src(extra_body={"temperature": 0})], ["ok"])
assert client._sources[0].extra_body == {}
assert any("extra_body" in m for m in warnings)
assert any("dimensions" in m for m in warnings) # 文案须指路,不能只说不支持
await client.embed(["hi"]) # 装配后可正常工作
async def test_telemetry_never_records_a_parameter_that_was_not_sent(self):
"""剥离的真正理由: embed payload 硬编码 {model, input},不剥离则审计表
会显示这次调用带了 temperature=0——那是数据造假,比参数失效更坏。
"""
rec = _MemoryRecorder()
client, _ = _embed_client(
[_src(extra_body={"temperature": 0})], ["ok"], telemetry=rec
)
await client.embed(["hi"])
assert rec.rows[0]["sampling"] is None
async def test_no_warning_without_extra_body(self):
with _captured_warnings() as warnings:
client, _ = _embed_client([_src()], ["ok"])
assert client._sources[0].extra_body == {}
assert not [m for m in warnings if "extra_body" in m]
class TestEmbeddingSettings: class TestEmbeddingSettings:
_ENV = { _ENV = {
"EMBED__QWEN__1__BASE_URL": "https://gw.example/v1", "EMBED__QWEN__1__BASE_URL": "https://gw.example/v1",
+24
View File
@@ -7,6 +7,7 @@ retry_exhausted/circuit_open/stalled 三组断言即设计 §6 ③ 的 G1 契约
import asyncio import asyncio
import pytest import pytest
from loguru import logger
from polygateway.backends.memory.breaker import InMemoryGate from polygateway.backends.memory.breaker import InMemoryGate
from polygateway.backends.memory.limiter import InMemoryLimiter from polygateway.backends.memory.limiter import InMemoryLimiter
@@ -377,6 +378,29 @@ class TestCheckHealth:
await task await task
class TestExtraBodyStripped:
"""issue #4 决策 G: OCR 路径只发 multipart 表单,剥离 extra_body 并 warning。"""
def test_stripped_with_warning_but_assembly_succeeds(self):
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
client, _, _ = _client([_src(extra_body={"temperature": 0})], ["text"])
finally:
logger.remove(sink_id)
assert client._sources[0].extra_body == {}
assert any("extra_body" in m for m in messages)
async def test_telemetry_never_records_a_parameter_that_was_not_sent(self):
"""不剥离则审计表会显示这次 OCR 带了 temperature=0——数据造假。"""
recorder = _MemoryRecorder()
client, _, _ = _client(
[_src(extra_body={"temperature": 0})], ["text"], telemetry=recorder
)
await client.recognize_text(b"IMG")
assert recorder.rows[0]["sampling"] is None
class TestTelemetry: class TestTelemetry:
async def test_success_and_failure_recorded_without_image_bytes(self): async def test_success_and_failure_recorded_without_image_bytes(self):
recorder = _MemoryRecorder() recorder = _MemoryRecorder()