在上一篇中我们讲述了agent，其中我们说agent动态绑定工具和动态设置提示词都需要中间件，现在我们就看一下什么是中间件。这里先简单用一句话概述就是中间件是Agent的“中枢”，在执行流程的关键节点进行拦截、控制和增强。

概述

Middleware(中间件)，简单说就是Agent 执行过程中的钩子函数，是 LangChain 1.x 的“王牌”工程化能力。借助中间件，开发者可以高度定制和控制Agent运行的每一个环节，是处理 Agent 生命周期的标准方式。

在 LangChain 的 Agent 执行循环中，比如 “模型调用前”、“模型调用后”、“工具调用前后” 设置一些钩子（hooks），让你在不改 Agent 主体逻辑的情况下实现策略与治理。

为什么需要中间件

如果没有中间件，Agent 的执行流程通常比较直接：

plaintext
用户输入 → 拼接提示词/消息 → 调用模型 → 如有需要调用工具 → 返回结果

这种方式对于简单场景已经足够，但一旦进入真实项目，往往会遇到很多额外需求，例如：

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

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

• 主流程会迅速变乱

  Agent 本身只需要关心“理解用户需求、决定是否调用工具、生成结果”，但一旦把日志、鉴权、重试、风控、审计都塞进去，主逻辑就会变得臃肿。

• 很多逻辑是横切需求，难以复用

  例如日志、重试、风控、权限控制，通常不是某一个 Agent 独有的，而是多个 Agent 都需要。如果直接写死在每个 Agent 里，会产生大量重复代码。

• 流程控制粒度不够细

  有些逻辑必须发生在“模型调用前”，有些要发生在“工具调用后”，如果没有统一的执行拦截点，开发者只能手动改主流程，既麻烦又容易出错。

• 后期维护成本高

  当你需要增加一个新规则，例如“所有外部工具调用前都先做审计”，如果系统没有中间件机制，往往需要修改很多处代码。

总结： 中间件的价值就在于把这些与业务无关、但与执行过程强相关的横切逻辑，从 Agent 主流程中分离出 来。让Agent 主体代码 聚焦业务 ，而借助中间件，实现“ 拦截流程、修改流程、增强流程 ”。 简言之，LangChain 1.x 的中间件能实现如下功能：

• ✅ 日志与分析 - 追踪行为、调试、性能监控
• ✅ 转换 - 修改提示词、工具选择、输出格式
• ✅ 容错 - 重试、降级、早期终止
• ✅ 安* - 限流、守护规则、PII 检测

Langchinn内置中间件概述

LangChain提供的和模型供应商无关的内置中间件分为六个类别

1. 类型1：成本与资源控制类

核心目标：控成本、控配额、避免无限调用 这类中间件主要解决“ Agent太贵、太能跑、停不下来 ”的问题。 包含：

• Model call limit：限制模型调用次数，防止一次任务反复请求 LLM，导致费用失控
• Tool call limit：限制工具调用次数，避免 Agent 无限试错、死循环调工具
• Summarization：在上下文快满时自动总结历史，减少 token 消耗
• Context editing：裁剪上下文、清理工具调用痕迹，本质上也是为了节省上下文成本

业务场景理解： 适合生产环境的成本治理、配额治理、长会话优化、SaaS 产品控费。

2. 类型2：稳定性与容错保障类

核心目标：保证服务不中断、失败后尽量自动恢复 这类中间件主要解决“ 调用失败怎么办、模型挂了怎么办、工具超时怎么办 ”。 包含：

• Model fallback：主模型失败时切换备用模型
• Model retry：模型调用失败后自动重试
• Tool retry：工具调用失败后自动重试

业务场景理解： 适合线上生产系统，尤其是多模型、多工具依赖的 Agent。 本质上是在做 高可用、容灾、鲁棒性建设。

3. 类型3：安全与合规风控类

核心目标：让 Agent 可控、可审、合规 这类中间件主要解决“ Agent乱执行、泄露敏感信息、做危险操作 ”的问题。 包含：

