压缩与容错:研究压缩、token 超限重试、最终报告生成

30 秒导读: 一个 researcher 搜完一堆资料后,消息历史会大到爆上下文窗口;把它们喂给模型"蒸馏成摘要""拼成最终报告"时,随时可能撞上"prompt 太长"这种错。本章讲这条流水线尾端的三处 LLM 产出——研究压缩、跨供应商 token 超限识别、最终报告生成——以及把它们裹起来的那套**"认出超限→砍上下文→重试"**的容错逻辑。这是本库工程含量最高、最能体现 bitter-lesson 式健壮性的部分。

前置:研究引擎怎么把 supervisor 委派、researcher 怎么跑 ReAct 循环并并行 fan-out,见 02-supervisor-researcher-loop.md;整张 LangGraph 的三层编排见 01-orchestration-graph.md。本章只讲这条链的收尾与容错。

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

一句话定义: 研究做完之后,把"一大堆对话消息"变成"一段干净摘要",再变成"一篇报告"——而且做这两步的时候,假设它会因为内容太长而失败,并准备好一层层降级重试。

它要解决的真实痛点。 深度研究天然会产生海量文本:

一个 researcher 可能调了十几次搜索工具,每次工具返回几千字。
这些工具消息 + 模型的思考消息全堆在一个消息列表里。
到了要"总结"或"写报告"的时候,这个列表塞进 prompt,很容易超过模型的上下文窗口(比如 GPT-4o 的 128k、Claude 的 200k)。
一旦超,API 直接抛异常——如果不管,整个研究任务前功尽弃。

一句话直觉/类比。 把它想成做读书笔记再写论文:

压缩(compress) = 把你翻过的一摞书、划的线,浓缩成一页笔记(但底稿留着)。
写报告(final report) = 把所有人的笔记拼起来,写成一篇成文。
容错 = 如果笔记多到桌子摆不下,就先撤掉最近翻的那几本;如果论文素材多到写不动,就按比例先删掉一截再写。

为什么值得单独讲。 因为"调用模型"从来不是难点,"在真实供应商、真实长文本下不崩"才是。这一章的代码,大半不是在产出内容,而是在兜住失败。

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

三处 LLM 产出,分布在两个位置:

产出点	在哪	输入	输出
研究压缩 `compress_research`	researcher 子图收尾	一个 researcher 的全部消息	`compressed_research`(摘要)+ `raw_notes`(底稿)
最终报告 `final_report_generation`	主图收尾	所有 `notes` 拼成的 `findings`	`final_report`(成文)
(第三处是"token 超限识别",不产内容,是前两处共用的容错判据)	`utils.py`	一个异常 + 模型名	是/否超限

主线走一遍(高层,不进代码):

researcher 搜完
   │  researcher_messages = [AI思考, Tool结果, AI思考, Tool结果, ...]
   ▼
┌─────────────────────────────────────────────┐
│ compress_research(researcher 子图收尾)       │
│  · 追加一句"请清理这些发现"的 human message   │
│  · while 循环最多 3 次:                       │
│      调压缩模型 → 成功则返回摘要+底稿          │
│      失败且是 token 超限 → 砍掉最后一段 AI 之后│
│                            的消息,再试        │
│  · 3 次都败 → 返回一句错误串当摘要(不崩)     │
└─────────────────────────────────────────────┘
   │  compressed_research 汇成 supervisor 的 notes
   ▼   (多个 researcher 并行,notes 逐条累加)
┌─────────────────────────────────────────────┐
│ final_report_generation(主图收尾)           │
│  · findings = 所有 notes 拼一起               │
│  · while 循环最多 3 次:                       │
│      调报告模型 → 成功则返回报告               │
│      失败且是 token 超限 → 渐进截断 findings   │
│           第1次:按 模型上限×4 字符 截          │
│           之后:上次长度 ×0.9 再截             │
│  · 查不到模型上限 / 非超限错 → 立刻返回错误串   │
└─────────────────────────────────────────────┘
   │
   ▼  final_report(总能返回一个字符串)

贯穿两处的同一条暗线: 每一次 LLM 调用都被 try/except + while 包住,except 里先问一句"这是不是 token 超限?"——是,就砍上下文重试;不是,就当普通失败处理。这句"是不是超限"的判断,就是第三处的主角。

3. 核心原理之一:研究压缩 `compress_research`

它要解决的小问题

