路径约定：本文档中模块路径均相对于 src/coding/proxy/，例如 backends/base.py 指 src/coding/proxy/backends/base.py。

1. 项目概述

1.1 项目动机与背景

Claude Code 作为日常 AI 编程助手，其底层依赖 Anthropic Messages API (/v1/messages)。在实际使用过程中，以下场景时有发生：

• 限流 (Rate Limiting)：高频请求触发 429 rate_limit_error，导致短时间内无法继续使用
• 配额耗尽 (Usage Cap)：月度/日度配额用尽后返回 403 错误，含 "usage cap" 等提示
• 服务过载 (Overloaded)：Anthropic 服务端高峰期返回 503 overloaded_error
• 网络波动：国际链路不稳定导致连接超时

与此同时，多个 Anthropic 兼容或可转换的 API 通道为构建多后端容灾体系提供了可能：

• GitHub Copilot：提供 Anthropic 兼容的 Claude API 端点，通过 GitHub OAuth token / PAT 完成 token 交换
• Google Antigravity Claude：通过 Google Gemini/Vertex AI 端点提供 Claude 模型访问，需 Anthropic ↔ Gemini 双向格式转换
• 智谱 (Zhipu)：提供与 Anthropic 兼容的 GLM API 接口（/api/anthropic），作为终端兜底

coding-proxy 的核心诉求：当任一后端不可用时，自动、无缝地沿 N-tier 层级链降级到下一个可用后端，对 Claude Code 客户端完全透明，用户无需手动干预。

1.2 设计目标

2. 系统架构总览

2.1 架构分层图

graph TD
    Client["Claude Code 客户端"]
    Server["FastAPI Server<br/><code>server/app.py</code>"]
    Router["RequestRouter<br/><code>routing/router.py</code>"]

    Client -->|"POST /v1/messages"| Server
    Server --> Router

    Router --> T0
    Router --> T1
    Router --> T2
    Router --> TN

    subgraph T0["BackendTier 0: Anthropic"]
        direction LR
        A_BE["AnthropicBackend"]
        A_CB["CB"]
        A_QG["QG"]
    end

    subgraph T1["BackendTier 1: Copilot"]
        direction LR
        C_BE["CopilotBackend"]
        C_CB["CB"]
        C_QG["QG"]
    end

    subgraph T2["BackendTier 2: Antigravity"]
        direction LR
        G_BE["AntigravityBackend"]
        G_CB["CB"]
        G_QG["QG"]
    end

    subgraph TN["BackendTier N: Zhipu（终端，无 CB/QG）"]
        Z_BE["ZhipuBackend"]
    end

    A_BE --> API_A["Anthropic API"]
    C_BE --> API_C["GitHub Copilot API"]
    G_BE --> API_G["Google Gemini API"]
    Z_BE --> API_Z["智谱 GLM API"]

    style T0 fill:#1a5276,color:#fff
    style T1 fill:#1a5276,color:#fff
    style T2 fill:#1a5276,color:#fff
    style TN fill:#7b241c,color:#fff

术语表：

2.2 模块职责一览

2.3 技术选型

3. 设计模式详解

3.1 Template Method（模板方法模式）

经典出处：GoF《Design Patterns: Elements of Reusable Object-Oriented Software》— 定义算法骨架，将某些步骤延迟到子类实现。

应用位置：backends/base.py — BaseBackend 抽象基类

设计要点：

BaseBackend 定义了请求处理的算法骨架，将差异化的逻辑延迟到子类：

BaseBackend（模板）
├── send_message()          ← 固定流程：prepare → send → parse response
├── send_message_stream()   ← 固定流程：prepare → stream → yield chunks
├── _get_client()           ← 公共逻辑：惰性初始化 httpx.AsyncClient
├── close()                 ← 公共逻辑：关闭 HTTP 客户端
│
├── _prepare_request()      ← 【抽象·异步】子类实现请求转换（支持 token 刷新等异步操作）
├── _get_endpoint()         ← 【钩子】端点路径（默认 /v1/messages）
├── _on_error_status()      ← 【钩子】错误状态码处理（如 token 失效标记）
├── get_name()              ← 【抽象】子类返回后端标识
└── should_trigger_failover() ← 基于 FailoverConfig 判断故障转移

