Agent 进阶：Middleware 全流程拦截（LangChain 执行机制解析）

前言

在前几篇中，我已经实现了：

• 单工具调用
• 多工具协作
• ReAct 多步推理

但一直有一个问题：Agent 内部到底是怎么执行的？能不能“拦截”和“控制”它？

这篇我们来做一件更底层的事情：通过 Middleware（中间件）接管 Agent 执行流程

一、完整代码（Agent Middleware）

from langchain.agents import create_agent
from langchain.agents.middleware import before_agent, after_agent, before_model, after_model, wrap_model_call, wrap_tool_call
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
from langchain.agents import AgentState
from langgraph.runtime import Runtime


@tool(description="查询天气,传入城市名称字符串，返回字符串天气信息")
def get_weather(city: str) -> str:
    return f"{city}:晴天"


@before_agent
def before_agent_middleware(state: AgentState, runtime: Runtime) -> None:
    print(f"[before_agent]，消息数量：{len(state['messages'])}")


@after_agent
def after_agent_middleware(state: AgentState, runtime: Runtime) -> None:
    print(f"[after_agent]，消息数量：{len(state['messages'])}")


@before_model
def before_model_middleware(state: AgentState, runtime: Runtime) -> None:
    print(f"[before_model]，消息数量：{len(state['messages'])}")


@after_model
def after_model_middleware(state: AgentState, runtime: Runtime) -> None:
    print(f"[after_model]，消息数量：{len(state['messages'])}")


@wrap_model_call
def wrap_model_call(request, handler):
    print("模型调用了")
    return handler(request)


@wrap_tool_call
def wrap_tool_call(request, handler):
    print(f"工具执行：{request.tool_call['name']}")
    print(f"工具参数：{request.tool_call['args']}")
    return handler(request)


agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weather],
    middleware=[
        before_agent_middleware,
        after_agent_middleware,
        before_model_middleware,
        after_model_middleware,
        wrap_model_call,
        wrap_tool_call,
    ],
)

res = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "北京明天天气如何？"},
        ]
    }
)

print("**********\n", res)

二、什么是 Middleware？

Middleware（中间件）可以理解为：在 Agent 执行过程中插入“钩子函数”

类比（很好理解）

就像 Java Web：

• Filter（过滤器）
• Interceptor（拦截器）

Agent Middleware 本质一样：在关键节点“插一手”

三、6 大拦截点（核心重点）

1. Agent 执行前

@before_agent

触发时机：

• 刚收到用户请求

用途：

• 日志记录
• 参数校验

2. Agent 执行后

@after_agent

用途：

• 结果统计
• 日志收集

3. 模型调用前

@before_model

模型还没推理

4. 模型调用后

@after_model

已经生成结果

5. 包装模型调用（最强）

@wrap_model_call

可以：

• 改请求
• 改返回值
• 加日志

6. 包装工具调用（重点！）

@wrap_tool_call

可以：

• 监控工具调用
• 修改参数
• 做权限控制

四、执行流程图（非常重要）

整个 Agent 执行流程

用户输入
   ↓
before_agent
   ↓
before_model
   ↓
wrap_model_call（模型执行）
   ↓
after_model
   ↓
wrap_tool_call（工具执行）
   ↓
...（可能多轮）
   ↓
after_agent
   ↓
返回结果

五、结合运行结果，彻底看懂 Agent 执行流程（重点）

我们来看程序的真实输出：

[before_agent]，并附带运行时信息数量：1
[before_model]，并附带运行时信息数量：1
模型调用了
[after_model]，并附带运行时信息数量：2
工具执行：get_weather
工具参数：{'city': '北京'}
[before_model]，并附带运行时信息数量：3
模型调用了
[after_model]，并附带运行时信息数量：4
[after_agent]，并附带运行时信息数量：4

一、整体流程一句话总结

这段日志本质是在描述：模型思考 → 调用工具 → 再思考 → 输出结果

二、逐行拆解执行过程（非常关键）

Step 1：Agent 开始执行

[before_agent]，消息数量：1

当前状态：

messages = [用户问题]

说明：

• 只有一条消息（用户输入）
• Agent 刚刚开始运行

Step 2：模型第一次推理前

[before_model]，消息数量：1

模型准备开始“思考”

Step 3：模型第一次调用

模型调用了

此时模型在做：“这个问题需要查天气 → 调用工具”

Step 4：模型推理结束

[after_model]，消息数量：2

messages 变成：

[
  用户问题,
  AIMessage（包含 tool_calls）
]

重点：这里 AI 还没有回答，而是：决定调用工具

Step 5：工具执行

工具执行：get_weather
工具参数：{'city': '北京'}

执行：

get_weather("北京")

返回：

北京:晴天

Step 6：模型第二次推理前

[before_model]，消息数量：3

当前 messages：

[
  用户问题,
  AIMessage（调用工具）,
  ToolMessage（工具返回结果）
]

Step 7：模型第二次调用

模型调用了

此时模型在做：“我已经拿到天气结果，可以生成最终答案了”

Step 8：模型最终输出

[after_model]，消息数量：4

messages：

[
  用户问题,
  AI调用工具,
  工具返回结果,
  最终回答
]

Step 9：Agent 执行结束

[after_agent]，消息数量：4

整个流程完成 ✅