一个 researcher 跑完,researcher_messages 里塞满了 AI 思考和工具返回。这些原始素材太长、太乱,不能直接往上层传。要把它蒸馏成一段结构化摘要,同时别丢信息——所以还要把原始底稿也留一份。

思路/直觉

两件事一起做,缺一不可:

蒸馏:让一个"压缩模型"读全部消息,输出干净摘要 compressed_research。
留底:把所有 tool + ai 消息原样拼成 raw_notes 存档,防止摘要漏掉细节时还能回溯。

而蒸馏这一步本身可能因为"消息太长"失败,所以外面裹一层重试。

真实实现

配置压缩专用模型,并追加一句"请清理这些发现"的人类消息把模型从"研究模式"切到"压缩模式":

# deep_researcher.py:527-538  compress_research
synthesizer_model = configurable_model.with_config({
    "model": configurable.compression_model,
    "max_tokens": configurable.compression_model_max_tokens,
    ...
})
researcher_messages = state.get("researcher_messages", [])
researcher_messages.append(HumanMessage(content=compress_research_simple_human_message))

compression_model 是可独立配置的字段(configuration.py:173 compression_model),压缩可以用比研究更便宜的模型。

核心是这段 while 重试循环——成功即返回,超限则砍上下文继续:

# deep_researcher.py:544-574  compress_research(节选)
while synthesis_attempts < max_attempts:          # max_attempts = 3
    try:
        compression_prompt = compress_research_system_prompt.format(date=get_today_str())
        messages = [SystemMessage(content=compression_prompt)] + researcher_messages
        response = await synthesizer_model.ainvoke(messages)
        raw_notes_content = "\n".join([                       # 留底:tool+ai 原样拼接
            str(message.content)
            for message in filter_messages(researcher_messages, include_types=["tool", "ai"])
        ])
        return {"compressed_research": str(response.content),
                "raw_notes": [raw_notes_content]}
    except Exception as e:
        synthesis_attempts += 1
        if is_token_limit_exceeded(e, configurable.research_model):
            researcher_messages = remove_up_to_last_ai_message(researcher_messages)   # 砍上下文
            continue
        continue                                              # 其它错也重试

filter_messages(..., include_types=["tool", "ai"])(deep_researcher.py:556)把工具结果和 AI 思考挑出来当底稿。

"砍上下文"到底砍掉什么。 remove_up_to_last_ai_message 从尾往前找到最后一条 AIMessage,返回它之前的所有消息(把它和它之后的都丢掉):

# utils.py:848-866  remove_up_to_last_ai_message
for i in range(len(messages) - 1, -1, -1):
    if isinstance(messages[i], AIMessage):
        return messages[:i]      # 丢掉最后一段 AI 及其之后的消息
return messages                  # 没有 AI 消息就原样返回

直觉:最后一轮"AI 思考 + 它触发的工具返回"往往是压垮上下文的那一截,先砍掉它,消息短了,下一次 ainvoke 就可能塞得下。每重试一次砍一段,循环最多 3 次。

永不崩的兜底。 3 次都失败,不抛异常,而是返回一句错误字符串当作摘要,底稿照留:

# deep_researcher.py:582-585
return {"compressed_research": "Error synthesizing research report: Maximum retries exceeded",
        "raw_notes": [raw_notes_content]}

关键细节/坑

超限判据用的是 research_model 而不是 compression_model(deep_researcher.py:569 传的是 configurable.research_model)。两者不一致时,识别用的模型名可能和真正发压缩请求的模型对不上——属于代码里的小瑕疵,但因为 is_token_limit_exceeded 在 provider 未知时会遍历所有供应商(见 §5),多数情况仍能识别。
摘要如何上浮成 notes。 researcher 子图返回的 compressed_research 会被 supervisor 侧读出、拼进 notes(deep_researcher.py:310、324-330);notes / raw_notes 都用 override_reducer 累加(state.py:70-71)。最终报告吃的就是这堆 notes。

4. 核心原理之二:最终报告生成 `final_report_generation`

它要解决的小问题

所有 researcher 的摘要都汇进了 notes。现在要把它们拼成 findings,连同研究简报一起喂给"报告模型"写成一篇成文。同样的老问题:findings 可能长到爆上下文。

思路/直觉

跟压缩一样裹 while 重试,但降级方式不同:压缩是"砍消息",报告是"按字符渐进截断 findings"。因为报告的输入是一整块拼好的文本,没有"最后一条 AI 消息"可砍,只能直接截字符串;而且要估算该截到多长——这里用了一个很务实的近似。

