内置中间件（下）：工具选择、重试与上下文

3.4 LLMToolSelectorMiddleware中间件

智能工具筛选。

当工具太多时，用子模型筛选最相关的几个工具。

参数：

• model ：用于工具筛选的子模型
• max_tools ：限定可以调用的工具总数
• always_include ：指定的工具不被计数

from langchain.agents.middleware import LLMToolSelectorMiddleware

tool_selector = LLMToolSelectorMiddleware(
    model="openai:gpt-5.4-mini",
    max_tools=5,  # 最多选择 5 个工具
    always_include=["get_weather"]
)

agent = create_agent(
    model="deepseek-v4-flash",
    tools=[...100个工具...],  # 很多工具
    middleware=[tool_selector]
)

举例1

提供必要的大模型：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从.env文件中加载环境变量
load_dotenv(override=True)

model_out = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从.env文件中加载环境变量
load_dotenv(override=True)

model_in = init_chat_model(
    model="gpt-4o-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

继续：

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain.messages import HumanMessage
from langchain.tools import tool

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

@tool
def get_news():
    """查询今日国内新闻概要"""
    return ("今日国内新闻概要："
            "中方三艘油轮过航霍尔木兹海峡")

@tool
def calculate(num1: int, num2: int) -> int:
    """
    执行数学计算

    Args:
        num1: 第一个加数
        num2: 第二个加数
    """
    return num1 + num2

@tool
def search_stock(symbol: str):
    """
    查询股票行情

    Args:
        symbol: 股票代码
    """
    return "该股票今天行情不错"
agent = create_agent(
    model=model_out,
    tools=[get_weather, get_news, calculate, search_stock],
    middleware=[
        LLMToolSelectorMiddleware(
            model=model_in,
            max_tools=0,
            always_include=["get_weather"]
        ),
    ],
)

response = agent.invoke({
    "messages": HumanMessage("北京今天天气如何？今日新闻概要")
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

北京今天天气如何？今日新闻概要
================================== Ai Message ==================================
Tool Calls:
get_weather (call_dZzxfUgchN0iXX7HXXwCEman)
Call ID: call_dZzxfUgchN0iXX7HXXwCEman
 Args:
  city: 北京
================================= Tool Message =================================
Name: get_weather

北京今天天气晴朗
================================== Ai Message ==================================

北京今天天气：**晴朗**。

关于**今日新闻概要**：我目前无法直接获取实时新闻源。
如果你愿意，我可以帮你：
1. **按你关心的领域**整理新闻概要（如国内、国际、财经、科技、体育）
2. **根据你提供的新闻链接/标题**做摘要
3. 你也可以告诉我想看的**新闻时间范围**和**主题**，我来帮你生成一版简明概要。

举例2

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain.messages import HumanMessage
from langchain.tools import tool
@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

@tool
def get_news():
    """查询今日国内新闻概要"""
    return ("今日国内新闻概要："
            "中方三艘油轮过航霍尔木兹海峡")

@tool
def calculate(num1: int, num2: int) -> int:
    """
    执行数学计算

    Args:
        num1: 第一个加数
        num2: 第二个加数
    """
    return num1 + num2

@tool
def search_stock(symbol: str):
    """
    查询股票行情

    Args:
        symbol: 股票代码
    """
    return "该股票今天行情不错"

agent = create_agent(
    model=model_out,
    tools=[get_weather, get_news, calculate, search_stock],
    middleware=[
        LLMToolSelectorMiddleware(
            model=model_in,
            max_tools=0,
            always_include=["get_news"]
        ),
    ],
)

response = agent.invoke({
    "messages": HumanMessage("北京今天天气如何？今日新闻概要")
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

北京今天天气如何？今日新闻概要
================================== Ai Message ==================================
Tool Calls:
get_news (call_cDdh1r1WCuyd3Y0QtRT4gnLo)
Call ID: call_cDdh1r1WCuyd3Y0QtRT4gnLo
 Args:
================================= Tool Message =================================
Name: get_news

今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡
================================== Ai Message ==================================

我可以先给你**今日新闻概要**：

- **中方三艘油轮过航霍尔木兹海峡**

另外，**北京今天天气**我这边目前**无法直接获取实时天气数据**。
如果你愿意，我可以帮你：

1. **根据你提供的天气信息**来解读穿衣建议；或
2. 你告诉我你想看 **北京的气温、降雨、风力**，我可以给你一个实用的出门建议模板。

举例3

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain.messages import HumanMessage
from langchain.tools import tool

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

@tool
def get_news():
    """查询今日国内新闻概要"""
    return ("今日国内新闻概要："
            "中方三艘油轮过航霍尔木兹海峡")

@tool
def calculate(num1: int, num2: int) -> int:
    """
    执行数学计算

    Args:
        num1: 第一个加数
        num2: 第二个加数
    """
    return num1 + num2

@tool
def search_stock(symbol: str):
    """
    查询股票行情

    Args:
        symbol: 股票代码
    """
    return "该股票今天行情不错"

agent = create_agent(
    model=model_out,
    tools=[get_weather, get_news, calculate, search_stock],
    middleware=[
        LLMToolSelectorMiddleware(
            model=model_in,
            max_tools=0,
            always_include=["get_weather", "get_news"]
        ),
    ],
)

response = agent.invoke({
    "messages": HumanMessage("北京今天天气如何？今日新闻概要")
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

北京今天天气如何？今日新闻概要
================================== Ai Message ==================================
Tool Calls:
get_weather (call_0tvrXvRQEC6Hov7FuSYZjzyh)
Call ID: call_0tvrXvRQEC6Hov7FuSYZjzyh
 Args:
  city: 北京
 get_news (call_QuQ0Tw3nFrWamdnRsDdFcAdO)
 Call ID: call_QuQ0Tw3nFrWamdnRsDdFcAdO
 Args:
================================= Tool Message =================================
Name: get_weather

北京今天天气晴朗
================================= Tool Message =================================
Name: get_news

今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡
================================== Ai Message ==================================

北京今天天气：晴朗。
今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡。

举例4

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain.messages import HumanMessage
from langchain.tools import tool

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

@tool
def get_news():
    """查询今日国内新闻概要"""
    return ("今日国内新闻概要："
            "中方三艘油轮过航霍尔木兹海峡")

@tool
def calculate(num1: int, num2: int) -> int:
    """
    执行数学计算

    Args:
        num1: 第一个加数
        num2: 第二个加数
    """
    return num1 + num2

@tool
def search_stock(symbol: str):
    """
    查询股票行情

    Args:
        symbol: 股票代码
    """
    return "该股票今天行情不错"

agent = create_agent(
    model=model_out,
    tools=[get_weather, get_news, calculate, search_stock],
    middleware=[
        LLMToolSelectorMiddleware(
            model=model_in,
            max_tools=1,
            always_include=["get_weather"]
        ),
    ],
)

response = agent.invoke({
    "messages": HumanMessage("北京今天天气如何？今日新闻概要")
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

北京今天天气如何？今日新闻概要
================================== Ai Message ==================================
Tool Calls:
get_weather (call_lskyUyDve7vnvc4mdNOWtPGf)
Call ID: call_lskyUyDve7vnvc4mdNOWtPGf
 Args:
  city: 北京
 get_news (call_vyD9kylFNc6wAEW3B9GT8APB)
 Call ID: call_vyD9kylFNc6wAEW3B9GT8APB
 Args:
================================= Tool Message =================================
Name: get_weather

北京今天天气晴朗
================================= Tool Message =================================
Name: get_news

今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡
================================== Ai Message ==================================

北京今天天气：晴朗。
今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡。

举例5

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain.messages import HumanMessage
from langchain.tools import tool

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

@tool
def get_news():
    """查询今日国内新闻概要"""
    return ("今日国内新闻概要："
            "中方三艘油轮过航霍尔木兹海峡")

@tool
def calculate(num1: int, num2: int) -> int:
    """
    执行数学计算

    Args:
        num1: 第一个加数
        num2: 第二个加数
    """
    return num1 + num2

@tool
def search_stock(symbol: str):
    """
    查询股票行情

    Args:
        symbol: 股票代码
    """
    return "该股票今天行情不错"

agent = create_agent(
    model=model_out,
    tools=[get_weather, get_news, calculate, search_stock],
    middleware=[
        LLMToolSelectorMiddleware(
            model=model_in,
            max_tools=1,
            always_include=["get_news"]
        ),
    ],
)

response = agent.invoke({
    "messages": HumanMessage("北京今天天气如何？今日新闻概要")
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================
北京今天天气如何？今日新闻概要
================================== Ai Message ==================================
Tool Calls:
get_weather (call_KBQkrc4VvwAWGqJ0Zp1UYqSw)
Call ID: call_KBQkrc4VvwAWGqJ0Zp1UYqSw
 Args:
  city: 北京
 get_news (call_QTXo7yh9nilwVZr6RFTeSRyl)
 Call ID: call_QTXo7yh9nilwVZr6RFTeSRyl
 Args:
================================= Tool Message =================================
Name: get_weather

北京今天天气晴朗
================================= Tool Message =================================
Name: get_news

今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡
================================== Ai Message ==================================

北京今天天气：晴朗。
今日国内新闻概要：中方三艘油轮过航霍尔木兹海峡。

3.5 ToolRetryMiddleware中间件

基于指数退避算法，设置工具调用失败时的重试策略。

指数退避（Exponential Backoff） 的核心思想就是：当某个操作失败（通常是网络请求、API 调用或数据库连接）时，系统不会立刻重试，也不会每次都等待相同的固定时间，而是让每一次重试的延迟时间按指数级增长。

为什么不直接重试？

想象一下，某个热门网站的服务器因为瞬间流量太大（比如抢票或秒杀）崩溃了。如果所有失败的客户端都立刻或每隔1秒就重试一次，这无异于对已经瘫痪的服务器进行了一场持续的 DDoS（分布式拒绝服务）攻击，服务器可能永远也缓不过来。

jitter是为了避免大量工具的重试请求集中在固定的时间点，引入抖动。

假设：按照策略，两次工具调用请求的时间间隔应为10秒，加入抖动后，可能为8.9秒，也可能为10.2秒。

举例1：带抖动

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.messages import HumanMessage
import datetime

def write_times(s):
    """将每次工具调用的时间戳和间隔写入本地文件，方便观察退避策略"""
    with open("call_times_with_jitter.txt", "a", encoding="utf-8") as f:
        f.write(s + "\n")

count = 1
start_time = None

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    global count
    global start_time
    interval = 0
    current_time = datetime.datetime.now()
    if not start_time:
        interval = 0
    else:
        # 计算当前调用与上一次调用之间的时间差（秒）
        interval = (current_time - start_time).total_seconds()
    start_time = current_time
    res_str = f"第 {count} 次调用，当前时间： {start_time}, 和上次调用间隔 {interval} 秒"
    count += 1
    # 记录日志
    write_times(res_str)
    # 故意抛出 TimeoutError，以此触发中间件的重试机制
    raise TimeoutError("Not Implemented")

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        # ToolRetryMiddleware 用于捕获工具执行中的异常并自动重试
        ToolRetryMiddleware(
            max_retries=6, # 最大重试次数（不包含初始的那次调用，一共最多调 1 + 6 = 7 次）
            backoff_factor=2.0, # 指数退避因子（每次重试等待时间乘以 2）
            initial_delay=1.0, # 第一次重试前的初始等待时间（1 秒）
            max_delay=10.0, # 最大等待延迟上限（防止指数增长无限大，限制在 10 秒）
            jitter=True, # 开启抖动（在等待时间中加入随机性，防止并发请求时出现“惊群效 应”）
            retry_on=(TimeoutError,), # 仅针对捕获到特定的 TimeoutError 异常时才 触发重试
            on_failure="continue" # 当达到最大重试次数依然失败时，Agent 的行 为："continue" 表示将错误信息包装后塞回对话历史，让大模型知道失败了并继续决策
        ),
    ],
)

response = agent.invoke({
    "messages": [HumanMessage("今天北京天气如何？")]
})

# 1. 你的提问 -> 2. AI 决定调用工具 -> 3. 重试失败后的错误反馈 -> 4. AI 最终给出的兜 底回复
for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

今天北京天气如何？
================================== Ai Message ==================================
Tool Calls:
get_weather (call_bzolqQ74DGFvIBel5CRt6DlA)
Call ID: call_bzolqQ74DGFvIBel5CRt6DlA
 Args:
  city: 北京
================================= Tool Message =================================
Name: get_weather

Tool 'get_weather' failed after 7 attempts with TimeoutError: Not Implemented. Please try again.
================================== Ai Message ==================================

抱歉，我这边查询北京天气时工具超时了，暂时没拿到实时结果。

如果你愿意，我可以：
1. 你稍后再问一次，我再帮你查；
2. 先告诉你北京今天常见天气注意事项，或者帮你看“适合穿什么”；
3. 你也可以直接发我你看到的天气截图/数据，我帮你解读。

call_times_with_jitter.txt 文件内容

为了精确查看工具调用的时间间隔，以理解各项参数的含义，我们将工具调用时间写入本地文件

 2026-03-31 16:32:15.290785, 和上次调用间隔 0 毫秒
 2026-03-31 16:32:16.334696, 和上次调用间隔 1.043911 毫秒
 2026-03-31 16:32:17.878361, 和上次调用间隔 1.543665 毫秒
 2026-03-31 16:32:21.170138, 和上次调用间隔 3.291777 毫秒
 2026-03-31 16:32:30.240411, 和上次调用间隔 9.070273 毫秒
 2026-03-31 16:32:41.034344, 和上次调用间隔 10.793933 毫秒
 2026-03-31 16:32:50.508267, 和上次调用间隔 9.473923 毫秒

举例2：无抖动

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.messages import HumanMessage
import datetime

def write_times(s):
    """将每次工具调用的时间戳和间隔写入本地文件，方便观察退避策略"""
    with open("call_times_with_jitter.txt", "a", encoding="utf-8") as f:
        f.write(s + "\n")
count = 1
start_time = None

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    global count
    global start_time
    interval = 0
    current_time = datetime.datetime.now()
    if not start_time:
        interval = 0
    else:
        # 计算当前调用与上一次调用之间的时间差（秒）
        interval = (current_time - start_time).total_seconds()
    start_time = current_time
    res_str = f"第 {count} 次调用，当前时间： {start_time}, 和上次调用间隔 {interval} 秒"
    count += 1
    # 记录日志
    write_times(res_str)
    # 故意抛出 TimeoutError，以此触发中间件的重试机制
    raise TimeoutError("Not Implemented")

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        # ToolRetryMiddleware 用于捕获工具执行中的异常并自动重试
        ToolRetryMiddleware(
            max_retries=6, # 最大重试次数（不包含初始的那次调用，一共最多调 1 + 6 = 7 次）
            backoff_factor=2.0, # 指数退避因子（每次重试等待时间乘以 2）
            initial_delay=1.0, # 第一次重试前的初始等待时间（1 秒）
            max_delay=10.0, # 最大等待延迟上限（防止指数增长无限大，限制在 10 秒）
            jitter=False, # 关闭抖动，意味着重试机制从“随机化的指数退避”退化成了“严格 固定的指数退避”
            retry_on=(TimeoutError,), # 仅针对捕获到特定的 TimeoutError 异常时才 触发重试
            on_failure="continue" # 当达到最大重试次数依然失败时，Agent 的行 为："continue" 表示将错误信息包装后塞回对话历史，让大模型知道失败了并继续决策
        ),
    ],
)

response = agent.invoke({
    "messages": [HumanMessage("今天北京天气如何？")]
})

# 1. 你的提问 -> 2. AI 决定调用工具 -> 3. 重试失败后的错误反馈 -> 4. AI 最终给出的兜 底回复
for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

今天北京天气如何？
================================== Ai Message ==================================
Tool Calls:
get_weather (call_KnGIyEMkbvToYPTCEofCu3Ub)
Call ID: call_KnGIyEMkbvToYPTCEofCu3Ub
 Args:
  city: 北京
================================= Tool Message =================================
Name: get_weather

Tool 'get_weather' failed after 7 attempts with TimeoutError: Not Implemented. Please try again.
================================== Ai Message ==================================

抱歉，我暂时无法获取到北京的实时天气（天气查询接口未成功返回结果）。

如果你愿意，我可以：
1. 帮你整理一个“北京天气查询”的快捷方式；
2. 根据季节给你一份今天出门穿衣建议；
3. 你也可以稍后再问我一次，我再帮你查。

call_times_without_jitter.txt文件内容

 2026-03-31 16:32:58.801876, 和上次调用间隔 0 毫秒
 2026-03-31 16:32:59.808856, 和上次调用间隔 1.00698 毫秒
 2026-03-31 16:33:01.817964, 和上次调用间隔 2.009108 毫秒
 2026-03-31 16:33:05.828532, 和上次调用间隔 4.010568 毫秒
 2026-03-31 16:33:13.835262, 和上次调用间隔 8.00673 毫秒
 2026-03-31 16:33:23.845651, 和上次调用间隔 10.010389 毫秒
 2026-03-31 16:33:33.855016, 和上次调用间隔 10.009365 毫秒

自动重试失败的工具调用。

第 1 次重试（retry_number=1）: 等待 ~ 1.0 * (2.0 ** 1) = 2.0 秒

第 2 次重试（retry_number=2）: 等待 ~ 1.0 * (2.0 ** 2) = 4.0 秒

第 3 次重试（retry_number=3）: 等待 ~ 1.0 * (2.0 ** 3) = 8.0 秒

也就是说，等待时间以指数方式增长 —— 每失败一次，下次再试之前等待更长时间。

如果你把 backoff_factor = 0，就意味着不使用指数增长，重试之间始终用固定的 initial_delay。

from langchain.agents.middleware import ToolRetryMiddleware

retry = ToolRetryMiddleware(
    max_retries=3,
    backoff_factor=2.0,  # 指数退避因子
    retry_on_exceptions=[ConnectionError, TimeoutError]
)

agent = create_agent(
    model="openai:gpt-4o",
    tools=[web_search, api_call],
    middleware=[retry]
)

将 jitter 从 True 改为 False （关闭抖动），意味着重试机制从“随机化的指数退避”退化成了“严格固定的指数退避”。

为了更直观理解，看一下这两种状态下的核心区别：

1. 理论上的等待时间对比

在这段代码中，你设置了 initial_delay=1.0 （初始延迟 1 秒）、backoff_factor=2.0 （倍数是 2）以及 max_delay=10.0 （最大延迟 10 秒）。

当工具持续报错时，关闭抖动（jitter=False ）与开启抖动（jitter=True ）的等待延迟（Interval）对比如下：

重试轮	理想基础延迟（秒）	关闭抖动 (jitter=False) 的实际等待	开启抖动 (jitter=True) 的实际等待
第 1 次重试	1.0	严格等于 1.0 秒	在随机区间内取值
第 2 次重试	2.0	严格等于 2.0 秒	在随机区间内取值
第 3 次重试	4.0	严格等于 4.0 秒	在随机区间内取值
第 4 次重试	8.0	严格等于 8.0 秒	在随机区间内取值
第 5 次重试	10.0（受限于 max_delay）	严格等于 10.0 秒	在随机区间内取值
第 6 次重试	10.0（受限于 max_delay）	严格等于 10.0 秒	在随机区间内取值

💡 现象结论：

关闭抖动后，查看生成的 call_times_with_jitter.txt 日志，你会发现输出的 interval 数字会极其精准地趋近于 1.0 、2.0 、4.0 、8.0 、10.0 、10.0 。

2. 为什么要引入 Jitter（抖动）？关闭它会有什么问题？

在单用户、单并发的测试环境下，关闭 jitter没有任何副作用，甚至能让等待时间非常规律、可预测。

但在高并发的生产环境中，关闭 jitter会引发灾难性的“惊群效应（Thundering Herd Problem）” ：

• 没有 Jitter 的惨剧（jitter=False ）：

假设某刻天气 API 服务突然宕机了 1 秒。此时刚好有 1000 个用户同时发起了查询。因为这 1000

个请求同时失败，并且它们都严格死板地等待 1 秒、2 秒、4 秒……

这意味着，在第 1 秒、第 3 秒、第 7 秒的那个精准的时间点上，这 1000 个请求会整整齐齐地再次

同时轰炸服务器。刚刚复活的服务器瞬间又被这波整齐的峰值流量压垮，形成恶性循环。

• 引入 Jitter 的优势（jitter=True ）：

通过给重试时间加上随机性，这 1000 个请求会在 秒、 秒的区间内均匀地错开（削峰

填谷）。流量被平摊到了整条时间轴上，服务器就能轻松地分批处理完这些请求。

总结：

• jitter=False （你当前的代码）：重试间隔死板、精准、可预测。适合本地调试、测试重试逻辑是否生效。
• jitter=True ：重试间隔随机、错开、更安全。适合线上生产环境，防止把下游第三方 API 或数据库冲垮。

3.6 ModelRetryMiddleware中间件

模型调用失败时重试，策略和工具调用的重试一样，都是基于指数退避算法。

因此，本节案例不再重点观察指数退避算法，而是测试不同的退出模式。

举例1：继续运行

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware
from langchain.messages import HumanMessage

from dotenv import load_dotenv

load_dotenv(override=True)

agent = create_agent(
    model="deepseek-cat",
    middleware=[
        ModelRetryMiddleware(
            max_retries=6,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=10.0,
            on_failure="continue",
            jitter=False,
        ),
    ],
)

response = agent.invoke({
    "messages": [HumanMessage("你好")]
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

你好
================================== Ai Message ==================================

Model call failed after 7 attempts with BadRequestError: Error code: 400 - {'error': {'message': 'The supported API model names are deepseek-v4- pro or deepseek-v4-flash, but you passed deepseek-cat.', 'type': 'invalid_request_error', 'param': None, 'code': 'invalid_request_error'}}

举例2：抛异常

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware
from langchain.messages import HumanMessage

from dotenv import load_dotenv

load_dotenv(override=True)

agent = create_agent(
    model="deepseek-cat",
    middleware=[
        ModelRetryMiddleware(
            max_retries=6,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=10.0,
            on_failure="error",
            jitter=False,
        ),
    ],
)

response = agent.invoke({
    "messages": [HumanMessage("你好")]
})

for msg in response["messages"]:
    msg.pretty_print()

输出

Traceback...
BadRequestError: Error code: 400 - {'error': {'message': 'Model Not Exist', 'type': 'invalid_request_error', 'param': None, 'code': 'invalid_request_error'}}
During task with name 'model' and id 'd056329f-a93f-5a8d-456f- 63d194b87e6a'

3.7 LLMToolEmulator中间件

某些情况下，工具尚未开发完成，我们希望先测试工具调用，可以用LLM tool emulator模拟工具。

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
from langchain.messages import HumanMessage

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

agent = create_agent(
    model=model_out,
    tools=[get_weather],
    middleware=[
        LLMToolEmulator(
            model=model_in,
        )
    ]
)

response = agent.invoke({
    "messages": [HumanMessage("今天北京天气如何")]
})

for msg in response["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

今天北京天气如何
================================== Ai Message ==================================
Tool Calls:
get_weather (call_91l1vNSrhngCocth4jm1ILni)
Call ID: call_91l1vNSrhngCocth4jm1ILni
Args:
 city: 北京
================================= Tool Message =================================
Name: get_weather

{
"city": "北京",
"date": "2025-04-12",
"weather": "多云转晴",
"temperature": {
 "current": 18,
 "high": 22,
 "low": 11
},
"humidity": "45%",
"wind": {
 "direction": "西北风",
 "speed": "3-4级"
},
"aqi": 85,
"sunrise": "05:37",
"sunset": "18:49",
"recommendation": "昼夜温差较大，建议携带外套"
}
================================== Ai Message ==================================

北京今天天气：**多云转晴**。

- **当前气温**：18℃
- **最高/最低**：22℃ / 11℃
- **湿度**：45%
- **风力**：西北风 3-4级
- **空气质量**：AQI 85，属于一般
- **日出/日落**：05:37 / 18:49

**建议**：昼夜温差较大，出门最好带件外套。

3.8 ContextEditingMiddleware中间件

上下文编辑中间件，该中间件提供了上下文管理的一种方式。

通过更改发送给模型的消息列表来控制成本。

注意：不会更改消息列表。因此我们只能通过token用量来推测是否对消息列表进行了裁剪。

1. 实验组-启用上下文编辑

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit
from langchain.messages import HumanMessage, AIMessage
from langgraph.checkpoint.memory import InMemorySaver

from dotenv import load_dotenv

load_dotenv()

count = 0

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    global count
    return (f"当前是第 {count} 次调用工具，{city}今天天气晴朗"
            f"天气非常好，北风，非常适合出行，盼望着，盼望着，"
            f"春天来了。我喜欢春天，你喜欢吗，天气真的很不错"
            f"万里无云，天气晴朗，春和景明，哈哈哈哈哈哈，这是凑字数的"
            f"真不错，天气非常好，适合出行，这里token挺多的"
            f"可以出门玩，尅有跑步，钓鱼，爬山，一切都很好哈哈哈")
agent = create_agent(
    model="deepseek-chat",
    tools=[get_weather],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=50,
                    keep=0,
                ),
            ],
        ),
    ],
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "1"}}

for i in range(3):
    print("=" * 30, f"当前是第 {i + 1} 轮调用", "=" * 30)
    count = i + 1
    response = agent.invoke({
        "messages": [HumanMessage(f"第 {i + 1} 次询问：今天北京天气如何，一句话回 答")]},
        config=config
    )
    print("---- 本次返回的 messages ----")
    for msg in response["messages"]:
        # msg.pretty_print()
        if isinstance(msg, AIMessage):
            if not msg.tool_calls:
                print(f"本次token用量：{msg.usage_metadata}")

输出

============================== 当前是第 1 轮调用 ==============================
---- 本次返回的 messages ----
本次token用量：{'input_tokens': 169, 'output_tokens': 17, 'total_tokens': 186, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
============================== 当前是第 2 轮调用 ==============================
---- 本次返回的 messages ----
本次token用量：{'input_tokens': 169, 'output_tokens': 17, 'total_tokens': 186, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
本次token用量：{'input_tokens': 239, 'output_tokens': 18, 'total_tokens': 257, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
============================== 当前是第 3 轮调用 ==============================
---- 本次返回的 messages ----
本次token用量：{'input_tokens': 169, 'output_tokens': 17, 'total_tokens': 186, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
本次token用量：{'input_tokens': 239, 'output_tokens': 18, 'total_tokens': 257, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
本次token用量：{'input_tokens': 309, 'output_tokens': 18, 'total_tokens': 327, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

说明：

1. ContextEditingMiddleware 的价值：大模型多轮对话时，如果频繁调用产生大量文本的工具

（如代码执行、网页爬取），历史记录会急剧膨胀。这个中间件就像一个“上下文抽脂手术”，在不

影响当前对话的前提下，自动在后台删掉之前沉淀的工具调用废话，从而极大地节省 Token 费用

并防止超出模型最大上下文窗口（Context Window）。

2. InMemorySaver ：它在内存中开辟了一个空间。第二轮和第三轮提问时，Agent 能通过

thread_id 自动找回前几轮的记忆。

2. 对照组-不裁剪上下文

from langchain.agents import create_agent
from langchain.messages import HumanMessage, AIMessage
from langgraph.checkpoint.memory import InMemorySaver

# 全局计数器，用于在工具内部追踪这是第几次被触发
count = 0

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    global count

    # 故意返回一段非常冗长、包含大量 Token 的文本，用于测试中间件的 Token 清理/截断功能
    return (f"当前是第 {count} 次调用工具，{city}今天天气晴朗"
            f"天气非常好，北风，非常适合出行，盼望着，盼望着，"
            f"春天来了。我喜欢春天，你喜欢吗，天气真的很不错"
            f"万里无云，天气晴朗，春和景明，哈哈哈哈哈哈，这是凑字数的"
            f"真不错，天气非常好，适合出行，这里token挺多的"
            f"可以出门玩，尅有跑步，钓鱼，爬山，一切都很好哈哈哈")

agent = create_agent(
    model=model,
    tools=[get_weather],
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "1"}}

for i in range(3):
    print("=" * 30, f"当前是第 {i + 1} 轮调用", "=" * 30)
    count = i + 1
    response = agent.invoke({
        "messages": [HumanMessage(f"第 {i + 1} 次询问：今天北京天气如何，一句话回 答")]},
        config=config
    )
    print("---- 本次返回的 messages ----")
    for msg in response["messages"]:
        # msg.pretty_print()
        if isinstance(msg, AIMessage):
            if not msg.tool_calls:
                print(f"本次token用量：{msg.usage_metadata}")

输出

============================== 当前是第 1 轮调用 ==============================
---- 本次返回的 messages ----
本次token用量：{'input_tokens': 280, 'output_tokens': 17, 'total_tokens': 297, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
============================== 当前是第 2 轮调用 ==============================
---- 本次返回的 messages ----
本次token用量：{'input_tokens': 280, 'output_tokens': 17, 'total_tokens': 297, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
本次token用量：{'input_tokens': 460, 'output_tokens': 17, 'total_tokens': 477, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
============================== 当前是第 3 轮调用 ==============================
---- 本次返回的 messages ----
本次token用量：{'input_tokens': 280, 'output_tokens': 17, 'total_tokens': 297, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
本次token用量：{'input_tokens': 460, 'output_tokens': 17, 'total_tokens': 477, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
本次token用量：{'input_tokens': 640, 'output_tokens': 17, 'total_tokens': 657, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

可以观察到，对照组的input_tokens明显大于实验组。

3.9 FilesystemFileSearchMiddleware中间件

基于系统的Glob和Grep检索工具，为Agent赋予本地文件搜索和分析的能力。

• Glob根据文件路径检索
• Grep根据文件内容检索

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware
from langchain.messages import HumanMessage

agent = create_agent(
    model=model,
    tools=[],  # 自动添加 Glob 和 Grep 工具
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="../todo_workspace",  #搜索目录
            # 【可选】限制搜索的文件后缀，防止模型读取非代码或无关文件
            # allowed_extensions=[".py", ".ipynb", ".js", ".md"],  # 允许的文 件类型
            # 是否启用 ripgrep 搜索引擎：
            # 设为 True 可以获得比原生 Grep 更快的性能（前提是系统已安装 ripgrep）
            use_ripgrep=True,
            # 单个文件的最大读取限制（单位MB）：防止读取超大型日志或二进制文件导致 OOM
            max_file_size_mb=10
        ),
    ],
)

result = agent.invoke({
    "messages": [HumanMessage("找到包含add函数的Python或Jupyter文件")]
})

for msg in result["messages"]:
    msg.pretty_print()

输出

================================ Human Message =================================

找到包含add函数的Python或Jupyter文件
================================== Ai Message ==================================
Tool Calls:
glob_search (call_9EaFoPbjcs4Am5MQDcDXySd9)
Call ID: call_9EaFoPbjcs4Am5MQDcDXySd9
 Args:
  pattern: **/*.py
   path: /
  glob_search (call_MpTXhDYcuduxnVjdx6yTuKq9)
Call ID: call_MpTXhDYcuduxnVjdx6yTuKq9
Args:
 pattern: **/*.ipynb
 path: /
================================= Tool Message =================================
Name: glob_search

/my_add.py
/test_my_add.py
================================= Tool Message =================================
Name: glob_search

No files found
================================== Ai Message ==================================

找到包含 `add` 函数的 Python 文件：

- `/my_add.py`
- `/test_my_add.py`

没有找到 Jupyter 文件（`.ipynb`）。

3.10 Shell tool中间件

为Agent提供一个可以执行命令的Shell环境。

Windows下无法测试。

3.11 Filesystem中间件

这是源自deepagents（基于LangChain的另一个框架）的中间件

内置了四个工具，分别用于查看目录、读文件、写文件和改文件。

3.12 Subagent中间件

也是来自deepagents的中间件

用于便捷地创建子Agent。