"""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 @dataclass(frozen=True) class ProviderProfile: """单个 provider 的能力与差异声明。 thinking_on/thinking_off 分别是 `SourceConfig.enable_thinking` 为 True/False 时并入请求体的参数片段(None 时二者都不注入,用模型默认); strip_think_tags 声明响应 content 需剥离 ```` 标签(qwen 系); supports_native_schema 供 D14 阶梯选择原生 response_format 策略。 注: 某个 provider 的两档若皆为空字典(如 openai/minimax),说明该 provider 无已知的推理开关参数——此时 `enable_thinking` 对它**不产生任何效果**, 而非静默生效。需要下发自定义参数时用 `SourceConfig.extra_body`。 """ name: str thinking_on: dict[str, Any] thinking_off: dict[str, Any] 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, ), # 两档皆空 ⇒ `enable_thinking` 对本 provider **不产生任何效果**(调用方 # 以为关掉了实际没关)。真需要控制推理时经 `SourceConfig.extra_body` 下发 "openai": ProviderProfile( name="openai", thinking_on={}, thinking_off={}, strip_think_tags=False, ), # OpenAI 兼容基线,无已知注入差异;reasoning_content 由 transport 通用处理。 # 同上: 两档皆空 ⇒ `enable_thinking` 对 MiniMax 源不产生任何效果 "minimax": ProviderProfile( name="minimax", thinking_on={}, thinking_off={}, strip_think_tags=False, ), } ) 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