真实实现

先把 notes 拼成 findings,配置报告模型:

# deep_researcher.py:621-632  final_report_generation
notes = state.get("notes", [])
cleared_state = {"notes": {"type": "override", "value": []}}   # 用完清空 notes
findings = "\n".join(notes)
writer_model_config = {
    "model": configurable.final_report_model,
    "max_tokens": configurable.final_report_model_max_tokens,
    ...
}

核心是这段渐进降级重试:

# deep_researcher.py:639-683  final_report_generation(节选)
while current_retry <= max_retries:                # max_retries = 3
    try:
        final_report_prompt = final_report_generation_prompt.format(
            research_brief=state.get("research_brief", ""),
            messages=get_buffer_string(state.get("messages", [])),
            findings=findings, date=get_today_str())
        final_report = await configurable_model.with_config(writer_model_config).ainvoke(
            [HumanMessage(content=final_report_prompt)])
        return {"final_report": final_report.content, "messages": [final_report], **cleared_state}
    except Exception as e:
        if is_token_limit_exceeded(e, configurable.final_report_model):
            current_retry += 1
            if current_retry == 1:
                model_token_limit = get_model_token_limit(configurable.final_report_model)
                if not model_token_limit:
                    return {"final_report": "Error ... could not determine the model's maximum context length ...", ...}
                findings_token_limit = model_token_limit * 4     # 4 字符 ≈ 1 token 的近似
            else:
                findings_token_limit = int(findings_token_limit * 0.9)   # 之后每次再砍 10%
            findings = findings[:findings_token_limit]
            continue
        else:
            return {"final_report": f"Error generating final report: {e}", ...}   # 非超限错,立刻返回

降级阶梯看这张表就清楚(设模型上限 L):

第几次重试	`findings` 截断到多少字符	依据
第 1 次超限	`L × 4`	`deep_researcher.py:676`,4 字符≈1 token 的粗略近似
第 2 次超限	上一次 × 0.9	`deep_researcher.py:679`
第 3 次超限	再 × 0.9	同上,逐次递减
4 次都败	返回 "Maximum retries exceeded"	`deep_researcher.py:693-696`

"4 字符 ≈ 1 token" 是一个刻意粗糙的近似:英文里一个 token 平均约 4 个字符,所以 L token 的上限大致对应 L×4 字符。它不追求精确,只要把明显过长的 findings 砍到一个大概能塞下的量级,再靠后续 ×0.9 逐步逼近即可。

关键细节/坑

查不到上限就直接放弃并给出可操作提示。 第 1 次超限时若 get_model_token_limit 返回 None(模型不在表里),不硬猜,而是返回一段错误串,明确提示去 utils.py 更新模型表(deep_researcher.py:669-674)。诚实降级胜过瞎截。
非 token 超限的错立刻返回(deep_researcher.py:684-690),不浪费重试次数——只有"超限"这种"砍一砍还能救"的错才值得重试。
notes 用完即清空(cleared_state,deep_researcher.py:622),避免状态残留;notes 走 override_reducer(state.py:71)支持这种整体覆写。

5. 核心原理之三:跨供应商 token 超限识别 `is_token_limit_exceeded`

这是本章的技术核心,也是前两处 except 里那句判断的实现。难点在于:不同供应商把"上下文超限"这件事,报成完全不同的异常类和错误文案。 没有统一标准,只能一家家认。

思路/直觉

两步走:

先按模型名前缀猜 provider(openai: / anthropic: / gemini: / google:),命中就只查那一家的模式——快。
猜不出 provider,就三家全查(OR 到一起)——稳。

# utils.py:665-701  is_token_limit_exceeded(节选)
error_str = str(exception).lower()
provider = None
if model_name:
    model_str = str(model_name).lower()
    if model_str.startswith('openai:'):       provider = 'openai'
    elif model_str.startswith('anthropic:'):   provider = 'anthropic'
    elif model_str.startswith('gemini:') or model_str.startswith('google:'):
        provider = 'gemini'
if provider == 'openai':      return _check_openai_token_limit(exception, error_str)
elif provider == 'anthropic': return _check_anthropic_token_limit(exception, error_str)
elif provider == 'gemini':    return _check_gemini_token_limit(exception, error_str)
return (_check_openai_token_limit(exception, error_str) or     # provider 未知:全查
        _check_anthropic_token_limit(exception, error_str) or
        _check_gemini_token_limit(exception, error_str))

