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")
if not api_key:
raise ValueError("未找到 DASHSCOPE_API_KEY 或 API_KEY 环境变量，请先在 .env 或系统环境中配置后再运行。")
os.environ["DASHSCOPE_API_KEY"] = api_key
chat = ChatTongyi(model="qwen3-max")
return chat
@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 state["messages"]:
print(f"[after_model] {state['messages'][-1].content}")
# ============================================================================
# 包装式钩子（针对工具和模型）
# ============================================================================
@wrap_model_call
def retry_on_error(request: Any, handler: Callable) -> Any:
"""模型调用的包装钩子，实现重试逻辑。"""
max_retries = 3
for attempt in range(max_retries):
try:
print("[wrap_model_call]")
return handler(request)
except Exception as e:
if attempt == max_retries - 1:
raise e
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:
result = handler(request)
print(f"[wrap_tool_call] Tool completed successfully")
return result
except Exception as e:
print(f"[wrap_tool_call] Tool failed: {e}")
raise
def create_agent_with_middleware() -> Any:
"""创建一个带有中间件的 Agent 智能体。"""
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, # Agent 执行后日志
log_before_agent, # Agent 执行前日志
],
)
return agent
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)
res = agent.invoke({"messages": [{"role": "user", "content": user_question},]})
print("-" * 80)
print("最终结果：")
print("**********")
print(res)
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()
# 创建带有中间件的 Agent
agent = create_agent_with_middleware()
# 调用 Agent，观察中间件的执行过程
user_question = "今天天气如何呀，如何穿衣"
invoke_agent_with_middleware(agent, user_question)
print("=" * 80)
print("示例执行完毕，你可以根据需要修改中间件逻辑，实现更多自定义功能。")
print("=" * 80)
if __name__ == "__main__":
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=[])]}================================================================================================================================================================ 示例执行完毕，你可以根据需要修改中间件逻辑，实现更多自定义功能。 ================================================================================