LangChain 和 Deep Agents 为常见用例提供预构建中间件。每个中间件都是生产就绪的,可以根据您的特定需求进行配置。

提供商无关的中间件

以下中间件适用于任何 LLM 提供商:
中间件描述
摘要当接近 token 限制时自动摘要对话历史。
人工介入暂停执行以等待人工批准工具调用。
模型调用限制限制模型调用次数以防止过度成本。
工具调用限制通过限制调用计数来控制工具执行。
模型回退当主模型失败时自动回退到替代模型。
PII 检测检测和处理个人身份信息(PII)。
待办事项列表为代理配备任务规划和跟踪功能。
LLM 工具选择器在调用主模型之前使用 LLM 选择相关工具。
工具重试使用指数退避自动重试失败的工具调用。
模型重试使用指数退避自动重试失败的模型调用。
LLM 工具模拟器使用 LLM 模拟工具执行以进行测试。
上下文编辑通过修剪或清除工具使用来管理对话上下文。
Shell 工具向代理公开持久性 shell 会话以执行命令。
文件搜索提供文件系统上的 Glob 和 Grep 搜索工具。
文件系统为代理提供文件系统以存储上下文和长期记忆。
子代理添加生成子代理的能力。

摘要

当接近 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),
        ),
    ],
)
下面显示的 fraction 条件用于 triggerkeep 依赖于聊天模型的配置文件数据(如果使用 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
触发摘要的条件。可以是:
  • 单个 ContextSize 元组(必须满足指定条件)
  • ContextSize 元组列表(必须满足任何条件 - 或逻辑)
条件应该是以下之一:
  • 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_counter
function
自定义 token 计数函数。默认为基于字符的计数。
summary_prompt
string
摘要的自定义提示词模板。如果未指定,使用内置模板。模板应包含 {messages} 占位符,对话历史将在此处插入。
trim_tokens_to_summarize
number
default:"4000"
生成摘要时包含的最大 token 数。消息将在摘要之前被修剪以适应此限制。
summary_prefix
string
deprecated
已弃用: 使用 summary_prompt 提供完整提示词。
max_tokens_before_summary
number
deprecated
已弃用: 改用 trigger: ("tokens", value)。触发摘要的 token 阈值。
messages_to_keep
number
deprecated
已弃用: 改用 keep: ("messages", value)。要保留的最近消息。
摘要中间件监视消息 token 计数,并在达到阈值时自动摘要较旧的消息。触发条件控制摘要何时运行:
  • 单个条件对象(必须满足指定条件)
  • 条件数组(必须满足任何条件 - 或逻辑)
  • 每个条件可以使用 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-4.1",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4.1-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
)

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

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

人工介入

暂停代理执行以等待人工批准、编辑或拒绝工具调用,然后才能执行。人工介入适用于以下情况:
  • 需要人工批准的高风险操作(例如数据库写入、金融交易)。
  • 必须有人类监督的合规工作流。
  • 人类反馈指导代理的长时间运行对话。
API 参考: HumanInTheLoopMiddleware
人工介入中间件需要一个 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:
    """读取电子邮件的模拟函数。"""
    return f"Email content, ID: {email_id}"

def your_send_email_tool(recipient: str, subject: str, body: str) -> str:
    """发送电子邮件的模拟函数。"""
    return f"Email sent to {recipient}, 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,
            }
        ),
    ],
)
有关完整示例、配置选项和集成模式,请参阅人工介入文档
观看此视频指南,了解人工介入中间件行为演示。

模型调用限制

限制模型调用次数以防止无限循环或过度成本。模型调用限制适用于以下情况:
  • 防止失控代理进行过多 API 调用。
  • 在生产部署上实施成本控制。
  • 在特定调用预算内测试代理行为。
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(),  # 线程限制需要
    tools=[],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,
            run_limit=5,
            exit_behavior="end",
        ),
    ],
)
观看此视频指南,了解模型调用限制中间件行为演示。
thread_limit
number
线程中所有运行的最大模型调用数。默认为无限制。
run_limit
number
单次调用的最大模型调用数。默认为无限制。
exit_behavior
string
default:"end"
达到限制时的行为。选项:'end'(正常终止)或 'error'(抛出异常)

工具调用限制

