上一篇文章我们介绍了 LangChain Middleware 以及它在 Agent 执行链中的位置，同时也介绍了 Node-style Hook 和 Wrap-style Hook 两种扩展方式。

在 LangChain 中内置了一些 Middleware，它们能够解决 Agent 开发中常见的一些问题，比如，如何在越来越长的对话中控制上下文和 Token 消耗、工具访问外部接口偶尔超时时如何自动重试、执行关键动作时的手工审批等。此外，当内置的 Middleware 不能满足需求外，我们还可以自定义 Middleware。

本文主要介绍常用的内置 Middleware，以及针对实际问题，如何自定义 Middleware。

一、内置 Middleware 介绍

LangChain 提供的 Middleware 很多。如下表：

问题常用 Middleware作用
Tool 调用偶发失败ToolRetryMiddleware对指定异常进行有限重试
模型调用失败ModelRetryMiddleware重试模型请求
主模型不可用ModelFallbackMiddleware切换备用模型
对话过长SummarizationMiddleware压缩较早的历史消息
Tool 结果太多ContextEditingMiddleware清理旧 Tool 输出
Agent 循环过多ModelCallLimitMiddleware、ToolCallLimitMiddleware限制调用次数
高风险操作HumanInTheLoopMiddlewareTool 执行前等待人工确认

实际开发时，我们根据具体问题，决定添加哪个 Middleware。并不需要为了“配置完整”，一开始就把所有 Middleware 都挂上。事实上，Middleware 越多，执行链越复杂。出现问题时，也越难判断到底是哪一层修改了 State、拦截了 Tool 或处理了异常。

二、先判断 Hook 的位置

遇到一个 Middleware，理解它最直接的方法不是看参数，而是先看：

它在哪个 Hook 上工作。

Node-style Hook 更像是在 Agent 执行到某个阶段时插入一段逻辑：

代码片段Text
before_agent
before_model
after_model
after_agent

Wrap-style Hook 则会直接包住一次模型调用或者 Tool 调用：

代码片段Text
wrap_model_call
wrap_tool_call

例如 SummarizationMiddleware。

它会在调用模型之前检查当前上下文。如果消息或 Token 已经超过阈值，就先整理较早的历史消息，再让模型继续执行。

过程大致是：

代码片段Text
检查 State
    ↓
压缩历史消息
    ↓
更新 State
    ↓
调用模型

它主要是在固定阶段修改 State，所以适合 Node-style Hook。

重试则不一样。

Tool 调用失败以后，Middleware 需要捕获异常，再决定是否重新执行：

代码片段Text
调用 Tool
   ↓
发生异常
   ↓
是否重试
   ↓
再次调用 Tool

这里必须直接控制原始调用，因此更适合 Wrap-style Hook。

可以先记住一个简单判断：

修改某个执行阶段的数据，优先考虑 Node-style。

控制一次模型或 Tool 的真实调用，优先考虑 Wrap-style。

后面很多 Middleware 都可以用这个思路理解。

三、场景一：外部接口要有限重试

Agent 只要开始调用外部服务，重试几乎是绕不开的问题。

例如订单 Agent 可能调用物流、库存、支付或者内部微服务。

这类接口偶尔会出现：

代码片段Text
ConnectionError
TimeoutError

重新请求一次，很可能就恢复了。

但参数错误、鉴权失败、权限不足这类问题，重试通常没有意义。

ToolRetryMiddleware 是一个用于控制工具调用重试行为的中间件。它可以把重试范围限制得比较明确，避免在不需要重试的场景下盲目重试，从而提升系统的容错性和可控性。

下面模拟一个物流接口。第一次连接失败，第二次恢复。

代码片段Python
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.tools import tool
 
 
attempts = {"count": 0}
 
 
@tool
def query_shipping(order_id: str) -> str:
    """查询订单的最新物流状态。"""
    attempts["count"] += 1
    print(f"物流接口调用次数：{attempts['count']}")
 
    if attempts["count"] == 1:
        raise ConnectionError("shipping service unavailable")
 
    return f"订单 {order_id} 已到达上海转运中心"
 
 
