LangChain 和 Deep Agents 为常见用例提供预置 middleware。每个 middleware 都可用于生产环境,并且可以根据你的具体需求配置。

Provider-agnostic middleware

以下 middleware 可配合任意 LLM 提供商使用:
Middleware描述
Summarization接近 token 限制时自动总结对话历史。
Human-in-the-loop暂停执行,等待人工批准工具调用。
Model call limit限制模型调用次数,避免成本过高。
Tool call limit通过限制调用次数来控制工具执行。
Model fallback主模型失败时自动回退到备用模型。
PII detection检测并处理个人身份信息 (PII)。
To-do list为 agent 提供任务规划和跟踪能力。
LLM tool selector调用主模型前,使用 LLM 选择相关工具。
Tool retry使用指数退避自动重试失败的工具调用。
Model retry使用指数退避自动重试失败的模型调用。
LLM tool emulator使用 LLM 模拟工具执行,便于测试。
Context editing通过修剪或清除工具使用来管理对话上下文。
Shell tool向 agent 暴露持久 shell 会话,用于执行命令。
File search为文件系统文件提供 Glob 和 Grep 搜索工具。
Filesystem为 agent 提供用于存储上下文和长期记忆的文件系统。
Subagent添加生成 subagent 的能力。

Summarization

接近 token 限制时自动总结对话历史,保留近期消息并压缩较早的上下文。Summarization 适用于以下场景:
  • 超出上下文窗口的长时间运行对话。
  • 包含大量历史记录的多轮对话。
  • 需要保留完整对话上下文的应用。
Summarization 是面向文本的上下文压缩。它不会调整图片、音频或视频 payload 的大小、对其降采样,或以其他方式压缩这些 payload。keep 保留的近期消息仍包含原始多模态块,而被总结的较早多模态消息只由生成的文本摘要表示。对于大量使用图片的应用,请将媒体存储在文件系统或对象存储中,并通过消息历史传递 URL 或文件引用。
API reference: SummarizationMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
)
使用 langchain>=1.1 时,triggerkeepfraction 条件(如下所示)依赖聊天模型的 profile data。如果数据不可用,请使用其他条件或手动指定:
from langchain.chat_models import init_chat_model

custom_profile = {
    "max_input_tokens": 100_000,
    # ...
}
model = init_chat_model("gpt-5.4", profile=custom_profile)
model
string | BaseChatModel
required
用于生成摘要的模型。可以是模型标识符字符串(例如 'openai:gpt-5.4-mini'),也可以是 BaseChatModel 实例。更多信息,请参阅 init_chat_model
trigger
ContextSize | list[ContextSize] | None
触发总结的条件。可以是:
  • 单个 ContextSize 元组(必须满足指定条件)
  • ContextSize 元组列表(满足任一条件即可,即 OR 逻辑)
条件应为以下之一:
  • fraction (float):模型上下文大小的比例 (0-1)
  • tokens (int):绝对 token 数
  • messages (int):消息数
必须至少指定一个条件。如果未提供,summarization 不会自动触发。更多信息,请参阅 ContextSize 的 API reference。
keep
ContextSize
default:"('messages', 20)"
总结后要保留多少上下文。请仅指定以下一个条件:
  • fraction (float):要保留的模型上下文大小比例 (0-1)
  • tokens (int):要保留的绝对 token 数
  • messages (int):要保留的近期消息数量
更多信息,请参阅 ContextSize 的 API reference。
token_counter
function
自定义 token 计数函数。默认使用基于字符的计数。
summary_prompt
string
用于总结的自定义 prompt 模板。如果未指定,则使用内置模板。模板应包含 {messages} 占位符,系统会在此插入对话历史。
trim_tokens_to_summarize
number
default:"4000"
生成摘要时包含的最大 token 数。总结前会修剪消息,使其符合此限制。
summary_prefix
string
deprecated
已弃用: 请改用 summary_prompt 提供完整 prompt。
max_tokens_before_summary
number
deprecated
已弃用: 请改用 trigger: ("tokens", value)。这是触发 summarization 的 token 阈值。
messages_to_keep
number
deprecated
已弃用: 请改用 keep: ("messages", value)。这是要保留的近期消息数量。
Summarization middleware 会监控消息 token 数,并在达到阈值时自动总结较早消息。触发条件 控制何时运行 summarization:
  • 单个条件对象(必须满足指定条件)
  • 条件数组(满足任一条件即可,即 OR 逻辑)
  • 每个条件都可以使用 fraction(模型上下文大小比例)、tokens(绝对数量)或 messages(消息数量)
