[06] Hermes Agent 模型调度源码拆解：40+ Provider 注册表、5 种 API 模式与动态运行时解析

人工智能Agent 2026-06-21 16
预计阅读时间：10 分钟

> TL;DR：从 /model claude-sonnet-4 敲下回车到请求发出去，中间经过 runtime_provider.py（1694 行）和 auth.py（7706 行）的精密调度——Provider 注册表解析 → 凭据池检索 → API 模式自动检测 → 请求参数拼装。这篇不是讲怎么配 model，是拆这套调度引擎的每一层。

上一篇拆了记忆系统，这篇拆谁都离不开的——模型调度。 你可能觉得这个没什么好拆的：配置里写个 model name + provider，Hermes 就去调 API 了。其实中间的链路比你想象的长得多——runtime_provider.py（1694 行）、auth.py（7706 行），两文件加起来将近一万行代码，就为了把一行 model: deepseek/deepseek-v4-pro 变成一条可执行的 HTTP 请求。

1. 注册表：40 个内置 Provider 的索引

auth.py 中定义了 PROVIDER_REGISTRY——一个包含约 40 个内置 Provider 的字典。每个 entry 是一个 ProviderConfig：

@dataclass
class ProviderConfig:
    id: str
    name: str
    auth_type: str          # "api_key" / "oauth_device_code" / "oauth_external" / "oauth_minimax" / "aws_sdk" / "external_process"
    portal_base_url: str = ""
    inference_base_url: str = ""
    client_id: str = ""
    scope: str = ""
    api_key_env_vars: tuple = ()   # 环境变量优先级列表
    base_url_env_var: str = ""

注册表里的典型 entry：

PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
    "anthropic": ProviderConfig(
        id="anthropic",
        name="Anthropic",
        auth_type="api_key",
        inference_base_url="https://api.anthropic.com",
        api_key_env_vars=("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"),
        base_url_env_var="ANTHROPIC_BASE_URL",
    ),
    "deepseek": ProviderConfig(
        id="deepseek",
        name="DeepSeek",
        auth_type="api_key",
        inference_base_url="https://api.deepseek.com/v1",
        api_key_env_vars=("DEEPSEEK_API_KEY",),
        base_url_env_var="DEEPSEEK_BASE_URL",
    ),
    "openai-codex": ProviderConfig(
        id="openai-codex",
        name="OpenAI Codex",
        auth_type="oauth_external",
        inference_base_url="https://chatgpt.com/backend-api/codex",
    ),
    # ... 约 40 个类似 entry
}

每个 Provider 的 api_key_env_vars 是优先级列表。比如 Anthropic 会依次检查 ANTHROPIC_API_KEY → ANTHROPIC_TOKEN → CLAUDE_CODE_OAUTH_TOKEN。这个设计让用户不需要把所有环境变量都配齐——有一个就行。 注册表不是静态写死的。文件尾部有一段自动扩展逻辑：

try:
    from providers import list_providers as _list_providers_for_registry
    for _pp in _list_providers_for_registry():
        if _pp.name in PROVIDER_REGISTRY:
            continue
        PROVIDER_REGISTRY[_pp.name] = ProviderConfig(
            id=_pp.name,
            name=_pp.display_name,
            auth_type="api_key",
            inference_base_url=_pp.base_url,
            api_key_env_vars=_api_key_vars,
            base_url_env_var=_base_url_var or "",
        )
        for _alias in _pp.aliases:
            if _alias not in PROVIDER_REGISTRY:
                PROVIDER_REGISTRY[_alias] = PROVIDER_REGISTRY[_pp.name]
except Exception:
    pass

插件安装的新 Provider 会自动加入注册表，不需要动 auth.py 的源码。

2. Provider 解析链：从配置名到 Provider 实例

当用户在 config.yaml 中配了 provider: anthropic，解析过程从头走到尾有三步。

2.1 resolve_requested_provider()

runtime_provider.py 的入口：

def resolve_requested_provider(requested=None) -> str:
    if requested and requested.strip():
        return requested.strip().lower()
    model_cfg = _get_model_config()
    cfg_provider = model_cfg.get("provider")
    if isinstance(cfg_provider, str) and cfg_provider.strip():
        return cfg_provider.strip().lower()
    env_provider = os.getenv("HERMES_INFERENCE_PROVIDER", "").strip().lower()
    if env_provider:
        return env_provider
    return "auto"

