LangChain 和 Deep Agents 提供了适用于常见场景的预置中间件。每个中间件均已生产就绪,并可根据您的具体需求进行配置。
与提供商无关的中间件
以下中间件适用于任何 LLM 提供商:
中间件 描述 Summarization(摘要) 在接近 token 限制时自动对对话历史进行摘要。 Human-in-the-loop(人工介入) 暂停执行以待人工审批工具调用。 Model call limit(模型调用限制) 限制模型调用次数以防止过高费用。 Tool call limit(工具调用限制) 通过限制调用次数来控制工具执行。 Model fallback(模型回退) 主模型失败时自动回退到备用模型。 PII detection(PII 检测) 检测并处理个人身份信息(PII)。 To-do list(待办列表) 为 Agent 提供任务规划和跟踪能力。 LLM tool selector(LLM 工具选择器) 在调用主模型前使用 LLM 选择相关工具。 Tool retry(工具重试) 以指数退避方式自动重试失败的工具调用。 Model retry(模型重试) 以指数退避方式自动重试失败的模型调用。 LLM tool emulator(LLM 工具模拟器) 使用 LLM 模拟工具执行,用于测试目的。 Context editing(上下文编辑) 通过裁剪或清除工具使用记录来管理对话上下文。 Shell tool(Shell 工具) 向 Agent 暴露持久 shell 会话以执行命令。 File search(文件搜索) 提供对文件系统文件的 Glob 和 Grep 搜索工具。 Filesystem(文件系统) 为 Agent 提供文件系统以存储上下文和长期记忆。 Subagent(子 Agent) 添加生成子 Agent 的能力。
Summarization
在接近 token 限制时自动对对话历史进行摘要,保留最近的消息并压缩较旧的上下文。摘要功能在以下场景非常有用:
超出上下文窗口的长对话。
具有大量历史记录的多轮对话。
需要保留完整对话上下文的应用。
API 参考: SummarizationMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import SummarizationMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [ your_weather_tool , your_calculator_tool ],
middleware = [
SummarizationMiddleware (
model = "gpt-4.1-mini" ,
trigger = ( "tokens" , 4000 ),
keep = ( "messages" , 20 ),
),
],
)
trigger 和 keep 的 fraction 条件(如下所示)在使用 langchain>=1.1 时依赖聊天模型的配置文件数据 。如果数据不可用,请使用其他条件或手动指定:from langchain . chat_models import init_chat_model
custom_profile = {
"max_input_tokens" : 100_000 ,
# ...
}
model = init_chat_model ( "gpt-4.1" , profile = custom_profile )
model
string | BaseChatModel
required
用于生成摘要的模型。可以是模型标识字符串(例如 'openai:gpt-4.1-mini')或 BaseChatModel 实例。更多信息请参阅 init_chat_model 。 trigger
ContextSize | list[ContextSize] | None
触发摘要的条件。可以是: 条件应为以下之一:
fraction(float):模型上下文大小的比例(0-1)
tokens(int):绝对 token 数量
messages(int):消息数量
至少必须指定一个条件。如果未提供,摘要将不会自动触发。 更多信息请参阅 ContextSize 的 API 参考。 keep
ContextSize
default: "('messages', 20)"
摘要后保留多少上下文。精确指定以下之一:
fraction(float):保留模型上下文大小的比例(0-1)
tokens(int):保留的绝对 token 数量
messages(int):保留的最近消息数量
更多信息请参阅 ContextSize 的 API 参考。 自定义 token 计数函数。默认使用基于字符的计数。
用于摘要的自定义提示模板。未指定时使用内置模板。模板应包含 {messages} 占位符,对话历史将插入其中。
生成摘要时包含的最大 token 数量。摘要前将裁剪消息以符合此限制。
已弃用: 请改用 summary_prompt 提供完整提示。
max_tokens_before_summary
已弃用: 请改用 trigger: ("tokens", value)。触发摘要的 token 阈值。
已弃用: 请改用 keep: ("messages", value)。要保留的最近消息数量。
Human-in-the-loop
在工具调用执行前暂停 Agent 执行,以待人工审批、编辑或拒绝。Human-in-the-loop(人工介入) 在以下场景非常有用:
需要人工审批的高风险操作(例如数据库写入、金融交易)。
强制要求人工监督的合规工作流。
需要人工反馈来引导 Agent 的长对话。
API 参考: HumanInTheLoopMiddleware
Human-in-the-loop 中间件需要检查点器 来在中断之间维护状态。
from langchain . agents import create_agent
from langchain . agents . middleware import HumanInTheLoopMiddleware
from langgraph . checkpoint . memory import InMemorySaver
def read_email_tool ( email_id : str ) -> str :
"""Mock function to read an email by its ID."""
return f "Email content for ID: { email_id } "
def send_email_tool ( recipient : str , subject : str , body : str ) -> str :
"""Mock function to send an email."""
return f "Email sent to { recipient } with subject ' { subject } '"
agent = create_agent (
model = "gpt-4.1" ,
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 中间件的行为。
Model call limit
限制模型调用次数以防止无限循环或过高费用。模型调用限制在以下场景非常有用:
防止失控 Agent 发出过多 API 调用。
在生产部署中强制执行成本控制。
在特定调用预算内测试 Agent 行为。
API 参考: ModelCallLimitMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import ModelCallLimitMiddleware
from langgraph . checkpoint . memory import InMemorySaver
agent = create_agent (
model = "gpt-4.1" ,
checkpointer = InMemorySaver (), # Required for thread limiting
tools = [],
middleware = [
ModelCallLimitMiddleware (
thread_limit = 10 ,
run_limit = 5 ,
exit_behavior = "end" ,
),
],
)
观看此视频指南 了解 Model Call Limit 中间件的行为。
一个线程中所有运行的最大模型调用次数。默认无限制。
达到限制时的行为。选项:'end'(优雅终止)或 'error'(抛出异常)
通过限制工具调用次数来控制 Agent 执行,可全局限制所有工具或针对特定工具。工具调用限制在以下场景非常有用:
防止对昂贵外部 API 的过度调用。
限制网络搜索或数据库查询。
对特定工具使用强制速率限制。
防止 Agent 进入失控循环。
API 参考: ToolCallLimitMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import ToolCallLimitMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool , database_tool ],
middleware = [
# Global limit
ToolCallLimitMiddleware ( thread_limit = 20 , run_limit = 10 ),
# Tool-specific limit
ToolCallLimitMiddleware (
tool_name = "search" ,
thread_limit = 5 ,
run_limit = 3 ,
),
],
)
观看此视频指南 了解 Tool Call Limit 中间件的行为。
要限制的特定工具名称。如果未提供,限制将应用于所有工具(全局) 。
一个线程(对话)中所有运行的最大工具调用次数。使用相同线程 ID 的多次调用之间持久存在。需要检查点器来维护状态。None 表示无线程限制。
每次单次调用(一个用户消息→响应周期)的最大工具调用次数。每条新用户消息后重置。None 表示无运行限制。 注意: thread_limit 和 run_limit 至少必须指定一个。
达到限制时的行为:
'continue'(默认)- 以错误消息阻止超限的工具调用,让其他工具和模型继续。模型根据错误消息决定何时结束。
'error' - 抛出 ToolCallLimitExceededError 异常,立即停止执行
'end' - 使用 ToolMessage 和 AI 消息立即停止执行。仅在限制单个工具时有效;如果其他工具有待处理的调用,则抛出 NotImplementedError。
使用以下方式指定限制:
线程限制 - 一个对话中所有运行的最大调用次数(需要检查点器)
运行限制 - 每次单次调用的最大调用次数(每轮重置)
退出行为:
'continue'(默认)- 以错误消息阻止超限调用,Agent 继续
'error' - 立即抛出异常
'end' - 使用 ToolMessage + AI 消息停止(仅适用于单工具场景)
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-4.1" ,
tools = [ search_tool , database_tool , scraper_tool ],
middleware = [ global_limiter , search_limiter , database_limiter , strict_limiter ],
)
Model fallback
主模型失败时自动回退到备用模型。模型回退在以下场景非常有用:
构建能够处理模型中断的弹性 Agent。
通过回退到更便宜的模型来优化成本。
跨 OpenAI、Anthropic 等提供商的冗余。
API 参考: ModelFallbackMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import ModelFallbackMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
ModelFallbackMiddleware (
"gpt-4.1-mini" ,
"claude-3-5-sonnet-20241022" ,
),
],
)
观看此视频指南 了解 Model Fallback 中间件的行为。
first_model
string | BaseChatModel
required
主模型失败时首先尝试的备用模型。可以是模型标识字符串(例如 'openai:gpt-4.1-mini')或 BaseChatModel 实例。
PII detection
使用可配置策略检测并处理对话中的个人身份信息(PII)。PII 检测在以下场景非常有用:
有合规要求的医疗和金融应用。
需要对日志进行脱敏的客户服务 Agent。
任何处理敏感用户数据的应用。
API 参考: PIIMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import PIIMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
PIIMiddleware ( "email" , strategy = "redact" , apply_to_input = True ),
PIIMiddleware ( "credit_card" , strategy = "mask" , apply_to_input = True ),
],
)
自定义 PII 类型
您可以通过提供 detector 参数来创建自定义 PII 类型。这允许您检测超出内置类型的特定用例模式。
三种创建自定义检测器的方式:
正则表达式模式字符串 - 简单模式匹配
自定义函数 - 带验证逻辑的复杂检测
from langchain . agents import create_agent
from langchain . agents . middleware import PIIMiddleware
import re
# Method 1: Regex pattern string
agent1 = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
PIIMiddleware (
"api_key" ,
detector = r " sk- [ a-zA-Z0-9 ] {32} " ,
strategy = "block" ,
),
],
)
# Method 2: Compiled regex pattern
agent2 = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
PIIMiddleware (
"phone_number" ,
detector = re . compile ( r "\+ ? \d {1,3} [ \s.- ] ? \d {3,4} [ \s.- ] ? \d {4} " ),
strategy = "mask" ,
),
],
)
# Method 3: Custom detector function
def detect_ssn ( content : str ) -> list [ dict [ str , str | int ]]:
"""Detect SSN with validation.
Returns a list of dictionaries with 'text', 'start', and 'end' keys.
"""
import re
matches = []
pattern = r " \d {3} -\d {2} -\d {4} "
for match in re . finditer ( pattern , content ):
ssn = match . group ( 0 )
# Validate: first 3 digits shouldn't be 000, 666, or 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-4.1" ,
tools = [],
middleware = [
PIIMiddleware (
"ssn" ,
detector = detect_ssn ,
strategy = "hash" ,
),
],
)
自定义检测器函数签名:
检测器函数必须接受一个字符串(内容)并返回匹配结果:
返回包含 text、start 和 end 键的字典列表:
def detector ( content : str ) -> list [ dict [ str , str | int ]]:
return [
{ "text" : "matched_text" , "start" : 0 , "end" : 12 },
# ... more matches
]
关于自定义检测器:
简单模式使用正则表达式字符串
需要标志(例如不区分大小写匹配)时使用 RegExp 对象
需要超出模式匹配的验证逻辑时使用自定义函数
自定义函数可完全控制检测逻辑,并可实现复杂的验证规则
要检测的 PII 类型。可以是内置类型(email、credit_card、ip、mac_address、url)或自定义类型名称。
处理检测到的 PII 的方式。选项:
'block' - 检测到时抛出异常
'redact' - 替换为 [REDACTED_{PII_TYPE}]
'mask' - 部分遮蔽(例如 ****-****-****-1234)
'hash' - 替换为确定性哈希值
自定义检测器函数或正则表达式模式。如果未提供,使用该 PII 类型的内置检测器。
To-do list
为 Agent 提供任务规划和跟踪能力,用于复杂的多步骤任务。待办列表在以下场景非常有用:
需要跨多个工具协调的复杂多步骤任务。
进度可见性很重要的长期运行操作。
此中间件会自动为 Agent 提供 write_todos 工具和系统提示,以引导有效的任务规划。
API 参考: TodoListMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import TodoListMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [ read_file , write_file , run_tests ],
middleware = [ TodoListMiddleware ()],
)
观看此视频指南 了解 To-do List 中间件的行为。
用于引导待办事项使用的自定义系统提示。未指定时使用内置提示。
write_todos 工具的自定义描述。未指定时使用内置描述。
在调用主模型前使用 LLM 智能选择相关工具。LLM 工具选择器在以下场景非常有用:
拥有大量工具(10+)且每次查询大多数工具不相关的 Agent。
通过过滤不相关工具来减少 token 使用。
提升模型的专注度和准确性。
此中间件使用结构化输出询问 LLM 哪些工具与当前查询最相关。结构化输出模式定义了可用的工具名称和描述。模型提供商通常会在后台将此结构化输出信息添加到系统提示中。
API 参考: LLMToolSelectorMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import LLMToolSelectorMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [ tool1 , tool2 , tool3 , tool4 , tool5 , ... ],
middleware = [
LLMToolSelectorMiddleware (
model = "gpt-4.1-mini" ,
max_tools = 3 ,
always_include = [ "search" ],
),
],
)
用于工具选择的模型。可以是模型标识字符串(例如 'openai:gpt-4.1-mini')或 BaseChatModel 实例。更多信息请参阅 init_chat_model 。 默认使用 Agent 的主模型。 最多选择的工具数量。如果模型选择的数量超过此值,只使用前 max_tools 个。未指定时无限制。
无论选择结果如何始终包含的工具名称。这些不计入 max_tools 限制。
以可配置的指数退避方式自动重试失败的工具调用。工具重试在以下场景非常有用:
处理外部 API 调用中的瞬时故障。
提高依赖网络的工具的可靠性。
构建能优雅处理临时错误的弹性 Agent。
API 参考: ToolRetryMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import ToolRetryMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool , database_tool ],
middleware = [
ToolRetryMiddleware (
max_retries = 3 ,
backoff_factor = 2.0 ,
initial_delay = 1.0 ,
),
],
)
要应用重试逻辑的工具或工具名称的可选列表。如果为 None,则应用于所有工具。
retry_on
tuple[type[Exception], ...] | callable
default: "(Exception,)"
要重试的异常类型元组,或接受异常并在应重试时返回 True 的可调用对象。
on_failure
string | callable
default: "return_message"
所有重试耗尽时的行为。选项:
'return_message' - 返回包含错误详情的 ToolMessage(允许 LLM 处理失败)
'raise' - 重新抛出异常(停止 Agent 执行)
自定义可调用对象 - 接受异常并返回 ToolMessage 内容字符串的函数
指数退避的乘数。每次重试等待 initial_delay * (backoff_factor ** retry_number) 秒。设为 0.0 表示固定延迟。
此中间件以指数退避方式自动重试失败的工具调用。 关键配置:
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-4.1" ,
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
以可配置的指数退避方式自动重试失败的模型调用。模型重试在以下场景非常有用:
处理模型 API 调用中的瞬时故障。
提高依赖网络的模型请求的可靠性。
构建能优雅处理临时模型错误的弹性 Agent。
API 参考: ModelRetryMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import ModelRetryMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool , database_tool ],
middleware = [
ModelRetryMiddleware (
max_retries = 3 ,
backoff_factor = 2.0 ,
initial_delay = 1.0 ,
),
],
)
retry_on
tuple[type[Exception], ...] | callable
default: "(Exception,)"
要重试的异常类型元组,或接受异常并在应重试时返回 True 的可调用对象。
on_failure
string | callable
default: "continue"
所有重试耗尽时的行为。选项:
'continue'(默认)- 返回包含错误详情的 AIMessage,允许 Agent 优雅处理失败
'error' - 重新抛出异常(停止 Agent 执行)
自定义可调用对象 - 接受异常并返回 AIMessage 内容字符串的函数
指数退避的乘数。每次重试等待 initial_delay * (backoff_factor ** retry_number) 秒。设为 0.0 表示固定延迟。
此中间件以指数退避方式自动重试失败的模型调用。 from langchain . agents import create_agent
from langchain . agents . middleware import ModelRetryMiddleware
# Basic usage with default settings (2 retries, exponential backoff)
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool ],
middleware = [ ModelRetryMiddleware ()],
)
# Custom exception filtering
class TimeoutError ( Exception ):
"""Custom exception for timeout errors."""
pass
class ConnectionError ( Exception ):
"""Custom exception for connection errors."""
pass
# Retry specific exceptions only
retry = ModelRetryMiddleware (
max_retries = 4 ,
retry_on = ( TimeoutError , ConnectionError ),
backoff_factor = 1.5 ,
)
def should_retry ( error : Exception ) -> bool :
# Only retry on rate limit errors
if isinstance ( error , TimeoutError ):
return True
# Or check for specific HTTP status codes
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 ,
)
# Return error message instead of raising
retry_continue = ModelRetryMiddleware (
max_retries = 4 ,
on_failure = "continue" , # Return AIMessage with error instead of raising
)
# Custom error message formatting
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 (no exponential growth)
constant_backoff = ModelRetryMiddleware (
max_retries = 5 ,
backoff_factor = 0.0 , # No exponential growth
initial_delay = 2.0 , # Always wait 2 seconds
)
# Raise exception on failure
strict_retry = ModelRetryMiddleware (
max_retries = 2 ,
on_failure = "error" , # Re-raise exception instead of returning message
)
使用 LLM 模拟工具执行以用于测试目的,将实际工具调用替换为 AI 生成的响应。LLM 工具模拟器在以下场景非常有用:
在不执行真实工具的情况下测试 Agent 行为。
在外部工具不可用或成本高昂时开发 Agent。
在实现实际工具之前对 Agent 工作流进行原型设计。
API 参考: LLMToolEmulator
from langchain . agents import create_agent
from langchain . agents . middleware import LLMToolEmulator
agent = create_agent (
model = "gpt-4.1" ,
tools = [ get_weather , search_database , send_email ],
middleware = [
LLMToolEmulator (), # Emulate all tools
],
)
要模拟的工具名称(str)或 BaseTool 实例列表。如果为 None(默认),将模拟所有工具。如果为空列表 [],不模拟任何工具。如果为包含工具名称/实例的数组,只模拟这些工具。
用于生成模拟工具响应的模型。可以是模型标识字符串(例如 'anthropic:claude-sonnet-4-6')或 BaseChatModel 实例。未指定时默认使用 Agent 的模型。更多信息请参阅 init_chat_model 。
此中间件使用 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 :
"""Get the current weather for a location."""
return f "Weather in { location } "
@tool
def send_email ( to : str , subject : str , body : str ) -> str :
"""Send an email."""
return "Email sent"
# Emulate all tools (default behavior)
agent = create_agent (
model = "gpt-4.1" ,
tools = [ get_weather , send_email ],
middleware = [ LLMToolEmulator ()],
)
# Emulate specific tools only
agent2 = create_agent (
model = "gpt-4.1" ,
tools = [ get_weather , send_email ],
middleware = [ LLMToolEmulator ( tools = [ "get_weather" ])],
)
# Use custom model for emulation
agent4 = create_agent (
model = "gpt-4.1" ,
tools = [ get_weather , send_email ],
middleware = [ LLMToolEmulator ( model = "claude-sonnet-4-6" )],
)
Context editing
通过在达到 token 限制时清除较旧的工具调用输出来管理对话上下文,同时保留最近的结果。这有助于在具有大量工具调用的长对话中保持上下文窗口的可管理性。上下文编辑在以下场景非常有用:
具有大量工具调用且超出 token 限制的长对话
通过删除不再相关的旧工具输出来降低 token 成本
在上下文中仅保留最近的 N 个工具结果
API 参考: ContextEditingMiddleware ,ClearToolUsesEdit
from langchain . agents import create_agent
from langchain . agents . middleware import ContextEditingMiddleware , ClearToolUsesEdit
agent = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
ContextEditingMiddleware (
edits = [
ClearToolUsesEdit (
trigger = 100000 ,
keep = 3 ,
),
],
),
],
)
edits
list[ContextEdit]
default: "[ClearToolUsesEdit()]"
token_count_method
string
default: "approximate"
Token 计数方法。选项:'approximate' 或 'model'
ClearToolUsesEdit 选项:触发编辑的 token 数量。当对话超过此 token 数量时,将清除较旧的工具输出。
编辑运行时要回收的最小 token 数量。设为 0 时按需清除尽可能多的内容。
必须保留的最近工具结果数量。这些将永远不会被清除。
是否清除 AI 消息上的原始工具调用参数。为 True 时,工具调用参数将替换为空对象。
要排除在清除之外的工具名称列表。这些工具的输出将永远不会被清除。
placeholder
string
default: "[cleared]"
插入到已清除工具输出中的占位符文本。这将替换原始工具消息内容。
此中间件在达到 token 限制时应用上下文编辑策略。最常见的策略是 ClearToolUsesEdit,它在保留最近结果的同时清除较旧的工具结果。 工作原理:
监控对话中的 token 数量
达到阈值时,清除较旧的工具输出
保留最近的 N 个工具结果
可选保留工具调用参数以提供上下文
from langchain . agents import create_agent
from langchain . agents . middleware import ContextEditingMiddleware , ClearToolUsesEdit
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool , your_calculator_tool , database_tool ],
middleware = [
ContextEditingMiddleware (
edits = [
ClearToolUsesEdit (
trigger = 2000 ,
keep = 3 ,
clear_tool_inputs = False ,
exclude_tools = [],
placeholder = "[cleared]" ,
),
],
),
],
)
向 Agent 暴露持久 shell 会话以执行命令。Shell 工具中间件在以下场景非常有用:
需要执行系统命令的 Agent
开发和部署自动化任务
测试和验证工作流
文件系统操作和脚本执行
安全注意事项 :请使用适当的执行策略(HostExecutionPolicy、DockerExecutionPolicy 或 CodexSandboxExecutionPolicy)以匹配您部署的安全要求。
限制 :持久 shell 会话目前不支持中断(human-in-the-loop)。我们预计将在未来添加对此的支持。
API 参考: ShellToolMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import (
ShellToolMiddleware ,
HostExecutionPolicy ,
)
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool ],
middleware = [
ShellToolMiddleware (
workspace_root = "/workspace" ,
execution_policy = HostExecutionPolicy (),
),
],
)
shell 会话的基础目录。如果省略,Agent 启动时会创建一个临时目录,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 沙箱以获得额外的系统调用/文件系统限制
redaction_rules
tuple[RedactionRule, ...] | list[RedactionRule] | None
在将命令输出返回给模型之前对其进行脱敏的可选脱敏规则。 脱敏规则在执行后应用,使用 HostExecutionPolicy 时不能防止密钥或敏感数据的泄露。
shell_command
Sequence[str] | str | None
用于启动持久会话的可选 shell 可执行文件(字符串)或参数序列。默认为 /bin/bash。
提供给 shell 会话的可选环境变量。值在命令执行前强制转换为字符串。
此中间件提供一个持久 shell 会话,Agent 可以用它依次执行命令。 执行策略:
HostExecutionPolicy(默认)- 具有完全主机访问权限的原生执行
DockerExecutionPolicy - 隔离的 Docker 容器执行
CodexSandboxExecutionPolicy - 通过 Codex CLI 的沙箱执行
from langchain . agents import create_agent
from langchain . agents . middleware import (
ShellToolMiddleware ,
HostExecutionPolicy ,
DockerExecutionPolicy ,
RedactionRule ,
)
# Basic shell tool with host execution
agent = create_agent (
model = "gpt-4.1" ,
tools = [ search_tool ],
middleware = [
ShellToolMiddleware (
workspace_root = "/workspace" ,
execution_policy = HostExecutionPolicy (),
),
],
)
# Docker isolation with startup commands
agent_docker = create_agent (
model = "gpt-4.1" ,
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 ,
),
),
],
)
# With output redaction (applied post execution)
agent_redacted = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
ShellToolMiddleware (
workspace_root = "/workspace" ,
redaction_rules = [
RedactionRule ( pii_type = "api_key" , detector = r " sk- [ a-zA-Z0-9 ] {32} " ),
],
),
],
)
File search
为 Agent 提供对文件系统的 Glob 和 Grep 搜索工具。文件搜索中间件在以下场景非常有用:
代码探索和分析
按名称模式查找文件
使用正则表达式搜索代码内容
需要文件发现的大型代码库
API 参考: FilesystemFileSearchMiddleware
from langchain . agents import create_agent
from langchain . agents . middleware import FilesystemFileSearchMiddleware
agent = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
FilesystemFileSearchMiddleware (
root_path = "/workspace" ,
use_ripgrep = True ,
),
],
)
是否使用 ripgrep 进行搜索。如果 ripgrep 不可用,则回退到 Python 正则表达式。
要搜索的最大文件大小(MB)。超过此大小的文件将被跳过。
此中间件为 Agent 添加两个搜索工具: Glob 工具 - 快速文件模式匹配:
支持 **/*.py、src/**/*.ts 等模式
按修改时间排序返回匹配的文件路径
Grep 工具 - 使用正则表达式的内容搜索:
完整的正则表达式语法支持
使用 include 参数按文件模式过滤
三种输出模式:files_with_matches、content、count
from langchain . agents import create_agent
from langchain . agents . middleware import FilesystemFileSearchMiddleware
from langchain . messages import HumanMessage
agent = create_agent (
model = "gpt-4.1" ,
tools = [],
middleware = [
FilesystemFileSearchMiddleware (
root_path = "/workspace" ,
use_ripgrep = True ,
max_file_size_mb = 10 ,
),
],
)
# Agent can now use glob_search and grep_search tools
result = agent . invoke ({
"messages" : [ HumanMessage ( "Find all Python files containing 'async def'" )]
})
# The agent will use:
# 1. glob_search(pattern="**/*.py") to find Python files
# 2. grep_search(pattern="async def", include="*.py") to find async functions
Filesystem 中间件
上下文工程是构建有效 Agent 的主要挑战。当使用返回长度可变的结果的工具(例如 web_search 和 RAG)时,这尤其困难,因为长工具结果会迅速填满您的上下文窗口。
来自 deep agents 的 FilesystemMiddleware 提供四种与短期和长期记忆交互的工具:
ls:列出文件系统中的文件
read_file:读取整个文件或文件中的若干行
write_file:向文件系统写入新文件
edit_file:编辑文件系统中的现有文件
from langchain . agents import create_agent
from deepagents . middleware . filesystem import FilesystemMiddleware
# FilesystemMiddleware is included by default in create_deep_agent
# You can customize it if building a custom agent
agent = create_agent (
model = "claude-sonnet-4-6" ,
middleware = [
FilesystemMiddleware (
backend = None , # Optional: custom backend (defaults to StateBackend)
system_prompt = "Write to the filesystem when..." , # Optional custom addition to the system prompt
custom_tool_descriptions = {
"ls" : "Use the ls tool when..." ,
"read_file" : "Use the read_file tool to..."
} # Optional: Custom descriptions for filesystem tools
),
],
)
短期与长期文件系统
默认情况下,这些工具写入图状态中的本地”文件系统”。要在线程间启用持久存储,请配置一个 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 = lambda rt : CompositeBackend (
default = StateBackend ( rt ),
routes = { "/memories/" : StoreBackend ( rt )}
),
custom_tool_descriptions = {
"ls" : "Use the ls tool when..." ,
"read_file" : "Use the read_file tool to..."
} # Optional: Custom descriptions for filesystem tools
),
],
)
当您为 /memories/ 配置带有 StoreBackend 的 CompositeBackend 时,任何以 /memories/ 为前缀的文件都会保存到持久存储,并在不同线程之间保留。没有此前缀的文件保留在临时状态存储中。
Subagent
将任务交给子 Agent 可以隔离上下文,在深入处理任务的同时保持主(监督者)Agent 的上下文窗口整洁。
来自 deep agents 的子 Agent 中间件允许您通过 task 工具提供子 Agent。
from langchain . tools import tool
from langchain . agents import create_agent
from deepagents . middleware . subagents import SubAgentMiddleware
@tool
def get_weather ( city : str ) -> str :
"""Get the weather in a city."""
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-4.1" ,
"middleware" : [],
}
],
)
],
)
子 Agent 由名称 、描述 、系统提示 和工具 定义。您还可以为子 Agent 提供自定义模型 或额外的中间件 。当您希望给子 Agent 一个额外的状态键与主 Agent 共享时,这特别有用。
对于更复杂的使用场景,您也可以提供自己预构建的 LangGraph 图作为子 Agent。
from langchain . agents import create_agent
from deepagents . middleware . subagents import SubAgentMiddleware
from deepagents import CompiledSubAgent
from langgraph . graph import StateGraph
# Create a custom LangGraph graph
def create_weather_graph ():
workflow = StateGraph ( ... )
# Build your custom graph
return workflow . compile ()
weather_graph = create_weather_graph ()
# Wrap it in a 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 ],
)
],
)
除了任何用户定义的子 Agent 外,主 Agent 始终可以访问一个 general-purpose(通用)子 Agent。此子 Agent 具有与主 Agent 相同的指令和所有可访问工具。general-purpose 子 Agent 的主要目的是上下文隔离——主 Agent 可以将复杂任务委托给此子 Agent,并获得简洁的回答,而不会有中间工具调用带来的冗余信息。
特定提供商中间件
这些中间件针对特定 LLM 提供商进行了优化。有关完整详情和示例,请参阅各提供商的文档。
通过 MCP 将这些文档接入 Claude、VSCode 等以获取实时答案。