实践｜Error Recovery，只重发这一轮模型请求

先让报告失败，看看程序会不会多跑一次

用户要求检查 main.py 与 expected.txt。允许执行时，程序输出 42，结果保存以后，固定模型发送器才在报告请求上注入一次 429。恢复成功应得到核对结论，程序执行次数仍然为 1。思想篇解释了为什么重试范围应包住模型请求。

源码在独立公开仓库 learn-agent-harness，Python3.10+，默认标准库、无需 API Key：

bash

git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s11-error-recovery/user_code/practice.py --decision allow

默认输出中，父逻辑调用为 5，父请求尝试为 6，计划等待 0.25 秒，最终 answered 并报告实际输出 42。子任务仍使用前课的 3 次独立固定判断；它们没有混入父请求次数。文件和获准后的进程真实执行，模型与故障注入是固定替身，等待只记录秒数。

如果改成 --decision deny，运行请求在原权限入口被拒绝，程序不执行，固定模型报告未执行。恢复层没有收到 API 故障，也不会把权限拒绝当作再试一次就能放行的情况。

classify_failure：先决定该交回、等待、调整还是停下

发送器的出口有两种基本数据：正常响应保留 stop_reason 和 content；类型化 SDK 异常由 sdk_send 转为 failure 字段，带 HTTP 状态或连接/超时类型。普通 Python 异常继续暴露，不被适配器一概吞掉。

classify_failure 不依赖异常文字，返回 action 与 reason。完整 end_turn/tool_use 可以交回；max_tokens 只建议扩大预算；明确的临时状态建议等待；400、401、403、413 与未知响应停下。分类尚未扣额度或执行恢复，所以图中的“恢复建议”不代表已发出第二次请求。

真实执行轨迹回放；模型响应为固定队列，等待只记录秒数，不发API请求或执行工具。

Pys11/core/runtime.py · classify_failure

未运行

def classify_failure(outcome):
    """接收适配器的基本数据，不匹配任意SDK异常文字。"""
    if type(outcome) is not dict:
        return {'action': 'stop', 'reason': 'invalid_response'}
    if 'failure' in outcome:
        failure = outcome['failure']
        if type(failure) is not dict:
            return {'action': 'stop', 'reason': 'invalid_failure'}
        status = failure.get('status')
        if type(status) is int and status in (429, 500, 502, 503, 504, 529):
            return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
        if failure.get('transport') in ('connection', 'timeout') and status is None:
            return {'action': 'wait', 'reason': failure['transport']}
        return {'action': 'stop', 'reason': 'request_error:' + str(status)}
    reason = outcome.get('stop_reason')
    if reason == 'max_tokens':
        return {'action': 'expand', 'reason': 'max_tokens'}
    if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
        return {'action': 'accept', 'reason': reason}
    return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
import math
import random
import time
scenario='ready'

class RecoveryStopped(RuntimeError):
    """恢复未取得完整响应；由旧循环作为model_error保留已有历史。"""

def checked_state(state):
    if type(state) is not dict or set(state) != {'remaining', 'waited', 'events'}:
        raise ValueError('恢复状态必须由宿主持有。')
    if type(state['remaining']) is not int or not 0 <= state['remaining'] <= 20 or type(state['waited']) not in (int, float) or not math.isfinite(state['waited']) or not 0 <= state['waited'] <= 3 or type(state['events']) is not list or len(state['events']) > 20:
        raise ValueError('请求或等待额度无效。')

def classify_failure(outcome):
    """接收适配器的基本数据，不匹配任意SDK异常文字。"""
    if type(outcome) is not dict:
        return {'action': 'stop', 'reason': 'invalid_response'}
    if 'failure' in outcome:
        failure = outcome['failure']
        if type(failure) is not dict:
            return {'action': 'stop', 'reason': 'invalid_failure'}
        status = failure.get('status')
        if type(status) is int and status in (429, 500, 502, 503, 504, 529):
            return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
        if failure.get('transport') in ('connection', 'timeout') and status is None:
            return {'action': 'wait', 'reason': failure['transport']}
        return {'action': 'stop', 'reason': 'request_error:' + str(status)}
    reason = outcome.get('stop_reason')
    if reason == 'max_tokens':
        return {'action': 'expand', 'reason': 'max_tokens'}
    if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
        return {'action': 'accept', 'reason': reason}
    return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}