保留条件 控制要保留多少上下文(仅指定一个):
  • fraction:要保留的模型上下文大小比例
  • tokens:要保留的绝对 token 数
  • messages:要保留的近期消息数量
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware


# 单个条件:tokens >= 4000 时触发
agent = create_agent(
    model="gpt-5.4",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
)

# 多个条件:tokens >= 3000 或 messages >= 6 时触发
agent2 = create_agent(
    model="gpt-5.4",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=[
                ("tokens", 3000),
                ("messages", 6),
            ],
            keep=("messages", 20),
        ),
    ],
)

# 使用比例限制
agent3 = create_agent(
    model="gpt-5.4",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("fraction", 0.8),
            keep=("fraction", 0.3),
        ),
    ],
)

Human-in-the-loop

在工具调用执行前暂停 agent 执行,以便人工批准、编辑或拒绝工具调用。Human-in-the-loop 适用于以下场景:
  • 需要人工批准的高风险操作(例如数据库写入、金融交易)。
  • 必须由人工监督的合规工作流。
  • 由人工反馈指导 agent 的长时间运行对话。
API reference: HumanInTheLoopMiddleware
Human-in-the-loop middleware 需要 checkpointer,以便在中断之间维护状态。
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver


def your_read_email_tool(email_id: str) -> str:
    """按 ID 读取电子邮件的模拟函数。"""
    return f"Email content for ID: {email_id}"

def your_send_email_tool(recipient: str, subject: str, body: str) -> str:
    """发送电子邮件的模拟函数。"""
    return f"Email sent to {recipient} with subject '{subject}'"

agent = create_agent(
    model="gpt-5.4",
    tools=[your_read_email_tool, your_send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "your_send_email_tool": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "your_read_email_tool": False,
            }
        ),
    ],
)
如需完整示例、配置选项和集成模式,请参阅 Human-in-the-loop 文档
观看这个视频指南,了解 Human-in-the-loop middleware 的行为。

Model call limit

限制模型调用次数,防止无限循环或成本过高。Model call limit 适用于以下场景:
  • 防止失控的 agent 发起过多 API 调用。
  • 在生产部署中执行成本控制。
  • 在特定调用预算内测试 agent 行为。
API reference: ModelCallLimitMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="gpt-5.4",
    checkpointer=InMemorySaver(),  # thread 限制需要此项
    tools=[],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,
            run_limit=5,
            exit_behavior="end",
        ),
    ],
)
观看这个视频指南,了解 Model Call Limit middleware 的行为。
thread_limit
number
一个 thread 中所有运行的最大模型调用次数。默认不限制。
run_limit
number
单次调用的最大模型调用次数。默认不限制。
exit_behavior
string
default:"end"
达到限制时的行为。选项:'end'(优雅终止)或 'error'(引发异常)

Tool call limit

通过限制工具调用次数来控制 agent 执行,可以对所有工具全局限制,也可以针对特定工具限制。Tool call limit 适用于以下场景:
  • 防止过度调用成本较高的外部 API。
  • 限制 web 搜索或数据库查询。
  • 对特定工具的使用执行速率限制。
  • 防止失控的 agent 循环。
API reference: ToolCallLimitMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool, database_tool],
    middleware=[
        # 全局限制
        ToolCallLimitMiddleware(thread_limit=20, run_limit=10),
        # 工具专属限制
        ToolCallLimitMiddleware(
            tool_name="search",
            thread_limit=5,
            run_limit=3,
        ),
    ],
)
观看这个视频指南,了解 Tool Call Limit middleware 的行为。
tool_name
string
要限制的特定工具名称。如果未提供,限制将全局应用于所有工具
thread_limit
number
一个 thread(对话)中所有运行的最大工具调用次数。它会在相同 thread ID 的多次调用之间保留。需要 checkpointer 维护状态。None 表示不设置 thread 限制。
run_limit
number
单次调用(一次用户消息到响应的循环)的最大工具调用次数。每条新用户消息都会重置。None 表示不设置运行限制。注意: 必须至少指定 thread_limitrun_limit 之一。
exit_behavior
string
default:"continue"
达到限制时的行为:
  • 'continue'(默认):用错误消息阻止超限工具调用,让其他工具和模型继续。模型会根据错误消息决定何时结束。
  • 'error':引发 ToolCallLimitExceededError 异常,立即停止执行
  • 'end':用超限工具调用对应的 ToolMessage 和 AI message 立即停止执行。仅在限制单个工具时有效;如果其他工具仍有待处理调用,则引发 NotImplementedError