三家各自怎么认

每个 _check_* 都走同一套路:先看异常类型/模块名确认是不是这家的异常,再匹配类名 + 关键字。三家的具体判据各不相同:

供应商	先确认"是这家的异常"	再看类名	关键判据	源码
OpenAI	类型串或模块名含 `openai`	`BadRequestError` / `InvalidRequestError`	错误文案含 `token`/`context`/`length`/`maximum context`/`reduce` 任一;或 `code=='context_length_exceeded'` / `type=='invalid_request_error'`	`utils.py:703-734`
Anthropic	类型串或模块名含 `anthropic`	`BadRequestError`	错误文案含 `prompt is too long`	`utils.py:736-757`
Gemini/Google	类型串或模块名含 `google`	`ResourceExhausted` / `GoogleGenerativeAIFetchError`	命中类名即算;或类型串含 `google.api_core.exceptions.resourceexhausted`	`utils.py:759-785`

以 OpenAI 为例,能看到"类名 + 关键字"和"错误码"两条并行判据:

# utils.py:717-732  _check_openai_token_limit(节选)
is_request_error = class_name in ['BadRequestError', 'InvalidRequestError']
if is_openai_exception and is_request_error:
    token_keywords = ['token', 'context', 'length', 'maximum context', 'reduce']
    if any(keyword in error_str for keyword in token_keywords):
        return True
if hasattr(exception, 'code') and hasattr(exception, 'type'):
    if (getattr(exception, 'code', '') == 'context_length_exceeded' or
        getattr(exception, 'type', '') == 'invalid_request_error'):
        return True

