5、Agent中间件

5.1 中间件概念

上一章结尾提到了记忆带来的消息膨胀问题。在解决它之前，我们先理解一个更通用的概念：中间件（Middleware）。

中间件是一种插入Agent执行流程中的"拦截器"。它可以在Agent执行的各个关键节点上介入，对数据进行加工处理。

中间件可以在以下位置介入：

介入位置	时机	典型用途
before_model	消息发给LLM之前	压缩历史消息、注入额外上下文
after_model	LLM返回结果之后	记录日志、过滤敏感内容
wrap_tool	工具执行前后	人工审核、权限控制

使用中间件的方式也很简单——创建中间件实例，通过middleware参数传入create_agent：

agent = create_agent(
    model=llm,
    tools=tools,
    checkpointer=checkpointer,
    middleware=[middleware_a, middleware_b],   # ← 中间件列表
)

5.2 消息压缩中间件

这就是上一章留下的问题的解决方案。SummarizationMiddleware会在消息发给LLM之前，自动检查消息列表的长度。当消息量超过设定的阈值时，它会用一个独立的LLM调用，把旧消息压缩成一段摘要，从而大幅减少Token消耗。

from langchain.agents.middleware import SummarizationMiddleware
from langchain_openai import ChatOpenAI

summary_middleware = SummarizationMiddleware(
    model=ChatOpenAI(model="gpt-4o-mini"),   # 用于生成摘要的LLM
    trigger=("messages", 100),                # 触发条件：消息数量达到100条时压缩
)

压缩的效果示意：

压缩前（100条消息）：
  [用户:你好, AI:你好, 用户:天气?, AI:晴天, ..., 用户:最新问题]
                                ↓ SummarizationMiddleware 介入
压缩后（2条消息）：
  [系统:以下是之前对话的摘要：用户询问了天气、订单状态等问题..., 用户:最新问题]

trigger参数支持三种触发策略：

触发策略	写法	含义
按消息数量	("messages", 100)	消息数量达到100条时触发压缩
按Token比例	("fraction", 0.5)	Token数达到模型上下文窗口的50%时触发
按Token绝对值	("tokens", 3000)	Token数达到3000时触发

5.3 人工审核中间件

有些操作是高风险的——比如转账、删除数据、发送邮件。即使LLM决定要执行这些操作，我们也希望先让人类确认一下再真正执行。

HumanInTheLoopMiddleware就是做这件事的。它会在指定的工具执行前暂停Agent，等待人类审核。

from langchain.agents.middleware import HumanInTheLoopMiddleware

# 配置哪些工具需要人工审核
hitl_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "transfer_money": True,    # 转账 → 需要审核
        "delete_record": True,     # 删除 → 需要审核
        "get_weather": False,      # 查天气 → 不需要审核
    }
)

完整示例

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

# ===== 1. 定义工具 =====
@tool
def get_weather(city: str) -> str:
    """查询天气"""
    return f"{city}的天气晴朗，气温25度。"

@tool
def transfer_money(amount: int, to_account: str) -> str:
    """转账操作（敏感操作，需要审核）

    Args:
        amount: 转账金额（元）
        to_account: 收款账户名称
    """
    print(f">>> 正在执行转账: {amount}元 → {to_account}")
    return f"成功转账 {amount} 元给 {to_account}。"

# ===== 2. 配置中间件 =====
hitl_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "transfer_money": True,    # 转账需要审核
        "get_weather": False,      # 查天气不需要
    }
)

# ===== 3. 创建Agent =====
llm = init_chat_model("gpt-4o-mini", model_provider="openai")
checkpointer = InMemorySaver()

agent = create_agent(
    model=llm,
    tools=[get_weather, transfer_money],
    middleware=[hitl_middleware],
    checkpointer=checkpointer,        # 人工审核必须配合checkpointer使用
)

def run_demo():
    config = {"configurable": {"thread_id": "thread-1"}}

    # 改为同步的 invoke
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "请帮我转账 100 元给 Alice"}]},
        config=config,
    )
    print("------",result)

    if "__interrupt__" in result:
        interrupt_value = result["__interrupt__"][0].value
        print(f"⚠️ 操作被拦截，等待审核")

        decision = "approve"

        action_requests = interrupt_value.get("action_requests", [])
        decisions = [{"type": decision} for _ in action_requests]
        result = agent.invoke(
            Command(resume={"decisions": decisions}),
            config=config,
        )

    for msg in result.get("messages", []):
        if hasattr(msg, "type") and msg.type == "ai" and not getattr(msg, "tool_calls", None):
            print(f"[Agent]: {msg.content}")

run_demo()

执行流程图示：

5.4 本章小结

1. 中间件是Agent的"拦截器"： 可以在LLM调用前后、工具执行前后介入处理。
2. SummarizationMiddleware： 解决长对话中消息无限增长的问题，自动压缩历史消息为摘要。
3. HumanInTheLoopMiddleware： 对高风险操作增加人工审核环节，避免Agent自动执行敏感操作。
4. 中间件可以叠加使用： 在middleware列表中传入多个中间件，它们会按顺序依次执行。