Anthropic SDK 错误处理与可观测

调用跑通只是起点。生产环境的每一次请求都走在"会失败"的边缘：参数拼错是 400，发得太快是 429，赶上高峰是 529，网络抖一下连响应都拿不到。本系列前面各篇零散地处置过它们——第 1 篇的报错速查、第 4 篇的 tool_result 层级错配、第 5 篇的超长输入、第 6 篇的思考块签名校验，都是"遇到了再对症"。本篇把这些碎片拼成一张完整的图：失败怎么分类、SDK 替你重试了什么、异常怎么精确捕获、超时怎么配、出了问题拿什么找官方，最后落成一个能自己扛住故障的批跑器。

本文是《Anthropic SDK 从入门到实战》第 11 篇，倒数第二篇。全文按"失败分类 → 自动重试 → 异常捕获 → 超时配置 → 排查工具 → 可观测 → 实战 → 总结"组织。

1、失败分类全景

错误处理的第一课不是写 try，而是收到异常后 30 秒内判断"错在哪一层、要不要重试"。日常失败逃不出五类：请求错（4xx，请求本身有问题）、限流（429，发得太快）、超载（5xx 与 529，服务端扛不住）、超时（客户端等不及了）、网络（压根没连上）。判定流程如下：

解读：这棵树最值钱的是第一个分叉——先分"有没有拿到 HTTP 响应"。拿到了，走状态码分支：4xx 改请求，429 与 5xx 退避重试；没拿到，要么超时要么断连，都属于传输层抖动，重试通常有效。第 1 篇的速查表回答"这是什么错"，这棵树回答"下一步干什么"，两张图配合用。各类错误码的官方口径见错误码文档。

2、SDK 自动重试的边界

很多 429 从没到你眼前——SDK 已经默默重试并成功了。Python SDK 默认 max_retries=2，一次调用最多尝试 3 次；自动重试的范围是连接错误、408、409、429 与全部 5xx，指数退避，并尊重响应里的 retry-after 头。这两个默认值定义在 SDK 源码的 _constants.py 里。配置方式如下：

import anthropic

client = anthropic.Anthropic(max_retries=4)          # 全局：最多尝试 5 次
client.with_options(max_retries=0).messages.create(  # 单次调用：关掉自动重试
    model="claude-sonnet-5-5", max_tokens=1024,
    messages=[{"role": "user", "content": "你好"}],
)

解读：构造参数管全局，with_options 管单次，max_retries=0 彻底关闭。默认 2 次对多数场景够用，真正要想清楚的是"谁来重试"。两层重试的分工边界如下：

分界线只有一条：SDK 只管传输层抖动，从不替你改请求。400 重发一百次还是 400，第 4 篇的层级错配、第 5 篇的超长输入、第 6 篇的签名失败都在此列；"模型输出违反业务规则"这类失败 SDK 根本看不见，得走第 3 篇的校验反馈循环。限流档位的构成见官方限流文档。

注意：自动重试会与外层重试叠乘——外层再包 3 轮，最坏 3 × 3 共 9 次请求、9 倍延迟。调用方已有重试机制时，给客户端传 max_retries=0 只留一层。

3、异常类地图与精确捕获

自动重试放弃后，异常才抛到你手里。SDK 为每个状态码定义了专属异常类，全部继承自 APIError。Python SDK 的完整映射如下：

按这张表写捕获链，从最具体的子类排到基类：

try:
    response = client.messages.create(
        model="claude-sonnet-5-5", max_tokens=1024,
        messages=[{"role": "user", "content": "用一句话解释退避"}],
    )
except anthropic.RateLimitError as e:
    wait = e.response.headers.get("retry-after", "60")  # 服务端建议的秒数
    print(f"限流，{wait} 秒后再试")
except anthropic.BadRequestError as e:
    print(f"请求有错：{e.message}")      # 逐句读，通常直接点名缺什么
except anthropic.APIStatusError as e:
    print(f"其他 HTTP 错误 {e.status_code}：{e.type}")
except anthropic.APIConnectionError as e:
    print(f"网络异常：{e.message}")