这种字符串/类名匹配是典型的"脆而实用":没有优雅接口能问"这个异常是不是上下文超限",于是干脆按经验把各家已知的错误特征硬编码进来,并留了一句诚实的注释——它可能过时,请自行更新(utils.py:787 上方 # NOTE: This may be out of date ...)。这正是 bitter lesson 的味道:与其等一个统一抽象,不如把现实里的脏东西一条条兜住。

模型上限表与子串匹配

报告降级要知道"模型上限是多少",靠一张硬编码表:

# utils.py:788-829  MODEL_TOKEN_LIMITS(节选)
MODEL_TOKEN_LIMITS = {
    "openai:gpt-4.1": 1047576,
    "openai:gpt-4o": 128000,
    "anthropic:claude-opus-4": 200000,
    "google:gemini-1.5-pro": 2097152,
    ...   # 覆盖 openai/anthropic/google/cohere/mistral/ollama/bedrock
}

查表用的是子串包含而非精确相等——model_key in model_string,只要模型字符串里含某个 key 就命中:

# utils.py:831-846  get_model_token_limit
for model_key, token_limit in MODEL_TOKEN_LIMITS.items():
    if model_key in model_string:      # 子串匹配,容忍带后缀/日期的模型名
        return token_limit
return None                            # 查不到 → 上层给出"请更新模型表"的提示

为什么用子串。 真实模型名常带日期/版本后缀(如 anthropic:claude-sonnet-4-20250514),精确相等会漏;子串匹配让 "anthropic:claude-sonnet-4" 这样的 key 仍能命中。代价是要注意 key 的顺序与包含关系(短 key 可能先命中),但对当前这张表够用。

6. 巧妙之处(可带走的技术)

"每次 LLM 调用都假设它会因太长而失败"。压缩和报告两处都是 try 里调模型、except 里先问"是不是超限"、while 里降级重试——把脆弱的单次调用磨成可降级流程。(deep_researcher.py:544-574、639-683)
两种降级手段对应两种输入形态:压缩的输入是消息列表 → 砍"最后一段 AI 及其后"(remove_up_to_last_ai_message);报告的输入是一整块字符串 → 按字符渐进截断。同一目标,因数据结构不同而选不同刀法。
"4 字符≈1 token"的粗近似 + 逐次 ×0.9:不追求精确 token 计数,用最便宜的估算迈第一大步,再用几何递减慢慢逼近能塞下的边界。(deep_researcher.py:676、679)
provider 先猜后兜底:能从模型前缀确定就只查一家(快),否则三家全 OR(稳)。(utils.py:677-701)
永远返回点东西,绝不把异常抛穿:重试耗尽也返回一句错误串当结果(deep_researcher.py:582、693),让上层图能继续收尾而非整体崩溃。
查不到就给可操作指令:模型不在表里时,报告直接返回"请去 utils.py 更新模型表"(deep_researcher.py:669-674)——诚实地把维护责任显式暴露出来。

7. 边界与局限(诚实)

超限识别本质是字符串/类名匹配,会漂移。 供应商改文案、改异常类名,或新增供应商(Groq/DeepSeek 等未在三家 _check_* 里),就可能认不出——代码自己在 utils.py:787 上方注明了这点。认不出时,超限会被当作"其它错":压缩仍会盲目重试(deep_researcher.py:573-574),报告则会当非超限错立刻返回(deep_researcher.py:684-690)。
压缩重试对"非超限错"也会重试到耗尽(deep_researcher.py:573),但每次不缩短上下文,若错误是确定性的(如鉴权失败),3 次都会白跑。
压缩超限判据用了 research_model 而非压缩实际用的 compression_model(deep_researcher.py:569),两者模型不同时识别可能偏差,靠"provider 未知则全查"兜底。
MODEL_TOKEN_LIMITS 是静态硬编码,新模型/新上限需手动维护;get_model_token_limit 的子串匹配对 key 顺序敏感。

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

主题	文件路径	符号名
研究压缩 + 压缩侧重试循环	`src/open_deep_research/deep_researcher.py:511`	`compress_research`
最终报告生成 + 渐进截断降级	`src/open_deep_research/deep_researcher.py:607`	`final_report_generation`
压缩摘要上浮成 supervisor 的 notes	`src/open_deep_research/deep_researcher.py:310`	(`compressed_research` → `notes`)
跨供应商超限识别(入口)	`src/open_deep_research/utils.py:665`	`is_token_limit_exceeded`
OpenAI 超限模式匹配	`src/open_deep_research/utils.py:703`	`_check_openai_token_limit`
Anthropic 超限模式匹配	`src/open_deep_research/utils.py:736`	`_check_anthropic_token_limit`
Gemini/Google 超限模式匹配	`src/open_deep_research/utils.py:759`	`_check_gemini_token_limit`
模型上限表	`src/open_deep_research/utils.py:788`	`MODEL_TOKEN_LIMITS`
上限查表(子串匹配)	`src/open_deep_research/utils.py:831`	`get_model_token_limit`
压缩侧砍上下文	`src/open_deep_research/utils.py:848`	`remove_up_to_last_ai_message`
从工具消息抽 notes	`src/open_deep_research/utils.py:599`	`get_notes_from_tool_calls`
notes/raw_notes 累加 reducer	`src/open_deep_research/state.py:55`	`override_reducer`
压缩/报告模型配置字段	`src/open_deep_research/configuration.py:173`	`compression_model` / `final_report_model`
压缩系统/人类提示词、报告提示词	`src/open_deep_research/prompts.py:186`	`compress_research_system_prompt` / `compress_research_simple_human_message` / `final_report_generation_prompt`

相邻章节:研究引擎(supervisor 委派 + researcher ReAct + 并行 fan-out)见 02-supervisor-researcher-loop.md;三层 LangGraph 编排见 01-orchestration-graph.md;工具装配(搜索抽象、Tavily 摘要、MCP)见 04-tools-search-mcp.md;全库总览见 index.md。

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

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

3. 核心原理之一:研究压缩 compress_research​

它要解决的小问题​

思路/直觉​

真实实现​

关键细节/坑​

4. 核心原理之二:最终报告生成 final_report_generation​

它要解决的小问题​

思路/直觉​

真实实现​

关键细节/坑​

5. 核心原理之三:跨供应商 token 超限识别 is_token_limit_exceeded​

思路/直觉​

三家各自怎么认​

模型上限表与子串匹配​

6. 巧妙之处(可带走的技术)​

7. 边界与局限(诚实)​

8. 代码地图(导航索引)​

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

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

3. 核心原理之一:研究压缩 `compress_research`

它要解决的小问题

思路/直觉

真实实现

关键细节/坑

4. 核心原理之二:最终报告生成 `final_report_generation`

它要解决的小问题

思路/直觉

真实实现

关键细节/坑

5. 核心原理之三:跨供应商 token 超限识别 `is_token_limit_exceeded`

思路/直觉

三家各自怎么认

模型上限表与子串匹配

6. 巧妙之处(可带走的技术)

7. 边界与局限(诚实)

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