健壮性:重试、模型回退、Token 裁剪与成本追踪
30 秒导读: 一个自主编码 agent 要连续跑几十分钟、几百步,中间会撞上 API 限流、超时、 弱模型调不动工具、上下文塞爆窗口、成本悄悄冲上天花板——任何一个都能让整轮任务白跑。 本章讲 RA.Aid 用来"扛住这些"的五道防线,以及它们如何在 三阶段主线 的每一次 agent 运行外面兜底。
本章不重复讲 数据库 schema 和 后端选择(ReAct vs CIAYN), 只讲"让运行不崩"的那些机制。读完你应该能回答:RA.Aid 在弱模型、 超长上下文、API 抖动、 成本失控这四种压力下,分别靠什么坚持跑完。
1. 这是什么(零基础也能懂)
一句话定义: 这是包在"每一次 agent 运行"外面的一层保命外壳——它不负责让 agent 更聪明,只负责让 agent 别在半路挂掉。
想象一下你让 AI 帮你改一个大项目的代码,它要连续工作 40 分钟。这 40 分钟里几乎必然发生:
- API 抖一下:供应商返回 429(请求太多)、超时、或 500,这一步的调用失败了。
- 模型太弱:你为了省钱用了个小模型,它就是学不会按格式把工具调对,连错好几次。
- 上下文塞爆:聊到第 30 步,历史消息累积到 20 万 token,超过模型窗口,直接报错。
- 钱烧超了:你只想花 5 美元,但 agent 不知道,闷头跑到 50 美元。
它能做什么(五道防线):
| 防线 | 解决的压力 | 一句话做法 |
|---|---|---|
| 重试 + 中断 | API 抖动 / 用户想停 | 指数退避重试(最多 20 次),Ctrl-C 能干净打断 |
| 完成/退出/崩溃标志 | 何时该停、遇到不可重试错误 | 上下文里挂一组布尔量,流式循环每 chunk 检查 |
| 模型回退 | 弱模型调不动工具 | 连错到阈值就换"更会调工具"的模型重试这一步 |
| Token 裁剪 | 上下文超长 | 估算 token、按模型窗口裁掉最老的历史,永远保留头两条 |
| 成本追踪 + 限额 | 花钱失控 | 每次调用记 token/花费,超过 --max-cost 就停或问 |
一句话直觉: 把这层想成飞机的自动配平 + 熔断器——飞行员(模型)负责往哪飞, 这层负责"气流颠簸时别散架、油量见底时报警"。
2. 顶层全景(它大概怎么转)
RA.Aid 的核心入口是 run_agent_with_retry(ra_aid/agent_utils.py:567)。它是一个
最多 20 次的重试循环,循环体里调 _run_agent_stream 真正跑一步 agent,外面用
try/except 把各种异常分门别类地接住。五道防线就挂在这个循环的不同位置上。
怎么读这张图: 从上到下是"一次重试尝试"的生命周期;右侧标出每种异常被哪道防线接住。
run_agent_with_retry (agent_utils.py:567, 最多 20 次)
┌──────────────────────────────────────────────┐
│ 进入 InterruptibleSection + agent_context() │ ← Ctrl-C 可打断 / 挂完成标志
│ │
每轮 │ ① 检查崩溃标志 is_crashed → 直接返回 │
尝试 │ ② 检查停止信号 has_received_stop_signal │
│ ③ _run_agent_stream(agent, msg_list) ────────┼──► 流式跑一步 agent
│ 每个 chunk: check_interrupt() │ (内部按需 Token 裁剪)
│ 检查 is_completed / should_exit │
│ ④ 跑测试命令 execute_test_command │
└───────────────┬────────────────────────────────┘
│ 抛异常?按类型分流:
┌────────────────┼─────────────────┬──────────────────┐
▼ ▼ ▼ ▼
ToolExecutionError API 抖动类 400/BadRequest KeyboardInterrupt
→ 模型回退 → _handle_api_error → 标记崩溃 / AgentInterrupt
(fallback) 指数退避后 continue 返回,不重试 → 向上抛,退出
部件一句话职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
run_agent_with_retry | 20 次重试主循环 + 异常分流 | agent_utils.py:567 |
_run_agent_stream | 流式跑一步、逐 chunk 检查中断/完成 | agent_utils.py:512 |
_handle_api_error | 判定是否限流、指数退避、记轨迹 | agent_utils.py:284 |
InterruptibleSection / check_interrupt | Ctrl-C 打断机制 | agent_utils.py:245、254 |
AgentContext | 完成/退出/崩溃标志的载体 | agent_context.py:14 |
FallbackHandler | 弱模型调不动工具时换模型 | fallback_handler.py:25 |
anthropic_token_limiter | 按模型窗口裁剪历史消息 | anthropic_token_limiter.py |
DefaultCallbackHandler | 记 token/成本、检查限额 | callbacks/default_callback_handler.py:148 |
3. 核心原理(逐个机制,由浅入深)
3.1 重试与中断:让 API 抖动不致命,让用户能喊停
要解决的小问题: LLM API 会随机返回 429/超时/500。这些是暂时的——过几秒再试往往就好了。 但如果一遇错就整轮崩,自主运行根本没法用。同时,用户随时可能想按 Ctrl-C 停下,不能让它卡死。
思路/直觉: 把每一步 agent 运行放进一个重试循环;错误来了先判断"这值不值得重试", 值得就等一会儿再试,等待时间指数增长(第 1 次等 1 秒、第 2 次 2 秒、第 3 次 4 秒……), 这样既给供应商喘息、又不会疯狂空转。
原理演示:
# 示意,非源码:指数退避重试的骨架
base_delay = 1
for attempt in range(max_retries): # 最多 20 次
try:
run_one_agent_step() # 真正干活
return "done"
except TransientAPIError as e: # 只对"暂时性"错误退避
if attempt == max_retries - 1:
raise # 最后一次还错,才真失败
delay = base_delay * (2 ** attempt) # 1, 2, 4, 8...
sleep_but_stay_interruptible(delay) # 等待期间仍能被 Ctrl-C 打断
# 重点看:退避时间随尝试次数翻倍,且等待是"可打断"的
真实实现:
主循环在 run_agent_with_retry(agent_utils.py:567),max_retries = 20、base_delay = 1。
它把异常分成几类接住(agent_utils.py:648-691):
ToolExecutionError→ 交给模型回退(见 3.3),然后continue重试。- 一批 API 异常(
InternalServerError、APITimeoutError、各家RateLimitError、ResourceExhausted、ServiceUnavailableError……)→ 交给_handle_api_error退避。 - HTTP 400 / BadRequest → 视为不可重试,标记 agent 崩溃直接返回(
agent_utils.py:681-689), 因为请求本身就是坏的,重试只会一直坏。
退避的核心在 _handle_api_error(agent_utils.py:284)。它先用一长串规则判断"这是不是限流错误"
——检查异常类型、status_code == 429、以及错误文本里有没有 "rate limit"/"too many requests"/
"quota exceeded"/"429" 等短语(agent_utils.py:287-336),只为在轨迹和控制台里用不同标题显示。
然后算延迟并等待:
delay = base_delay * (2**attempt) # agent_utils.py:345
...
start = time.monotonic()
while time.monotonic() - start < delay: # agent_utils.py:371-374
check_interrupt() # 等待期间反复检查中断
time.sleep(0.1)
关键细节/坑:
- 退避等待是"可打断"的:它不是
time.sleep(delay)一睡到底,而是切成 0.1 秒的小片、 每片都check_interrupt()(agent_utils.py:371-374),所以退避期间按 Ctrl-C 也能立刻响应。 - 中断机制 靠三件套:
_setup_interrupt_handling(agent_utils.py:260)在主线程把SIGINT接管到_request_interrupt(agent_utils.py:233);后者把"当前上下文"记进全局;check_interrupt(agent_utils.py:254)一旦发现当前上下文被标记就抛AgentInterrupt。 进出用InterruptibleSection(agent_utils.py:245)上下文管理器压栈/出栈,finally里_restore_interrupt_handling恢复原信号处理器(agent_utils.py:692-693)。 reset_agent_completion_flags(agent_utils.py:273)在需要时清掉上一轮的完成标志,避免 串味(它转调agent_context.reset_completion_flags)。
3.2 完成 / 退出 / 崩溃标志:agent 怎么知道"该停了"
要解决的小问题: 流式循环会一个 chunk 一个 chunk 地吐输出。可"任务已完成""用户要退出" "agent 崩了"这些信号是在别处产生的(工具里、信号处理器里、异常里)——循环怎么及时知道并停下?
思路/直觉: 在当前agent 上下文上挂一组布尔量当"信号灯",谁想让 agent 停就点亮对应的灯;
流式循环每收到一个 chunk 就扫一眼这些灯,亮了就退出。上下文用 contextvars 存,天然按线程/任务 隔离。
图示(信号灯):
AgentContext (agent_context.py:14) ← 挂在 contextvar 上,进 with 块时创建
┌─────────────────────────────┐
│ task_completed / plan_completed → is_completed "任务/计划做完了"
│ agent_should_exit → should_exit "用户/客户端要退"
│ agent_has_crashed + message → is_crashed "撞上不可重试错误"
└─────────────────────────────┘
▲ ▲
工具调 mark_task_completed _run_agent_stream 每 chunk 检查
信号/客户端 mark_should_exit 主循环开头检查 is_crashed
400 错误 mark_agent_crashed
真实实现: _run_agent_stream(agent_utils.py:512)的内层 for chunk in agent.stream(...)
里,每个 chunk 都 check_interrupt()、打印输出,然后:
if is_completed() or should_exit(session_id): # agent_utils.py:538
reset_completion_flags()
return True
标志本身定义在 AgentContext:mark_task_completed/mark_plan_completed(agent_context.py:38、47)、
mark_should_exit(agent_context.py:63,还支持沿父上下文向上传播 N 层)、mark_agent_crashed
(agent_context.py:86)。
关键细节/坑:
- 崩溃不传播、退出可传播:
mark_agent_crashed只标记当前上下文(agent_context.py:89注释明说), 而mark_should_exit可按propagation_depth向父上下文传递——因为"退出"往往是整棵子 agent 树都该停,"崩溃"则是局部的。 - 两种后端在这里分叉:
_run_agent_stream结尾,若是 ReAct(CompiledGraph)就靠agent.get_state().next判断要不要继续下一段(agent_utils.py:554-562);若是 CIAYN 则改看has_received_stop_signal(agent_utils.py:545-552)。后端差异见 第 2 章。
3.3 模型回退:弱模型调不动工具,就换个更会调的
要解决的小问题: 便宜/小模型的通病是调不好工具——参数格式错、JSON 不合法、 函数名拼错,同一个工具连错好几次。整轮任务不该因为这一步就崩。
思路/直觉: 给失败计数;同一个工具连续失败到阈值(默认 3 次),就临时把这一步 "外 包"给一串专门擅长调工具的模型:逐个试,谁先成功就用谁的结果继续。这串候选来自一份 工具调用能力排行榜。
图示(回退流水线):
工具调用失败 → handle_failure() 计数 +1
│
│ 同名工具连续失败 >= max_failures(默认3)?
▼ 是
attempt_fallback():遍历候选模型 (fallback_handler.py:147)
候选① → 绑定这个工具、强制/提示它调 → with_retry(3) → 成功?→ 用结果,返回
候选② → ... 否 ↓
候选③ → ...
│ 全部失败
▼
抛 FallbackToolExecutionError
候选模型从哪来: _load_fallback_tool_models(fallback_handler.py:57)遍历
supported_top_tool_models(tool_leaderboard.py:520),按你有没有配对应供应商的 API key
(validate_provider_env)过滤,取前 FALLBACK_TOOL_MODEL_LIMIT(默认 5)个。
这份排行榜是从伯克利 Gorilla 工具调用榜(BFCL)抽的数据,按 overall_acc 排序
(tool_leaderboard.py:1-5 注释),所以"更靠前 = 更会调工具"。
真实实现: FallbackHandler.handle_failure(fallback_handler.py:102)累加
tool_failure_consecutive_failures,达到 max_failures 且有候选就调 attempt_fallback。
每个候选走 invoke_fallback(fallback_handler.py:286):
simple_model = initialize_llm(fallback_model["provider"], fallback_model["model"])
bound_model = self._bind_tool_model(simple_model, fallback_model) # fc 强制调 / prompt 提示调
retry_model = bound_model.with_retry(stop_after_attempt=RETRY_FALLBACK_COUNT) # 每个候选自身再重试 3 次
response = retry_model.invoke(self.construct_prompt_msg_list())
_bind_tool_model(fallback_handler.py:274)按候选的 type 决定手法:"fc" 用
bind_tools(..., tool_choice=名字) 强制只能调这个工具;否则用 prompt 方式引导它调
(fallback_handler.py:275-283)。
关键细节/坑:
- 默认关闭:回退是实验特性,
init_fallback_handler(agent_utils.py:389)只有在配置experimental_fallback_handler为真、且是 ReAct agent 时才建 handler,否则返回None。 - 换了工具名就重置:
_reset_on_new_failure(fallback_handler.py:222)发现当前失败的工具名 和上次不同,就清零计数——阈值针对的是"同一个工具连错",不是"总错次数"。 - 两条后端接线:ReAct 侧由
_handle_fallback_response(agent_utils.py:416)把回退结果转成HumanMessage塞回消息列表;CIAYN 侧在自己的stream里调handle_fallback_response(ciayn_agent.py:784、调用点ciayn_agent.py:1119)。成功一步就reset_fallback_handler清零。