agent = create_agent(
    model="gpt-5.5",
    tools=[query_shipping],
    system_prompt="处理物流问题时必须调用 query_shipping 获取最新状态。",
    middleware=[
        ToolRetryMiddleware(
            tools=["query_shipping"],
            max_retries=2,
            retry_on=(ConnectionError, TimeoutError),
            initial_delay=0.0,
            backoff_factor=0.0,
            jitter=False,
            on_failure="continue",
        )
    ],
)
 
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "查询订单 A1024 的物流"}
    ]
})
 
print(result["messages"][-1].content)

运行结果

代码片段Text
物流接口调用次数：1
物流接口调用次数：2
订单 A1024 已到达上海转运中心。

第一次调用失败后，ToolRetryMiddleware 捕获 ConnectionError，然后再次执行 Tool。

第二次成功以后，模型拿到正常的 ToolMessage，继续完成回答。

这里真正重要的是：

代码片段Python
retry_on=(ConnectionError, TimeoutError)

而不是单纯把：

代码片段Python
max_retries=2

设得更大。

线上项目不要写成“所有异常都自动重试”。

例如：

代码片段Text
401 Unauthorized
403 Forbidden
参数校验失败
业务规则拒绝

这些错误继续重试基本没有意义。

需要注意的是，除了要明确重试条件外，有些重要业务接口也要考虑幂等性设计。

比如，退款、创建订单、发送消息这类 Tool 要更加谨慎。

假设第一次退款其实已经成功，只是客户端没有收到响应。

如果 Middleware 自动再执行一次，就可能重复退款。

因此这类 Tool 除了设置合理的重试条件，还应该在业务层设计幂等机制。

四、场景二：长对话要主动控制上下文

Agent 运行一段时间之后，另一个很常见的问题就是上下文越来越大。

尤其是带 Tool 的 Agent。

一次任务可能不断产生：

代码片段Text
HumanMessage
AIMessage
ToolMessage
AIMessage
ToolMessage
AIMessage
...

如果 Tool 返回的是 JSON、搜索结果或者数据库记录，上下文增长会更快。

问题不只是 Token 成本增加。

大量已经失去价值的旧 Tool 结果，还可能干扰模型当前的判断。

LangChain 中比较常用的两个 Middleware 是：

代码片段Text
SummarizationMiddleware
ContextEditingMiddleware

SummarizationMiddleware 更适合处理长对话，把较早的消息整理成摘要。

ContextEditingMiddleware 更适合工具调用频繁的 Agent，用来清理旧 Tool 输出。

下面给一个故障排查 Agent 加入历史摘要，同时限制一次运行中的模型调用次数。

代码片段Python
from langchain.agents import create_agent
from langchain.agents.middleware import (
    ModelCallLimitMiddleware,
    SummarizationMiddleware,
)
 
 
agent = create_agent(
    model="gpt-5.5",
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("messages", 8),
            keep=("messages", 4),
        ),
        ModelCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ],
)
 
conversation_history = [
    {"role": "user", "content": "旧版客户端开始出现请求失败。"},
    {"role": "assistant", "content": "请提供服务端错误码。"},
    {"role": "user", "content": "错误码为 INVALID_FIELD。"},
    {"role": "assistant", "content": "可能仍在发送废弃字段。"},
    {"role": "user", "content": "新版客户端偶尔也会失败。"},
    {"role": "assistant", "content": "需要继续检查路由和配置。"},
    {"role": "user", "content": "失败请求命中了下线节点。"},
    {"role": "assistant", "content": "缓存中可能还有旧路由配置。"},
]
 
result = agent.invoke({
    "messages": conversation_history
    + [{"role": "user", "content": "整理目前确认的故障原因"}]
})
 
print(result["messages"][-1].content)

运行结果

一次典型结果类似：

代码片段Text
目前确认了两个原因：旧版客户端仍在发送已废弃字段，
部分请求又因缓存中的旧配置被路由到了下线节点。

达到设定阈值以后，较早的消息会被压缩。

主模型看到的上下文不再是：

代码片段Text
全部历史消息
+
当前问题

而更接近：

代码片段Text
历史摘要
+
最近几条原始消息
+
当前问题

线上系统通常更适合按 Token 数量或者上下文使用比例设置触发条件，而不是简单按照消息条数。

摘要不能替代数据存储

这一点很重要。

摘要一定会丢失细节。

例如：

代码片段Text
订单金额
合同原文
完整代码
SQL
Tool 返回的完整 JSON

