nanobot 源码解读 · 07 | Provider 与模型路由：一个模型名如何落到正确的后端

@dataclass(frozen=True)
class ProviderSpec:
    # 身份
    name: str                 # 配置字段名，如 "dashscope"
    keywords: tuple[str, ...] # 模型名关键字，用于匹配
    env_key: str              # API key 的环境变量名
    display_name: str = ""

    # 用哪个实现
    backend: str = "openai_compat"   # openai_compat | anthropic | azure_openai | ...

    # 网关 / 本地检测
    is_gateway: bool = False          # 能路由任意模型（OpenRouter 等）
    is_local: bool = False            # 本地部署（Ollama）
    detect_by_key_prefix: str = ""    # 按 api_key 前缀识别，如 "sk-or-"
    detect_by_base_keyword: str = ""  # 按 api_base URL 子串识别
    default_api_base: str = ""        # 该厂商默认的 OpenAI 兼容 base URL

    # 网关行为
    strip_model_prefix: bool = False  # 发送前剥掉 "provider/" 前缀
    supports_prompt_caching: bool = False
    # 思考/reasoning 相关的一堆开关...
    thinking_style: str = ""
    reasoning_effort_remap: tuple[tuple[str, str], ...] = ()
    implicit_reasoning_models: tuple[str, ...] = ()
    strip_history_reasoning_content: bool = False

def _resolve_provider_setup(config, *, preset, model=None):
    model = model or preset.model
    provider_name = config.get_provider_name(model, preset=preset)  # 配置里查 maps
    p = config.get_provider(model, preset=preset)
    spec = find_by_name(provider_name)                              # 在 PROVIDERS 表查 spec
    if not spec and p:
        if not p.api_base:
            raise ValueError(...)                                   # 没 spec 必须有 api_base
        spec = create_dynamic_spec(provider_name, ...)              # 动态造一个
    backend = spec.backend if spec else "openai_compat"
    # 一堆校验：Azure 必须 api_base、OAuth/本地/直连可免 key 等
    return _ProviderSetup(model=model, provider_name=provider_name,
                          provider_config=p, spec=spec, backend=backend)

if backend == "openai_codex":   # OAuth 流程
    ...
elif backend == "xai_grok":     # OAuth 流程
    ...
elif backend == "azure_openai":
    ...
elif backend == "anthropic":
    provider = AnthropicProvider(...)
elif backend == "bedrock":
    ...
else:  # openai_compat（绝大多数厂商）
    provider = OpenAICompatProvider(...)

def __init__(self, api_key, api_base, default_model="gpt-4o", spec=None, ...):
    ...
    # 懒加载：OpenAI client + httpx transport 创建很贵（Windows 上约 700ms）
    self._client = None
    self._client_lock = asyncio.Lock()

def _build_client(self):
    import httpx
    # 本地模型（Ollama/vLLM）容易在两次调用间让 keepalive 连接失效，
    # 于是关掉 keepalive、不走代理，每次请求开新连接（LAN 上很便宜）
    if self._is_local:
        _local_limits = httpx.Limits(keepalive_expiry=0)
        http_client = httpx.AsyncClient(limits=_local_limits, transport=httpx.AsyncHTTPTransport(proxy=None, ...))
    self._client = AsyncOpenAI(api_key=..., base_url=self._effective_base,
                               max_retries=0, timeout=timeout_s, http_client=http_client)

