首页 / 文章 / 使用七种内置中间件增强 Python LangChain 智能体安全性

LangChain AI Agents Python Gemini Middleware LLM

使用七种内置中间件增强 Python LangChain 智能体安全性

了解 LangChain 1.0 中间件如何在不改动 Gemini 智能体核心逻辑的情况下，为其添加摘要生成、调用限制、重试机制、模型备用方案、个人身份信息屏蔽以及人工审核功能。

2568 词

在笔记本中让 LangChain 智能体回答问题只需几分钟时间。但要打造一个可在生产环境中可靠使用的智能体则更为困难：它绝不能无限循环直至耗尽 API 预算，也不能将客户的卡号转发给模型提供商，或是发送未经任何人批准的邮件。LangChain 1.0 通过中间件解决了这些运行方面的问题，这类中间件位于智能体循环周围，以简单列表的形式供用户配置。本指南首先阐述相关概念，接着将七种中间件应用于基于 Gemini 的 Python 智能体，并最后介绍一种自定义中间件，这样你就能清楚地了解每个中间件会带来什么变化以及何时该使用它。

智能体循环周围的检查点

想象一下机场。人们的目的是从一座城市飞往另一座城市，但围绕这一核心行为存在着一系列检查点：值机确认身份，安检扫描行李，登机口核对登机牌，着陆后则由行李提取处处理。这些环节中没有哪个能直接驾驶飞机，飞行员也不会检查行李。每个环节都在核心操作之前或之后执行特定的任务。

中间件将同样的理念应用到智能体上。智能体的核心是一个循环：调用模型，让模型选择工具并执行这些工具，不断重复直到模型给出最终答案。中间件会在不修改该循环结构的情况下在其周围插入检查点：

在文本传递给大语言模型之前去除卡号，这就是安检扫描器。
要求人工批准发送的邮件，则相当于登机口。
在模型调用十次后停止以控制花费，这就是断路器。

如果您使用 TypeScript，LangChain guardrails and middleware 这篇相关文章从 JavaScript 角度阐述了相同的概念；本指南则聚焦 Python 并详细介绍其内置的各类类。

中间件可使用的钩子

该循环在每个阶段都会提供相应的钩子，中间件可以绑定到一个或多个钩子上：

before_agent 和 after_agent 会在每次调用的最开始和最结束时各执行一次。
before_model 和 after_model 会在循环即将调用模型或刚刚调用完模型时触发。
wrap_model_call 和 wrap_tool_call 用于封装调用过程，从而实现重试、替换或从缓存中获取数据的功能。

这就是完整的思维模型。你可以通过向 create_agent 传递列表来添加中间件，就像这个示例中结合了邮件内容屏蔽、通话时长限制以及工具重试功能：

agent = create_agent(
    model=model,
    tools=[my_tool],
    middleware=[
        PIIMiddleware("email", strategy="redact"),
        ModelCallLimitMiddleware(run_limit=5),
        ToolRetryMiddleware(max_retries=3),
    ],
)

列表的顺序很重要。中间件会按顺序应用，就像包裹在代理周围的洋葱层一样，因此首先列出的屏蔽步骤会在其他任何操作之前处理原始输入。

设置与基础代理

中间件需要 LangChain 1.0 或更高版本，因此请使用 -U 标志进行安装以升级旧版本。示例中使用的是 Gemini 的免费套餐；你可以在 Google AI Studio 中创建密钥。

!pip install -qU langchain langchain-google-genai

导入 os 以及 Gemini 聊天模型类：

import os
from langchain_google_genai import ChatGoogleGenerativeAI

接着配置模型。此代码片段仅出于演示目的而直接设置密钥；在实际代码中，应从环境变量中导出 GOOGLE_API_KEY 或从密钥管理器中加载该密钥，而非将其写在源代码中。将温度值设为零可确保输出结果具有重复性：

os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)

每个实验的主体都是一个不包含中间件的最小化智能体。它需要 create_agent 函数以及 tool 装饰器：

from langchain.agents import create_agent
from langchain_core.tools import tool

该智能体包含一个模拟的天气工具，始终会报告晴天状况，并且仅通过一条用户消息即可触发该工具的运行：

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)

下面的每个部分都会为这个智能体添加另一个检查点。

1. 用于限制内存的摘要中间件

在长时间的对话中，消息历史会不断累积，直至超出上下文窗口大小，每多一个标记都会被计费。SummarizationMiddleware会在before_model阶段检查历史记录的大小；一旦超过阈值，它就会将较旧的消息汇总为摘要，仅保留最新的消息原文。

