diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 91f6c6d..7cff61f 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -285,6 +285,9 @@ class RetryMW: stream=request.stream, overlay=request.overlay, call_id=call_id, + # 逐次尝试原样重传: 换源不改变调用方要的档位(源级默认由 transport + # 自己按选中的源解析,两者在 effective_effort 里汇合) + reasoning_effort=request.reasoning_effort, ) if result.usage_source == "unavailable": # 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回, diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index 09a6c3f..020e943 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -12,6 +12,7 @@ from typing import Any, Protocol, runtime_checkable from .types import ( ChatRequest, + Effort, EmbeddingTransportResult, LLMResponse, OcrLayoutResult, @@ -36,7 +37,16 @@ class Middleware(Protocol): @runtime_checkable class Transport(Protocol): - """一次原始调用的协议细节(请求组装/流式解析/错误翻译);不含任何治理。""" + """一次原始调用的协议细节(请求组装/流式解析/错误翻译);不含任何治理。 + + `reasoning_effort` 是本次调用要求的推理档位(`None` = 不表态,随源级配置)。 + 它必须走**协议参数**而不能让 transport 自己去读 `ChatRequest`: 端口只收拆开的 + 请求要素,是为了让 transport 不依赖洋葱内部的请求类型(P7 端口最内层)。 + + 该参数**不设默认值**,与 `TelemetryRecorder.record_llm_call` 同一既有约定: + 库外无第三方实现者,写全签名的成本为零,而默认值会把"某一层漏传"变成静默的 + "调用方没表态"——一次本该报错的漏配就此变成一次悄悄涨价的调用。 + """ async def complete( self, @@ -46,6 +56,7 @@ class Transport(Protocol): stream: bool, overlay: dict[str, Any], call_id: str, + reasoning_effort: Effort | None, ) -> TransportResult: ... diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index a86f7af..d2e6ad9 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -28,6 +28,7 @@ from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_ti from polygateway.thinking import ( ThinkingCapability, ThinkingUnsupportedError, + effective_effort, get_capability, observe_thinking, reconcile_thinking, @@ -351,6 +352,7 @@ class OpenAICompatTransport: profile: ProviderProfile, stream: bool, overlay: dict[str, Any], + reasoning_effort: Effort | None, ) -> dict[str, Any]: payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream} if stream: @@ -360,17 +362,19 @@ class OpenAICompatTransport: capability = get_capability(source.model, table=self._capabilities) first_time = source.model not in self._warned_models self._warned_models.add(source.model) - # `enable_thinking` 的档位语法糖(True → auto,False → none,None 不表态); - # 就地转换是过渡形态,T5 起由 thinking.effective_effort() 统一收口并接上 - # 源级/请求级档位(设计 §4.2) - enabled = source.enable_thinking - effort = None if enabled is None else (Effort.AUTO if enabled else Effort.NONE) + # 三层优先级在此汇合: 请求级 > 源级 > enable_thinking 语法糖(设计 §4.2)。 + # 判定与装配守卫共用同一个纯函数,两处分叉就会变成"装配期放行、运行期报错" payload.update( resolve_thinking( profile, capability, - effort, + effective_effort( + request_effort=reasoning_effort, + source_effort=source.reasoning_effort, + enable_thinking=source.enable_thinking, + ), model=source.model, + fallback=source.effort_fallback, warn_unregistered=first_time, ).payload ) @@ -388,12 +392,22 @@ class OpenAICompatTransport: stream: bool, overlay: dict[str, Any], call_id: str, + reasoning_effort: Effort | None, ) -> TransportResult: - """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。""" + """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。 + + `reasoning_effort` 是**请求级**档位(`None` = 不表态);它与源级配置的优先级 + 在 `_build_payload` 里由 `effective_effort` 裁定,本层只负责把它送到。 + """ profile = get_provider(source.provider, registry=self._registry) try: payload = self._build_payload( - messages=messages, source=source, profile=profile, stream=stream, overlay=overlay + messages=messages, + source=source, + profile=profile, + stream=stream, + overlay=overlay, + reasoning_effort=reasoning_effort, ) except ThinkingUnsupportedError as exc: # 推理开关不可满足是**请求本身**的问题: 换源重试都救不了它。只捕这个 diff --git a/tests/e2e/test_thinking_live.py b/tests/e2e/test_thinking_live.py index db35e85..707dc34 100644 --- a/tests/e2e/test_thinking_live.py +++ b/tests/e2e/test_thinking_live.py @@ -514,6 +514,7 @@ class TestAssemblyGuardAgainstRealConfig: stream=True, overlay={}, call_id="e2e-guard", + reasoning_effort=None, ) finally: await client.aclose() diff --git a/tests/integration/test_redis_cross_connection.py b/tests/integration/test_redis_cross_connection.py index b3cd507..4b37cb6 100644 --- a/tests/integration/test_redis_cross_connection.py +++ b/tests/integration/test_redis_cross_connection.py @@ -73,7 +73,7 @@ class ScriptedTransport: self.hang = hang self.calls: list[str] = [] - async def complete(self, *, messages, source, stream, overlay, call_id): + async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort): self.calls.append(source.name) if self.hang: await asyncio.Event().wait() diff --git a/tests/unit/test_backpressure.py b/tests/unit/test_backpressure.py index bde7e93..479ea05 100644 --- a/tests/unit/test_backpressure.py +++ b/tests/unit/test_backpressure.py @@ -210,7 +210,7 @@ class ClockAdvancingTransport: self.clock = clock self.calls = [] - async def complete(self, *, messages, source, stream, overlay, call_id): + async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort): self.calls.append((source.name, call_id)) advance, action = self.script.pop(0) self.clock.advance(advance) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 90e3608..4b3482e 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -24,6 +24,7 @@ from polygateway.transports.openai_compat import OpenAICompatTransport from polygateway.types import ( BackpressurePolicy, BreakerConfig, + Effort, GlobalLimits, RetryPolicy, SourceConfig, @@ -219,6 +220,31 @@ class TestReasoningEffortPriority: overrides.setdefault("model", "glm-5.3") return _source(provider="zhipu", **overrides) + async def test_request_effort_wins_over_source(self): + captured = [] + source = self._zhipu(reasoning_effort=Effort.LOW) + async with self._capturing_client(captured, sources=[source]) as client: + await client.chat([{"role": "user", "content": "hi"}], reasoning_effort=Effort.MAX) + assert captured[0]["reasoning_effort"] == "max" + assert captured[0]["thinking"] == {"type": "enabled"} + + async def test_none_request_does_not_clear_source(self): + """请求级不表态 ≠ 请求级要求"不推理": 前者必须让源级默认继续生效。""" + captured = [] + source = self._zhipu(reasoning_effort=Effort.LOW) + async with self._capturing_client(captured, sources=[source]) as client: + await client.chat([{"role": "user", "content": "hi"}]) + assert captured[0]["reasoning_effort"] == "low" + + async def test_request_none_tier_is_an_opinion_not_an_absence(self): + """请求级 `none` 是"要求不推理",不得被当成"没表态"而回落到源级档位。""" + captured = [] + source = self._zhipu(model="glm-5.2", reasoning_effort=Effort.MAX) + async with self._capturing_client(captured, sources=[source]) as client: + await client.chat([{"role": "user", "content": "hi"}], reasoning_effort=Effort.NONE) + assert captured[0]["thinking"] == {"type": "disabled"} + assert "reasoning_effort" not in captured[0] + @pytest.mark.parametrize( ("provider", "model", "fragment"), [ diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 5116cb2..3fcae73 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -69,13 +69,14 @@ def _transport_for(handler, *, registry=None): ) -async def _complete(transport, source, *, stream=True, overlay=None): +async def _complete(transport, source, *, stream=True, overlay=None, reasoning_effort=None): return await transport.complete( messages=[{"role": "user", "content": "hi"}], source=source, stream=stream, overlay=overlay or {}, call_id="cid-1", + reasoning_effort=reasoning_effort, ) diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index 89aea7c..b0d8c5d 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -1,15 +1,18 @@ """ports.py 端口冻结测试(M1 设计 §4): Protocol 结构性检查 + Gate 快照校验。""" +import inspect from typing import Any import pytest from polygateway.ports import ( CacheBackend, + EmbeddingTransport, GateDecision, GateState, GateUpdate, Middleware, + OcrTransport, Permit, ProviderGate, RateLimiter, @@ -69,7 +72,7 @@ class _DummyMw: class _DummyTransport: - async def complete(self, *, messages, source, stream, overlay, call_id): + async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort): raise NotImplementedError @@ -142,6 +145,34 @@ def test_protocols_are_runtime_checkable(impl, protocol): assert isinstance(impl, protocol) +class TestReasoningTierIsOnlyOnTheChatPort: + """档位属于 chat 端口,且**只属于**它(Task 5b)。 + + `@runtime_checkable` 只查方法名不查签名,故协议签名本身必须被显式断言—— + 否则实现漏改一个参数,要到运行期调用才会以 `TypeError` 现形,而那时的现场 + 离根因已经很远。 + """ + + def test_chat_transport_carries_the_per_call_tier(self): + params = inspect.signature(Transport.complete).parameters + assert "reasoning_effort" in params + # 不给默认值是有意的(与 TelemetryRecorder 同一既有约定): 库外无第三方 + # 实现者,写全签名成本为零,而默认值会把"漏传"变成静默的"不表态" + assert params["reasoning_effort"].default is inspect.Parameter.empty + + @pytest.mark.parametrize( + ("protocol", "method"), + [ + (EmbeddingTransport, "embed"), + (OcrTransport, "recognize_text"), + (OcrTransport, "parse_layout"), + ], + ) + def test_other_transports_have_no_reasoning_tier(self, protocol, method): + """embedding 与 OCR 没有推理语义,给它们加档位只会静默无效(issue #4 同款决策)。""" + assert "reasoning_effort" not in inspect.signature(getattr(protocol, method)).parameters + + class _DummyStatusProvider(_DummyRecorder): @property def telemetry_status(self) -> TelemetryStatus: diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index c11a939..4a7c852 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -24,6 +24,7 @@ from polygateway.types import ( BackpressurePolicy, BreakerConfig, ChatRequest, + Effort, GlobalLimits, RetryPolicy, SourceConfig, @@ -63,14 +64,21 @@ def _ok(content="ok"): class FakeTransport: - """按脚本逐次返回结果或抛异常;记录每次 (source_name, call_id)。""" + """按脚本逐次返回结果或抛异常;记录每次 (source_name, call_id) 与收到的档位。 + + `reasoning_effort` 刻意**不给默认值**,与 `Transport` 协议保持逐字一致: + `@runtime_checkable` 只查方法名不查签名,fake 上多一个默认值就会把"中间件漏传" + 这类缺口伪装成"调用方没表态",而报错现场离根因很远。 + """ def __init__(self, script): self.script = list(script) self.calls = [] + self.efforts = [] - async def complete(self, *, messages, source, stream, overlay, call_id): + async def complete(self, *, messages, source, stream, overlay, call_id, reasoning_effort): self.calls.append((source.name, call_id)) + self.efforts.append(reasoning_effort) action = self.script.pop(0) if isinstance(action, Exception): raise action @@ -135,6 +143,37 @@ def _harness( _REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}]) +class TestRequestTierReachesTransport: + """请求级档位必须一路穿过洋葱到达 transport(Task 5b)。 + + `ChatRequest` 上填了字段而中间件不搬运,是"看起来配了、实际没发出去"的静默 + 失效——正是 issue #20 里下游改用 extra_body 绕过治理的成因。 + """ + + async def test_request_tier_reaches_transport(self): + mw, _, _, transport, *_ = _harness([_src("a")], [_ok()]) + await mw( + ChatRequest(messages=[{"role": "user", "content": "hi"}], reasoning_effort=Effort.HIGH) + ) + assert transport.efforts == [Effort.HIGH] + + async def test_absent_tier_is_carried_as_none(self): + """不表态也要显式传下去: 漏传与"传了 None"在协议上必须区分不开才安全。""" + mw, _, _, transport, *_ = _harness([_src("a")], [_ok()]) + await mw(_REQ) + assert transport.efforts == [None] + + async def test_tier_is_carried_on_every_retry_attempt(self): + """换源重试时档位不得在第二次尝试上丢失。""" + mw, _, _, transport, *_ = _harness( + [_src("a"), _src("b")], [TransientError("boom", source_name="a"), _ok()] + ) + await mw( + ChatRequest(messages=[{"role": "user", "content": "hi"}], reasoning_effort=Effort.LOW) + ) + assert transport.efforts == [Effort.LOW, Effort.LOW] + + class TestSuccessPath: async def test_first_attempt_success_builds_response(self): mw, limiter, gate, transport, sleep, _ = _harness([_src("a")], [_ok("hello")]) diff --git a/tests/unit/test_usage_source_domain.py b/tests/unit/test_usage_source_domain.py index f1c58d8..7dd3459 100644 --- a/tests/unit/test_usage_source_domain.py +++ b/tests/unit/test_usage_source_domain.py @@ -134,6 +134,7 @@ async def test_salvage_override_stays_in_domain(usage): stream=True, overlay={}, call_id="cid", + reasoning_effort=None, ) assert result.usage_source in USAGE_SOURCES