• Human-in-the-loop：在关键工具调用前暂停，等人工审批
• PII detection：检测和处理个人敏感信息
• Model call limit / Tool call limit：某种意义上也可归到风控，因为它能防止异常滥用

业务场景理解： 适合企业内部系统、客服系统、审批流、数据查询类 Agent。 尤其是涉及：发邮件、调数据库、调财务/人事系统、导出敏感信息、执行外部动作等

4. 类型4：决策增强与智能编排类

核心目标：提升 Agent 的决策质量和任务拆解能力 这类中间件主要解决“ Agent不够聪明、不会规划、不会先筛工具 ”的问题。 包含：

• To-do list：给 Agent 增加任务规划、分步骤执行和状态跟踪能力
• LLM tool selector：当工具太多时，用子模型筛选最相关的几个工具交给主模型
• Subagent：允许生成子Agent，把复杂任务拆给不同角色处理

业务场景理解： 适合复杂任务流，比如：研究型 Agent、多步骤分析、报告生成、多角色协作、长链路任务编排等。 这类本质上是在增强 Agent的“脑子”与“组织能力”。

5. 类型5：执行能力扩展类

核心目标：给 Agent 更多“手脚” 这类中间件主要解决“ Agent只能聊天，不能真正操作环境 ”的问题。 包含：

• Shell tool：给 Agent 持久 shell，会执行命令
• File search：给 Agent 文件搜索能力，能做 Glob/Grep
• Filesystem：给 Agent 文件系统读写与长期存储能力

业务场景理解： 适合工程 Agent、代码 Agent、本地自动化 Agent、运维 Agent。 本质上是把 Agent 从“纯推理”扩展成“能操作环境的执行体”。

6. 类型6：开发调试与测试辅助类

核心目标：方便开发、测试、验证 Agent 行为 这类中间件主要不是直接服务业务，而是服务于 研发和调试阶段 。 包含：

• LLM tool emulator：用 LLM 模拟工具执行，便于测试（最典型）
• Summarization：有时也可辅助调试长会话表现
• Context editing：可用于测试上下文裁剪效果
• Human-in-the-loop：也常用于调试高风险步骤

业务场景理解： 适合开发阶段快速验证流程、做 mock、减少真实工具依赖。

常用的内置中间件使用

SummarizationMiddleware中间件

作用：对历史消息列表进行 摘要&总结 ，达到 压缩上下文 的效果。

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

部分参数说明：

1. 参数1：model —用于摘要的模型

   可以是模型名称也可以是模型对象，如果传递的是模型名称，底层会调用 init_chat_model 初始化模型。
   

2. 参数2：trigger —摘要触发条件

    是一个列表，每个元素对应一个条件，当 任一条件满足 时，触发摘要。
    1. tokens ：token的数量，历史token的累计数量达到该值触发摘要。
    2. messages ：历史消息数量，历史消息条数达到该值触发摘要。
    3. fraction ：上下文长度比例。历史token的累计数量达到模型的      `max_input_tokens*fraction` 触发摘要。
   
    如果条件包含 fraction ，要求模型的profile包含 max_input_tokens ，Deepseek模型的profile为空，此时需要手动添加该配置项。Deepseek-V3.2的上下文长度为128K。
    
   

3. 参数3：keep —摘要时保留的原始消息

支持三种条件，但和trigger不同，keep同一时间只接收一种条件。

1. tokens ：摘要时保留的token数量。
2. messages ：摘要时保留的历史消息条数。
3. fraction ：摘要时保留 max_input_tokens*fraction 个token。

4. 参数4：token_counter —统计token数量的函数

  默认使用LangChain提供的 count_tokens_approximately ，一般不用更改。
对于纯文本消息，该函数的大致思路是先统计消息的字符数，也就是 len(字符串) ，然后再除以每
个token大致的字符数，转换为粗略的token数，再加一些额外开销。作为估算的token数。

5. 参数5：summary_prompt —摘要时的自定义提示词

