PolyGateway/src/polygateway/providers.py

"""provider 注册表(D11): 消灭 `"qwen" in provider` 式字符串猜测。

每个 provider 显式声明 thinking 参数注入形态与响应处理差异;查找按名字
**精确匹配**,未注册即装配期报错。注册是纯函数——返回新表,不修改共享
状态(纯 asyncio 中立铁律);client 经 `registry` 参数持有自己的表。
"""

from collections.abc import Mapping
from dataclasses import dataclass
from types import MappingProxyType
from typing import Any

from loguru import logger


@dataclass(frozen=True)
class ProviderProfile:
    """单个 provider 的能力与差异声明。

    thinking_on/thinking_off 分别是 `SourceConfig.enable_thinking` 为
    True/False 时并入请求体的参数片段(`enable_thinking` 为 None 时二者都不
    注入,用模型默认);strip_think_tags 声明响应 content 需剥离 ``<think>``
    标签(qwen 系);supports_native_schema 供 D14 阶梯选择原生 response_format。

    两档各有三种取值,**语义互不重叠**(issue #5):

    ==========  ==========================================================
    ``{...}``   已知的注入片段
    ``{}``      已知**无需注入**任何参数即处于该档
    ``None``    **未知**: 本库不知道该 provider 如何表达这一档
    ==========  ==========================================================

    `None` 与 `{}` 必须分开: 二者曾同为空字典,导致 `enable_thinking=False`
    对 minimax/openai 源静默失效——调用方以为关掉了推理,实际什么都没发生。
    现在 `None` 会在装配期显式报错并指路 `register_provider` / `extra_body`。

    注: 本类只声明**形态**(参数长什么样,按 provider 变);某个具体模型能否
    关闭推理属**能力**(按 model 变),见 `ThinkingCapability`。
    """

    name: str
    thinking_on: Mapping[str, Any] | None
    thinking_off: Mapping[str, Any] | None
    strip_think_tags: bool
    supports_native_schema: bool = False


DEFAULT_PROFILES: Mapping[str, ProviderProfile] = MappingProxyType(
    {
        # 注入片段出处: VT llm.py:130-144(开启形态)与 CHS invokers.py:230-238(关闭形态)
        "qwen": ProviderProfile(
            name="qwen",
            thinking_on={"enable_thinking": True},
            thinking_off={"enable_thinking": False},
            strip_think_tags=True,
        ),
        "deepseek": ProviderProfile(
            name="deepseek",
            thinking_on={"thinking": {"type": "enabled"}},
            thinking_off={"thinking": {"type": "disabled"}},
            strip_think_tags=False,
        ),
        # OpenAI 兼容基线段名: 实践中被复用为**任意**兼容厂商的兜底(下游把
        # kimi-k3 挂在 provider=openai 下),故不能下发任何厂商方言参数——发给
        # 不认识它的厂商会 400。两档标 None(未知): 配了 enable_thinking 即在
        # 装配期报错并指路,真 OpenAI 推理模型的用户走 register_provider
        "openai": ProviderProfile(
            name="openai",
            thinking_on=None,
            thinking_off=None,
            strip_think_tags=False,
        ),
        # 注入形态出处: 2026-08-02 经自建 new-api 中转实测(findings §2),
        # **直连官方端点未验证**。实测 enable_thinking / thinking 两种写法均被
        # 静默丢弃(prompt_tokens 恒定不变),reasoning_effort 才是真开关。
        # "开"取 medium: qwen 的 enable_thinking:true 与 deepseek 的
        # thinking:{enabled} 都不指定预算、由模型自定,medium 是五档里语义最接近
        # "厂商正常强度"的一档;取 high 等于替下游做"加钱换质量"的业务判断。
        # 要精确控制档位经 `SourceConfig.extra_body`(优先级高于本片段)
        "minimax": ProviderProfile(
            name="minimax",
            thinking_on={"reasoning_effort": "medium"},
            thinking_off={"reasoning_effort": "none"},
            strip_think_tags=False,
        ),
    }
)


class ThinkingUnsupportedError(ValueError):
    """推理开关无法满足: 形态未知或该模型不支持该方向(issue #5)。

    是 `ValueError` 的子类而非 `errors.py` 四分类之一——它描述的是**配置**
    不可满足(装配期就该炸),不是一次调用的运行时失败。transport 在请求期
    捕获它并翻译为 `RequestRejectedError` 再进四分类。单列一个类型是为了让
    捕获点能精确到它,而不是宽catch 整个 `ValueError`(那会把序列化等无关
    错误误贴成"推理开关无法满足")。
    """


@dataclass(frozen=True)
class ThinkingCapability:
    """某个**具体模型**能否关闭推理(issue #5);登记必须附实测证据与日期。

    与 `ProviderProfile` 的分工: 后者声明**形态**(参数长什么样,按 provider 变,
    数年不变一次),本类声明**能力**(按 model 变,同一 provider 每代都变)。二者
    合一在 provider 级表达不了代际差异——实测 MiniMax-M3 可关闭推理,而同厂的
    M2.7/M2.5 三种参数形态全部无效(findings §2.3),profile 一格管不住三个模型。

    `evidence` 不是装饰: 能力表过期是必然事件,没有出处就无从判断该不该信它。
    """

    can_disable: bool
    evidence: str


