From 6bb64ca9387f0d76466637507392e0b4f0f2dd50 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 31 Jul 2026 21:17:49 -0400 Subject: [PATCH] feat: accept sampling overlay on chat() and per-source extra_body Priority is structured injection > per-call overlay > per-source config, and extra_body now takes part in the cache fingerprint. --- src/polygateway/client.py | 47 ++++++++- src/polygateway/transports/openai_compat.py | 3 + tests/unit/test_client.py | 101 ++++++++++++++++++++ tests/unit/test_openai_compat.py | 21 ++++ 4 files changed, 167 insertions(+), 5 deletions(-) diff --git a/src/polygateway/client.py b/src/polygateway/client.py index 7ea7837..7ce8e67 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -9,6 +9,8 @@ from __future__ import annotations import asyncio +import hashlib +import json import random import time from typing import TYPE_CHECKING, Any, Literal, TypeVar @@ -32,7 +34,7 @@ from polygateway.sources import ( SourceCooldownMemo, ) from polygateway.transports.openai_compat import OpenAICompatTransport -from polygateway.types import ChatRequest, LLMResponse +from polygateway.types import ChatRequest, LLMResponse, validate_request_overlay if TYPE_CHECKING: from collections.abc import Awaitable, Iterable, Mapping @@ -59,6 +61,29 @@ if TYPE_CHECKING: _T = TypeVar("_T") +def build_model_fingerprint(sources: Iterable[SourceConfig]) -> str: + """缓存 key 的模型身份: 多源 scope = 排序去重的 model 合集。 + + 配置级采样参数(`extra_body`)必须参与,否则把 temperature 从 0 改成 1 + 后重启仍会读到旧缓存(issue #4 设计决策 C)。全源 `extra_body` 皆空时 + 字面量与历史实现逐字相同,不触发存量缓存冷启动。 + """ + fingerprint = ",".join(sorted({s.model for s in sources})) + # 按 (model, extra_body) 而非源名摘要: 语义是"本 scope 会用哪些 + # (模型, 解码参数)组合",改源名不该误触全量冷启动 + marks = sorted( + { + json.dumps([s.model, dict(s.extra_body)], sort_keys=True, ensure_ascii=False) + for s in sources + if s.extra_body + } + ) + if marks: + digest = hashlib.sha256("".join(marks).encode("utf-8")).hexdigest() + fingerprint = f"{fingerprint}|{digest}" + return fingerprint + + class GatewayClient: """统一治理入口;构造函数全量注入(测试/高级),工厂覆盖 90% 场景。""" @@ -113,12 +138,11 @@ class GatewayClient: if cache is not None: if cache_namespace is None or cache_ttl_s is None: raise ValueError("启用缓存必须提供 cache_namespace 与 cache_ttl_s") - # 多源 scope 的 key 身份 = 排序去重的 model 合集;源集合变化 → 一次性冷启动 - fingerprint = ",".join(sorted({s.model for s in sources})) + # 多源 scope 的 key 身份;源集合或其 extra_body 变化 → 一次性冷启动 middlewares.append( CacheMW( backend=cache, - model_fingerprint=fingerprint, + model_fingerprint=build_model_fingerprint(sources), default_namespace=cache_namespace, ttl_s=cache_ttl_s, strategy=structured_strategy, @@ -150,12 +174,23 @@ class GatewayClient: cache_namespace: str | None = None, structured: type[BaseModel] | Literal["json"] | None = None, stream: bool = True, + overlay: Mapping[str, Any] | None = None, ) -> LLMResponse: - """一次治理调用(签名冻结,ARCH §5.2;与三项目 LLMProvider 协议兼容)。""" + """一次治理调用(签名冻结,ARCH §5.2;与三项目 LLMProvider 协议兼容)。 + + `overlay` 是采样参数覆盖层(`temperature`/`seed`/`max_tokens` 等),优先级 + 高于源级 `extra_body`、低于结构化输出的注入。带默认值的 keyword-only + 参数不影响既有调用点(issue #4)。 + """ if structured is not None and not self._structured_available: raise ImportError( "结构化输出未启用: 安装 pip install 'polygateway[structured]' 后重新装配" ) + # 进洋葱之前校验并拷贝: 保护键/不可序列化值在此收口(否则会在 CacheMW + # 的降级 try 之外抛裸 TypeError);拷贝防调用方复用同一 dict 逐次改 seed + # 造成的竞态。同一份快照填 overlay 与 sampling——前者会被结构化注入, + # 后者跨层恒定,供缓存 key 与遥测读取(设计决策 A/B/E) + sampling = validate_request_overlay(overlay or {}, origin="chat(overlay=...)") request = ChatRequest( messages=messages, session_id=session_id, @@ -164,6 +199,8 @@ class GatewayClient: cache_namespace=cache_namespace, structured=structured, stream=stream, + overlay=sampling, + sampling=sampling, ) return await self._handler(request) diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index f0164e9..c8111df 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -294,6 +294,9 @@ class OpenAICompatTransport: payload.update(profile.thinking_on) elif source.enable_thinking is False: payload.update(profile.thinking_off) + # 顺序即优先级(issue #4 设计决策 A): 配置级 extra_body 在前,调用级 + # overlay(含结构化注入)在后覆盖之。两行不可调换 + payload.update(source.extra_body) payload.update(overlay) return payload diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index f7ceaec..637c668 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -19,6 +19,7 @@ from polygateway.backends.memory.cache import InMemoryCache from polygateway.backends.memory.limiter import InMemoryLimiter from polygateway.sources import RoundRobinSelector from polygateway.structured.json_repair import JsonRepairStrategy +from polygateway.structured.native_schema import NativeSchemaStrategy from polygateway.transports.openai_compat import OpenAICompatTransport from polygateway.types import ( BackpressurePolicy, @@ -117,6 +118,106 @@ class TestChatEndToEnd: await client.chat([{"role": "user", "content": "hi"}], structured="json") +class TestSamplingOverlay: + """调用级采样参数入口(issue #4 Task 3)。""" + + def _capturing_client(self, captured, **overrides): + def handler(request): + captured.append(json.loads(request.content)) + return _sse() + + return _client(handler=handler, **overrides) + + async def test_overlay_reaches_request_body(self): + captured = [] + async with self._capturing_client(captured) as client: + await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 42}) + assert captured[0]["seed"] == 42 + + async def test_call_level_beats_config_level(self): + """优先级: 调用级 > 配置级(设计决策 A)。""" + captured = [] + source = _source(extra_body={"temperature": 0, "top_p": 0.9}) + async with self._capturing_client(captured, sources=[source]) as client: + await client.chat([{"role": "user", "content": "hi"}], overlay={"temperature": 1}) + assert captured[0]["temperature"] == 1 # 调用级覆盖 + assert captured[0]["top_p"] == 0.9 # 配置级未被顶掉的键保留 + + async def test_structured_injection_beats_call_level(self): + """结构化注入优先级最高: 它关系到响应能否被解析(设计决策 A)。""" + captured = [] + client = self._capturing_client(captured, structured_strategy=NativeSchemaStrategy()) + async with client: + await client.chat( + [{"role": "user", "content": "hi"}], + structured="json", + overlay={"response_format": {"type": "text"}}, + ) + assert captured[0]["response_format"] != {"type": "text"} + + async def test_protected_key_rejected_before_onion(self): + """保护键在进洋葱之前就报错,transport 一次都不该被碰到。""" + captured = [] + async with self._capturing_client(captured) as client: + with pytest.raises(ValueError, match="stream"): + await client.chat([{"role": "user", "content": "hi"}], overlay={"stream": False}) + assert captured == [] + + async def test_unserializable_value_rejected_before_onion(self): + """裸 TypeError 会在 CacheMW 的降级 try 之外炸且无遥测(设计决策 B)。""" + captured = [] + async with self._capturing_client(captured) as client: + with pytest.raises(ValueError, match="JSON"): + await client.chat( + [{"role": "user", "content": "hi"}], overlay={"temperature": object()} + ) + assert captured == [] + + async def test_caller_dict_mutation_does_not_leak(self): + """调用方逐次改 seed 复用同一 dict 是预期模式(设计决策 E)。""" + captured = [] + caller_overlay = {"seed": 1} + async with self._capturing_client(captured) as client: + await client.chat([{"role": "user", "content": "hi"}], overlay=caller_overlay) + caller_overlay["seed"] = 2 + await client.chat([{"role": "user", "content": "hi"}], overlay=caller_overlay) + assert [c["seed"] for c in captured] == [1, 2] + + +class TestModelFingerprint: + """配置级采样参数须进缓存身份,否则改 temperature 后仍读旧缓存(决策 C)。""" + + def test_empty_extra_body_keeps_legacy_fingerprint(self): + """全源无 extra_body 时字面量与旧实现逐字相同,不触发存量缓存冷启动。""" + from polygateway.client import build_model_fingerprint + + sources = [_source(), _source(name="qwen_2", model="qwen-plus")] + assert build_model_fingerprint(sources) == "qwen-max,qwen-plus" + + def test_extra_body_changes_fingerprint(self): + from polygateway.client import build_model_fingerprint + + plain = build_model_fingerprint([_source()]) + tuned = build_model_fingerprint([_source(extra_body={"temperature": 0})]) + assert plain != tuned + assert tuned.startswith("qwen-max|") # 旧字面量仍是前缀,便于人眼辨认 + + def test_source_rename_does_not_change_fingerprint(self): + """指纹按 (model, extra_body) 而非源名: 改名不该误触全量冷启动。""" + from polygateway.client import build_model_fingerprint + + a = build_model_fingerprint([_source(name="qwen_1", extra_body={"temperature": 0})]) + b = build_model_fingerprint([_source(name="renamed", extra_body={"temperature": 0})]) + assert a == b + + def test_differing_extra_body_across_sources_is_distinguished(self): + from polygateway.client import build_model_fingerprint + + a = build_model_fingerprint([_source(extra_body={"temperature": 0})]) + b = build_model_fingerprint([_source(extra_body={"temperature": 1})]) + assert a != b + + class TestFactories: def test_from_env_assembles(self): client = GatewayClient.from_env("LLM", env=_ENV) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 65c5f48..5bce879 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -445,6 +445,27 @@ class TestRequestShaping: ) assert seen["response_format"] == {"type": "json_object"} + async def test_extra_body_merged_and_outranked_by_overlay(self): + """顺序即优先级: thinking profile → extra_body → overlay(issue #4)。""" + seen = {} + + def handler(request): + seen.update(json.loads(request.content)) + return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE)) + + await _complete( + _transport_for(handler), + _source(extra_body={"temperature": 0, "top_p": 0.9}), + overlay={"temperature": 1}, + ) + assert seen["temperature"] == 1 # 调用级覆盖配置级 + assert seen["top_p"] == 0.9 # 未被顶掉的配置级键保留 + + async def test_extra_body_cannot_break_governed_keys(self): + """治理键由 payload 骨架拥有;extra_body 的保护键在构造期已被拦下。""" + with pytest.raises(ValueError, match="stream"): + _source(extra_body={"stream": False}) + class TestErrorTranslation: @pytest.mark.parametrize(