解读：三处细节。其一，except 从子类写到基类，Python 按顺序匹配，把 APIStatusError 排在前面会吞掉子类的区分度；其二，APIConnectionError 与 APIStatusError 在 Python SDK 里是兄弟而非父子——前者根本没有响应可读，捕获链两者都要列（TypeScript SDK 里它是 APIError 的子类，跨语言迁移时留意）；其三，e.type 返回 invalid_request_error、overloaded_error 这类官方错误类型串，比状态码更细——529 超载与普通 500 同落 InternalServerError，区分它们就靠这个字段。顺带补两个少见码：402 是计费问题、403 是权限不足，.type 分别返回 billing_error 与 permission_error——都在"改代码没用、去控制台办事"之列。

4、超时配置

默认超时 10 分钟（600 秒总限，建连 5 秒），对"等一份长回复"宽容，对交互场景残忍——没有用户愿意盯着转圈等十分钟。三种配置粒度如下：

import anthropic

client = anthropic.Anthropic(timeout=30.0)             # 总超时 30 秒
client = anthropic.Anthropic(                          # 分段控制
    timeout=anthropic.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)
client.with_options(timeout=300.0).messages.create(    # 单次调用覆盖
    model="claude-sonnet-5-5", max_tokens=4096,
    messages=[{"role": "user", "content": "写一首长诗"}],
)

解读：传浮点数设总超时，单位是秒；anthropic.Timeout 拆开建连、读、写各段，read 管"服务端多久不吐字节"。超时抛出的是 APITimeoutError，第 3 节捕获链的 APIConnectionError 分支天然接得住。一个容易被忽略的乘法：超时同样计入自动重试——600 秒默认值配 2 次重试，最坏 1800 秒才见分晓；反过来把超时压到 30 秒去跑长输出，会收获一串超时，正确解法是换第 2 篇的流式，而不是干等。

5、排查工具箱

分类与捕获解决"我这边的处置"，排查要回答"为什么错"。两件工具排在最前面。

第一件是 request-id。每个响应都带 request-id 头：成功响应读 response._request_id（下划线开头，但官方明确它是公开属性），异常从 e.response.headers 里取。向 Anthropic 支持报障时附上它，对方能直接定位那次请求的服务端日志，省掉几轮来回追问。统一出口的打印示例如下：

try:
    response = client.messages.create(
        model="claude-sonnet-5-5", max_tokens=1024,
        messages=[{"role": "user", "content": "查一下订单状态"}])
except anthropic.APIStatusError as e:
    rid = e.response.headers.get("request-id", "无")
    print(f"状态码 {e.status_code}，类型 {e.type}，request-id {rid}")
    print(e.message)                   # 错误正文逐句读

解读：四个字段一行打齐，贴进工单或内部告警群即可。除了响应头，错误返回体里也带同样的 request_id 字段，两条路取一条即可；429 的响应头另有 x-ratelimit-remaining 系列字段，能直接读出配额余量，接进告警比事后翻日志及时。第二件工具是 message 本身——Anthropic 的错误正文以直白著称，缺哪个参数、哪条消息角色不合法、思考块签名为什么不通过，通常一句话写明；第 1 篇"先状态码、再 type、再正文"的读法在这里兑现。遇到持续性 5xx，先看官方状态页 status.anthropic.com，分清"我的问题"与"大家的问题"再动手。

6、可观测三件套

错误处理守住下限，可观测抬高上限：在事故发生前看见趋势。三件事按投入产出排序：调用日志、usage 落账、轨迹平台。给 messages.create 套一层日志中间件，前两件一次到手：

import logging
import time
import anthropic

logger = logging.getLogger("anthropic.calls")

def logged_create(client, **kwargs):
    """带耗时、用量与 request-id 的调用包装。"""
    start = time.monotonic()
    try:
        r = client.messages.create(**kwargs)
    except anthropic.APIError as e:
        logger.error("调用失败 类型=%s", type(e).__name__)
        raise
    logger.info(
        "model=%s 耗时=%.0fms in=%d out=%d cache_read=%d rid=%s",
        r.model, (time.monotonic() - start) * 1000,
        r.usage.input_tokens, r.usage.output_tokens,
        r.usage.cache_read_input_tokens, r._request_id,
    )
    return r

解读：日志字段与第 7、9 篇的账本口径一致——input_tokens 与 output_tokens 是钱，cache_read_input_tokens 是缓存命中，攒一周就能回答"限流前兆长什么样""成本环比涨在哪"。调试 SDK 自身行为时，环境变量 ANTHROPIC_LOG=debug 可打开 SDK 内部日志。第三件轨迹平台：官方提供 OpenTelemetry 追踪，见可观测文档；LangChain 生态的项目打开 LangSmith 就有全链路轨迹，这里点到为止。