def retry_delay(attempt, retry_after=None, jitter=0.0):
    """优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
    if type(attempt) is not int or not 0 <= attempt < 4:
        raise ValueError('尝试序号必须在0到3之间。')
    if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
        raise ValueError('抖动比例必须在0到1之间。')
    seconds = None
    if type(retry_after) in (str, int, float):
        try:
            value = float(retry_after)
            if math.isfinite(value) and value >= 0:
                seconds = value
        except ValueError:
            pass
    if seconds is not None:
        return seconds
    base = min(0.25 * 2 ** attempt, 2.0)
    return base + base * 0.25 * jitter

ready={'stop_reason':'end_turn','content':[{'type':'text','text':'已有本次执行证据，输出为42。'}]}
partial={'stop_reason':'max_tokens','content':[{'type':'tool_use','id':'partial','name':'run_main','input':{}}]}
request={'messages':[{'role':'user','content':'检查main.py与expected.txt。'}],'max_tokens':1024}
state={'remaining':20,'waited':0.0,'events':[]}
sent=[]
waits=[]
outcome={'ready':ready,'transient':{'failure':{'status':529}},'truncated':partial,'invalid_request':{'failure':{'status':400}},'denied':{'failure':{'status':403}},'unknown':{'stop_reason':'new_reason','content':[]}}.get(scenario,ready)
attempt=1
retry_after={'header':'0.7','invalid':'not-a-number','long':'9'}.get(scenario)
jitter=2 if scenario=='bad_jitter' else 0.5
queue=[{'failure':{'status':529}},ready]
if scenario=='truncated':queue=[partial,ready]
if scenario=='twice':queue=[partial,partial]
if scenario=='fatal':queue=[{'failure':{'status':401}}]
if scenario=='exhausted':queue=[{'failure':{'status':529}}]*4
if scenario=='long_wait':queue=[{'failure':{'status':429,'retry_after':'9'}}]
if scenario=='budget':state['remaining']=0
if scenario=='mutation':queue=[{'failure':{'status':529}},ready]
def send(payload):
    sent.append(deepcopy(payload))
    if scenario=='mutation':payload['messages'].clear()
    return deepcopy(queue.pop(0))
before=deepcopy(request)


# 调用入口
try:
    observed=classify_failure(outcome)
    print(observed)
except (ValueError,RecoveryStopped) as error:
    observed={"error":str(error)}
    print(str(error))

分类完整响应、临时失败、输出截断与必须停止的原因。
分类只建议动作，不执行恢复。

选“输出截断”，能看到候选里放了一个 run_main；分类结果仍是 expand，工具没有运行。选“请求有误”与“API权限不足”，则交出停止原因。400 可能对应多种请求问题，413 指请求字节过大，本例没有把它们自动改写成“删历史以后重发”。

上面显示的是固定公开提交中完整函数正文。点击行号设置断点，再点调试，使用继续或下一步推进；三组图都跟随同一个代码轨迹游标，接口替身与调用入口可展开查看。

retry_delay：计算一个秒数，不负责等过去

输入是从 0 开始的尝试序号、可选 Retry-After 和 0 到 1 的抖动比例。先校验序号与比例，再尝试解析有限非负秒数；有效值直接返回，包括 0。日期格式或无效值回退到指数退避。这个教学解析器只支持秒数格式。

退避公式是 min(0.25 × 2^attempt, 2) 加上基础值的 0 到 25%。这里使用调用者给出的随机比例，便于测试固定结果；正式默认随机源为 random.random。等待动作由第三个函数执行，计算图不会把返回秒数画成真实经过的时间。

真实执行轨迹回放；模型响应为固定队列，等待只记录秒数，不发API请求或执行工具。

Pys11/core/runtime.py · retry_delay

未运行

def retry_delay(attempt, retry_after=None, jitter=0.0):
    """优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
    if type(attempt) is not int or not 0 <= attempt < 4:
        raise ValueError('尝试序号必须在0到3之间。')
    if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
        raise ValueError('抖动比例必须在0到1之间。')
    seconds = None
    if type(retry_after) in (str, int, float):
        try:
            value = float(retry_after)
            if math.isfinite(value) and value >= 0:
                seconds = value
        except ValueError:
            pass
    if seconds is not None:
        return seconds
    base = min(0.25 * 2 ** attempt, 2.0)
    return base + base * 0.25 * jitter

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
import math
import random
import time
scenario='backoff'

