三种防止智能体泄露机密的LangChain中间件防护机制

将敏感信息粘贴到聊天窗口中既容易操作，也容易被遗忘。也许有人为了调试把文件传入了ChatGPT，发送后才发现问题，开始担心其中是否包含API密钥、数据库密码或客户端令牌。于是重新阅读消息，关闭标签页，觉得应该没什么问题——大概也不算违反政策。

智能体让情况更糟，因为它会在每次会话中自动多次执行类似操作，且不会告知用户。只要给它访问代码库的权限用于审查，它最终就会打开一个包含类似OPENAI_API_KEY=sk-proj-...内容的.env文件。由于没有防护措施，这些密钥会原封不动地传送到语言模型中，出现在追踪日志里，并保存在服务提供商的记录中。至于这些密钥是否会进一步流入微调数据集或日志服务，用户根本无从知晓。

几轮操作后，同一个智能体调用了清理工具，删除了一个你从未提及的目录，悄无声息地记录下了delete_directory(path="./tmp_backups")，然后继续执行后续操作。当你发现需要的文件不见了时才察觉到这一问题。

这其实是两起独立的故障，既不存在漏洞，也没有出现严重的错误幻觉。该智能体只是没有明确区分自己能够做什么与应该做什么。解决方法是使用中间件，Cursor和GitHub Copilot等工具中的保护机制也是基于同样的原理。以下是将其整合到你自己开发的智能体中的方法。

中间件的作用

可以把智能体想象成一条生产线。数据从您那里、文件以及搜索结果中进入，随后传递给模型，模型做出决策，这些决策再转化为具体行动。中间件则是这条生产线上的检查点——它拦截数据，对其进行检测或转换，然后要么原样传递，要么对其修改，亦或是阻止其继续前进。

时机至关重要。如果过晚，敏感数据早已越过了安全边界；如果过早，则可能会检查到错误的内容。

Copilot的秘密扫描功能位于输出阶段。在建议出现在编辑器中之前，模式匹配器会对其进行检查，凡是以sk-、ghp_或AKIA开头的键形字符串都会被过滤掉。Cursor的“审查更改”功能则在任何文件写入操作之前启动：程序会完全暂停，显示差异对比，直到你批准后才能继续。如果你拒绝，写入操作就不会发生，程序会寻找其他方式。

一个功能负责防止模型输出敏感信息，另一个则防止模型执行某些操作。LangChain允许你同时实现这两种功能，还能设置资源限制。

捕获敏感信息：PIIMiddleware

你的智能体从不直接操作文件系统。它只能看到显示在其上下文窗口中的内容，也就是你输入的文字以及其工具返回的结果。因此，敏感数据只有两种进入方式：要么由你手动粘贴，要么通过工具从磁盘读取，比如目录扫描时会打开 .env 文件。无论哪种方式，在模型看到任何内容之前都会先经过中间件层，而 PIIMiddleware 正是在这个关键节点发挥作用的。

有一个名为 apply_to_tool_results 的标志，大多数教程都会使用它却从不解释。它的默认值为 False。只需将 apply_to_input=True 设为开启状态，就能处理你粘贴敏感数据的情况，但智能体读取文件的情形依然存在风险。具备文件访问权限的智能体会直接将整个 .env 文件作为工具结果获取，进而传递给模型。

最简的正确配置如下：

from langchain.agents.middleware import PIIMiddleware

PIIMiddleware(
    "email",
    strategy="redact",            # replaces match with [REDACTED_EMAIL]
    apply_to_input=True,          # scans what you type
    apply_to_tool_results=True,   # scans what tools return - never skip this
)

API密钥才是导致实际问题的根源，而没有任何股票检测工具知道你的密钥格式，因此你需要提供正则表达式。这个正则可以覆盖常见的前缀：

import re
API_KEY_PATTERN = r"(?:sk-|ghp_|AKIA)[a-zA-Z0-9]{20,48}"
# sk-   → OpenAI and Anthropic keys
# ghp_  → GitHub personal access tokens
# AKIA  → AWS access key IDs
PIIMiddleware(
    "api_key",
    detector=API_KEY_PATTERN,
    strategy="redact",
    apply_to_input=True,
    apply_to_tool_results=True,
)

该中间件提供了四种处理策略：

• redact 会将匹配内容替换为 [REDACTED_FIELDNAME]，适用于日志、追踪数据以及单向数据流。
• mask 用星号替换字符但保持长度不变，适合在需要表明某个值存在而不暴露其具体内容时使用。
• block 会抛出异常并终止整个处理过程，适用于零容忍的高安全级别环境。
• hash 会用确定的哈希值替代原内容，适合在无需存储实体值的情况下追踪该实体时使用。

