跳到主要内容

上下文管理与答案收尾

30 秒导读: 一个深度研究 Agent 可能跑几百个 turn、塞进 256K 上下文窗口。本章讲两件事: 一是怎么不爆窗口——裁掉旧的工具结果、在超窗前主动"退一步"收尾;二是怎么拿到一个干净的 答案——把整段失败压成结构化复盘喂回下一次尝试,并反复重试直到抠出符合格式的 \boxed{}

本章聚焦"省 token"和"收尾"这两条暗线。工具本身怎么实现(搜索/抓取/沙箱)见 05-tools-mcp.md;主循环怎么转、子 Agent 怎么调,见 02-orchestrator-loop.md;对模型脏输出的容错修复见 03-robustness-rollback.md


1. 这是什么(零基础也能懂)

问题很朴素: LLM 的上下文窗口是有限的(比如 256K token)。深度研究 Agent 每调一次工具, 就往对话里塞一大段返回结果(一个网页正文可能好几万字)。跑上几十、几百个 turn,历史越堆越长, 迟早塞不下。塞不下会怎样?要么 API 报错,要么把关键信息挤出窗口。

更朴素的第二个问题: 就算你跑完了,最后怎么拿到一个能判分的答案?评测要求答案写成 \boxed{...} 的固定格式。可模型经常"找到答案了却忘了包 \boxed{}",或者干脆跑到一半迷路了。

MiroThinker 用四招应对,由浅入深:

白话解决什么
① 上下文裁剪只留最近 N 个工具结果,旧的换成一句占位话历史太长,省 token
② 窗口守卫每 turn 前估一下 token,快超了就"退一步"提前收尾别等 API 报错才发现爆窗
③ 失败经验重试把整段失败压成"复盘报告",喂回下一次重来一次不行,带着教训再来
④ 答案收尾反复要模型总结,直到抠出干净的 \boxed{}拿到能判分的答案

一句话直觉: 把上下文窗口当一块有限的白板。招①是"擦掉旧草稿只留最近几笔";招②是 "白板快满时先把结论写下来别再往上画";招③是"这块白板报废了,但先抄一张便签带到下一块新白板"; 招④是"结论必须写进那个红框里,写歪了就重写"。


2. 顶层全景(它大概怎么转)

这几招分布在四个层次上,从每次 LLM 调用一直串到评测外层的重试循环。先看一张"谁在哪一层"的图。

外层:评测重试循环 benchmarks/common_benchmark.py::run_single_task
┌──────────────────────────────────────────────────────────────┐
│ while format_retry_count <= max_format_retries: │
│ current_task_description = 原始任务 + 累积的失败经验 │ ← 招③
│ └─► execute_task_pipeline(..., is_final_retry) │
│ │ │
│ 中层:一次完整跑 orchestrator.py::_run_main_agent_loop │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 每个 turn: │ │
│ │ ensure_summary_context() ── 快超窗?退一步收尾 ───┼──┐ │ ← 招②
│ │ create_message() ── 发给 LLM 前裁掉旧工具结果 ────┼┐ │ │ ← 招①
│ │ 循环结束(跑满 turn / 超窗): ││ │ │
│ │ generate_and_finalize_answer() ──收尾决策四象限──┼┼─┼──┼─► 招④
│ └────────────────────────────────────────────────────┘│ │ │
│ │ 拿不到 \boxed{} 且开了压缩 → generate_failure_summary ← 招③素材
│ └──────────────► 返回 failure_experience_summary ──┘ │
└──────────────────────────────────────────────────────────────┘

怎么读这张图: 从里到外三层。最内是"发一次 LLM 请求"(招①在这里裁历史);中层是"一次完整 的解题"(招②守着窗口,循环结束后招④收尾);最外是评测的"多次尝试"(招③把上一次的失败压成 经验,拼进任务描述里重来)。

部件一句话职责:

部件干什么在哪
_remove_tool_result_from_messages发送前裁掉旧工具结果,换占位文本src/llm/base_client.py
ensure_summary_context用 tiktoken 估 token,超窗就 pop 掉最后一对消息src/llm/providers/anthropic_client.py
AnswerGenerator收尾决策 + 失败复盘 + boxed 重试src/core/answer_generator.py
generate_agent_summarize_prompt收尾时那段"严格格式"的总结指令src/utils/prompt_utils.py
run_single_task评测外层的 format 重试循环benchmarks/common_benchmark.py