不要只依赖摘要长期保存。

更合理的结构是：

代码片段Text
原始数据
   ↓
数据库或对象存储
   ↓
Agent 上下文保存摘要和定位信息
   ↓
需要时通过 Tool 读取原文

Middleware 负责控制上下文。

数据库负责保存事实。

两者解决的是不同问题。

给 Agent 设置执行上限

示例中还有：

代码片段Python
ModelCallLimitMiddleware(
    run_limit=6
)

它解决的是另一种常见问题：

代码片段Text
模型
 ↓
Tool
 ↓
模型
 ↓
Tool
 ↓
模型
 ↓
Tool
 ↓
……

如果 Agent 没有找到合适的退出条件，可能持续调用模型和 Tool。

生产环境最好给这类执行链设置一个明确上限。

它的作用很像保险丝。

五、场景三：高风险 Tool 要增加人工审批

查询订单或者查询天气，一般可以直接让 Agent 执行。

但下面这些 Tool 就不一样：

代码片段Text
退款
删除文件
发送邮件
修改数据库
执行 SQL
修改权限
生产环境操作

一旦执行，外部系统就发生了真实变化。

这时真正需要防止的不是：

模型回答错了。

而是：

模型把错误判断执行了。

HumanInTheLoopMiddleware 可以在模型已经决定调用 Tool 之后，在 Tool 真正运行之前暂停 Agent。

过程大致是：

代码片段Text
模型生成 Tool Call
        ↓
Middleware 检查
        ↓
暂停 Agent
        ↓
人工审核
        ↓
approve / edit / reject
        ↓
继续执行

例如只审批超过 500 元的退款：

代码片段Python
from langchain.agents import create_agent
from langchain.agents.middleware import (
    HumanInTheLoopMiddleware,
    ToolCallRequest,
)
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
 
 
@tool
def refund_order(order_id: str, amount: float) -> str:
    """为指定订单退款。"""
    return f"订单 {order_id} 已退款 {amount:.2f} 元"
 
 
def large_refund(request: ToolCallRequest) -> bool:
    return float(request.tool_call["args"]["amount"]) > 500
 
 
agent = create_agent(
    model="gpt-5.5",
    tools=[refund_order],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "refund_order": {
                    "allowed_decisions": [
                        "approve",
                        "edit",
                        "reject",
                    ],
                    "when": large_refund,
                }
            }
        )
    ],
    checkpointer=InMemorySaver(),
)
 
config = {
    "configurable": {
        "thread_id": "refund-A1024"
    }
}
 
pending = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "为订单 A1024 退款 880 元",
            }
        ]
    },
    config=config,
    version="v2",
)
 
print(pending.interrupts[0].value["action_requests"])
 
completed = agent.invoke(
    Command(
        resume={
            "decisions": [
                {"type": "approve"}
            ]
        }
    ),
    config=config,
    version="v2",
)
 
print(completed.value["messages"][-1].content)

运行结果

代码片段Text
[{'name': 'refund_order',
  'arguments': {'order_id': 'A1024', 'amount': 880}}]

订单 A1024 已退款 880.00 元。

第一次执行 invoke() 时，模型虽然已经生成了：

代码片段Text
refund_order(
    order_id="A1024",
    amount=880
)

但是 Tool 还没有真正执行。

Agent 会停在中断位置。

人工审批完成以后，再通过：

代码片段Python
Command(resume=...)

继续之前的任务。

这也是 HITL 为什么需要 Checkpointer。

Agent 暂停之后，需要保存：

代码片段Text
当前 State
执行位置
待调用的 Tool
Tool 参数

否则下一次请求没有办法恢复之前的执行。

示例中的：

代码片段Python
InMemorySaver()

适合本地演示。

生产环境如果存在真正的人工审批流程，应该使用持久化 Checkpointer，否则进程重启后等待审批的任务可能无法继续。

不要让所有 Tool 都进入审批

HITL 也不能滥用。

如果查询一个订单都要求人工点击一次确认，Agent 很快就失去自动化价值。

更合理的是按风险划分。

例如：

代码片段Text
查询订单
→ 自动执行

小额退款
→ 自动执行

大额退款
→ 人工审批

HITL 的作用不是让人参与所有步骤。

而是在真正高风险的节点保留人的最终决定权。