通过限制工具调用次数来控制代理执行,可以全局限制所有工具或针对特定工具限制。工具调用限制适用于以下情况:
  • 防止对昂贵的外部 API 进行过多调用。
  • 限制网络搜索或数据库查询。
  • 对特定工具使用实施速率限制。
  • 防止失控的代理循环。
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=[
        # 全局限制
        ToolCallLimitMiddleware(thread_limit=20, run_limit=10),
        # 工具特定限制
        ToolCallLimitMiddleware(
            tool_name="search",
            thread_limit=5,
            run_limit=3,
        ),
    ],
)
观看此视频指南,了解工具调用限制中间件行为演示。
tool_name
string
要限制的特定工具名称。如果未提供,限制适用于所有工具全局
thread_limit
number
线程(对话)中所有运行的最大工具调用数。在具有相同线程 ID 的多次调用中持续。需要 checkpointer 维护状态。None 表示无线程限制。
run_limit
number
单次调用(一个用户消息 → 响应周期)的最大工具调用数。每次新用户消息时重置。None 表示无运行限制。注意: 必须指定 thread_limitrun_limit 中的至少一个。
exit_behavior
string
default:"continue"
达到限制时的行为:
  • 'continue'(默认)- 用错误消息阻止超出的工具调用,让其他工具和模型继续。模型根据错误消息决定何时结束。
  • 'error' - 立即抛出 ToolCallLimitExceededError 异常,停止执行
  • 'end' - 立即停止,用 ToolMessage 和 AI 消息表示超出的工具调用。仅在限制单个工具时有效;如果其他工具有待处理调用则抛出 NotImplementedError
指定限制:
  • 线程限制 - 对话中所有运行的最大调用数(需要 checkpointer)
  • 运行限制 - 每次单独调用的最大调用数(每次轮次重置)
退出行为:
  • 'continue'(默认)- 用错误消息阻止超出的调用,代理继续
  • '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],
)

模型回退

当主模型失败时自动回退到替代模型。模型回退适用于以下情况:
  • 构建能够处理模型停机的弹性代理。
  • 通过回退到更便宜的模型进行成本优化。
  • 跨 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",
        ),
    ],
)
观看此视频指南,了解模型回退中间件行为演示。
first_model
string | BaseChatModel
required
主模型失败时尝试的第一个回退模型。可以是模型标识符字符串(例如 'openai:gpt-4.1-mini')或 BaseChatModel 实例。
*additional_models
string | BaseChatModel
如果前面的模型失败,按顺序尝试的其他回退模型

PII 检测

使用可配置策略检测和处理对话中的个人身份信息(PII)。PII 检测适用于以下情况:
  • 具有合规要求的医疗和金融应用程序。
  • 需要清理日志的客户服务代理。
  • 处理敏感用户数据的任何应用程序。
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 类型。这允许您检测超出内置类型的特定于用例的模式。 创建自定义检测器的三种方式:
  1. 正则表达式字符串 - 简单模式匹配
  2. 自定义函数 - 带验证的复杂检测逻辑
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
import re


# 方法 1:正则表达式字符串
agent1 = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",
        ),
    ],
)

# 方法 2:编译的正则表达式模式
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",
        ),
    ],
)

# 方法 3:自定义检测器函数
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-4.1",
    tools=[],
    middleware=[
        PIIMiddleware(
            "ssn",
            detector=detect_ssn,
            strategy="hash",
        ),
    ],
)
自定义检测器函数签名: 检测器函数必须接受字符串(内容)并返回匹配项: 返回包含 textstartend 键的字典列表:
def detector(content: str) -> list[dict[str, str | int]]:
    return [
        {"text": "matched_text", "start": 0, "end": 12},
        # ... 更多匹配项
    ]
对于自定义检测器:
  • 使用正则表达式字符串进行简单模式
  • 当需要标志时使用 RegExp 对象(例如不区分大小写匹配)
  • 当需要超出模式匹配的验证逻辑时使用自定义函数
pii_type
string
required
PII 类型标识符。内置类型包括:"email""phone_number""ip_address""credit_card""ssn""us_passport""us_driver_license""us_bank_routing_number""us_bank_account""international_bank_account""custom_api_key"
strategy
string
required
检测到 PII 时的处理策略。选项:
  • "redact" - 用 [REDACTED] 替换 PII
  • "mask" - 用 **** 部分掩盖 PII
  • "hash" - 用 SHA256 哈希替换 PII
  • "block" - 如果在输入中检测到则阻止请求
detector
string | Pattern | callable
deprecated
已弃用。 改用新的检测器参数。
detector
str | Pattern | Callable[[str], list[dict[str, str | int]]]
自定义检测器。可以是:
  • 正则表达式字符串(例如 r"sk-[a-zA-Z0-9]{32}"
  • 编译的正则表达式对象(例如 re.compile(r"pattern")
  • 自定义检测器函数
apply_to_input
boolean
default:"True"
是否将策略应用于输入消息
apply_to_output
boolean
default:"False"
是否将策略应用于模型输出

待办事项列表

为代理配备任务规划和跟踪功能。待办事项列表适用于以下情况:
  • 需要完成多个步骤的复杂任务。
  • 按顺序跟踪子任务进展。
  • 确保所有必需步骤都完成。
API 参考: ToDoListMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import ToDoListMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[ToDoListMiddleware()],
)
task_extraction_prompt
string
用于从用户请求中提取待办事项的自定义提示词模板
task_update_prompt
string
用于更新待办事项状态的自定义提示词模板

LLM 工具选择器

在调用主模型之前使用 LLM 选择相关工具。LLM 工具选择器适用于以下情况:
  • 大量工具可用的情况。
  • 减少主模型的工具选择开销。
  • 过滤不相关工具以减少上下文大小。
API 参考: LLMToolSelectorMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[get_weather, search_web, send_email, get_calendar],
    middleware=[
        LLMToolSelectorMiddleware(
            model="gpt-4.1-mini",
            max_tools=5,
        ),
    ],
)
model
string | BaseChatModel
required
用于选择工具的模型
max_tools
number
选择的最大工具数量。如果未指定,所有相关工具都会被选择。