该提示词需要包含 {messages} 占位符，使得历史消息列表可以被插入。不指定则使用内置提示词。

6. 参数6：trim_token_to_summarize —摘要时历史消息的最大token数

如果历史消息token数大于该值，则会被裁剪。默认为" 4000 "。
如果trigger用token作为度量，调大触发阈值时，当前配置项应相应调整，否则会丢失信息。

代码示例:

python
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain.agents.middleware import SummarizationMiddleware
from langchain.agents import create_agent

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

# 核心多语言自适应Prompt
AUTO_LANG_SUMMARY_PROMPT = """
下面是一段人机对话历史，请生成简洁对话摘要。
重要规则：
1. 识别对话主体使用的语言；
2. **摘要必须使用和对话完全相同的语言输出**；
3. 不要切换语言，不要中英混杂；
4. 保留关键问题、核心诉求、重要结论。
5. **输出的摘要开头必须固定带上标签：中文会话输出「对话摘要：」，英文会话输出「Conversation Summary:」**

对话内容：
{messages}
"""

agent = create_agent(
    model=model,
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=[
                ("tokens", 100),
                ("messages", 6),
                ("fraction", 0.001)
            ],
            keep=("messages", 2),
            summary_prompt=AUTO_LANG_SUMMARY_PROMPT,
            # summary_prefix="对话摘要："
        )
    ]
)
response = agent.invoke({
    "messages": messages
})

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

HumanInTheLoopMiddleware中间件

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

部分参数说明：

1. 参数1：interrupt_on —工具名和中断策略的映射

策略可以是True、False或InterruptOnConfig对象，精细控制决策选项。

python
interrupt_on={
    "get_weather": True,
    "read_email_tool": False,
    "send_email_tool": {
        "allowed_decisions": ["approve", "reject"],
    },
}

True表示所有决策(approve, edit, reject) 都可以选择，相当于"get_weather": { "allowed_decisions": ["approve", "edit", "reject"], }

False表示不中断，即无需审批即可执行。

InterruptOnConfig 是一个TypedDict的子类，可以用字典直接赋值。支持的Key有：

• ① allowed_decisions 精细控制中断后允许的决策。
• ② description ：特定工具的中断描述信息，优先级高于description_prefix，后 者会更 改 所有 工具中断的描述。

2. 参数2：description_prefix —自定义中断描述

默认为 "Tool execution requires approval" ，下面的举例可以看到效果。

举例过程1:调用前中断

本例需要从中断的位置让Agent继续运行，这就需要用到短期记忆，这里先使用即可。 创建Agent时通过checkpointer参数启用了短期记忆，在调用时通过传递相同的config加载记忆。记住固定用法即可。

python
agent = create_agent(
    model=model,
    tools=[get_weather, get_news, read_email_tool, send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
                "get_news": True,
                "read_email_tool": False,
                "send_email_tool": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "发送邮件中断啦"
                },
            },
            description_prefix="中断啦"
        ),
    ],
)
config = {"configurable": {"thread_id": "1"}}

# 第一次调用：会暂停在发送邮件前
response = agent.invoke(
    {
        "messages": [
            HumanMessage(content="请帮我查询今天北京的天气"
                                 "查询今日新闻"
                                 "查看ID为 'sk2131421' 的邮件内容，"
                                 "向15641685664@qq.com发送邮件，标题是'哈哈哈'，内容是：'你好啊'"
                                 "同时做这四件事")
        ]
    },
    config=config,
)

print("==== 第一次 invoke 返回 ====")
print("========= 原始响应 =========")
rprint(response)
print("========= 美化输出 =========")
for msg in response["messages"]:
    msg.pretty_print()
# 关键：看中断信息
interrupts = response.get("__interrupt__", [])
print("========== interrupts ==========")
rprint(interrupts)
# print("==== 逐个打印 interrupt 请求 ====")
action_requests = interrupts[0].value["action_requests"]
for action_request in action_requests:
    rprint(action_request)

