fix: validate ownership of managed reasoning parameters
This commit is contained in:
@@ -441,6 +441,62 @@ def effective_effort(
|
||||
return Effort.AUTO if enable_thinking else Effort.NONE
|
||||
|
||||
|
||||
_RAW_THINKING_ROOTS = frozenset(
|
||||
{
|
||||
"reasoning_effort",
|
||||
"enable_thinking",
|
||||
"thinking",
|
||||
"thinking_budget",
|
||||
"reasoning",
|
||||
"thinkingConfig",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _has_output_effort(raw: Mapping[str, Any]) -> bool:
|
||||
"""标准嵌套强度只检查明确的路径,不猜私有方言。"""
|
||||
output = raw.get("output_config")
|
||||
return isinstance(output, Mapping) and "effort" in output
|
||||
|
||||
|
||||
def validate_thinking_wire(wire: ThinkingWire, *, model: str) -> None:
|
||||
"""拒绝开启片段代选强度,未知形态仍交由请求方向检查。"""
|
||||
base = wire.on_base
|
||||
if base is not None and (
|
||||
"reasoning_effort" in base
|
||||
or (wire.effort_key is not None and wire.effort_key in base)
|
||||
or _has_output_effort(base)
|
||||
):
|
||||
raise ThinkingUnsupportedError(
|
||||
f"模型 {model!r} 的 on_base 不得包含强度档位;请移除强度并显式传 reasoning_effort"
|
||||
)
|
||||
|
||||
|
||||
def validate_thinking_raw(
|
||||
raw: Mapping[str, Any],
|
||||
*,
|
||||
effort: Effort | None,
|
||||
wire: ThinkingWire | None,
|
||||
origin: str,
|
||||
) -> None:
|
||||
"""受管推理只有一个来源;同值或被遮蔽的 raw 控制也拒绝。"""
|
||||
if effort is None:
|
||||
return
|
||||
roots = set(_RAW_THINKING_ROOTS)
|
||||
if wire is not None:
|
||||
for fragment in (wire.on_base, wire.off):
|
||||
if fragment is not None:
|
||||
roots.update(fragment)
|
||||
if wire.effort_key is not None:
|
||||
roots.add(wire.effort_key)
|
||||
if roots.intersection(raw) or _has_output_effort(raw):
|
||||
# 不打印 raw 或自定义键名,防配置中夹带秘密。
|
||||
raise ThinkingUnsupportedError(
|
||||
f"{origin} 与受管 reasoning_effort={effort.value!r} 冲突;"
|
||||
"请删除 raw 推理控制,或移除源级/请求级推理表态后仅用 raw"
|
||||
)
|
||||
|
||||
|
||||
def resolve_thinking(
|
||||
profile: ProviderProfile,
|
||||
capability: ThinkingCapability | None,
|
||||
@@ -490,6 +546,7 @@ def resolve_thinking(
|
||||
而是**静默判否**: Phase 2 按开启方向取字段、Phase 4 整条被绕过,最后在拼错误
|
||||
文案时才以 `AttributeError` 现形(一个未文档化、也不属四分类的异常)。
|
||||
"""
|
||||
validate_thinking_wire(profile.thinking, model=model)
|
||||
# Phase 0: 归一 —— 判据全是身份比较,入口不归一则后面每一关都在拿裸串比枚举
|
||||
if effort is not None:
|
||||
effort = coerce_effort(effort, origin=f"resolve_thinking(model={model!r})")
|
||||
|
||||
@@ -847,3 +847,67 @@ class TestEffectiveEffort:
|
||||
assert (
|
||||
effective_effort(request_effort=None, source_effort=None, enable_thinking=None) is None
|
||||
)
|
||||
|
||||
|
||||
class TestThinkingWireOwnership:
|
||||
"""开启片段只能表达开启,不能携带隐式强度。"""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"base",
|
||||
[
|
||||
{"reasoning_effort": "high"},
|
||||
{"reasoning_effort": None},
|
||||
{"depth": "auto"},
|
||||
{"output_config": {"effort": "low"}},
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("effort", [None, Effort.AUTO, Effort.HIGH])
|
||||
def test_on_base_cannot_hide_a_tier(self, base, effort):
|
||||
profile = ProviderProfile(
|
||||
name="custom",
|
||||
thinking=ThinkingWire(off={"depth": "none"}, on_base=base, effort_key="depth"),
|
||||
strip_think_tags=False,
|
||||
)
|
||||
with pytest.raises(ThinkingUnsupportedError, match="on_base"):
|
||||
resolve_thinking(profile, None, effort, model="custom-model")
|
||||
|
||||
|
||||
class TestThinkingRawOwnership:
|
||||
"""纯规则覆盖标准、自定义根与无意图逃生口。"""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw",
|
||||
[
|
||||
{"reasoning_effort": "high"},
|
||||
{"enable_thinking": True},
|
||||
{"thinking": {}},
|
||||
{"thinking_budget": 100},
|
||||
{"reasoning": {}},
|
||||
{"thinkingConfig": {}},
|
||||
{"output_config": {"effort": None}},
|
||||
{"depth.key": None},
|
||||
{"off_control": {}},
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("effort", [Effort.NONE, Effort.AUTO, Effort.HIGH])
|
||||
def test_control_roots_rejected_without_mutating_input(self, raw, effort):
|
||||
from copy import deepcopy
|
||||
|
||||
from polygateway.thinking import validate_thinking_raw
|
||||
|
||||
wire = ThinkingWire(off={"off_control": False}, on_base={}, effort_key="depth.key")
|
||||
before = deepcopy(raw)
|
||||
with pytest.raises(ThinkingUnsupportedError):
|
||||
validate_thinking_raw(raw, effort=effort, wire=wire, origin="test")
|
||||
assert raw == before
|
||||
validate_thinking_raw(raw, effort=None, wire=wire, origin="test")
|
||||
assert raw == before
|
||||
|
||||
def test_output_format_is_not_effort_unless_wire_owns_root(self):
|
||||
from polygateway.thinking import validate_thinking_raw
|
||||
|
||||
raw = {"output_config": {"format": "json"}, "temperature": 0, "seed": 7}
|
||||
validate_thinking_raw(raw, effort=Effort.AUTO, wire=None, origin="test")
|
||||
wire = ThinkingWire(off=None, on_base={"output_config": {"enabled": True}}, effort_key=None)
|
||||
with pytest.raises(ThinkingUnsupportedError):
|
||||
validate_thinking_raw(raw, effort=Effort.AUTO, wire=wire, origin="test")
|
||||
|
||||
Reference in New Issue
Block a user