1. 本章目标 #

前面几章一直在往上下文里加东西:系统提示词、记忆、技能、文件、子智能体报告。这一章反过来——上下文满了怎么办。

好消息是 Deep Agents 已经替你处理了大部分情况:create_deep_agent 默认就带卸载和摘要,你不加任何 middleware 它也在工作。坏消息是这些机制的阈值和行为你不了解的话,会在两个地方吃亏:一是不知道钱花在哪了,二是任务被压缩坏了不知道为什么。

学完你应能:

前置依赖: 第 42 章(文件系统,卸载全靠它)、第 44 章和第 45 章(技能与记忆的开销,本章把三笔账合起来算)、第 40 章(子智能体,它是上下文隔离的手段)。

参考文档:

建议阅读顺序: §4 到 §6 是核心。§5 和 §6 各有一个和直觉不符的地方,分别是预览的形式和「摘要不改 state」,建议重点看。§9 那份清单可以直接抄。

本章验证环境:deepagents 0.7.12。本章几乎所有实验都是零 token 的——上下文压缩的行为可以用假模型完整观察,这也是研究这套机制的最佳方式。

2. 上下文的五种类型 #

官方把上下文分成五类,各自有不同的控制手段:

类型 你控制什么 作用范围
输入上下文 启动时进提示词的东西(系统提示词、记忆、技能) 静态,每次运行都加载
运行时上下文 调用时传入的静态配置(用户元数据、API key、连接) 每次运行,会传播给子智能体
上下文压缩 内置的卸载和摘要 自动,接近限制时触发
上下文隔离 用子智能体隔离重活,只回结果 每个子智能体,委派时
长期记忆 通过虚拟文件系统跨线程持久化 跨会话

前两章讲的是第一类,第 40 章讲的是第四类,第 45 章讲的是第五类。本章主攻第三类,顺带把第二类补上。

运行时上下文有个容易误解的点:它不会自动进模型提示词。你传 context= 之后,只有工具、middleware 或其它逻辑主动读它并写进消息,模型才看得到。它的用途是给你的代码用,不是给模型看:

"""运行时上下文:给工具用的配置,模型看不到。"""
from dataclasses import dataclass

from langchain.tools import ToolRuntime, tool

from deepagents import create_deep_agent


@dataclass
class Context:
    user_id: str
    api_key: str


@tool
def fetch_user_data(query: str, runtime: ToolRuntime[Context]) -> str:
    """查询当前用户的数据。"""
    # 从运行时上下文取,不作为工具参数暴露给模型
    user_id = runtime.context.user_id
    return f"用户 {user_id} 的数据:{query}"


agent = create_deep_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[fetch_user_data],
    context_schema=Context,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "查我的订单"}]},
    context=Context(user_id="u-123", api_key="sk-..."),
)

这个模式在第 45 章 §9.1 见过——user_id 从 runtime 取而不做成参数,是为了防止模型(被诱导)去查别人的数据。API key 这类东西尤其要走这条路,绝不能进提示词。

3. 系统提示词是怎么拼出来的 #

前面几章各自量过自己那部分的开销,现在把完整的拼装顺序摆出来。官方给的 8 段:

  1. 你的 system_prompt(如果提供)
  2. 基础 agent 提示词
  3. 记忆提示词:AGENTS.md 内容 + 使用指南(仅当提供 memory)
  4. 技能提示词:技能位置 + 技能清单 + 使用说明(仅当提供 skills)
  5. 虚拟文件系统提示词(文件工具文档,沙箱后端时含 execute)
  6. 子智能体提示词(task 工具用法)
  7. 你的自定义 middleware 的提示词(如果有)
  8. Human-in-the-loop 提示词(仅当设置 interrupt_on)

这个顺序解释了第 44 章和第 45 章那些数字的来源:传 memory= 多出的 5245 字符是第 3 段,传 skills= 多出的 2011 字符是第 4 段。

顺带说明一个细节:你自己写的 system_prompt 在最前面。官方文档里的措辞是最终提示词按 USER → BASE → SUFFIX 组装——USER 就是你传的那段。所以你的指令优先级最高,但后面几段会补充大量工具用法说明,可能和你的指令产生张力。写 system_prompt 时不用重复解释工具怎么用,框架会说。

4. 三笔看不见的基线开销 #

在讲压缩之前,先把「还没开始干活就已经花掉的钱」算清楚。这三笔都在每一轮请求里重复发送。

第一笔:工具 schema。 这是最容易被忽略的一笔,因为它不在系统提示词里,而是在请求的 tools 字段:

"""零 token:量一下内置工具的 schema 开销。"""
import json

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langgraph.checkpoint.memory import InMemorySaver

from deepagents import create_deep_agent
from deepagents.backends import StateBackend

CAP = {}


