理解 LangChain Middleware：Agent 从“能跑”到“可控”的核心机制

在前面的几篇文章中，我们已经陆续介绍了 LangChain Agent 开发中的一些基础能力，包括模型配置、Tool 调用、结构化输出和流式输出等。

到这一步，一个 Agent 已经能够完成不少任务。

但进入实际项目后，问题往往会从“Agent 能不能执行”转向“Agent 应该怎样执行”。

例如，模型失败后是否自动重试，不同任务是否切换模型，工具执行前是否检查参数，敏感操作是否需要人工审批，以及模型和工具调用如何统一记录。

这些逻辑通常不属于某一个具体的 Tool，也不适合全部放进 System Prompt。LangChain 提供的 Middleware（中间件），就是用来处理这类贯穿 Agent 执行过程的横切逻辑。

Middleware 关注的不是 Agent 要完成什么任务，而是执行过程应该如何被控制、修改和扩展。

本文将从 Agent 的运行链开始，介绍 Middleware 的位置、生命周期与常见 Hook，以及如何拦截模型和 Tool 的调用过程，并说明它与 Tool、Memory、HITL 等能力之间的关系。

先看一个简化后的 Agent 执行过程：

01用户输入
02   ↓
03Model
04   ↓
05是否调用 Tool
06   ↓
07Tool
08   ↓
09Tool Result
10   ↓
11Model
12   ↓
13最终结果

create_agent() 创建的 Agent 底层运行在 LangGraph Runtime 上。模型节点和工具节点会在循环中不断执行，直到模型不再发起工具调用，或者满足其他终止条件。

Middleware 并不会替代这个循环，而是把自己的逻辑插入执行过程：

01用户输入
02   ↓
03[Middleware]
04   ↓
05Model Request
06   ↓
07[Middleware]
08   ↓
09Model
10   ↓
11[Middleware]
12   ↓
13Tool Call
14   ↓
15[Middleware]
16   ↓
17Tool
18   ↓
19...

因此，可以把 Middleware 理解为：

位于 Agent 执行链不同阶段，用来观察、修改或控制执行过程的一层扩展机制。

它适合处理那些“每个模型调用都可能需要”“多个工具都应该遵守”或者“整个 Agent 都要统一执行”的逻辑。

例如：

• 日志与监控；
• Prompt 和上下文注入；
• 动态模型选择；
• Tool 权限过滤；
• 模型和工具重试；
• Fallback；
• Guardrails；
• PII 检查；
• Human-in-the-loop；
• 调用次数限制。

LangChain 本身的多种预构建能力，也是通过 Middleware 实现的，包括 Summarization、Human-in-the-loop、Model Fallback、Tool Retry、Model Retry 和 PII Detection 等。

理解 Middleware，最重要的是理解它到底可以插在哪里。

当前 LangChain Python Middleware API 提供两类 Hook：Node-style Hook 和 Wrap-style Hook。

Node-style Hook 在 Agent 生命周期的特定位置执行，包括 before_agent、before_model、after_model 和 after_agent。它们可以读取和更新 Agent State，因此适合上下文准备、消息处理、状态记录和生命周期控制等逻辑。

Wrap-style Hook 则围绕一次实际的模型调用或工具调用执行，目前主要包括 wrap_model_call 和 wrap_tool_call。它类似装饰器，可以在调用前修改请求、调用后处理结果，也可以决定是否调用、重复调用或替换原始调用，因此特别适合重试、降级、异常处理、权限检查和日志监控等场景。

两者的核心区别可以理解为：Node-style 是在 Agent 执行链中插入一个处理步骤，Wrap-style 是把一次 Model/Tool 调用包起来并控制它的执行过程。

可以通过这个图来进一步理解：

01              Agent Loop
02
03                 │
04        ┌────────▼────────┐
05        │ before_model    │ ← Node-style
06        └────────┬────────┘
07                 │
08        ┌────────▼─────────────┐
09        │ wrap_model_call      │
10        │                      │
11        │   ┌──────────────┐   │
12        │   │ Model Call   │   │
13        │   └──────────────┘   │
14        │                      │
15        └────────┬─────────────┘
16                 │
17        ┌────────▼────────┐
18        │ after_model     │ ← Node-style
19        └────────┬────────┘
20                 │
21                 ▼

Node-style Hook 对应 Agent 生命周期中的固定节点：

例如一次 Agent 执行中调用了模型三次，那么：

01before_agent
02    ↓
03before_model
04Model
05after_model
06    ↓
07Tool
08    ↓
09before_model
10Model
11after_model
12    ↓
13Tool
14    ↓
15before_model
16Model
17after_model
18    ↓
19after_agent