六、场景四：权限校验要放在 Tool 执行入口

假设客服系统有两个角色：

代码片段Text
support
finance

客服只能：

代码片段Text
query_order

财务可以：

代码片段Text
query_order
refund_order

当然可以在 System Prompt 中告诉模型：

代码片段Text
客服不能执行退款。

但这只能降低模型发出错误 Tool Call 的概率。

不能把它当成真正的权限系统。

更可靠的做法是在 Tool 执行之前，再做一次系统级权限检查。

下面实现一个简单的权限 Middleware。

在实现之前，先看一下自定义 Middleware 的基本语法。AgentMiddleware 是 LangChain 提供的一个基类，可以通过继承它来创建自己的中间件。如下：

代码片段Python
class MyMiddleware(AgentMiddleware):
    def before_agent(self, state, runtime):
        # 在 Agent 运行之前执行
        return None
 
    def wrap_tool_call(self, request, handler):
        # 在 Tool 被调用时执行
        return handler(request)

其中 before_agent() 接收当前状态 state 和运行时 runtime，适合做前置校验；wrap_tool_call() 接收 ToolCallRequest 和原始的处理函数 handler，handler 就是真正执行 Tool 的函数，我们需要决定是调用它还是跳过它。创建好的中间件实例放入 middleware 列表，create_agent 会按顺序执行它们。

有了上面的了解，我们继续实现这个 Middleware。

代码片段Python
from collections.abc import Callable
from dataclasses import dataclass
 
from langchain.agents import create_agent
from langchain.agents.middleware import (
    AgentMiddleware,
    AgentState,
    ToolCallRequest,
)
from langchain.messages import ToolMessage
from langchain.tools import tool
from langgraph.runtime import Runtime
 
 
@dataclass
class AccessContext:
    user_id: str
    role: str
 
 
class AccessMiddleware(AgentMiddleware):
    role_tools = {
        "support": {"query_order"},
        "finance": {
            "query_order",
            "refund_order",
        },
    }
 
    def before_agent(
        self,
        state: AgentState,
        runtime: Runtime,
    ) -> dict | None:
        if not runtime.context.user_id:
            raise PermissionError("缺少当前用户身份")
 
        return None
 
    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable,
    ):
        context = request.runtime.context
        tool_name = request.tool_call["name"]
 
        allowed = self.role_tools.get(
            context.role,
            set(),
        )
 
        if tool_name not in allowed:
            return ToolMessage(
                content=(
                    f"当前角色 {context.role} "
                    f"无权调用 {tool_name}"
                ),
                tool_call_id=request.tool_call["id"],
                name=tool_name,
                status="error",
            )
 
        return handler(request)
 
 
@tool
def query_order(order_id: str) -> str:
    """查询订单详情。"""
    return f"订单 {order_id} 当前状态为已签收"
 
 
@tool
def refund_order(
    order_id: str,
    amount: float,
) -> str:
    """为订单发起退款。"""
    return f"订单 {order_id} 已退款 {amount:.2f} 元"
 
 
agent = create_agent(
    model="gpt-5.5",
    tools=[
        query_order,
        refund_order,
    ],
    middleware=[
        AccessMiddleware()
    ],
    context_schema=AccessContext,
)
 
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "为订单 A1024 退款 100 元",
            }
        ]
    },
    context=AccessContext(
        user_id="u-2048",
        role="support",
    ),
)
 
print(result["messages"][-1].content)

运行结果

代码片段Text
当前客服角色没有退款权限，本次操作未执行。

这里用了两个 Hook。

before_agent() 在 Agent 开始运行时检查当前用户身份。

而：

代码片段Python
wrap_tool_call()

真正负责 Tool 权限。

客服请求退款时：

代码片段Text
模型生成 refund_order
        ↓
AccessMiddleware
        ↓
检查当前角色
        ↓
权限不足
        ↓
不执行 handler()
        ↓
返回错误 ToolMessage

因此真正的 refund_order() 根本不会运行。

这就是 Prompt 和 Middleware 的区别。

Prompt 可以告诉模型：

不要这么做。

Middleware 可以保证：

即使模型这么做了也执行不了。

真实项目中，这一层还可以继续接入已有的 RBAC、租户权限或者业务权限服务，而不是把所有规则直接写死在 Middleware 代码中。