DEFAULT_CAPABILITIES: Mapping[str, ThinkingCapability] = MappingProxyType(
    {
        "MiniMax-M3": ThinkingCapability(
            can_disable=True,
            evidence="2026-08-02 经 new-api 中转实测 N=10: reasoning_effort=none 稳定关闭,零跳变",
        ),
        "MiniMax-M2.7": ThinkingCapability(
            can_disable=False,
            evidence=(
                "2026-08-02 实测 reasoning_effort=none / thinking:{disabled} / thinking:{adaptive} "
                "各 N=3 全部无效;OpenRouter 注册表登记 mandatory:true,models.dev 登记无控制手段"
            ),
        ),
        "MiniMax-M2.5": ThinkingCapability(
            can_disable=False,
            evidence="2026-08-02 实测同 M2.7: 三种形态各 N=3 全部无效;外部注册表同样登记为强制推理",
        ),
        "qwen3.7-plus": ThinkingCapability(
            can_disable=True,
            evidence="2026-08-02 实测 enable_thinking=false 关闭(completion 5 token,无推理)",
        ),
        "deepseek-v4-pro": ThinkingCapability(
            can_disable=True,
            evidence="2026-08-02 实测 thinking:{type:disabled} 关闭(completion 3 token,无推理)",
        ),
    }
)
"""在用模型的推理能力登记(YAGNI: 不覆盖全世界,未登记走 `resolve_thinking` 退化)。"""


def get_capability(
    model: str, *, table: Mapping[str, ThinkingCapability] | None = None
) -> ThinkingCapability | None:
    """按模型名精确查找;未登记返回 None(= 能力未知,由调用方决定如何退化)。

    与 `get_provider` 未注册即报错不同: provider 是配置里写死的少数几个值,
    写错就是配置错误;而模型名千变万化,新模型上线不该被库挡住(设计 §5 R4)。
    """
    return (DEFAULT_CAPABILITIES if table is None else table).get(model)


def register_capability(
    model: str,
    capability: ThinkingCapability,
    *,
    base: Mapping[str, ThinkingCapability] | None = None,
) -> dict[str, ThinkingCapability]:
    """纯函数注册: 返回 base(缺省 DEFAULT_CAPABILITIES)+ 新条目的新表,同名覆盖。"""
    table = dict(DEFAULT_CAPABILITIES if base is None else base)
    table[model] = capability
    return table


def resolve_thinking(
    profile: ProviderProfile,
    capability: ThinkingCapability | None,
    enable_thinking: bool | None,
    *,
    model: str,
    warn_unregistered: bool = True,
) -> Mapping[str, Any]:
    """三态 + 两层能力 → 请求体注入片段;不可满足时 ValueError。

    调用点负责翻译: 装配期直接冒泡(配置错误),transport 内翻译为
    `RequestRejectedError`(四分类之一)。判定顺序即语义,不可调换——形态未知时
    无从注入,能力如何无关紧要,故 Phase 2 必须先于 Phase 4;未登记模型没有
    `can_disable` 可读,故 Phase 3 必须先于 Phase 4。

    `model` 只用于错误与告警文案: 报错能定位到具体模型才有可操作性,而
    `capability` 为 None(未登记)时无从从别处取得模型名。

    `warn_unregistered=False` 供请求热路径去重用: 装配期已经喊过一次,逐次
    调用再喊只会刷屏。判定结果不受此参数影响。
    """
    # Phase 1: 调用方不表态 —— 与 False 严格区分,用模型默认档
    if enable_thinking is None:
        return {}
    slot = profile.thinking_on if enable_thinking else profile.thinking_off
    direction = "thinking_on" if enable_thinking else "thinking_off"
    # Phase 2: 形态未知 —— 提供了开关却不知道怎么发,静默放行就是欺骗调用方
    if slot is None:
        raise ThinkingUnsupportedError(
            f"provider {profile.name!r} 的 {direction} 形态未知(模型 {model!r}): "
            f"本库不知道该 provider 如何表达这一档。请用 register_provider 注册形态,"
            f"或改用 SourceConfig.extra_body 直接下发供应商参数"
        )
    # Phase 3: 能力未登记 —— 新模型上线不该被库挡住,但也不该假装成功
    if capability is None:
        if warn_unregistered:
            _warn_unregistered(model, profile, slot)
        return slot
    # Phase 4: 明确不支持关闭 —— 调用方要的是"不推理"的语义保证,给不了必须说
    if enable_thinking is False and not capability.can_disable:
        raise ThinkingUnsupportedError(
            f"模型 {model!r} 无法关闭推理,enable_thinking=False 无法满足: "
            f"{capability.evidence}。该模型的推理是固有属性,任何参数都关不掉——"
            f"需要关闭思维链请换用支持关闭的模型"
        )
    return slot


def _warn_unregistered(model: str, profile: ProviderProfile, slot: Mapping[str, Any]) -> None:
    logger.warning(
        "模型 {} 的推理能力未登记,按 provider {} 的形态尽力注入 {};"
        "若该模型实际不支持这一档,本次设置将静默失效。实测后请用 register_capability 登记",
        model,
        profile.name,
        dict(slot),
    )


def get_provider(
    name: str, *, registry: Mapping[str, ProviderProfile] | None = None
) -> ProviderProfile:
    """按名字精确查找 profile;未注册直接报错(严禁默认值掩盖配置错误)。"""
    table = DEFAULT_PROFILES if registry is None else registry
    profile = table.get(name)
    if profile is None:
        raise ValueError(
            f"未注册的 provider: {name!r}(已注册: {sorted(table)});"
            f"新 provider 用 register_provider(ProviderProfile(...)) 注册后经 registry 参数传入"
        )
    return profile


def register_provider(
    profile: ProviderProfile, *, base: Mapping[str, ProviderProfile] | None = None
) -> dict[str, ProviderProfile]:
    """纯函数注册: 返回 base(缺省 DEFAULT_PROFILES)+ 新条目的新表,同名覆盖。"""
    table = dict(DEFAULT_PROFILES if base is None else base)
    table[profile.name] = profile
    return table