before_agent 和 after_agent 通常只执行一次，而 before_model、after_model 会跟随 Agent Loop 多次执行。

这使得它们适合处理不同粒度的逻辑。

例如用户鉴权可以放在 before_agent，因为一次请求检查一次即可；模型调用次数统计更适合 before_model；模型输出审查则可以放在 after_model。

另一类是 Wrap-style Hook：

01wrap_model_call
02wrap_tool_call

如前所述，它们不是简单地在某个节点“运行一下”，而是把模型调用或工具调用包起来。

结构类似：

01def middleware(request, handler):
02    # 调用前
03    result = handler(request)
04    # 调用后
05    return result

正因为拿到了 handler，Middleware 可以决定：

• 是否真正继续调用；
• 调用多少次；
• 使用修改后的 request 调用；
• 捕获异常以后重新调用；
• 修改返回结果。

因此，重试、Fallback、缓存、动态路由等需要控制调用过程的逻辑，通常更适合 Wrap-style Hook。

Middleware 一个非常重要的用途，是在模型真正执行之前修改 ModelRequest。

例如，同一个 Agent 中，希望普通问题使用较便宜的模型，而长对话切换到能力更强的模型：

01from collections.abc import Callable
02 
03from langchain.agents import create_agent
04from langchain.agents.middleware import (
05    ModelRequest,
06    ModelResponse,
07    wrap_model_call,
08)
09from langchain.chat_models import init_chat_model
10 
11 
12fast_model = init_chat_model("openai:gpt-5.5-mini")
13strong_model = init_chat_model("openai:gpt-5.5")
14 
15 
16@wrap_model_call
17def select_model(
18    request: ModelRequest,
19    handler: Callable[[ModelRequest], ModelResponse],
20) -> ModelResponse:
21    if len(request.messages) > 10:
22        model = strong_model
23    else:
24        model = fast_model
25 
26    return handler(request.override(model=model))
27 
28 
29agent = create_agent(
30    model=fast_model,
31    tools=[],
32    middleware=[select_model],
33)
34 
35result = agent.invoke({
36    "messages": [
37        {"role": "user", "content": "解释一下 Python 中的生成器"}
38    ]
39})
40 
41print(result["messages"][-1].content)

模型生成内容可能略有不同，下面展示一次典型结果：

01Python 生成器是一种按需产生数据的迭代器……

这里真正需要关注的不是回答内容，而是：

01messages <= 10
02      ↓
03gpt-5.5-mini
04
05messages > 10
06      ↓
07gpt-5.5

request.override() 创建修改后的请求，再交给 handler()。原来的 Agent Loop 不需要知道模型为什么发生了变化。动态模型选择正是 wrap_model_call 的典型使用方式。

同样的机制也可以修改：

• System Prompt；
• Messages；
• Tools；
• Response Format；
• Model。

例如可以根据当前用户角色，只向模型暴露当前用户有权限调用的 Tool。

这里有一个容易忽略的区别：通过 wrap_model_call 临时修改 Messages、Prompt 或 Tools，通常只影响当前这一次模型调用，并不等于修改 Agent State。需要跨后续步骤持续保留的信息，应明确写入 State 或 Store。

很多 Agent 一开始会使用固定 Prompt：

01agent = create_agent(
02    model="openai:gpt-5.5",
03    system_prompt="你是一名企业知识库助手",
04)

但实际系统中，Prompt 往往取决于运行时信息。

例如：

01普通员工
02只能查询公开资料
03
04部门管理员
05可以查询部门内部资料
06
07系统管理员
08允许执行管理操作

如果为每一种角色创建一个 Agent，会产生大量重复配置。

更合适的方式是动态生成 Prompt。

LangChain 提供了 @dynamic_prompt：

01from dataclasses import dataclass
02 
03from langchain.agents import create_agent
04from langchain.agents.middleware import dynamic_prompt, ModelRequest
05 
06 
07@dataclass
08class UserContext:
09    role: str
10 
11 
12@dynamic_prompt
13def build_prompt(request: ModelRequest) -> str:
14    role = request.runtime.context.role
15 
16    return (
17        "你是一名企业内部助手。"
18        f"当前用户角色为：{role}。"
19        "请严格按照该角色的权限回答问题。"
20    )
21 
22 
23agent = create_agent(
24    model="openai:gpt-5.5",
25    middleware=[build_prompt],
26    context_schema=UserContext,
27)
28 
29result = agent.invoke(
30    {
31        "messages": [
32            {"role": "user", "content": "我当前是什么角色？"}
33        ]
34    },
35    context=UserContext(role="管理员"),
36)
37 
38print(result["messages"][-1].content)

