跳到主要内容

健壮性:重试、模型回退、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_retry20 次重试主循环 + 异常分流agent_utils.py:567
_run_agent_stream流式跑一步、逐 chunk 检查中断/完成agent_utils.py:512
_handle_api_error判定是否限流、指数退避、记轨迹agent_utils.py:284
InterruptibleSection / check_interruptCtrl-C 打断机制agent_utils.py:245254
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 = 20base_delay = 1。 它把异常分成几类接住(agent_utils.py:648-691):

  • ToolExecutionError → 交给模型回退(见 3.3),然后 continue 重试。
  • 一批 API 异常(InternalServerErrorAPITimeoutError、各家 RateLimitErrorResourceExhaustedServiceUnavailableError……)→ 交给 _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:3847)、 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 清零。

3.4 Token 预算与裁剪:上下文塞爆前先瘦身

要解决的小问题: 对话越滚越长,历史消息迟早超过模型上下文窗口,再发就直接报错。 必须在每次调用前把历史裁到窗口以内,同时别把关键的开头(系统提示、原始任务)裁掉。

思路/直觉: 先算出"这个模型的窗口有多大",再把消息历史从最老的开始丢,直到估算 token 落回窗口内;永远保留最前面几条(系统提示 + 原始任务),那是 agent 的"根"。

第一步:窗口多大 —— get_model_token_limit(anthropic_token_limiter.py:332)先问 litellm 的 get_model_infomax_input_tokens;拿不到就退回本地字典 models_params (models_params.py:18,每个模型一条 token_limit,如 Claude 系 200000)按名字查,还带一个 "去掉连字符再查一次"的归一化兜底(anthropic_token_limiter.py:392-399);再查不到就用 DEFAULT_TOKEN_LIMIT = 100000(models_params.py:13)。Claude 3.7 会额外减掉 max_tokens 预留输出空间(adjust_claude_37_token_limit,anthropic_token_limiter.py:306)。

第二步:怎么裁 —— 有两条路,按模型选(build_agent_kwargs,agent_utils.py:106-125):

裁剪器用于怎么估 token怎么裁
state_modifierClaude 3.7 / 4 系litellm 精确 token_counteranthropic_trim_messages,固定留头 2 条
base_state_modifier其它模型CiaynAgent._estimate_tokens 粗估LangChain trim_messages,strategy=last

原理演示(base 版的核心):

# 示意,非源码:留住第一条,只在"剩下的"里按 token 裁
first = messages[0] # 系统提示/原始任务:必留
budget = max_input_tokens - tokens(first) # 给第一条先扣掉预算
kept = trim_from_oldest(messages[1:], budget, strategy="last") # 从最老的往前丢
return [first] + kept
# 重点看:预算先减去"必留的头",再在剩余消息上做裁剪

真实实现: base_state_modifier(anthropic_token_limiter.py:227)正是上面这套:取 first_message、算 first_tokensnew_max_tokens = max_input_tokens - first_tokens, 再对 messages[1:]trim_messages。token 估算用 estimate_messages_tokens (anthropic_token_limiter.py:32),它内部就是 CiaynAgent._estimate_tokens

CIAYN 后端有一套自己的裁剪 _trim_chat_history(ciayn_agent.py:872):先按 max_history_messages(默认 50)砍消息条数,再 while 循环从头 pop(0) 丢最老的,直到 _estimate_tokens 累加 ≤ max_tokens(ciayn_agent.py:896-905)。它的 token 估算简单粗暴:

return len(text.encode("utf-8")) // 2.0 # ciayn_agent.py:927 —— 按 UTF-8 字节数除以 2 粗估

关键细节/坑:

  • 两套估算精度不同:Anthropic 系用 litellm 精确计数(create_token_counter_wrapper, anthropic_token_limiter.py:154,还得把 LangChain 消息转成 litellm 格式),其它模型用 "字节数 // 2"的粗估。粗估便宜但可能偏,所以只在非 Anthropic 路径用。
  • 可整体关掉:配置 limit_tokens=Falsebuild_agent_kwargs 干脆不挂裁剪器 (agent_utils.py:106),适合窗口够大或想自己管的场景。