四个具体子类的差异化实现：

3.2 Circuit Breaker（熔断器模式）

经典出处：Martin Fowler "CircuitBreaker" (2014)；Michael Nygard《Release It! Design and Deploy Production-Ready Software》第 5 章 — 通过快速失败防止级联故障。

应用位置：routing/circuit_breaker.py — CircuitBreaker 类

状态机：

                  连续 N 次失败
 ┌────────┐ ──────────────────→ ┌────────┐
 │ CLOSED │                     │  OPEN  │ ◄──┐
 │ (正常)  │                     │ (熔断)  │    │
 └────────┘                     └───┬────┘    │
      ▲                             │          │
      │ 连续 M 次成功      超时恢复  │          │ 失败（退避×2）
      │                             ▼          │
      │                       ┌──────────┐     │
      └───────────────────────│HALF_OPEN │─────┘
                              │ (试探)    │
                              └──────────┘

状态转换条件：

指数退避 (Exponential Backoff)：每次从 HALF_OPEN 回退到 OPEN 时，恢复等待时间翻倍（recovery_timeout *= 2），上限为 max_recovery_seconds（默认 3600 秒）。避免对仍未恢复的后端频繁重试。

线程安全：所有状态变更通过 threading.Lock 保护，确保并发请求下状态一致。

3.3 Priority Chain（优先级匹配链）

应用位置：routing/model_mapper.py — ModelMapper 类

设计要点：

ModelMapper 采用三级优先级匹配链，按精确度递减依次尝试：

输入模型名称
    │
    ▼
 1. 精确匹配（Exact Match）
    │ pattern == model（无通配符、非正则）
    ├─ 命中 → 返回 target
    │
    ▼
 2. 模式匹配（Regex / Glob Match）
    │ is_regex=true → re.fullmatch()
    │ 含 * → fnmatch.fnmatch()
    ├─ 命中 → 返回 target
    │
    ▼
 3. 默认值（Default Fallback）
    └─ 返回 "glm-5.1"

默认映射规则：

正则表达式在 __init__ 时预编译（re.compile()），map() 调用时直接使用编译后的对象，避免重复编译开销。

3.4 Factory Method（工厂方法模式）

经典出处：GoF《Design Patterns》— 定义创建对象的接口，由子类决定实例化哪个类。

应用位置：server/app.py — create_app() 函数

组装顺序：

create_app(config)
    │
    ├─ 1. TokenLogger(config.db_path)
    ├─ 2. ModelMapper(config.model_mapping)
    ├─ 3. 构建 BackendTier 链（每个 tier 使用独立的 CB/QG 配置）：
    │     ├─ Tier 0: AnthropicBackend
    │     │    + _build_circuit_breaker(config.circuit_breaker)
    │     │    + _build_quota_guard(config.quota_guard)
    │     ├─ Tier 1: CopilotBackend（enabled 时）
    │     │    + _build_circuit_breaker(config.copilot_circuit_breaker)
    │     │    + _build_quota_guard(config.copilot_quota_guard)
    │     ├─ Tier 2: AntigravityBackend（enabled 时）
    │     │    + _build_circuit_breaker(config.antigravity_circuit_breaker)
    │     │    + _build_quota_guard(config.antigravity_quota_guard)
    │     └─ Tier N: ZhipuBackend（终端，无 CB/QG）
    ├─ 4. RequestRouter(tiers, token_logger)
    │
    └─ 5. FastAPI(lifespan=lifespan)
         ├─ app.state.router = router
         ├─ app.state.token_logger = token_logger
         └─ app.state.config = config

3.5 Proxy（代理模式）

经典出处：GoF《Design Patterns》— 为其他对象提供一种代理以控制对这个对象的访问。

应用位置：整体架构

coding-proxy 本身即是一个代理服务：

• 对外暴露与 Anthropic Messages API 完全兼容的 POST /v1/messages 接口
• Claude Code 客户端只需将 ANTHROPIC_BASE_URL 指向代理地址
• 代理在幕后完成后端选择、故障转移、模型映射、用量记录等增值逻辑
• 支持流式（SSE text/event-stream）和非流式（JSON）两种响应模式

