#!/usr/bin/env python3 """带超时预算 + 熔断 + 降级 + 人工升级的 Claude 调用编排模拟器(无需 API key)。 模拟一个生产编排器如何处置官方错误分类(platform.claude.com/docs/en/api/errors): - 429 rate_limit_error(带 retry-after): 等待后重试 - 429 rate_limit_error(无 retry-after, enforced_spend_limit_reached): 重试无效 -> 升级告警 + 降级 - 500 api_error / 529 overloaded_error: 指数退避 + 抖动后重试 - 400/401/403: 确定性错误, 不重试 -> 升级 - 超时: 消耗超时预算 熔断器: closed -> open(连续失败达阈值) -> half-open(冷却后单探测) -> closed/再 open 降级: 主模型失败且预算耗尽 -> 切换 fallback 模型并标记 degraded 升级: 降级也失败或不可恢复错误 -> 进入人工升级队列(带结构化上下文) 运行: python3 01_resilient_call_orchestrator.py """ from __future__ import annotations import random from dataclasses import dataclass, field # 模拟时钟(毫秒), 避免真实 sleep CLOCK = {"now_ms": 0.0} def advance(ms: float) -> None: CLOCK["now_ms"] += ms # ---------------------------------------------------------------- 错误模型 class ApiCallError(Exception): """模拟 Claude API 错误响应。kind 对应官方 error type 语义。""" def __init__(self, status: int, kind: str, retry_after_ms: int | None = None, detail_code: str | None = None): super().__init__(f"HTTP {status} {kind}" + (f" retry_after={retry_after_ms}ms" if retry_after_ms else "") + (f" detail={detail_code}" if detail_code else "")) self.status = status self.kind = kind self.retry_after_ms = retry_after_ms self.detail_code = detail_code class SimulatedUpstream: """脚本化上游: 按顺序抛出预先编排的故障, 最后(或按需)成功。 latency_ms: 每次调用的模拟耗时; 超过 per-attempt 超时则算超时失败。 """ def __init__(self, script: list[ApiCallError | None], latency_ms: float = 400.0): self.script = list(script) self.latency_ms = latency_ms self.calls = 0 def call(self) -> dict: self.calls += 1 advance(self.latency_ms) if self.script: step = self.script.pop(0) if step is not None: raise step return {"content": "primary answer", "model": "primary"} class SimulatedFallback: """降级模型: 模拟稳定可用。""" def __init__(self, latency_ms: float = 300.0, fail: bool = False): self.latency_ms = latency_ms self.fail = fail def call(self) -> dict: advance(self.latency_ms) if self.fail: raise ApiCallError(529, "overloaded_error") return {"content": "cached-template answer (degraded)", "model": "fallback"} # ---------------------------------------------------------------- 超时预算 @dataclass class TimeoutBudget: """请求级总时限, 向下分配给每次尝试与退避等待。""" total_ms: float start_ms: float = field(default_factory=lambda: CLOCK["now_ms"]) def remaining_ms(self) -> float: return self.total_ms - (CLOCK["now_ms"] - self.start_ms) def spend(self, ms: float) -> bool: """尝试消耗 ms; 返回 False 表示预算已不够。""" if ms > self.remaining_ms(): advance(self.remaining_ms()) return False advance(ms) return True # ---------------------------------------------------------------- 熔断器 class CircuitBreaker: """closed -> open(连续失败>=threshold) -> half-open(cooldown 后) -> closed/再 open。""" def __init__(self, failure_threshold: int = 3, cooldown_ms: float = 10_000.0): self.failure_threshold = failure_threshold self.cooldown_ms = cooldown_ms self.state = "closed" self.failures = 0 self.opened_at_ms: float | None = None def allow(self) -> tuple[bool, str]: if self.state == "closed": return True, "closed" if self.state == "open": elapsed = CLOCK["now_ms"] - (self.opened_at_ms or 0) if elapsed >= self.cooldown_ms - 1e-6: # 容忍浮点误差 self.state = "half-open" return True, "half-open(probe)" return False, f"open(fast-fail, 还需冷却 {self.cooldown_ms - elapsed:.0f}ms)" return True, "half-open(probe)" def record(self, ok: bool) -> None: if ok: self.failures = 0 self.state = "closed" else: self.failures += 1 if self.failures >= self.failure_threshold or self.state == "half-open": self.state = "open" self.opened_at_ms = CLOCK["now_ms"] # ---------------------------------------------------------------- 编排器 @dataclass class Attempt: n: int event: str waited_ms: float = 0.0 spent_ms: float = 0.0 class Orchestrator: def __init__(self, breaker: CircuitBreaker, max_retries: int = 2, backoff_base_ms: float = 500.0, per_attempt_ms: float = 3_000.0, rng: random.Random | None = None): self.breaker = breaker self.max_retries = max_retries # 与官方 SDK 默认一致: 2 次重试 self.backoff_base_ms = backoff_base_ms self.per_attempt_ms = per_attempt_ms self.rng = rng or random.Random(42) # -- 错误分类: 哪些可重试, 怎么等 ------------------------------------ def classify(self, err: ApiCallError) -> str: if err.status == 429: if err.detail_code == "enforced_spend_limit_reached": return "spend_cap" # 无 retry-after, 重试无效 return "rate_limit" # 带 retry-after if err.status in (500, 502, 503, 529): return "transient" # 指数退避重试 if err.status == 504: return "timeout" # 重试或改 streaming return "permanent" # 400/401/403/404 def backoff_ms(self, attempt: int, retry_after_ms: int | None) -> float: if retry_after_ms is not None: return float(retry_after_ms) # 尊重 retry-after jitter = self.rng.uniform(0, self.backoff_base_ms * 0.25) return self.backoff_base_ms * (2 ** attempt) + jitter # -- 主入口 ---------------------------------------------------------- def handle(self, primary: SimulatedUpstream, fallback: SimulatedFallback, budget: TimeoutBudget, log: list[Attempt]) -> dict: allowed, breaker_state = self.breaker.allow() if not allowed: log.append(Attempt(0, f"circuit {breaker_state} -> 直接降级")) return self._fallback(fallback, log) attempt = 0 while attempt <= self.max_retries: if budget.remaining_ms() <= 0: log.append(Attempt(attempt, "timeout-budget exhausted -> 降级")) self.breaker.record(False) return self._fallback(fallback, log) per = min(self.per_attempt_ms, budget.remaining_ms()) t0 = CLOCK["now_ms"] try: result = primary.call() if per >= primary.latency_ms else (_ for _ in ()).throw( ApiCallError(504, "timeout_error")) self.breaker.record(True) log.append(Attempt(attempt, "primary OK", spent_ms=CLOCK["now_ms"] - t0)) return {**result, "degraded": False} except ApiCallError as err: kind = self.classify(err) if kind in ("spend_cap", "permanent"): # 确定性/预算性错误: 重试无意义 -> 升级 + 降级 log.append(Attempt(attempt, f"{err} -> 不重试, escalate + fallback")) self.breaker.record(False) out = self._fallback(fallback, log) out["escalation"] = ("budget_alert" if kind == "spend_cap" else "config_error") return out wait = self.backoff_ms(attempt, err.retry_after_ms) if not budget.spend(wait): log.append(Attempt(attempt, f"{err}; 退避 {wait:.0f}ms 超预算 -> 降级")) self.breaker.record(False) return self._fallback(fallback, log) log.append(Attempt(attempt, f"{err}; 退避 {wait:.0f}ms 后重试", waited_ms=wait)) attempt += 1 log.append(Attempt(self.max_retries, "重试次数用尽 -> 熔断记失败并降级")) self.breaker.record(False) return self._fallback(fallback, log) def _fallback(self, fb: SimulatedFallback, log: list[Attempt]) -> dict: try: result = fb.call() log.append(Attempt(99, "fallback OK, 响应标记 degraded")) return {**result, "degraded": True} except ApiCallError as err: log.append(Attempt(99, f"fallback 也失败({err}) -> escalation")) return {"content": None, "model": None, "degraded": True, "escalation": "all_paths_failed"} # ---------------------------------------------------------------- 场景演示 def run_scenario(title: str, script: list[ApiCallError | None], fallback_fail: bool = False, budget_ms: float = 12_000.0) -> dict: print(f"\n=== 场景: {title} ===") breaker = CircuitBreaker(failure_threshold=3, cooldown_ms=10_000) orch = Orchestrator(breaker) upstream = SimulatedUpstream(script) fb = SimulatedFallback(fail=fallback_fail) log: list[Attempt] = [] result = orch.handle(upstream, fb, TimeoutBudget(budget_ms), log) for a in log: parts = [f"[#{a.n}] {a.event}"] if a.waited_ms: parts.append(f"等待 {a.waited_ms:.0f}ms") if a.spent_ms: parts.append(f"耗时 {a.spent_ms:.0f}ms") print(" " + "; ".join(parts)) print(f" => 最终: model={result.get('model')} degraded={result.get('degraded')}" f" escalation={result.get('escalation', '-')} " f"上游调用 {upstream.calls} 次, 熔断器={breaker.state}") return result def main() -> None: print("Claude 调用编排模拟器 — 错误处置依据 platform.claude.com/docs/en/api/errors (截至 2026-08-30)") print("SDK 默认自动重试 2 次(指数退避, 尊重 retry-after) — 本模拟器显式化同一策略") # 1) 529 过载两次后成功: 指数退避重试救回 r1 = run_scenario("529 overloaded 两次后成功", [ ApiCallError(529, "overloaded_error"), ApiCallError(529, "overloaded_error"), ]) assert r1["model"] == "primary" and not r1["degraded"] # 2) 429 限流(带 retry-after): 按 retry-after 等待后成功 r2 = run_scenario("429 rate limit (retry-after=600ms)", [ ApiCallError(429, "rate_limit_error", retry_after_ms=600), ]) assert r2["model"] == "primary" and not r2["degraded"] # 3) 429 spend cap(无 retry-after): 重试无效 -> 升级 + 降级 r3 = run_scenario("429 spend cap (enforced_spend_limit_reached)", [ ApiCallError(429, "rate_limit_error", detail_code="enforced_spend_limit_reached"), ]) assert r3["degraded"] and r3["escalation"] == "budget_alert" # 4) 400 确定性错误: 不重试, 升级 + 降级 r4 = run_scenario("400 invalid_request_error", [ ApiCallError(400, "invalid_request_error"), ]) assert r4["degraded"] and r4["escalation"] == "config_error" # 5) 持续 529 + 预算耗尽: 降级链生效 r5 = run_scenario("持续 529 直到超时预算耗尽", [ ApiCallError(529, "overloaded_error")] * 5) assert r5["degraded"] and "escalation" not in r5 # 6) 主路径与降级全灭: 人工升级 r6 = run_scenario("主路径 500 且 fallback 也 529", [ ApiCallError(500, "api_error")] * 3, fallback_fail=True) assert r6["escalation"] == "all_paths_failed" and r6["content"] is None # 7) 熔断器: 连续失败打开后快速失败(直接降级, 不再打上游) print("\n=== 场景: 熔断器状态机 ===") breaker = CircuitBreaker(failure_threshold=2, cooldown_ms=10_000) for i in range(2): breaker.record(False) ok, why = breaker.allow() print(f" 连续 2 次失败后 allow() -> {ok} ({why})") assert not ok and breaker.state == "open" advance(10_000) # 模拟冷却期流逝 ok, why = breaker.allow() print(f" 冷却 10s 后 allow() -> {ok} ({why})") assert ok and breaker.state == "half-open" breaker.record(True) print(f" 探测成功后 state -> {breaker.state}") assert breaker.state == "closed" print("\n全部断言通过: 预算/重试/熔断/降级/升级行为符合官方错误处置建议。") if __name__ == "__main__": main()