中间件

中间件概述

什么是中间件

中间件是 Agent 的 “中枢”，在执行流程的关键节点进行拦截、控制和增强。
中间件是 Agent 执行过程中的钩子函数，是 LangChain 1.x 的王牌工程化能力。
下面两张图是没有中间件和有中间件 Agent 结构图

为什么需要中间件

想根据问题复杂度动态 切换模型。
想 限制 某些用户只能调用部分工具。
想在工具报错时 自动重试 或返回兜底结果。
想在模型调用前 插入额外的系统提示。
想记录每一步的 执行日志，方便排查问题。
想在敏感信息出现时 阻断执行。
想在正式执行工具前增加 人工审批 。

这些需求有一个共同特点：它们不是 Agent 的核心业务逻辑，但又会影响 Agent 的执行过程。如果把这些逻辑全部直接写进主流程，会带来几个问题：

主流程会迅速变乱。
很多逻辑是横切需求，难以复用。
流程控制粒度不够细。
后期维护成本高。

中间件的分类

自定义中间件：允许开发者自定义，从而实现更加灵活的 Agent 行为管理。
内置中间件：LangChain 提供的与供应商无关的中间件。

官方文档：https://docs.langchain.com/oss/python/langchain/middleware/overview

中间件	描述	常用
SummarizationMiddleware	对历史消息列表进行摘要&总结，达到压缩上下文的效果。	✅
HumanInTheLoopMiddleware	在工具调用前中断 Agent 的运行	✅
PIIMiddleware	敏感信息保护	✅
TodoListMiddleware	赋予了 Agent `任务规划` 和 `追踪进度` 的能力	✅
ModelCallLimitMiddleware	限制`模型`调用次数	❌
ToolCallLimitMiddleware	限制`工具`调用次数	❌
ModelFallbackMiddleware	备用模型	❌
LLMToolSelectorMiddleware	智能工具筛选	❌
ToolRetryMiddleware	工具调用失败的重试策略	❌
ModelRetryMiddleware	模型调用失败的重试策略	❌
LLMToolEmulator	测试工具调用	❌
ContextEditingMiddleware	管理对话上下文长度	❌
FilesystemFileSearchMiddleware	本地文件搜索和分析	❌
ToolErrorMiddleware	捕获工具异常并转为模型可读的错误信息	❌
ProviderToolSearchMiddleware	按需加载工具 schema，减小请求体积	❌
ShellToolMiddleware	为 Agent 提供持久化 Shell 会话	❌
FilesystemMiddleware	为 Agent 提供文件读写与搜索工具集	❌
SubagentMiddleware	让主 Agent 能够委派子任务给子 Agent	❌
RubricGradingMiddleware	让 Agent 按评分标准自我迭代	❌

SummarizationMiddleware

作用：对历史消息列表进行 摘要&总结，达到 压缩上下文 的效果。
原理：在 达到触发条件 时，调用大模型对历史消息进行摘要，将摘要的结果作为 HumanMessage，放到消息列表最开始的位置。

参数：

model：用于摘要的模型。可以是字符串或模型对象。
trigger：摘要的触发条件。下面三个条件满足一个即触发：
- tokens：token的数量，历史token的累计数量达到该值触发摘要。
- messages：历史消息数量，历史消息条数达到该值触发摘要。
- fraction：上下文长度比例。历史token的累计数量达到模型的 max_input_tokens * fraction 触发摘要。
keep：摘要时保留的原始消息。下面三个条件只接受一种：
- tokens：摘要时保留的token数量。
- messages：摘要时保留的历史消息条数。
- fraction：摘要时保留 max_input_tokens*fraction 个token。
token_counter：统计token数量的函数。
summary_prompt：摘要时的自定义提示词。该提示词需要包含 {messages} 占位符，使得历史消息列表可以被插入。
trim_token_to_summarize：摘要时历史消息的最大token数，默认为 4000。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

load_dotenv(verbose=True)

# 用于摘要的模型
model = init_chat_model(
    model="deepseek-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    profile={
        "max_input_tokens": 128_000
    }
)

messages = [
    SystemMessage("你是个非常有好的AI助手"),
    HumanMessage("你好啊，我是老王，你是谁？"),
    AIMessage("你好老王，我是小王"),
    HumanMessage("好的小王，很高兴认识你"),
    AIMessage("你高兴得太早了"),
    HumanMessage("呵呵，你什么意思，你是谁？")
]

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=[
                ("tokens", 100),
                ("messages", 6),
                ("fraction", 0.0001)
            ],
            keep=("messages", 2),
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}"
        )
    ]
)

res = agent.invoke({
    "messages": messages
})

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

""" 输出结果：
================================ Human Message =================================

Here is a summary of the conversation to date:

历史消息摘要：
- 用户自称“老王”，询问 AI 身份。
- AI 自称“小王”，与老王互相问候。
- 老王表示很高兴认识小王。
- 当前关系：双方已完成初步认识，气氛友好。
================================== Ai Message ==================================

你高兴得太早了
================================ Human Message =================================

呵呵，你什么意思，你是谁？
================================== Ai Message ==================================

老王，别误会，我是你的 AI 助手。刚才那句“你高兴得太早了”是随口接梗，开玩笑开过头了，不合适，我道个歉。

之前说叫我“小王”也行，但我先说明白：我是人工智能，不是真人。你要愿意继续叫我小王，我就应着；你要问正事，我也正经答。
"""

HumanInTheLoopMiddleware

在工具调用前中断 Agent 的运行，等待用户对工具调用请求决策。可选的决策有：approve（同意执行）、edit（编辑调用配置后执行）、reject（拒绝执行）。

参数：

interrupt_on：工具名和中断策略的映射，可选值：
- True：所有决策都可以选。
- False：不中断。
- InterruptOnConfig：是一个TypedDict的子类，可以用字典直接赋值。支持的Key有：
  - allowed_decisions：精细控制中断后允许的决策
  - description：特定工具的中断描述信息，优先级高于 description_prefix。
description_prefix：自定义中断描述

main.py

python

HumanInTheLoopMiddleware(
   interrupt_on={
      "tool_name_1": True,
      "tool_name_2": False,
      "tool_name_3": {
         "allowed_decisions": ["approve", "reject"],
         "description": "tool_name_3 中断了，可以选择放行或拒绝执行"
      }
   },
   description_prefix="工具调用中断"   
)

调用前中断

创建 Agent 时通过 checkpointer 参数启用了短期记忆，在调用时通过传递相同的 config 加载记忆。

调用前.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv(verbose=True)

@tool
def get_weather(city: str, next_day: bool):
    """天气查询工具

    Args:
        city: 城市
        next_day: 是否显示次日天气
    """
    res = f"{city}天气很好"
    if next_day:
        res += "，明天有雨"
    return res

@tool
def get_news():
    """查询今日新闻"""
    return "今天没啥新闻"

