《智能体栈走向一体化》-2026-08-16

③ Claude 模型系列第 9 期：主备供应商切换时记录请求 ID，避免盲目重试

• 难度： 进阶｜动手： 30—45 分钟｜成本： 脚本免费；API 按实际词元计费
• 环境： Python 3.10+；安装 anthropic 当前稳定版；设置 ANTHROPIC_API_KEY；备用线路通过 ANTHROPIC_FALLBACK_BASE_URL 与 ANTHROPIC_FALLBACK_API_KEY 配置
• 场景： 承接第 8 期风险路由与费用熔断，在临时网络错误、429、5xx 或 529 时切到备用兼容线路，并把 Anthropic request-id 写入账本。官方说明每个响应都有唯一请求 ID，官方 SDK默认会对瞬时错误重试两次；本例关闭 SDK 内部重试，避免与外层切换叠加。

安装依赖：

python -m pip install "anthropic>=0.60,<1"

保存为 claude_failover.py：

import json, os, sys, time, uuid
from datetime import datetime, timezone
from pathlib import Path

import anthropic

MODEL = os.getenv("CLAUDE_MODEL", "claude-sonnet-5")
LEDGER = Path(os.getenv("CLAUDE_FAILOVER_LEDGER", "claude_failover.jsonl"))

providers = [
    {
        "name": "主线路",
        "api_key": os.environ["ANTHROPIC_API_KEY"],
        "base_url": os.getenv("ANTHROPIC_BASE_URL", "https://api.anthropic.com"),
    }
]
if os.getenv("ANTHROPIC_FALLBACK_API_KEY") and os.getenv("ANTHROPIC_FALLBACK_BASE_URL"):
    providers.append({
        "name": "备用线路",
        "api_key": os.environ["ANTHROPIC_FALLBACK_API_KEY"],
        "base_url": os.environ["ANTHROPIC_FALLBACK_BASE_URL"],
    })

RETRYABLE = (
    anthropic.RateLimitError,
    anthropic.InternalServerError,
    anthropic.APIConnectionError,
    anthropic.APITimeoutError,
)

def log(row):
    with LEDGER.open("a", encoding="utf-8") as f:
        f.write(json.dumps(row, ensure_ascii=False) + "\n")

def ask(prompt):
    operation_id = str(uuid.uuid4())
    last_error = None
    for provider in providers:
        client = anthropic.Anthropic(
            api_key=provider["api_key"],
            base_url=provider["base_url"],
            max_retries=0,  # 外层负责切换，避免双重重试
            timeout=90,
        )
        try:
            message = client.messages.create(
                model=MODEL,
                max_tokens=800,
                messages=[{"role": "user", "content": prompt}],
            )
            request_id = getattr(message, "_request_id", None)
            row = {
                "time": datetime.now(timezone.utc).isoformat(),
                "operation_id": operation_id,
                "provider": provider["name"],
                "request_id": request_id,
                "model": MODEL,
                "input_tokens": message.usage.input_tokens,
                "output_tokens": message.usage.output_tokens,
                "status": "success",
            }
            log(row)
            text = "".join(block.text for block in message.content if block.type == "text")
            return row, text
        except RETRYABLE as exc:
            last_error = exc
            request_id = getattr(exc, "request_id", None)
            log({
                "time": datetime.now(timezone.utc).isoformat(),
                "operation_id": operation_id,
                "provider": provider["name"],
                "request_id": request_id,
                "status": "retryable_error",
                "error_type": type(exc).__name__,
            })
            time.sleep(1)
            continue
        except anthropic.APIStatusError as exc:
            # 400、401、403、404 等通常不是换线路能解决的问题，立即停止。
            log({
                "time": datetime.now(timezone.utc).isoformat(),
                "operation_id": operation_id,
                "provider": provider["name"],
                "request_id": getattr(exc, "request_id", None),
                "status": "fatal_error",
                "http_status": exc.status_code,
            })
            raise
    raise RuntimeError(f"主备线路均失败：{type(last_error).__name__}")

if __name__ == "__main__":
    prompt = " ".join(sys.argv[1:]) or "用三点总结主备模型切换的风险。"
    meta, answer = ask(prompt)
    print(json.dumps(meta, ensure_ascii=False, indent=2))
    print(answer)

运行：

export ANTHROPIC_API_KEY="主线路环境变量密钥"
export ANTHROPIC_FALLBACK_BASE_URL="https://你的备用兼容线路/v1"
export ANTHROPIC_FALLBACK_API_KEY="备用线路环境变量密钥"
python claude_failover.py "审查这份发布计划，列出不可逆操作、回滚条件和人工批准点。"

成功标志： 正常时输出主线路和 request_id；主线路遇到限流、连接错误或服务端错误时尝试备用线路；账本用同一 operation_id 串起所有尝试。常见坑： 备用线路未必暴露 Anthropic 请求 ID；401/403 多为密钥或权限错误，不应盲目切换；对付款、发信、下单等有副作用的请求，收到超时后不能直接重放，必须先查询业务状态。不该用： 不能容忍供应商差异、数据不能离开指定区域、或请求具有不可重复副作用时，不应自动跨供应商切换。

参考：https://platform.claude.com/docs/en/api/errors

https://platform.claude.com/docs/en/cli-sdks-libraries/sdks/python

下期预告： 第 10 期加入幂等键、业务状态查询与补偿动作，专门处理“请求超时但服务端可能已经成功”的灰色状态。