class RecoveryStopped(RuntimeError):
    """恢复未取得完整响应；由旧循环作为model_error保留已有历史。"""

def checked_state(state):
    if type(state) is not dict or set(state) != {'remaining', 'waited', 'events'}:
        raise ValueError('恢复状态必须由宿主持有。')
    if type(state['remaining']) is not int or not 0 <= state['remaining'] <= 20 or type(state['waited']) not in (int, float) or not math.isfinite(state['waited']) or not 0 <= state['waited'] <= 3 or type(state['events']) is not list or len(state['events']) > 20:
        raise ValueError('请求或等待额度无效。')

def classify_failure(outcome):
    """接收适配器的基本数据，不匹配任意SDK异常文字。"""
    if type(outcome) is not dict:
        return {'action': 'stop', 'reason': 'invalid_response'}
    if 'failure' in outcome:
        failure = outcome['failure']
        if type(failure) is not dict:
            return {'action': 'stop', 'reason': 'invalid_failure'}
        status = failure.get('status')
        if type(status) is int and status in (429, 500, 502, 503, 504, 529):
            return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
        if failure.get('transport') in ('connection', 'timeout') and status is None:
            return {'action': 'wait', 'reason': failure['transport']}
        return {'action': 'stop', 'reason': 'request_error:' + str(status)}
    reason = outcome.get('stop_reason')
    if reason == 'max_tokens':
        return {'action': 'expand', 'reason': 'max_tokens'}
    if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
        return {'action': 'accept', 'reason': reason}
    return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}