3.6 Composite（组合模式）— BackendTier

经典出处：GoF《Design Patterns》— 将对象组合成树形结构以表示"部分-整体"的层次结构。

应用位置：routing/tier.py — BackendTier 数据类

设计要点：

BackendTier 将三个正交关注点聚合为路由器的最小调度单元：

┌───────────────┐
│  BackendTier  │ ← 路由器的最小调度单元
├───────────────┤
│  backend      │ ← BaseBackend 实例（实际请求执行）
│  circuit_breaker │ ← CircuitBreaker | None（弹性熔断）
│  quota_guard  │ ← QuotaGuard | None（配额管控）
└───────────────┘

关键方法：

3.7 State Machine（状态机模式）— QuotaGuard

应用位置：routing/quota_guard.py — QuotaGuard 类

设计要点：

基于滑动窗口的双态状态机，通过 Token 预算追踪主动避免触发上游配额限制：

                 窗口用量 ≥ budget × threshold%
                 或检测到 cap error
 ┌──────────────┐ ─────────────────────→ ┌──────────────────┐
 │ WITHIN_QUOTA │                        │ QUOTA_EXCEEDED   │
 │ （正常）      │                        │ （超限）          │
 └──────────────┘ ←───────────────────── └──────────────────┘
                 窗口用量自然滑出 < threshold%
                 或探测请求成功

核心机制：

• 滑动窗口：deque[(timestamp, tokens)]，_expire() 清除超出 window_seconds 的条目
• 探测恢复：QUOTA_EXCEEDED 状态下，每隔 probe_interval_seconds 放行一个探测请求
• cap error 模式：由外部 _is_cap_error() 触发，不做预算自动恢复，仅允许探测恢复
• 基线加载：启动时从数据库加载历史用量，防止重启后误判配额状态
• 线程安全：所有状态变更通过 threading.Lock 保护

3.8 Double-Check Locking（双重检查锁模式）

应用位置：backends/copilot.py — CopilotTokenManager；backends/antigravity.py — GoogleOAuthTokenManager

设计要点：

两个 Token Manager 均采用相同的异步 DCL 模式，确保高并发下 token 交换仅执行一次：

# 快速路径（无锁）
if self._access_token and time.monotonic() < self._expires_at:
    return self._access_token

# 慢路径（加锁后二次检查）
async with self._lock:
    if self._access_token and time.monotonic() < self._expires_at:
        return self._access_token
    await self._exchange()  # 或 self._refresh()

两者均支持被动刷新：当后端返回 401/403 时，通过 _on_error_status() 调用 invalidate() 标记 token 失效，下次请求自动触发重新获取。

4. 请求生命周期

4.1 完整请求流程

Client POST /v1/messages
        │
        ▼
 app.messages() 解析 body + headers
        │
        ├─ stream=true ──→ route_stream()
        └─ stream=false ─→ route_message()
                │
                ▼
     ┌── for tier in tiers ──────────┐
     │                                │
     │  tier.can_execute()?           │
     │  = CB.can_execute()            │
     │    AND QG.can_use_primary()    │
     │        │                       │
     │   ┌────┴────┐                  │
     │   │ Yes      │ No              │
     │   ▼          │(skip, 非终端)   │
     │ tier.backend │  │              │
     │ .send_xxx()  │  │              │
     │   │          │  │              │
     │ ┌─┴──┐       │  │              │
     │ │成功 │ 失败   │  │              │
     │ │    │        │  │              │
     │ │ record     should            │
     │ │ success   failover?          │
     │ │    │        │                │
     │ │    │   ┌────┴───┐            │
     │ │    │   │Yes     │No          │
     │ │    │   │        │            │
     │ │    │  record   return        │
     │ │    │  failure  error         │
     │ │    │  (+cap?)   │            │
     │ │    │   ▼        │            │
     │ │    │  next tier  │            │
     └─┼────┼────────────┘            │
       │    │                         │
       ▼    ▼                         │
    TokenLogger.log()                 │
            │                         │
            ▼                         │
      Response → Client ◄─────────────┘