@tool
def read_email(email_id: str):
    """查看邮件

    Args:
        email_id: 邮件 ID
    """
    return f"邮件ID：{email_id} 是空的"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[get_weather, get_news, read_email],
    checkpointer=InMemorySaver(),   # 短期记忆
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
                "get_news": False,
                "read_email": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "read_email 中断了，可以选择放行或拒绝执行"
                },
            },
            description_prefix="工具调用中断了"
        )
    ]
)

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

res = agent.invoke({
    "messages": ["请帮我查询今天北京的天气，查询今日新闻，查看ID为 'sk2131421' 的邮件内容。这几件事同时执行"]
}, config=agent_config)

rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(...),
        AIMessage(
            content='我来同时执行这三项任务：',
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'city': '北京', 'next_day': False},
                    'id': 'call_00_uwsc27OYMLan9ZTZo7kZ6565',
                    'type': 'tool_call'
                },
                {
                    'name': 'get_news',
                    'args': {},
                    'id': 'call_01_no5rlxNYL01JjmiFUKxM4799',
                    'type': 'tool_call'
                },
                {
                    'name': 'read_email',
                    'args': {'email_id': 0},
                    'id': 'call_02_w1Nf30lxqrjKAuLRLBp24852',
                    'type': 'tool_call'
                }
            ],
            ...
        )
    ],
    '__interrupt__': [
        Interrupt(
            value={
                'action_requests': [
                    {
                        'name': 'get_weather',
                        'args': {'city': '北京', 'next_day': False},
                        'description': "工具调用中断了\n\nTool: get_weather\nArgs: {'city': '北京', 'next_day': False}"
                    },
                    {
                        'name': 'read_email',
                        'args': {'email_id': 0},
                        'description': 'read_email 中断了，可以选择放行或拒绝执行'
                    }
                ],
                'review_configs': [
                    {
                        'action_name': 'get_weather',
                        'allowed_decisions': ['approve', 'edit', 'reject']
                    },
                    {
                        'action_name': 'read_email',
                        'allowed_decisions': ['approve', 'reject']
                    }
                ]
            },
            id='d045f4851cb3aa54e7939762282949ae'
        )
    ]
}
"""

处理中断

处理中断.py

完整代码.py

python

# 处理中断
action_requests = res.get("__interrupt__")[0].value["action_requests"]
if action_requests:
    decisions = []
    for ar in action_requests:
        if ar["name"] == "get_weather":
            decisions.append({
                "type": "edit",
                "edited_action": {
                    "name": "get_weather",
                    "args": {"city": "大连市", "next_day": True}
                }
            })
        elif ar["name"] == "read_email":
            decisions.append({"type": "approve"})
    res = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=agent_config
    )
    for msg in res["messages"]:
        msg.pretty_print()

""" 输出结果：
================================ Human Message =================================

请帮我查询今天北京的天气，查询今日新闻，查看ID为 'sk2131421' 的邮件内容。这几件事同时执行
================================== Ai Message ==================================

我来同时为您执行这三个查询。
Tool Calls:
  get_weather (call_00_9fYSOeH4CXKF7rXh4bvj3299)
 Call ID: call_00_9fYSOeH4CXKF7rXh4bvj3299
  Args:
    city: 大连市
    next_day: True
  get_news (call_01_pNOEPWqvIIqEpsnTUNeR6082)
 Call ID: call_01_pNOEPWqvIIqEpsnTUNeR6082
  Args:
  read_email (call_02_vwkmov1PlnzrSG1cU0ms4445)
 Call ID: call_02_vwkmov1PlnzrSG1cU0ms4445
  Args:
    email_id: sk2131421
================================= Tool Message =================================
Name: get_weather

大连市天气很好，明天有雨
================================= Tool Message =================================
Name: get_news

今天没啥新闻
================================= Tool Message =================================
Name: read_email

邮件ID：sk2131421 是空的
================================== Ai Message ==================================

抱歉，我刚才把城市填错了（填成了大连），我重新查询北京的天气：
Tool Calls:
  get_weather (call_00_UJ5GyfhTNkVAlz1BxsXw6913)
 Call ID: call_00_UJ5GyfhTNkVAlz1BxsXw6913
  Args:
    city: 北京市
    next_day: True
"""

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

load_dotenv(verbose=True)

@tool
def get_weather(city: str, next_day: bool):
    """天气查询工具

    Args:
        city: 城市
        next_day: 是否显示次日天气
    """
    res = f"{city}天气很好"
    if next_day:
        res += "，明天有雨"
    return res

@tool
def get_news():
    """查询今日新闻"""
    return "今天没啥新闻"

@tool
def read_email(email_id: str):
    """查看邮件

    Args:
        email_id: 邮件 ID
    """
    return f"邮件ID：{email_id} 是空的"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[get_weather, get_news, read_email],
    checkpointer=InMemorySaver(),   # 短期记忆
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
                "get_news": False,
                "read_email": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "read_email 中断了，可以选择放行或拒绝执行"
                },
            },
            description_prefix="工具调用中断了"
        )
    ]
)

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

res = agent.invoke({
    "messages": ["请帮我查询今天北京的天气，查询今日新闻，查看ID为 'sk2131421' 的邮件内容。这几件事同时执行"]
}, config=agent_config)

# 处理中断
action_requests = res.get("__interrupt__")[0].value["action_requests"]
if action_requests:
    decisions = []
    for ar in action_requests:
        if ar["name"] == "get_weather":
            decisions.append({
                "type": "edit",
                "edited_action": {
                    "name": "get_weather",
                    "args": {"city": "大连市", "next_day": True}
                }
            })
        elif ar["name"] == "read_email":
            decisions.append({"type": "approve"})
    res = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=agent_config
    )
    for msg in res["messages"]:
        msg.pretty_print()

""" 输出结果：
================================ Human Message =================================

请帮我查询今天北京的天气，查询今日新闻，查看ID为 'sk2131421' 的邮件内容。这几件事同时执行
================================== Ai Message ==================================

我来同时为您执行这三个查询。
Tool Calls:
  get_weather (call_00_9fYSOeH4CXKF7rXh4bvj3299)
 Call ID: call_00_9fYSOeH4CXKF7rXh4bvj3299
  Args:
    city: 大连市
    next_day: True
  get_news (call_01_pNOEPWqvIIqEpsnTUNeR6082)
 Call ID: call_01_pNOEPWqvIIqEpsnTUNeR6082
  Args:
  read_email (call_02_vwkmov1PlnzrSG1cU0ms4445)
 Call ID: call_02_vwkmov1PlnzrSG1cU0ms4445
  Args:
    email_id: sk2131421
================================= Tool Message =================================
Name: get_weather

大连市天气很好，明天有雨
================================= Tool Message =================================
Name: get_news