工具重试

使用指数退避自动重试失败的工具调用。工具重试适用于以下情况:
  • 处理暂时性故障(例如网络问题、超时)。
  • 提高依赖外部 API 的工具的可靠性。
  • 构建能够优雅处理临时错误的弹性代理。
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, api_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=60.0,
        ),
    ],
)
观看此视频指南,了解工具重试中间件行为演示。
tools
list[str | BaseTool]
要重试的工具名称列表或工具实例。如果为 None(默认),重试所有工具。如果为空列表 [],不重试任何工具。
max_retries
number
default:"2"
初始调用后的最大重试次数(默认总共 3 次尝试)
retry_on
tuple[type[Exception], ...] | callable
default:"(Exception,)"
要重试的异常类型元组,或接受异常并返回 True 的可调用对象(如果应该重试)。
on_failure
string | callable
default:"return_message"
所有重试耗尽时的行为。选项:
  • 'return_message'(默认)- 返回带有错误详情的 ToolMessage,允许代理可能优雅地处理失败
  • 'raise' - 重新抛出异常(停止代理执行)
  • 自定义可调用对象 - 函数返回 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"
是否添加随机抖动(±25%)以避免雷鸣般的群体效应
中间件使用指数退避自动重试失败的工具调用。关键配置:
  • 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",
        ),
    ],
)

模型重试

使用可配置的指数退避自动重试失败的模型调用。模型重试适用于以下情况:
  • 处理模型 API 调用中的暂时性故障。
  • 提高网络相关模型请求的可靠性。
  • 构建能够优雅处理临时模型错误的弹性代理。
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,
        ),
    ],
)
max_retries
number
default:"2"
初始调用后的最大重试次数(默认总共 3 次尝试)
retry_on
tuple[type[Exception], ...] | callable
default:"(Exception,)"
要重试的异常类型元组,或接受异常并返回 True 的可调用对象。
on_failure
string | callable
default:"continue"
所有重试耗尽时的行为。选项:
  • 'continue'(默认)- 返回带有错误详情的 AIMessage,允许代理可能优雅地处理失败
  • 'error' - 重新抛出异常(停止代理执行)
  • 自定义可调用对象 - 函数返回 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"
是否添加随机抖动(±25%)以避免雷鸣般的群体效应
中间件使用指数退避自动重试失败的模型调用。
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware


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

# 自定义异常过滤
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 状态码
    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 工具模拟器

使用 LLM 模拟工具执行以进行测试,用 AI 生成的响应替换实际工具调用。LLM 工具模拟器适用于以下情况:
  • 在不执行真实工具的情况下测试代理行为。
  • 当外部工具不可用或昂贵时开发代理。
  • 在实现实际工具之前原型化代理工作流。
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(),  # 模拟所有工具
    ],
)
tools
list[str | BaseTool]
要模拟的工具名称列表或 BaseTool 实例。如果为 None(默认),所有工具都会被模拟。如果为空列表 [],不模拟任何工具。如果数组包含工具名称/实例,则仅模拟那些工具。
model
string | BaseChatModel
用于生成模拟工具响应的模型。可以是模型标识符字符串(例如 'anthropic:claude-sonnet-4-6')或 BaseChatModel 实例。如果未指定,默认为代理的模型。更多信息请参阅 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:
    """获取位置当前的天气。"""
    return f"Weather in {location}"

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


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

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

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

上下文编辑

通过清除达到 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()]"
要应用的 ContextEdit 策略列表
token_count_method
string
default:"approximate"
Token 计数方法。选项:'approximate''model'
ClearToolUsesEdit 选项:
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 消息上的源工具调用参数。当为 True 时,工具调用参数将被替换为空对象。
exclude_tools
list[string]
default:"()"
从清除中排除的工具名称列表。这些工具的输出永远不会被清除。
placeholder
string
default:"[cleared]"
为清除的工具输出插入的占位符文本。这将替换原始工具消息内容。
中间件在达到 token 限制时应用上下文编辑策略。最常见的策略是 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-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]",
                ),
            ],
        ),
    ],
)

