LangChain Middleware 实战：上下文压缩、人工审批、安全防护与自定义 Hook

当 Agent 只调用一次模型时，代码通常很简单；但进入真实项目后，我们还需要处理很多“模型调用之外”的问题：

对话太长时自动总结
高风险工具执行前等待人工审批
隐藏手机号、邮箱和 API Key
限制模型与工具调用次数
模型或工具失败后重试
记录日志、统计耗时、修改请求和响应

如果把这些逻辑全部写进工具或系统提示词，代码会越来越混乱。LangChain Middleware（中间件）提供了统一的 Agent 生命周期扩展机制。

一、Middleware 到底是什么？

中间件位于 Agent 执行流程的关键位置，可以在模型或工具调用前后读取、修改或拦截数据。

用户输入
  ↓
中间件 before_agent
  ↓
中间件 before_model
  ↓
模型调用
  ↓
中间件 after_model
  ↓
工具调用（如果模型要求）
  ↓
中间件 after_agent
  ↓
最终结果

创建 Agent 时，通过 middleware 列表挂载：

from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[middleware_a, middleware_b],
)

中间件本身不等于工具。工具是模型可选择的业务能力，中间件则由 Agent 框架在特定生命周期自动执行。

二、SummarizationMiddleware：自动压缩长对话

对话消息不断追加后，可能超过模型上下文窗口，也会持续增加 token 成本。SummarizationMiddleware 会在达到阈值时总结较早的消息，并保留近期消息。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=model,
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger={"tokens": 4000},
            keep={"messages": 6},
        )
    ],
)

核心参数可以这样理解：

trigger：什么时候触发总结
keep：总结后保留多少近期内容
summary_prompt：用什么要求生成摘要

运行逻辑是：

消息尚未达到阈值 -> 原样交给模型
消息达到阈值 -> 总结旧消息 -> 保留近期消息 -> 再调用模型

摘要不是删除所有历史，而是把大量旧消息压缩成一段更短的上下文。

三、HumanInTheLoopMiddleware：工具执行前人工审批

天气查询等只读工具风险较低，但发邮件、删除文件、执行 SQL、提交订单等工具不应该由模型直接执行。

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model,
    tools=[get_weather, get_news, read_email_tool, send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
                "get_news": True,
                "read_email_tool": False,
                "send_email_tool": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "发送邮件前需要人工确认",
                },
            },
            description_prefix="工具调用已暂停",
        )
    ],
)

interrupt_on 的含义：

配置	行为
True	该工具调用前触发中断
False	不拦截，直接执行
字典	自定义允许的决策和提示信息

人工审批依赖检查点，因为 Agent 暂停后必须保存运行位置。因此需要 checkpointer 和固定的 thread_id：

config = {"configurable": {"thread_id": "approval-001"}}
result = agent.invoke(input_data, config=config)

恢复时通过 Command(resume=...) 把审批结果送回原来的中断点。这里的中断不是异常，而是工作流主动等待外部决策。

四、PIIMiddleware：保护敏感信息

PII 是 Personally Identifiable Information，即个人可识别信息，例如手机号、邮箱、银行卡和 API Key。

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model=model,
    middleware=[
        PIIMiddleware(
            "api_key",
            strategy="hash",
            apply_to_input=True,
            detector=r"sk-[a-zA-Z0-9]+",
        ),
        PIIMiddleware(
            "phone_number",
            strategy="mask",
            apply_to_input=True,
            detector=detect_phone_number,
        ),
    ],
)

常见处理策略包括遮盖、替换、哈希或阻止。apply_to_input=True 表示数据发送给模型前先检测和处理。

自定义手机号检测器：

import re

def detect_phone_number(content: str):
    return [
        {
            "text": match.group(0),
            "start": match.start(),
            "end": match.end(),
        }
        for match in re.finditer(r"[0-9]{11}", content)
    ]

text 是匹配内容，start 和 end 是它在原字符串中的起止位置，中间件据此准确替换敏感片段。

五、TodoListMiddleware：让复杂任务先规划再执行

面对“读取文件、修改代码、运行测试”这样的多步骤任务，Agent 容易遗漏步骤。TodoListMiddleware 为 Agent 提供待办管理能力。