模型生成内容可能略有不同，下面展示一次典型结果：

01你当前的角色是管理员。

这里的 context 并不是聊天消息，而是 Runtime Context。

Runtime 中可以保存用户 ID、权限、数据库连接等一次运行所需要的静态依赖；Middleware 可以读取这些信息，再决定本次模型究竟看到什么。

这也是 Middleware 和上下文工程联系非常紧密的原因。

Prompt 不再只是 Agent 创建时写死的一段文字，而可以根据：

01State
02Runtime Context
03Store

动态组合。

Middleware 不只能控制模型，还可以包裹工具调用。

例如一个搜索 Tool 偶尔会出现网络错误：

01@tool
02def search(query: str) -> str:
03    ...

如果异常直接向上传播，整次 Agent 执行可能因此失败。

可以使用 wrap_tool_call：

01from langchain.agents.middleware import wrap_tool_call
02from langchain.messages import ToolMessage
03 
04 
05@wrap_tool_call
06def handle_tool_error(request, handler):
07    try:
08        return handler(request)
09    except Exception as exc:
10        return ToolMessage(
11            content=f"工具调用失败：{exc}",
12            tool_call_id=request.tool_call["id"],
13        )

假设 Tool 抛出网络异常，预期结果类似：

01工具调用失败：connection timeout

此时 Agent 得到的不是未处理异常，而是一条 ToolMessage。

模型可以根据这个结果决定下一步：

01重新调用工具
02    ↓
03换一个工具
04    ↓
05告诉用户当前无法获取数据

LangChain 的 Tool Retry、Model Retry、Model Fallback 等预构建 Middleware，本质上解决的也是类似问题：把“失败以后怎么办”从业务 Tool 中移到统一的执行层。

相比把 try/except 分散写在每一个 Tool 中，这种方式更容易统一配置重试次数、退避策略和错误记录。

但也需要注意：并不是所有错误都应该自动重试。

参数错误、权限不足、明确的业务拒绝通常不应该反复执行。网络抖动、限流或部分临时服务异常，才更适合有限次数的重试和降级。

Agent 与普通函数调用的一个区别，是它会自己决定是否调用 Tool。

这意味着权限不能只写在 Prompt 中：

01“你不能调用 delete_user”

Prompt 是给模型看的指令，不是可靠的权限边界。

真正涉及权限的操作，更适合在执行层控制。例如在 wrap_model_call 中根据用户角色过滤可用 Tool，或者在 Tool 执行前再次校验权限。

Middleware 也适合放置 Guardrails（护栏，是指为模型输出设置的安全边界与校验规则，用于自动拦截、修正或提示不符合预期的生成内容）。

例如：

01before_agent
02检查输入与身份
03
04before_model
05裁剪上下文或检测敏感信息
06
07after_model
08检查模型输出
09
10wrap_tool_call
11检查工具参数和权限
12
13after_agent
14记录最终结果

LangChain 的 Guardrails 可以通过 Middleware 插入 Agent 生命周期，既可以使用规则、正则等确定性检查，也可以使用模型或分类器完成语义检查。

Human-in-the-loop 也建立在类似机制上。

例如 HumanInTheLoopMiddleware 会在模型已经产生 Tool Call、但 Tool 尚未真正执行时检查调用。如果某个 Tool 需要人工审批，就暂停执行，等待用户批准、修改或拒绝后再继续。

日志和监控则更直接。

可以在 before_model 记录：

01请求时间
02模型
03消息数量
04用户 ID

在 after_model 记录：

01Token 使用量
02模型响应
03耗时

再在 wrap_tool_call 中记录 Tool 名称、参数、耗时和异常。

这样观测逻辑不会污染模型、Tool 和业务代码。

简单场景可以直接使用装饰器：

01@before_model
02def log_request(...):
03    ...

如果一个 Middleware 同时需要多个 Hook，或者带有较多配置，则更适合继承 AgentMiddleware。

01from langchain.agents.middleware import AgentMiddleware, AgentState
02from langgraph.runtime import Runtime
03 
04 
05class LoggingMiddleware(AgentMiddleware):
06 
07    def before_model(
08        self,
09        state: AgentState,
10        runtime: Runtime,
11    ) -> dict | None:
12        print(f"Model input messages: {len(state['messages'])}")
13        return None
14 
15    def after_model(
16        self,
17        state: AgentState,
18        runtime: Runtime,
19    ) -> dict | None:
20        print("Model call completed")
21        return None

然后注册：

01agent = create_agent(
02    model="openai:gpt-5.5",
03    tools=[...],
04    middleware=[
05        LoggingMiddleware(),
06    ],
07)

