"""真实 API 验证推理开关与推理可观测性(issue #5 + #6;判据于 #16/#17 重建)。 本组用例**必须真跑**: 改动的正确性与具体模型强相关,mock 只能验证代码路径, 验证不了"这个参数在这个模型上到底关没关掉推理"。 三条判据纪律(第 1、2 条来自 findings §4c,第 1 条的推翻与第 3 条来自 `findings/2026-08-25-thinking-observability-regression.md`): 1. **判别量是库裁定的三态 `thinking_observation`,既不是 `reasoning_tokens` 也不是 `completion_tokens`。** 长度判据早已排除: 两档的输出长度分布**是 重叠的**(实测关闭档最高 46 token、开启档最低 13 token),按阈值判两边都会 误判。而 `reasoning_tokens` 这个曾经"干净分开"的判据也已失效——MiniMax 这一路上游不再返回 `usage.completion_tokens_details`,该字段恒 `None`;同一 次调用里库明明拿得到 185 字符推理正文,单看 token 计数却把"推理正常"读成 "没推理"(2026-08-25 findings §3.4/结论③,四条用例因此假红)。三态裁定同时 看正文与计数: **正文是事实本身,token 计数只是对事实的转述**。 2. **另配一个不含魔数的确定性锚点**(见 L2b、L5): 同一模型上,关闭档的 `prompt_tokens` 严格小于开启档——供应商在开启时注入了推理指令,输入侧 token 数随之变大。这是相对比较,不硬编码任何具体数值;且它不依赖上游是否 回传推理正文,所以在"观测不到推理"的非流式路径上依然作数。 3. **`UNKNOWN` 不等于"没推理",不能拿它判红。** 关闭方向要求每轮"未观测到 推理"(`UNKNOWN` 计入满足——它没有证伪力),其证伪力来自: 模型若偷偷推理了, 可观测路径会翻成 `OBSERVED`。开启方向只要求多数轮 `OBSERVED`;M3 非流式 路径整片观测不到,该档由 L5 用另一套断言覆盖。 源不可用一律 `skip` 并在报告中记为「未覆盖」,**绝不静默计入通过**。 """ import asyncio import dataclasses import json import os from collections import Counter from collections.abc import Mapping from datetime import datetime from pathlib import Path from types import MappingProxyType import pytest from dotenv import dotenv_values from polygateway import GatewayClient, GatewaySettings, ThinkingObservation from polygateway.errors import ( AllSourcesExhausted, GatewayUnavailableError, RequestRejectedError, SourceDeadError, TransientError, ) from polygateway.thinking import DEFAULT_CAPABILITIES, ThinkingCapability, get_capability from polygateway.types import EFFORT_ORDER, Effort _ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None} _HAS_SOURCE = any(k.split("__")[0] == "LLM" and k.endswith("__API_KEY") for k in _ENV) # slow: 本组 92 次真实调用、约 4 分半(2026-08-26 判据换三态后实测;此前记的 # "137 次、约 7 分钟"已被证伪,别照旧值估 CI 预算),且判据是统计性的——网络抖动 # 会让它偶发失败(实测有一次 network_error 连续三次耗尽源)。让它阻断 `make ci` # 会把测试变成噪声源,故沿用项目既有的 slow 标记默认排除,合并前用 `-m slow` # 显式真跑并存档报告。"不自动门控"不等于"可跳过"。 pytestmark = [ pytest.mark.slow, pytest.mark.skipif( not _HAS_SOURCE, reason="需真实网关凭据: 在 .env 配置 LLM__{PROVIDER}__1__*(本组必须真跑)" ), ] _OUT_DIR = Path("tests/outputs/e2e") _ROUNDS = int(os.environ.get("PGW_E2E_THINKING_ROUNDS", "10")) # 需要一点推理才能答对,但答案极短: 关掉推理时 completion 稳定在个位数, # 开着时则是几百——两档之间隔着一个数量级,判据不必卡在噪声里 _PROMPT = "一个笼子里有若干鸡和兔,共 35 个头、94 只脚。鸡和兔各有多少只?只输出两个数字。" _ROWS: list[dict] = [] # 显式映射,不按模型名猜 provider —— 那正是 D11 要消灭的东西(providers.py 开篇)。 # 漏登记会被 test_every_capability_has_a_provider_mapping 当场抓住,而不是 # 在 L8 里被"源不可用"这个假理由吞掉 _MODEL_PROVIDER = { "MiniMax-M3": "minimax", "MiniMax-M2.7": "minimax", "MiniMax-M2.5": "minimax", "qwen3.7-plus": "qwen", "qwen3.7-max": "qwen", "qwen3.6-plus": "qwen", "qwen3.5-flash": "qwen", "qwen-plus-latest": "qwen", "deepseek-v4-pro": "deepseek", "deepseek-v4-flash": "deepseek", "deepseek-v4-flash-vision-exp": "deepseek", "glm-5.3": "zhipu", "glm-5.3-flash": "zhipu", "glm-5.2": "zhipu", "glm-5.1": "zhipu", "glm-5": "zhipu", "glm-4.6v": "zhipu", "kimi-k3": "moonshot", "kimi-for-coding": "moonshot", "gpt-5.4": "openai", "gpt-5.5": "openai", "claude-opus-5": "anthropic", "claude-sonnet-5": "anthropic", "claude-haiku-5": "anthropic", "gemini-3.1-pro": "google", "gemini-3-flash": "google", } def _base_settings() -> GatewaySettings: # 强制关缓存: 多轮测量要求每一轮都真的打到供应商,命中缓存会把后续轮次 # 变成对第一轮的回放,整组判据随之失效 return GatewaySettings.from_env("LLM", env={**_ENV, "PGW_CACHE_BACKEND": "none"}) def _settings(**source_overrides) -> GatewaySettings: base = _base_settings() source = dataclasses.replace(base.sources[0], **source_overrides) return dataclasses.replace(base, sources=(source,)) async def _run_rounds(rounds: int, *, stream: bool = True, **source_overrides) -> list[dict]: """跑 N 轮真实调用,返回逐轮观测;任一轮抛错即向上冒泡由用例决定处置。""" client = GatewayClient.from_settings(_settings(**source_overrides)) observations = [] try: for i in range(rounds): resp = await client.chat( [{"role": "user", "content": _PROMPT}], stream=stream, # 每轮独立 salt: 即便某层缓存意外开着也不会回放 cache_salt=f"thinking-live-{i}", ) observations.append( { "round": i + 1, "prompt_tokens": resp.prompt_tokens, "completion_tokens": resp.completion_tokens, "reasoning_tokens": resp.reasoning_tokens, # 结论与证据一起入报告: 只记 observation 会让"为什么这么判" # 不可复核,而 thinking_chars 正是本次改判的直接证据 "thinking_observation": resp.thinking_observation, "thinking_chars": len(resp.thinking), "content": resp.content[:60], } ) finally: await client.aclose() return observations def _record(matrix_id: str, desc: str, status: str, detail, observations=None) -> None: _ROWS.append( { "matrix": matrix_id, "desc": desc, "status": status, "detail": detail, "observations": observations or [], } ) def _reasoning_off(obs: dict) -> bool: """关闭方向: 只要没观测到推理即算满足。 `UNKNOWN` 计入满足是有意的: 它没有证伪力(本次无任何信号,判不出来),拿它 判红等于每次关闭调用都喊一遍。本判据真正的证伪力在于——模型若偷偷推理了, 可观测路径会把裁定翻成 `OBSERVED`。 **刻意不设 completion_tokens 上限**: 实测关闭档偶尔会到 46 token(模型没照做 "只输出两个数字",把解题过程写进了正文),而那是正文不是推理。加长度门只会 把这种正常波动误判成"没关掉"。 """ return obs["thinking_observation"] != ThinkingObservation.OBSERVED def _reasoning_on(obs: dict) -> bool: """开启方向: 观测到推理即为真。 判据从 `reasoning_tokens` 换成库的三态裁定,因为 MiniMax 这一路已不再上报 `completion_tokens_details`(2026-08-25 findings 结论②),该字段恒 `None`; 而库在同一次调用里拿得到 185 字符推理正文(findings §3.4)——旧判据看不见 它,L2/L3b/L4/L5 四条因此假红。 也不能退回 completion_tokens 当判据: medium 档的推理量方差极大(实测 15 轮 跨 7-170 token),两档分布还与关闭档重叠,按长度阈值判会把"推理了但想得少" 误判成没推理。 """ return obs["thinking_observation"] == ThinkingObservation.OBSERVED def _skip_if_unreachable(exc: Exception, matrix_id: str, desc: str): """源不可用(渠道下线/模型未开通)→ 跳过并记为未覆盖,不伪装成通过。""" _record(matrix_id, desc, "SKIP(源不可用)", str(exc)[:200]) pytest.skip(f"{matrix_id} 源不可用,已记为未覆盖: {str(exc)[:120]}") @pytest.fixture(scope="module", autouse=True) def _write_report(): yield _OUT_DIR.mkdir(parents=True, exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") path = _OUT_DIR / f"test_thinking_live_{ts}.md" lines = [ "# 推理开关与推理可观测性真实 API 验证", "", f"- 时间: {ts}", f"- 每档轮数: {_ROUNDS}", "- 判别量: 库裁定的三态 `thinking_observation`(OBSERVED/ABSENT/UNKNOWN)," "由推理正文与 reasoning_tokens 共同裁定 —— 正文是事实,token 计数只是转述", "- 关闭判据: **每轮** observation != OBSERVED(UNKNOWN 计入满足,它没有证伪力);" "刻意不设输出长度上限(两档的 completion 分布重叠: 实测关闭档最高 46、开启档最低 13)", "- 开启判据: **多数轮** observation == OBSERVED", "- 确定性锚点(L2b、L5): 关闭档 prompt_tokens 最大值 < 开启档最小值,相对比较无魔数", "- L5(非流式): M3 该路径推理已计费却不回传正文,故不断言「观测到推理」," "改断锚点可分 + 开启档不被误判为 ABSENT", "", "## 矩阵结论", "", "| 矩阵 | 场景 | 结论 | 说明 |", "|---|---|---|---|", ] total_calls = 0 for row in _ROWS: detail = str(row["detail"]).replace("|", "\\|").replace("\n", " ")[:160] lines.append(f"| {row['matrix']} | {row['desc']} | {row['status']} | {detail} |") total_calls += len(row["observations"]) lines += ["", f"**总真实调用次数: {total_calls}**", "", "## 逐轮原始观测", ""] for row in _ROWS: if not row["observations"]: continue lines += [f"### {row['matrix']} — {row['desc']}", "", "```json"] lines.append(json.dumps(row["observations"], ensure_ascii=False, indent=2)) lines += ["```", ""] uncovered = [r["matrix"] for r in _ROWS if r["status"].startswith("SKIP")] if uncovered: lines += ["## 未覆盖", "", f"以下矩阵行未跑到: {', '.join(uncovered)}", ""] path.write_text("\n".join(lines), encoding="utf-8") print(f"\n[e2e 报告] {path}") class TestMiniMaxM3: """M3 是唯一实测可关闭推理的 MiniMax 模型,修复的地基压在它身上。""" async def test_l1_disable_actually_disables(self): obs = await _run_rounds(_ROUNDS, model="MiniMax-M3", enable_thinking=False) offs = [o for o in obs if _reasoning_off(o)] _record( "L1", "enable_thinking=False(流式)", "PASS" if len(offs) == len(obs) else "FAIL", f"{len(offs)}/{len(obs)} 轮未观测到推理", obs, ) assert len(offs) == len(obs), f"关闭方向要求每轮满足: {obs}" async def test_l2_enable_actually_enables(self): obs = await _run_rounds(_ROUNDS, model="MiniMax-M3", enable_thinking=True) ons = [o for o in obs if _reasoning_on(o)] _record( "L2", "enable_thinking=True(流式,注入 medium)", "PASS" if len(ons) * 2 > len(obs) else "FAIL", f"{len(ons)}/{len(obs)} 轮观察到推理", obs, ) assert len(ons) * 2 > len(obs), f"开启方向要求多数轮满足: {obs}" async def test_l2b_off_and_on_are_distinguishable_without_magic_numbers(self): """确定性锚点: 开启档的 prompt_tokens 严格大于关闭档。 供应商在开启推理时会向模板注入推理指令,输入侧 token 数随之变大。这是 本组唯一不依赖输出侧噪声的证据,且是相对比较——不硬编码任何具体数值, 供应商改模板也不会让它假红。 """ rounds = max(3, _ROUNDS // 3) off = await _run_rounds(rounds, model="MiniMax-M3", enable_thinking=False) on = await _run_rounds(rounds, model="MiniMax-M3", enable_thinking=True) off_max = max(o["prompt_tokens"] for o in off) on_min = min(o["prompt_tokens"] for o in on) _record( "L2b", "关闭/开启的 prompt_tokens 可分", "PASS" if off_max < on_min else "FAIL", f"关闭档最大 {off_max} < 开启档最小 {on_min}", off + on, ) assert off_max < on_min, ( f"两档的 prompt_tokens 未分开(关闭最大 {off_max},开启最小 {on_min}): 注入可能没到达模型" ) async def test_l3_no_opinion_is_the_model_default(self): obs = await _run_rounds(_ROUNDS, model="MiniMax-M3", enable_thinking=None) # M3 的默认档实测就是不推理(findings §2.1),所以不干预时也应观测不到推理。 # 注意这**不能**反过来证明关闭方向生效 —— L1 与本行同分布,区分二者的是 # L2b 的 prompt_tokens 与 L3b 的乱码值反证 quiet = [o for o in obs if _reasoning_off(o)] _record( "L3", "enable_thinking=None(不干预,基线)", "PASS" if len(quiet) == len(obs) else "FAIL", f"{len(quiet)}/{len(obs)} 轮未观测到推理(M3 默认档本就不推理)", obs, ) assert len(quiet) == len(obs), f"M3 默认档不应推理: {obs}" async def test_l3b_none_is_recognised_not_silently_dropped(self): """反证: 关闭方向的观测必须排除"参数被静默丢弃"这一伪解释。 L1(关闭)与 L3(不干预)在 M3 上**同分布**——因为 M3 默认档本就不推理。 所以 L1 单独看不能区分"`none` 真的被消费"与"`none` 被中转吞了",而后者 正是 issue #5 的原始故障形态(`enable_thinking` 就是这么被吞的)。 判别方法: 发一个**非法值**。若未知值会被静默丢弃,它的表现应与"不注入" 一致(不推理);实测它反而开启了推理,说明网关认这个键、只是不认这个值。 既然非法值与 `none` 的表现不同,`none` 就必然是被识别的枚举值。 **该手法不可移植,只对"认这个键但不校验值"的 provider 成立**: minimax 对 非法 `reasoning_effort` 返回 200 且照常推理(2026-08-25 findings §5: prompt 207,介于基线 194 与 medium 216 之间,走了第三条模板路径);而 qwen 对同样的值直接返回 **HTTP 400**。把本用例套到 qwen 那类会校验值的 provider 上,拿到的会是异常而非"不推理",是假红。 """ rounds = max(3, _ROUNDS // 3) bogus = await _run_rounds( rounds, model="MiniMax-M3", enable_thinking=None, extra_body={"reasoning_effort": "definitely-not-a-real-level"}, ) off = await _run_rounds(rounds, model="MiniMax-M3", enable_thinking=False) bogus_on = [o for o in bogus if _reasoning_on(o)] off_quiet = [o for o in off if _reasoning_off(o)] ok = len(bogus_on) * 2 > len(bogus) and len(off_quiet) == len(off) _record( "L3b", "非法值反证 none 被识别", "PASS" if ok else "FAIL", f"非法值 {len(bogus_on)}/{len(bogus)} 轮推理,none {len(off_quiet)}/{len(off)} 轮不推理" "(两者表现不同 ⇒ none 非被丢弃)", bogus + off, ) assert len(bogus_on) * 2 > len(bogus), ( f"非法值未开启推理,无法排除'未知值被静默丢弃'这一伪解释: {bogus}" ) assert len(off_quiet) == len(off), f"none 未关闭推理: {off}" async def test_l4_extra_body_overrides_the_profile(self): """profile 注入 none,extra_body 要求 high —— 后者必须赢(优先级不可调换)。 判据是行为而非报文: 若 extra_body 没赢,拿到的就是 none 的结果(不推理)。 """ rounds = max(3, _ROUNDS // 2) obs = await _run_rounds( rounds, model="MiniMax-M3", enable_thinking=False, extra_body={"reasoning_effort": "high"}, ) ons = [o for o in obs if _reasoning_on(o)] _record( "L4", "extra_body 覆盖 profile 注入", "PASS" if len(ons) * 2 > len(obs) else "FAIL", f"{len(ons)}/{len(obs)} 轮观察到推理(证明 high 生效而非 none)", obs, ) assert len(ons) * 2 > len(obs), f"extra_body 未能覆盖 profile: {obs}" async def test_l5_non_stream_path_is_distinguishable_and_honestly_unknown(self): """非流式快路径: 参数确实到达了模型,而推理信号被如实标成"观测不到"。 **本用例不能断言"非流式开启档观测到推理"——那永远不成立**: M3 在非流式 路径下推理段确实产生并计费(2026-08-25 findings §3.4: 开启档 completion 53 vs 关闭档 3),但 `message` 里没有 `reasoning_content`、`usage` 里也没有 `completion_tokens_details`,推理内容整体不回传。**这是上游行为,库修不了; 库能做也必须做的是让它可见**——下游在为看不见的东西付费,不该由库替它 沉默。 故改断两件在非流式下真实成立的事: 其一 `prompt_tokens` 锚点仍把两档分开(判据形态照抄 L2b,证明注入到达了模型, 排除"非流式路径把参数弄丢了"这一伪解释); 其二开启档的裁定**不是 `ABSENT`**——`ABSENT` 的语义是"上游明确上报未推理", 而实情是"判不出来"(`UNKNOWN`),库若把后者伪装成前者,正是 issue #16/#17 里 那个静默错觉。这里断 `!= ABSENT` 而非 `== UNKNOWN`,是为了留出上游哪天开始 回传正文的余地: 那时裁定会翻成 `OBSERVED`,是好事,不该让它把测试判红。 """ rounds = max(3, _ROUNDS // 2) off = await _run_rounds(rounds, stream=False, model="MiniMax-M3", enable_thinking=False) on = await _run_rounds(rounds, stream=False, model="MiniMax-M3", enable_thinking=True) offs = [o for o in off if _reasoning_off(o)] off_max = max(o["prompt_tokens"] for o in off) on_min = min(o["prompt_tokens"] for o in on) not_absent = [o for o in on if o["thinking_observation"] != ThinkingObservation.ABSENT] on_states = Counter(str(o["thinking_observation"]) for o in on) ok = len(offs) == len(off) and off_max < on_min and len(not_absent) == len(on) _record( "L5", "非流式: prompt 锚点可分 + 开启档如实标 UNKNOWN 而非 ABSENT", "PASS" if ok else "FAIL", f"关闭 {len(offs)}/{len(off)} 轮未观测到推理;" f"关闭档 prompt 最大 {off_max} < 开启档最小 {on_min};" f"开启档裁定分布 {dict(on_states)}", off + on, ) assert len(offs) == len(off), f"非流式关闭方向未满足: {off}" assert off_max < on_min, ( f"非流式两档 prompt_tokens 未分开(关闭最大 {off_max},开启最小 {on_min}): " f"开启参数可能没到达模型" ) assert len(not_absent) == len(on), ( f"非流式开启档被裁成 ABSENT(声称上游明确上报未推理),而实情是观测不到: {on}" ) class TestOtherProviders: """qwen / deepseek 的 profile 是既有实现,本组防的是"改 minimax 时误伤它们"。""" @pytest.mark.parametrize( ("matrix", "provider", "model"), [("L6", "qwen", "qwen3.7-plus"), ("L7", "deepseek", "deepseek-v4-pro")], ) async def test_existing_profiles_still_disable(self, matrix, provider, model): desc = f"{provider} enable_thinking=False" try: obs = await _run_rounds(_ROUNDS, provider=provider, model=model, enable_thinking=False) except (AllSourcesExhausted, SourceDeadError, TransientError) as exc: # 只吞网关/网络类失败。**不吞 ValueError / RequestRejected** —— # 那两类正是本次改动最可能的误伤方向,吞掉就成了纪律(c)要防的静默 _skip_if_unreachable(exc, matrix, desc) offs = [o for o in obs if _reasoning_off(o)] _record( matrix, desc, "PASS" if len(offs) == len(obs) else "FAIL", f"{len(offs)}/{len(obs)} 轮未观测到推理", obs, ) assert len(offs) == len(obs), f"{provider} 关闭方向未满足: {obs}" async def test_qwen_enabled_is_observed(self): """设计 §14 验收: qwen 开启档必须裁定为 `OBSERVED`,不是 `UNKNOWN`。 本条是三态裁定的**跨供应商对照组**: MiniMax 这一路两个信号都可能缺失 (非流式档整片 `UNKNOWN`),若只按它调判据,很容易把"观测不到"当成常态; qwen 在同一网关同一 key 上照常返回推理信号(findings 2026-08-25 §2), 故这里能且必须要求正面结论——它一旦掉成 `UNKNOWN`,说明的是库的组装路径 丢了信号,而不是上游行为变了。 """ matrix, provider, model = "L6b", "qwen", "qwen3.7-plus" desc = f"{provider} enable_thinking=True" try: obs = await _run_rounds(_ROUNDS, provider=provider, model=model, enable_thinking=True) except (AllSourcesExhausted, SourceDeadError, TransientError) as exc: _skip_if_unreachable(exc, matrix, desc) ons = [o for o in obs if _reasoning_on(o)] _record( matrix, desc, "PASS" if len(ons) * 2 > len(obs) else "FAIL", f"{len(ons)}/{len(obs)} 轮观测到推理(OBSERVED)", obs, ) assert len(ons) * 2 > len(obs), f"{provider} 开启方向要求多数轮 OBSERVED: {obs}" class TestCapabilityDrift: """L8 漂移哨兵: 能力表过期是必然事件,这里是它的过期告警。""" def test_every_capability_has_a_provider_mapping(self): """能力表新增条目必须同步本测试的映射,否则该行会被静默跳过。""" missing = sorted(set(DEFAULT_CAPABILITIES) - set(_MODEL_PROVIDER)) assert not missing, f"这些模型缺 provider 映射,L8 会漏测: {missing}" @pytest.mark.parametrize("model", sorted(DEFAULT_CAPABILITIES)) async def test_declared_capability_matches_reality(self, model): cap = get_capability(model) provider = _MODEL_PROVIDER[model] rounds = max(3, _ROUNDS // 2) desc = f"{model} 声明 can_disable={cap.can_disable}" if not cap.can_disable: # 声明关不掉: 装配期就该炸,炸了即与声明一致(不必真调用) with pytest.raises(ValueError, match=model): GatewayClient.from_settings( _settings(provider=provider, model=model, enable_thinking=False) ) _record("L8", desc, "PASS", "装配期按声明拒绝,与实测一致") return try: obs = await _run_rounds(rounds, provider=provider, model=model, enable_thinking=False) except (AllSourcesExhausted, SourceDeadError, TransientError) as exc: _skip_if_unreachable(exc, "L8", desc) offs = [o for o in obs if _reasoning_off(o)] verdict = Counter(_reasoning_off(o) for o in obs) _record( "L8", desc, "PASS" if len(offs) == len(obs) else "FAIL(能力表已漂移)", f"实测未观测到推理 {dict(verdict)}(True=满足);声明 can_disable=True 要求每轮满足", obs, ) assert len(offs) == len(obs), ( f"能力表漂移: {model} 声明可关闭推理,实测未关掉 —— 请复测后更新 DEFAULT_CAPABILITIES" ) class TestAssemblyGuardAgainstRealConfig: """L9: 纯本地,但用的是 .env 里的真实配置形态,防"守卫只在合成配置上生效"。""" def test_l9_m27_rejected_at_assembly(self): with pytest.raises(ValueError, match="MiniMax-M2.7"): GatewayClient.from_settings( _settings(provider="minimax", model="MiniMax-M2.7", enable_thinking=False) ) _record("L9", "M2.7 + enable_thinking=False", "PASS", "装配期报错,未发出任何请求") def test_l9_unknown_shape_rejected_at_assembly(self): with pytest.raises(ValueError, match="register_provider"): GatewayClient.from_settings( _settings(provider="openai", model="kimi-k3", enable_thinking=False) ) _record("L9", "provider=openai 形态未知", "PASS", "装配期报错并指路") async def test_transport_layer_rejects_when_guard_is_bypassed(self): """构造函数全量注入这条路绕过装配守卫,transport 必须兜住并归四分类。""" settings = _settings(provider="minimax", model="MiniMax-M2.7", enable_thinking=False) client = GatewayClient.from_settings( dataclasses.replace( settings, sources=(dataclasses.replace(settings.sources[0], enable_thinking=None),) ) ) try: # 装配用 None 绕过守卫,再把源换成 False 直接喂给 transport bad = dataclasses.replace(settings.sources[0], enable_thinking=False) with pytest.raises(RequestRejectedError, match="MiniMax-M2.7"): await client._terminal._transport.complete( messages=[{"role": "user", "content": _PROMPT}], source=bad, stream=True, overlay={}, call_id="e2e-guard", reasoning_effort=None, ) finally: await client.aclose() _record("L9", "绕过装配守卫时 transport 兜底", "PASS", "RequestRejectedError,属四分类") # ══════════════════════════════════════════════════════════════════════════════ # T10: 逐模型档位实测(方法论沿用 issue #20) # # 本节与上面的 L1-L9 分工不同: 上面验的是**库的行为**(注入到没到、观测准不准), # 这里验的是**能力表的内容**(`DEFAULT_CAPABILITIES` 里那 20 多条声明是不是真的)。 # 二者判据可以共用,数据源却必须分开——能力表实测要**绕过能力表**才有意义, # 否则拿待验证的声明去挡请求,等于用结论证明前提。 # # 判据(三条,均沿用已有纪律): # ① 关闭方向: 每轮 `thinking_observation != OBSERVED` 才算真关掉;任一轮 # OBSERVED 即证伪(推理正文是事实本身,不需要多数票)。 # ② **短提示词的"关掉了"必须经长上下文复核**: issue #20 实测 GLM 系在短提示词 # 下 reasoning_tokens≈1.2 像是关了,5552 token 长上下文下跳到 0/54/167 即露馅。 # 短提示词下推理量本就趋近于 0,分不出"关了"与"没什么可想的"。 # ③ 开启方向: 多数轮 OBSERVED(单轮抖动不判红,与 L2 同口径)。 # ④ 关闭结论**不许只靠 `UNKNOWN`**: 上游整片不回传推理信号时(kimi、MiniMax 两路 # 都是),"没看见"不是"没发生"。此时补一个不含魔数的锚点——关闭档的 # `completion_tokens` 必须严格小于 `max` 档,否则结论记为「判不出来」。 # ══════════════════════════════════════════════════════════════════════════════ _TIER_OUT_DIR = Path("tests/outputs/thinking") _TIER_ROUNDS = int(os.environ.get("PGW_E2E_TIER_ROUNDS", "5")) _TIER_LONG_ROUNDS = int(os.environ.get("PGW_E2E_TIER_LONG_ROUNDS", "3")) # 共用生产网关,宁慢勿冲(人类 2026-09-05 指令): 默认 3,可下调不建议上调 _TIER_CONCURRENCY = int(os.environ.get("PGW_E2E_TIER_CONCURRENCY", "3")) # 固定短提示词: 答案本身约 4 token,推理 token 的信噪比高(issue #20 同款) _TIER_PROMPT = "23 乘以 47 等于多少?只回答一个数字,不要解释。" # 长上下文对照组(判据②)。填充文本与题目无关且不含任何业务领域词汇(零业务假设 # 铁律),只为把输入撑到数千 token;题目放在最后,避免被当成"读完就忘"的前缀 _TIER_LONG_PROMPT = ( "\n".join( f"{i:04d}. 这是一段与题目无关的填充文字,仅用于把上下文撑到数千 token," "以复核短提示词下得到的关闭结论在长上下文下是否依然成立。" for i in range(120) ) + "\n\n" + _TIER_PROMPT ) _ALL_EFFORTS: tuple[Effort, ...] = (*EFFORT_ORDER, Effort.AUTO) _PROBE_ROWS: list[dict] = [] def _tier_settings(model: str) -> GatewaySettings: """探测用配置: 生产口径的超时,但**重试预算压到 1 次**。 压重试是因为探测里"这一轮失败"本身就是数据(逐轮进报告),库替它重试只会 把"渠道当下不可用"变成三倍等待——2026-09-05 实测 claude 系 7 天限额用尽时 每轮 429,三次重试让单个模型阻塞三分钟以上,26 个模型跑不完。 **单次请求的超时不动**(仍是 .env 的生产值 300s): §4.6 那条"测试超时不得紧于 生产配置"防的是把慢而正常的模型误判成不可用,那个风险在这里照旧存在。重试次数 与背压窗口不属于同一类——它们决定"失败之后还等多久",而不是"多慢算失败"; 一个真在出字的模型永远碰不到这两者。 """ base = GatewaySettings.from_env( "LLM", env={ **_ENV, "PGW_CACHE_BACKEND": "none", "LLM_MAX_RETRIES": "1", # 探测是**单源**的,没有别的源可换。生产值 1200s 的 stall window 在这里 # 只会把"这个模型当下不可用"拖成 20 分钟一轮: 2026-09-05 实测 claude 系 # 7 天限额用尽返回 429 且不带 Retry-After,库据此判"无可运行源"并按背压 # 语义等到窗口耗尽(实测把窗口调到 45s 即在 46.7s 报 stalled)。多源生产 # 场景下这段等待是有意义的(等别的源恢复),探测场景下等不到任何东西 "LLM__BACKPRESSURE__STALL_WINDOW_S": "60", }, ) source = dataclasses.replace( base.sources[0], provider=_MODEL_PROVIDER[model], model=model, enable_thinking=None, reasoning_effort=None, ) return dataclasses.replace(base, sources=(source,)) def _probe_capabilities(model: str) -> dict[str, ThinkingCapability]: """临时全档能力表: **实测的对象正是能力表本身**,不能拿它当前提去挡请求。 不传 `capabilities={}`(即"未登记")的理由是噪声: 那条路会走 Phase 3,每轮都 warning 一句"能力未登记",几百轮下来把真正的告警淹没。全档表让五关全部放行, 请求原样发出去,由上游而不是由库来回答"这一档到底行不行"。 """ return {model: ThinkingCapability(_ALL_EFFORTS, evidence="T10 实测临时表(不进 DEFAULT)")} async def _probe_effort( model: str, effort: Effort, *, rounds: int, prompt: str, prompt_kind: str ) -> list[dict]: """对一个 (模型, 档位) 打 N 轮真实请求,逐轮记录;失败轮记 `error` 而不冒泡。 失败不冒泡是本函数与 `_run_rounds` 的唯一区别: 这里"上游拒绝这一档"本身就是 **实测结论**(HTTP 400 = 该档不被接受),把它抛出去会让数据采集半途而废。 只吞四分类与 `AllSourcesExhausted`——库自身的 `ValueError` 等仍然冒泡,那是 bug 不是数据。 """ client = GatewayClient.from_settings( _tier_settings(model), capabilities=_probe_capabilities(model) ) semaphore = asyncio.Semaphore(_TIER_CONCURRENCY) async def _one(index: int) -> dict: base = {"round": index + 1, "effort": effort.value, "prompt_kind": prompt_kind} async with semaphore: try: resp = await client.chat( [{"role": "user", "content": prompt}], stream=True, reasoning_effort=effort, cache_salt=f"tier-probe-{model}-{effort.value}-{prompt_kind}-{index}", ) except ( RequestRejectedError, GatewayUnavailableError, SourceDeadError, TransientError, ) as exc: # 捕 `GatewayUnavailableError` 而不是只捕 `AllSourcesExhausted`: # 某个模型在网关上不通时,连续失败会把熔断门打开,后续轮次抛的是 # `CircuitOpenError`(同一父类的兄弟)。只捕子类会让"源不可用"这 # 件事在第 N 轮换个类型冒出去,把数据采集打断成一次红测 return {**base, "error": f"{type(exc).__name__}: {str(exc)[:160]}"} return { **base, "error": None, "prompt_tokens": resp.prompt_tokens, "completion_tokens": resp.completion_tokens, "reasoning_tokens": resp.reasoning_tokens, "thinking_chars": len(resp.thinking), "thinking_observation": resp.thinking_observation, "applied_effort": resp.applied_effort, # 核对模型身份: issue #20 记录本渠道对 glm-5.2 的请求 6/6 回报 # model=glm-5.3。凡结论依赖模型身份的,对不上即数据不可信 "model_reported": resp.model_reported, "content": resp.content[:40], } try: return list(await asyncio.gather(*(_one(i) for i in range(rounds)))) finally: await client.aclose() def _probe_ok(obs: dict) -> bool: return obs["error"] is None def _probe_quiet(obs: dict) -> bool: """成功且未观测到推理(判据①的满足条件);失败轮不算"安静",它没有观测。""" return _probe_ok(obs) and obs["thinking_observation"] != ThinkingObservation.OBSERVED def _probe_observed(obs: dict) -> bool: return _probe_ok(obs) and obs["thinking_observation"] == ThinkingObservation.OBSERVED def _rt_summary(observations: list[dict]) -> str: """报告里的一行摘要: rt 观测值序列 + 裁定分布 + 身份核对,三样缺一不可复核。""" ok = [o for o in observations if _probe_ok(o)] if not ok: return f"全部 {len(observations)} 轮失败: {observations[0]['error']}" rts = [o["reasoning_tokens"] for o in ok] verdicts = Counter(str(o["thinking_observation"]) for o in ok) reported = sorted({str(o["model_reported"]) for o in ok}) failed = len(observations) - len(ok) tail = f";{failed} 轮失败" if failed else "" return ( f"rt={rts};裁定 {dict(verdicts)};thinking_chars=" f"{[o['thinking_chars'] for o in ok]};model_reported={reported}{tail}" ) # 已知的合法别名: 供应商回报的名字与配置里的别名本就可以不同(月之暗面回 # `k3`、Google 回 `-preview` 后缀)。**显式登记而不是按前缀猜**——猜的话 # `glm-5.2 → glm-5.3` 这种真·串台也会被当成"同族别名"放过,而那正是本表要抓的 _MODEL_REPORTED_ALIASES: Mapping[str, frozenset[str]] = MappingProxyType( { "kimi-k3": frozenset({"k3"}), "kimi-for-coding": frozenset({"k3"}), "gemini-3-flash": frozenset({"gemini-3-flash-preview"}), "gemini-3.1-pro": frozenset({"gemini-3.1-pro-preview"}), } ) def _identity_mismatch(model: str, observations: list[dict]) -> list[str]: """响应体里的 `model` 与请求的模型对不上 → 本次数据说的不是这个模型。 issue #20 就栽在这里: 该渠道对 `glm-5.2` 的请求 6/6 回报 `model=glm-5.3`, 照单全收的话,能力表里 glm-5.2 那一行记的其实是 glm-5.3 的行为。凡结论依赖 模型身份的,对不上就必须当场作废,而不是打个折扣继续用。 `None`(上游未上报)不算不符: 那是"没说",不是"说了别的"。 """ allowed = {model, *_MODEL_REPORTED_ALIASES.get(model, frozenset())} return sorted( { o["model_reported"] for o in observations if _probe_ok(o) and o["model_reported"] is not None and o["model_reported"] not in allowed } ) async def _anchor_off_against_on( model: str, off_observations: list[dict] ) -> tuple[list[dict], Effort | None, bool]: """判据④: 拿"开启档的 completion 明显更大"给关闭结论补一个正面证据。 需要它是因为 `UNKNOWN` 的语义: 它是"本次没有任何信号,判不出来",不是"没推理" (`observe_thinking` 的 docstring 把这条写死了)。kimi 与 MiniMax 这两路上游都 不回传 `completion_tokens_details`,关闭档整片 `UNKNOWN`——此时若直接把"没看见" 读成"关掉了",库就会登记一个自己从未验证过的 `none`,而下游据此以为省了钱。 锚点取 `completion_tokens` 的相对比较(关闭档最大值 < 开启档最小值),**不含 任何魔数**: 推理段计在 completion 里,真开着时两档差一个数量级(实测 kimi-k3 关闭档恒 9 token)。取 `max` 档而非 `auto`: 后者对 minimax 一路等于"什么都不注入" (`on_base={}`),那是模型默认档而不是"开",拿它当对照组会把 M3 这种默认不推理的 模型判成"分不开"。`max` 打不通时才退到 `auto`。 """ off_usable = [o for o in off_observations if _probe_ok(o)] for tier in (Effort.MAX, Effort.AUTO): anchor = await _probe_effort( model, tier, rounds=_TIER_LONG_ROUNDS, prompt=_TIER_PROMPT, prompt_kind=f"anchor({tier.value})", ) on_usable = [o for o in anchor if _probe_ok(o)] if not on_usable: continue off_max = max(o["completion_tokens"] for o in off_usable) on_min = min(o["completion_tokens"] for o in on_usable) return anchor, tier, off_max < on_min return [], None, False def _probe_record(model: str, phase: str, verdict: str, detail: str, observations: list[dict]): _PROBE_ROWS.append( { "model": model, "provider": _MODEL_PROVIDER[model], "phase": phase, "verdict": verdict, "detail": detail, "observations": observations, } ) @pytest.fixture(scope="module", autouse=True) def _write_tier_report(): """T10 报告独立成文件: 它的读者是"能力表该怎么改",与 L1-L9 的"库对不对"不同。""" yield if not _PROBE_ROWS: return _TIER_OUT_DIR.mkdir(parents=True, exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") path = _TIER_OUT_DIR / f"tier_probe_{ts}.md" lines = [ "# 推理档位能力表实测(T10,经 new-api 中转)", "", f"- 时间: {ts}", f"- 短提示词轮数: {_TIER_ROUNDS};长上下文复核轮数: {_TIER_LONG_ROUNDS};" f"并发: {_TIER_CONCURRENCY}(共用生产网关,宁慢勿冲)", f"- 短提示词: `{_TIER_PROMPT}`", f"- 长上下文: 同题 + {len(_TIER_LONG_PROMPT)} 字符无关填充(判据②)", "- 判据: 关闭方向要求**每轮**未观测到推理,且短提示词的「关掉了」必须经长上下文复核;" "开启方向要求多数轮 OBSERVED", "- 能力表在探测时被临时替换为全档表: 实测的对象正是它,不能拿它挡请求", "", "## 逐模型结论", "", "| 模型 | provider | 阶段 | 结论 | 观测 |", "|---|---|---|---|---|", ] total = 0 for row in _PROBE_ROWS: detail = str(row["detail"]).replace("|", "\\|").replace("\n", " ")[:220] lines.append( f"| {row['model']} | {row['provider']} | {row['phase']} | {row['verdict']} | {detail} |" ) total += len(row["observations"]) lines += ["", f"**总真实调用次数: {total}**", "", "## 逐轮原始观测", ""] for row in _PROBE_ROWS: if not row["observations"]: continue lines += [f"### {row['model']} — {row['phase']}", "", "```json"] lines.append(json.dumps(row["observations"], ensure_ascii=False, indent=2, default=str)) lines += ["```", ""] path.write_text("\n".join(lines), encoding="utf-8") print(f"\n[T10 报告] {path}") class TestTierProbe: """能力表实测。可只跑单个模型: `-k "test_t10 and glm-5.3"`。""" @pytest.mark.parametrize("model", sorted(_MODEL_PROVIDER)) async def test_t10_none_direction_matches_declaration(self, model): """「这个模型到底关不关得掉」——能力表里唯一会**报错**的那条声明。 它是本节最要紧的一条: `Effort.NONE` 在不在清单里,决定 Phase 4 是放行还是 当场报错。声明错了,两个方向的代价都很实在——多写了 `none` 会让下游以为 关掉了(issue #20 的静默失效),漏写了会把一条本来可用的路堵死。 """ short = await _probe_effort( model, Effort.NONE, rounds=_TIER_ROUNDS, prompt=_TIER_PROMPT, prompt_kind="short" ) # 上游拒绝这一档(400)是**结论**而非故障: 它等价于"关不掉"; # 其余失败(渠道下线/超时)才是源不可用,按既有纪律记为未覆盖 rejected = [o for o in short if o["error"] and o["error"].startswith("RequestRejected")] usable = [o for o in short if _probe_ok(o)] # **按可用轮判,而不是一有失败就整条跳过**: 共用网关上偶发 429/503 是常态, # 一票否决会让整张表因为一次抖动而没有数据。样本低于 3 轮才是真的没结论 if not rejected and len(usable) < min(3, _TIER_ROUNDS): broken = [o for o in short if o["error"]] _probe_record(model, "none 方向", "SKIP(源不可用)", _rt_summary(short), short) pytest.skip(f"{model} 源不可用,已记为未覆盖: {broken[0]['error'][:120]}") strangers = _identity_mismatch(model, short) if strangers: _probe_record( model, "none 方向", "SKIP(身份不符,数据不可信)", f"该渠道把请求回报成 {strangers};{_rt_summary(short)}", short, ) pytest.skip(f"{model} 被该渠道路由到 {strangers},本次观测说的不是这个模型") observations = list(short) measured_can_disable = not rejected and all(_probe_quiet(o) for o in usable) note = "" if measured_can_disable: # 判据②: 短提示词下"看起来关了"必须过长上下文这一关 long_ctx = await _probe_effort( model, Effort.NONE, rounds=_TIER_LONG_ROUNDS, prompt=_TIER_LONG_PROMPT, prompt_kind="long", ) observations += long_ctx usable = [o for o in long_ctx if _probe_ok(o)] if not usable: note = ";长上下文复核未跑通,结论只在短提示词下成立" else: measured_can_disable = all(_probe_quiet(o) for o in usable) note = ";长上下文复核" + ("同样未观测到推理" if measured_can_disable else "露馅") if measured_can_disable and not any( o["thinking_observation"] is ThinkingObservation.ABSENT for o in observations ): # 判据④: 全程 `UNKNOWN` 时,"关掉了"是一句没有正面证据的话 anchor, anchor_tier, separable = await _anchor_off_against_on(model, observations) observations += anchor if anchor_tier is None: note += ";锚点未跑通,关闭结论缺正面证据" elif separable: note += f";锚点可分(关闭档 completion 严格小于 {anchor_tier.value} 档)" else: measured_can_disable = None note += f";**锚点不可分**(与 {anchor_tier.value} 档的 completion 分不开),判不出来" detail = f"实测 can_disable={measured_can_disable}{note}。短: {_rt_summary(short)}" + ( f" ‖ 后续: {_rt_summary(observations[len(short) :])}" if len(observations) > len(short) else "" ) if measured_can_disable is None: _probe_record(model, "none 方向", "INCONCLUSIVE(无正面证据)", detail, observations) pytest.skip(f"{model} 判不出来,已记为未覆盖: {detail[:160]}") capability = get_capability(model) if capability is None: _probe_record(model, "none 方向", "DATA(未登记)", detail, observations) pytest.skip(f"{model} 未登记(设计 §8 第三档),本条只采数据: {detail[:120]}") agrees = measured_can_disable == capability.can_disable _probe_record( model, "none 方向", "PASS" if agrees else "FAIL(能力表已漂移)", f"声明 can_disable={capability.can_disable};{detail}", observations, ) assert agrees, ( f"{model} 的能力表与实测不符: 声明 can_disable={capability.can_disable}," f"实测 {measured_can_disable}。{detail}" ) @pytest.mark.parametrize("model", sorted(DEFAULT_CAPABILITIES)) async def test_t10_declared_tiers_actually_reason(self, model): """已登记的每个**开启档**都必须被上游接受,且真的推理。 证伪力只在"被拒"与"没推理"两件事上——**不断言档位之间的 rt 高低**: 设计 §4.3 已定,同一档 rt 实测在 8~56 之间跳,拿它比大小必然是噪声。 故本条能证伪的是"登记了一个上游根本不认的档",不是"档位排序对不对"。 """ capability = get_capability(model) tiers = [e for e in capability.supported_efforts if e is not Effort.NONE] if not tiers: pytest.skip(f"{model} 只登记了 none,没有开启档可验") failures = [] for tier in tiers: observations = await _probe_effort( model, tier, rounds=_TIER_ROUNDS, prompt=_TIER_PROMPT, prompt_kind="short" ) rejected = [ o for o in observations if o["error"] and o["error"].startswith("RequestRejected") ] usable = [o for o in observations if _probe_ok(o)] observed = [o for o in observations if _probe_observed(o)] strangers = _identity_mismatch(model, observations) if strangers: # 与 none 方向同一条纪律: 回报的不是这个模型,这组数就不是它的 _probe_record( model, f"档位 {tier.value}", "SKIP(身份不符,数据不可信)", f"该渠道把请求回报成 {strangers};{_rt_summary(observations)}", observations, ) pytest.skip(f"{model} 被该渠道路由到 {strangers},本次观测说的不是这个模型") if rejected: verdict, problem = "FAIL(上游拒绝该档)", f"{tier.value}: 上游拒绝" elif not usable: verdict, problem = "SKIP(源不可用)", None elif len(observed) * 2 > len(usable): verdict, problem = "PASS", None else: verdict, problem = "FAIL(该档未推理)", f"{tier.value}: 多数轮未观测到推理" if problem: failures.append(problem) _probe_record( model, f"档位 {tier.value}", verdict, _rt_summary(observations), observations ) assert not failures, f"{model} 登记的档位与实测不符: {failures}" @pytest.mark.parametrize("model", ["gemini-3.1-pro", "gpt-5.5", "glm-5.3"]) async def test_t10_no_opinion_stays_no_opinion(self, model): """不表态时库**不推定**模型自己的默认档(Phase 1),顺带采下默认档的 rt 基线。 为什么给这三个模型单列一条: 它们的「厂商默认档」是 evidence 里写着、却最容易 写错的一格(Gemini 3.1 Pro 官方文档说 HIGH、OpenRouter 说 medium,两源打架), 而默认档写错会误导下游估成本。库本身不依赖这个值——**它不表态就什么都不注入**, 这正是本条断言的东西;默认档的 rt 观测只作报告里的旁证,**不作断言**: 单一模型上 rt 与档位没有可判定的函数关系(设计 §4.3),拿它反推默认档只能存疑,不能定论。 2026-09-05: gemini 一路当下在本渠道上游报错,claude 一路 7 天限额用尽,故把 另两格换成当下可测的 gpt-5.5 与 glm-5.3;gemini 留着,渠道恢复即有数。 """ client = GatewayClient.from_settings( _tier_settings(model), capabilities=_probe_capabilities(model) ) observations = [] try: for i in range(_TIER_ROUNDS): try: resp = await client.chat( [{"role": "user", "content": _TIER_PROMPT}], stream=True, cache_salt=f"tier-default-{model}-{i}", ) except ( RequestRejectedError, GatewayUnavailableError, SourceDeadError, TransientError, ) as exc: observations.append( { "round": i + 1, "effort": "(不表态)", "prompt_kind": "short", "error": f"{type(exc).__name__}: {str(exc)[:160]}", } ) continue observations.append( { "round": i + 1, "effort": "(不表态)", "prompt_kind": "short", "error": None, "prompt_tokens": resp.prompt_tokens, "completion_tokens": resp.completion_tokens, "reasoning_tokens": resp.reasoning_tokens, "thinking_chars": len(resp.thinking), "thinking_observation": resp.thinking_observation, "applied_effort": resp.applied_effort, "model_reported": resp.model_reported, "content": resp.content[:40], } ) finally: await client.aclose() usable = [o for o in observations if _probe_ok(o)] if not usable: _probe_record( model, "默认档基线(不表态)", "SKIP(源不可用)", _rt_summary(observations), observations, ) pytest.skip(f"{model} 源不可用,已记为未覆盖: {observations[0]['error'][:120]}") leaked = [o for o in usable if o["applied_effort"] is not None] _probe_record( model, "默认档基线(不表态)", "PASS" if not leaked else "FAIL(库替模型推定了默认档)", _rt_summary(observations), observations, ) assert not leaked, f"{model}: 不表态时 applied_effort 应为 None,实测 {leaked}"