使用以下方式指定限制:
  • Thread limit:一个对话中所有运行的最大调用次数(需要 checkpointer)
  • Run limit:单次调用的最大调用次数(每轮重置)
退出行为:
  • 'continue'(默认):用错误消息阻止超限调用,agent 继续运行
  • 'error':立即引发异常
  • 'end':用 ToolMessage + AI message 停止(仅限单工具场景)
from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware


global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)
search_limiter = ToolCallLimitMiddleware(tool_name="search", thread_limit=5, run_limit=3)
database_limiter = ToolCallLimitMiddleware(tool_name="query_database", thread_limit=10)
strict_limiter = ToolCallLimitMiddleware(tool_name="scrape_webpage", run_limit=2, exit_behavior="error")

agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool, database_tool, scraper_tool],
    middleware=[global_limiter, search_limiter, database_limiter, strict_limiter],
)

Model fallback

主模型失败时自动回退到备用模型。Model fallback 适用于以下场景:
  • 构建能处理模型中断的弹性 agent。
  • 通过回退到更便宜的模型来优化成本。
  • 在 OpenAI、Anthropic 等提供商之间提供冗余。
API reference: ModelFallbackMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-5.4-mini",
            "claude-3-5-sonnet-20241022",
        ),
    ],
)
观看这个视频指南,了解 Model Fallback middleware 的行为。
first_model
string | BaseChatModel
required
主模型失败时首先尝试的 fallback 模型。可以是模型标识符字符串(例如 'openai:gpt-5.4-mini'),也可以是 BaseChatModel 实例。
*additional_models
string | BaseChatModel
当前面的模型失败时,按顺序尝试的其他 fallback 模型

PII detection

使用可配置策略检测并处理对话中的个人身份信息 (PII)。PII detection 适用于以下场景:
  • 有合规要求的医疗和金融应用。
  • 需要清理日志的客服 agent。
  • 任何处理敏感用户数据的应用。
使用 apply_to_output=True 时,PIIMiddleware 还会通过注册的 stream transformer 编辑 streamed wire output,包括文本增量、工具调用参数、工具输出和状态快照。需要 langchain>=1.3.2。请参阅在 middleware 上注册 transformer
API reference: PIIMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
    ],
)

Custom PII types

你可以通过提供 detector 参数创建自定义 PII 类型。这样可以检测内置类型之外、与你的用例相关的模式。 创建自定义 detector 的三种方式:
  1. Regex pattern string:简单模式匹配
  2. Custom function:包含验证的复杂检测逻辑
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
import re


# 方法 1:regex pattern string
agent1 = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",
        ),
    ],
)

# 方法 2:已编译的 regex pattern
agent2 = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        PIIMiddleware(
            "phone_number",
            detector=re.compile(r"\+?\d{1,3}[\s.-]?\d{3,4}[\s.-]?\d{4}"),
            strategy="mask",
        ),
    ],
)

# 方法 3:自定义 detector 函数
def detect_ssn(content: str) -> list[dict[str, str | int]]:
    """检测 SSN,并进行验证。

    返回包含 'text'、'start' 和 'end' 键的字典列表。
    """
    import re
    matches = []
    pattern = r"\d{3}-\d{2}-\d{4}"
    for match in re.finditer(pattern, content):
        ssn = match.group(0)
        # 验证:前 3 位不应为 000、666 或 900-999
        first_three = int(ssn[:3])
        if first_three not in [0, 666] and not (900 <= first_three <= 999):
            matches.append({
                "text": ssn,
                "start": match.start(),
                "end": match.end(),
            })
    return matches

agent3 = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        PIIMiddleware(
            "ssn",
            detector=detect_ssn,
            strategy="hash",
        ),
    ],
)
自定义 detector 函数签名: detector 函数必须接受字符串(content)并返回匹配项: 返回包含 textstartend 键的字典列表:
def detector(content: str) -> list[dict[str, str | int]]:
    return [
        {"text": "matched_text", "start": 0, "end": 12},
        # ... 更多匹配项
    ]
对于自定义 detector:
  • 对简单模式使用 regex 字符串
  • 需要 flag(例如不区分大小写匹配)时使用 RegExp 对象
  • 需要模式匹配之外的验证逻辑时使用自定义函数
  • 自定义函数可让你完全控制检测逻辑,并实现复杂的验证规则