from langchain.agents.middleware import SummarizationMiddleware

下面的配置使用相同的模型来生成摘要，当消息数量达到10条时触发汇总操作，并保留最新的4条消息不变。测试会构建一个包含六个城市（共十二条消息）的虚假历史记录，然后询问哪个城市最先出现：

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        SummarizationMiddleware(
            model=model,               # which LLM writes the summary
            trigger=("messages", 10),  # summarize when history hits 10 messages
            keep=("messages", 4),      # keep the 4 most recent messages intact
        ),
    ],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
    long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
    long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history))   # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"]))   # 6

有十三条消息进入，之后剩下六条：摘要、四条保留的消息以及新的回复。由于该事实被包含在摘要中，模型仍然会回答“德里”。在演示中通过统计消息数量可以直观了解这种行为，但在实际生产环境中，使用基于令牌的触发条件如(“tokens”, 3000)或上下文窗口的比例值如(“fraction”, 0.8)来监控实际成本并设置更合理的限制会更为有效。需注意，摘要生成过程存在信息丢失风险：对话初期提到的精确数字或标识符可能无法保留。

2. 将调用限制作为成本断路器

最昂贵的代理故障类型是无限循环，即模型与工具不断互相调用，持续耗费资源数分钟才有人察觉。有两种中间件可以对此设置严格限制：

from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware

此处模型每轮最多调用三次，且设置了exit_behavior="end"，这样智能体会正常停止而不会抛出异常；工具则最多调用两次。提示语特意要求逐一查询六个城市：

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        # "end" = stop gracefully instead of raising an error
        ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
        ToolCallLimitMiddleware(run_limit=2),
    ],
)result = agent.invoke(
    {"messages": [{"role": "user", "content":
        "Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)

智能体需要执行六次查询，达到上限后便用已有的部分结果优雅地完成任务。此外还设有thread_limit参数，可用于限制整个对话线程中的调用次数，而不仅仅是单次运行。这两项设置属于简单的预防措施；应将限制值设定得高于合法请求的实际需求，这样只有在真正出现异常情况时才会触发限制。

3. 用于处理不稳定依赖的ToolRetryMiddleware

实际工具会出现故障：HTTP 请求超时且连接中断。ToolRetryMiddleware通过wrap_tool_call捕获这些故障，并以指数退避策略进行重试。

from langchain.agents.middleware import ToolRetryMiddleware

为演示这一功能，一个股票价格查询工具会记录尝试次数，在前两次失败后抛出ConnectionError，之后才会成功。该中间件允许最多进行三次重试，首次延迟为一秒，此后每次延迟时间翻倍：

attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
    """Get the current stock price for a ticker symbol."""
    attempt_counter["count"] += 1
    print(f"  [tool called — attempt #{attempt_counter['count']}]")
    if attempt_counter["count"] < 3:
        raise ConnectionError("API timeout — please retry")
    return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
    model=model,
    tools=[flaky_stock_price],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,       # retry a failed tool up to 3 times
            initial_delay=1.0,   # wait 1s before first retry
            backoff_factor=2.0,  # double the wait each time: 1s, 2s, 4s
        ),
    ],
)result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)

该工具两次失败后，中间件会等待并再次尝试，第三次尝试最终成功。从代理的角度来看并未出现问题。重试仅适用于读取等幂等操作；对于会扣款或发送消息的工具，重试可能会重复产生副作用。

4. 用于服务提供商故障的ModelFallbackMiddleware

同样的弹性机制也能保护模型调用。当主模型因速率限制或故障等原因失效时，该中间件会按照您指定的顺序，对备用模型重新发起请求：

from langchain.agents.middleware import ModelFallbackMiddleware

示例中将较轻量的 Gemini 模型设为主模型，并添加另一个 Gemini 模型作为备用：

backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
    model=model,  # primary: gemini-3.5-flash-lite
    tools=[get_weather],
    middleware=[ModelFallbackMiddleware(backup_model)],
)

只要主模型正常运行，您就不会察觉到任何异常，这正是设计初衷。只有在服务提供商出现问题的那天，才会显示出其作用。为了获得更强的保护机制，建议选择其他提供商的备用模型，因为同一 API 后端的所有模型往往都会受到故障影响。

5. 用于敏感数据的 PIIMiddleware

通常情况下，你根本不希望将电子邮件地址、卡号或 IP 地址发送给模型提供方。PIIMiddleware会在模型看到相关内容之前，在before_model阶段扫描文本，并采用四种策略中的一种：redact、mask、hash或block。

from langchain.agents.middleware import PIIMiddleware