今天没啥新闻
================================= Tool Message =================================
Name: read_email

邮件ID：sk2131421 是空的
================================== Ai Message ==================================

抱歉，我刚才把城市填错了（填成了大连），我重新查询北京的天气：
Tool Calls:
  get_weather (call_00_UJ5GyfhTNkVAlz1BxsXw6913)
 Call ID: call_00_UJ5GyfhTNkVAlz1BxsXw6913
  Args:
    city: 北京市
    next_day: True
"""

PIIMiddleware

敏感信息保护。PII中间件用于检测和处理对话中的个人身份信息，支持自定义处理策略。

参数：

pii_type：检测的PII数据类型。
- email：电子邮箱地址
- credit_card：信用卡号
- url：网址
- mac_address：设备MAC地址
- ip：IP地址

strategy：处理PII信息的策略。

取值	描述	适用场景
`redact`	用字符串 `[REDACTED_[PII_TYPE]]` 替换，完全 `擦除/隐藏` 真实内容	日志清洗、合规需求、公开输出时隐藏敏感内容。
`mask`	用 `***` 将PII信息的前面一部分信息遮蔽。	用户服务界面 / 前端显示 / 需要部分可识别但不泄露完整敏感内容的场景。
`hash`	用哈希值替代原值。	analytics、调试 (debug)、统计分析、匿名追踪等场景。
`block`	直接抛出异常	对隐私要求极高、绝不允许泄露任何敏感信息的场景。

detector：自定义 PII检测函数 或者 正则表达式。
apply_to_input：是否在调用模型前检测，默认为 True。
apply_to_output：是否在模型调用后检测，默认为 False。
apply_to_tool_results：是否在工具调用后检测其输出，默认为 False。

内置检测器.py

自定义检测器.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        PIIMiddleware(pii_type="email", strategy="redact"),
        PIIMiddleware(pii_type="credit_card", strategy="mask"),
        PIIMiddleware(pii_type="url", strategy="hash"),
        PIIMiddleware(pii_type="mac_address", strategy="mask"),
        PIIMiddleware(pii_type="ip", strategy="block"),
    ]
)

res = agent.invoke({
    "messages": ["帮我向 156168188@qq.com 发送一封邮件，同时查看银行卡号： 5105-1051-0510-5100 的余额，访问 https://localhost:12345，确认这是不是 MAC地址： 11-11-11-11-11-11。"]
})

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

try:
    res = agent.invoke({
        "messages": ["看看这个 IP 能不能 ping 通：192.168.10.1"]
    })
except Exception as e:
    print('=' * 33, '-> 抛异常 <-', '=' * 33)
    print(f"检测到IP，抛出异常：{e}")

""" 输出结果：
================================ Human Message =================================

帮我向 [REDACTED_EMAIL] 发送一封邮件，同时查看银行卡号： ****-****-****-5100 的余额，访问 <url_hash:de479c42> MAC地址： **-**-**-**-**-11。
================================== Ai Message ==================================

......
================================= -> 抛异常 <- ==================================
检测到IP，抛出异常：Detected 1 instance(s) of ip in text content
"""

python

import os
import re
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

def detect_phone_number(content: str):
    return [
        {
            "text": m.group(0),  # 提取出具体匹配到的 11 位数字文本（例如"13800138000"）
            "start": m.start(),  # 这段数字在原文本中的“起始索引位置”（从 0 开始算）
            "end": m.end()  # 这段数字在原文本中的“结束索引位置”
        } for m in re.finditer(r"[0-9]{11}", content)
    ]

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        PIIMiddleware(pii_type="api_key", strategy="hash", detector=r"sk-[a-zA-Z0-9]+"),
        PIIMiddleware(pii_type="phone_number", strategy="mask", detector=detect_phone_number),
    ]
)

res = agent.invoke({
    "messages": ["这是不是有效的 API_KEY： sk-awef23AFEfaafaefa，帮我给这个号码打电话： 12345612345，访问 https://localhost:12345。"]
})

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

""" 输出结果：
================================ Human Message =================================

这是不是有效的 API_KEY： <api_key_hash:6c678cc0>，帮我给这个号码打电话： ****2345，访问 https://localhost:12345。
================================== Ai Message ==================================

......
"""

TodoListMiddleware

TodoListMiddleware 中间件赋予了 Agent 任务规划 和 追踪进度 的能力，可以 应对复杂的多步任务。

你的任务是否需要拆解？
   ├── 否 (比如：问答、翻译、单次函数调用) ──> ❌ 绝不需要，浪费算力
   └── 是 (比如：写一个包含多文件的工程)
      └── 步骤是否多变且需要应对失败？
         ├── 否 (步骤完全固定，如 A->B->C) ──> ❌ 传统的LangGraph线性节点即可
         └── 是 (AI 需要边做边调计划) ──> 引入 TodoListMiddleware

经典场景：

任务链路长、步骤多，且有严格的先后依赖关系。
需要在前端 UI 界面实时展示 Agent 的“思考与执行进度”

参数：

system_prompt：自定义指导todo列表使用的提示词，通常不用设置。
tool_description：自定义write_tools工具的描述信息，通常不用设置。

需求：创建目录，创建文件 html 文件并写入内容。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from pathlib import Path
from rich import print as rprint

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def create_dir(dir_name: str):
    """创建目录工具

    Args:
        dir_name: 目录名
    """
    dp = f"/Users/igma/Mac/Python/LangChain1.2/{dir_name}"
    Path(dp).mkdir(parents=True, exist_ok=True)
    return dp

@tool(parse_docstring=True)
def create_file(dir_path: str, file_name: str):
    """创建文件

    Args:
        dir_path: 文件路径
        file_name: 文件名
    """
    p = Path(Path(dir_path) / file_name)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.touch()
    return p

@tool(parse_docstring=True)
def append_to_file(file_path: str, content: str) -> str:
    """追加写入文件

    Args:
        file_path: 文件路径
        content: 文件内容
    """
    path = Path(file_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as f:
        f.write(content)
    return f"写入成功：{path}"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    tools=[create_dir, create_file, append_to_file],
    middleware=[TodoListMiddleware()],
    system_prompt="你是一个程序员。无论任务简单与否，每次执行前都必须先调用 write_todos 工具列出待办事项，然后再执行具体操作。工作全部在工作区下进行。"
)

res = agent.invoke({
    "messages": ["1. 创建一个 web 目录；2. 在目录中创建一个 index.html 文件；3.文件内容帮我写一个Hello页面。不需要样式"]
})

rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='1. 创建一个 web 目录；2. 在目录中创建一个 index.html 文件；3.文件内容帮我写一个Hello页面。不需要样式',
            additional_kwargs={},
            response_metadata={},
            id='f5db4831-e976-4351-ae50-f9d17e19baa5'
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'write_todos',
                    ...
                    'type': 'tool_call'
                }
            ],
        ),
        ToolMessage(...),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'create_dir',
                    ...
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(...),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'write_todos',
                    ...
                    'type': 'tool_call'
                },
                {
                    'name': 'create_file',
                    ...
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(...),
        ToolMessage(...),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'write_todos',
                    ...
                    'type': 'tool_call'
                },
                {
                    'name': 'append_to_file',
                    ...
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(...),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'write_todos',
                    ...
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(...),
        AIMessage(
            content='全部完成 ✅\n\n1. ',
            ...
            tool_calls=[],
            ...
        )
    ],
    'todos': [
        {'content': '创建 web 目录', 'status': 'completed'},
        {
            'content': '在 web 目录中创建 index.html 文件',
            'status': 'completed'
        },
        {
            'content': '向 index.html 写入 Hello 页面内容',
            'status': 'completed'
        }
    ]
}
"""

