📦 源码仓库

https://github.com/NousResearch/hermes-agent

本地源码：~/.hermes/hermes-agent/

本讲核心文件：

providers/base.py（214 行）— ProviderProfile 基类

providers/__init__.py（191 行）— 注册表与发现

plugins/model-providers/（30+ 提供商插件）

hermes_cli/auth.py（7926 行）— 多源认证系统

hermes_cli/runtime_provider.py（1775 行）— 运行时路由

agent/credential_pool.py（2184 行）— 凭证池

agent/transports/chat_completions.py（704 行）— 传输层

agent/auxiliary_client.py（5949 行）— 辅助客户端路由

# providers/base.py:38-95
@dataclass
class ProviderProfile:
    """Base provider profile — subclass or instantiate with overrides."""

    # ── Identity ──
    name: str                        # 唯一标识: "anthropic", "openrouter"
    api_mode: str = "chat_completions"  # 协议模式
    aliases: tuple = ()              # 别名: ("claude", "claude-oauth")

    # ── 人类可读元数据 ──
    display_name: str = ""           # 显示名称: "GMI Cloud"
    description: str = ""            # 描述: "GMI Cloud (multi-model direct API)"
    signup_url: str = ""             # 注册链接

    # ── 认证与端点 ──
    env_vars: tuple = ()             # 环境变量: ("ANTHROPIC_API_KEY",)
    base_url: str = ""               # API 基础 URL
    models_url: str = ""             # 模型列表端点（可选覆盖）
    auth_type: str = "api_key"       # api_key|oauth_device_code|oauth_external|copilot|aws_sdk
    supports_health_check: bool = True

    # ── 视觉支持 ──
    supports_vision: bool = False
    supports_vision_tool_messages: bool = True

    # ── 模型目录 ──
    fallback_models: tuple = ()      # 离线时的备选模型列表
    hostname: str = ""

    # ── 客户端级特性 ──
    default_headers: dict = field(default_factory=dict)

    # ── 请求级特性 ──
    fixed_temperature: Any = None    # None=默认, OMIT_TEMPERATURE=不发送
    default_max_tokens: int | None = None
    default_aux_model: str = ""      # 辅助任务用廉价模型

钩子方法                                    作用                      典型消费者
─────────────────────────────────────────────────────────────────────────────
get_hostname()                              URL 反查                model_metadata.py
prepare_messages(msgs)                      消息预处理              Qwen 归一化 + cache_control
build_extra_body(**ctx)                     extra_body 字段         OpenRouter provider prefs
build_api_kwargs_extras(**ctx)              (extra_body, top_level) Kimi reasoning_effort
fetch_models(*, api_key)                    实时模型目录              模型选择器

# plugins/model-providers/anthropic/__init__.py:13-52
class AnthropicProfile(ProviderProfile):
    def fetch_models(self, *, api_key, timeout=8.0):
        req = urllib.request.Request("https://api.anthropic.com/v1/models")
        req.add_header("x-api-key", api_key)        # ← 非 Bearer!
        req.add_header("anthropic-version", "2023-06-01")
        ...

anthropic = AnthropicProfile(
    name="anthropic",
    aliases=("claude", "claude-oauth", "claude-code"),
    api_mode="anthropic_messages",
    env_vars=("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"),
    base_url="https://api.anthropic.com",
    auth_type="api_key",
    default_aux_model="claude-haiku-4-5-20251001",
)
register_provider(anthropic)

# providers/__init__.py:43-73
_REGISTRY: dict[str, ProviderProfile] = {}
_ALIASES: dict[str, str] = {}
_discovered = False

def get_provider_profile(name: str) -> ProviderProfile | None:
    if not _discovered:
        _discover_providers()          # ← 懒加载触发点
    canonical = _ALIASES.get(name, name)
    return _REGISTRY.get(canonical)

