diff --git a/src/polygateway/thinking.py b/src/polygateway/thinking.py index e72742b..36c17d0 100644 --- a/src/polygateway/thinking.py +++ b/src/polygateway/thinking.py @@ -179,3 +179,65 @@ def _warn_unregistered(model: str, profile: ProviderProfile, slot: Mapping[str, profile.name, dict(slot), ) + + +def reconcile_thinking( + *, + enable_thinking: bool | None, + observation: ThinkingObservation, + capability: ThinkingCapability | None, + model: str, +) -> str | None: + """把静态声明与运行时观测对账;矛盾返回告警文案,无矛盾返回 None。 + + 能力表过期是必然事件(M3 的 evidence 曾停在 8-02 整整 23 天),而过期的 + 表现是静默错觉。本函数把它变成可报警事件,代价是一次枚举比较。 + + **只判定、不打日志**: 文案作为返回值交给调用点,单测才能直接断言告警内容, + 而不必去解析日志格式;节流也才能留在握有实例状态的 transport 里。 + + **不抛错**: 一次观测不足以否决一次成功的调用;可观测性属遥测方向,降级即 + warning(P5 的"报错而非放行"只约束限流/熔断)。矛盾结果已随 `LLMResponse` + 与遥测落地,处置权归下游。 + """ + # Phase 1: 调用方不表态 —— 没提要求就无从谈"违背" + if enable_thinking is None: + return None + # Phase 2: 要求关闭 —— 只有 OBSERVED 能证伪。UNKNOWN 没有证伪力,拿它报警 + # 等于每次关闭调用都喊一遍(M3 关闭档恒落此档),噪声即等于没有告警 + if enable_thinking is False: + if observation is not ThinkingObservation.OBSERVED: + return None + return _off_but_observed(model, capability) + # Phase 3: 要求开启 —— ABSENT 是正面证伪,UNKNOWN 是"看不见",两者文案不可混 + if observation is ThinkingObservation.ABSENT: + return ( + f"模型 {model!r} 的 enable_thinking=True 未生效: 已注入开启参数," + f"上游却明确上报本次未推理(reasoning_tokens=0)" + ) + if observation is ThinkingObservation.UNKNOWN: + return ( + f"模型 {model!r} 的 enable_thinking=True 无法确认是否生效: 已注入开启参数," + f"但本次响应观测不到任何推理信号(推理正文与 usage 明细双缺)。" + f"若走的是非流式路径,推理内容可能已计费却不回传" + ) + return None + + +def _off_but_observed(model: str, capability: ThinkingCapability | None) -> str: + """关闭请求未被满足的两种说法;登记与否决定该说哪一句。 + + 两者必须分开: `resolve_thinking` 对未登记模型的告警是**事前猜测**,这里是 + **事后实证**。对未登记模型说"能力表声称可关闭"是错的——它根本没登记。 + """ + if capability is None: + return ( + f"模型 {model!r} 的 enable_thinking=False 未被满足: 实测观测到推理发生," + f"且该模型的推理能力尚未登记(本次按 provider 形态尽力注入)。" + f"请实测后用 register_capability 登记其真实能力" + ) + return ( + f"模型 {model!r} 的 enable_thinking=False 未被满足: 实测观测到推理发生," + f"而能力表登记 can_disable={capability.can_disable}(evidence: {capability.evidence})。" + f"能力表可能已过期——请复测后用 register_capability 更新登记" + ) diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index aab2e46..bc34bb6 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -14,6 +14,7 @@ import time from typing import TYPE_CHECKING, Any import httpx +from loguru import logger from polygateway.errors import ( PolyGatewayError, @@ -29,6 +30,7 @@ from polygateway.thinking import ( ThinkingUnsupportedError, get_capability, observe_thinking, + reconcile_thinking, resolve_thinking, ) from polygateway.transports._http_errors import compose_message, summarize_body @@ -320,6 +322,9 @@ class OpenAICompatTransport: # 未登记模型只喊一次: 装配期已喊过,逐次调用再喊是日志洪水。 # 实例级而非模块级 —— 模块级可变状态违反纯 asyncio 中立铁律 self._warned_models: set[str] = set() + # 对账告警独立节流,**不复用** `_warned_models`: 那个 set 的语义是"未登记 + # 能力已告警过",两件事共用一个开关会互相压制——一方喊过就把另一方静音 + self._warned_mismatches: set[tuple[str, bool | None]] = set() self._client_factory = client_factory or _default_client_factory self._clients: dict[str, httpx.AsyncClient] = {} @@ -390,8 +395,9 @@ class OpenAICompatTransport: ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"} try: if stream: - return await self._complete_stream(client, url, payload, source, profile) - return await self._complete_once(client, url, payload, source, profile) + result = await self._complete_stream(client, url, payload, source, profile) + else: + result = await self._complete_once(client, url, payload, source, profile) except StreamLivenessTimeout as exc: raise TransientError(f"{source.name} 流活性超时({exc.kind})", **ctx) from exc except httpx.TimeoutException as exc: @@ -399,6 +405,29 @@ class OpenAICompatTransport: except httpx.TransportError as exc: # VT 宽集: 覆盖断连/协议错误/读写失败(设计 §9 行 8) raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc + # 此处是唯一同时握有请求方向与响应结果的地方,对账只能落在这里 + self._warn_on_thinking_mismatch(source, result) + return result + + def _warn_on_thinking_mismatch(self, source: SourceConfig, result: TransportResult) -> None: + """声明与观测矛盾即 warning;按 (model, direction) 节流,同组合只喊一次。 + + 节流键必须含方向: 同一模型的开、关两档是两个独立的矛盾,合并键会让先出现 + 的那一档把另一档永久静音。逐次调用刷屏会把告警变成噪声,噪声等于没有告警。 + """ + message = reconcile_thinking( + enable_thinking=source.enable_thinking, + observation=result.thinking_observation, + capability=get_capability(source.model, table=self._capabilities), + model=source.model, + ) + if message is None: + return + key = (source.model, source.enable_thinking) + if key in self._warned_mismatches: + return + self._warned_mismatches.add(key) + logger.warning(message) async def embed( self, *, texts: list[str], source: SourceConfig, call_id: str diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index cd7838d..e2f365b 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -546,6 +546,56 @@ class TestThinkingObservationVerdict: assert result.thinking_observation is ThinkingObservation.ABSENT +class TestThinkingReconciliation: + """对账告警按 (model, direction) 节流(设计 §5)。 + + 节流键必须含方向: 同一模型的开、关两档是两个独立的矛盾,合并键会让先出现 + 的那一档把另一档永久静音。 + """ + + def _handler(self, request): + payload = json.loads(request.content) + if payload.get("reasoning_effort") == "none": + # 关闭档却回了推理正文 → OBSERVED,与"要求关闭"矛盾 + return _sse_stream( + _chunk(reasoning="偷偷想了"), _chunk(content="ok"), _chunk(usage=_USAGE) + ) + # 开启档却零信号 → UNKNOWN,无法确认是否生效(M3 实测形态) + return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE)) + + def _minimax(self, enable_thinking): + return _source( + name="mm", provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking + ) + + async def test_same_model_and_direction_warns_only_once(self): + transport = _transport_for(self._handler) + source = self._minimax(False) + messages: list[str] = [] + sink_id = logger.add(messages.append, level="WARNING") + try: + await _complete(transport, source) + await _complete(transport, source) + finally: + logger.remove(sink_id) + hits = [m for m in messages if "MiniMax-M3" in m] + assert len(hits) == 1, f"同一 (model, direction) 应只告警一次,实得 {len(hits)} 次" + + async def test_switching_direction_earns_a_second_warning(self): + transport = _transport_for(self._handler) + messages: list[str] = [] + sink_id = logger.add(messages.append, level="WARNING") + try: + await _complete(transport, self._minimax(False)) + await _complete(transport, self._minimax(False)) + await _complete(transport, self._minimax(True)) + await _complete(transport, self._minimax(True)) + finally: + logger.remove(sink_id) + hits = [m for m in messages if "MiniMax-M3" in m] + assert len(hits) == 2, f"两个方向各应告警一次,实得 {len(hits)} 次" + + class TestNonStreamFastPath: async def test_non_stream_parses_message(self): def handler(request): diff --git a/tests/unit/test_thinking.py b/tests/unit/test_thinking.py index 08dada7..75cacb6 100644 --- a/tests/unit/test_thinking.py +++ b/tests/unit/test_thinking.py @@ -14,6 +14,7 @@ from polygateway.thinking import ( ThinkingCapability, get_capability, observe_thinking, + reconcile_thinking, register_capability, resolve_thinking, ) @@ -160,3 +161,119 @@ class TestResolveThinking: cap = ThinkingCapability(can_disable=False, evidence="构造") with pytest.raises(ValueError, match="register_provider"): resolve_thinking(get_provider("openai"), cap, False, model="whatever") + + +class TestReconcileThinking: + """声明 × 观测对账(设计 §5): 矛盾出文案,不表态出 None。 + + 文案本身是被断言对象——判定与日志分离正是为此: 告警内容可直接比对,不必 + 去解析日志格式。 + """ + + _CAP = ThinkingCapability( + can_disable=True, evidence="2026-08-02 实测 reasoning_effort=none 可关闭" + ) + + def test_off_but_observed_with_a_registered_capability_blames_the_table(self): + """已登记却实测推理了 = 能力表漂移: 必须附 evidence 与更新指路。""" + msg = reconcile_thinking( + enable_thinking=False, + observation=ThinkingObservation.OBSERVED, + capability=self._CAP, + model="MiniMax-M3", + ) + assert msg is not None + assert "MiniMax-M3" in msg + assert "2026-08-02 实测 reasoning_effort=none 可关闭" in msg + assert "register_capability" in msg + + def test_off_but_observed_unregistered_never_claims_a_table_entry(self): + """未登记模型没有"能力表声称"这回事——说它就是撒谎。""" + msg = reconcile_thinking( + enable_thinking=False, + observation=ThinkingObservation.OBSERVED, + capability=None, + model="MiniMax-M9", + ) + assert msg is not None + assert "MiniMax-M9" in msg + assert "能力表" not in msg + assert "register_capability" in msg + + def test_registered_and_unregistered_wordings_differ(self): + registered = reconcile_thinking( + enable_thinking=False, + observation=ThinkingObservation.OBSERVED, + capability=self._CAP, + model="MiniMax-M3", + ) + unregistered = reconcile_thinking( + enable_thinking=False, + observation=ThinkingObservation.OBSERVED, + capability=None, + model="MiniMax-M3", + ) + assert registered != unregistered + + @pytest.mark.parametrize("capability", [None, _CAP]) + def test_on_but_absent_is_a_contradiction(self, capability): + """上游明确上报未推理: 这是唯一的正面证伪,与能力表登记与否无关。""" + msg = reconcile_thinking( + enable_thinking=True, + observation=ThinkingObservation.ABSENT, + capability=capability, + model="qwen3.7-plus", + ) + assert msg is not None + assert "qwen3.7-plus" in msg + + @pytest.mark.parametrize("capability", [None, _CAP]) + def test_on_but_unknown_admits_it_cannot_confirm(self, capability): + """issue #17 的诚实版本: 明说"我注入了,但我看不见结果"。""" + msg = reconcile_thinking( + enable_thinking=True, + observation=ThinkingObservation.UNKNOWN, + capability=capability, + model="MiniMax-M3", + ) + assert msg is not None + assert "MiniMax-M3" in msg + + def test_off_and_unknown_stays_silent(self): + """UNKNOWN 没有证伪力: 拿它报警等于每次关闭调用都喊(M3 关闭档恒落此档)。""" + assert ( + reconcile_thinking( + enable_thinking=False, + observation=ThinkingObservation.UNKNOWN, + capability=self._CAP, + model="MiniMax-M3", + ) + is None + ) + + @pytest.mark.parametrize( + "observation", + [ThinkingObservation.OBSERVED, ThinkingObservation.ABSENT, ThinkingObservation.UNKNOWN], + ) + def test_no_request_no_grievance(self, observation): + """调用方不表态,就无从谈"违背"。""" + assert ( + reconcile_thinking( + enable_thinking=None, + observation=observation, + capability=self._CAP, + model="MiniMax-M3", + ) + is None + ) + + def test_on_and_observed_is_exactly_what_was_asked_for(self): + assert ( + reconcile_thinking( + enable_thinking=True, + observation=ThinkingObservation.OBSERVED, + capability=self._CAP, + model="MiniMax-M3", + ) + is None + )