class Fake(BaseChatModel):
    @property
    def _llm_type(self) -> str:
        return "fake"

    def bind_tools(self, tools, **kw):
        # 工具的描述 + JSON schema,这些每轮都要发
        names, total = [], 0
        for t in tools:
            names.append(getattr(t, "name", "?"))
            try:
                sch = t.args_schema.model_json_schema() if getattr(t, "args_schema", None) else {}
            except Exception:
                sch = {}
            total += len(getattr(t, "description", "") or "") + len(json.dumps(sch))
        CAP["names"] = names
        CAP["chars"] = total
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kw) -> ChatResult:
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content="ok"))])


agent = create_deep_agent(model=Fake(), backend=StateBackend(),
                          checkpointer=InMemorySaver())
agent.invoke({"messages": [{"role": "user", "content": "hi"}]},
             {"configurable": {"thread_id": "t"}})
print(f"工具 {len(CAP['names'])} 个: {CAP['names']}")
print(f"描述 + schema 合计 {CAP['chars']} 字符")
工具 8 个: ['ls', 'read_file', 'write_file', 'edit_file', 'delete', 'glob', 'grep', 'task']
描述 + schema 合计 10894 字符

默认 8 个内置工具,每轮 10894 字符(约 2700 token)。 这笔钱和你的任务无关,纯粹是「有这些工具可用」的代价。

想砍掉它,用 HarnessProfile 的 excluded_tools。注意它不是 create_deep_agent 的参数,而要通过注册生效:

"""只读智能体:把写操作工具从基线里去掉。"""
from deepagents import HarnessProfile, register_harness_profile

# key 是 provider 名或具体模型名;注册是累加的
register_harness_profile(
    "deepseek",
    HarnessProfile(excluded_tools=frozenset(["write_file", "edit_file", "delete"])),
)

效果:

默认(全部 8 个内置工具)
  工具 8 个: ['ls', 'read_file', 'write_file', 'edit_file', 'delete', 'glob', 'grep', 'task']
  描述+schema 合计 10894 字符

注册 profile 后,排除三个写工具
  工具 5 个: ['ls', 'read_file', 'glob', 'grep', 'task']
  描述+schema 合计 8308 字符
  -> 省 2586 字符(24%)

这是配置,不是压缩。 它在压缩机制介入之前就把基线缩小了,所以对整个运行期都有效。只读智能体、或者明确不用沙箱的场景,值得配一下。(HarnessProfile 的完整能力留到第 50 章。)

第二笔和第三笔是前两章量过的:

开销项 大小 条件
工具 schema 10894 字符 默认 8 个内置工具,每轮
记忆固定说明 5245 字符 传了 memory=,每轮
技能固定说明 2011 字符 传了 skills=,每轮

三笔全开:18150 字符(约 4500 token)起步,此时你的业务内容还一个字都没进去。

这不是说这些机制不该用,而是提醒:在评估「上下文快满了」之前,先知道有多少是固定成本。 一个 8K 上下文的小模型,光这三笔就吃掉一半多。

5. 卸载:大工具结果自动落盘 #

现在进入压缩。第一个机制是卸载(offloading):工具的输入或结果超过阈值时,内容存到文件系统,上下文里只留一个引用。

这个机制默认就开着,不需要加任何 middleware。

5.1. 阈值实测 #

内部参数叫 tool_token_limit_before_evict,默认 20000。用一个能返回任意大小结果的工具验证:

"""零 token:卸载的触发阈值和替换形式。"""
from typing import List

from langchain.tools import tool
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langgraph.checkpoint.memory import InMemorySaver

from deepagents import create_deep_agent
from deepagents.backends import StateBackend

NUM_CHARS_PER_TOKEN = 4      # 框架内部的 token 估算比例


def make_payload(n_tokens):
    """造一个约 n_tokens 个 token 的文本,每行带行号方便看预览。"""
    line = "这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。"
    out, i = [], 1
    while sum(len(x) for x in out) < n_tokens * NUM_CHARS_PER_TOKEN:
        out.append(f"[{i:05d}] {line}")
        i += 1
    return "\n".join(out)


SIZES = {"small": 1000, "just_under": 19000, "over": 25000, "huge": 60000}
PAYLOADS = {k: make_payload(v) for k, v in SIZES.items()}


@tool
def dump_logs(size: str) -> str:
    """导出日志。size 可选 small / just_under / over / huge。"""
    return PAYLOADS[size]


class Scripted(BaseChatModel):
    """按脚本调一次工具就收尾,不消耗 token。"""

    script: List[dict] = []
    step: int = 0

    @property
    def _llm_type(self) -> str:
        return "scripted"

    def bind_tools(self, tools, **kw):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kw) -> ChatResult:
        i = self.step
        self.step = i + 1
        if i < len(self.script):
            s = self.script[i]
            msg = AIMessage(content="", tool_calls=[
                {"name": s["name"], "args": s["args"], "id": f"c{i}", "type": "tool_call"}])
        else:
            msg = AIMessage(content="done")
        return ChatResult(generations=[ChatGeneration(message=msg)])