大多数人认为block是最安全的选择，但对于工具处理结果而言，它往往并非正确选项。假设智能体在扫描过程中读取了30个文件，其中有一个文件的注释中包含电子邮件地址。使用block会直接终止整个会话。redact则会删除该敏感信息并让工作继续，这几乎总是人们希望看到的结果。只有当敏感数据的存在本身就意味着智能体进入了不应进入的区域时，才应使用block。

自我测试一下：如果设置了apply_to_input=True而apply_to_tool_results保持默认值，那么哪种具体情况会得不到保护？你应该能够用一句话回答。

阻止隐式删除：HumanInTheLoopMiddleware

返回被删除的目录。HumanInTheLoopMiddleware其实是Cursor中“审核更改”机制转化为的代码实现。在任何具有破坏性或修改性的工具运行之前，该代理会暂停并保存当前状态。这样你就能清楚地看到它打算做什么，必须在任何操作执行前给予批准。

暂停相对容易，而恢复则较为困难。即便在数小时或数天后、进程重新启动之后，要让代理恢复到之前的确切状态，就必须将对话状态、待处理的操作以及代理的推理进度存储在某种持久化的介质中。这种存储机制被称为检查点。如果没有检查点，一旦出现中断，状态就会丢失，下一次调用将从零开始，因此审批机制只会终止当前会话。

暂停发生在模型决定要执行什么操作之后、任何工具开始运行之前。这是唯一可行的时机，因为此时决策仍是可见的，且尚未发生任何不可逆转的操作。

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.sqlite import SqliteSaver

# The checkpointer is not optional. Without it, resume is impossible.
with SqliteSaver.from_conn_string("./review_agent.db") as checkpointer:
    agent = create_agent(
        model="anthropic:claude-sonnet-4-20250514",
        tools=[read_file, list_directory, write_file, search_codebase],
        middleware=[
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "write_file": True,       # always pause before writing
                    "read_file": False,        # reading is safe - no pause needed
                    "list_directory": False,
                    "search_codebase": False,
                }
            ),
        ],
        checkpointer=checkpointer,            # saved to SQLite, persists across restarts
    )

获得批准后的继续操作如下：

# First call — agent hits the interrupt at write_file and pauses
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Review and fix the config files"}]},
    config={"configurable": {"thread_id": "session-001"}}
    # thread_id ties the saved state to this specific session
)


# result.interrupted == True
# result.pending_tool_call == {"name": "write_file", "args": {"path": "src/config.py", ...}}
# You show this to the user and wait for approval
# User approves - resume the same thread
final_result = agent.invoke(
    Command(resume=True),
    config={"configurable": {"thread_id": "session-001"}}
    # same thread_id - loads state from the checkpointer and picks up where it stopped
)

一个小问题就能耗费整个下午的时间。interrupt_on={"write_file": True}会以精确字符串的形式匹配工具名称。如果你的工具实际名为file_write或update_file，则中断不会触发，调用会悄无声息地通过。当工具被调用时，请查看跟踪日志以获取真实名称，并据此进行匹配。

两种故障模式下的两项限制

每个步骤中，智能体都会调用模型，模型会选择某个动作，有时还会调用工具，而且一次模型调用可以同时启动多次工具调用。这些都是独立的资源，彼此之间不会相互计算。

ModelCallLimitMiddleware用于统计模型调用次数，而ToolCallLimitMiddleware则用于统计工具执行次数。10次模型调用若每次都会触发3次并行工具调用，那么总执行次数为30次，仅通过模型调用次数的限制是无法察觉到这一点的。反之亦然：如果智能体一直进行自我推理而不使用任何工具，那么工具调用计数器将始终保持在零，不会触发限制。

from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware

ModelCallLimitMiddleware(
    max_calls=30,
    on_limit="raise",   # raises MaxCallsExceeded - catch this in your application
)
ToolCallLimitMiddleware(
    max_calls=60,
    on_limit="raise",
)

两者都应使用，因为每一种都能覆盖另一种无法处理的情况。

关于 on_limit="raise" 与 on_limit="return" 的选择：在需要部分结果的交互式会话中，比如展示目前已审核内容的摘要时，应选择 "return"；而当未完成的输出会破坏后续流程时，则应选择 "raise"。

数值设置也需要谨慎。max_calls=10 这一限制在快速测试时看似安全，但具有多步骤执行计划的智能体在第一步就可能消耗8次调用机会。此时限制机制会迫使它在任务进行到一半时停止，返回不完整的结果，而你可能要花费二十分钟才能从日志中理解问题所在。解决之道并非降低限制，而是根据任务实际需求来设定合适的数值。

顺序是安全保障

中间件会按照你列出的顺序运行，这绝非无关紧要的细节。如果任何用于重写或概括消息内容的中间件位于 PIIMiddleware 之前，隐私信息检测就会在已经被修改过的文本上进行扫描。当内容经过压缩、改写或重新结构化后导致原有匹配模式不再适用时，删除功能就会漏掉这些信息，因为敏感数据早已存在于模型的上下文中，而删除操作是在损害发生之后才执行的。