# providers/__init__.py:140-191
def _discover_providers() -> None:
    """三阶段发现，后者覆盖前者:
      1. Bundled plugins: <repo>/plugins/model-providers/<name>/
      2. User plugins:    $HERMES_HOME/plugins/model-providers/<name>/
      3. Legacy modules:  providers/<name>.py (向后兼容)
    """
    global _discovered
    _discovered = True
    # 阶段 1: bundled
    for child in sorted(_BUNDLED_PLUGINS_DIR.iterdir()):
        _import_plugin_dir(child, "bundled")
    # 阶段 2: user (覆盖同名 bundled)
    for child in sorted(user_dir.iterdir()):
        _import_plugin_dir(child, "user")
    # 阶段 3: legacy single-file
    for _, modname, _ in pkgutil.iter_modules(_pkg.__path__):
        importlib.import_module(f"providers.{modname}")

下游消费者                    读取内容
────────────────────────────────────────────
hermes_cli/auth.py            → PROVIDER_REGISTRY 扩展 api-key profile
hermes_cli/models.py          → 调用 profile.fetch_models()
hermes_cli/doctor.py          → 对 auth_type="api_key" 做 /models 健康检查
hermes_cli/config.py          → 注入 env_vars 到 OPTIONAL_ENV_VARS
hermes_cli/runtime_provider.py → profile.api_mode 作为 URL 检测失败时的回退
agent/model_metadata.py       → profile.get_hostname() 做 hostname→provider 映射
agent/auxiliary_client.py     → profile.default_aux_model 优先于硬编码 dict
agent/transports/chat_completions.py → 调用 prepare_messages(),
                                       build_extra_body(), build_api_kwargs_extras()
run_agent.py                  → 传递 provider_profile= 取代旧标志路径

认证类型                    机制                          典型提供商
─────────────────────────────────────────────────────────────────
api_key                   环境变量 / config.yaml          OpenRouter, Gemini, Kimi
oauth_device_code         RFC 8628 设备码流程             Nous Portal, OpenAI Codex
oauth_external            外部 OAuth 流程                 Qwen, xAI Grok, MiniMax
copilot                   GitHub Copilot ACP              Copilot, Copilot ACP
aws_sdk                   AWS SDK 凭证                    Bedrock

# hermes_cli/auth.py:150-200
@dataclass
class ProviderConfig:
    id: str                        # 唯一 ID: "nous", "openai-codex"
    name: str                      # 显示名: "Nous Portal"
    auth_type: str                 # 认证类型
    portal_base_url: str = ""      # OAuth 门户 URL
    inference_base_url: str = ""   # 推理端点 URL
    client_id: str = ""            # OAuth client_id
    scope: str = ""                # OAuth scope
    extra: Dict[str, Any] = field(default_factory=dict)
    api_key_env_vars: tuple = ()   # API key 环境变量
    base_url_env_var: str = ""     # 可选的 base_url 环境变量

PROVIDER_REGISTRY = {
    "nous": ProviderConfig(
        id="nous", name="Nous Portal",
        auth_type="oauth_device_code",
        portal_base_url="https://portal.nousresearch.com",
        inference_base_url="https://inference-api.nousresearch.com/v1",
        client_id="hermes-cli", scope="inference:invoke",
    ),
    "openai-api": ProviderConfig(
        id="openai-api", name="OpenAI API",
        auth_type="api_key",
        inference_base_url="https://api.openai.com/v1",
        api_key_env_vars=("OPENAI_API_KEY",),
        base_url_env_var="OPENAI_BASE_URL",
    ),
    ...
}

认证状态结构 (auth.json):
{
  "version": 1,
  "active_provider": "nous",
  "providers": {
    "nous": {
      "tokens": {
        "access_token": "eyJ...",
        "refresh_token": "...",
        "agent_key": "eyJ..."        ← Nous 专用 invoke JWT
      },
      "expires_at": "2026-07-18T...",
      "last_refresh": "2026-06-18T..."
    }
  },
  "credential_pools": {
    "openrouter": [ ... ],           ← 多 key 凭证池
    "custom:together-ai": [ ... ]
  }
}