def retry_delay(attempt, retry_after=None, jitter=0.0):
    """优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
    if type(attempt) is not int or not 0 <= attempt < 4:
        raise ValueError('尝试序号必须在0到3之间。')
    if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
        raise ValueError('抖动比例必须在0到1之间。')
    seconds = None
    if type(retry_after) in (str, int, float):
        try:
            value = float(retry_after)
            if math.isfinite(value) and value >= 0:
                seconds = value
        except ValueError:
            pass
    if seconds is not None:
        return seconds
    base = min(0.25 * 2 ** attempt, 2.0)
    return base + base * 0.25 * jitter

ready={'stop_reason':'end_turn','content':[{'type':'text','text':'已有本次执行证据，输出为42。'}]}
partial={'stop_reason':'max_tokens','content':[{'type':'tool_use','id':'partial','name':'run_main','input':{}}]}
request={'messages':[{'role':'user','content':'检查main.py与expected.txt。'}],'max_tokens':1024}
state={'remaining':20,'waited':0.0,'events':[]}
sent=[]
waits=[]
outcome={'ready':ready,'transient':{'failure':{'status':529}},'truncated':partial,'invalid_request':{'failure':{'status':400}},'denied':{'failure':{'status':403}},'unknown':{'stop_reason':'new_reason','content':[]}}.get(scenario,ready)
attempt=1
retry_after={'header':'0.7','invalid':'not-a-number','long':'9'}.get(scenario)
jitter=2 if scenario=='bad_jitter' else 0.5
queue=[{'failure':{'status':529}},ready]
if scenario=='truncated':queue=[partial,ready]
if scenario=='twice':queue=[partial,partial]
if scenario=='fatal':queue=[{'failure':{'status':401}}]
if scenario=='exhausted':queue=[{'failure':{'status':529}}]*4
if scenario=='long_wait':queue=[{'failure':{'status':429,'retry_after':'9'}}]
if scenario=='budget':state['remaining']=0
if scenario=='mutation':queue=[{'failure':{'status':529}},ready]
def send(payload):
    sent.append(deepcopy(payload))
    if scenario=='mutation':payload['messages'].clear()
    return deepcopy(queue.pop(0))
before=deepcopy(request)


# 调用入口
try:
    observed=retry_delay(attempt,retry_after,jitter)
    print(observed)
except (ValueError,RecoveryStopped) as error:
    observed={"error":str(error)}
    print(str(error))

有效服务器秒数优先，无效时退避加抖动。
返回计划秒数，由调用者检查额度。

“服务器秒数”返回 0.7；“无效等待头”按退避计算；“要求等待过长”仍返回 9。后者不是计算出错，而是后续恢复入口需要判断：当前任务只有 3 秒计划等待额度，应该停止，不能私自把服务器值缩短。

无效抖动则明确报错，发送与等待都没有发生。把等待策略拆成纯计算函数，可以直接检查零秒、非有限数、非法头和增长过程，而不必用实际等待来猜公式是否正确。

request_with_recovery：候选响应留在恢复层，完整后才返回

函数接收 send、request 与宿主持有的 state。state 保存 remaining、waited 和 events，父会话的连续逻辑调用共享这一份。请求先深复制，每次交给发送器时再复制，避免 SDK 适配器或测试发送器修改调用者历史。

每次尝试先扣请求额度，再发出请求并分类。accept 才返回响应；stop 直接抛出带原因的 RecoveryStopped。wait 在检查次数与累计秒数以后预留等待额度，再调用 pause；expand 仅把当前副本的 max_tokens 从 1024 调到 2048，第二次截断即停止。

真实执行轨迹回放；模型响应为固定队列，等待只记录秒数，不发API请求或执行工具。

Pys11/core/runtime.py · request_with_recovery

未运行

def request_with_recovery(send, request, state, pause=time.sleep, random_unit=random.random):
    """每次逻辑请求最多4次尝试；父会话共享20次请求/3秒计划等待额度。"""
    checked_state(state)
    pending = deepcopy(request)
    if pending.get('max_tokens') != 1024 or type(pending.get('max_tokens')) is not int:
        raise ValueError('教学输出预算从1024开始。')
    expanded = False
    for attempt in range(1, 5):
        if state['remaining'] == 0:
            raise RecoveryStopped('request_budget')
        state['remaining'] -= 1
        outcome = send(deepcopy(pending))
        decision = classify_failure(outcome)
        state['events'].append({'attempt': attempt, 'action': decision['action'], 'reason': decision['reason']})
        if decision['action'] == 'accept':
            return outcome
        if decision['action'] == 'stop':
            raise RecoveryStopped(decision['reason'])
        if attempt == 4 or state['remaining'] == 0:
            raise RecoveryStopped('request_budget')
        if decision['action'] == 'expand':
            if expanded:
                raise RecoveryStopped('output_still_truncated')
            pending['max_tokens'] = 2048
            expanded = True
            continue
        delay = retry_delay(attempt - 1, decision.get('retry_after'), random_unit())
        if state['waited'] + delay > 3:
            raise RecoveryStopped('wait_budget')
        state['waited'] += delay
        state['events'][-1]['delay'] = delay
        pause(delay)
    raise RecoveryStopped('request_budget')

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
import math
import random
import time
scenario='transient'

class RecoveryStopped(RuntimeError):
    """恢复未取得完整响应；由旧循环作为model_error保留已有历史。"""

def checked_state(state):
    if type(state) is not dict or set(state) != {'remaining', 'waited', 'events'}:
        raise ValueError('恢复状态必须由宿主持有。')
    if type(state['remaining']) is not int or not 0 <= state['remaining'] <= 20 or type(state['waited']) not in (int, float) or not math.isfinite(state['waited']) or not 0 <= state['waited'] <= 3 or type(state['events']) is not list or len(state['events']) > 20:
        raise ValueError('请求或等待额度无效。')

def classify_failure(outcome):
    """接收适配器的基本数据，不匹配任意SDK异常文字。"""
    if type(outcome) is not dict:
        return {'action': 'stop', 'reason': 'invalid_response'}
    if 'failure' in outcome:
        failure = outcome['failure']
        if type(failure) is not dict:
            return {'action': 'stop', 'reason': 'invalid_failure'}
        status = failure.get('status')
        if type(status) is int and status in (429, 500, 502, 503, 504, 529):
            return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
        if failure.get('transport') in ('connection', 'timeout') and status is None:
            return {'action': 'wait', 'reason': failure['transport']}
        return {'action': 'stop', 'reason': 'request_error:' + str(status)}
    reason = outcome.get('stop_reason')
    if reason == 'max_tokens':
        return {'action': 'expand', 'reason': 'max_tokens'}
    if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
        return {'action': 'accept', 'reason': reason}
    return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}

def retry_delay(attempt, retry_after=None, jitter=0.0):
    """优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
    if type(attempt) is not int or not 0 <= attempt < 4:
        raise ValueError('尝试序号必须在0到3之间。')
    if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
        raise ValueError('抖动比例必须在0到1之间。')
    seconds = None
    if type(retry_after) in (str, int, float):
        try:
            value = float(retry_after)
            if math.isfinite(value) and value >= 0:
                seconds = value
        except ValueError:
            pass
    if seconds is not None:
        return seconds
    base = min(0.25 * 2 ** attempt, 2.0)
    return base + base * 0.25 * jitter