举例过程2：指明工具调用请求决策

python
from langgraph.types import Command

# 如果有中断，说明进入人在环了
weather_decision = {
    "type": "edit",
    "edited_action": {
        "name": "get_weather",
        "args": {"city": "中国北京市", "is_forcast": True}
    }
}
news_decision = {
    "type": "approve",
}
send_email_decision = {
    "type": "approve"
}
decisions = {
    "decisions": []
}
# 决策的顺序必须和返回的中断请求顺序一致
for action_request in action_requests:
    if action_request["name"] == "get_weather":
        decisions["decisions"].append(weather_decision)
    if action_request["name"] == "get_news":
        decisions["decisions"].append(news_decision)
    if action_request["name"] == "send_email_tool":
        decisions["decisions"].append(send_email_decision)

if interrupts:
    # 审批通过
    resumed_response = agent.invoke(
        Command(resume=decisions),
        config=config,  # 必须是同一个 thread_id
    )
    print("==== 审批后继续执行 ====")
    for msg in resumed_response["messages"]:
        msg.pretty_print()

上面两个案例是配合一起来看。把一个案例拆成两部分了。

PIIMiddleware中间件

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

部分参数说明：

1. 参数1：pii_type —检测的PII数据类型

   可以是内置类型或自定义类型，自定义类型有

   • email ：电子邮箱地址
   • credit_card ：信用卡号
   • url ：网址
   • mac_address ：设备MAC地址
   • ip ：IP地址

2. 参数2：strategy —处理PII信息的策略

支持四种选项：

• redact ：将检测到的PII信息用字符串 [REDACTED_[PII_TYPE]] 替换，其中的 PII_TYPE是上面提到的具体类型，比如 [REDACTED_EMAIL] 、 [REDACTED_CREDIT_CARD] 这样的标签。完全 “擦除/隐藏” 真实内容。适合日志清洗、合规需求、公开输出时隐藏敏感内容。

• mask ：用 *** 将PII信息的前面一部分信息遮蔽。比如信用卡号可能变成 -- ****-1234 （只保留最后几位/部分可见），邮箱可能保留域名部分 + 隐藏用户名的一部分等 — 既 隐藏大部分敏感信息，又保留了一点“可辨识性”（比如账号后四位、域名等），适合用户服务界面 / 前端显示 / 需要部分可识别但不泄露完整敏感内容的场景。

• hash ：用检测到的PII信息的 哈希值 替代原值。比如 <email_hash:a1b2c3d4> 。适合 analytics、调试 (debug)、统计分析、匿名追踪等场景。

• block ：如果检测到PII信息， 直接抛出异常 。适合对隐私要求极高、绝不允许泄露任何敏感信息 的场景。

3. 参数3：detector —自定义 PII检测函数 或者 正则表达式

如果没有提供则使用内置的检测函数。LangChain为每种PII信息定制了专门的检测函数.

pythn
BUILTIN_DETECTORS: dict[str, Detector] = {
    "email": detect_email,
    "credit_card": detect_credit_card,
    "ip": detect_ip,
    "mac_address": detect_mac_address,
    "url": detect_url,
}

4. 参数4：apply_to_input —是否在调用模型前检测

默认为True。

5. 参数5：apply_to_output —是否在模型调用后检测

默认为False。

6. 参数6：apply_to_tool_results —是否在工具调用后检测其输出

默认为False。 通常我们 只在模型调用前 检测。因为PII检测的主要目的是避免将敏感信息发送给模型服务导致信息泄露。

举例1:内置检测器

python
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        PIIMiddleware("url", strategy="hash", apply_to_input=True),
        PIIMiddleware("mac_address", strategy="mask", apply_to_input=True),
        PIIMiddleware("ip", strategy="block", apply_to_input=True),
    ]
)
response = agent.invoke({
    "messages": [HumanMessage("""
    帮我向 156168188@qq.com 发送一封邮件
    同时查看银行卡号： 5105-1051-0510-5100 的余额
    访问 https://localhost:12345
    确认这是不是 MAC地址： 11-11-11-11-11-11
""")]
})
for msg in response["messages"]:
    msg.pretty_print()

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