4.2 流式请求处理

流式请求使用 StreamingResponse + 异步生成器 _stream_proxy()：

1. RequestRouter.route_stream() 返回 AsyncIterator[tuple[bytes, str]]
2. 每个 SSE chunk 通过 _parse_usage_from_chunk() 提取 Token 用量：
   • message_start 事件：提取 input_tokens、cache_creation_input_tokens、cache_read_input_tokens、request_id
   • message_delta 事件：提取 output_tokens
3. chunk 原样透传给客户端
4. 流结束后：
   • tier.record_success(input_tokens + output_tokens) — 通知 CB 成功 + QG 记录用量
   • _record_usage() — 记录完整用量到 TokenLogger

故障转移时：清空已收集的 usage 数据，从下一个 tier 重新开始流式传输。

4.3 非流式请求处理

非流式请求直接调用 send_message() 获取完整响应：

1. 主后端返回成功（status_code < 400）→ record_success() → 返回
2. 主后端返回错误 → 检查 should_trigger_failover()：
   • 是 → record_failure(is_cap_error=_is_cap_error(resp)) → 尝试下一个 tier
   • 否 → 直接返回错误响应
3. 捕获 httpx.TimeoutException / httpx.ConnectError → record_failure() → 故障转移

cap error 检测（RequestRouter._is_cap_error()）：当 status_code in (429, 403) 且 error_message 包含 "usage cap" / "quota" / "limit exceeded" 时，判定为配额上限错误，通过 record_failure(is_cap_error=True) 联动 QG 进入 QUOTA_EXCEEDED 状态。

4.4 故障转移判定逻辑

故障转移的判定在 BaseBackend.should_trigger_failover() 中实现，依据三层条件（可通过配置文件自定义）：

特殊规则：对于 429 和 503 状态码，即使无法解析响应体（body），也会强制触发故障转移。

终端后端行为：ZhipuBackend 因构造时不传入 failover_config，should_trigger_failover() 始终返回 False。

4.5 生命周期管理

通过 lifespan 异步上下文管理器管理应用生命周期：

启动 (Startup)：

1. TokenLogger.init() — 创建 SQLite 数据库表和索引
2. 为每个启用了 QuotaGuard 的 tier 加载基线：
   • token_logger.query_window_total(window_hours, backend=tier.name) — 查询历史用量
   • quota_guard.load_baseline(total_tokens) — 初始化滑动窗口

关闭 (Shutdown)：

1. router.close() — 关闭所有后端 HTTP 客户端（含 TokenManager 客户端）
2. token_logger.close() — 关闭数据库连接

5. 模块详细设计

5.1 backends/ — 后端模块

数据结构（backends/base.py）：

@dataclass
class UsageInfo:
    input_tokens: int = 0
    output_tokens: int = 0
    cache_creation_tokens: int = 0
    cache_read_tokens: int = 0
    request_id: str = ""

@dataclass
class BackendResponse:
    status_code: int = 200
    usage: UsageInfo = field(default_factory=UsageInfo)
    is_streaming: bool = False
    raw_body: bytes = b"{}"
    error_type: str | None = None
    error_message: str | None = None

AnthropicBackend（Tier 0，backends/anthropic.py）：

• 构造：AnthropicBackend(config: AnthropicConfig, failover_config: FailoverConfig)
• 过滤 hop-by-hop 头（host、content-length、transfer-encoding、connection）
• 透传客户端的 OAuth token（authorization 头）和请求体
• 不修改请求体

CopilotBackend（Tier 1，backends/copilot.py）：

• 构造：CopilotBackend(config: CopilotConfig, failover_config: FailoverConfig)
• 内置 CopilotTokenManager：GitHub token → Copilot chat token 交换（兼容 token/refresh_in 与 access_token/expires_in 两种响应）
• 过滤 hop-by-hop 头并注入 Authorization: Bearer {copilot_token}
• 透传请求体（Claude 模型名原生支持，无需映射）
• 401/403 时自动 invalidate token 触发被动刷新
• 覆写 close() 释放 TokenManager 客户端资源

AntigravityBackend（Tier 2，backends/antigravity.py）：