from langchain.agents.middleware import TodoListMiddleware

agent = create_agent(
    model=model,
    tools=[list_files, read_file, write_file, run_tests],
    middleware=[TodoListMiddleware()],
    system_prompt=(
        "你是代码修复助手。遇到多步骤任务时，"
        "先使用 write_todos 制定计划，再读取文件、修复代码并运行测试。"
    ),
)

它适合代码修复、报告生成、资料研究等步骤多且执行时间长的任务。工具本身仍要做好路径限制和权限保护，不能只依靠提示词保证安全。

六、限制模型和工具调用次数

1. ModelCallLimitMiddleware

防止模型循环调用，控制 token 和费用：

from langchain.agents.middleware import ModelCallLimitMiddleware

ModelCallLimitMiddleware(
    thread_limit=10,
    run_limit=5,
    exit_behavior="end",
)

thread_limit：同一个 thread 的累计模型调用上限
run_limit：单次 agent.invoke() 的模型调用上限
exit_behavior="end"：达到上限后结束
exit_behavior="error"：达到上限后抛出异常

使用 thread_limit 时需要检查点存储器，因为中间件必须跨多轮调用累计次数。

2. ToolCallLimitMiddleware

限制全部工具或指定工具的调用次数：

from langchain.agents.middleware import ToolCallLimitMiddleware

ToolCallLimitMiddleware(
    run_limit=3,
    exit_behavior="end",
)

它可以避免昂贵 API 被频繁调用、数据库遭到重复查询，以及 Agent 陷入工具循环。

exit_behavior 常见取值：

error：直接抛出异常
end：结束本次 Agent
continue：把超限信息交给模型，让模型决定下一步

continue 使用不当仍可能形成循环，因此生产环境通常要同时配置模型调用上限。

七、失败重试与故障转移

1. ToolRetryMiddleware

工具请求可能因网络波动临时失败。ToolRetryMiddleware 可以使用指数退避进行重试：

第一次失败 -> 等待较短时间
第二次失败 -> 等待更长时间
第三次失败 -> 再增加等待时间

jitter（抖动）会给等待时间增加随机偏移，避免大量请求在同一时刻集中重试。

2. ModelRetryMiddleware

用于模型 API 超时、限流或临时连接错误，其重试思想与工具重试相同。需要设置最大次数，不能无限重试。

3. ModelFallbackMiddleware

主模型不可用时切换到备用模型：

主模型调用成功 -> 使用主模型结果
主模型调用失败 -> 尝试备用模型 1
仍然失败 -> 尝试备用模型 2 或抛出错误

故障转移能提升可用性，但不同模型的工具调用、结构化输出和提示词行为可能不完全一致，需要提前测试兼容性。

八、其他实用内置中间件

LLMToolSelectorMiddleware

当 Agent 绑定几十个工具时，全部发送给主模型会增加 token，也会降低选择准确率。该中间件先用一个子模型筛选最相关的少量工具。

常用参数：

model：负责筛选工具的模型
max_tools：最多保留多少工具
always_include：始终保留的工具

LLMToolEmulator

工具还没开发完成时，使用 LLM 模拟工具结果，适合原型验证和流程测试。它不能代替真实接口的集成测试。

ContextEditingMiddleware

在消息发送给模型之前裁剪或编辑上下文，用来减少 token 成本。它通常只修改本次模型请求看到的上下文，不一定修改 Agent 状态中保存的完整消息。

FilesystemFileSearchMiddleware

自动提供 Glob 和 Grep 类工具，让 Agent 可以按文件路径和内容搜索本地工作区：

from langchain.agents.middleware import FilesystemFileSearchMiddleware

FilesystemFileSearchMiddleware(
    root_path="../todo_workspace",
    allowed_extensions=[".py", ".ipynb", ".md"],
    use_ripgrep=True,
    max_file_size_mb=10,
)

root_path 和文件大小限制非常重要，避免 Agent 搜索工作区外的敏感目录或读取超大文件。

九、自定义中间件：Node-style Hooks

Node-style Hook 在生命周期的某个时间点执行一次，适合修改状态、记录日志或改变流程。

