Agent的中间件，本质上是AOP（面向切面编程）思想在Agent执行流程中的具体应用。 它让你能够在Agent执行核心任务（调用LLM、使用工具等）的“前后”或“左右”，插入可复用、可组合的额外逻辑，从而横切（Cross-cutting）整个处理过程。

        一个基础的Agent通常只负责“接收输入 -> 调用LLM -> 调用工具 -> 返回输出”。但在真实应用中，我们几乎总是需要处理一些与核心业务逻辑无关的“横切关注点”，例如：

• 日志记录：记录每次Agent调用的输入、输出、耗时，用于调试和监控。

• 权限校验：在Agent执行动作前，检查当前用户是否有权限调用某个特定工具。

• 缓存：对于相同的或相似的LLM请求，直接返回缓存结果，节省成本和延迟。

• 限流/熔断：防止Agent过度调用昂贵的LLM或外部API。

• 输入/输出预处理/后处理：例如对用户输入进行敏感词过滤，或将Agent的输出格式化为统一的JSON结构。

• 错误处理与重试：当LLM调用或工具调用失败时，统一捕获异常并执行重试或降级策略。

        如果没有中间件，这些逻辑就需要直接写在Agent的核心执行循环里，导致代码臃肿、难以维护、无法复用。中间件提供了一个优雅的解决方案：解耦横切关注点，并支持插件化、可组合的架构。

Middleware的核心工作原理

        Middleware的核心是一个洋葱模型（Onion Model）。这与许多Web框架（如Express.js、Koa、Flask的before/after request钩子）类似，但作用在Agent的执行链路上。

核心概念：

• Handler：Agent的核心处理函数，它接收一个请求（输入），经过一系列处理，返回一个响应（输出）。handler 就是洋葱的中心。

• Middleware：一个接收 next 参数的函数。next 指向“链中的下一个中间件”或最终的 handler。

• 执行流程：当一个请求到来时，它会从第一个中间件开始，逐层进入，直到核心Handler，然后再逐层返回。每一个中间件都可以在调用 next 之前（前置逻辑）和 next 返回之后（后置逻辑）执行代码。

@before_model中间件

        @before_model是 LangChain Agent 执行流程中的前置拦截器，在模型调用前触发。其核心功能包括：

• 消息预处理：裁剪/删除/总结历史消息，控制上下文长度，可以参考“超出LLM 上下文解决方案”小节内容。
• 状态管理：读取或修改 Agent 的短期记忆（AgentState）。

@after_model中间件操作短期记忆

@after_model是 LangChain Agent 执行流程中的后置拦截器，在模型生成响应后触发。其核心功能包括：

• 输出校验：验证模型响应的合规性（如敏感词过滤）
• 消息预处理：多轮对话中，进行裁剪/删除/总结历史消息，控制上下文长度，可以参考“超出LLM 上下文解决方案”小节内容。
• 状态管理：基于模型输出动态修改 Agent 的短期记忆（AgentState）。

🔁 Agent 级别

💬 模型调用 (LLM) 级别

🛠️ 工具调用 (Tool) 级别

⚙️ 函数/节点级别

⛓️ 调用链级别 (Wrap-Style)

 

示例：

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime

from agent.my_llm import llm


@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的天气
    Args:
        city (str): 要查询天气的城市名称

    Returns:
        str: 包含城市天气信息的字符串
    """
    return f"{city}的天气是晴朗的，温度是25摄氏度"

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict:
    # 打印当前消息
    print("当前state：", state)
    print("当前runtime：", runtime)

    messages = state["messages"]
    if len(messages) > 5:
        # 保留最后3条消息
        retain_msg_count = 3

        # 如果倒数第3条消息是工具调用，那么就保留最后2条消息
        if messages[-retain_msg_count].type == "tool":
            retain_msg_count =2

        #删除的消息
        print("删除的消息：", messages[:-retain_msg_count])

        # 删除消息
        return {"messages": [RemoveMessage(id=msg.id) for msg in messages[:-retain_msg_count]]}
    return None


agent = create_agent(
    model=llm,
    tools=[get_weather],
    middleware=[trim_messages],
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "session_001"}}

# 模拟对话
response1 = agent.invoke({"messages": [{"role": "user", "content": "你好，我是张三"}]}, config=config)
print(response1["messages"][-1].content)
print("***"*20)
response2 = agent.invoke({"messages": [{"role": "user", "content": "今天北京天气好吗？"}]}, config=config)
print(response2["messages"][-1].content)
print("***"*20)
response3 = agent.invoke({"messages": [{"role": "user", "content": "上海天气怎么样？"}]}, config=config)
print(response3["messages"][-1].content)
print("***"*20)
final_response = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫什么？"}]}, config=config)
print(final_response["messages"][-1].content)