ModelCallLimitMiddleware

限制模型调用次数。

参数：

thread_limit：线程级限制。整个对话线程（同一 thread_id）内允许的最大模型调用次数。
run_limit：运行级限制。单次 invoke 调用内允许的最大模型调用次数，每轮对话重置。
exit_behavior：达到限制后的运行方式，可选值：
- end：直接结束
- error：抛异常

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from rich import print as rprint

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def get_weather(dt: str):
    """查询天气工具

    Args:
        dt: 查询模式，可选值：今天 or 明天
    """
    return f"{dt}天气很好"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        ModelCallLimitMiddleware(
            run_limit=1,
            exit_behavior="end"
        )
    ],
    tools=[get_weather]
)

res = agent.invoke({
    "messages": ["帮我做两件事：1.查一下今天的天气；2.查一下明天的天气。两件事同时进行"]
})
rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='帮我做两件事：1.查一下今天的天气；2.查一下明天的天气。两件事同时进行',
            additional_kwargs={},
            response_metadata={},
            id='56bd8435-c42a-4f90-9bc5-d24b84459485'
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'dt': '今天'},
                    'id': 'call_00_dByD0Y82kCBQncf2Oiy59319',
                    'type': 'tool_call'
                },
                {
                    'name': 'get_weather',
                    'args': {'dt': '明天'},
                    'id': 'call_01_jbMHr8645iZKbLyHom1Z2337',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='今天天气很好',
            ...
        ),
        ToolMessage(
            content='明天天气很好',
            ...
        ),
        AIMessage(
            content='Model call limits exceeded: run limit (1/1)',
            ...
        )
    ]
}
"""

ToolCallLimitMiddleware

限制工具调用次数，可以 限制所有工具 调用的总次数，也可以 限制特定工具 的调用次数。

参数：

tool_name：限制特定工具。为 None 时，限制所有工具的总调用次数。
thread_limit：线程级限制。整个对话（同一 thread_id）内允许的最大调用次数。
run_limit：运行级限制。单次 invoke（从最后一条 HumanMessage 开始算起）内允许的最大调用次数，每轮对话重置。
exit_behavior：达到限制后的行为，可选值：
- continue：默认值，阻止超限的工具调用，注入一条错误 ToolMessage，但其他工具和模型继续运行，由模型自行决定何时结束。
- error：直接抛出 ToolCallLimitExceededError 异常，立即中断执行。
- end：立即终止执行。仅当限制的是单个具体工具时有效；如果存在多个待处理的并行工具调用，会抛出 NotImplementedError。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from rich import print as rprint

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def get_weather(dt: str):
    """查询天气工具

    Args:
        dt: 查询模式，可选值：今天 or 明天
    """
    return f"{dt}天气很好"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        ToolCallLimitMiddleware(
            run_limit=1,
            exit_behavior="end"
        )
    ],
    tools=[get_weather]
)

res = agent.invoke({
    "messages": ["帮我做两件事：1.查一下今天的天气；2.查一下明天的天气。两件事同时进行"]
})
rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='帮我做两件事：1.查一下今天的天气；2.查一下明天的天气。两件事同时进行',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'dt': '今天'},
                    'id': 'call_00_pq03x88JWsS5TptSjHyq0388',
                    'type': 'tool_call'
                },
                {
                    'name': 'get_weather',
                    'args': {'dt': '明天'},
                    'id': 'call_01_w8LUMoH0vdTsACkcfZFy5652',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='Tool call limit exceeded. Do not make additional tool calls.',
            name='get_weather',
            id='55e2d81c-f84b-4cb4-be31-cc2f408e0bdb',
            tool_call_id='call_01_w8LUMoH0vdTsACkcfZFy5652',
            status='error'
        ),
        AIMessage(
            content='Tool call limit reached: run limit exceeded (2/1 calls).',
            ...
        )
    ]
}
"""

ModelFallbackMiddleware

用于故障转移，当主模型无法访问时，启用备用模型。

参数：参数可以是模型对象，也可以是模型名字符串。

main.py

python

agent = create_agent(
    model="openai:gpt-5.5",  # 主模型
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            "anthropic:claude-sonnet-4-5-20250929",     # 第一个备用
            "google_genai:gemini-3.6-flash",            # 第二个备用
        )
    ]
)

LLMToolSelectorMiddleware

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

参数：

model：用于筛选工具的子模型。
max_tools：可调用工具的总数。
always_include：指定的工具不被计数。若 max_tools = 1，always_include 里有两个工具，那么主模型实际可以选择三个工具。

main.py

python

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        LLMToolSelectorMiddleware(
            model="deepseek-flash",
            max_tools=1,
            always_include=["get_weather"]
        )
    ],
    tools=[get_weather, tool_a, tool_b]
)

ToolRetryMiddleware

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

指数退避：工具调用失败，重复调用的时间间隔成指数级增长。
抖动效果：指数退避执行一组任务时，可以将这一组任务拆成不同时间段执行。

参数：

参数	类型	默认值	说明
max_retries	int	2	最大重试次数。设为 0 表示不重试。
backoff_factor	float	2.0	指数退避的乘数。
initial_delay	float	1.0	第一次重试前的初始等待时间（秒）。
max_delay	float	60.0	等待时间的上限（秒）防止退避时间无限增长。
jitter	bool	True	是否开启抖动。
retry_on	tuple 或 callable	(Exception,)	指定哪些异常发生时重试。
on_failure	str 或 callable	"continue"	重试耗尽后的行为。"continue"：返回带错误信息的 ToolMessage，让模型自行处理；"error"：重新抛出异常，中断执行；也可以传自定义函数，返回错误信息字符串。
tools	list[BaseTool \| str] \| None	None	指定只对哪些工具生效。可以是工具实例或工具名字符串列表。None 表示对所有工具生效。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from datetime import datetime

load_dotenv(verbose=True)

count = 0