举例2：自定义检测器/函数

python
import re


# 自定义检测函数
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=model,
    tools=[],
    middleware=[
        PIIMiddleware("api_key", strategy="hash", apply_to_input=True,
                      detector=r"sk-[a-zA-Z0-9]+"),
        PIIMiddleware("phone_number", strategy="mask", apply_to_input=True,
                      detector=detect_phone_number)
    ]
)
response = agent.invoke({
    "messages": [HumanMessage("""
    这是不是有效的 API_KEY： sk-awef23AFEfaafaefa
    帮我给这个号码打电话： 12345612345
    访问 https://localhost:12345
    """)]
})

TodoListMiddleware中间件

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

比如，当一个大任务需要被拆解为 3 个以上的子任务，且前面的步骤是后面步骤的前提时，如果不列Todo 列表，大模型在执行到第 3 步时，很容易忘记自己最初的目标，或者在工具返回大量报错信息后“应激”，直接跳过验证去回答用户。

此时， TodoListMiddleware 中间件强制它把计划挂在全局状态里，时刻提醒它“下一步该干什么”。

参数说明

1. system_prompt —自定义指导todo列表使用的提示词

   不提供则使用内置提示词，通常不必提供。

2. tool_description —自定义write_tools工具的描述信息

   不提供则使用内置描述，通常不必提供。

案例演示：

我们设计一个较为复杂的任务：

1. 任务目标

   扫描工作目录，测试并修复工作区下的my_add.py文件。

2. 工具列表

   为了实现上述任务，提供一系列工具。

   list_files：扫描工作目录，列出其中的所有文件

   read_file：扫描指定文件，返回文件内容

   write_file：向指定文件写入内容

   run_tests：运行测试，底层基于pytest实现

工具定义：

python
from langchain.tools import tool
from pathlib import Path

WORKSPACE = Path("../todo_workspace")


@tool
def list_files(path: str = ".") -> str:
   """
   列出工作区指定目录下的文件和子目录。path 只能是相对路径。

   Args:
       path: 工作区下的相对路径，一定指向目录，默认为.，表示工作区根路径，不能访问工作区外的目录
   """

   target = (WORKSPACE / path).resolve()
   workspace_root = WORKSPACE.resolve()

   if not str(target).startswith(str(workspace_root)):
       return "错误：只允许访问工作区内的目录。"
   if not target.exists():
       return f"错误：目录不存在: {path}"
   if not target.is_dir():
       return f"错误：不是目录: {path}"

   items = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))

   if not items:
       return f"目录为空: {path}"

   lines = []
   for item in items:
       rel = item.relative_to(workspace_root)
       kind = "[DIR]" if item.is_dir() else "[FILE]"
       lines.append(f"{kind} {rel.as_posix()}")

   return "\n".join(lines)


@tool
def read_file(path: str) -> str:
   """
   读取工作区中的文本文件内容。path 只能是相对路径。

   Args:
       path: 工作区内的文件名
   """

   file_path = (WORKSPACE / path).resolve()
   if not str(file_path).startswith(str(WORKSPACE.resolve())):
       return "错误：只允许读取工作区内的文件。"
   if not file_path.exists():
       return f"错误：文件不存在: {path}"
   return file_path.read_text(encoding="utf-8")


@tool
def write_file(path: str, content: str) -> str:
   """
   写入工作区中的文本文件。path 只能是相对路径。

   Args:
       path: 工作区内的文件名
       content: 写入文件的内容
   """

   file_path = (WORKSPACE / path).resolve()
   if not str(file_path).startswith(str(WORKSPACE.resolve())):
       return "错误：只允许写入工作区内的文件。"
   file_path.write_text(content, encoding="utf-8")
   return f"已写入文件: {path}"