该处理工具没有其他功能，它会将电子邮件地址完全替换为占位符，同时仅保留卡号的最后四位数字来遮掩其余部分，这两种规则都会应用于用户输入的内容：

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        # Replace emails entirely with [REDACTED_EMAIL]
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # Mask credit cards — keeps last 4 digits
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
    ],
)result = agent.invoke(
    {"messages": [{"role": "user", "content":
        "Draft a support reply to priya.sharma@example.com confirming her card "
        "4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)

模型在生成回复时根本不会收到真实的地址或完整的卡号。你还可以注册自定义的 PII 类型并使用自己的正则表达式，例如用来屏蔽任何看起来像内部 API 密钥的文本：

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

基于模式的检测能够识别格式正确的值，但无法捕捉所有创新的拼写方式，因此应将其视为防御层而非合规性保障。

6. 用于审批环节的HumanInTheLoopMiddleware

某些操作后果严重，无法完全由系统自主处理：发送邮件、删除记录或进行支付等。HumanInTheLoopMiddleware会在敏感工具执行前立即暂停代理，等待人工决策后再继续运行。

其工作方式与前述中间件不同。被暂停的代理不会被终止，检查点会保存其完整状态，而线程ID则作为日后查找并恢复该进程的依据。审核人员可以批准请求、修改其参数或予以拒绝。

这些导入项包含了中间件、来自 LangGraph 的内存检查点器，以及用于恢复运行的 Command 类型：

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

下面的智能体拥有一个 send_email 工具，它会将该工具标记为可中断，并将暂停状态存储在 InMemorySaver 中。在 demo-1 线程上的首次调用会要求该智能体给经理发送邮件：

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the given recipient."""
    return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
    model=model,
    tools=[send_email],
    middleware=[
        # Pause and ask a human whenever the agent wants to call send_email
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
    ],
    checkpointer=InMemorySaver(),   # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
    {"messages": [{"role": "user", "content":
        "Send an email to boss@company.com saying the report is ready."}]},
    config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])

工具尚未运行时执行就中止了。结果中的 __interrupt__ 条目显示了待处理的调用信息，包括接收者、主题和内容，但邮件并未实际发送。随后在同一线程上通过 Command 发送批准指令，从而恢复了暂停的运行状态：

# Step 2: approve and resume
result = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config,  # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)

除了使用 approve，您还可以发送带有原因的 reject 或者携带修改后参数的 edit。在真实应用中，此处正是渲染审批界面的地方。有两点需要注意：InMemorySaver在进程重启时会丢失状态，因此生产环境需要使用持久化的检查点；另外，不同LangChain版本之间的恢复数据格式有所变化，需在您所使用的版本的中间件参考文档中确认相关格式。

7. 使用装饰器编写自定义中间件

当内置功能无法满足需求时，自定义中间件其实很简单，因为每个钩子都有对应的装饰器。所需导入的元素为 before_model 装饰器和 AgentState 类型：

from langchain.agents.middleware import before_model, AgentState

此示例会记录每次模型调用时即将发送的消息数量，并像其他预构建的中间件一样被添加到列表中：

@before_model
def log_before_model(state: AgentState, runtime) -> None:
    print(f"  [middleware] Calling model with {len(state['messages'])} messages")
    # Returning None = observe only.
    # Returning a dict would UPDATE the agent's state (e.g., trim messages)
    return Noneagent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[log_before_model],  # plugs in like any prebuilt middleware
)

返回值是重要的设计考量。返回 None 表示该中间件仅负责观察；而返回字典则可以更新智能体状态，这样就可以筛选消息、注入上下文或实施自定义约束。对于其他钩子函数也有相应的装饰器：@before_agent、@after_model、@wrap_model_call、@wrap_tool_call，还有用于在运行时构建系统提示语的 @dynamic_prompt。

关键要点

中间件将操作层面的问题与智能体逻辑分开：循环结构保持不变，各个检查点则作为普通列表层层叠加，最终传递给 create_agent 函数。

请有意识地安排列表顺序；在将文本转发到其他地方之前应先进行内容截取处理。

摘要生成与调用次数限制用于控制成本，工具重试机制与模型回退策略用于保障可靠性，个人身份信息处理措施用于防止数据泄露，而人工干预机制则用于控制那些不可逆的操作。

每种方法都有需要牢记的局限性：摘要会丢失细节，对于非幂等工具而言重试并不安全，正则表达式检测存在缺陷，且内存中的检查点在重启后会消失。

包括TodoListMiddleware、LLMToolSelectorMiddleware和ContextEditingMiddleware在内的更多中间件，其详细说明可见于官方参考文档。