@tool(description="查询天气情况")
def get_weather():
    global count
    count += 1
    print(f"第{count}次调用，当前时间：{datetime.now().strftime("%H:%M:%S")}")
    raise TimeoutError("Not Implemented")

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        ToolRetryMiddleware(
            max_retries=5,
            max_delay=10,
            retry_on=(TimeoutError,),
        )
    ],
    tools=[get_weather]
)

res = agent.invoke({
    "messages": ["帮我查一下今天的天气情况"]
})

""" 输出结果：
第1次调用，当前时间：15:01:34
第2次调用，当前时间：15:01:35
第3次调用，当前时间：15:01:37
第4次调用，当前时间：15:01:41
第5次调用，当前时间：15:01:47
第6次调用，当前时间：15:01:59
"""

ModelRetryMiddleware

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

参数：

参数	类型	默认值	说明
max_retries	int	2	最大重试次数。设为 0 表示不重试。
backoff_factor	float	2.0	指数退避的乘数。
initial_delay	float	1.0	第一次重试前的初始等待时间（秒）。
max_delay	float	60.0	等待时间的上限（秒）防止退避时间无限增长。
jitter	bool	True	是否开启抖动。
retry_on	tuple 或 callable	(Exception,)	指定哪些异常发生时重试。
on_failure	str 或 callable	"continue"	重试耗尽后的行为："continue" 返回带错误信息的 AIMessage 让 Agent 继续；"error" 重新抛出异常终止执行；也可传函数自定义错误信息。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

agent = create_agent(
    model=init_chat_model(
        model="deepseek-Demo",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        ModelRetryMiddleware(
            max_retries=5,
            max_delay=10,
        )
    ],
)

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

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

""" 输出结果：
================================ Human Message =================================

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

Model call failed after 6 attempts with BadRequestError: Error code: 400 - {'error': {'message': 'The supported API model names are deepseek-flash, deepseek-v4-pro, but you passed deepseek-Demo. (request_id: 4d362f5f-5f52-46f3-adcf-d6da26bcd849)', 'type': 'invalid_request_error', 'param': None, 'code': 'invalid_request_error'}}
"""

LLMToolEmulator

用于测试和开发阶段的中间件。它的核心作用是：拦截对指定工具的调用，不执行真实代码，而是让另一个 LLM 根据工具描述和参数“编造”一个看起来合理的返回结果。

参数：

tools：要模拟的工具列表。
model：子模型。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

load_dotenv(verbose=True)

@tool
def get_weather(city: str):
    """查询天气工具

    Args:
        city: 城市名
    """
    return f"{city}天气良好"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        LLMToolEmulator(
            model=init_chat_model(
                model="deepseek-flash",
                api_key=os.getenv("DEEPSEEK_API_KEY"),
                base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
            ),
        )
    ],
    tools=[get_weather]
)

res = agent.invoke({
    "messages": ["帮我查一下北京的天气"]
})

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

""" 输出结果：
================================ Human Message =================================

帮我查一下北京的天气
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_00_ET_gapKEeVhQ63fZFmhc1gH0044)
 Call ID: call_00_ET_gapKEeVhQ63fZFmhc1gH0044
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

{"city":"北京","weather":"晴","temperature":26,"unit":"°C","feels_like":27,"humidity":42,"wind":"东南风 3级","air_quality":"良","updated_at":"2025-06-18 14:30"}
================================== Ai Message ==================================

北京当前天气如下：

- **天气**：晴
- **气温**：26°C（体感 27°C）
- **湿度**：42%
- **风力**：东南风 3 级
- **空气质量**：良
- **更新时间**：2025-06-18 14:30

整体来说天气挺舒服的，晴晒、温度适中、空气也不错，适合外出活动。白天阳光较强，可以注意防晒补水～
"""

ContextEditingMiddleware

管理对话上下文长度的中间件，核心手段是在发给模型之前，清空或裁剪旧的工具调用结果，从而控制 token 消耗。

参数：

edits：要应用的编辑策略序列。
token_count_method：用近似估算还是模型精确计数；精确计数仅部分模型支持。

main.py

python

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=50000,  # 输入超过 5 万 token 时触发
                    keep=2,         # 保留最近 2 次工具结果
                )
            ],
        )
    ],
    tools=[get_weather]
)

FilesystemFileSearchMiddleware

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

Glob：基于文件路径的模式匹配。用于快速查找符合特定命名模式的文件。
Grep：基于文件内容的搜索。用于在文件内部查找匹配特定文本或正则表达式的内容。

参数：

root_path：必填，搜索的根目录。
use_ripgrep：是否优先使用 ripgrep 进行内容搜索。
max_file_size_mb：搜索时允许的最大文件大小（单位 MB）超过此大小的文件会被跳过，避免处理超大文件导致性能问题。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

agent = create_agent(
   model=init_chat_model(
      model="deepseek-flash",
      api_key=os.getenv("DEEPSEEK_API_KEY"),
      base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
   ),
   middleware=[
      FilesystemFileSearchMiddleware(
         root_path="../AI/",
         use_ripgrep=True,
         max_file_size_mb=10
      ),
   ],
)

res = agent.invoke({
   "messages": ["找一下Python或Jupyter文件"]
})

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

""" 输出结果：
================================ Human Message =================================

找一下Python或Jupyter文件
================================== Ai Message ==================================

我来帮你查找 Python 和 Jupyter 文件。
Tool Calls:
  glob_search (call_00_FoIxSdhH8EPK7MiHrREV0825)
 Call ID: call_00_FoIxSdhH8EPK7MiHrREV0825
  Args:
    pattern: **/*.py
  glob_search (call_01_UQmS0HB40MW3TV700FSo9705)
 Call ID: call_01_UQmS0HB40MW3TV700FSo9705
  Args:
    pattern: **/*.ipynb
================================= Tool Message =================================
Name: glob_search

/main.py
================================= Tool Message =================================
Name: glob_search

/main.ipynb
================================== Ai Message ==================================

找到了 2 个文件（都在根目录 `/` 下）：

| 文件 | 类型 |
|---|---|
| `/main.py` | Python 脚本 |
| `/main.ipynb` | Jupyter Notebook |

需要我查看其中某个文件的内容，或者搜索特定子目录吗？
"""

多个中间件组合及执行顺序

多个中间件的书写顺序很重要，会按照中间件顺序执行各自的生命周期函数。

python

middleware = [
   TrimmerMiddleware(),         # 1. 先修剪消息
   SummarizationMiddleware(),   # 2. 再摘要
   LoggingMiddleware()          # 3. 最后记录日志
]

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

class Middleware1(AgentMiddleware):
    def before_model(self, state, runtime):
        print("[中间件1] before_model")
        return None

    def after_model(self, state, runtime):
        print("[中间件1] after_model")
        return None

class Middleware2(AgentMiddleware):
    def before_model(self, state, runtime):
        print("[中间件2] before_model")
        return None

    def after_model(self, state, runtime):
        print("[中间件2] after_model")
        return None

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[Middleware1(), Middleware2()],
)