pii_type
string
required
要检测的 PII 类型。可以是内置类型(emailcredit_cardipmac_addressurl),也可以是自定义类型名称。
strategy
string
default:"redact"
如何处理检测到的 PII。选项:
  • 'block':检测到时引发异常
  • 'redact':替换为 [REDACTED_{PII_TYPE}]
  • 'mask':部分遮盖(例如 ****-****-****-1234
  • 'hash':替换为确定性 hash
detector
function | regex
自定义 detector 函数或 regex pattern。如果未提供,则使用该 PII 类型的内置 detector。
apply_to_input
boolean
default:"True"
在模型调用前检查用户消息
apply_to_output
boolean
default:"False"
在模型调用后检查 AI messages。使用 langchain>=1.3.2 时,还会通过注册的 stream transformer 编辑 streamed wire output(文本增量、工具调用参数、工具输出、状态快照)。请参阅 event streaming
apply_to_tool_results
boolean
default:"False"
执行后检查工具结果消息

To-do list

为 agent 提供面向复杂多步骤任务的任务规划和跟踪能力。To-do list 适用于以下场景:
  • 需要协调多个工具的复杂多步骤任务。
  • 进度可见性很重要的长时间运行操作。
此 middleware 会自动为 agent 提供 write_todos 工具和 system prompts,以指导有效的任务规划。
API reference: TodoListMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[read_file, write_file, run_tests],
    middleware=[TodoListMiddleware()],
)
观看这个视频指南,了解 To-do List middleware 的行为。
system_prompt
string
用于指导 todo 使用的自定义 system prompt。如果未指定,则使用内置 prompt。
tool_description
string
write_todos 工具的自定义描述。如果未指定,则使用内置描述。

LLM tool selector

在调用主模型前,使用 LLM 智能选择相关工具。LLM tool selector 适用于以下场景:
  • 拥有大量工具(10+)且每个查询通常只需要其中少数工具的 agent。
  • 通过过滤无关工具减少 token 使用量。
  • 提升模型专注度和准确性。
此 middleware 使用结构化输出询问 LLM 哪些工具与当前查询最相关。结构化输出 schema 定义可用的工具名称和描述。模型提供商通常会在后台将这些结构化输出信息添加到 system prompt 中。 API reference: LLMToolSelectorMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[tool1, tool2, tool3, tool4, tool5, ...],
    middleware=[
        LLMToolSelectorMiddleware(
            model="gpt-5.4-mini",
            max_tools=3,
            always_include=["search"],
        ),
    ],
)
model
string | BaseChatModel
用于选择工具的模型。可以是模型标识符字符串(例如 'openai:gpt-5.4-mini'),也可以是 BaseChatModel 实例。更多信息,请参阅 init_chat_model默认使用 agent 的主模型。
system_prompt
string
给选择模型的指令。如果未指定,则使用内置 prompt。
max_tools
number
可选择的最大工具数。如果模型选择了更多工具,则只使用前 max_tools 个。如果未指定,则不限制。
always_include
list[string]
无论选择结果如何都始终包含的工具名称。这些工具不计入 max_tools 限制。

Tool retry

使用可配置的指数退避自动重试失败的工具调用。Tool retry 适用于以下场景:
  • 处理外部 API 调用中的瞬时失败。
  • 提升依赖网络的工具的可靠性。
  • 构建能优雅处理临时错误的弹性 agent。
API reference: ToolRetryMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool, database_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
        ),
    ],
)
max_retries
number
default:"2"
初始调用后的最大重试次数(默认总共尝试 3 次)
tools
list[BaseTool | str]
可选的工具或工具名称列表,用于指定应用重试逻辑的工具。如果为 None,则应用于所有工具。
retry_on
tuple[type[Exception], ...] | callable
default:"(Exception,)"
可以是要重试的异常类型元组,也可以是接收异常并在应重试时返回 True 的 callable。
on_failure
string | callable
default:"return_message"
所有重试都耗尽时的行为。选项:
  • 'return_message':返回包含错误详情的 ToolMessage(允许 LLM 处理失败)
  • 'raise':重新引发异常(停止 agent 执行)
  • 自定义 callable:接收异常并返回 ToolMessage 内容字符串的函数
backoff_factor
number
default:"2.0"
指数退避的乘数。每次重试等待 initial_delay * (backoff_factor ** retry_number) 秒。设置为 0.0 可使用固定延迟。
initial_delay
number
default:"1.0"
第一次重试前的初始延迟,单位为秒
max_delay
number
default:"60.0"
两次重试之间的最大延迟,单位为秒(限制指数退避增长)
jitter
boolean
default:"true"
是否向延迟添加随机 jitter(±25%),以避免 thundering herd
该 middleware 会使用指数退避自动重试失败的工具调用。关键配置:
  • max_retries:重试次数(默认:2)
  • backoff_factor:指数退避乘数(默认:2.0)
  • initial_delay:起始延迟,单位为秒(默认:1.0)
  • max_delay:延迟增长上限(默认:60.0)
  • jitter:添加随机变化(默认:True)
