diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index 81e155f..b2093c7 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -29,7 +29,7 @@ if TYPE_CHECKING: class TelemetryEmitter: - """从请求与结果组装 20 字段并写入 recorder;一切写失败降级 warning。""" + """从请求与结果组装 21 字段并写入 recorder;一切写失败降级 warning。""" def __init__(self, recorder: TelemetryRecorder, *, pricing: PricingTable | None = None) -> None: self._recorder = recorder @@ -65,9 +65,7 @@ class TelemetryEmitter: cached_prompt_tokens=response.cached_prompt_tokens if response else None, model_reported=response.model_reported if response else None, # 唯一有"生效源"的入口,故是唯一能并上 extra_body 的(设计决策 D) - sampling=canonical_sampling_json( - merge_sampling(source.extra_body, request.sampling) - ), + sampling=canonical_sampling_json(merge_sampling(source.extra_body, request.sampling)), ) async def emit_cache_hit(self, *, request: ChatRequest, response: LLMResponse) -> None: diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 96f71e5..723131f 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -59,9 +59,7 @@ def validate_request_overlay(overlay: Mapping[str, Any], *, origin: str) -> dict return dict(overlay) -def merge_sampling( - extra_body: Mapping[str, Any], sampling: Mapping[str, Any] -) -> dict[str, Any]: +def merge_sampling(extra_body: Mapping[str, Any], sampling: Mapping[str, Any]) -> dict[str, Any]: """合并配置级与调用级采样参数;调用级优先(issue #4 设计决策 A)。""" return {**extra_body, **sampling} @@ -237,9 +235,7 @@ class SourceConfig: object.__setattr__(self, "extra_body", MappingProxyType(validated)) -def strip_unsupported_extra_body( - sources: list[SourceConfig], *, path: str -) -> list[SourceConfig]: +def strip_unsupported_extra_body(sources: list[SourceConfig], *, path: str) -> list[SourceConfig]: """剥离非 chat 路径不消费的 `extra_body` 并 warning(issue #4 决策 G)。 剥离是必需的而非顺手清理: embedding 的 payload 硬编码 `{model, input}`、 diff --git a/tests/integration/test_governance_stack.py b/tests/integration/test_governance_stack.py index 86eb237..1b31d69 100644 --- a/tests/integration/test_governance_stack.py +++ b/tests/integration/test_governance_stack.py @@ -4,6 +4,7 @@ """ import asyncio +import dataclasses import json import sqlite3 @@ -204,3 +205,74 @@ class TestTransientErrorExport: assert isinstance(ei.value, polygateway.AllSourcesExhausted) assert isinstance(ei.value.__cause__, TransientError) + + +class TestSamplingThroughStack: + """issue #4: 采样参数经完整洋葱到达请求体,且缓存/遥测口径一致。""" + + async def test_reaches_wire_and_lands_in_telemetry(self, tmp_path): + seen = [] + + def handler(request): + seen.append(json.loads(request.content)) + return _sse() + + db = tmp_path / "t.db" + recorder = SQLiteRecorder(db) + client = _full_client(handler, telemetry=recorder) + await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 42}) + recorder.close() + + assert seen[0]["seed"] == 42 # 穿过全栈到达线上 + rows = sqlite3.connect(db).execute("SELECT sampling FROM llm_calls").fetchall() + assert json.loads(rows[0][0]) == {"seed": 42} + + async def test_config_level_merges_and_records(self, tmp_path): + """源级 extra_body 只有 emit_attempt 记得到(唯一有生效源的入口)。""" + seen = [] + + def handler(request): + seen.append(json.loads(request.content)) + return _sse() + + src = dataclasses.replace(_source(), extra_body={"temperature": 0}) + db = tmp_path / "t.db" + recorder = SQLiteRecorder(db) + client = GatewayClient( + scope="llm", + sources=[src], + selector=RoundRobinSelector(), + limiter=InMemoryLimiter( + scope="llm", sources={src.name: src}, global_limits=GlobalLimits(0, 0, 0) + ), + breaker=InMemoryGate(config=_BREAKER), + transport=OpenAICompatTransport( + client_factory=lambda s: httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ), + retry=RetryPolicy(2, 2.0, 30.0), + backpressure=BackpressurePolicy(300.0, 0.01), + telemetry=recorder, + structured_strategy=JsonRepairStrategy(), + sleep=_noop_sleep, + ) + await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 1}) + recorder.close() + + assert seen[0]["temperature"] == 0 and seen[0]["seed"] == 1 + rows = sqlite3.connect(db).execute("SELECT sampling FROM llm_calls").fetchall() + assert json.loads(rows[0][0]) == {"seed": 1, "temperature": 0} + + async def test_differing_seed_bypasses_cache_end_to_end(self): + """issue 场景全栈回归: 逐 rollout 变 seed 必须真的回源。""" + calls = [] + + def handler(request): + calls.append(json.loads(request.content)["seed"]) + return _sse() + + client = _full_client(handler, cache=InMemoryCache()) + await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 1}) + await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 2}) + second_same = await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 1}) + assert calls == [1, 2] # 两个不同 seed 各自回源 + assert second_same.cache_hit is True # 同 seed 才命中 diff --git a/tests/unit/test_embedding.py b/tests/unit/test_embedding.py index c95b9b7..a33f5a8 100644 --- a/tests/unit/test_embedding.py +++ b/tests/unit/test_embedding.py @@ -381,9 +381,7 @@ class TestExtraBodyStripped: 会显示这次调用带了 temperature=0——那是数据造假,比参数失效更坏。 """ rec = _MemoryRecorder() - client, _ = _embed_client( - [_src(extra_body={"temperature": 0})], ["ok"], telemetry=rec - ) + client, _ = _embed_client([_src(extra_body={"temperature": 0})], ["ok"], telemetry=rec) await client.embed(["hi"]) assert rec.rows[0]["sampling"] is None diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index ba1f6d9..1c737d6 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -394,9 +394,7 @@ class TestExtraBodyStripped: 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 - ) + client, _, _ = _client([_src(extra_body={"temperature": 0})], ["text"], telemetry=recorder) await client.recognize_text(b"IMG") assert recorder.rows[0]["sampling"] is None