优先级链：显式参数 > config.yaml > 环境变量 > "auto"。

2.2 _resolve_runtime_from_pool_entry()

拿到 provider 名后，进入 _resolve_runtime_from_pool_entry()——这是最长的函数，根据不同的 Provider 参数注入。关键逻辑：

def _resolve_runtime_from_pool_entry(*, provider, entry, ...):
    base_url = getattr(entry, "runtime_base_url", None) or getattr(entry, "base_url", "") or ""
    api_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
    api_mode = "chat_completions"    # 默认模式
    if provider == "openai-codex":
        api_mode = "codex_responses"
        base_url = base_url or DEFAULT_CODEX_BASE_URL
    elif provider == "anthropic":
        api_mode = "anthropic_messages"
        base_url = base_url or "https://api.anthropic.com"
    elif provider == "minimax-oauth":
        api_mode = "anthropic_messages"  # MiniMax OAuth 只谈 Anthropic Messages 协议
    elif provider == "xai":
        api_mode = "codex_responses"
    # ... 约 20 个 elif 分支

每个内置 Provider 都有自己的 api_mode 和默认 base_url。即便用户配错了 api_mode，opencode-go 这类 Provider 也会在运行时强制纠正：

if provider in {"opencode-zen", "opencode-go"}:
    from hermes_cli.models import opencode_model_api_mode
    api_mode = opencode_model_api_mode(provider, effective_model)

OpenCode Go 同时服务多个模型族——GLM/Kimi 走 chat_completions，MiniMax/Qwen 走 anthropic_messages。如果用户从 MiniMax 切到 Kimi，残留的 anthropic_messages 模式会导致 404。opencode_model_api_mode() 根据当前模型名动态修正。

2.3 API 模式自动检测

对于没有显式指定 api_mode 的 Provider，runtime 会尝试从 base_url 推断：

def _detect_api_mode_for_url(base_url):
    normalized = (base_url or "").strip().lower().rstrip("/")
    hostname = base_url_hostname(base_url)
    if hostname == "api.x.ai":
        return "codex_responses"
    if hostname == "api.openai.com":
        return "codex_responses"
    if normalized.endswith("/anthropic"):
        return "anthropic_messages"
    if hostname == "api.kimi.com" and "/coding" in normalized:
        return "anthropic_messages"
    return None

3. 5 种 API 模式

注册和解析完后，runtime 输出一个 api_mode 字段——它决定 Agent Loop 用哪种协议与模型通信： | api_mode | 协议 | 适用 Provider | |----------|------|--------------| | chat_completions | OpenAI /v1/chat/completions | DeepSeek、走 OpenAI 协议的 Anthropic 网关、本地模型 | | codex_responses | Codex Responses API | OpenAI Codex、xAI Grok、api.openai.com | | anthropic_messages | Anthropic /v1/messages | Anthropic 原生、MiniMax OAuth、Kimi Coding | | bedrock_converse | AWS Bedrock Converse | AWS Bedrock | | codex_app_server | 子进程全权 | 将整个 turn 交给 Codex 子进程（实验性） | 不同的 api_mode 在 Agent Loop 中走的 HTTP 调用路径完全不同——anthropic_adapter.py 负责把 System Prompt 格式从 OpenAI 转 Anthropic，model_metadata.py 负责估算 token 长度。每条路径都有自己的工具格式转换。

4. 凭据池：多 Key 轮换

runtime_provider.py 通过 CredentialPool 管理多组 API Key：

from agent.credential_pool import CredentialPool, PooledCredential
pool_key = get_custom_provider_pool_key(base_url, provider_name=provider_name)
pool = load_pool(pool_key)
entry = pool.select()  # 轮选一个有效凭证

凭据池的典型场景是：你配置了 3 个 OpenAI API Key，CredentialPool 会在每次请求时自动轮换。如果某个 key 返回 429（rate limit），pool 会标记它不可用并切换到下一个。 对于不是自定义 Provider 的内置 Provider，凭据解析走 resolve_api_key_provider_credentials()——它按 api_key_env_vars 的优先级列表依次检查环境变量。

5. 本地模型自动发现