七、多个 Middleware 要注意执行顺序

一个实际 Agent 通常不会只有一个 Middleware。

例如：

代码片段Python
middleware=[
    AccessMiddleware(),
    ToolRetryMiddleware(...),
    SummarizationMiddleware(...),
    ModelCallLimitMiddleware(...),
]

这时候顺序会开始影响行为。

Node-style Hook 中：

代码片段Text
before_*

一般按照 Middleware 的注册顺序执行。

而：

代码片段Text
after_*

则反向执行。

Wrap-style Middleware 则是一层一层包裹。

假设：

代码片段Python
middleware=[
    A,
    B,
    C,
]

可以近似理解为：

代码片段Text
A(
    B(
        C(
            原始调用
        )
    )
)

也就是说，第一个 Middleware 位于最外层。

这在实际开发中很重要。

例如权限检查和重试同时存在。

权限不足属于明确的业务拒绝，不应该被 Retry Middleware 当成“临时失败”反复执行。

再比如敏感信息处理和日志同时存在。

必须想清楚日志记录的是：

代码片段Text
原始数据

还是：

代码片段Text
脱敏后的数据

否则即使最后传给模型的数据已经脱敏，敏感信息也可能提前进入日志系统。

当 Middleware 数量开始增加时，一个很实用的方法是直接把执行链画出来：

代码片段Text
用户请求
   ↓
身份检查
   ↓
上下文处理
   ↓
Model
   ↓
Tool Call
   ↓
权限检查
   ↓
人工审批
   ↓
Retry
   ↓
Tool

很多组合问题，看执行链比盯着 middleware=[] 配置更容易发现。

八、项目中的选择建议

真正开发时，可以直接从问题反推 Middleware。

外部接口偶发超时

使用：

代码片段Text
ToolRetryMiddleware

只重试临时异常，并检查 Tool 是否具备幂等性。

模型调用偶发失败

使用：

代码片段Text
ModelRetryMiddleware

设置合理的重试次数。

主模型不可用

使用：

代码片段Text
ModelFallbackMiddleware

提前准备能力和成本都能接受的备用模型。

上下文持续增长

根据实际情况使用：

代码片段Text
SummarizationMiddleware
ContextEditingMiddleware

长对话优先考虑摘要。

ToolMessage 特别多时，再考虑清理旧 Tool 结果。

Agent 容易循环执行

使用：

代码片段Text
ModelCallLimitMiddleware
ToolCallLimitMiddleware

给一次运行设置明确边界。

Tool 会修改外部系统

例如退款、删除、发送、写数据库。

优先评估：

代码片段Text
HumanInTheLoopMiddleware

是否需要加入审批。

规则和自己的业务权限相关

例如：

代码片段Text
用户角色
部门
租户
额度
业务状态

通常更适合实现自定义 Middleware。

九、理解 Middleware 的实用方法

学习 Middleware 没必要先记住大量类名。

更实用的是每次遇到需求时先回答三个问题。

这段逻辑应该发生在哪里

例如：

代码片段Text
Agent 开始前
模型调用前
模型调用后
Tool 执行前
Tool 执行过程中

位置先确定，Hook 的范围基本也就确定了。

这是修改 State 还是控制调用

如果主要是：

代码片段Text
补充上下文
裁剪消息
更新 State

优先考虑 Node-style Hook。

如果需要：

代码片段Text
阻止调用
重试
捕获异常
切换模型
检查权限

通常更适合 Wrap-style Hook。

这是通用问题还是业务规则

如果是：

代码片段Text
重试
Fallback
摘要
调用次数限制
HITL

先找 LangChain 已有的 Middleware。

如果是：

代码片段Text
部门权限
租户隔离
业务额度
审批规则

再考虑自定义 Middleware。

这样理解以后，Middleware 就不再是一堆需要记住的 API。

它本质上是一组插入 Agent 执行链的控制点。

总结

Middleware 是用来处理那些贯穿 Agent 执行过程的控制逻辑。当我们遇到调用失败后要重试、上下文过长、HITL 和权限校验等问题时，可以考虑使用相应的 Middleware。

本文介绍的重试、上下文控制、HITL 和权限校验，看起来解决的是不同问题，但最终目标其实一致：让 Agent 的执行过程能够被限制、暂停、恢复和拦截。