对于此处的安全架构，中间件的顺序为：

middleware=[
    # PIIMiddleware always first — it must see raw, untransformed data
    PIIMiddleware("api_key", detector=API_KEY_PATTERN, strategy="redact",
                  apply_to_input=True, apply_to_tool_results=True),
    PIIMiddleware("email", strategy="redact",
                  apply_to_input=True, apply_to_tool_results=True),

# Limits next - exact position within the group is flexible
    ModelCallLimitMiddleware(max_calls=30, on_limit="raise"),
    ToolCallLimitMiddleware(max_calls=60, on_limit="raise"),
    # Human-in-the-loop last in the safety group
    # (it fires in the after_model hook regardless of list position,
    # but last is a readable convention)
    HumanInTheLoopMiddleware(interrupt_on={"write_file": True}),
]

应将此列表视为流水线而非插件注册表。第四个元素不会收到原始消息，而只会收到前三个元素处理后留下的内容。若将PIIMiddleware放在任何处理消息内容的组件之后，这一保障就会悄然消失。

再做一次验证：如果PIIMiddleware位于负责汇总对话历史的中间件之后，会出现什么问题？如果你无需向上滚动就能回答，那就说明排序规则已确定。

构建代码审查代理

现在将所有组件整合成一个行为类似商业工具的代理。它包含四个工具：read_file、list_directory、write_file和search_codebase。另有三层安全机制对其进行保护：敏感信息永远不会传递给模型，写入操作需要你的批准，且资源使用有严格上限。

from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware,
    HumanInTheLoopMiddleware,
    ModelCallLimitMiddleware,
    ToolCallLimitMiddleware,
)
from langgraph.checkpoint.sqlite import SqliteSaver

# Custom pattern for secrets common in codebases
API_KEY_PATTERN = r"(?:sk-|ghp_|AKIA)[a-zA-Z0-9]{20,48}"
# sk-   → OpenAI / Anthropic keys
# ghp_  → GitHub personal access tokens
# AKIA  → AWS access key IDs
with SqliteSaver.from_conn_string("./review_agent.db") as checkpointer:
    agent = create_agent(
        model="anthropic:claude-sonnet-4-20250514",
        tools=[read_file, list_directory, write_file, search_codebase],
        middleware=[
            # 1. Catch secrets before they reach the model - on both surfaces
            PIIMiddleware(
                "api_key",
                detector=API_KEY_PATTERN,
                strategy="redact",
                apply_to_input=True,
                apply_to_tool_results=True,  # this is the one that catches .env reads
            ),
            PIIMiddleware(
                "email",
                strategy="redact",
                apply_to_input=True,
                apply_to_tool_results=True,
            ),
            # 2. Hard resource limits - stops infinite loops and runaway sessions
            ModelCallLimitMiddleware(max_calls=30, on_limit="raise"),
            ToolCallLimitMiddleware(max_calls=60, on_limit="raise"),
            # 3. Approval gate - nothing gets written without your explicit sign-off
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "write_file": True,
                    "read_file": False,
                    "list_directory": False,
                    "search_codebase": False,
                }
            ),
        ],
        checkpointer=checkpointer,  # required for interrupt/resume to work
    )

当代理想要写入文件时，流程如下：

Agent: I'd like to update src/config.py to fix the circular import.
       Here is what I plan to write:
--- src/config.py ---
       from typing import Optional
       from pydantic import BaseModel
       class Settings(BaseModel):
           debug: bool = False
           database_url: str = "sqlite:///app.db"
       ...
       Approve this change? (yes/no)
User: yes
Agent: Written. Moving to tests/test_config.py next.

The agent cannot touch anything silently. Every write surfaces for review before it happens. Secrets in any file the agent reads are replaced with [REDACTED_API_KEY] before reaching the model — which also means they never appear in the approval prompt you read. What you are reviewing is already clean.

有两个简单的扩展值得了解。如果代码库中存储了用户数据，比如测试用例中的电话号码，应在处理电子邮件相关代码之前添加 PIIMiddleware("phone", strategy="redact") 这一行代码。所有的个人身份信息检测器都会对相同的输入独立运行，因此它们在组内的顺序并不重要。另外，如果您希望以平滑的方式降级而非突然停止运行，可以将两个限制处理中间件中的 on_limit="raise" 更改为 on_limit="return"，这样代理就会返回它已处理出的部分结果。

现在，该智能体无法向模型泄露秘密，未经批准不得修改文件，也不能超出用户设定的资源上限运行。仍存在一个限制：它无法在长时间的会话中保持一致性。若需要审查数百个文件并使用数十次工具调用，上下文窗口会在任务完成前就被填满，从而导致智能体忘记之前处理的内容，其建议也会变得偏离主题。这与此处解决的“控制智能体可操作范围”的问题属于不同的问题。