• 构造：AntigravityBackend(config: AntigravityConfig, failover_config: FailoverConfig)
• 内置 GoogleOAuthTokenManager：refresh_token → Google access_token（有效期 ~1 小时，提前 120 秒刷新）
• 调用 convert/ 模块进行 Anthropic → Gemini 请求格式转换
• 覆写 send_message()：POST /{model_endpoint}:generateContent，响应经 convert_response() 逆转换
• 覆写 send_message_stream()：POST /{model_endpoint}:streamGenerateContent?alt=sse，经 adapt_sse_stream() 适配
• 401/403 时自动 invalidate OAuth token
• 覆写 close() 释放 TokenManager 客户端资源

ZhipuBackend（终端 Tier N，backends/zhipu.py）：

• 构造：ZhipuBackend(config: ZhipuConfig, model_mapper: ModelMapper)（无 failover_config）
• 浅拷贝请求体（{**request_body}），避免修改原始数据
• 调用 ModelMapper.map() 映射模型名称（claude-* → glm-*）
• 替换认证头为 x-api-key，保持 anthropic-version 头兼容

HTTP 客户端管理：

• 惰性初始化 httpx.AsyncClient（首次调用时创建）
• 自动检测并重建已关闭的客户端（is_closed 检查）
• close() 方法释放连接资源

5.2 routing/ — 路由模块

BackendTier（routing/tier.py）：

@dataclass
class BackendTier:
    backend: BaseBackend
    circuit_breaker: CircuitBreaker | None = None
    quota_guard: QuotaGuard | None = None

RequestRouter（routing/router.py）：

• 持有 tiers: list[BackendTier] + token_logger，支持 N-tier 链式降级
• route_message() → 非流式路由，按优先级逐层尝试，返回 BackendResponse
• route_stream() → 流式路由，按优先级逐层尝试，yield (chunk, backend_name)
• _record_usage() → 统一记录用量到 TokenLogger
• _is_cap_error() → 静态方法，检测 429/403 + 配额相关关键词，联动 QG

CircuitBreaker 参数（routing/circuit_breaker.py）：

QuotaGuard 参数（routing/quota_guard.py）：

QuotaGuard 公共方法：

5.3 config/ — 配置模块

配置搜索优先级（config/loader.py）：

CLI --config 参数（显式指定）
    ↓ 未指定时
./config.yaml（项目根目录）
    ↓ 不存在时
~/.coding-proxy/config.yaml（用户目录）
    ↓ 不存在时
Pydantic 默认值

环境变量展开：

• 语法：${VARIABLE_NAME}
• 实现：正则 \$\{([^}]+)\} 匹配，递归处理 dict/list/str
• 未定义的变量保留原始 ${VAR} 文本
• 适用场景：API Key 等敏感信息不宜写入配置文件

向后兼容（config/schema.py @model_validator）：

• 配置中的 anthropic 字段自动迁移为 primary
• 配置中的 zhipu 字段自动迁移为 fallback

5.4 logging/ — 日志模块

usage_log 表结构（logging/db.py）：

索引：idx_usage_ts（时间戳）、idx_usage_backend（后端名）

SQLite 优化：WAL (Write-Ahead Logging) 模式，支持读写并发而不互相阻塞。

查询方法：

统计展示（logging/stats.py）：

• show_usage() 使用 Rich Table 格式化展示统计结果
• 列：日期、后端、请求模型、实际模型、请求数、输入 Token、输出 Token、故障转移数、平均耗时

5.5 server/ — 服务模块

API 端点（server/app.py）：

6. 配置系统设计

6.1 完整配置字段参考

server — 服务器配置

primary — 主后端（Anthropic）配置

copilot — GitHub Copilot 后端配置

antigravity — Google Antigravity Claude 后端配置

fallback — 备选后端（智谱）配置

circuit_breaker — 熔断器配置

除全局 circuit_breaker 外，还支持 copilot_circuit_breaker 和 antigravity_circuit_breaker 独立配置，字段与全局相同。

quota_guard — 配额守卫配置

同理支持 copilot_quota_guard 和 antigravity_quota_guard 独立配置。

failover — 故障转移触发条件