def _build_kwargs(self, messages, tools, model, max_tokens, temperature, reasoning_effort, tool_choice):
    model_name = model or self.default_model
    spec = self._spec

    # 1) prompt caching：Anthropic/Claude 打 cache_control
    if spec and spec.supports_prompt_caching:
        if model_name.lower().startswith(("anthropic/", "claude")):
            messages, tools = self._apply_cache_control(messages, tools)

    model_name = self._request_model_name(model_name)   # 网关剥前缀

    kwargs = {"model": model_name,
              "messages": self._sanitize_messages(self._sanitize_empty_content(messages))}

    # 2) GPT-5 / o 系列 reasoning 激活时拒绝 temperature，按需省略
    if self._supports_temperature(model_name, reasoning_effort):
        kwargs["temperature"] = temperature

    # 3) 有的厂商要 max_completion_tokens，有的要 max_tokens
    if spec and spec.supports_max_completion_tokens or _requires_max_completion_tokens(model_name):
        kwargs["max_completion_tokens"] = max(1, max_tokens)
    else:
        kwargs["max_tokens"] = max(1, max_tokens)

    # 4) 每模型参数覆盖（如 kimi-k2.7 强制 temperature=1.0）
    if spec:
        for pattern, overrides in spec.model_overrides:
            if pattern in model_name.lower():
                kwargs.update(overrides); break

    # 5) reasoning_effort 方言翻译（minimal→minimum、Mistral→high/none、
    #    Magistral 隐式思考直接 strip 该参数、Kimi 去掉冗余 reasoning_effort...）
    ...
    if wire_effort and semantic_effort != "none":
        kwargs["reasoning_effort"] = wire_effort

    # 6) 思考开关注入 extra_body（火山/通义/阶跃各自的 thinking 形状）
    for thinking_style in _thinking_styles_for(spec, model_name):
        extra = _thinking_extra_body(thinking_style, thinking_enabled)
        if extra: kwargs.setdefault("extra_body", {}).update(extra)

    # 7) 工具
    if tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = tool_choice or "auto"

    # 8) DeepSeek 思考模式历史必须补 reasoning_content=""，否则 400
    if explicit_thinking or implicit_deepseek_thinking:
        for msg in kwargs["messages"]:
            if msg.get("role") == "assistant" and "reasoning_content" not in msg:
                msg["reasoning_content"] = ""
    return kwargs

async def _try_with_fallback(self, call, kwargs, has_streamed, on_stream_recover=None):
    primary_model = kwargs.get("model") or self._primary.get_default_model()
    if self._primary_available():
        response = await call(self._primary, kwargs)
        if response.finish_reason != "error":
            self._primary_failures = 0
            return response                      # 主模型成功，直接返回
        # 主模型报错了，判断是否「可降级」
        if not self._should_fallback(response):
            return response                      # 不可降级（如鉴权失败）→ 直接返回错误
        self._primary_failures += 1
        if self._primary_failures >= _PRIMARY_FAILURE_THRESHOLD:
            self._primary_tripped_at = time.monotonic()   # 连续失败太多 → 熔断器打开
    # 依次试每个 fallback 预设
    for fallback in self._fallback_presets:
        fb_provider = self._provider_factory(fallback)
        await self._notify_fallback_model(fallback.model)
        fb_kwargs = {**kwargs, "model": fallback.model,
                     "max_tokens": fallback.max_tokens, "temperature": fallback.temperature}
        # 切换模型时还要处理对话状态能否跨 Provider 续接、是否需要原生压缩
        ...
        fb_response = await call(fb_provider, fb_kwargs)
        if fb_response.finish_reason != "error":
            return fb_response                  # 某个降级成功即返回
    # 全失败，返回最后一个错误

async def _run_with_retry(self, call, kw, original_messages, *, retry_mode, on_retry_wait, ...):
    delays = list(self._CHAT_RETRY_DELAYS)
    persistent = retry_mode == "persistent"
    while True:
        attempt += 1
        response = await call(**kw)
        if response.finish_reason != "error":
            return response
        # 流式已吐字的情况：超时则续接重试，否则直接返回
        ...
        if not self.is_transient_response(response):
            # 非瞬态错误：若消息含图片，尝试「去掉图片重试一次」；否则直接返回
            ...
            return response
        # 瞬态错误：按 Retry-After 头或退避表退避后重试
        retry_after = self._extract_retry_after_from_response(response)
        delay = retry_after + BUFFER if retry_after else delays[min(attempt-1, ...)]
        await self._sleep_with_heartbeat(delay, ...)
    return last_response