【Claude】服务器返回 5xx 错误时的通用处理与重试策略 bug报错已解决关键词: Claude Code、5xx 错误、HTTP 500、HTTP 502、HTTP 503、HTTP 504、HTTP 529、服务器错误、通用重试、指数退避、错误分类、状态码、服务降级、故障转移、断路器、重试策略、优雅降级一、问题描述在使用 Claude API 或 Claude Code 时服务器可能会返回各种 5xx 错误。这些错误表示服务端出现了问题与客户端的请求内容无关。当这些错误反复出现时需要一个通用的处理框架来应对。常见的 5xx 错误包括状态码含义描述推荐处理500Internal Server Error服务端内部错误短延迟后重试502Bad Gateway网关错误通常上游服务不可用稍等后重试503Service Unavailable服务暂时不可用等待后重试504Gateway Timeout网关超时重试529Overloaded服务器过载Anthropic 自定义等待后重试具体表现API 调用随机返回 500/502/503/504/529错误不是每次都出现而是间歇性出现同一请求重试后可能成功在 Claude Code 中表现为工具调用失败或响应中断日志中混杂多种 5xx 错误码二、根因分析2.1 5xx 错误的分类5xx 错误可以分为两类类型描述重试成功率瞬时错误服务端临时波动通常几秒后恢复高80%持续错误服务端持续故障需要较长时间恢复低30%2.2 不同 5xx 的触发原因错误码常见原因典型持续时间500代码 bug、数据异常、内部状态损坏数分钟到数小时502负载均衡器无法连接上游服务数秒到数分钟503服务维护、容量不足、主动限流数秒到数小时504上游服务响应超时数秒到数分钟529GPU 集群过载、推理队列满数秒到数分钟2.3 为什么需要通用处理不同 5xx 错误的应对策略有相似之处都需要重试都需要退避等待都需要设置最大重试次数都需要记录和监控实现一个通用框架可以统一处理所有 5xx 错误避免为每个错误码单独写处理逻辑。三、实际操练3.1 分类收集 5xx 错误import anthropic from collections import Counter, defaultdict client anthropic.Anthropic(api_keyyour-api-key) class ErrorCollector: 收集和分析 5xx 错误 def __init__(self): self.errors defaultdict(Counter) def record(self, status_code, model, timestampNone): from datetime import datetime if timestamp is None: timestamp datetime.now() hour timestamp.strftime(%H) self.errors[status_code][hour] 1 self.errors[status_code][total] 1 def report(self): print( 5xx 错误分布 ) for code in sorted(self.errors.keys()): print(f\nHTTP {code}:) print(f 总计: {self.errors[code][total]}) # 按小时分布 hours {k: v for k, v in self.errors[code].items() if k ! total} for hour, count in sorted(hours.items()): print(f {hour}:00 - {count} 次) # 使用示例 collector ErrorCollector() for i in range(100): try: response client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens100, messages[{role: user, content: f测试 {i}}] ) except anthropic.APIStatusError as e: if e.status_code 500: collector.record(e.status_code, sonnet) print(f遇到 {e.status_code}: {e.message}) collector.report()3.2 测试不同 5xx 的重试效果import time def test_retry_effectiveness(error_code, max_retries5): 测试特定 5xx 错误的重试效果 success_count 0 total_attempts 0 for _ in range(20): # 测试 20 次 for attempt in range(max_retries): total_attempts 1 # 模拟重试实际应用中这里是真实的 API 调用 # 如果成功则 break # 这里用随机数模拟 import random if random.random() 0.7: # 70% 重试成功 success_count 1 break time.sleep(1) success_rate success_count / 20 avg_attempts total_attempts / 20 print(f错误码 {error_code}: 成功率 {success_rate:.1%}, 平均尝试次数 {avg_attempts:.1f}) return success_rate # 测试 test_retry_effectiveness(500) test_retry_effectiveness(529)3.3 检查服务状态import urllib.request def check_service_status(): 检查 Anthropic 服务状态 # 检查 API 端点是否可达 try: req urllib.request.Request( https://api.anthropic.com/v1/health, methodGET, headers{anthropic-version: 2023-06-01} ) response urllib.request.urlopen(req, timeout5) print(f服务状态: {response.status}) return True except urllib.error.HTTPError as e: print(fHTTP 错误: {e.code}) return False except Exception as e: print(f连接失败: {e}) return False check_service_status()四、解决方案4.1 方案一统一重试装饰器实现一个通用的重试装饰器处理所有 5xx 错误import time import random from functools import wraps import anthropic class RetryConfig: 重试配置 def __init__(self, max_retries5, base_delay1.0, max_delay60.0, retry_on(500, 502, 503, 504, 529), exponential_base2.0): self.max_retries max_retries self.base_delay base_delay self.max_delay max_delay self.retry_on retry_on self.exponential_base exponential_base def retry_on_5xx(configNone): 重试装饰器处理所有 5xx 错误 if config is None: config RetryConfig() def decorator(func): wraps(func) def wrapper(*args, **kwargs): last_error None for attempt in range(config.max_retries 1): try: return func(*args, **kwargs) except anthropic.APIStatusError as e: if e.status_code in config.retry_on: if attempt config.max_retries: raise # 计算退避时间 delay min( config.base_delay * (config.exponential_base ** attempt), config.max_delay ) # 添加抖动10%-30% jitter random.uniform(0.1, 0.3) * delay total_delay delay jitter print(fHTTP {e.status_code}等待 {total_delay:.1f}s 后重试 f({attempt1}/{config.max_retries})) time.sleep(total_delay) last_error e else: # 不在重试列表中的错误直接抛出 raise except Exception as e: # 非 API 错误如网络错误也重试 if attempt config.max_retries: raise delay config.base_delay * (config.exponential_base ** attempt) print(f网络错误: {e}等待 {delay:.1f}s 后重试) time.sleep(delay) raise last_error or Exception(Max retries exceeded) return wrapper return decorator # 使用 retry_on_5xx(RetryConfig(max_retries5, base_delay2.0, max_delay30.0)) def call_claude_api(client, messages, modelclaude-3-5-sonnet-20241022): return client.messages.create( modelmodel, max_tokens1000, messagesmessages ) # 调用 response call_claude_api(client, [{role: user, content: Hello}])4.2 方案二断路器模式当 5xx 错误持续出现时暂时停止请求避免加剧服务端压力import time from enum import Enum class CircuitState(Enum): CLOSED closed # 正常允许请求 OPEN open # 断开拒绝请求 HALF_OPEN half_open # 半开测试性允许 class CircuitBreaker: 断路器防止持续请求失败的服务 def __init__(self, failure_threshold5, recovery_timeout60, half_open_max_calls3): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.half_open_max_calls half_open_max_calls self.state CircuitState.CLOSED self.failure_count 0 self.last_failure_time None self.half_open_calls 0 def can_execute(self): 判断当前是否允许执行请求 if self.state CircuitState.CLOSED: return True elif self.state CircuitState.OPEN: # 检查是否过了恢复时间 if time.time() - self.last_failure_time self.recovery_timeout: self.state CircuitState.HALF_OPEN self.half_open_calls 0 print(断路器进入半开状态尝试恢复) return True return False elif self.state CircuitState.HALF_OPEN: # 半开状态只允许有限次数 if self.half_open_calls self.half_open_max_calls: self.half_open_calls 1 return True return False def record_success(self): 记录成功 self.failure_count 0 if self.state CircuitState.HALF_OPEN: self.state CircuitState.CLOSED self.half_open_calls 0 print(断路器关闭服务恢复正常) def record_failure(self): 记录失败 self.failure_count 1 self.last_failure_time time.time() if self.state CircuitState.HALF_OPEN: # 半开状态失败重新断开 self.state CircuitState.OPEN print(断路器重新打开服务仍不可用) elif self.failure_count self.failure_threshold: self.state CircuitState.OPEN print(f断路器打开连续失败 {self.failure_count} 次) class ResilientClient: 带断路器和重试的客户端 def __init__(self, client): self.client client self.breaker CircuitBreaker(failure_threshold5, recovery_timeout60) def create_message(self, messages, modelclaude-3-5-sonnet-20241022, max_tokens1000): if not self.breaker.can_execute(): raise Exception(服务暂时不可用请稍后重试) try: response self.client.messages.create( modelmodel, max_tokensmax_tokens, messagesmessages ) self.breaker.record_success() return response except anthropic.APIStatusError as e: if e.status_code 500: self.breaker.record_failure() raise except Exception as e: self.breaker.record_failure() raise # 使用 resilient ResilientClient(client) try: response resilient.create_message([{role: user, content: 测试}]) except Exception as e: print(f请求失败: {e})4.3 方案三降级与故障转移当主服务持续返回 5xx 时切换到备用方案class FallbackStrategy: 故障降级策略 def __init__(self, client): self.client client self.fallback_models [ claude-3-5-sonnet-20241022, claude-3-5-haiku-20241022, claude-3-haiku-20240307 ] def create_with_fallback(self, messages, preferred_modelclaude-3-5-sonnet-20241022, max_tokens1000): 主模型失败时降级到备用模型 models [preferred_model] [m for m in self.fallback_models if m ! preferred_model] for model in models: try: response self.client.messages.create( modelmodel, max_tokensmax_tokens, messagesmessages ) print(f使用模型: {model}) return response except anthropic.APIStatusError as e: if e.status_code 500: print(f模型 {model} 返回 {e.status_code}尝试下一个) continue raise raise Exception(所有模型均不可用) def create_with_cache_fallback(self, messages, cache): 当 API 不可用时返回缓存结果 cache_key str(messages) try: response self.client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens1000, messagesmessages ) # 缓存成功结果 cache[cache_key] response.content[0].text return response except anthropic.APIStatusError as e: if e.status_code 500 and cache_key in cache: print(fAPI 不可用返回缓存结果) return type(obj, (object,), {content: [{text: cache[cache_key]}]}) raise # 使用 fallback FallbackStrategy(client) response fallback.create_with_fallback( messages[{role: user, content: 分析代码}] )4.4 方案四批量请求的重试对于批量请求需要特殊处理部分失败的情况class BatchRetryHandler: 批量请求重试处理器 def __init__(self, client, max_retries3): self.client client self.max_retries max_retries def process_batch(self, requests, modelclaude-3-5-sonnet-20241022): 处理批量请求自动重试失败项 results [None] * len(requests) failed_indices list(range(len(requests))) for attempt in range(self.max_retries 1): if not failed_indices: break new_failed [] for idx in failed_indices: try: response self.client.messages.create( modelmodel, max_tokensrequests[idx].get(max_tokens, 1000), messagesrequests[idx][messages] ) results[idx] response except anthropic.APIStatusError as e: if e.status_code 500 and attempt self.max_retries: new_failed.append(idx) else: results[idx] e failed_indices new_failed if failed_indices and attempt self.max_retries: wait 2 ** attempt random.uniform(0, 1) print(f批量请求: {len(failed_indices)} 个失败等待 {wait:.1f}s 后重试) time.sleep(wait) return results # 使用 handler BatchRetryHandler(client) requests [ {messages: [{role: user, content: f任务 {i}}]} for i in range(10) ] results handler.process_batch(requests)五、验证测试5.1 验证重试装饰器# 测试重试逻辑 retry_on_5xx(RetryConfig(max_retries3, base_delay1.0)) def test_retry_func(): import random if random.random() 0.5: raise anthropic.APIStatusError( Test 500, responsetype(obj, (object,), {status_code: 500})(), bodyNone ) return Success try: result test_retry_func() print(f成功: {result}) except Exception as e: print(f最终失败: {e})5.2 验证断路器# 测试断路器状态转换 breaker CircuitBreaker(failure_threshold3, recovery_timeout5) # 模拟失败 for i in range(5): if breaker.can_execute(): print(f第 {i1} 次: 执行请求) breaker.record_failure() else: print(f第 {i1} 次: 断路器阻止) # 等待恢复 time.sleep(6) if breaker.can_execute(): print(恢复后: 允许执行) breaker.record_success()5.3 回归测试清单检查项操作预期结果重试装饰器触发 500指数退避后重试成功断路器连续失败达到阈值后断开故障转移主模型失败降级到备用模型批量重试部分失败失败项自动重试缓存回退API 不可用返回缓存结果六、最佳实践速查表实践优先级描述统一重试高所有 5xx 使用同一重试逻辑指数退避高2^attempt 抖动最大重试高设置上限建议 5 次断路器中连续失败时暂时断开故障降级中切换到备用模型或缓存批量重试低批量任务只重试失败项监控告警高5xx 频率超过阈值时告警日志记录高记录所有 5xx 错误和时间七、综合框架企业级错误处理class EnterpriseErrorHandler: 企业级 Claude API 错误处理框架 def __init__(self, client, configNone): self.client client self.config config or { retry: {max_retries: 5, base_delay: 2.0, max_delay: 60.0}, circuit_breaker: {threshold: 5, timeout: 60}, fallback: {enabled: True, models: [sonnet, haiku]}, cache: {enabled: True, ttl: 3600} } self.breaker CircuitBreaker(**self.config[circuit_breaker]) self.cache {} def call(self, messages, modelclaude-3-5-sonnet-20241022, max_tokens1000, use_cacheFalse): 统一调用入口 # 检查缓存 if use_cache and self.config[cache][enabled]: cache_key self._cache_key(messages, model) if cache_key in self.cache: return self.cache[cache_key] # 检查断路器 if not self.breaker.can_execute(): raise Exception(服务暂不可用请稍后重试) # 执行请求带重试 retry_config RetryConfig(**self.config[retry]) retry_on_5xx(retry_config) def _do_call(): return self.client.messages.create( modelmodel, max_tokensmax_tokens, messagesmessages ) try: response _do_call() self.breaker.record_success() # 缓存结果 if use_cache: self.cache[cache_key] response return response except anthropic.APIStatusError as e: if e.status_code 500: self.breaker.record_failure() # 尝试故障降级 if self.config[fallback][enabled]: return self._fallback(messages, max_tokens) raise def _cache_key(self, messages, model): return f{model}:{hash(str(messages))} def _fallback(self, messages, max_tokens): 故障降级 fallback_models [ claude-3-5-sonnet-20241022, claude-3-5-haiku-20241022 ] for m in fallback_models: try: return self.client.messages.create( modelm, max_tokensmax_tokens, messagesmessages ) except: continue raise Exception(故障降级失败) # 使用 handler EnterpriseErrorHandler(client) response handler.call( messages[{role: user, content: 分析代码}], use_cacheTrue )八、总结服务器返回 5xx 错误时的通用处理策略统一重试所有 5xx 使用指数退避重试不要区分错误码断路器连续失败超过阈值时暂停请求避免雪崩故障降级主模型不可用时降级到备用模型缓存回退API 完全不可用时返回缓存结果批量重试批量任务只重试失败项避免全部重试对于生产系统建议使用 EnterpriseErrorHandler 这样的综合框架将重试、断路器、降级和缓存集成在一起。这样无论遇到什么 5xx 错误系统都能优雅地处理而不是直接崩溃。记住5xx 错误是服务端问题客户端能做的只有优雅地重试和降级。不要试图修复服务端而是确保你的应用在服务端不稳定时仍能正常运行。
【Claude】服务器返回 5xx 错误时的通用处理与重试策略 bug报错已解决
【Claude】服务器返回 5xx 错误时的通用处理与重试策略 bug报错已解决关键词: Claude Code、5xx 错误、HTTP 500、HTTP 502、HTTP 503、HTTP 504、HTTP 529、服务器错误、通用重试、指数退避、错误分类、状态码、服务降级、故障转移、断路器、重试策略、优雅降级一、问题描述在使用 Claude API 或 Claude Code 时服务器可能会返回各种 5xx 错误。这些错误表示服务端出现了问题与客户端的请求内容无关。当这些错误反复出现时需要一个通用的处理框架来应对。常见的 5xx 错误包括状态码含义描述推荐处理500Internal Server Error服务端内部错误短延迟后重试502Bad Gateway网关错误通常上游服务不可用稍等后重试503Service Unavailable服务暂时不可用等待后重试504Gateway Timeout网关超时重试529Overloaded服务器过载Anthropic 自定义等待后重试具体表现API 调用随机返回 500/502/503/504/529错误不是每次都出现而是间歇性出现同一请求重试后可能成功在 Claude Code 中表现为工具调用失败或响应中断日志中混杂多种 5xx 错误码二、根因分析2.1 5xx 错误的分类5xx 错误可以分为两类类型描述重试成功率瞬时错误服务端临时波动通常几秒后恢复高80%持续错误服务端持续故障需要较长时间恢复低30%2.2 不同 5xx 的触发原因错误码常见原因典型持续时间500代码 bug、数据异常、内部状态损坏数分钟到数小时502负载均衡器无法连接上游服务数秒到数分钟503服务维护、容量不足、主动限流数秒到数小时504上游服务响应超时数秒到数分钟529GPU 集群过载、推理队列满数秒到数分钟2.3 为什么需要通用处理不同 5xx 错误的应对策略有相似之处都需要重试都需要退避等待都需要设置最大重试次数都需要记录和监控实现一个通用框架可以统一处理所有 5xx 错误避免为每个错误码单独写处理逻辑。三、实际操练3.1 分类收集 5xx 错误import anthropic from collections import Counter, defaultdict client anthropic.Anthropic(api_keyyour-api-key) class ErrorCollector: 收集和分析 5xx 错误 def __init__(self): self.errors defaultdict(Counter) def record(self, status_code, model, timestampNone): from datetime import datetime if timestamp is None: timestamp datetime.now() hour timestamp.strftime(%H) self.errors[status_code][hour] 1 self.errors[status_code][total] 1 def report(self): print( 5xx 错误分布 ) for code in sorted(self.errors.keys()): print(f\nHTTP {code}:) print(f 总计: {self.errors[code][total]}) # 按小时分布 hours {k: v for k, v in self.errors[code].items() if k ! total} for hour, count in sorted(hours.items()): print(f {hour}:00 - {count} 次) # 使用示例 collector ErrorCollector() for i in range(100): try: response client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens100, messages[{role: user, content: f测试 {i}}] ) except anthropic.APIStatusError as e: if e.status_code 500: collector.record(e.status_code, sonnet) print(f遇到 {e.status_code}: {e.message}) collector.report()3.2 测试不同 5xx 的重试效果import time def test_retry_effectiveness(error_code, max_retries5): 测试特定 5xx 错误的重试效果 success_count 0 total_attempts 0 for _ in range(20): # 测试 20 次 for attempt in range(max_retries): total_attempts 1 # 模拟重试实际应用中这里是真实的 API 调用 # 如果成功则 break # 这里用随机数模拟 import random if random.random() 0.7: # 70% 重试成功 success_count 1 break time.sleep(1) success_rate success_count / 20 avg_attempts total_attempts / 20 print(f错误码 {error_code}: 成功率 {success_rate:.1%}, 平均尝试次数 {avg_attempts:.1f}) return success_rate # 测试 test_retry_effectiveness(500) test_retry_effectiveness(529)3.3 检查服务状态import urllib.request def check_service_status(): 检查 Anthropic 服务状态 # 检查 API 端点是否可达 try: req urllib.request.Request( https://api.anthropic.com/v1/health, methodGET, headers{anthropic-version: 2023-06-01} ) response urllib.request.urlopen(req, timeout5) print(f服务状态: {response.status}) return True except urllib.error.HTTPError as e: print(fHTTP 错误: {e.code}) return False except Exception as e: print(f连接失败: {e}) return False check_service_status()四、解决方案4.1 方案一统一重试装饰器实现一个通用的重试装饰器处理所有 5xx 错误import time import random from functools import wraps import anthropic class RetryConfig: 重试配置 def __init__(self, max_retries5, base_delay1.0, max_delay60.0, retry_on(500, 502, 503, 504, 529), exponential_base2.0): self.max_retries max_retries self.base_delay base_delay self.max_delay max_delay self.retry_on retry_on self.exponential_base exponential_base def retry_on_5xx(configNone): 重试装饰器处理所有 5xx 错误 if config is None: config RetryConfig() def decorator(func): wraps(func) def wrapper(*args, **kwargs): last_error None for attempt in range(config.max_retries 1): try: return func(*args, **kwargs) except anthropic.APIStatusError as e: if e.status_code in config.retry_on: if attempt config.max_retries: raise # 计算退避时间 delay min( config.base_delay * (config.exponential_base ** attempt), config.max_delay ) # 添加抖动10%-30% jitter random.uniform(0.1, 0.3) * delay total_delay delay jitter print(fHTTP {e.status_code}等待 {total_delay:.1f}s 后重试 f({attempt1}/{config.max_retries})) time.sleep(total_delay) last_error e else: # 不在重试列表中的错误直接抛出 raise except Exception as e: # 非 API 错误如网络错误也重试 if attempt config.max_retries: raise delay config.base_delay * (config.exponential_base ** attempt) print(f网络错误: {e}等待 {delay:.1f}s 后重试) time.sleep(delay) raise last_error or Exception(Max retries exceeded) return wrapper return decorator # 使用 retry_on_5xx(RetryConfig(max_retries5, base_delay2.0, max_delay30.0)) def call_claude_api(client, messages, modelclaude-3-5-sonnet-20241022): return client.messages.create( modelmodel, max_tokens1000, messagesmessages ) # 调用 response call_claude_api(client, [{role: user, content: Hello}])4.2 方案二断路器模式当 5xx 错误持续出现时暂时停止请求避免加剧服务端压力import time from enum import Enum class CircuitState(Enum): CLOSED closed # 正常允许请求 OPEN open # 断开拒绝请求 HALF_OPEN half_open # 半开测试性允许 class CircuitBreaker: 断路器防止持续请求失败的服务 def __init__(self, failure_threshold5, recovery_timeout60, half_open_max_calls3): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.half_open_max_calls half_open_max_calls self.state CircuitState.CLOSED self.failure_count 0 self.last_failure_time None self.half_open_calls 0 def can_execute(self): 判断当前是否允许执行请求 if self.state CircuitState.CLOSED: return True elif self.state CircuitState.OPEN: # 检查是否过了恢复时间 if time.time() - self.last_failure_time self.recovery_timeout: self.state CircuitState.HALF_OPEN self.half_open_calls 0 print(断路器进入半开状态尝试恢复) return True return False elif self.state CircuitState.HALF_OPEN: # 半开状态只允许有限次数 if self.half_open_calls self.half_open_max_calls: self.half_open_calls 1 return True return False def record_success(self): 记录成功 self.failure_count 0 if self.state CircuitState.HALF_OPEN: self.state CircuitState.CLOSED self.half_open_calls 0 print(断路器关闭服务恢复正常) def record_failure(self): 记录失败 self.failure_count 1 self.last_failure_time time.time() if self.state CircuitState.HALF_OPEN: # 半开状态失败重新断开 self.state CircuitState.OPEN print(断路器重新打开服务仍不可用) elif self.failure_count self.failure_threshold: self.state CircuitState.OPEN print(f断路器打开连续失败 {self.failure_count} 次) class ResilientClient: 带断路器和重试的客户端 def __init__(self, client): self.client client self.breaker CircuitBreaker(failure_threshold5, recovery_timeout60) def create_message(self, messages, modelclaude-3-5-sonnet-20241022, max_tokens1000): if not self.breaker.can_execute(): raise Exception(服务暂时不可用请稍后重试) try: response self.client.messages.create( modelmodel, max_tokensmax_tokens, messagesmessages ) self.breaker.record_success() return response except anthropic.APIStatusError as e: if e.status_code 500: self.breaker.record_failure() raise except Exception as e: self.breaker.record_failure() raise # 使用 resilient ResilientClient(client) try: response resilient.create_message([{role: user, content: 测试}]) except Exception as e: print(f请求失败: {e})4.3 方案三降级与故障转移当主服务持续返回 5xx 时切换到备用方案class FallbackStrategy: 故障降级策略 def __init__(self, client): self.client client self.fallback_models [ claude-3-5-sonnet-20241022, claude-3-5-haiku-20241022, claude-3-haiku-20240307 ] def create_with_fallback(self, messages, preferred_modelclaude-3-5-sonnet-20241022, max_tokens1000): 主模型失败时降级到备用模型 models [preferred_model] [m for m in self.fallback_models if m ! preferred_model] for model in models: try: response self.client.messages.create( modelmodel, max_tokensmax_tokens, messagesmessages ) print(f使用模型: {model}) return response except anthropic.APIStatusError as e: if e.status_code 500: print(f模型 {model} 返回 {e.status_code}尝试下一个) continue raise raise Exception(所有模型均不可用) def create_with_cache_fallback(self, messages, cache): 当 API 不可用时返回缓存结果 cache_key str(messages) try: response self.client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens1000, messagesmessages ) # 缓存成功结果 cache[cache_key] response.content[0].text return response except anthropic.APIStatusError as e: if e.status_code 500 and cache_key in cache: print(fAPI 不可用返回缓存结果) return type(obj, (object,), {content: [{text: cache[cache_key]}]}) raise # 使用 fallback FallbackStrategy(client) response fallback.create_with_fallback( messages[{role: user, content: 分析代码}] )4.4 方案四批量请求的重试对于批量请求需要特殊处理部分失败的情况class BatchRetryHandler: 批量请求重试处理器 def __init__(self, client, max_retries3): self.client client self.max_retries max_retries def process_batch(self, requests, modelclaude-3-5-sonnet-20241022): 处理批量请求自动重试失败项 results [None] * len(requests) failed_indices list(range(len(requests))) for attempt in range(self.max_retries 1): if not failed_indices: break new_failed [] for idx in failed_indices: try: response self.client.messages.create( modelmodel, max_tokensrequests[idx].get(max_tokens, 1000), messagesrequests[idx][messages] ) results[idx] response except anthropic.APIStatusError as e: if e.status_code 500 and attempt self.max_retries: new_failed.append(idx) else: results[idx] e failed_indices new_failed if failed_indices and attempt self.max_retries: wait 2 ** attempt random.uniform(0, 1) print(f批量请求: {len(failed_indices)} 个失败等待 {wait:.1f}s 后重试) time.sleep(wait) return results # 使用 handler BatchRetryHandler(client) requests [ {messages: [{role: user, content: f任务 {i}}]} for i in range(10) ] results handler.process_batch(requests)五、验证测试5.1 验证重试装饰器# 测试重试逻辑 retry_on_5xx(RetryConfig(max_retries3, base_delay1.0)) def test_retry_func(): import random if random.random() 0.5: raise anthropic.APIStatusError( Test 500, responsetype(obj, (object,), {status_code: 500})(), bodyNone ) return Success try: result test_retry_func() print(f成功: {result}) except Exception as e: print(f最终失败: {e})5.2 验证断路器# 测试断路器状态转换 breaker CircuitBreaker(failure_threshold3, recovery_timeout5) # 模拟失败 for i in range(5): if breaker.can_execute(): print(f第 {i1} 次: 执行请求) breaker.record_failure() else: print(f第 {i1} 次: 断路器阻止) # 等待恢复 time.sleep(6) if breaker.can_execute(): print(恢复后: 允许执行) breaker.record_success()5.3 回归测试清单检查项操作预期结果重试装饰器触发 500指数退避后重试成功断路器连续失败达到阈值后断开故障转移主模型失败降级到备用模型批量重试部分失败失败项自动重试缓存回退API 不可用返回缓存结果六、最佳实践速查表实践优先级描述统一重试高所有 5xx 使用同一重试逻辑指数退避高2^attempt 抖动最大重试高设置上限建议 5 次断路器中连续失败时暂时断开故障降级中切换到备用模型或缓存批量重试低批量任务只重试失败项监控告警高5xx 频率超过阈值时告警日志记录高记录所有 5xx 错误和时间七、综合框架企业级错误处理class EnterpriseErrorHandler: 企业级 Claude API 错误处理框架 def __init__(self, client, configNone): self.client client self.config config or { retry: {max_retries: 5, base_delay: 2.0, max_delay: 60.0}, circuit_breaker: {threshold: 5, timeout: 60}, fallback: {enabled: True, models: [sonnet, haiku]}, cache: {enabled: True, ttl: 3600} } self.breaker CircuitBreaker(**self.config[circuit_breaker]) self.cache {} def call(self, messages, modelclaude-3-5-sonnet-20241022, max_tokens1000, use_cacheFalse): 统一调用入口 # 检查缓存 if use_cache and self.config[cache][enabled]: cache_key self._cache_key(messages, model) if cache_key in self.cache: return self.cache[cache_key] # 检查断路器 if not self.breaker.can_execute(): raise Exception(服务暂不可用请稍后重试) # 执行请求带重试 retry_config RetryConfig(**self.config[retry]) retry_on_5xx(retry_config) def _do_call(): return self.client.messages.create( modelmodel, max_tokensmax_tokens, messagesmessages ) try: response _do_call() self.breaker.record_success() # 缓存结果 if use_cache: self.cache[cache_key] response return response except anthropic.APIStatusError as e: if e.status_code 500: self.breaker.record_failure() # 尝试故障降级 if self.config[fallback][enabled]: return self._fallback(messages, max_tokens) raise def _cache_key(self, messages, model): return f{model}:{hash(str(messages))} def _fallback(self, messages, max_tokens): 故障降级 fallback_models [ claude-3-5-sonnet-20241022, claude-3-5-haiku-20241022 ] for m in fallback_models: try: return self.client.messages.create( modelm, max_tokensmax_tokens, messagesmessages ) except: continue raise Exception(故障降级失败) # 使用 handler EnterpriseErrorHandler(client) response handler.call( messages[{role: user, content: 分析代码}], use_cacheTrue )八、总结服务器返回 5xx 错误时的通用处理策略统一重试所有 5xx 使用指数退避重试不要区分错误码断路器连续失败超过阈值时暂停请求避免雪崩故障降级主模型不可用时降级到备用模型缓存回退API 完全不可用时返回缓存结果批量重试批量任务只重试失败项避免全部重试对于生产系统建议使用 EnterpriseErrorHandler 这样的综合框架将重试、断路器、降级和缓存集成在一起。这样无论遇到什么 5xx 错误系统都能优雅地处理而不是直接崩溃。记住5xx 错误是服务端问题客户端能做的只有优雅地重试和降级。不要试图修复服务端而是确保你的应用在服务端不稳定时仍能正常运行。
