【Bug已解决】Codex capacity errors 自动重试与意图保留解决方案-育师

【Bug已解决】Codex Desktop: capacity errors should preserve intent and auto-retry instead of handing model routing back to the user 解决方案

原始报错：Codex Desktop: capacity errors should preserve intent and auto-retry instead of handing model routing back to the user 场景：向模型服务发请求时遇到容量错误（服务端繁忙 / 限流 / 容量不足），客户端不做任何重试，直接把"请求失败，请选择另一个模型或重试"的锅甩给用户，让用户手动去点。用户本来只是想完成一次任务，却被中途打断去当"调度员"。关键词：容量错误、自动重试、意图保留、故障转移、退避策略、不暴露给用户。

一、现象长什么样

用户发出一条消息，后台返回容量类错误（典型是 HTTP 429 或 503，或业务层capacity_error）。此时客户端的行为是：

弹出一个对话框："模型当前不可用，请选择其他模型或稍后重试"；
把请求中断，把"接下来怎么办"的决策丢回给用户；
用户要么手动换模型，要么手动点重试，要么放弃。

问题在于：容量错误通常是瞬时且可恢复的。服务端只是那一秒满了，过几秒就有空位。如果客户端在后台悄悄重试几次，绝大多数请求本可以成功，用户根本不会察觉。把决策甩给用户，等于把一个"应该用代码自动处理"的瞬时故障，硬变成了一个"需要人类介入"的产品断点。

二、背景：容量错误到底是什么

容量错误和"请求本身有错"是两回事：

请求错误（400 / 认证失败 / 参数非法）：重试也没用，必须让用户改请求。
容量错误（429 限流 / 503 不可用 / 业务层 capacity）：服务端暂时处理不了，但请求本身没问题，稍后重试大概率成功。

正确的设计是：对容量错误，客户端应在不惊动用户的前提下自动重试（可能换一个等价端点或同一端点退避重试），保留用户最初的完整意图（用哪个模型、发什么内容、哪些参数）；只有当重试彻底耗尽仍失败，才把问题交给用户。

三、根因：错误分类缺失 + 意图未保留

把问题拆开，常见根因：

没有错误分类：客户端把所有非 2xx 都当成"用户错误"弹窗，不区分容量错误和可恢复错误。
意图未保留：每次重试都重新构造请求体，或者把"当前选中的模型"当作可变状态在弹窗里被改掉，重试时用的已经不是用户最初要的模型。
重试暴露给用户：重试逻辑要么没有，要么把"重试按钮"交给用户点，没有后台自动重试。
无限/无退避重试：少数实现有重试但没退避、没上限，把服务端打得更满，反而加重容量问题。

下面用最小模型复现"没有重试、直接抛给用户"的错误写法，再给正确实现。

四、最小可运行复现

下面模拟一个请求函数：有 60% 概率返回容量错误。错误写法是不重试、直接把错误抛给调用方（用户）：

import random class CapacityError(Exception): pass def call_model(body: dict) -> dict: # 模拟：70% 概率容量不足 if random.random() < 0.7: raise CapacityError("capacity_error: model busy") return {"ok": True, "echo": body["input"]} def handle_user_send(body: dict): try: return call_model(body) except CapacityError: # 错误写法：直接把锅甩给用户，让他去选模型/重试 return {"need_user_action": True, "message": "请选择其他模型或重试"} if __name__ == "__main__": body = {"model": "gpt-x", "input": "总结这段日志"} result = handle_user_send(body) print(result) # 大概率输出 {'need_user_action': True, ...} —— 用户被迫介入

运行几次会发现，明明请求没问题，却频繁要求用户手动处理。这正是因为容量错误被当成终态了。

五、方案：带指数退避的自动重试

第一层修复：遇到容量错误，后台自动重试，用指数退避避免踩踏服务端：