当 base_url 指向 localhost 或 127.0.0.1 时，Hermes 会自动探测本地运行的模型名：

def _auto_detect_local_model(base_url) -> str:
    try:
        import requests
        url = base_url.rstrip("/")
        if not url.endswith("/v1"):
            url += "/v1"
        resp = requests.get(url + "/models", timeout=5)
        if resp.ok:
            models = resp.json().get("data", [])
            if len(models) == 1:
                return models[0].get("id", "")
    except Exception:
        pass
    return ""

如果本地服务只加载了一个模型，Hermes 会直接用它的 ID 作为默认模型名，不需要用户在配置里写模型名。

6. 自定义 Provider 解析：新式 vs 旧式

_get_named_custom_provider() 负责解析用户在 config.yaml 中自定义的 Provider。

新式（providers: dict）

providers:
  my-deepseek:
    api: "https://api.deepseek.com/v1"
    key_env: "DEEPSEEK_API_KEY"
    default_model: "deepseek-chat"
    transport: "chat_completions"

代码路径：

providers = config.get("providers")
if isinstance(providers, dict):
    for ep_name, entry in providers.items():
        name_norm = _normalize_custom_provider_name(ep_name)
        key_env = str(entry.get("key_env", "") or "").strip()
        resolved_api_key = os.getenv(key_env, "").strip() if key_env else ""
        if not resolved_api_key:
            resolved_api_key = str(entry.get("api_key", "") or "").strip()
        if requested_norm in {ep_name, name_norm, f"custom:{name_norm}"}:
            base_url = entry.get("api") or entry.get("url") or entry.get("base_url") or ""
            result = {
                "name": entry.get("name", ep_name),
                "base_url": base_url.strip(),
                "api_key": resolved_api_key,
                "model": entry.get("default_model", ""),
            }
            api_mode = _parse_api_mode(entry.get("api_mode") or entry.get("transport"))
            if api_mode:
                result["api_mode"] = api_mode
            return result

旧式（custom_providers: list）

custom_providers = get_compatible_custom_providers(config)
for entry in custom_providers:
    name = entry.get("name")
    base_url = entry.get("base_url")
    result = {
        "name": name.strip(),
        "base_url": base_url.strip(),
        "api_key": str(entry.get("api_key", "") or "").strip(),
    }
    api_mode = _parse_api_mode(entry.get("api_mode"))
    if api_mode:
        result["api_mode"] = api_mode
    return result

旧式 entry 的 api_key 直接写在配置里，新式推荐用 key_env 指向环境变量——安全一些。

7. 一条完整的调度链

把全部串起来，从 /model 命令到 HTTP 请求的完整路径：

用户：/model deepseek/deepseek-v4-pro
    ↓
model_switch.py → 更新 config 中的 model.default
    ↓
runtime_provider.resolve_requested_provider("deepseek")
    → "deepseek" （直接匹配，不走 fallback）
    ↓
resolve_api_key_provider_credentials("deepseek")
    → 查 PROVIDER_REGISTRY["deepseek"].api_key_env_vars
    → "DEEPSEEK_API_KEY" → os.getenv("DEEPSEEK_API_KEY") → sk-xxx
    ↓
_resolve_runtime_from_pool_entry(provider="deepseek", ...)
    → api_mode = "chat_completions"
    → base_url = "https://api.deepseek.com/v1"
    ↓
Agent Loop 拿到运行时配置：
    {"provider": "deepseek", "api_mode": "chat_completions",
     "base_url": "https://api.deepseek.com/v1", "api_key": "sk-xxx"}
    ↓
Prompt Builder 构造请求 payload
    → model_metadata.py 查上下文长度
    → 拼装 OpenAI 格式 tool_definitions
    ↓
HTTP POST https://api.deepseek.com/v1/chat/completions

本系列基于 Hermes Agent v0.15.2 源码。模型调度文件：hermes_cli/runtime_provider.py（1694 行）、hermes_cli/auth.py（7706 行）、hermes_cli/model_switch.py、agent/model_metadata.py。

本文由 admin 原创，转载请注明出处。

Hermes Agent Provider 模型调度 源码分析

相关推荐

评论

0
暂无评论，来发表第一条评论吧

发表评论

登录 后发表评论

发现更多