@tool
def run_tests() -> str:
   """
   在工作区运行 pytest -q，并返回输出。
   不接收任何参数，返回格式为
   returncode=0|1
   STDOUT:
   STDERR:
   """
   try:
       result = subprocess.run(
           ["pytest", "-q"],
           cwd=str(WORKSPACE),
           capture_output=True,
           text=True,
           timeout=20,
       )
       return (
           f"returncode={result.returncode}\n\n"
           f"STDOUT:\n{result.stdout}\n\n"
           f"STDERR:\n{result.stderr}"
       )
   except Exception as e:
       return f"运行测试失败: {e}"

主程序

python
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.messages import HumanMessage
from rich import print as rprint

# 1. 初始化 Agent
agent = create_agent(
   model=model,
   # write_todos 等工具，TodoListMiddleware 需要配合这些工具使用
   tools=[list_files, read_file, write_file, run_tests],
   # 引入 Todo 列表中间件
   middleware=[TodoListMiddleware()],
   system_prompt=(
       "你是一个代码修复助手。遇到多步骤任务时，先使用 write_todos 制定待办事项；"
       "然后读取文件、修复代码并运行测试。工作全部在工作区下进行。"
   ),
)
# 2. 使用invoke进行同步调用
print("正在执行 Agent 任务...")
final_state = agent.invoke(
   {
       "messages": [
           HumanMessage(content="请测试并修复工作区下 my_add.py 文件中的代码")
       ]
   }
)
rprint(final_state)

自定义中间件

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

Hook 函数，中文常叫 钩子函数 ，指的是：在某个既定流程的特定时机，被框架、系统或主程序 自动 调用 的扩展函数。

核心特点：

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

LangChain的hook函数分类

官方将六个钩子函数按照风格分为两类

类型1：Node-style hooks(节点风格钩子)

顾名思义，它们在流程的 特定节点 运行。 适合顺序逻辑，如记录日志、验证 包括

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

类型2：Wrap-style hooks(包装风格钩子)

顾名思义，它们在 模型或工具调用前后 运行。 适合控制流，如重试、回退、缓存。 包括

• wrap_model_call (包裹模型调用)
• wrap_tool_call (包裹工具调用)

Node-style hooks函数用法

支持两种用法

• 装饰器是函数式挂载，把一个hook快速挂载到Agent的某个节点。
• 类写法是对象化中间件，把中间件封装为一个可配置、可复用、可扩展的组件。

基于装饰器实现

python
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, before_agent, after_agent, AgentState, after_model
from langgraph.runtime import Runtime
from typing import Any


# 1. 定义 before_model 钩子
@before_model
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_model <- "
    return None


# 2. 定义 after_model 钩子
@after_model
def after_model_middleware(state: AgentState, runtime: Runtime) -> dict[str,
Any] | None:
    state["messages"][-1].content += " -> after_model <- "
    return None


# 3. 定义 before_agent 钩子
@before_agent
def before_agent_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_agent <- "
    return None


# 4. 定义 after_agent 钩子
@after_agent
def after_agent_middleware(state: AgentState, runtime: Runtime) -> None:
    state["messages"][-1].content += " -> after_agent <- "
    return None


agent = create_agent(
    model=model,
    middleware=[before_model_middleware, after_model_middleware,
                before_agent_middleware, after_agent_middleware]  # 👈 添加中间件
)
response = agent.invoke({
    "messages": [HumanMessage("你好啊")],
})

基于类的实现

关键规则：

1. 必须继承 AgentMiddleware ← 这个固定
2. 方法名固定 ( before_model , after_model ) ← 这个固定
3. 类名随意 ← 这个不固定

LangGraph 只看：

• 是否继承 AgentMiddleware？
• 是否有 before_model / after_model 等方法？

python
from langchain.agents.middleware import AgentMiddleware, AgentState
from langgraph.runtime import Runtime
from typing import Any