如果一次 Agent 调用了模型两次，可以看到类似：

01Model input messages: 1
02Model call completed
03Model input messages: 3
04Model call completed

这说明 Middleware 不是 Agent 外层的一次性回调，而是实际参与 Agent Loop。

当存在多个 Middleware 时，顺序也很重要。

假设：

01middleware=[
02    middleware_a,
03    middleware_b,
04    middleware_c,
05]

before_* 按注册顺序执行：

01A → B → C

after_* 则反向执行：

01C → B → A

Wrap Hook 会像函数调用一样逐层嵌套：

01A before
02  B before
03    C before
04      Model
05    C after
06  B after
07A after

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

因此，Middleware 列表并不是简单的“功能清单”。

例如一个系统同时包含：

01权限检查
02PII 脱敏
03Prompt 构造
04模型重试
05日志记录

应该明确谁看到原始数据、谁看到处理后的数据，以及异常究竟应该先被谁捕获。

Middleware 很容易和 Tool、Memory、HITL 混在一起，但它们并不是同一个层级的概念。

Tool 是能力。

例如：

01search
02send_email
03query_database

它告诉 Agent 可以执行哪些动作。

Middleware 则控制这些能力如何被使用：

01谁可以调用 send_email
02调用前是否审批
03失败后是否重试
04调用参数是否记录

Memory 保存跨步骤或跨会话需要使用的信息。

Middleware 可以读取 Memory，也可以在适当阶段将信息写入 State 或 Store，但 Middleware 自身不是 Memory。

HITL 同样不是 Middleware 的同义词。

HITL 描述的是“执行过程中需要人工决策”这种机制；在 LangChain 当前实现中，HumanInTheLoopMiddleware 正好使用 Middleware 把这种机制插入 Agent 的执行流程。

实际项目中，它们通常是组合关系：

01Memory
02提供用户偏好和历史信息
03        ↓
04Middleware
05构造当前 Prompt 和 Tool 权限
06        ↓
07Model
08决定是否调用 Tool
09        ↓
10Middleware
11检查调用是否合法
12        ↓
13HITL
14敏感操作请求人工确认
15        ↓
16Tool
17真正执行操作

这样理解以后，各个组件之间的职责会清楚很多。

Middleware 很适合解决横切逻辑，但并不意味着所有代码都应该放进去。

一个简单判断方法是看这段逻辑是否属于“某一个具体业务动作”。

如果代码本身就是一个能力，例如：

01查询订单
02发送邮件
03读取文件

应该实现成 Tool。

如果是长期保存的用户偏好、历史事实或会话状态，应交给 State、Store 或相应 Memory 机制。

如果逻辑控制的是“Agent 在某个执行阶段应该怎样处理”，才更适合 Middleware。

实际项目中可以遵循几个原则：

第一，一个 Middleware 尽量只负责一种明确职责。

不要创建一个同时负责权限、日志、重试、Prompt 修改和内容审核的巨大 Middleware。拆分以后，执行顺序和测试都会更容易理解。

第二，优先使用预构建 Middleware。

重试、Fallback、Summarization、PII、HITL 等已经有对应实现时，没有必要重新维护一套类似逻辑。

第三，区分临时上下文和持久状态。

只想改变当前一次模型看到的内容，可以修改 ModelRequest；希望后续步骤继续使用，则应该更新 State 或 Store。

第四，不要依靠 Prompt 实现真正的权限控制。

Prompt 可以影响模型决策，但涉及数据库写入、邮件发送、删除资源等操作，仍需要在工具执行层进行确定性检查。

第五，Middleware 不宜无限堆叠。

Middleware 越多，执行链越长，也越容易产生顺序依赖。某个 Middleware 修改 Prompt，另一个修改 Tool，再由第三个重试模型，如果缺少清晰边界，问题反而更难定位。

Middleware 最适合承担的，是那些与具体业务能力无关，却必须贯穿 Agent 执行过程的控制逻辑。

Tool 解决 Agent“能做什么”，Memory 解决“记住什么”，Middleware 解决“执行过程中怎样控制”，HITL 则解决“什么时候必须交给人决定”。

本文理解 Middleware 最重要的一点，就是不要把它看成 Agent 外层普通的请求拦截器。它直接参与模型调用、工具调用和 Agent 生命周期，因此能够修改一次模型看到的上下文，也可以改变后续执行路径。

当 Agent 从简单 Demo 走向包含权限、可靠性、安全与监控要求的真实应用后，很多原本散落在 Model、Tool 和业务代码中的控制逻辑，都可以通过 Middleware 获得更清晰的边界。