1. 装饰器写法

from langchain.agents.middleware import before_model, after_model

@before_model
def inspect_input(state, runtime):
    print("模型调用前：", state["messages"][-1].content)
    return None

@after_model
def inspect_output(state, runtime):
    print("模型调用后：", state["messages"][-1].content)
    return None

2. 类写法

from langchain.agents.middleware import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def before_agent(self, state, runtime):
        print("Agent 开始")
        return None

    def before_model(self, state, runtime):
        print("模型调用前")
        return None

    def after_model(self, state, runtime):
        print("模型调用后")
        return None

    def after_agent(self, state, runtime):
        print("Agent 结束")
        return None

返回 None 表示不更新状态、继续正常流程；返回字典则可以写入状态更新。

十、can_jump_to：由中间件改变执行路径

Hook 不只能观察状态，还能声明可以跳转到 tools、model 或 end：

from langchain.agents.middleware import before_model
from langchain.messages import AIMessage

@before_model(can_jump_to=["end"])
def block_overflow(state, runtime):
    if "overflow" in state["messages"][-1].content:
        return {
            "messages": [AIMessage("上下文窗口溢出，流程终止")],
            "jump_to": "end",
        }
    return None

典型用途：

jump_to="tools"：绕过模型，直接进入工具节点
jump_to="model"：让模型重新生成一次
jump_to="end"：安全拦截并提前终止

任何重试跳转都必须设计停止条件，否则会形成无限循环。

十一、自定义中间件：Wrap-style Hooks

Wrap-style Hook 把实际调用包在中间，既能处理调用前逻辑，也能处理调用后逻辑和异常。

from typing import Callable
from langchain.agents.middleware import (
    wrap_model_call,
    ModelRequest,
    ModelResponse,
)

@wrap_model_call
def log_model_call(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    print("模型调用前")
    response = handler(request)
    print("模型调用后")
    return response

handler(request) 是真正执行模型调用的语句。如果不调用它，模型就不会被请求。

同理，wrap_tool_call 可以包裹工具执行，用于超时、重试、缓存、日志、权限检查和统一异常转换。

Node-style：在某个时间点插入逻辑
Wrap-style：完整包裹一次模型或工具调用

十二、多个中间件的执行顺序

假设配置：

middleware=[MW1(), MW2(), MW3()]

执行顺序类似多层包装：

进入：MW1 before -> MW2 before -> MW3 before -> 真正调用
返回：真正调用 -> MW3 after -> MW2 after -> MW1 after

也就是“进入正序，返回逆序”。可以想象成穿三层外套：先穿第一层、再穿第二层、最后穿第三层；脱的时候顺序相反。

顺序会影响结果。例如 PII 清洗应在日志记录和模型调用之前执行，否则原始敏感信息可能已经被日志保存。

十三、如何选择中间件？

需求	推荐中间件
长对话压缩	SummarizationMiddleware
高风险工具审批	HumanInTheLoopMiddleware
手机号、邮箱、密钥保护	PIIMiddleware
多步骤任务规划	TodoListMiddleware
控制模型成本	ModelCallLimitMiddleware
控制外部 API 次数	ToolCallLimitMiddleware
模型或工具临时失败	Retry 中间件
主模型故障切换	ModelFallbackMiddleware
工具太多	LLMToolSelectorMiddleware
搜索本地代码	FilesystemFileSearchMiddleware
自定义日志、缓存、路由	Node-style / Wrap-style Hook

十四、总结

Middleware 是 Agent 生命周期的扩展层，不是模型工具。
内置中间件解决上下文、安全、审批、重试、限流和文件检索问题。
Node-style Hook 适合在特定时点读取或更新状态。
Wrap-style Hook 适合包裹完整调用，实现日志、重试、缓存和异常处理。
can_jump_to 可以改变 Agent 路径，但必须防止无限循环。
多个中间件进入时正序执行，返回时逆序执行。

当 Agent 从教学示例进入真实业务时，中间件往往比再增加一个工具更重要。它把安全、成本、可靠性和可观测性从业务函数中分离出来，让 Agent 的核心逻辑保持清晰。