ready={'stop_reason':'end_turn','content':[{'type':'text','text':'已有本次执行证据，输出为42。'}]}
partial={'stop_reason':'max_tokens','content':[{'type':'tool_use','id':'partial','name':'run_main','input':{}}]}
request={'messages':[{'role':'user','content':'检查main.py与expected.txt。'}],'max_tokens':1024}
state={'remaining':20,'waited':0.0,'events':[]}
sent=[]
waits=[]
outcome={'ready':ready,'transient':{'failure':{'status':529}},'truncated':partial,'invalid_request':{'failure':{'status':400}},'denied':{'failure':{'status':403}},'unknown':{'stop_reason':'new_reason','content':[]}}.get(scenario,ready)
attempt=1
retry_after={'header':'0.7','invalid':'not-a-number','long':'9'}.get(scenario)
jitter=2 if scenario=='bad_jitter' else 0.5
queue=[{'failure':{'status':529}},ready]
if scenario=='truncated':queue=[partial,ready]
if scenario=='twice':queue=[partial,partial]
if scenario=='fatal':queue=[{'failure':{'status':401}}]
if scenario=='exhausted':queue=[{'failure':{'status':529}}]*4
if scenario=='long_wait':queue=[{'failure':{'status':429,'retry_after':'9'}}]
if scenario=='budget':state['remaining']=0
if scenario=='mutation':queue=[{'failure':{'status':529}},ready]
def send(payload):
    sent.append(deepcopy(payload))
    if scenario=='mutation':payload['messages'].clear()
    return deepcopy(queue.pop(0))
before=deepcopy(request)


# 调用入口
try:
    observed=request_with_recovery(send,request,state,pause=waits.append,random_unit=lambda:0)
    print(observed)
except (ValueError,RecoveryStopped) as error:
    observed={"error":str(error)}
    print(str(error))