import random import time class CapacityError(Exception): pass def call_model(body: dict) -> dict: if random.random() < 0.7: raise CapacityError("capacity_error: model busy") return {"ok": True, "echo": body["input"]} def send_with_retry(body: dict, max_retries: int = 5, base_delay: float = 0.5) -> dict: last_err = None for attempt in range(max_retries): try: return call_model(body) except CapacityError as e: last_err = e if attempt == max_retries - 1: break # 指数退避 + 一点抖动，避免所有客户端同时重试 delay = base_delay * (2 ** attempt) jitter = random.uniform(0, delay * 0.3) time.sleep(delay + jitter) # 重试耗尽才抛出 raise last_err if __name__ == "__main__": body = {"model": "gpt-x", "input": "总结这段日志"} try: print(send_with_retry(body)) except CapacityError: print({"need_user_action": True, "message": "服务持续繁忙，请稍后再试"})

现在同样的 70% 失败率，经过几次退避重试后，绝大多数请求能自动成功，用户无感知。只有在连续多次都失败时，才降级为"请稍后再试"。

六、方案：保留意图——重试用同一份请求体

第二层修复：重试时绝不重新构造请求体，原样复用用户最初提交的body，保证"模型、输入、参数"一模一样。如果请求体里含随机成分（如某些采样种子），要把种子固化为请求的一部分，避免重试两次结果不同造成困惑：

import copy def send_preserving_intent(body: dict, max_retries: int = 5, base_delay: float = 0.5) -> dict: intent = copy.deepcopy(body) # 冻结用户最初意图 last_err = None for attempt in range(max_retries): try: # 每次都用同一份 intent，绝不在重试里改模型/输入 return call_model(intent) except CapacityError as e: last_err = e if attempt == max_retries - 1: break time.sleep(base_delay * (2 ** attempt)) raise last_err

关键点是intent在循环外冻结。即便用户在等待期间切换了下拉框里的"当前模型"，重试用的依旧是他点击发送那一刻的模型——这正是"保留意图"的含义。

七、方案：故障转移也不暴露给用户

第三层修复：如果同一模型多端点，或同一能力有多个等价模型，可以在后台做故障转移（failover），同样对用户透明。故障转移要满足：转移到的目标能力等价、请求体不变、次数有限：

def send_with_failover(body: dict, endpoints: list, max_retries: int = 5): intent = copy.deepcopy(body) last_err = None for attempt in range(max_retries): ep = endpoints[attempt % len(endpoints)] # 轮询等价端点 try: return call_model_on(intent, ep) except CapacityError as e: last_err = e if attempt == max_retries - 1: break time.sleep(0.5 * (2 ** attempt)) raise last_err def call_model_on(body: dict, endpoint: str) -> dict: if random.random() < 0.7: raise CapacityError(f"capacity_error @ {endpoint}") return {"ok": True, "endpoint": endpoint, "echo": body["input"]} if __name__ == "__main__": body = {"model": "gpt-x", "input": "总结这段日志"} eps = ["ep-a", "ep-b", "ep-c"] try: print(send_with_failover(body, eps)) except CapacityError: print({"need_user_action": True, "message": "所有等价端点暂不可用"})

用户全程只发了一次请求，至于中途重试了几遍、换过哪个端点，他都看不到——这正是"不把路由决策交回用户"的目标。

八、验证：把重试与意图保留锁进测试

import itertools def test_retry_recovers_from_transient_capacity(): # 前两次失败，第三次成功 states = {"n": 0} def flaky(body): states["n"] += 1 if states["n"] <= 2: raise CapacityError("busy") return {"ok": True} out = send_with_retry_wrapped(flaky, {"input": "x"}) assert out == {"ok": True} assert states["n"] == 3 def test_intent_preserved_across_retries(): seen = [] def recorder(body): seen.append(body["model"]) if len(seen) < 3: raise CapacityError("busy") return {"ok": True} send_with_retry_wrapped(recorder, {"model": "gpt-x", "input": "y"}) assert seen == ["gpt-x", "gpt-x", "gpt-x"] # 模型始终不变 def send_with_retry_wrapped(fn, body, max_retries=5, base_delay=0): last = None for i in range(max_retries): try: return fn(body) except CapacityError as e: last = e if i == max_retries - 1: break time.sleep(base_delay) raise last if __name__ == "__main__": test_retry_recovers_from_transient_capacity() test_intent_preserved_across_retries() print("容量错误重试与意图保留测试通过。")