7、实战：稳健的异步批跑器

把全篇零件拼成一台整机：批处理 500 条数据，要求并发执行、瞬时失败自动退避重试、最终失败按类别落盘留档。第 8 篇的 Batch 接口适合不赶时间的半价场景，这里走同步批跑器——结果即时可取、失败即时可见。先看异常分类与单任务执行：

import asyncio, random
import anthropic

client = anthropic.AsyncAnthropic(max_retries=2, timeout=120.0)

def classify(exc: Exception) -> str:
    """把异常压成四类标签，决定是否值得重试。"""
    if isinstance(exc, anthropic.RateLimitError):
        return "rate_limit"
    if isinstance(exc, anthropic.InternalServerError):
        return "server_error"            # 529 超载也落在这里
    if isinstance(exc, anthropic.APIConnectionError):
        return "network"                 # 含超时
    return "fatal"                       # 400 等：请求本身的问题

async def run_one(task, sem):
    async with sem:
        for attempt in range(4):         # 外层再兜 4 轮
            try:
                r = await client.messages.create(**task["params"])
                return {"id": task["id"], "ok": True, "out": r.usage.output_tokens}
            except anthropic.APIError as e:
                label = classify(e)
                if label == "fatal" or attempt == 3:
                    headers = getattr(e.response, "headers", None)
                    return {"id": task["id"], "ok": False, "label": label,
                            "code": getattr(e, "status_code", None),
                            "rid": headers and headers.get("request-id"),
                            "msg": e.message}
                await asyncio.sleep(min(2 ** attempt + random.random(), 30))

解读：classify 把异常压成四个标签，fatal 一票否决重试——第 2 节分工表的代码化，与第 8 篇在批处理结果里"请求非法与瞬时错误分开"是同一思路的同步版。外层只兜 4 轮且带随机抖动，与 SDK 内层 2 次叠乘后最坏 12 次请求，第 2 节提醒过的乘法在这里显式取舍：批跑器要的是熬过限流窗口，不是无限重试。429 时还可以读 retry-after 头替换固定退避，留给练习。再看并发调度与落盘：

import json

async def main(tasks):
    sem = asyncio.Semaphore(4)           # 并发上限压在限流档位之下
    results = await asyncio.gather(
        *(run_one(t, sem) for t in tasks), return_exceptions=True)
    failed = [r for r in results if not (isinstance(r, dict) and r["ok"])]
    with open("batch_errors.jsonl", "w", encoding="utf-8") as f:
        for r in failed:
            if isinstance(r, dict):
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print(f"共 {len(tasks)} 条：成功 {len(tasks) - len(failed)}，"
          f"失败 {len(failed)} 已落盘")

if __name__ == "__main__":
    tasks = [{"id": f"t{i}", "params": {
        "model": "claude-sonnet-5-5", "max_tokens": 1024,
        "messages": [{"role": "user", "content": f"处理第 {i} 条数据"}],
    }} for i in range(500)]
    asyncio.run(main(tasks))

解读：Semaphore(4) 把并发压在限流档位之下，比"炸了再退避"便宜得多；gather 收拢全部结果，失败记录带 id、类别、状态码与 request-id 写进 jsonl——第 5 节的四字段口径原样复用，事后既能重跑 fatal 之外的条目，也能拿 request-id 去找支持。跑完打印一行汇总，"成功多少、失败几何、错在哪类"这份最小报告，就是批任务的可观测起点。

8、总结

本篇把散落十篇的报错知识收进一个体系：失败分五类，第一刀切在"有没有拿到响应"；SDK 默认重试 2 次、只管传输层抖动，4xx 永远要你改请求；异常按状态码精确捕获，except 链从子类写到基类；超时默认 600 秒且会被重试放大；排查靠 request-id 加 message 细读；可观测从一层日志中间件起步。三句话带走：分类先于处置，分工先于重试，落账先于优化。

留三个练习：其一，把第 4 节超时改成 5 秒跑一次长输出，亲眼看一次 APITimeoutError，再换流式对比首字延迟；其二，给第 7 节的 run_one 加上 retry-after 解析，429 时按服务端建议等待；其三，把第 6 节日志改成追加写 jsonl，跑一天批任务后用 pandas 汇总每小时调用量与失败率。对于准备上生产的调用代码，建议今天就把 except 链与日志中间件补上——出事之后再补，补的是复盘，不是止损。下一篇是系列收官：把 12 篇的全部零件组装成一个完整的小项目。