class MyMiddleware(AgentMiddleware):

    def __init__(self):
        super().__init__()

    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> before_model <- "
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> after_model <- "
        return None

    def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> before_agent <- "
        return None

    def after_agent(self, state: AgentState, runtime: Runtime) -> None:
        state["messages"][-1].content += " -> after_agent <- "
        return None
    

my_middleware = MyMiddleware()
agent = create_agent(
    model=model,
    middleware=[my_middleware]
)
response = agent.invoke({
    "messages": [HumanMessage("你好啊")],
})
for msg in response["messages"]:
    msg.pretty_print()

总结

1. before_model 通常的场景：

   • 消息修剪（trim messages）
   • PII 脱敏
   • 输入验证
   • 条件路由

2. after_model 通常的场景：

   • 输出验证
   • 格式化响应
   • 统计信息
   • 状态更新

3. 参数说明

Node-style hooks函数有两个参数

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

2. 返回值说明

返回 None：不修改状态（不修改Agent状态）

```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} # 更新状态中的 count

返回 {"jump_to": "..."}：控制流程

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

这里就涉及到Node-style的四个hook函数可以接收额外参数 can_jump_to 。 钩子函数可以 改变Agent正常的运行轨迹 。比如：发现上下文窗口溢出，直接跳转至结尾，提前终止整个Agent。 can_jump_to 决定了钩子函数可以直接跳转至流程的哪些位置，可取值如下：

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

基于装饰器实现

python
from langchain_core.messages import AIMessage


# 在模型（LLM）执行前触发。允许跳转到 "tools" 节点。
@before_model(can_jump_to=["tools"])
def force_tool_first(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """
    【业务场景：强行拦截并触发工具】
    如果用户输入包含 "direct tool"，则跳过本次大模型的思考/生成阶段，
    直接伪造一个大模型的 tool_calls 意图，强行把控制权移交给工具执行节点。
    """

    text = state["messages"][-1].content
    # 检查关键词，满足条件则强行干预流程
    if isinstance(text, str) and "direct tool" in text.lower():
        print("[MIDDLEWARE] before_model: jump_to='tools'")
    # 人工构造一个大模型的消息对象（AIMessage）
    # 欺骗系统，让系统误以为这是模型自己决定要调用的工具
    fake_tool_call = AIMessage(
        content="人工构造的消息",
        tool_calls=[
            {
                "name": "get_news",
                "args": {},
                "id": "call_force_weather_001",
            }
        ],
    )
    # 返回更新后的状态：注入伪造的消息，并明确指定下一步跳转到 "tools" 节点
    return {
        "messages": [fake_tool_call],
        "jump_to": "tools",
    }
    # 如果不满足触发条件，返回 None，流程正常向下流转（继续让 LLM 思考）
    return None

基于类实现

python
from langchain.agents.middleware import hook_config,can_jump_to


class MyMiddleware(AgentMiddleware):

    @hook_config(can_jump_to=["tools", "end"])
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:

        text = state["messages"][-1].content
        # 假装溢出
        if "overflow" in text:
            print("[MIDDLEWARE] before_model: jump_to='end' when contenxt window overflow")
            return {
                "messages": [
                    AIMessage("上下文窗口溢出，终止")
                ],
                "jump_to": "end",
            }
        
        if isinstance(text, str) and "direct tool" in text.lower():
            print("[MIDDLEWARE] before_model: jump_to='tools'")
            fake_tool_call = AIMessage(
                content="人工构造的消息",
                tool_calls=[
                    {
                        "name": "get_news",
                        "args": {},
                        "id": "call_force_weather_001",
                    }
                ],
            )
            return {
                "messages": [fake_tool_call],
                "jump_to": "tools",
            }
        return None

Wrap-style hooks函数用法

wrap_model_call

① 基于装饰器实现 我们可以同时在模型调用前后做事，所以命名为 wrap_model_call ，wrap意为 包裹 。

python
from typing import Callable

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse


@wrap_model_call
def wrap_model_call_middleware(
        request: ModelRequest,  # 包含即将发送给大模型的所有请求数据（如消息列表、温度等）
        handler: Callable[[ModelRequest], ModelResponse],  # 核心句柄：代表下一个中间
        件或最终的大模型调用服务
) -> ModelResponse | None:
    # 动态篡改用户发出的最后一条消息的内容，悄悄往里面追加字符串。
    # 典型应用：统一在底层为所有请求追加特殊的 Prompt 提示词（例如：“请用中文回答”、“禁止透漏公司机密”等）。
    request.messages[-1].content += " -> wrap_model_call_before <- "
    # 将修改后的请求传递给 handler，真正去调用大模型（或者流转到下一个拦截器）
    # 这一步会产生真实的 Token 消耗并等待大模型响应
    response = handler(request)
    # 大模型返回响应后，在将响应交付给 Agent 状态机之前，对其内容进行直接篡改
    # `response.result` 是一个消息列表，修改其第一条返回消息的内容
    # 典型应用：做底层的文本敏感词过滤、输出格式强行格式化、或是统一添加某些后处理标记。
    response.result[0].content += " -> wrap_model_call_after <- "
    # 将修改完的响应体返回，继续维持 Agent 生命周期流转
    return response

② 基于类实现

python
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest,ModelResponse
from typing import Callable


class WrapModelCallMiddleWare(AgentMiddleware):

    def wrap_model_call(
            self,
            request: ModelRequest,
            handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse | None:


        request.messages[-1].content += " -> wrap_model_call_before <- "
        response = handler(request)
        response.result[0].content += " -> wrap_model_call_after <- "
        return response
    
agent = create_agent(
    model=model,
    middleware=[WrapModelCallMiddleWare()]
)

使用场景：用于拦截、重试、缓存模型调用。

响应缓存案例

python
from langchain.agents.middleware import wrap_model_call, ModelRequest,ModelResponse
from typing import Callable
import hashlib
import json


class ModelCache:
    """模型响应缓存"""

    def __init__(self):
        self.cache = {}


    def create_hook(self):
        @wrap_model_call
        def cache_model(
                request: ModelRequest,
                handler: Callable[[ModelRequest], ModelResponse]
        ) -> ModelResponse:


            # 生成缓存键
            cache_key = hashlib.md5(
                json.dumps({
                    "messages": [str(m) for m in request.messages],
                    "system": str(request.system_message)
                }).encode()
            ).hexdigest()
            
            # 检查缓存
            if cache_key in self.cache:
                print("💾 缓存命中！")
                return self.cache[cache_key]
            
            # 调用模型
            print("🔍 缓存未命中，调用模型")
            response = handler(request)
            
            # 存入缓存
            self.cache[cache_key] = response
            return response
        return cache_model
# 使用
cache = ModelCache()
agent = create_agent(
    model=model,
    middleware=[cache.create_hook()]
)

修改系统提示词案例

python
@wrap_model_call
def add_context(
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """动态添加上下文信息到系统提示"""
    # 获取当前时间
    from datetime import datetime
    
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    # 构建新的系统消息
    original_content = request.system_message.content if request.system_message else ""
    new_content = f"""{original_content}
    当前时间：{current_time}
    用户位置：中国
    语言偏好：中文
    """
    # 创建新的系统消息
    new_system_message = SystemMessage(content=new_content)
    # 使用 override 方法修改请求
    modified_request = request.override(system_message=new_system_message)
    return handler(modified_request)

wrap_tool_call

我们可以同时在工具调用前后做事，所以命名为 wrap_tool_call 。

python
from langchain_core.messages import ToolMessage
from langgraph.prebuilt.tool_node import ToolCallRequest
from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command

@wrap_tool_call
def wrap_tool_call_middleware(
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
    result = handler(request)
    print(f"原始参数：{request.tool_call['args']}")
    print(f"原始参数调用结果： {result}")
    request.tool_call["args"]["is_forcast"] = True
    result = handler(request)
    print(f"更新后的参数：{request.tool_call['args']}")
    print(f"更新参数调用结果： {result}")
    return result