for size in SIZES:
    agent = create_deep_agent(
        model=Scripted(script=[{"name": "dump_logs", "args": {"size": size}}]),
        tools=[dump_logs],
        backend=StateBackend(),
        checkpointer=InMemorySaver(),
    )
    r = agent.invoke({"messages": [{"role": "user", "content": "导出日志"}]},
                     {"configurable": {"thread_id": size}})
    original = PAYLOADS[size]
    for m in r["messages"]:
        if type(m).__name__ == "ToolMessage":
            body = m.content if isinstance(m.content, str) else str(m.content)
            print(f"{size:11s} 原始 {len(original):7d} 字符 -> 模型收到 {len(body):6d} 字符 "
                  f"({'已卸载' if len(body) < len(original) * 0.5 else '原样'})")
small       原始    4141 字符 -> 模型收到   4141 字符 (原样)
just_under  原始   78089 字符 -> 模型收到  78089 字符 (原样)
over        原始  102713 字符 -> 模型收到   1051 字符 (已卸载)
huge        原始  246505 字符 -> 模型收到   1051 字符 (已卸载)

19000 token 原样通过,25000 token 被卸载——阈值就是 20000 token。

注意压缩比:不管原始是 10 万还是 24 万字符,替换后都是固定的 1051 字符。

5.2. 替换成了什么 #

这是模型实际收到的完整内容:

Tool result too large, the result of this tool call c0 was saved in the filesystem
at this path: /large_tool_results/c0

You can read the result from the filesystem by using the read_file tool, but make
sure to only read part of the result at a time...

You can do this by specifying an offset and limit in the read_file tool call. For
example, to read the first 100 lines...

Here is a preview showing the head and tail of the result (lines of the form
`... [N lines truncated] ...` indicate omitted content):

1  [00001] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
2  [00002] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
3  [00003] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
4  [00004] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
5  [00005] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
... [2693 lines truncated] ...
2699  [02699] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
2700  [02700] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
2701  [02701] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
2702  [02702] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。
2703  [02703] 这一行是模拟的日志内容,用来把工具结果撑大到超过卸载阈值。

三个要点:

  1. 文件路径是 /large_tool_results/<tool_call_id>。 用工具调用 ID 命名,所以每次调用一个文件、不会互相覆盖
  2. 提示明确教模型用 offset + limit 分段读,而不是一次读全(否则又炸回去了)
  3. 预览是「头 5 行 + 尾 5 行」,中间用 ... [N lines truncated] ... 标记

第 3 点和官方文档不一致。文档的说法是「a preview of the first 10 lines」(前 10 行的预览),实测是头尾各 5 行。提示文本自己也写明了是 a preview showing the head and tail。

这个差异有实际意义:头尾预览比只看开头有用得多。日志、报表这类内容的结尾往往有汇总信息,能看到尾部就能判断要不要细读、以及大概该读哪一段。

内容完整保存在文件里:

state 里的文件: ['/large_tool_results/c0']
  /large_tool_results/c0: 102713 字符(和原始一样长: True)

一个字都没丢,模型随时能 read_file 或 grep 回查。

5.3. 另一半:工具输入的卸载 #

上面测的是工具结果。工具输入也会被处理,但机制不同:

写文件、编辑文件这类操作,会在对话历史里留下包含完整文件内容的工具调用。这些内容既然已经落盘了,留在历史里就是冗余。所以:

当会话上下文超过模型窗口的 85% 时,旧的工具调用会被截断,替换成指向磁盘文件的指针。

注意触发条件是两个:工具输入超过 20000 token,且上下文到了 85%。所以它比结果卸载更晚介入——毕竟内容已经安全落盘了,不急着清。

最后一个限制要记住:内置压缩不处理图片。 它不会缩放图片、不降分辨率、不生成视觉嵌入。多模态内容的上下文管理要单独考虑。

6. 摘要:历史压缩的双轨机制 #

卸载解决的是「单条消息太大」。当消息条数本身太多、且没有更多可卸载的内容时,第二个机制介入:摘要。

create_deep_agent 的默认栈里就有 SummarizationMiddleware,同样不需要你配置。

6.1. 触发阈值从哪来 #

框架有个函数直接算这些默认值,可以调出来看:

"""零 token:看框架给你的模型算出了什么阈值。"""
import inspect

from deepagents.middleware.summarization import compute_summarization_defaults

print(inspect.getsource(compute_summarization_defaults))

源码非常直白:

def compute_summarization_defaults(model: BaseChatModel) -> SummarizationDefaults:
    has_profile = (
        model.profile is not None
        and isinstance(model.profile, dict)
        and "max_input_tokens" in model.profile
        and isinstance(model.profile["max_input_tokens"], int)
    )

    if has_profile:
        return {
            "trigger": ("fraction", 0.85),
            "keep": ("fraction", 0.10),
            "truncate_args_settings": {
                "trigger": ("fraction", 0.85),
                "keep": ("fraction", 0.10),
            },
        }

    # Defaults for models without profile info are more conservative to avoid
    # overshooting context limits.
    return {
        "trigger": ("tokens", 170000),
        "keep": ("messages", 6),
        "truncate_args_settings": {
            "trigger": ("messages", 20),
            "keep": ("messages", 20),
        },
    }

所以规则是:

情况 触发 保留
模型有 profile(含 max_input_tokens) 窗口的 85% 窗口的 10%
模型没有 profile 固定 170000 token 只留 6 条消息

没有 profile 时的回退值明显更保守(源码注释也说了是「avoid overshooting context limits」)。这有个实际影响:如果你用自定义模型类或者不常见的模型,会走到这条路径上,摘要在 17 万 token 就触发、而且只给你留 6 条消息。上下文窗口更大的模型会白白浪费空间,更小的模型则可能在 17 万之前就炸了。

实际算一下常见模型:

deepseek:deepseek-v4-flash
  trigger = ('fraction', 0.85)
  keep    = ('fraction', 0.1)
  (model profile max_input_tokens = 1000000)

anthropic:claude-sonnet-4-6
  trigger = ('fraction', 0.85)
  keep    = ('fraction', 0.1)
  (model profile max_input_tokens = 1000000)

两个模型的窗口都是 100 万 token,85% 就是 85 万 token 才触发摘要。这个数字很关键:日常任务根本碰不到它。你如果在小任务里看到摘要被触发,说明有别的问题(比如某个工具在疯狂返回内容)。

要自己指定阈值,trigger 和 keep 都接受三种形式:

from deepagents.middleware.summarization import SummarizationMiddleware

SummarizationMiddleware(
    model=model,
    backend=backend,
    trigger=("fraction", 0.70),    # 窗口的 70% 就触发(比默认更早)
    keep=("tokens", 20000),        # 保留最近 20000 token
)
# trigger / keep 都支持:("fraction", 0.0~1.0) / ("tokens", int) / ("messages", int)

6.2. 双轨:state 全量、模型收到压缩版 #

这是本章最重要的一个机制,也是最容易误解的地方。把 trigger 调到很小,观察每一轮里模型实际收到什么和state 里有什么:

"""零 token:摘要到底改写了什么。"""
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langgraph.checkpoint.memory import InMemorySaver

from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.middleware.summarization import SummarizationMiddleware

CALLS = []
N = {"i": 0}


class Spy(BaseChatModel):
    """记录每次模型调用实际收到的消息列表。"""

    @property
    def _llm_type(self) -> str:
        return "spy"

    def bind_tools(self, tools, **kw):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kw) -> ChatResult:
        N["i"] += 1
        # 摘要请求的特征:消息里含摘要提示词的标记
        is_sum = any("Messages to summarize" in str(m.content) for m in messages)
        CALLS.append({
            "is_sum": is_sum,
            "items": [(type(m).__name__, str(m.content)[:52]) for m in messages],
        })
        if is_sum:
            # 假装模型产出了一份四段式摘要
            return ChatResult(generations=[ChatGeneration(message=AIMessage(
                content="## SESSION INTENT\n用户在连续提问。\n\n"
                        "## SUMMARY\n已回答前几个问题。\n\n"
                        "## ARTIFACTS\nNone\n\n"
                        "## NEXT STEPS\n继续回答。"))])
        return ChatResult(generations=[ChatGeneration(
            message=AIMessage(content=f"回复{N['i']}"))])


be = StateBackend()
model = Spy()
agent = create_deep_agent(
    model=model,
    backend=be,
    middleware=[SummarizationMiddleware(
        model=model, backend=be,
        trigger=("messages", 4),    # 4 条消息就触发,方便观察
        keep=("messages", 2),       # 只保留最近 2 条
    )],
    checkpointer=InMemorySaver(),
)

cfg = {"configurable": {"thread_id": "s"}}
for k in range(1, 6):
    CALLS.clear()
    out = agent.invoke({"messages": [{"role": "user", "content": f"问题{k}"}]}, cfg)
    print(f"\n第 {k} 轮 · state 里 {len(out['messages'])} 条消息")
    for rec in CALLS:
        print(f"  [{'摘要调用' if rec['is_sum'] else '正常推理'}] 模型收到 {len(rec['items'])} 条:")
        for kind, body in rec["items"]:
            print(f"     {kind:14s} {body!r}")

关键的第 3 轮开始:

第 3 轮 · state 里 6 条消息
  [摘要调用] 模型收到 1 条:
     HumanMessage   '<role>\nContext Extraction Assistant\n</role>...'
  [正常推理] 模型收到 4 条:
     SystemMessage  ''
     HumanMessage   'You are in the middle of a conversation that has bee'
     AIMessage      '回复2'
     HumanMessage   '问题3'

第 5 轮 · state 里 10 条消息
  [摘要调用] 模型收到 1 条:
     HumanMessage   '<role>\nContext Extraction Assistant\n</role>...'
  [正常推理] 模型收到 4 条:
     SystemMessage  ''
     HumanMessage   'You are in the middle of a conversation that has bee'
     AIMessage      '回复6'
     HumanMessage   '问题5'

注意这个对比:

state 里的完整历史: 10 条
  ['问题1', '回复1', '问题2', '回复2', '问题3', '回复4', '问题4', '回复6', '问题5', '回复8']

模型最后一次实际收到: 4 条
  SystemMessage  ''
  HumanMessage   'You are in the middle of a conversation that has bee...'
  AIMessage      '回复6'
  HumanMessage   '问题5'

state 里的历史从 2 条一路长到 10 条,模型每次实际收到的始终是 4 条。

这就是双轨机制:

摘要不修改 state,它只改写「这一次发给模型的消息列表」。

框架内部是通过 request.override(messages=truncated_messages) 做的——只影响这一次调用,不产生状态变更。

这个设计有几个直接后果,都很重要:

  1. result["messages"] 永远是完整历史。 你的前端拿到的是全量对话,用户看不出被压缩过
  2. checkpointer 里存的是全量。 恢复会话时历史不丢
  3. 摘要不是一次性的。 每轮都重新算一遍,所以上面第 3、4、5 轮各触发了一次
  4. 调试时别看 result["messages"] 判断上下文大小,那是全量的;要看模型实际收到了什么

6.3. 摘要长什么样 #

摘要不是简单裁剪,而是让模型产出一份结构化的四段式提取。框架的摘要提示词要求这四段:

## SESSION INTENT
用户的主要目标是什么?整个会话想完成什么任务?

## SUMMARY
提取对话历史里最重要的上下文。包括重要选择、结论、策略,以及关键决策背后的理由。
记录被否决的方案和否决原因。

## ARTIFACTS
本次对话创建、修改、访问了哪些产物、文件或资源?文件修改要列出具体路径和改了什么。
这一节防止产物信息的静默丢失。

## NEXT STEPS
为达成会话目标还剩哪些具体任务?下一步该做什么?

这四段的设计意图很清楚:SESSION INTENT 防止跑偏、ARTIFACTS 防止忘记自己写过什么文件、NEXT STEPS 防止重复已完成的工作。提示词里还专门交代「确保不要重复你已经完成的动作」。

摘要产出后,被包装成一条 HumanMessage 塞进消息列表。完整原文(360 字符):

You are in the middle of a conversation that has been summarized.

The full conversation history has been saved to
/conversation_history/session_125d0a6a8cca4279852e0e74b46effa9.md
should you need to refer back to it for details.

A condensed summary follows:

<summary>
## SESSION INTENT
测试。

## SUMMARY
已问答数轮。

## ARTIFACTS
None

## NEXT STEPS
继续。
</summary>

关键在第二段:它明确告诉模型完整历史存在哪个文件。所以摘要不是有损丢弃,是可回溯的压缩——模型发现摘要里信息不够时,可以 read_file 或 grep 那个文件把细节捞回来。

6.4. 文件系统里的规范记录 #

那个文件长这样:

/conversation_history/session_c21040e5e2f249b2b15ef2cbdd1f46f5.md(499 字符)

## Summarized at 2026-09-02T19:24:04.374890+00:00

<message type="human">问题1</message>
<message type="ai">回复1</message>
<message type="human">问题2</message>

## Summarized at 2026-09-02T19:24:04.380536+00:00

<message type="ai">回复2</message>
<message type="human">问题3</message>

## Summarized at 2026-09-02T19:24:04.385351+00:00

<message type="ai">回复4</message>
<message type="human">问题4</message>

几个细节:

所以这个文件是「被压缩掉的内容」的归档。加上 state 里的全量历史,同一份对话实际存了三份:state 全量、模型收到的压缩版、磁盘上的归档。用 StateBackend 时后者也在 state 里,会让 checkpoint 变大——长会话要留意存储成本。

6.5. 兜底与流式过滤 #

除了阈值触发,还有一条兜底路径:

任何模型调用抛出 ContextOverflowError 时,框架立即回退到摘要,然后用「摘要 + 保留的最近消息」重试。

也就是说即使阈值判断失灵(比如 token 估算不准),也不会直接失败。这对没有 profile 的模型尤其有用。

最后一个实用细节:摘要步骤产生的 token 会混进你的流式输出。因为摘要也是一次模型调用。要过滤掉:

for chunk in agent.stream(
    {"messages": [...]},
    stream_mode="messages",
    version="v2",
):
    token, metadata = chunk["data"]
    # 摘要步骤的 token 带这个标记,前端不该展示给用户
    if metadata.get("lc_source") == "summarization":
        continue
    ...

不过滤的话,用户会在界面上看到一段莫名其妙的 ## SESSION INTENT。这个坑第 49 章讲事件流时还会碰到。

7. 按需压缩:compact_conversation #

自动摘要在阈值处触发,但阈值处不一定是好的压缩时机——可能正卡在一个任务中间。

更好的时机是任务之间的间隙。为此可以给模型一个工具,让它自己决定什么时候压缩:

"""给模型一个手动压缩的工具。"""
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.middleware.summarization import create_summarization_tool_middleware

backend = StateBackend()
model = "deepseek:deepseek-v4-flash"

agent = create_deep_agent(
    model=model,
    backend=backend,
    middleware=[create_summarization_tool_middleware(model, backend)],
)

工具表的变化:

不加时(8 个): ['ls', 'read_file', 'write_file', 'edit_file', 'delete', 'glob', 'grep', 'task']

加了之后(9 个): ['ls', 'read_file', 'write_file', 'edit_file', 'delete', 'glob',
                   'grep', 'task', 'compact_conversation']

三个要点:

  1. 加了它不会关掉自动摘要。 两者共享同一个摘要引擎和状态,是互补而非替代
  2. 自定义 middleware 插在 PatchToolCallsMiddleware 之后(装配顺序见第 50 章)
  3. 它给模型多了一个工具,也就多了一点 schema 开销和一个可能被误用的动作

什么时候值得加?长流程、多阶段的任务——比如「调研 5 个主题然后写报告」,让模型在每个主题之间压一次,比等到 85% 再压要干净得多。短任务不用加。

8. 提示缓存 #

前面算的所有基线开销,有一个便宜得多的解法:提示缓存。系统提示词、工具 schema 这些每轮都一样的内容,如果供应商支持缓存,就不用每次全价重算。

Deep Agents 的做法是:默认就装上,不用你配。

"""零 token:看框架默认装了哪些提示缓存中间件。"""
import inspect

from deepagents.middleware._prompt_caching import (
    AnthropicPromptCachingMiddleware, append_prompt_caching_middleware)

for n, p in inspect.signature(AnthropicPromptCachingMiddleware.__init__).parameters.items():
    if n != "self":
        d = "必填" if p.default is inspect.Parameter.empty else repr(p.default)
        print(f"{n:28s} 默认={d}")

lst = []
append_prompt_caching_middleware(lst)
for m in lst:
    print(f"\n实际装载: {type(m).__name__}")
    for a in ("ttl", "type", "min_messages_to_cache", "unsupported_model_behavior"):
        if hasattr(m, a):
            print(f"   {a} = {getattr(m, a)}")
type                         默认='ephemeral'
ttl                          默认='5m'
min_messages_to_cache        默认=0
unsupported_model_behavior   默认='warn'

实际装载: AnthropicPromptCachingMiddleware
   ttl = 5m
   type = ephemeral
   min_messages_to_cache = 0
   unsupported_model_behavior = ignore

几个要点:

  1. AnthropicPromptCachingMiddleware 是无条件装载的,对非 Anthropic 模型静默失效(no-op)
  2. 默认 TTL 是 5 分钟('5m'),可选 '1h'
  3. 框架把 unsupported_model_behavior 从默认的 'warn' 改成了 'ignore'。 这是必要的——既然无条件装载,对其它模型就不该刷警告
  4. BedrockPromptCachingMiddleware 和 FireworksPromptCachingMiddleware 只在装了对应的包时才追加(langchain-aws / langchain-fireworks)

第 3 点有个实际含义:你用非 Anthropic 模型时,缓存中间件在那里但什么都不做,而且不会告诉你。 别以为默认开着就等于所有模型都省钱了。

那个 5 分钟 TTL 也值得琢磨:

后两种情况可以考虑显式装一个 TTL 更长的实例:

from deepagents.middleware._prompt_caching import AnthropicPromptCachingMiddleware

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    # 显式装一个 1 小时 TTL 的(1h 通常比 5m 单价更高,要算一下是否划算)
    middleware=[AnthropicPromptCachingMiddleware(ttl="1h")],
)

顺带一个第 45 章留下的伏笔:MemoryMiddleware 有个 add_cache_control 参数(默认 False)。开了它,那 5245 字符的记忆说明书就能进缓存段——记忆内容基本不变,是缓存的理想对象。

另外,从一次异常的调用栈里能看到 middleware 的实际嵌套顺序,正好印证了装配关系:

filesystem.py    wrap_model_call
  └─ subagents.py    wrap_model_call
      └─ summarization.py    wrap_model_call
          │   return handler(request.override(messages=truncated_messages))
          └─ prompt_caching.py    wrap_model_call
              └─ 真正的模型调用

提示缓存在最内层,也就是最靠近模型的一层——它要给「摘要之后的最终消息列表」打缓存标记,顺序上必须在摘要之后。

9. 长任务配置清单 #

把前面所有东西汇总成一份可以直接抄的清单。

默认就有的(什么都不用做):

机制 阈值 行为
工具结果卸载 20000 token 存到 /large_tool_results/<id>,留头尾各 5 行预览
工具输入截断 20000 token 且上下文 85% 旧工具调用换成文件指针
自动摘要 窗口 85%(无 profile 时 17 万 token) 四段式摘要 + 归档到 /conversation_history/
溢出兜底 ContextOverflowError 立即摘要并重试
Anthropic 提示缓存 — TTL 5 分钟,非 Anthropic 模型静默失效

按需要加的:

"""长任务的推荐配置。"""
from deepagents import (HarnessProfile, create_deep_agent,
                        register_harness_profile)
from deepagents.backends import FilesystemBackend
from deepagents.middleware.summarization import create_summarization_tool_middleware

MODEL = "deepseek:deepseek-v4-flash"

# 1. 砍掉用不到的内置工具,缩小每轮基线(本例:只读智能体)
register_harness_profile(
    "deepseek",
    HarnessProfile(excluded_tools=frozenset(["write_file", "edit_file", "delete"])),
)

# 2. 用落盘后端,让卸载的内容真正离开 state(避免 checkpoint 膨胀)
backend = FilesystemBackend(root_dir="./workspace", virtual_mode=True)

agent = create_deep_agent(
    model=MODEL,
    backend=backend,
    middleware=[
        # 3. 给模型手动压缩的能力,让它在任务间隙自己压
        create_summarization_tool_middleware(MODEL, backend),
    ],
    # 4. 重活派给子智能体,主上下文只收结果
    subagents=[
        {
            "name": "researcher",
            "description": "深度调研单个主题",
            # 明确要求产出摘要而非原始材料,这是控制回传体积的关键
            "system_prompt": "你是调研员。调研完成后只回一份不超过 500 字的"
                             "结论摘要,原始材料写进文件并在摘要里给出路径。",
        },
    ],
)

四条判断原则:

  1. 先砍基线,再谈压缩。 excluded_tools 是纯收益的配置,压缩是有代价的补救
  2. 重活派出去。 子智能体隔离是最有效的手段——主智能体连那几十次工具调用都看不到,不存在压缩问题
  3. 约束子智能体的输出体积。 官方专门提了这条:调试时发现子智能体输出很长,就在它的 system_prompt 里要求产出综合结论而非原始材料
  4. 用文件系统当外部内存。 大产出写文件,上下文里只留路径;需要细节时 grep 回查

官方给的最佳实践还有两条值得抄:

10. 实战约定与坑 #

  1. 卸载和摘要默认就开着,create_deep_agent 自带,不需要加 middleware(§5、§6)
  2. 卸载阈值是 20000 token。 19000 原样、25000 被卸载,实测确认(§5.1)
  3. 卸载后的预览是头 5 行 + 尾 5 行,不是官方文档说的「前 10 行」。头尾预览比只看开头有用得多(§5.2)
  4. 卸载文件路径是 /large_tool_results/<tool_call_id>,内容一字不丢,可 read_file / grep 回查
  5. 摘要不修改 state。 它只改写这一次发给模型的消息列表;result["messages"] 永远是全量(§6.2)
  6. 别用 result["messages"] 判断上下文大小——那是全量历史,不是模型实际收到的
  7. 摘要默认在窗口 85% 触发、保留 10%。 100 万窗口的模型就是 85 万 token,日常任务碰不到(§6.1)
  8. 模型没有 profile 时走保守回退:17 万 token 触发、只留 6 条消息。 用自定义模型类要留意这条(§6.1)
  9. 摘要是可回溯的:包装文本里明确写了完整历史的文件路径,模型能自己捞回细节(§6.3)
  10. 同一份对话实际存了三份:state 全量、模型收到的压缩版、/conversation_history/ 归档。用 StateBackend 时归档也在 state 里,长会话要留意 checkpoint 体积(§6.4)
  11. 流式输出要过滤摘要 token,否则用户会看到莫名的 ## SESSION INTENT。判据是 metadata.get("lc_source") == "summarization"(§6.5)
  12. ContextOverflowError 会自动回退到摘要重试,阈值失灵也不会直接失败(§6.5)
  13. 8 个内置工具每轮 10894 字符 schema。 这笔开销不在系统提示词里,最容易被忽略(§4)
  14. excluded_tools 要通过 register_harness_profile 注册,不是 create_deep_agent 的参数。排除三个写工具省 24%(§4)
  15. 提示缓存默认装载但只对 Anthropic 生效,其它模型静默无效(unsupported_model_behavior 被设成 ignore),别误以为所有模型都在省钱(§8)
  16. 默认 TTL 只有 5 分钟。 批处理、定时任务、或有长工具调用的场景基本命中不了,需要显式配 ttl="1h"(§8)
  17. 内置压缩不处理图片:不缩放、不降分辨率、不做视觉嵌入(§5.3)
  18. 运行时上下文不会自动进提示词。 API key 这类东西走 context=,让工具从 runtime 读(§2)
  19. 子智能体是最有效的上下文手段,但要在它的 system_prompt 里明确约束回传体积(§9)