失败处理:
  • on_failure='return_message':返回错误消息
  • on_failure='raise':重新引发异常
  • 自定义函数:返回错误消息的函数
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware


agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool, database_tool, api_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=60.0,
            jitter=True,
            tools=["api_tool"],
            retry_on=(ConnectionError, TimeoutError),
            on_failure="continue",
        ),
    ],
)

Model retry

使用可配置的指数退避自动重试失败的模型调用。Model retry 适用于以下场景:
  • 处理模型 API 调用中的瞬时失败。
  • 提升依赖网络的模型请求的可靠性。
  • 构建能优雅处理临时模型错误的弹性 agent。
API reference: ModelRetryMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool, database_tool],
    middleware=[
        ModelRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
        ),
    ],
)
max_retries
number
default:"2"
初始调用后的最大重试次数(默认总共尝试 3 次)
retry_on
tuple[type[Exception], ...] | callable
default:"(Exception,)"
可以是要重试的异常类型元组,也可以是接收异常并在应重试时返回 True 的 callable。
on_failure
string | callable
default:"continue"
所有重试都耗尽时的行为。选项:
  • 'continue'(默认):返回包含错误详情的 AIMessage,允许 agent 尝试优雅处理失败
  • 'error':重新引发异常(停止 agent 执行)
  • 自定义 callable:接收异常并返回 AIMessage 内容字符串的函数
backoff_factor
number
default:"2.0"
指数退避的乘数。每次重试等待 initial_delay * (backoff_factor ** retry_number) 秒。设置为 0.0 可使用固定延迟。
initial_delay
number
default:"1.0"
第一次重试前的初始延迟,单位为秒
max_delay
number
default:"60.0"
两次重试之间的最大延迟,单位为秒(限制指数退避增长)
jitter
boolean
default:"true"
是否向延迟添加随机 jitter(±25%),以避免 thundering herd
该 middleware 会使用指数退避自动重试失败的模型调用。
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware


# 使用默认设置的基本用法(2 次重试,指数退避)
agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool],
    middleware=[ModelRetryMiddleware()],
)

# Custom exception filtering
class TimeoutError(Exception):
    """用于超时错误的自定义异常。"""
    pass

class ConnectionError(Exception):
    """用于连接错误的自定义异常。"""
    pass

# 仅重试特定异常
retry = ModelRetryMiddleware(
    max_retries=4,
    retry_on=(TimeoutError, ConnectionError),
    backoff_factor=1.5,
)


def should_retry(error: Exception) -> bool:
    # 仅对速率限制错误重试
    if isinstance(error, TimeoutError):
        return True
    # 或检查特定 HTTP status code
    if hasattr(error, "status_code"):
        return error.status_code in (429, 503)
    return False

retry_with_filter = ModelRetryMiddleware(
    max_retries=3,
    retry_on=should_retry,
)

# 返回错误消息而不是引发异常
retry_continue = ModelRetryMiddleware(
    max_retries=4,
    on_failure="continue",  # 返回包含错误的 AIMessage,而不是引发异常
)

# 自定义错误消息格式
def format_error(error: Exception) -> str:
    return f"Model call failed: {error}. Please try again later."

retry_with_formatter = ModelRetryMiddleware(
    max_retries=4,
    on_failure=format_error,
)

# 固定退避(无指数增长)
constant_backoff = ModelRetryMiddleware(
    max_retries=5,
    backoff_factor=0.0,  # 无指数增长
    initial_delay=2.0,  # 始终等待 2 秒
)

# 失败时引发异常
strict_retry = ModelRetryMiddleware(
    max_retries=2,
    on_failure="error",  # 重新引发异常,而不是返回消息
)

LLM tool emulator

使用 LLM 模拟工具执行以便测试,用 AI 生成的响应替代实际工具调用。LLM tool emulator 适用于以下场景:
  • 不执行真实工具也能测试 agent 行为。
  • 在外部工具不可用或成本较高时开发 agent。
  • 在实现实际工具前,为 agent 工作流制作原型。
API reference: LLMToolEmulator
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator

agent = create_agent(
    model="gpt-5.4",
    tools=[get_weather, search_database, send_email],
    middleware=[
        LLMToolEmulator(),  # 模拟所有工具
    ],
)
tools
list[str | BaseTool]
要模拟的工具名称(str)或 BaseTool 实例列表。如果为 None(默认),则模拟所有工具。如果为空列表 [],则不模拟任何工具。如果是包含工具名称或实例的数组,则只模拟这些工具。
model
string | BaseChatModel
用于生成模拟工具响应的模型。可以是模型标识符字符串(例如 'google_genai:gemini-3.5-flash'),也可以是 BaseChatModel 实例。如果未指定,则默认使用 agent 的模型。更多信息,请参阅 init_chat_model
该 middleware 使用 LLM 为工具调用生成合理的响应,而不是执行实际工具。
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
from langchain.tools import tool


@tool
def get_weather(location: str) -> str:
    """获取某个地点的当前天气。"""
    return f"Weather in {location}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件。"""
    return "Email sent"


# 模拟所有工具(默认行为)
agent = create_agent(
    model="gpt-5.4",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator()],
)

# 仅模拟特定工具
agent2 = create_agent(
    model="gpt-5.4",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator(tools=["get_weather"])],
)

# 使用自定义模型进行模拟
agent4 = create_agent(
    model="gpt-5.4",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator(model="claude-sonnet-4-6")],
)

Context editing

达到 token 限制时,通过清除较早的工具调用输出来管理对话上下文,同时保留近期结果。这有助于在包含大量工具调用的长对话中保持上下文窗口可控。Context editing 适用于以下场景:
  • 包含大量工具调用并超出 token 限制的长对话
  • 通过移除不再相关的较早工具输出降低 token 成本
  • 在上下文中只保留最近 N 个工具结果
API reference: ContextEditingMiddleware, ClearToolUsesEdit
from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

agent = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=100000,
                    keep=3,
                ),
            ],
        ),
    ],
)
edits
list[ContextEdit]
default:"[ClearToolUsesEdit()]"
要应用的 ContextEdit 策略列表
token_count_method
string
default:"approximate"
Token 计数方法。选项:'approximate''model'
ClearToolUsesEdit options:
trigger
number
default:"100000"
触发编辑的 token 数。当对话超过此 token 数时,较早的工具输出会被清除。
clear_at_least
number
default:"0"
编辑运行时至少回收的 token 数。如果设置为 0,则按需尽可能清除。
keep
number
default:"3"
必须保留的最近工具结果数量。这些结果永远不会被清除。
clear_tool_inputs
boolean
default:"False"
是否清除 AI message 上原始工具调用参数。为 True 时,工具调用参数会被替换为空对象。
exclude_tools
list[string]
default:"()"
要排除在清除之外的工具名称列表。这些工具的输出永远不会被清除。
placeholder
string
default:"[cleared]"
为已清除工具输出插入的占位文本。它会替换原始工具消息内容。
达到 token 限制时,该 middleware 会应用上下文编辑策略。最常见的策略是 ClearToolUsesEdit,它会清除较早的工具结果,同时保留近期结果。工作方式:
  1. 监控对话中的 token 数
  2. 达到阈值时,清除较早的工具输出
  3. 保留最近 N 个工具结果
  4. 可选保留工具调用参数作为上下文
from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit


agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool, your_calculator_tool, database_tool],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=2000,
                    keep=3,
                    clear_tool_inputs=False,
                    exclude_tools=[],
                    placeholder="[cleared]",
                ),
            ],
        ),
    ],
)

Shell tool

向 agent 暴露持久 shell 会话,用于执行命令。Shell tool middleware 适用于以下场景:
  • 需要执行系统命令的 agent
  • 开发和部署自动化任务
  • 测试和验证工作流
  • 文件系统操作和脚本执行
安全注意事项:使用合适的执行策略(HostExecutionPolicyDockerExecutionPolicyCodexSandboxExecutionPolicy)来匹配部署环境的安全要求。
限制:持久 shell 会话目前不能与 interrupts(human-in-the-loop)一起工作。未来会添加对此的支持。
API reference: ShellToolMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import (
    ShellToolMiddleware,
    HostExecutionPolicy,
)

agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            execution_policy=HostExecutionPolicy(),
        ),
    ],
)
workspace_root
str | Path | None
shell 会话的基础目录。如果省略,则在 agent 启动时创建临时目录,并在结束时移除。
startup_commands
tuple[str, ...] | list[str] | str | None
会话启动后按顺序执行的可选命令
shutdown_commands
tuple[str, ...] | list[str] | str | None
会话关闭前执行的可选命令
execution_policy
BaseExecutionPolicy | None
控制超时、输出限制和资源配置的执行策略。选项:
  • HostExecutionPolicy:完整主机访问权限(默认);最适合 agent 已在容器或 VM 内运行的可信环境
  • DockerExecutionPolicy:为每次 agent 运行启动单独的 Docker 容器,提供更强隔离
  • CodexSandboxExecutionPolicy:复用 Codex CLI sandbox,提供额外的 syscall 和文件系统限制
redaction_rules
tuple[RedactionRule, ...] | list[RedactionRule] | None
可选的编辑规则,用于在将命令输出返回给模型前清理输出。
Redaction rules 在执行后应用,使用 HostExecutionPolicy 时,它们无法防止 secret 或敏感数据被外传。
tool_description
str | None
已注册 shell 工具描述的可选覆盖
shell_command
Sequence[str] | str | None
用于启动持久会话的可选 shell 可执行文件(字符串)或参数序列。默认为 /bin/bash
env
Mapping[str, Any] | None
提供给 shell 会话的可选环境变量。命令执行前,值会被强制转换为字符串。
该 middleware 提供单个持久 shell 会话,agent 可用它按顺序执行命令。执行策略:
  • HostExecutionPolicy(默认):拥有完整主机访问权限的原生执行
  • DockerExecutionPolicy:隔离的 Docker 容器执行
  • CodexSandboxExecutionPolicy:通过 Codex CLI 进行 sandbox 执行
from langchain.agents import create_agent
from langchain.agents.middleware import (
    ShellToolMiddleware,
    HostExecutionPolicy,
    DockerExecutionPolicy,
    RedactionRule,
)


# 使用主机执行的基本 shell 工具
agent = create_agent(
    model="gpt-5.4",
    tools=[search_tool],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            execution_policy=HostExecutionPolicy(),
        ),
    ],
)

# 带启动命令的 Docker 隔离
agent_docker = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            startup_commands=["pip install requests", "export PYTHONPATH=/workspace"],
            execution_policy=DockerExecutionPolicy(
                image="python:3.11-slim",
                command_timeout=60.0,
            ),
        ),
    ],
)

# 带输出编辑(执行后应用)
agent_redacted = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            redaction_rules=[
                RedactionRule(pii_type="api_key", detector=r"sk-[a-zA-Z0-9]{32}"),
            ],
        ),
    ],
)
在文件系统上提供 Glob 和 Grep 搜索工具。File search middleware 适用于以下场景:
  • 代码探索和分析
  • 按名称模式查找文件
  • 使用 regex 搜索代码内容
  • 需要文件发现的大型代码库
API reference: FilesystemFileSearchMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware

agent = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="/workspace",
            use_ripgrep=True,
        ),
    ],
)
root_path
str
required
要搜索的根目录。所有文件操作都相对于此路径。
use_ripgrep
bool
default:"True"
是否使用 ripgrep 进行搜索。如果 ripgrep 不可用,则回退到 Python regex。
max_file_size_mb
int
default:"10"
要搜索的最大文件大小,单位为 MB。大于此大小的文件会被跳过。
该 middleware 会向 agent 添加两个搜索工具:Glob tool:快速文件模式匹配:
  • 支持 **/*.pysrc/**/*.ts 等模式
  • 返回按修改时间排序的匹配文件路径
Grep tool:使用 regex 搜索内容:
  • 支持完整 regex 语法
  • 使用 include 参数按文件模式过滤
  • 三种输出模式:files_with_matchescontentcount
from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware
from langchain.messages import HumanMessage


agent = create_agent(
    model="gpt-5.4",
    tools=[],
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="/workspace",
            use_ripgrep=True,
            max_file_size_mb=10,
        ),
    ],
)

# Agent 现在可以使用 glob_search 和 grep_search 工具
result = agent.invoke({
    "messages": [HumanMessage("Find all Python files containing 'async def'")]
})

# Agent 会使用:
# 1. glob_search(pattern="**/*.py") 查找 Python 文件
# 2. grep_search(pattern="async def", include="*.py") 查找 async 函数

Filesystem middleware

上下文工程是构建有效 agent 的主要挑战。使用会返回可变长度结果的工具(例如 web_search 和 RAG)时尤其困难,因为较长的工具结果会快速填满上下文窗口。 Deep AgentsFilesystemMiddleware 提供四个工具,用于与短期记忆和长期记忆交互:
  • ls:列出文件系统中的文件
  • read_file:读取整个文件,或读取文件中的指定行数
  • write_file:向文件系统写入新文件
  • edit_file:编辑文件系统中的现有文件