Shell 工具

向代理公开持久性 shell 会话以执行命令。Shell 工具中间件适用于以下情况:
  • 需要执行系统命令的代理
  • 开发和部署自动化任务
  • 测试和验证工作流
  • 文件系统操作和脚本执行
安全考虑:使用适当的执行策略(HostExecutionPolicyDockerExecutionPolicyCodexSandboxExecutionPolicy)以匹配您的部署安全要求。
限制:持久性 shell 会话目前不适用于中断(人工介入)。我们预计将来会添加对此的支持。
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(),
        ),
    ],
)
workspace_root
str | Path | None
shell 会话的基目录。如果省略,将在代理启动时创建临时目录,并在结束时删除。
startup_commands
tuple[str, ...] | list[str] | str | None
会话启动后按顺序执行的可选命令
shutdown_commands
tuple[str, ...] | list[str] | str | None
会话关闭前执行的可选命令
execution_policy
BaseExecutionPolicy | None
控制超时、输出限制和资源配置的执行策略。选项:
  • HostExecutionPolicy - 完全主机访问(默认);最适合代理已在容器或 VM 内运行的受信任环境
  • DockerExecutionPolicy - 为每次代理运行启动单独的 Docker 容器,提供更严格的隔离
  • CodexSandboxExecutionPolicy - 通过 Codex CLI 沙箱重用以获得额外的系统调用/文件系统限制
redaction_rules
tuple[RedactionRule, ...] | list[RedactionRule] | None
可选的编辑规则,用于在返回给模型之前清理命令输出。
编辑规则在执行后应用,在使用 HostExecutionPolicy 时不会阻止泄露机密或敏感数据。
tool_description
str | None
注册的 shell 工具描述的可选覆盖
shell_command
Sequence[str] | str | None
可选的 shell 可执行文件(字符串)或用于启动持久会话的参数序列。默认为 /bin/bash
env
Mapping[str, Any] | None
要提供给 shell 会话的可选环境变量。值在命令执行前被强制转换为字符串。
中间件提供单个持久性 shell 会话,代理可以使用它来按顺序执行命令。执行策略:
  • HostExecutionPolicy(默认)- 具有完全主机访问的本机执行
  • DockerExecutionPolicy - 隔离的 Docker 容器执行
  • CodexSandboxExecutionPolicy - 通过 Codex CLI 沙箱执行
from langchain.agents import create_agent
from langchain.agents.middleware import (
    ShellToolMiddleware,
    HostExecutionPolicy,
    DockerExecutionPolicy,
    RedactionRule,
)


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

# 带启动命令的 Docker 隔离
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,
            ),
        ),
    ],
)

# 带输出编辑(在执行后应用)
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}"),
            ],
        ),
    ],
)

文件搜索

提供文件系统上的 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,
        ),
    ],
)
root_path
str
required
要搜索的根目录。所有文件操作都相对于此路径。
use_ripgrep
bool
default:"True"
是否使用 ripgrep 进行搜索。如果 ripgrep 不可用,则回退到 Python 正则表达式。
max_file_size_mb
int
default:"10"
要搜索的最大文件大小(MB)。大于此值的文件将被跳过。
中间件向代理添加两个搜索工具:Glob 工具 - 快速文件模式匹配:
  • 支持 **/*.pysrc/**/*.ts 等模式
  • 返回按修改时间排序的匹配文件路径
Grep 工具 - 带正则表达式的內容搜索:
  • 完整的正则表达式语法支持
  • 使用 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-4.1",
    tools=[],
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="/workspace",
            use_ripgrep=True,
            max_file_size_mb=10,
        ),
    ],
)

# 代理现在可以使用 glob_search 和 grep_search 工具
result = agent.invoke({
    "messages": [HumanMessage("查找所有包含 'async def' 的 Python 文件")]
})

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

文件系统中间件

为代理提供文件系统以存储上下文和长期记忆。文件系统中间件适用于以下情况:
  • 需要持久存储的代理
  • 需要文件系统访问的代理
  • 需要长期记忆的代理
API 参考: FilesystemMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        FilesystemMiddleware(
            root_path="/workspace",
        ),
    ],
)
root_path
str
required
文件系统的根目录
allow_reads
bool
default:"True"
是否允许读取文件
allow_writes
bool
default:"True"
是否允许写入文件

子代理

为代理添加生成子代理的能力。子代理中间件适用于以下情况:
  • 需要将复杂任务分解为更小子任务的代理
  • 需要并行处理多个任务的代理
  • 需要隔离不同任务上下文的代理
API 参考: SubagentMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import SubagentMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        SubagentMiddleware(),
    ],
)
max_subagents
number
最大子代理数量。默认为无限制。
default_model
string
子代理使用的默认模型