model_mapping — 模型名称映射规则

database — 数据库配置

logging — 日志配置

向后兼容：配置中的 anthropic 字段会自动迁移为 primary，zhipu 字段自动迁移为 fallback。

7. 可扩展性设计

7.1 添加新后端

1. 在 backends/ 下创建新模块，继承 BaseBackend
2. 实现必需的抽象方法：
   • get_name() — 返回后端标识字符串
   • _prepare_request() — 转换请求体和请求头（异步，支持 token 刷新）
3. 按需覆写钩子方法：
   • _on_error_status() — 错误时标记 token 失效
   • send_message() / send_message_stream() — 自定义端点或格式转换
   • close() — 释放额外资源
4. 在 config/schema.py 中添加后端配置模型和对应的 CB/QG 配置字段
5. 在 server/app.py 的 create_app() 中构建 BackendTier 并插入到 tiers 链中

7.2 添加新映射规则

在配置文件的 model_mapping 节添加规则即可，无需修改代码：

model_mapping:
  - pattern: "claude-sonnet-4-6"   # 精确匹配
    target: "glm-5.1"
  - pattern: "claude-haiku-.*"      # 正则匹配
    target: "glm-4.5-air"
    is_regex: true
  - pattern: "claude-*"             # Glob 通配符
    target: "glm-5.1"

匹配优先级确保精确匹配不会被通配符覆盖。

7.3 自定义故障转移策略

通过配置文件调整 failover 节的三个字段即可自定义触发条件：

• 增减 status_codes 控制哪些 HTTP 状态码触发切换
• 增减 error_types 控制哪些 Anthropic 错误类型触发切换
• 增减 error_message_patterns 控制哪些错误消息关键词触发切换

如需更复杂的策略，可重写子类的 should_trigger_failover() 方法。

7.4 自定义配额管理策略

通过配置文件分别调整 quota_guard / copilot_quota_guard / antigravity_quota_guard 的参数，可为每个后端独立配置配额管理策略：

quota_guard:                # Anthropic 主后端
  enabled: true
  token_budget: 5000000
  window_hours: 5.0
  threshold_percent: 95.0
  probe_interval_seconds: 600

copilot_quota_guard:        # Copilot 中间层
  enabled: true
  token_budget: 3000000
  window_hours: 4.0

8. convert/ 模块设计

8.1 请求转换（Anthropic → Gemini）

应用位置：convert/anthropic_to_gemini.py — convert_request()

转换映射：

不支持的字段（静默剥离并记录 WARNING）：tools、tool_choice、metadata、extended_thinking、thinking

8.2 响应转换（Gemini → Anthropic）

应用位置：convert/gemini_to_anthropic.py — convert_response() / extract_usage()

finishReason 映射：

Parts 转换：

• text → {"type": "text", "text": "..."}
• functionCall → {"type": "tool_use", "id": "toolu_...", "name": "...", "input": {...}}

Usage 提取：

• usageMetadata.promptTokenCount → input_tokens
• usageMetadata.candidatesTokenCount → output_tokens
• 缓存字段填 0（Gemini 不直接暴露缓存信息）

8.3 SSE 流适配

应用位置：convert/gemini_sse_adapter.py — adapt_sse_stream()

将 Gemini SSE 流重构为 Anthropic 消息生命周期事件序列：

Gemini SSE chunks
    │
    ▼
message_start           ← 首次收到内容时发出
    │
content_block_start     ← 内容块开始
    │
content_block_delta*    ← 增量文本（每个 text part 一个）
    │
content_block_stop      ← 内容块结束
    │
message_delta           ← stop_reason + output_tokens
    │
message_stop            ← 消息结束

边界情况处理：

• 空 parts 后跟有 text 的 chunk → 延迟发出 message_start + content_block_start
• 流结束但未收到 finishReason → 补发默认 message_delta（stop_reason: "end_turn"）+ message_stop

9. 测试策略

9.1 单元测试覆盖

9.2 测试工具

• pytest (>=9.0) — 测试框架
• pytest-asyncio (>=1.3) — 异步测试支持
• monkeypatch — 环境变量和工作目录隔离
• tmp_path — 临时文件测试