assert request==before
assert all(not x.get("content") for x in request["messages"] if x["role"]=="assistant")

发送前扣请求额度，只发送模型请求副本。
完整响应才交回旧循环，半截工具请求不执行。
最多4次尝试，重复截断或额度不足停下，已有历史保留。

在 pending 的输出预算赋值行设断点，停下时仍是 1024，下一步以后才是 2048。原来的 messages 没有加入半截响应，截断内容中的 partial 调用 ID 不会出现在外部历史里。选“再次截断”，可以看到第二次候选最终变为停止原因。

“连续失败”会消耗四次尝试，最后一次失败以后没有额外等待；“等待超过额度”在第一次失败后就停止；“请求额度耗尽”一次也不发送。发送前扣数、等待前预留值，意味着发送异常或等待器中断不会让已经消耗的额度自动返还。

接回原循环时，哪一层还负责什么？

恢复函数中没有工具执行器，也不写 messages。装配层仍沿第十课收集真实状态、组装 system、准备计划与记忆副本；第八课整理仍发生在原逻辑调用之前。当前父请求被包装以后，唯一的旧循环照常负责保存响应和执行合法工具：

python

# s11/user_code/practice.py 的 parent 适配器
request = {
    "model": "teaching-model", "system": system,
    "tools": tool_definitions(tools), "messages": history,
    "max_tokens": 1024,
}
return request_with_recovery(send, request, recovery, pause=wait)

# s10 的装配继续把 parent 交给原有 run_loop；
# parent 返回完整响应后，run_loop 才保存、分发、记录工具结果。

恢复耗尽时异常进入旧循环的 model_error 出口，不回到工具执行阶段。默认故障发生在 run_main 的结果已经保存以后，所以可以验证“工具执行成功、报告请求失败”的状态，不把两种失败混在一句“任务失败”里。

s10 的公开入口仅把自身 core 导入改为按路径绑定，以免逐课目录中的同名 runtime 冲突；公开 run 签名和循环行为未变。子任务与摘要没有接入这一父请求包装器，不能从父恢复日志推断它们也已具备同样策略。

用失败反例检查这个恢复范围

切换故障时，既看最终状态，也看程序是否真的多执行了。标准库验证会将真实 main.py 改成每次运行就在计数文件追加一个 x；恢复后的内容必须恰好为 x。

bash

python3 examples/s11-error-recovery/user_code/practice.py --decision allow --scenario truncated
python3 examples/s11-error-recovery/user_code/practice.py --decision allow --scenario truncated_twice
python3 examples/s11-error-recovery/user_code/practice.py --decision allow --scenario long_wait
python3 examples/s11-error-recovery/user_code/verify_practice.py
python3 examples/s11-error-recovery/user_code/verify_fixtures.py

第一次截断后成功，父尝试为 6；再次截断以 model_error 停止，但已保存的程序结果仍在，程序计数仍为 1。long_wait 也停下，且没有发生那次 9 秒等待。exhausted 在首轮连续过载，四次请求后停止，尚未执行任何工具。失败场景 CLI 退出码为 1。

SDK 验证使用 requirements.txt 固定的 Anthropic1.13.0 与模拟 HTTP，覆盖临时状态、连接失败、截断、认证与过长等待，并核对 400/413 的类型化状态。SDK 设置 max_retries=0，观察到的出站尝试与宿主扣数一致；没有调用在线模型。浏览器则直接执行三个函数正文，模型响应来自可查看的固定队列，等待器只记录秒数。

本课的 4 次尝试、20 次父请求、3 秒计划等待和 1024→2048 都是教学参数。生产项目还要按任务成本、延迟和目标模型配置网络超时、取消及输出上限。若需要自动缩小上下文，应先证明候选请求更小、工具交接仍完整；若需要重跑有副作用的工具，应另外设计幂等与执行确认。先把“再次发送模型请求”这条边界守住，才能判断后来增加的恢复是否可靠。