feat: carry the per-call tier down to the transport that must send it
The Transport port took the request apart into five arguments, so a tier placed on ChatRequest could never reach _build_payload: the field was set, read by nobody, and silently ignored - the exact shape of failure that sent downstream to extra_body in the first place. complete() now takes reasoning_effort with no default, matching the TelemetryRecorder convention: a default would turn a missing hand-off into a silent 'no opinion'. All four fakes move with it, since @runtime_checkable checks method names and not signatures. EmbeddingTransport and OcrTransport are deliberately left alone - they have no reasoning semantics - and a test now holds that line. _build_payload drops its inline sugar conversion for effective_effort(), so the guard and the hot path share one judgement, and passes the source's effort_fallback for the same reason.
This commit is contained in:
@@ -285,6 +285,9 @@ class RetryMW:
|
|||||||
stream=request.stream,
|
stream=request.stream,
|
||||||
overlay=request.overlay,
|
overlay=request.overlay,
|
||||||
call_id=call_id,
|
call_id=call_id,
|
||||||
|
# 逐次尝试原样重传: 换源不改变调用方要的档位(源级默认由 transport
|
||||||
|
# 自己按选中的源解析,两者在 effective_effort 里汇合)
|
||||||
|
reasoning_effort=request.reasoning_effort,
|
||||||
)
|
)
|
||||||
if result.usage_source == "unavailable":
|
if result.usage_source == "unavailable":
|
||||||
# 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回,
|
# 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回,
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ from typing import Any, Protocol, runtime_checkable
|
|||||||
|
|
||||||
from .types import (
|
from .types import (
|
||||||
ChatRequest,
|
ChatRequest,
|
||||||
|
Effort,
|
||||||
EmbeddingTransportResult,
|
EmbeddingTransportResult,
|
||||||
LLMResponse,
|
LLMResponse,
|
||||||
OcrLayoutResult,
|
OcrLayoutResult,
|
||||||
@@ -36,7 +37,16 @@ class Middleware(Protocol):
|
|||||||
|
|
||||||
@runtime_checkable
|
@runtime_checkable
|
||||||
class Transport(Protocol):
|
class Transport(Protocol):
|
||||||
"""一次原始调用的协议细节(请求组装/流式解析/错误翻译);不含任何治理。"""
|
"""一次原始调用的协议细节(请求组装/流式解析/错误翻译);不含任何治理。
|
||||||
|
|
||||||
|
`reasoning_effort` 是本次调用要求的推理档位(`None` = 不表态,随源级配置)。
|
||||||
|
它必须走**协议参数**而不能让 transport 自己去读 `ChatRequest`: 端口只收拆开的
|
||||||
|
请求要素,是为了让 transport 不依赖洋葱内部的请求类型(P7 端口最内层)。
|
||||||
|
|
||||||
|
该参数**不设默认值**,与 `TelemetryRecorder.record_llm_call` 同一既有约定:
|
||||||
|
库外无第三方实现者,写全签名的成本为零,而默认值会把"某一层漏传"变成静默的
|
||||||
|
"调用方没表态"——一次本该报错的漏配就此变成一次悄悄涨价的调用。
|
||||||
|
"""
|
||||||
|
|
||||||
async def complete(
|
async def complete(
|
||||||
self,
|
self,
|
||||||
@@ -46,6 +56,7 @@ class Transport(Protocol):
|
|||||||
stream: bool,
|
stream: bool,
|
||||||
overlay: dict[str, Any],
|
overlay: dict[str, Any],
|
||||||
call_id: str,
|
call_id: str,
|
||||||
|
reasoning_effort: Effort | None,
|
||||||
) -> TransportResult: ...
|
) -> TransportResult: ...
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_ti
|
|||||||
from polygateway.thinking import (
|
from polygateway.thinking import (
|
||||||
ThinkingCapability,
|
ThinkingCapability,
|
||||||
ThinkingUnsupportedError,
|
ThinkingUnsupportedError,
|
||||||
|
effective_effort,
|
||||||
get_capability,
|
get_capability,
|
||||||
observe_thinking,
|
observe_thinking,
|
||||||
reconcile_thinking,
|
reconcile_thinking,
|
||||||
@@ -351,6 +352,7 @@ class OpenAICompatTransport:
|
|||||||
profile: ProviderProfile,
|
profile: ProviderProfile,
|
||||||
stream: bool,
|
stream: bool,
|
||||||
overlay: dict[str, Any],
|
overlay: dict[str, Any],
|
||||||
|
reasoning_effort: Effort | None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream}
|
payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream}
|
||||||
if stream:
|
if stream:
|
||||||
@@ -360,17 +362,19 @@ class OpenAICompatTransport:
|
|||||||
capability = get_capability(source.model, table=self._capabilities)
|
capability = get_capability(source.model, table=self._capabilities)
|
||||||
first_time = source.model not in self._warned_models
|
first_time = source.model not in self._warned_models
|
||||||
self._warned_models.add(source.model)
|
self._warned_models.add(source.model)
|
||||||
# `enable_thinking` 的档位语法糖(True → auto,False → none,None 不表态);
|
# 三层优先级在此汇合: 请求级 > 源级 > enable_thinking 语法糖(设计 §4.2)。
|
||||||
# 就地转换是过渡形态,T5 起由 thinking.effective_effort() 统一收口并接上
|
# 判定与装配守卫共用同一个纯函数,两处分叉就会变成"装配期放行、运行期报错"
|
||||||
# 源级/请求级档位(设计 §4.2)
|
|
||||||
enabled = source.enable_thinking
|
|
||||||
effort = None if enabled is None else (Effort.AUTO if enabled else Effort.NONE)
|
|
||||||
payload.update(
|
payload.update(
|
||||||
resolve_thinking(
|
resolve_thinking(
|
||||||
profile,
|
profile,
|
||||||
capability,
|
capability,
|
||||||
effort,
|
effective_effort(
|
||||||
|
request_effort=reasoning_effort,
|
||||||
|
source_effort=source.reasoning_effort,
|
||||||
|
enable_thinking=source.enable_thinking,
|
||||||
|
),
|
||||||
model=source.model,
|
model=source.model,
|
||||||
|
fallback=source.effort_fallback,
|
||||||
warn_unregistered=first_time,
|
warn_unregistered=first_time,
|
||||||
).payload
|
).payload
|
||||||
)
|
)
|
||||||
@@ -388,12 +392,22 @@ class OpenAICompatTransport:
|
|||||||
stream: bool,
|
stream: bool,
|
||||||
overlay: dict[str, Any],
|
overlay: dict[str, Any],
|
||||||
call_id: str,
|
call_id: str,
|
||||||
|
reasoning_effort: Effort | None,
|
||||||
) -> TransportResult:
|
) -> TransportResult:
|
||||||
"""一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。"""
|
"""一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。
|
||||||
|
|
||||||
|
`reasoning_effort` 是**请求级**档位(`None` = 不表态);它与源级配置的优先级
|
||||||
|
在 `_build_payload` 里由 `effective_effort` 裁定,本层只负责把它送到。
|
||||||
|
"""
|
||||||
profile = get_provider(source.provider, registry=self._registry)
|
profile = get_provider(source.provider, registry=self._registry)
|
||||||
try:
|
try:
|
||||||
payload = self._build_payload(
|
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:
|
except ThinkingUnsupportedError as exc:
|
||||||
# 推理开关不可满足是**请求本身**的问题: 换源重试都救不了它。只捕这个
|
# 推理开关不可满足是**请求本身**的问题: 换源重试都救不了它。只捕这个
|
||||||
|
|||||||
@@ -514,6 +514,7 @@ class TestAssemblyGuardAgainstRealConfig:
|
|||||||
stream=True,
|
stream=True,
|
||||||
overlay={},
|
overlay={},
|
||||||
call_id="e2e-guard",
|
call_id="e2e-guard",
|
||||||
|
reasoning_effort=None,
|
||||||
)
|
)
|
||||||
finally:
|
finally:
|
||||||
await client.aclose()
|
await client.aclose()
|
||||||
|
|||||||
@@ -73,7 +73,7 @@ class ScriptedTransport:
|
|||||||
self.hang = hang
|
self.hang = hang
|
||||||
self.calls: list[str] = []
|
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)
|
self.calls.append(source.name)
|
||||||
if self.hang:
|
if self.hang:
|
||||||
await asyncio.Event().wait()
|
await asyncio.Event().wait()
|
||||||
|
|||||||
@@ -210,7 +210,7 @@ class ClockAdvancingTransport:
|
|||||||
self.clock = clock
|
self.clock = clock
|
||||||
self.calls = []
|
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))
|
self.calls.append((source.name, call_id))
|
||||||
advance, action = self.script.pop(0)
|
advance, action = self.script.pop(0)
|
||||||
self.clock.advance(advance)
|
self.clock.advance(advance)
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ from polygateway.transports.openai_compat import OpenAICompatTransport
|
|||||||
from polygateway.types import (
|
from polygateway.types import (
|
||||||
BackpressurePolicy,
|
BackpressurePolicy,
|
||||||
BreakerConfig,
|
BreakerConfig,
|
||||||
|
Effort,
|
||||||
GlobalLimits,
|
GlobalLimits,
|
||||||
RetryPolicy,
|
RetryPolicy,
|
||||||
SourceConfig,
|
SourceConfig,
|
||||||
@@ -219,6 +220,31 @@ class TestReasoningEffortPriority:
|
|||||||
overrides.setdefault("model", "glm-5.3")
|
overrides.setdefault("model", "glm-5.3")
|
||||||
return _source(provider="zhipu", **overrides)
|
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(
|
@pytest.mark.parametrize(
|
||||||
("provider", "model", "fragment"),
|
("provider", "model", "fragment"),
|
||||||
[
|
[
|
||||||
|
|||||||
@@ -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(
|
return await transport.complete(
|
||||||
messages=[{"role": "user", "content": "hi"}],
|
messages=[{"role": "user", "content": "hi"}],
|
||||||
source=source,
|
source=source,
|
||||||
stream=stream,
|
stream=stream,
|
||||||
overlay=overlay or {},
|
overlay=overlay or {},
|
||||||
call_id="cid-1",
|
call_id="cid-1",
|
||||||
|
reasoning_effort=reasoning_effort,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,15 +1,18 @@
|
|||||||
"""ports.py 端口冻结测试(M1 设计 §4): Protocol 结构性检查 + Gate 快照校验。"""
|
"""ports.py 端口冻结测试(M1 设计 §4): Protocol 结构性检查 + Gate 快照校验。"""
|
||||||
|
|
||||||
|
import inspect
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from polygateway.ports import (
|
from polygateway.ports import (
|
||||||
CacheBackend,
|
CacheBackend,
|
||||||
|
EmbeddingTransport,
|
||||||
GateDecision,
|
GateDecision,
|
||||||
GateState,
|
GateState,
|
||||||
GateUpdate,
|
GateUpdate,
|
||||||
Middleware,
|
Middleware,
|
||||||
|
OcrTransport,
|
||||||
Permit,
|
Permit,
|
||||||
ProviderGate,
|
ProviderGate,
|
||||||
RateLimiter,
|
RateLimiter,
|
||||||
@@ -69,7 +72,7 @@ class _DummyMw:
|
|||||||
|
|
||||||
|
|
||||||
class _DummyTransport:
|
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
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
@@ -142,6 +145,34 @@ def test_protocols_are_runtime_checkable(impl, protocol):
|
|||||||
assert isinstance(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):
|
class _DummyStatusProvider(_DummyRecorder):
|
||||||
@property
|
@property
|
||||||
def telemetry_status(self) -> TelemetryStatus:
|
def telemetry_status(self) -> TelemetryStatus:
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ from polygateway.types import (
|
|||||||
BackpressurePolicy,
|
BackpressurePolicy,
|
||||||
BreakerConfig,
|
BreakerConfig,
|
||||||
ChatRequest,
|
ChatRequest,
|
||||||
|
Effort,
|
||||||
GlobalLimits,
|
GlobalLimits,
|
||||||
RetryPolicy,
|
RetryPolicy,
|
||||||
SourceConfig,
|
SourceConfig,
|
||||||
@@ -63,14 +64,21 @@ def _ok(content="ok"):
|
|||||||
|
|
||||||
|
|
||||||
class FakeTransport:
|
class FakeTransport:
|
||||||
"""按脚本逐次返回结果或抛异常;记录每次 (source_name, call_id)。"""
|
"""按脚本逐次返回结果或抛异常;记录每次 (source_name, call_id) 与收到的档位。
|
||||||
|
|
||||||
|
`reasoning_effort` 刻意**不给默认值**,与 `Transport` 协议保持逐字一致:
|
||||||
|
`@runtime_checkable` 只查方法名不查签名,fake 上多一个默认值就会把"中间件漏传"
|
||||||
|
这类缺口伪装成"调用方没表态",而报错现场离根因很远。
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(self, script):
|
def __init__(self, script):
|
||||||
self.script = list(script)
|
self.script = list(script)
|
||||||
self.calls = []
|
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.calls.append((source.name, call_id))
|
||||||
|
self.efforts.append(reasoning_effort)
|
||||||
action = self.script.pop(0)
|
action = self.script.pop(0)
|
||||||
if isinstance(action, Exception):
|
if isinstance(action, Exception):
|
||||||
raise action
|
raise action
|
||||||
@@ -135,6 +143,37 @@ def _harness(
|
|||||||
_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}])
|
_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:
|
class TestSuccessPath:
|
||||||
async def test_first_attempt_success_builds_response(self):
|
async def test_first_attempt_success_builds_response(self):
|
||||||
mw, limiter, gate, transport, sleep, _ = _harness([_src("a")], [_ok("hello")])
|
mw, limiter, gate, transport, sleep, _ = _harness([_src("a")], [_ok("hello")])
|
||||||
|
|||||||
@@ -134,6 +134,7 @@ async def test_salvage_override_stays_in_domain(usage):
|
|||||||
stream=True,
|
stream=True,
|
||||||
overlay={},
|
overlay={},
|
||||||
call_id="cid",
|
call_id="cid",
|
||||||
|
reasoning_effort=None,
|
||||||
)
|
)
|
||||||
assert result.usage_source in USAGE_SOURCES
|
assert result.usage_source in USAGE_SOURCES
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user