# hermes_cli/runtime_provider.py:426-442
def resolve_requested_provider(requested=None) -> str:
    """Provider 决议优先级链:
      1. 显式参数 (hermes --provider xxx)
      2. config.yaml model.provider
      3. HERMES_INFERENCE_PROVIDER 环境变量
      4. "auto" (自动检测)
    """
    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"

# runtime_provider.py:76-100
def _detect_api_mode_for_url(base_url: str) -> Optional[str]:
    """通过 URL 自动检测 api_mode:
      - api.openai.com → codex_responses (GPT-5.x 需要 Responses API)
      - api.x.ai → codex_responses
      - URL 以 /anthropic 结尾 → anthropic_messages
      - api.kimi.com/coding → anthropic_messages
    """
    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

# 有效的 api_mode 集合 (runtime_provider.py:240-251)
_VALID_API_MODES = {
    "chat_completions",      # OpenAI 兼容 (默认)
    "codex_responses",       # OpenAI Codex Responses API
    "anthropic_messages",    # Anthropic Messages API
    "bedrock_converse",      # AWS Bedrock
    "codex_app_server",      # 通过 codex app-server 子进程
}

# agent/credential_pool.py:129-163
@dataclass
class PooledCredential:
    provider: str                    # 提供商: "openrouter", "nous"
    id: str                          # 唯一 ID (6-char hex)
    label: str                       # 标签 (JWT email 或来源)
    auth_type: str                   # "oauth" 或 "api_key"
    priority: int                    # 优先级 (数字越小越优先)
    source: str                      # 来源: "device_code", "manual", "loopback_pkce"
    access_token: str                # 主凭证
    refresh_token: Optional[str]     # OAuth 刷新令牌
    last_status: Optional[str]       # "ok", "exhausted", "dead"
    last_status_at: Optional[float]  # 状态时间戳
    last_error_code: Optional[int]   # HTTP 错误码
    last_error_reason: Optional[str] # 错误原因
    last_error_reset_at: Optional[float]  # 恢复时间
    base_url: Optional[str]          # 端点 URL
    expires_at: Optional[str]        # 过期时间
    request_count: int = 0           # 请求计数
    extra: Dict[str, Any] = None     # 扩展字段

# credential_pool.py:96-105
SUPPORTED_POOL_STRATEGIES = {
    "fill_first",     # 填满第一个 key 的配额后再切下一个 (默认)
    "round_robin",    # 轮询: 均匀分配请求到所有 key
    "random",         # 随机选择
    "least_used",     # 选择请求数最少的 key
}

# 在 config.yaml 中配置:
credential_pool_strategies:
  openrouter: round_robin
  custom:my-api: least_used

同步路径 (credential_pool.py:541-770):

  _sync_anthropic_entry_from_credentials_file()
    → 检测 ~/.claude/.credentials.json 变化
    → OAuth refresh token 是单次使用的，外部刷新后必须同步

  _sync_codex_entry_from_auth_store()
    → 检测 auth.json 中 openai-codex 凭证更新
    → 防止 pool entry 冻结在 last_error_reset_at 之前

  _sync_xai_oauth_entry_from_auth_store()
    → 检测 xAI OAuth 凭证更新

  _sync_nous_entry_from_auth_store()
    → 检测 Nous OAuth 凭证 + agent_key 更新

  _sync_device_code_entry_to_auth_store()
    → 将 pool 刷新后的凭证写回 auth.json
    → 防止下次 load_pool() 用旧 token 覆盖新 token

# agent/transports/chat_completions.py:274-284
def build_kwargs(self, model, messages, tools=None, **params):
    sanitized = self.convert_messages(messages, model=model)

    # ── Profile 路径: 当 provider_profile 存在时 ──
    _profile = params.get("provider_profile")
    if _profile:
        return self._build_kwargs_from_profile(
            _profile, model, sanitized, tools, params
        )

    # ── 标志路径: 未注册/未知提供商的后备 ──
    # 使用 is_openrouter, is_nous, is_qwen_portal 等标志
    ...