3.5 推理辅助开关:给不带思维链的模型补一段"想一想"

要解决的小问题: 有些模型自己不会先规划再动手。RA.Aid 可以在规划/实现/研究前先让模型 产出一段"推理"来引导后续动作——但强模型不需要、反而费钱,得能按模型和用户意愿开关。

思路/直觉: 每个模型在 models_params 里带一个默认开关 reasoning_assist_default (models_params.py:3171387);用户可用 CLI 的 --reasoning-assistance / --no-reasoning-assistance 强制覆盖。三者优先级:强制开 > 强制关 > 模型默认

真实实现: 三个阶段的 agent(planning/implementation/research)里是同一段判定逻辑 (agents/planning_agent.py:137-148):

force_assistance = get_config_repository().get("force_reasoning_assistance", False)
disable_assistance = get_config_repository().get("disable_reasoning_assistance", False)
if force_assistance: reasoning_assist_enabled = True
elif disable_assistance: reasoning_assist_enabled = False
else: reasoning_assist_enabled = model_config.get("reasoning_assist_default", False)

开启后用 prompts/reasoning_assist_prompt.py 里对应阶段的模板(REASONING_ASSIST_PROMPT_PLANNING 等,reasoning_assist_prompt.py:3)拼 prompt。两个 CLI 开关在 __main__.py:1299-1301 等处写入配置。

3.6 成本追踪与限额:每次调用都记账,超预算就刹车

要解决的小问题: 自主 agent 会连发几百次 LLM 调用,用户很容易在不知不觉中烧掉远超预期的钱。 需要逐次记账并在触及上限时停下或征询

思路/直觉: 挂一个 LangChain callback,在每次调用结束时从响应里抠出 token 用量、 按单价算这次花费、累加进会话总额;每加一次就检查有没有超过 --max-cost/--max-tokens, 超了就按策略处理。

图示(记账 → 限额三分支):

每次 LLM 调用结束 on_llm_end (default_callback_handler.py:512)
│ 抠 token 用量 _extract_token_usage(兼容 Anthropic/Gemini 等多种格式)

_update_token_counts:算 cost、累加 session_totals、successful_requests++
│ _check_limits() 超限了吗?
├─ 用户之前已同意继续 → 继续
├─ exit_at_limit=True 或用户之前拒绝 → sys.exit(0) 直接退
└─ 否则 → 弹 Confirm 问"要继续吗?" → 记住这次决定

真实实现: DefaultCallbackHandler(callbacks/default_callback_handler.py:148,单例)。 on_llm_end(:512)算出本次耗时后调 _update_token_counts(:397),后者:

  • _get_tiered_cost_rates 拿单价(部分 Gemini 是"20 万 token 以上更贵"的阶梯价, :272),算 input_cost + output_cost,累加进 total_costsession_totals(:437-444)。
  • _check_limits(:616)比对配置里的 max_cost / max_tokens;超了就 _record_limit_reached 记一条轨迹,再按 exit_at_limit 和用户历史决定退出还是 Confirm.ask 弹窗询问(:454-510)。
  • 正常路径调 _handle_callback_update(:528),往数据库写一条 record_type="model_usage" 的 Trajectory(带 current_costinput_tokensoutput_tokensmodel)——这就是 轨迹表 里成本数据的来源。

限额从哪来:store_limit_config(__main__.py:99)把 CLI 的 --max-cost/--max-tokens/ --exit-at-limit 写进配置(__main__.py:106-108),回调再读它们。成本追踪本身可用 track_cost=False 关掉(_initialize_callback_handler_internal,:728)。

这些 Trajectory 记录最终被 ra_aid/server(FastAPI)+ frontend/ 消费,做成运行轨迹与成本的 可视化——本章不展开,只点到为止。


