Agent 中间件 (Middleware)

中间件是 LangChain 1.0 引入的核心机制，用于在 Agent 执行流程的关键环节（模型调用前后、工具调用前后等）插入可控的拦截与增强逻辑。它借鉴了 Web 框架（Express / Django）的中间件模式，使开发者无需修改 Agent 核心业务逻辑即可实现日志、安全、限流、上下文压缩等横切关注点。

middleware 模块文档
AgentMiddleware 基类文档

1. 中间件架构原理

中间件像洋葱一样层层包裹 Agent 核心循环。用户请求从外向内逐层通过各中间件的 before 逻辑，到达核心后再从内向外逐层通过 after 逻辑。wrap 类 Hook 则直接包裹模型 / 工具的调用过程。

中间件的四大分类：

分类	核心功能	典型中间件
Monitor (监控)	观察执行状态、日志、成本统计	自定义日志中间件
Modify (修改)	修改输入/输出、上下文压缩	`SummarizationMiddleware`、`ContextEditingMiddleware`
Control (控制)	流程阻断、人工介入、重试	`HumanInTheLoopMiddleware`、`ToolRetryMiddleware`
Enforce (强制)	安全过滤、限流、合规	`PIIMiddleware`、`ModelCallLimitMiddleware`

2. 中间件生命周期 — 6 个 Hook

Agent 执行过程中，中间件通过 6 个 Hook 切入。每个 Hook 有明确的职责和执行时机：

Hook	执行时机	典型用途
`before_agent`	Agent 启动前（仅一次）	全局初始化、环境校验、资源准备
`before_model`	每次模型调用前	输入预处理、PII 脱敏、上下文摘要、调用计数
`wrap_model_call`	包裹模型调用过程	缓存、熔断、降级、模型切换、上下文裁剪
`after_model`	每次模型调用后	输出校验、人工审批拦截、调用计数
`wrap_tool_call`	包裹工具调用过程	重试、权限校验、审计日志
`after_agent`	Agent 结束后（仅一次）	资源清理、最终状态记录、报告生成

使用中间件有两种方式：继承 AgentMiddleware 类或使用装饰器函数。

from langchain.agents import create_agent
from langchain.agents.middleware import before_model, wrap_model_call

# 方式一：装饰器（适合简单逻辑）
@before_model
def log_input(state, runtime):
    print(f"输入消息数: {len(state['messages'])}")

# 方式二：类（适合复杂逻辑，可组合多个 Hook）
from langchain.agents.middleware.types import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def before_model(self, state, runtime):
        print("before model")

    def after_model(self, state, runtime):
        print("after model")

3. 内置中间件

3.1 SummarizationMiddleware — 上下文摘要压缩

Hook: before_model | API 文档

当对话历史超出阈值时，自动将旧消息摘要为一段文本，保留最近的 N 条消息。大幅减少 Token 消耗。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import MemorySaver

agent = create_agent(
    "openai:gpt-4o",
    tools=[],
    checkpointer=MemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            trigger=("messages", 20),     # 消息数达到 20 时触发
            keep=("messages", 5),         # 保留最近 5 条，其余压缩为摘要
        )
    ],
)

trigger 支持多种阈值：("messages", N) 按消息数、("tokens", N) 按 Token 数、("fraction", 0.8) 按模型上下文窗口比例。也可以传列表，任一满足即触发。

3.2 PIIMiddleware — 敏感信息脱敏

Hook: before_model + after_model | API 文档

自动检测并处理对话中的 PII（个人身份信息），如邮箱、信用卡号、IP 地址等。

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    "openai:gpt-4o",
    middleware=[
        PIIMiddleware("credit_card", strategy="mask"),   # ****-****-****-1234
        PIIMiddleware("email", strategy="redact"),       # [REDACTED_EMAIL]
        PIIMiddleware("ip", strategy="hash"),            # <ip_hash:a1b2c3d4>
    ],
)

策略	效果	适用场景
`block`	检测到 PII 直接抛异常	完全禁止 PII 进入
`redact`	替换为 `[REDACTED_TYPE]`	日志脱敏、合规
`mask`	部分遮盖（保留末位）	客服 UI、可读性
`hash`	替换为确定性哈希	分析、调试（可关联）

支持自定义 PII 类型，通过正则或自定义检测函数：

PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")

3.3 ModelCallLimitMiddleware — 模型调用限制

Hook: before_model + after_model | API 文档

防止 Agent 陷入无限循环，限制单次运行或线程级别的模型调用次数。

from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    middleware=[
        ModelCallLimitMiddleware(
            run_limit=10,              # 单次运行最多调用 10 次模型
            thread_limit=100,          # 整个线程最多 100 次
            exit_behavior="end",       # 超限时优雅结束（也可选 "error" 抛异常）
        )
    ],
)