res = agent.invoke({
    "messages": ["测试"]
})

""" 输出结果：
[中间件1] before_model
[中间件2] before_model
[中间件2] after_model
[中间件1] after_model
"""

自定义中间件

某些复杂场景下，官方内置的中间件不能完全满足需求，此时可以通过实现LangChain暴露的中间件 钩子函数 构建自定义中间件

钩子函数

核心特点：

不是你主动在业务代码里随便调用的，而是当流程运行到某个“钩子点”时，系统自动触发它。
它依附于一个更大的执行流程。比如“请求开始前”“模型调用前”“任务结束后”“异常发生时”等。
它的作用是让你在不改主流程源码的前提下插入自己的逻辑。例如做 日志、鉴权、修改输入、拦截输出、清理资源 等

钩子函数分类

Node-style hooks(节点风格钩子)

在流程的 特定节点 运行。适合顺序逻辑，具体函数如下：

before_agent：在Agent开始运行之前执行。
before_model：在模型调用之前执行。
after_model：在模型调用之后执行。
after_agent：在Agent流程全部完成后执行。

Wrap-style hooks(包装风格钩子)

在 模型或工具调用前后 运行。适合控制流，具体函数如下：

wrap_model_call：包裹模型调用。
wrap_tool_call：包裹工具调用。

Node-style hooks

装饰器写法.py

类写法.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import before_agent, before_model, after_model, after_agent
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

@before_agent
def before_agent_middleware(state, runtime):
    print("before_agent")
    return None

@before_model
def before_model_middleware(state, runtime):
    print("before_model")
    return None

@after_model
def after_model_middleware(state, runtime):
    print("after_model")
    return None

@after_agent
def after_agent_middleware(state, runtime):
    print("after_agent")
    return None


agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[before_agent_middleware, before_model_middleware, after_model_middleware, after_agent_middleware],
)

res = agent.invoke({
    "messages": ["测试"]
})

""" 输出结果：
before_agent
before_model
after_model
after_agent
"""

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

class MyMiddleware(AgentMiddleware):
    def before_agent(self, state, runtime):
        print("before_agent")
        return None
    
    def before_model(self, state, runtime):
        print("before_model")
        return None

    def after_model(self, state, runtime):
        print("after_model")
        return None
    
    def after_agent(self, state, runtime):
        print("after_agent")
        return None

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[MyMiddleware()],
)

res = agent.invoke({
    "messages": ["测试"]
})

""" 输出结果：
before_agent
before_model
after_model
after_agent
"""

参数

state：是一个AgentState实例，维护Agent运行过程中的状态，这类状态会随着Agent的运行而发生变化，包括消息列表。
runtime: 是一个Runtime实例，维护Agent运行过程中的上下文环境，包括上下文、长期记忆等。

返回值

返回 None：不修改状态（不修改Agent状态）。
返回字典：更新状态。
返回 {"jump_to": "..."}：控制流程。

None.py

字典.py

jump_to.py

python

def before_model(self, state, runtime):
   print("日志记录")
   return None

python

def after_model(self, state, runtime):
   count = state.get("count", 0)
   return {"count": count + 1}

python

def before_model(self, state, runtime):
   if state.get("count", 0) > 10:
      return {"jump_to": "__end__"}
   return None

jump_to 目标：

"end" - 结束 Agent。
"tools" - 跳到工具节点。
其他自定义节点。

can_jump_to

钩子函数可以 改变Agent正常的运行轨迹 。比如：发现上下文窗口溢出，直接跳转至结尾，提前终止整个Agent。

can_jump_to 决定了钩子函数可以直接跳转至流程的哪些位置，可取值如下：

end：跳转至Agent流程末尾，或第一个after_agent钩子，直接终止整个流程。
tools：跳转至工具节点。
model：跳转至模型节点，或第一个before_model钩子。

装饰器写法.py

类写法.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool

load_dotenv(verbose=True)

@tool(description="查询天气工具")
def get_weather():
    return "天气很好"

@before_model(can_jump_to=["tools"])
def bm_a(state, runtime):
    """强制拦截并触发工具"""
    if "bm_a" in state["messages"][-1].content:
        return {
            "messages": [
                AIMessage(
                    content="这是一个伪造的消息，现在去调用工具",
                    tool_calls=[{
                        "name": "get_weather",
                        "args": {},
                        "id": "bm_a_001"
                    }]
                )
            ],
            "jump_to": "tools"
        }
    return None

@before_model(can_jump_to=["end"])
def bm_b(state, runtime):
    """模拟溢出场景"""
    if "bm_b" in state["messages"][-1].content:
        return {
            "messages": [
                AIMessage("上下文溢出，运行结束")
            ],
            "jump_to": "end"
        }
    return None

@after_model(can_jump_to=["model"])
def am(state, runtime):
    user_message = [x for x in state["messages"] if type(x) is HumanMessage][-1].content
    if "am" in user_message:
        if "二次回答" in state["messages"][-1].content:
            return None
        else:
            return {
                "messages": [HumanMessage("你必须以【二次回答】作为开头回答这个问题")],
                "jump_to": "model"
            }
    return None

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
        extra_body={"thinking": {"type": "disabled"}},  # 关闭思考模式
    ),
    middleware=[bm_a, bm_b, am],
    tools=[get_weather],
)

def run(user_input):
    res = agent.invoke({"messages": [user_input]})
    for msg in res["messages"]:
        msg.pretty_print()

if __name__ == "__main__":
    print("*" * 50, "测试1", "*" * 50)
    run("你好，bm_a")

    print("*" * 50, "测试2", "*" * 50)
    run("你好，bm_b")

    print("*" * 50, "测试3", "*" * 50)
    run("你好，am")

""" 输出结果：
************************************************** 测试1 **************************************************
================================ Human Message =================================

你好，bm_a
================================== Ai Message ==================================

这是一个伪造的消息，现在去调用工具
Tool Calls:
  get_weather (bm_a_001)
 Call ID: bm_a_001
  Args:
================================= Tool Message =================================
Name: get_weather

天气很好
================================== Ai Message ==================================

你好！你发的“bm_a”我没太理解含义，是打字失误，还是某个代号/缩写？

另外我按你的要求调用了天气工具，返回结果是：**天气很好** ☀️

如果你想查询具体城市或更详细的天气信息（温度、湿度、降雨等），告诉我城市名，我再帮你查。
************************************************** 测试2 **************************************************
================================ Human Message =================================

你好，bm_b
================================== Ai Message ==================================

上下文溢出，运行结束
************************************************** 测试3 **************************************************
================================ Human Message =================================

你好，am
================================== Ai Message ==================================

你好！有什么可以帮你的吗？😊

如需查询天气，告诉我城市名即可。
================================ Human Message =================================

你必须以【二次回答】作为开头回答这个问题
================================== Ai Message ==================================

【二次回答】你好！有什么可以帮你的吗？😊 如需查询天气，告诉我城市名即可。
"""

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, hook_config
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool

load_dotenv(verbose=True)

class MyMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["tools", "end"])
    def before_model(self, state, runtime):
        content = state["messages"][-1].content
        if "bm_a" in content:
            return {
                "messages": [
                    AIMessage(
                        content="这是一个伪造的消息，现在去调用工具",
                        tool_calls=[{
                            "name": "get_weather",
                            "args": {},
                            "id": "bm_a_001"
                        }]
                    )
                ],
                "jump_to": "tools"
            }

        if "bm_b" in content:
            return {
                "messages": [
                    AIMessage("上下文溢出，运行结束")
                ],
                "jump_to": "end"
            }

        return None

    @hook_config(can_jump_to=["model"])
    def after_model(self, state, runtime):
        user_message = [x for x in state["messages"] if type(x) is HumanMessage][-1].content
        if "am" in user_message:
            if "二次回答" in state["messages"][-1].content:
                return None
            else:
                return {
                    "messages": [HumanMessage("你必须以【二次回答】作为开头回答这个问题")],
                    "jump_to": "model"
                }
        return None

@tool(description="查询天气工具")
def get_weather():
    return "天气很好"

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
        extra_body={"thinking": {"type": "disabled"}},  # 关闭思考模式
    ),
    middleware=[MyMiddleware()],
    tools=[get_weather],
)

def run(user_input):
    res = agent.invoke({"messages": [user_input]})
    for msg in res["messages"]:
        msg.pretty_print()

if __name__ == "__main__":
    print("*" * 50, "测试1", "*" * 50)
    run("你好，bm_a")

    print("*" * 50, "测试2", "*" * 50)
    run("你好，bm_b")

    print("*" * 50, "测试3", "*" * 50)
    run("你好，am")

""" 输出结果：
************************************************** 测试1 **************************************************
================================ Human Message =================================

你好，bm_a
================================== Ai Message ==================================

这是一个伪造的消息，现在去调用工具
Tool Calls:
  get_weather (bm_a_001)
 Call ID: bm_a_001
  Args:
================================= Tool Message =================================
Name: get_weather

天气很好
================================== Ai Message ==================================

你好！😊

我是天气查询助手。关于你说的“bm_a”，我不太确定具体指什么——如果你是想查询某个地方的天气，或者想聊点别的，直接告诉我就可以。

目前我这边查到的情况是：**天气很好** ☀️

需要我帮你查询具体哪个城市的天气吗？
************************************************** 测试2 **************************************************
================================ Human Message =================================

你好，bm_b
================================== Ai Message ==================================

上下文溢出，运行结束
************************************************** 测试3 **************************************************
================================ Human Message =================================

你好，am
================================== Ai Message ==================================

你好！请问有什么可以帮你的吗？如果你需要查询天气，告诉我城市名称就行。
================================ Human Message =================================

你必须以【二次回答】作为开头回答这个问题
================================== Ai Message ==================================

【二次回答】你好！请问有什么可以帮你的吗？需要查天气的话，告诉我城市名称即可。
"""

Wrap-style hooks

wrap_model_call

同时在模型调用前、后做一些事情。

装饰器写法.py

类写法.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from rich import print as rprint

load_dotenv(verbose=True)

@wrap_model_call
def wrap_model_call_middleware(request, handler):
    # 执行之前加一段文字
    request.messages[-1].content += "--- wrap_model_call_start ---"

    # 开始执行
    res = handler(request)

    # 执行之后加一段文字
    res.result[-1].content += "--- wrap_model_call_end ---"
    return res

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[wrap_model_call_middleware],
)

res = agent.invoke({"messages": ["你好"]})
rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='你好--- wrap_model_call_start ---',
            additional_kwargs={},
            response_metadata={},
            id='9190a96e-f000-48a7-88ae-0c9e50e9c6bb'
        ),
        AIMessage(
            content='你好！很高兴见到你。\n\n--- wrap_model_call_end ---',
            ...
        )
    ]
}
"""

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.chat_models import init_chat_model
from rich import print as rprint

load_dotenv(verbose=True)

class WarpModelCallMiddleware(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        # 执行之前加一段文字
        request.messages[-1].content += "--- wrap_model_call_start ---"

        # 开始执行
        res = handler(request)

        # 执行之后加一段文字
        res.result[-1].content += "--- wrap_model_call_end ---"
        return res

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[WarpModelCallMiddleware()],
)

res = agent.invoke({"messages": ["你好"]})
rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='你好--- wrap_model_call_start ---',
            additional_kwargs={},
            response_metadata={},
            id='9190a96e-f000-48a7-88ae-0c9e50e9c6bb'
        ),
        AIMessage(
            content='你好！很高兴见到你。\n\n--- wrap_model_call_end ---',
            ...
        )
    ]
}
"""

wrap_tool_call

在工具执行前后、后做一些事情。

装饰器写法.py

类写法.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from rich import print as rprint

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def get_weather(city: str, next_day: bool):
    """查询天气情况的工具

    Args:
        city: 城市
        next_day: 是否包含明天的天气
    """
    res = f"{city}天气很好"
    if next_day:
        res += "明天有雨"
    return res

@wrap_tool_call
def wrap_tool_call_middleware(require, handler):
    if require.tool_call["name"] == "get_weather":
        require.tool_call["args"]["next_day"] = not require.tool_call["args"]["next_day"]
        res = handler(require)
        res.content += "--------------"
        return res

    return handler(require)

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[wrap_tool_call_middleware],
    tools=[get_weather],
)