11. 练习 #

  1. 量你自己的基线。 用 §4 那个假模型,量出你的项目实际的工具 schema 开销(含自定义工具)。加上 memory= 和 skills= 后再量一次,算出「业务内容还没进去就花掉多少」。

  2. 卡在阈值上。 造一个工具,分别返回 19000 和 21000 token 的内容,确认前者原样、后者被卸载。然后打印卸载后的完整替换文本,数一下预览了几行。

  3. 观察双轨。 按 §6.2 把 trigger 设成 ("messages", 4),跑 5 轮。每轮打印「state 消息数」和「模型实际收到的消息数」,确认前者增长、后者稳定。

  4. 回查归档。 摘要触发后,用 read_file 读 /conversation_history/session_*.md,确认里面是被压缩掉的那部分原文。再让模型回答一个只有归档里才有的细节,看它会不会自己去读文件。

  5. 测回退路径。 用一个自定义模型类(不带 profile),调 compute_summarization_defaults(),确认拿到的是 ("tokens", 170000) 和 ("messages", 6)。想一下如果你的模型窗口只有 3 万 token 会发生什么。

  6. 手动压缩。 加上 compact_conversation 工具,给模型一个三阶段任务,在提示里要求它「每完成一个阶段就压缩一次对话」,从轨迹里确认它调用了这个工具。

  7. (选做)缓存对比。 用 Anthropic 模型跑同一个多轮任务两次:一次默认配置、一次显式 AnthropicPromptCachingMiddleware(ttl="1h"),并在两轮之间 sleep 6 分钟。对比 usage 里的缓存命中数据。

12. 本章小结 #

  1. 五类上下文:输入、运行时、压缩、隔离、长期记忆。本章主攻压缩,顺带补上运行时上下文(它不会自动进提示词)。
  2. 系统提示词由 8 段拼成,你的 system_prompt 在最前面,后面几段会补充大量工具说明——不用自己重复解释工具怎么用。
  3. 三笔基线开销:工具 schema 10894 字符、记忆说明 5245 字符、技能说明 2011 字符。全开约 4500 token,业务内容还没进去。
  4. excluded_tools 是纯收益的配置(排除三个写工具省 24%),通过 register_harness_profile 注册,不是 create_deep_agent 的参数。
  5. 卸载和摘要默认就在工作,不需要任何配置。
  6. 卸载阈值 20000 token,替换成固定 1051 字符的引用,压缩比 99% 以上,内容完整存在 /large_tool_results/<tool_call_id>。
  7. 预览是头 5 行 + 尾 5 行(官方文档写的是「前 10 行」,实测不符)。头尾预览让模型能判断该读哪一段。
  8. 摘要不改 state,只通过 request.override(messages=...) 改写这一次的消息列表。所以 result["messages"] 是全量、模型收到的是压缩版。
  9. 默认 85% 触发、保留 10%;模型没 profile 时回退到 17 万 token / 6 条消息这个更保守的配置。
  10. 摘要是四段式结构化提取:SESSION INTENT / SUMMARY / ARTIFACTS / NEXT STEPS,分别防跑偏、防丢失理由、防忘记产物、防重复劳动。
  11. 摘要可回溯:包装文本里写明了完整历史的归档路径,模型可以 read_file 捞回细节。
  12. 归档在 /conversation_history/session_<id>.md,每次摘要追加一段带 UTC 时间戳,只记被压掉的部分。
  13. 流式输出要按 lc_source == "summarization" 过滤摘要 token,否则会漏给用户看。
  14. compact_conversation 让模型自己选压缩时机,适合多阶段长任务;它不会关掉自动摘要,两者共享同一引擎。
  15. 提示缓存默认装载、TTL 5 分钟、只对 Anthropic 生效,对其它模型静默无效(unsupported_model_behavior='ignore')。批处理场景要显式配长 TTL。
  16. 压缩不处理图片,多模态内容要单独考虑。
  17. 最有效的上下文手段是子智能体隔离,但必须在它的 system_prompt 里约束回传体积——否则隔离出来的重活会以另一种形式回到主上下文。