# chat_completions.py:458-597
def _build_kwargs_from_profile(self, profile, model, msgs, tools, params):
    """Profile 路径 — 所有特性来自 profile 对象"""

    # 1. 消息预处理
    msgs = profile.prepare_messages(msgs)

    # 2. Developer role swap (GPT-5/Codex)
    if any(p in model_lower for p in DEVELOPER_ROLE_MODELS):
        msgs[0]["role"] = "developer"

    # 3. Temperature
    if profile.fixed_temperature == OMIT_TEMPERATURE:
        pass  # 不发送 temperature
    elif profile.fixed_temperature is not None:
        kwargs["temperature"] = profile.fixed_temperature

    # 4. max_tokens (优先级: ephemeral > user > profile default)
    profile_max = profile.get_max_tokens(model)

    # 5. Provider-specific extras
    extra_body, top_level = profile.build_api_kwargs_extras(
        reasoning_config=..., supports_reasoning=..., model=model, ...
    )
    kwargs.update(top_level)

    # 6. extra_body 组装
    profile_body = profile.build_extra_body(
        session_id=..., provider_preferences=..., model=model, ...
    )
    extra_body.update(profile_body)

    # 7. Gemini native 特殊处理 — 过滤 extra_body 中的 OpenAI 特有字段
    if is_native_gemini_base_url(base_url):
        extra_body = {k: v for k, v in extra_body.items()
                      if k in ("thinking_config", "thinkingConfig")}

    return kwargs

# auxiliary_client.py:7-40
文本任务降级链:
  1. 用户主提供商 + 主模型
  2. OpenRouter  (OPENROUTER_API_KEY)
  3. Nous Portal (auth.json active provider)
  4. 自定义端点 (model.base_url + OPENAI_API_KEY)
  5. 原生 Anthropic
  6. 直连 API key 提供商 (z.ai/GLM, Kimi, MiniMax)
  7. None (无可用提供商)

视觉任务降级链:
  1. 主提供商 (如果支持视觉)
  2. OpenRouter
  3. Nous Portal
  4. 原生 Anthropic
  5. 自定义端点 (本地视觉模型: Qwen-VL, LLaVA, Pixtral)
  6. None

⚠️ OpenAI Codex OAuth 故意不在降级链中:
   OpenAI 对此端点使用未公开、不断变化的模型白名单，
   "硬编码模型尝试"会自行过期。

支付/额度耗尽降级 (auxiliary_client.py:36-41):
  当解析的提供商返回 HTTP 402 或额度相关错误时，
  call_llm() 自动用降级链中下一个可用提供商重试。
  处理常见场景: 用户耗尽 OpenRouter 余额但仍有 Codex OAuth
  或其他提供商可用。

Provider 抽象层使用的设计模式:

  ① 声明式配置 (Declarative Config)
     ProviderProfile dataclass 声明所有特性，
     而非散落在各层的硬编码标志。

  ② 插件注册 (Plugin Registry)
     三阶段懒发现，user 覆盖 bundled，
     后注册者胜出 (last-writer-wins)。

  ③ 策略模式 (Strategy Pattern)
     凭证池轮换: fill_first / round_robin / random / least_used
     传输层: chat_completions / codex_responses / anthropic_messages

  ④ 状态机 (State Machine)
     凭证状态: OK → EXHAUSTED → DEAD
     带 TTL 冷却和 provider-supplied reset_at 覆盖。

  ⑤ 钩子系统 (Hook System)
     Profile 的 5 个可覆盖钩子:
     get_hostname, prepare_messages, build_extra_body,
     build_api_kwargs_extras, fetch_models

  ⑥ 自动降级 (Automatic Fallback)
     辅助客户端的多层降级链 + 支付耗尽自动重试。

  ⑦ 跨进程同步 (Cross-process Sync)
     凭证池与 auth.json 的双向同步，
     防止 OAuth 刷新令牌被重复消费。