from langchain.agents import create_agent
from deepagents.middleware.filesystem import FilesystemMiddleware

# create_deep_agent 默认包含 FilesystemMiddleware
# 构建自定义 agent 时可以自定义它
agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        FilesystemMiddleware(
            backend=None,  # 可选:自定义 backend(默认为 StateBackend)
            system_prompt="Write to the filesystem when...",  # 可选:添加到 system prompt 的自定义内容
            custom_tool_descriptions={
                "ls": "Use the ls tool when...",
                "read_file": "Use the read_file tool to..."
            }  # 可选:filesystem 工具的自定义描述
        ),
    ],
)

Short-term vs. long-term filesystem

默认情况下,这些工具会写入 graph state 中的本地 “filesystem”。要启用跨 thread 的持久存储,请配置 CompositeBackend,将特定路径(例如 /memories/)路由到 StoreBackend
from langchain.agents import create_agent
from deepagents.middleware import FilesystemMiddleware
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

agent = create_agent(
    model="claude-sonnet-4-6",
    store=store,
    middleware=[
        FilesystemMiddleware(
            backend=CompositeBackend(
                default=StateBackend(),
                routes={"/memories/": StoreBackend()}
            ),
            custom_tool_descriptions={
                "ls": "Use the ls tool when...",
                "read_file": "Use the read_file tool to..."
            }  # 可选:filesystem 工具的自定义描述
        ),
    ],
)
当你为 /memories/ 配置带 StoreBackendCompositeBackend 时,所有以 /memories/ 为前缀的文件都会保存到持久存储中,并在不同 thread 之间保留。没有此前缀的文件会留在临时状态存储中。

Subagent

将任务交给 subagent 可以隔离上下文,让主(supervisor)agent 的上下文窗口保持干净,同时仍能深入处理任务。 Deep Agents 的 subagents middleware 允许你通过 task 工具提供 subagent。
from langchain.tools import tool
from langchain.agents import create_agent
from deepagents.middleware.subagents import SubAgentMiddleware


@tool
def get_weather(city: str) -> str:
    """获取某个城市的天气。"""
    return f"The weather in {city} is sunny."

agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        SubAgentMiddleware(
            default_model="claude-sonnet-4-6",
            default_tools=[],
            subagents=[
                {
                    "name": "weather",
                    "description": "This subagent can get weather in cities.",
                    "system_prompt": "Use the get_weather tool to get the weather in a city.",
                    "tools": [get_weather],
                    "model": "gpt-5.4",
                    "middleware": [],
                }
            ],
        )
    ],
)
subagent 由 namedescriptionsystem prompttools 定义。你也可以为 subagent 提供自定义 model 或额外的 middleware。如果你想给 subagent 一个额外的状态键,与主 agent 共享,这会特别有用。 对于更复杂的用例,也可以将自己的预置 LangGraph graph 作为 subagent 提供。
from langchain.agents import create_agent
from deepagents.middleware.subagents import SubAgentMiddleware
from deepagents import CompiledSubAgent
from langgraph.graph import StateGraph

# 创建自定义 LangGraph graph
def create_weather_graph():
    workflow = StateGraph(...)
    # 构建你的自定义 graph
    return workflow.compile()

weather_graph = create_weather_graph()

# 将其包装到 CompiledSubAgent 中
weather_subagent = CompiledSubAgent(
    name="weather",
    description="This subagent can get weather in cities.",
    runnable=weather_graph
)

agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        SubAgentMiddleware(
            default_model="claude-sonnet-4-6",
            default_tools=[],
            subagents=[weather_subagent],
        )
    ],
)
除了用户定义的 subagent,主 agent 始终可以访问 general-purpose subagent。该 subagent 拥有与主 agent 相同的指令,以及主 agent 可访问的所有工具。general-purpose subagent 的主要用途是上下文隔离,主 agent 可以将复杂任务委派给它,并获得简洁答案,而不会把中间工具调用的冗余内容带回主上下文。

Provider-specific middleware

这些 middleware 针对特定 LLM 提供商进行了优化。完整详情和示例,请参阅各提供商文档。
/images/providers/anthropic-icon.svg

Anthropic

面向 Claude 模型的 prompt caching、bash tool、text editor、memory 和 file search middleware。

AWS

面向 Amazon Bedrock 模型的 prompt caching middleware。

OpenAI

面向 OpenAI 模型的 content moderation middleware。