跳到主要内容LangChain Agent 中间件详解：节点式与包装式钩子应用 | 极客日志PythonAI算法
LangChain Agent 中间件详解：节点式与包装式钩子应用
综述由AI生成介绍 LangChain 中 Agent 中间件（Middleware）的核心概念与应用。中间件通过 Hooks 钩子实现对智能体执行流程的拦截与自定义控制，分为节点式钩子（如 before_agent、after_model）和包装式钩子（如 wrap_tool_call）。主要应用场景包括日志记录、提示词转换、重试逻辑及安全防护等。文中提供了基于通义千问模型的完整 Python 代码示例，演示了如何定义中间件函数并将其集成到 Agent 创建过程中，展示了从启动拦截到工具调用监控的完整生命周期管理。
AiEngineer27 浏览 Agent 中间件（Middleware）概述
中间件的作用是对智能体的每一步工作进行控制和自定义的执行。通过 Hooks 钩子实现拦截，自定义中间件可以简单的使用装饰器来定义。
核心概念
- 节点式钩子（执行点顺序拦截）
before_agent：Agent 执行之前拦截after_agent：Agent 执行后拦截before_model：模型执行前拦截after_model：模型执行后拦截
- 包装式钩子（针对工具和模型）
wrap_model_call：每个模型调用时候拦截wrap_tool_call：每个工具调用时候拦截
应用场景
- 日志记录、分析、调试
- 转换提示词、工具选择
- 重试、备用、提前终止等逻辑控制
- 安全防护、个人身份检测等
LangChain 中内置了一些基础的中间件，参见官方文档。
代码示例
""" LangChain Agent 中间件（Middleware）示例（基于通义 ChatTongyi）
本示例重点演示：
1. 中间件的作用：对智能体的每一步工作进行控制和自定义的执行
2. 节点式钩子（执行点顺序拦截）
3. 包装式钩子（针对工具和模型）
4. 中间件的应用场景
"""
import os
from typing import Any, Callable
from dotenv import load_dotenv
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import (
after_agent,
after_model,
before_agent,
before_model,
wrap_model_call,
wrap_tool_call,
)
from langgraph.runtime import Runtime
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
def init_chat_model() -> ChatTongyi:
"""初始化 ChatTongyi 聊天模型实例。"""
load_dotenv()
api_key = os.getenv("DASHSCOPE_API_KEY") or os.getenv("API_KEY")
api_key:
ValueError()
os.environ[] = api_key
chat = ChatTongyi(model=)
chat
() -> :
() -> :
()
() -> :
()
() -> :
()
() -> :
state[]:
()
() -> :
max_retries =
attempt (max_retries):
:
()
handler(request)
Exception e:
attempt == max_retries - :
e
()
() -> :
()
()
:
result = handler(request)
()
result
Exception e:
()
() -> :
model = init_chat_model()
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
monitor_tool,
retry_on_error,
log_latest_message,
log_before_model,
log_completion,
log_before_agent,
],
)
agent
() -> :
( * )
()
( * )
()
( * )
res = agent.invoke({: [{: , : user_question},]})
( * )
()
()
(res)
( * )
()
() -> :
( * )
()
( * )
()
()
()
()
()
()
()
()
()
()
()
()
()
()
()
()
()
()
( * )
()
agent = create_agent_with_middleware()
user_question =
invoke_agent_with_middleware(agent, user_question)
( * )
()
( * )
__name__ == :
main()
if
not
raise
"未找到 DASHSCOPE_API_KEY 或 API_KEY 环境变量，请先在 .env 或系统环境中配置后再运行。"
"DASHSCOPE_API_KEY"
"qwen3-max"
return
@tool(description="查询天气")
def
get_weather
str
"""一个简单的天气查询工具。"""
return
"晴天"
@before_agent
def
log_before_agent
state: AgentState, runtime: Runtime
None
"""Agent 执行之前的拦截钩子。"""
print
f"[before_agent] Starting agent with {len(state['messages'])} messages"
@after_agent
def
log_completion
state: AgentState, runtime: Runtime
None
"""Agent 执行之后的拦截钩子。"""
print
f"[after_agent] Agent completed with {len(state['messages'])} messages"
@before_model
def
log_before_model
state: AgentState, runtime: Runtime
None
"""模型执行之前的拦截钩子。"""
print
f"[before_model] About to call model with {len(state['messages'])} messages"
@after_model
def
log_latest_message
state: AgentState, runtime: Runtime
None
"""模型执行之后的拦截钩子。"""
if
"messages"
print
f"[after_model] {state['messages'][-1].content}"
@wrap_model_call
def
retry_on_error
request: Any, handler: Callable
Any
"""模型调用的包装钩子，实现重试逻辑。"""
3
for
in
range
try
print
"[wrap_model_call]"
return
except
as
if
1
raise
print
f"[wrap_model_call] Retry {attempt + 1}/{max_retries} due to: {e}"
@wrap_tool_call
def
monitor_tool
request: Any,
handler: Callable[[Any], Any],
Any
"""工具调用的包装钩子，实现监控和日志记录。"""
print
f"[wrap_tool_call] Executing tool: {request.tool_call['name']}"
print
f"[wrap_tool_call] Arguments: {request.tool_call['args']}"
try
print
f"[wrap_tool_call] Tool completed successfully"
return
except
as
print
f"[wrap_tool_call] Tool failed: {e}"
raise
def
create_agent_with_middleware
Any
"""创建一个带有中间件的 Agent 智能体。"""
return
def
invoke_agent_with_middleware
agent: Any, user_question: str
None
"""调用带有中间件的 Agent，并打印结果。"""
print
"="
80
print
"【示例】LangChain Agent 中间件（Middleware）"
print
"="
80
print
f"用户问题：{user_question}"
print
"-"
80
"messages"
"role"
"user"
"content"
print
"-"
80
print
"最终结果："
print
"**********"
print
print
"="
80
print
def
main
None
"""入口函数：演示 LangChain Agent 中间件的完整功能。"""
print
"="
80
print
"LangChain Agent 中间件（Middleware）示例（基于通义 ChatTongyi）"
print
"="
80
print
print
"中间件的作用：对智能体的每一步工作进行控制和自定义的执行"
print
print
"节点式钩子（执行点顺序拦截）："
print
" - before_agent: agent 执行之前拦截"
print
" - after_agent: agent 执行后拦截"
print
" - before_model: 模型执行前拦截"
print
" - after_model: 模型执行后拦截"
print
print
"包装式钩子（针对工具和模型）："
print
" - wrap_model_call: 每个模型调用时候拦截"
print
" - wrap_tool_call: 每个工具调用时候拦截"
print
print
"应用场景："
print
" - 日志记录、分析、调试"
print
" - 转换提示词、工具选择"
print
" - 重试、备用、提前终止等逻辑控制"
print
" - 安全防护、个人身份检测等"
print
"="
80
print
"今天天气如何呀，如何穿衣"
print
"="
80
print
"示例执行完毕，你可以根据需要修改中间件逻辑，实现更多自定义功能。"
print
"="
80
if
"__main__"
运行
python3 AI_LLM_RAG_Agent_Dev/38_LangChain_Agent_Middleware.py
运行结果
================================================================================ LangChain Agent 中间件（Middleware）示例（基于通义 ChatTongyi） ================================================================================ 中间件的作用：对智能体的每一步工作进行控制和自定义的执行 节点式钩子（执行点顺序拦截）： - before_agent: agent 执行之前拦截 - after_agent: agent 执行后拦截 - before_model: 模型执行前拦截 - after_model: 模型执行后拦截 包装式钩子（针对工具和模型）： - wrap_model_call: 每个模型调用时候拦截 - wrap_tool_call: 每个工具调用时候拦截 应用场景： - 日志记录、分析、调试 - 转换提示词、工具选择 - 重试、备用、提前终止等逻辑控制 - 安全防护、个人身份检测等 ================================================================================================================================================================ 【示例】LangChain Agent 中间件（Middleware） ================================================================================ 用户问题：今天天气如何呀，如何穿衣 -------------------------------------------------------------------------------- [before_agent] Starting agent with 1 messages [before_model] About to call model with 1 messages [wrap_model_call][after_model][wrap_tool_call] Executing tool: get_weather [wrap_tool_call] Arguments: {}[wrap_tool_call] Tool completed successfully [before_model] About to call model with 3 messages [wrap_model_call][after_model] 今天是晴天，建议穿着轻便、透气的衣物，比如短袖、薄长裤或裙子。如果白天阳光强烈，可以戴帽子和太阳镜，并注意防晒。同时，根据早晚温差适当增减衣物哦！ [after_agent] Agent completed with 4 messages -------------------------------------------------------------------------------- 最终结果： ********** {'messages':[HumanMessage(content='今天天气如何呀，如何穿衣', additional_kwargs={}, response_metadata={}, id='95a96b47-9198-4bd7-8df0-185ca2cc9711'), AIMessage(content='', additional_kwargs={'tool_calls':[{'function':{'arguments':'{}', 'name':'get_weather'}, 'id':'call_80cbf5d161424a3283cdfbcb', 'index':0, 'type':'function'}]}, response_metadata={'model_name':'qwen3-max', 'finish_reason':'tool_calls', 'request_id':'78350b41-f272-4445-9405-672655c7-0', 'token_usage':{'input_tokens':241, 'output_tokens':11, 'prompt_tokens_details':{'cached_tokens':0}, 'total_tokens':252}}, id='lc_run--019ca83f-f34c-79f0-a3cf-a383708555c7-0', tool_calls=[{'name':'get_weather', 'args':{}, 'id':'call_80cbf5d161424a3283cdfbcb', 'type':'tool_call'}], invalid_tool_calls=[]), ToolMessage(content='晴天', name='get_weather', id='d91e2407-c683-4893-84ab-48796487d2b4', tool_call_id='call_80cbf5d161424a3283cdfbcb'), AIMessage(content='今天是晴天，建议穿着轻便、透气的衣物，比如短袖、薄长裤或裙子。如果白天阳光强烈，可以戴帽子和太阳镜，并注意防晒。同时，根据早晚温差适当增减衣物哦！', additional_kwargs={}, response_metadata={'model_name':'qwen3-max', 'finish_reason':'stop', 'request_id':'1b01b40d-f423-442e-bd04-91de660350c5', 'token_usage':{'input_tokens':269, 'output_tokens':51, 'prompt_tokens_details':{'cached_tokens':0}, 'total_tokens':320}}, id='lc_run--019ca83f-fe1c-71d1-b747-4c68fcda965f-0', tool_calls=[], invalid_tool_calls=[])]}================================================================================================================================================================ 示例执行完毕，你可以根据需要修改中间件逻辑，实现更多自定义功能。 ================================================================================
相关免费在线工具
- 加密/解密文本
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
- RSA密钥对生成器
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
- Mermaid 预览与可视化编辑
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
- 随机西班牙地址生成器
随机生成西班牙地址（支持马德里、加泰罗尼亚、安达卢西亚、瓦伦西亚筛选），支持数量快捷选择、显示全部与下载。 在线工具，随机西班牙地址生成器在线工具，online
- Gemini 图片去水印
基于开源反向 Alpha 混合算法去除 Gemini/Nano Banana 图片水印，支持批量处理与下载。 在线工具，Gemini 图片去水印在线工具，online
- curl 转代码
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