3.4 ContextEditingMiddleware — 上下文裁剪

Hook: wrap_model_call | API 文档

当输入 Token 超过阈值时，自动裁剪旧的工具调用结果，释放上下文空间。默认使用 ClearToolUsesEdit 策略。

from langchain.agents.middleware import ContextEditingMiddleware

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    middleware=[ContextEditingMiddleware()],
)

3.5 ModelFallbackMiddleware — 模型故障降级

Hook: wrap_model_call | API 文档

主模型调用失败时，自动按顺序尝试备选模型，直到成功或全部耗尽。

from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="openai:gpt-4o",                # 主模型
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            "openai:gpt-4o-mini",          # 第一备选
            "anthropic:claude-sonnet-4-5-20250929",  # 第二备选
        )
    ],
)

3.6 LLMToolSelectorMiddleware — 智能工具筛选

Hook: wrap_model_call | API 文档

当 Agent 绑定大量工具时，先用一个轻量 LLM 筛选出与当前查询最相关的工具子集，减少 Token 消耗并提升准确率。

from langchain.agents.middleware import LLMToolSelectorMiddleware

agent = create_agent(
    "openai:gpt-4o",
    tools=[tool_1, tool_2, ..., tool_50],
    middleware=[
        LLMToolSelectorMiddleware(
            model="openai:gpt-4o-mini",
            max_tools=5,
            always_include=["search"],     # 始终保留的工具
        )
    ],
)

3.7 ToolRetryMiddleware — 工具调用自动重试

Hook: wrap_tool_call | API 文档

工具调用失败时自动重试，支持指数退避和抖动。

from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,            # 指数退避倍数
            initial_delay=1.0,             # 首次重试延迟（秒）
            on_failure="continue",         # 全部失败后返回错误消息让 LLM 处理
        )
    ],
)

3.8 HumanInTheLoopMiddleware — 人工审批

Hook: after_model | API 文档

当模型决定调用敏感工具时，自动中断执行并等待人工审批（批准 / 编辑 / 拒绝）。需配合 checkpointer 使用以持久化中断状态。

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import MemorySaver

agent = create_agent(
    "openai:gpt-4o",
    tools=[send_email, search],
    checkpointer=MemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"send_email": True},     # send_email 需审批，其他工具自动放行
        )
    ],
)

审批后通过 Command 恢复执行：

from langgraph.types import Command
from langchain.agents.middleware.human_in_the_loop import ApproveDecision

# 批准执行
agent.invoke(Command(resume=ApproveDecision()), config=config)

4. 中间件组合

多个中间件通过 middleware 列表传入，按列表顺序依次执行 before / wrap 逻辑，after 逻辑则按逆序执行。

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    checkpointer=MemorySaver(),
    middleware=[
        PIIMiddleware("credit_card", strategy="mask"),
        SummarizationMiddleware(model="openai:gpt-4o-mini", trigger=("messages", 20)),
        ModelCallLimitMiddleware(run_limit=15),
        ModelFallbackMiddleware("openai:gpt-4o-mini"),
        ToolRetryMiddleware(max_retries=2),
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
    ],
)

5. 自定义中间件

当内置中间件无法满足需求时，可以通过继承 AgentMiddleware 类或使用装饰器编写自定义中间件。

5.1 参数传递机制

不同 Hook 能够访问的参数不同，理解它们是编写自定义中间件的基础：

参数	类型	生命周期	可访问的 Hook
`state` (AgentState)	TypedDict	整个 Agent 会话	`before_agent`、`before_model`、`after_model`、`after_agent`
`request` (ModelRequest)	dataclass	单次模型/工具调用	`wrap_model_call`、`wrap_tool_call`、`dynamic_prompt`
`handler`	Callable	单次包装调用	`wrap_model_call`、`wrap_tool_call`
`runtime`	运行时对象	贯穿会话	所有 Hook（通过 `state` 或 `request.runtime`）

关键规则：

wrap_model_call / wrap_tool_call 中，可通过 request.override(...) 修改 model、messages、tools 等字段
before_model / after_model 等钩子中，只能读取 state，不能访问 request
handler(request) 是责任链中的下一环，调用它才会真正执行模型/工具调用

# ModelRequest 的核心字段
request.model       # 当前模型实例（可通过 override 替换）
request.messages    # 消息列表
request.tools       # 可用工具列表
request.state       # Agent 当前状态
request.runtime     # 运行时上下文，含 request.runtime.context

5.2 装饰器方式 — @dynamic_prompt 动态提示词