res = agent.invoke({"messages": ["帮我查一下北京今天的天气怎么样"]})
rprint(res)
print("*" * 60)
res = agent.invoke({"messages": ["帮我查一下北京今天的天气怎么样。明天天气也帮我查一下"]})
rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='帮我查一下北京今天的天气怎么样',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'city': '北京', 'next_day': True},
                    'id': 'call_00_IAEA7EfqBdiplrL2PL1o3215',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='北京天气很好明天有雨--------------',
            name='get_weather',
            id='9b545ce8-4d09-417f-a226-8795c910f2a7',
            tool_call_id='call_00_IAEA7EfqBdiplrL2PL1o3215'
        ),
        AIMessage(
            content='北京今天天气很好～☀️\n\n不过我顺便查了明天的天气：**明天北京有雨**，出门记得带伞哦。☔\n\n需要我帮你查其他城市或更详细的天气信息吗？',
            ...
        )
    ]
}
************************************************************
{
    'messages': [
        HumanMessage(
            content='帮我查一下北京今天的天气怎么样。明天天气也帮我查一下',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'city': '北京', 'next_day': False},
                    'id': 'call_00_ET_I2OhJrA1uLvCLaTfNLYZ9163',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='北京天气很好--------------',
            name='get_weather',
            id='b9095c54-3435-45eb-9a21-407e9d614603',
            tool_call_id='call_00_ET_I2OhJrA1uLvCLaTfNLYZ9163'
        ),
        AIMessage(
            content='北京天气很好--------------☀',
            ...
        ),
    ]
}
"""

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call, AgentMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from rich import print as rprint

load_dotenv(verbose=True)

@tool(parse_docstring=True)
def get_weather(city: str, next_day: bool):
    """查询天气情况的工具

    Args:
        city: 城市
        next_day: 是否包含明天的天气
    """
    res = f"{city}天气很好"
    if next_day:
        res += "明天有雨"
    return res

class MyMiddleware(AgentMiddleware):
    def wrap_tool_call(self, require, handler):
        if require.tool_call["name"] == "get_weather":
            require.tool_call["args"]["next_day"] = not require.tool_call["args"]["next_day"]
            res = handler(require)
            res.content += "--------------"
            return res

        return handler(require)

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[MyMiddleware()],
    tools=[get_weather],
)

res = agent.invoke({"messages": ["帮我查一下北京今天的天气怎么样"]})
rprint(res)
print("*" * 60)
res = agent.invoke({"messages": ["帮我查一下北京今天的天气怎么样。明天天气也帮我查一下"]})
rprint(res)

""" 输出结果：
{
    'messages': [
        HumanMessage(
            content='帮我查一下北京今天的天气怎么样',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'city': '北京', 'next_day': True},
                    'id': 'call_00_IAEA7EfqBdiplrL2PL1o3215',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='北京天气很好明天有雨--------------',
            name='get_weather',
            id='9b545ce8-4d09-417f-a226-8795c910f2a7',
            tool_call_id='call_00_IAEA7EfqBdiplrL2PL1o3215'
        ),
        AIMessage(
            content='北京今天天气很好～☀️\n\n不过我顺便查了明天的天气：**明天北京有雨**，出门记得带伞哦。☔\n\n需要我帮你查其他城市或更详细的天气信息吗？',
            ...
        )
    ]
}
************************************************************
{
    'messages': [
        HumanMessage(
            content='帮我查一下北京今天的天气怎么样。明天天气也帮我查一下',
            ...
        ),
        AIMessage(
            ...
            tool_calls=[
                {
                    'name': 'get_weather',
                    'args': {'city': '北京', 'next_day': False},
                    'id': 'call_00_ET_I2OhJrA1uLvCLaTfNLYZ9163',
                    'type': 'tool_call'
                }
            ],
            ...
        ),
        ToolMessage(
            content='北京天气很好--------------',
            name='get_weather',
            id='b9095c54-3435-45eb-9a21-407e9d614603',
            tool_call_id='call_00_ET_I2OhJrA1uLvCLaTfNLYZ9163'
        ),
        AIMessage(
            content='北京天气很好--------------☀',
            ...
        ),
    ]
}
"""

装饰器和类的选择

装饰器写法更适合单个 hook、逻辑简单、快速原型的场景。
类写法更适合多个 hook 组合、复杂配置、需要同时提供同步/异步实现、以及更强复用与可测试性的场景。

中间件执行顺序

中间件定义是乱序的，但传递给Agent的顺序是固定的。
中间件的执行遵循上述规律，只 和传递给Agent的顺序有关。
before_model 中间件的执行顺序和传递顺序一致。
after_model 中间件的执行顺序和传递顺序相反。
wrap_model_call 中间件的执行顺序是：先进后出。
before_model、after_model、wrap_model_call 同时出现时执行顺序是：before_model、wrap_model_call、wrap_model_call、after_model。

main.py

python

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model, wrap_model_call
from langchain.chat_models import init_chat_model

load_dotenv(verbose=True)

@before_model
def before_model_1(state, runtime):
    print("before_model_1")
    return None

@before_model
def before_model_2(state, runtime):
    print("before_model_2")
    return None

@after_model
def after_model_1(state, runtime):
    print("after_model_1")
    return None

@after_model
def after_model_2(state, runtime):
    print("after_model_2")
    return None

@wrap_model_call
def wrap_model_1(request, handler):
    print("wrap_model_1_before")
    res = handler(request)
    print("wrap_model_1_after")
    return res

@wrap_model_call
def wrap_model_2(request, handler):
    print("wrap_model_2_before")
    res = handler(request)
    print("wrap_model_2_after")
    return res

agent = create_agent(
    model=init_chat_model(
        model="deepseek-flash",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_API_BASE_URL"),
    ),
    middleware=[
        before_model_1,
        before_model_2,
        after_model_1,
        after_model_2,
        wrap_model_1,
        wrap_model_2,
    ],
)

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

""" 输出结果：
before_model_1
before_model_2
wrap_model_1_before
wrap_model_2_before
wrap_model_2_after
wrap_model_1_after
after_model_2
after_model_1
"""

中间件概述 ​

什么是中间件 ​

为什么需要中间件 ​

中间件的分类 ​

SummarizationMiddleware ​

HumanInTheLoopMiddleware ​

PIIMiddleware ​

TodoListMiddleware ​

ModelCallLimitMiddleware ​

ToolCallLimitMiddleware ​

ModelFallbackMiddleware ​

LLMToolSelectorMiddleware ​

ToolRetryMiddleware ​

ModelRetryMiddleware ​

LLMToolEmulator ​

ContextEditingMiddleware ​

FilesystemFileSearchMiddleware ​

多个中间件组合及执行顺序 ​

自定义中间件 ​

钩子函数 ​

钩子函数分类 ​

Node-style hooks(节点风格钩子) ​

Wrap-style hooks(包装风格钩子) ​

Node-style hooks ​

参数 ​

返回值 ​

can_jump_to ​

Wrap-style hooks ​

wrap_model_call ​

wrap_tool_call ​

装饰器和类的选择 ​

中间件执行顺序 ​

中间件概述

什么是中间件

为什么需要中间件

中间件的分类

SummarizationMiddleware

HumanInTheLoopMiddleware

PIIMiddleware

TodoListMiddleware

ModelCallLimitMiddleware

ToolCallLimitMiddleware

ModelFallbackMiddleware

LLMToolSelectorMiddleware

ToolRetryMiddleware

ModelRetryMiddleware

LLMToolEmulator

ContextEditingMiddleware

FilesystemFileSearchMiddleware

多个中间件组合及执行顺序

自定义中间件

钩子函数

钩子函数分类

Node-style hooks(节点风格钩子)

Wrap-style hooks(包装风格钩子)

Node-style hooks

参数

返回值

can_jump_to

Wrap-style hooks

wrap_model_call

wrap_tool_call

装饰器和类的选择

中间件执行顺序