4. 巧妙之处(可借鉴的技术)

  • 退避等待做成"可打断的小片睡眠":while 循环里 0.1 秒一片、每片 check_interrupt() (agent_utils.py:371-374),既实现指数退避又不牺牲 Ctrl-C 响应。比 time.sleep(delay) 好在: 用户不用等完整个退避才停下。
  • 回退候选来自真实排行榜 + API key 可用性双过滤:_load_fallback_tool_models (fallback_handler.py:57)先按 BFCL 分数排序,再按"你到底配了哪些供应商的 key"过滤 (tool_leaderboard.py:520)——保证换上的模型既更会调工具、又真的能用。
  • "必留头部 + 只裁中段"的裁剪不变式:两个 state_modifier 都固定保留前 2 条 (anthropic_token_limiter.py:216244-258),让系统提示和原始任务永不丢失,agent 不会 "裁着裁着忘了自己在干嘛"。
  • 400 错误显式判为不可重试:agent_utils.py:651681 专门把 BadRequest 从"值得重试"里 摘出来标记崩溃——避免对一个本质坏掉的请求空转 20 次。
  • 成本/限额判断在锁内、弹窗在锁外:_update_token_counts 把"是否需要询问用户"的决定放进 _lock 里算好,再到锁外 Confirm.ask(:482-510),避免持锁时阻塞在用户输入上。

5. 边界与局限(诚实)

  • 模型回退默认关,且只对 ReAct agent 生效(agent_utils.py:389-396);CIAYN 走自己的一套。 想用得显式开 experimental_fallback_handler,它仍是实验特性。
  • 非 Anthropic 的 token 估算很粗:"UTF-8 字节数 // 2"(ciayn_agent.py:927)只是近似, 对 CJK/代码可能偏差较大,裁剪边界不精确。
  • 成本单价靠内置表 + litellm:MODEL_COSTS(default_callback_handler.py:33)是硬编码快照, litellm 查不到时才用它;新模型或改价可能落表,查不到时默认按 0 计费(:254-270)—— 即成本可能被低估。
  • max_retries=20base_delay=1 写死run_agent_with_retry(agent_utils.py:576-577), 不可配置;第 20 次仍失败就抛 RuntimeError(agent_utils.py:339-341)。
  • 回退提示把消息都转成 SystemMessage:construct_prompt_msg_list(fallback_handler.py:348) 为规避各家 API 的消息结构校验,把历史统一转 SystemMessage,代码里自己标了 TODO (fallback_handler.py:346)承认这不理想。

6. 代码地图(导航索引)

主题文件路径符号名
重试主循环 + 异常分流ra_aid/agent_utils.pyrun_agent_with_retry
指数退避 / 限流判定ra_aid/agent_utils.py_handle_api_error
流式跑一步 + 逐 chunk 检查ra_aid/agent_utils.py_run_agent_stream
中断三件套ra_aid/agent_utils.pyInterruptibleSection / _request_interrupt / check_interrupt / _setup_interrupt_handling
完成/退出/崩溃标志ra_aid/agent_context.pyAgentContext / mark_should_exit / mark_agent_crashed / is_completed
重置完成标志ra_aid/agent_utils.pyreset_agent_completion_flags
模型回退核心ra_aid/fallback_handler.pyFallbackHandler / handle_failure / attempt_fallback / invoke_fallback
回退候选加载ra_aid/fallback_handler.py_load_fallback_tool_models
回退候选排行榜ra_aid/tool_leaderboard.pysupported_top_tool_models
回退初始化 / ReAct 接线ra_aid/agent_utils.pyinit_fallback_handler / _handle_fallback_response
CIAYN 回退接线ra_aid/agent_backends/ciayn_agent.pyhandle_fallback_response
模型窗口大小ra_aid/anthropic_token_limiter.pyget_model_token_limit / adjust_claude_37_token_limit
历史裁剪(两套)ra_aid/anthropic_token_limiter.pystate_modifier / base_state_modifier / estimate_messages_tokens
CIAYN 自有裁剪 + token 估算ra_aid/agent_backends/ciayn_agent.py_trim_chat_history / _estimate_tokens
各模型窗口/默认表ra_aid/models_params.pymodels_params / DEFAULT_TOKEN_LIMIT
推理辅助开关ra_aid/agents/planning_agent.pyforce_reasoning_assistance / disable_reasoning_assistance / reasoning_assist_default
成本/token 记账 + 限额ra_aid/callbacks/default_callback_handler.pyDefaultCallbackHandler / _update_token_counts / _check_limits
限额配置写入ra_aid/__main__.pystore_limit_config