两个配置开关(在 conf/agent/*.yaml,由 01-config-and-assembly.md 讲的 Hydra 装配注入)决定了所有行为:

开关含义例值
keep_tool_result发给 LLM 时保留最近几个工具结果,-1 表示全留5(见 mirothinker_1.7_keep5_max300.yaml:21)
context_compress_limit压缩/多尝试的开关兼上限,>0 开启、0 关闭5(同文件 :22)

3. 招① 上下文裁剪:只留最近 N 个工具结果

它要解决的小问题

对话历史里,占空间的大头是工具返回结果(网页正文、搜索结果 JSON、代码执行输出)。可模型 真正需要看清楚的,往往只是最近几步的结果——很早以前那次搜索返回的原文,现在多半用不上了。 于是:发给 LLM 前,把老的工具结果替换成一句占位话,骨架保留、内容清空。

思路(注意"发送副本"和"日志原本"分离)

一个关键设计:裁剪只作用于发给 LLM 的副本,原始 message_history 原封不动返回,这样日志里 仍能看到完整对话。_remove_tool_result_from_messages 一上来就 [m.copy() for m in messages] 拷贝一份再动手(src/llm/base_client.py:136)。

三种取值行为(呼应 conf 里的 keep_tool_result):

keep_tool_result行为
-1全留,原样返回,啥也不裁
0只留最初的任务消息,其余工具结果全替换成占位文本
N(>0)留最近 N 个工具结果,更早的替换成占位文本

有个容易忽略的细节:第一条 user 消息是"最初的任务",不是工具结果,永远保留。代码里先收集 所有 user/tool 角色的下标,再把第一条排除在"可裁"之外:

# 示意,非源码:重点看"第一条 user 是任务、不算工具结果"
user_indices = [i for i, m in enumerate(messages) if m["role"] in ("user", "tool")]
first_user_idx = user_indices[0] # 最初的任务,永远保留
tool_result_indices = user_indices[1:] # 从第二条 user 起才是工具结果
keep = tool_result_indices[-N:] # 只留最近 N 个

真实实现里,被裁掉的消息保留结构、只换内容——Anthropic 格式换成一个 text block,OpenAI 格式 换成纯字符串:

# src/llm/base_client.py:207-218 _remove_tool_result_from_messages
if isinstance(msg.get("content"), list):
msg["content"] = [{"type": "text",
"text": "Tool result is omitted to save tokens."}]
else:
msg["content"] = "Tool result is omitted to save tokens."

什么时候裁

每次 LLM 调用都裁。create_messagekeep_tool_result 一路透传下去 (src/llm/base_client.py:222-260),Anthropic 客户端在真正发请求前调用它, 拿到过滤副本 messages_for_llm,再返回未裁的原本:

# src/llm/providers/anthropic_client.py:120-124 _create_message
messages_for_llm = self._remove_tool_result_from_messages(
messages_history, keep_tool_result
)
processed_messages = self._apply_cache_control(messages_for_llm)
# ... 用 processed_messages 发请求,但 return response, messages_history(原本)

关键细节/坑

  • 占位文本本身也要占几个 token,但比起动辄几万字的网页正文,可忽略。
  • 裁剪只减小"发送体积",不改变 Agent 内部保存的历史,所以日志、回滚(03) 看到的都是完整对话。

4. 招② 窗口守卫:超窗前"退一步"提前收尾

它要解决的小问题

招①能省 token,但最近 N 个工具结果本身可能就很大——最后一次工具调用返回了一整篇长文,单它 就快撑爆窗口了。这时不能等 API 报 "context length exceeded",得在发请求前自己估一估,发现快 超了就主动结束循环、去收尾。

思路

ensure_summary_context 每个 turn 结束时算一笔"未来这次收尾请求大概要多少 token",跟 max_context_length 比。它不是精确计,而是用 tiktoken 粗估 + 1.5 倍缓冲 + 固定预留,宁可 保守一点提前收。

估算公式(src/llm/providers/anthropic_client.py:325-332):

estimated_total = 上次 LLM 调用的 input_tokens (last_call_tokens)
+ 上次 LLM 调用的 output_tokens
+ 最后一条 user 消息 tokens (× 1.5 缓冲)
+ summary_prompt tokens (× 1.5 缓冲)
+ max_tokens (给回答预留的空间)
+ 1000 (额外 buffer)
  • last_call_tokens 是上一次真实 API 返回的用量,在 _update_token_usage 里被写入 (anthropic_client.py:89-94,把 input + cache_creation + cache_read 加起来当 input)。
  • token 估算用 _estimate_tokens:优先 tiktoken.get_encoding("o200k_base"),取不到退回 cl100k_base,再不行就按"4 字符 ≈ 1 token"兜底(anthropic_client.py:281-300)。

超窗时怎么"退一步"

一旦 estimated_total >= max_context_length,它pop 掉最后一对 assistant-user 消息(即最后一次 工具调用的请求 + 结果),然后返回 False:

# src/llm/providers/anthropic_client.py:334-355 ensure_summary_context
if estimated_total >= self.max_context_length:
if message_history[-1]["role"] == "user": # 工具结果
message_history.pop()
if message_history[-1]["role"] == "assistant": # 工具调用请求
message_history.pop()
return False, message_history

为什么要 pop 掉这一对? 因为正是这最后一次(往往超大的)工具结果把窗口顶爆了。丢掉它,才腾得出 空间做收尾总结。主循环收到 False 就把 turn_count 直接拉到 max_turnsbreak:

# src/core/orchestrator.py:1124-1135
pass_length_check, message_history = self.llm_client.ensure_summary_context(
message_history, temp_summary_prompt)
if not pass_length_check:
turn_count = max_turns # 视同"跑满 turn"
break

关键细节

  • 循环退出后,reached_max_turns = turn_count >= max_turns(orchestrator.py:1141)——超窗和跑满 turn 在下游被当成同一种"没自然收尾"来处理,一起进招④的决策表。
  • 估算里没有把整段历史重新编码一遍(那样太慢),而是复用"上一次调用的实际用量 + 这次要新增的 部分",是个务实的近似。

5. 招③ 失败经验重试:把整段失败压成 post-mortem

它要解决的小问题

一次尝试失败了(跑满 turn 没找到、或找到了没写成 \boxed{})。整段对话历史又长又乱,不可能原样 带到下一次重来——那还是会爆窗口。怎么办?让模型自己写一份结构化的"事后复盘",把几百 turn 压缩 成几句话,只把这几句话带到下一次。

思路:让模型写结构化 post-mortem

generate_failure_summary 复用当前对话历史(去掉末尾多余的 user),追加两样东西:

  1. FAILURE_SUMMARY_PROMPT —— 一段 user 指令,要求不许调工具,按三段式写复盘。
  2. FAILURE_SUMMARY_ASSISTANT_PREFIX —— 一段预填的 assistant 开头(<think>...),把模型"引导"进 结构化格式。

复盘的三段式(src/utils/prompt_utils.py:44-52):

内容
Failure type四选一:incomplete(跑完 turn 没做完)/ blocked(工具失败或缺信息卡住)/ misdirected(走错方向)/ format_missed(找到答案但忘了 \boxed{})
What happened描述用了什么思路、为什么没到最终答案
Useful findings列出发现的事实、中间结果、结论——给下一次复用

预填 assistant 前缀是个巧妙点:模型续写时会顺着 <think> 里"我们要写一份 post-mortem…"的引导 走,格式更稳(prompt_utils.py:55-63)。生成后代码把前缀补回文本开头,再用 extract_failure_experience_summary 抠出 <think> 后、<use_mcp_tool> 前的正文 (answer_generator.py:230-247;抽取逻辑在 parsing_utils.py:228,</think> 后有正文取正文、 没正文就退回 think 内容)。

外层怎么把经验喂回下一次

评测的 run_single_task 是一个 while format_retry_count <= max_format_retries 循环 (benchmarks/common_benchmark.py:348),max_format_retries 就等于 context_compress_limit (:296)。每次拿到 FORMAT_ERROR_MESSAGEformat_retry_count += 1,把这次的 failure_experience_summary 累加进 failure_experiences,并从原始任务描述重新拼接增强版任务:

# 示意,非源码:每次都从干净的原始任务重拼,而不是在上次基础上叠
current_task_description = task_description
current_task_description += FAILURE_EXPERIENCE_HEADER # 一次:说明这是历史尝试
for idx, exp in enumerate(failure_experiences, 1):
current_task_description += FAILURE_EXPERIENCE_ITEM.format( # 每条:[Attempt N] + 复盘
attempt_number=idx, failure_summary=exp)
current_task_description += FAILURE_EXPERIENCE_FOOTER # 一次:"这次换个策略"

真实拼接见 common_benchmark.py:393-410。三段模板(prompt_utils.py:25-42):

  • FAILURE_EXPERIENCE_HEADER —— === Previous Attempts Analysis ===,开头出现一次。
  • FAILURE_EXPERIENCE_ITEM —— [Attempt {n}] {summary},每条失败一份。
  • FAILURE_EXPERIENCE_FOOTER —— === End of Analysis === + "这次该换个策略"。

为什么每次从原始任务重拼、而不是往上叠? 避免把 header/footer 嵌套累积;累积的只是纯粹的 failure_experiences 列表,拼接时一次性铺开。此外这循环还能断点续跑:format_retry_count > 0 时会从上几次的日志文件里把 failure_experience_summary 捞回来(common_benchmark.py:304-346)。

关键细节

  • retry_with_summary(conf,默认 True)控制是否生成复盘。若关掉(如 mirothinker_1.7_keep5_max300.yaml:23False),重试时就不带经验、裸重来。
  • 生成复盘本身也是一次 LLM 调用,step_idturn_count + 10 错开日志(answer_generator.py:225)。

6. 招④ 答案收尾:四象限决策 + boxed 重试

它要解决的小问题

循环结束了(自然收尾 / 跑满 turn / 超窗)。现在要么生成最终答案、要么直接放弃去写复盘—— 到底走哪条,取决于两个开关:开没开上下文压缩、是不是最后一次重试。

决策表(直接引自源码 docstring)

generate_and_finalize_answer 的 docstring 给了一张四象限表 (src/core/answer_generator.py:481-486):

Context ManagementReached Max Turns行为
OFF (limit=0)No生成答案 → 用中间答案兜底
OFF (limit=0)Yes生成答案 → 用中间答案兜底
ON (limit>0)No生成答案 → 兜底,失败则写复盘
ON (limit>0)Yes跳过生成 → 直接写复盘

背后的取舍很清楚:

  • 压缩关闭时,模型只有这一次机会——所以哪怕总结失败,也要拿历史里存过的中间 \boxed{} 答案兜底,最大化命中率(handle_no_context_management_fallback,answer_generator.py:355)。
  • 压缩开启时,还有下一次重试——所以绝不瞎猜:乱兜一个错答案反而拉低准确率,不如老实报 FORMAT_ERROR_MESSAGE,交给招③写复盘、下次带经验重来(handle_context_management_no_fallback, answer_generator.py:409)。
  • 压缩开启 + 跑满 turn + 不是最后一次:连生成都跳过——"任何答案都是盲猜",直接去写复盘 (answer_generator.py:507-528)。
  • 但如果是最后一次重试(is_final_retry):没有下次了,即使跑满 turn 也要搏一把生成答案 + 用 中间答案兜底(answer_generator.py:552-570)。is_final_retry 由外层 format_retry_count == max_format_retries 算出(common_benchmark.py:351)。

boxed 重试:抠到干净的 \boxed{} 为止

generate_final_answer_with_retries 先把 generate_agent_summarize_prompt(main) 这段严格格式指令 追加成一条 user(answer_generator.py:277-285),然后循环最多 max_final_answer_retries 次要模型 总结,每次抽 boxed:

# 示意,非源码:重点看"抽到非 FORMAT_ERROR 的 boxed 就停,否则回滚重试"
for retry_idx in range(self.max_final_answer_retries):
final_answer_text = await handle_llm_call(...) # 让模型总结
_, final_boxed_answer, _ = format_final_summary_and_log(...) # 抠 \boxed{}
if final_boxed_answer != FORMAT_ERROR_MESSAGE:
break # 拿到干净答案
if message_history[-1]["role"] == "assistant":
message_history.pop() # 撤掉这条失败的总结,重试

真实实现见 answer_generator.py:292-345。这里的重试次数也跟压缩挂钩:只有 keep_tool_result == -1(全留历史)时才给 3 次,否则 1 次(answer_generator.py:75-77)——历史被裁 过的场景,多要几次总结意义不大。

收尾时那段"严格格式"指令

generate_agent_summarize_prompt(agent_type="main") 是收尾的灵魂 (src/utils/prompt_utils.py:236-275)。它对模型下死命令,要点:

  • 若前文已有明确答案,别重算,直接抠出来改成要求格式。
  • 答案必须包进 \boxed{},且只能是"一个数字 / 尽量少的词 / 逗号分隔的列表"。
  • 严格遵循原题的格式要求(排序、单位、四舍五入、小数位…);数字用阿拉伯数字、不带单位、不带 千分位逗号;字符串不用冠词、结尾不带标点。
  • 绝对禁止再调任何工具——只能基于已检索到的信息和内部知识作答。

这段之所以这么啰嗦,是因为评测判分要精确匹配;格式差一点就判错。子 Agent(agent-browsing)有另一版 更侧重"把发现完整报告出来"的收尾 prompt(同函数 :276-294)。

兜底:中间 boxed 答案

intermediate_boxed_answers 是主循环过程中每次模型顺手给出 \boxed{} 时攒下来的列表。收尾若拿 不到干净 boxed 且允许兜底,就取最后一个中间答案顶上(answer_generator.py:392-401)——总比交白卷 或交个格式错误强。这只在"压缩关闭"或"最后一次重试"时启用,理由见上面的决策表。


7. 端到端走一遍(把四招串起来)

用一个 keep_tool_result=5 / context_compress_limit=5 的配置,想象一次典型的"跑满 turn 但没写对 boxed"的尝试:

turn 1..N:
每 turn 发请求前 → 招① 只把最近 5 个工具结果发给 LLM,更早的换占位
每 turn 结束 → 招② 估 token;某个 turn 最后一次工具返回超大,估算超窗
→ pop 掉最后一对消息,turn_count = max_turns,break
循环结束:reached_max_turns = True
招④ 决策:压缩 ON + 跑满 turn + 不是最后一次 → 跳过生成,直接:
招③ generate_failure_summary → 模型写三段式复盘 → 抽出 useful findings
返回 failure_experience_summary,boxed = FORMAT_ERROR_MESSAGE
外层 run_single_task:
收到 FORMAT_ERROR → format_retry_count += 1(仍 ≤ 5)
招③ 把这份复盘拼进 FAILURE_EXPERIENCE_HEADER/ITEM/FOOTER,增强任务描述
带着经验重跑 execute_task_pipeline …
…若到第 5 次(is_final_retry=True):招④ 改走"生成 + 中间答案兜底",搏最后一把

四招的分工一目了然:①②在一次尝试内部省 token、防爆窗;③在多次尝试之间传递教训;④在每次 尝试末尾把结论拧成干净的 \boxed{}


8. 边界与局限

  • token 估算是近似,不是精确。 ensure_summary_context 用"上次实际用量 + 1.5 倍缓冲"估当前,极端 情况下(比如单条工具结果暴增)仍可能与真实值有偏差;靠 +1000× 1.5 的保守缓冲兜。
  • 裁剪是"最近 N 个"的粗策略,不看语义相关性。真正重要但发生得早的工具结果一样会被换成占位—— 依赖模型在被裁前已把要点写进自己的思考/文本里。
  • 失败复盘由模型自己写,质量取决于模型;若复盘本身跑偏,喂回去的经验也可能误导下一次。
  • boxed 兜底只在压缩关闭 / 最后一次重试时启用,是刻意的准确率权衡(见 §6 决策表),不是 bug。
  • 格式严格性是双刃剑: generate_agent_summarize_prompt 那套规则为判分服务,若原题格式要求本身 含糊,严格 prompt 反而可能把对的答案格式化"过头"。

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

主题文件路径符号名
上下文裁剪(留最近 N 个工具结果)src/llm/base_client.py_remove_tool_result_from_messages
LLM 调用入口(透传 keep_tool_result)src/llm/base_client.pycreate_message
发请求前裁剪 + 返回原本src/llm/providers/anthropic_client.py_create_message
窗口守卫(超窗前退一步)src/llm/providers/anthropic_client.pyensure_summary_context
token 粗估(tiktoken + 兜底)src/llm/providers/anthropic_client.py_estimate_tokens
记录上次调用用量src/llm/providers/anthropic_client.py_update_token_usage
收尾主入口(四象限决策)src/core/answer_generator.pygenerate_and_finalize_answer
失败复盘生成src/core/answer_generator.pygenerate_failure_summary
boxed 重试循环src/core/answer_generator.pygenerate_final_answer_with_retries
压缩关闭:中间答案兜底src/core/answer_generator.pyhandle_no_context_management_fallback
压缩开启:不兜底src/core/answer_generator.pyhandle_context_management_no_fallback
复盘 prompt / 前缀src/utils/prompt_utils.pyFAILURE_SUMMARY_PROMPTFAILURE_SUMMARY_ASSISTANT_PREFIX
失败经验拼接模板src/utils/prompt_utils.pyFAILURE_EXPERIENCE_HEADER/_ITEM/_FOOTER
收尾严格格式指令src/utils/prompt_utils.pygenerate_agent_summarize_prompt
复盘正文抽取src/utils/parsing_utils.pyextract_failure_experience_summary
评测外层 format 重试循环benchmarks/common_benchmark.pyrun_single_task