@dynamic_prompt 是一个专门用于根据运行时上下文动态生成 System Prompt 的装饰器。

dynamic_prompt API 文档

from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from typing import TypedDict

class Context(TypedDict):
    user_role: str

@dynamic_prompt
def role_based_prompt(request: ModelRequest):
    role = request.runtime.context.get("user_role", "user")
    if role == "expert":
        return "你是一个专业气象分析师，提供详细数据"
    return "你是一个友善的助手，用简单语言回答"

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    middleware=[role_based_prompt],
    context_schema=Context,
)

# 不同上下文会注入不同的 System Prompt
agent.invoke(
    {"messages": [{"role": "user", "content": "北京天气"}]},
    context={"user_role": "expert"},
)

5.3 装饰器方式 — @wrap_model_call 模型动态切换

通过 @wrap_model_call 包装模型调用过程，可实现根据输入复杂度动态切换模型。

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain_openai import ChatOpenAI
from typing import Callable

small_model = ChatOpenAI(model="gpt-4o-mini")
large_model = ChatOpenAI(model="gpt-4o")

HARD_KEYWORDS = ("分析", "推导", "规划", "复杂", "多步骤")

@wrap_model_call
async def dynamic_model_router(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    messages = request.state.get("messages", [])
    last_content = messages[-1].content if messages else ""

    if any(kw in last_content for kw in HARD_KEYWORDS):
        request = request.override(model=large_model)

    return await handler(request)

agent = create_agent(
    model=small_model,
    tools=[...],
    middleware=[dynamic_model_router],
)

5.4 类方式 — 会话持久化中间件

继承 AgentMiddleware 可以在一个类中组合多个 Hook，适合复杂逻辑。

from langchain.agents.middleware import AgentMiddleware, AgentState
import os, json, time

class PersistSessionMiddleware(AgentMiddleware):
    def __init__(self, path: str = "./sessions"):
        super().__init__()
        self.path = path
        os.makedirs(path, exist_ok=True)

    def after_model(self, state: AgentState, runtime) -> None:
        messages = state.get("messages", [])
        if not messages:
            return
        filename = os.path.join(self.path, f"state_{int(time.time())}.json")
        data = [{"role": m.type, "content": m.content} for m in messages]
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

agent = create_agent(
    "openai:gpt-4o",
    tools=[],
    middleware=[PersistSessionMiddleware("./my_sessions")],
)

5.5 类方式 — 安全护栏中间件

利用 @hook_config(can_jump_to=["end"]) 可以在检测到危险输入时直接终止 Agent 执行。

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain_core.messages import AIMessage
from typing import Optional, Dict, Any

class SecurityGuardrail(AgentMiddleware):
    DANGEROUS_KEYWORDS = ["rm -rf", "drop database", "删除所有"]

    @hook_config(can_jump_to=["end"])
    def before_agent(self, state: AgentState, runtime) -> Optional[Dict[str, Any]]:
        messages = state.get("messages", [])
        if not messages:
            return None

        content = messages[-1].content.lower()
        for kw in self.DANGEROUS_KEYWORDS:
            if kw in content:
                return {
                    "messages": [AIMessage(content=f"安全警告：检测到危险操作 '{kw}'，已拦截。")],
                    "jump_to": "end",
                }
        return None

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    middleware=[SecurityGuardrail()],
)

5.6 类方式 — RBAC 权限控制中间件

通过 context_schema 传入用户上下文，在 before_agent 做权限校验，在 before_model 注入用户信息。

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain_core.messages import AIMessage
from typing import TypedDict, Optional, Dict, Any

class UserContext(TypedDict):
    user_id: str
    username: str
    role: str

ROLE_PERMISSIONS = {
    "admin": ["view", "restart", "config"],
    "operator": ["view", "restart"],
    "viewer": ["view"],
}

class RBACMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["end"])
    def before_agent(self, state: AgentState, runtime) -> Optional[Dict[str, Any]]:
        role = runtime.context.get("role", "guest")
        permissions = ROLE_PERMISSIONS.get(role, [])
        if not permissions:
            return {
                "messages": [AIMessage(content="权限不足，请联系管理员。")],
                "jump_to": "end",
            }
        return None

    def before_model(self, state: AgentState, runtime) -> Optional[Dict[str, Any]]:
        return {"user_info": dict(runtime.context)}

agent = create_agent(
    "openai:gpt-4o",
    tools=[...],
    middleware=[RBACMiddleware()],
    context_schema=UserContext,
)

agent.invoke(
    {"messages": [{"role": "user", "content": "查看服务器状态"}]},
    context={"user_id": "001", "username": "zhang", "role": "operator"},
)