外观
护栏(Guardrails)通过在智能体执行的关键节点对内容进行校验和过滤,帮助你构建安全、合规的 AI 应用。它们可以检测敏感信息、强制执行内容策略、验证输出,并在不安全行为造成问题之前加以阻止。
常见用例包括:
- 防止 PII 泄露
- 检测并阻止提示词注入攻击
- 阻止不当或有害内容
- 强制执行业务规则与合规要求
- 验证输出质量和准确性
你可以使用中间件实现护栏,在关键节点拦截执行——在智能体开始之前、在它完成之后,或在模型调用和工具调用周围。

护栏可以通过两种互补的方法实现:
- 确定性护栏 — 使用基于规则的逻辑,如正则表达式模式、关键字匹配或显式检查。快速、可预测且成本低,但可能漏掉微妙的违规行为。
- 基于模型的护栏 — 使用 LLM 或分类器以语义理解来评估内容。能捕获规则遗漏的细微问题,但速度更慢、成本更高。
LangChain 既提供内置护栏(例如 PII 检测、人在回路),也提供灵活的中间件系统,可用于采用上述任何一种方法构建自定义护栏。
内置护栏
PII 检测
LangChain 提供了内置中间件,用于检测和处理对话中的个人身份信息(PII)。该中间件可以检测常见的 PII 类型,如电子邮件、信用卡、IP 地址等。
PII 检测中间件对于以下场景很有帮助:有合规要求的医疗保健和金融应用、需要清理日志的客服智能体,以及任何处理敏感用户数据的应用。
PII 中间件支持多种处理已检测 PII 的策略:
| 策略 | 描述 | 示例 |
|---|---|---|
redact | 替换为 [REDACTED_{PII_TYPE}] | [REDACTED_EMAIL] |
mask | 部分遮盖(例如后 4 位) | ****-****-****-1234 |
hash | 替换为确定性哈希 | a8f5f167... |
block | 检测到时抛出异常 | 抛出错误 |
INFO
使用 apply_to_output=True 时,PIIMiddleware 还会通过注册的流式转换器(stream transformer)对流式传输的线上输出进行脱敏——包括文本增量、工具调用参数、工具输出和状态快照。需要 langchain>=1.3.2。请参阅在中间件上注册转换器。
python
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
agent = create_agent(
model="gpt-5.5",
tools=[customer_service_tool, email_tool],
middleware=[
# 在发送给模型之前脱敏用户输入中的电子邮件
PIIMiddleware(
"email",
strategy="redact",
apply_to_input=True,
),
# 对用户输入中的信用卡进行遮罩
PIIMiddleware(
"credit_card",
strategy="mask",
apply_to_input=True,
),
# 阻止 API 密钥 - 检测到时抛出错误
PIIMiddleware(
"api_key",
detector=r"sk-[a-zA-Z0-9]{32}",
strategy="block",
apply_to_input=True,
),
],
)
# 当用户提供 PII 时,将按策略处理
result = agent.invoke({
"messages": [{"role": "user", "content": "My email is john.doe@example.com and card is 5105-1051-0510-5100"}]
})typescript
import { createAgent, piiRedactionMiddleware } from "langchain";
const agent = createAgent({
model: "gpt-5.5",
tools: [customerServiceTool, emailTool],
middleware: [
// 在发送给模型之前脱敏用户输入中的电子邮件
piiRedactionMiddleware({
piiType: "email",
strategy: "redact",
applyToInput: true,
}),
// 对用户输入中的信用卡进行遮罩
piiRedactionMiddleware({
piiType: "credit_card",
strategy: "mask",
applyToInput: true,
}),
// 阻止 API 密钥 - 检测到时抛出错误
piiRedactionMiddleware({
piiType: "api_key",
detector: /sk-[a-zA-Z0-9]{32}/,
strategy: "block",
applyToInput: true,
}),
],
});
// 当用户提供 PII 时,将按策略处理
const result = await agent.invoke({
messages: [{
role: "user",
content: "My email is john.doe@example.com and card is 5105-1051-0510-5100"
}]
});内置 PII 类型与配置
内置 PII 类型:
email- 电子邮件地址credit_card- 信用卡号(Luhn 校验)ip- IP 地址mac_address- MAC 地址url- URL
配置选项:
| 参数 | 描述 | 默认值 |
|---|---|---|
pii_type | 要检测的 PII 类型(内置或自定义) | 必填 |
strategy | 如何处理检测到的 PII("block"、"redact"、"mask"、"hash") | "redact" |
detector | 自定义检测函数或正则表达式模式 | None(使用内置) |
apply_to_input | 在模型调用前检查用户消息 | True |
apply_to_output | 在模型调用后检查 AI 消息 | False |
apply_to_tool_results | 在执行后检查工具结果消息 | False |
| 参数 | 描述 | 默认值 |
|---|---|---|
piiType | 要检测的 PII 类型(内置或自定义) | 必填 |
strategy | 如何处理检测到的 PII("block"、"redact"、"mask"、"hash") | "redact" |
detector | 自定义检测器正则表达式模式 | undefined(使用内置) |
applyToInput | 在模型调用前检查用户消息 | true |
applyToOutput | 在模型调用后检查 AI 消息 | false |
applyToToolResults | 在执行后检查工具结果消息 | false |
有关 PII 检测能力的完整详细信息,请参阅中间件文档。
人在回路(Human-in-the-loop)
LangChain 提供了内置中间件,用于在执行敏感操作前要求人工批准。这是高风险决策最有效的护栏之一。
人在回路中间件对于以下场景很有帮助:金融交易和转账、删除或修改生产数据、向外部各方发送通信,以及任何具有重大业务影响的操作。
python
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, send_email_tool, delete_database_tool],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
# 敏感操作需要批准
"send_email": True,
"delete_database": True,
# 自动批准安全操作
"search": False,
}
),
],
# 在中断期间持久化状态
checkpointer=InMemorySaver(),
)
# 人在回路需要线程 ID 进行持久化
config = {"configurable": {"thread_id": "some_id"}}
# 智能体在执行敏感工具之前将暂停并等待批准
result = agent.invoke(
{"messages": [{"role": "user", "content": "Send an email to the team"}]},
config=config
)
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config=config # 使用相同的线程 ID 恢复已暂停的对话
)typescript
import { createAgent, humanInTheLoopMiddleware } from "langchain";
import { MemorySaver, Command } from "@langchain/langgraph";
const agent = createAgent({
model: "gpt-5.5",
tools: [searchTool, sendEmailTool, deleteDatabaseTool],
middleware: [
humanInTheLoopMiddleware({
interruptOn: {
// 敏感操作需要批准
send_email: { allowAccept: true, allowEdit: true, allowRespond: true },
delete_database: { allowAccept: true, allowEdit: true, allowRespond: true },
// 自动批准安全操作
search: false,
}
}),
],
checkpointer: new MemorySaver(),
});
// 人在回路需要线程 ID 进行持久化
const config = { configurable: { thread_id: "some_id" } };
// 智能体在执行敏感工具之前将暂停并等待批准
let result = await agent.invoke(
{ messages: [{ role: "user", content: "Send an email to the team" }] },
config
);
result = await agent.invoke(
new Command({ resume: { decisions: [{ type: "approve" }] } }),
config // 使用相同的线程 ID 恢复已暂停的对话
);TIP
有关实现批准工作流的完整详细信息,请参阅人在回路文档。
自定义护栏
对于更复杂的护栏,你可以创建在智能体执行之前或之后运行的自定义中间件。这让你完全掌控校验逻辑、内容过滤和安全检查。
智能体前护栏
使用"before agent(智能体前)"钩子,在每次调用的开始一次性校验请求。这对于会话级检查很有用,例如身份验证、速率限制,或在任何处理开始之前阻止不当请求。
python
from typing import Any
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime
class ContentFilterMiddleware(AgentMiddleware):
"""Deterministic guardrail: Block requests containing banned keywords."""
def __init__(self, banned_keywords: list[str]):
super().__init__()
self.banned_keywords = [kw.lower() for kw in banned_keywords]
@hook_config(can_jump_to=["end"])
def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
# 获取第一条用户消息
if not state["messages"]:
return None
first_message = state["messages"][0]
if first_message.type != "human":
return None
content = first_message.content.lower()
# 检查禁用关键字
for keyword in self.banned_keywords:
if keyword in content:
# 在任何处理之前阻止执行
return {
"messages": [{
"role": "assistant",
"content": "I cannot process requests containing inappropriate content. Please rephrase your request."
}],
"jump_to": "end"
}
return None
# 使用自定义护栏
from langchain.agents import create_agent
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, calculator_tool],
middleware=[
ContentFilterMiddleware(
banned_keywords=["hack", "exploit", "malware"]
),
],
)
# 该请求将在任何处理之前被阻止
result = agent.invoke({
"messages": [{"role": "user", "content": "How do I hack into a database?"}]
})python
from typing import Any
from langchain.agents.middleware import before_agent, AgentState, hook_config
from langgraph.runtime import Runtime
banned_keywords = ["hack", "exploit", "malware"]
@before_agent(can_jump_to=["end"])
def content_filter(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Deterministic guardrail: Block requests containing banned keywords."""
# 获取第一条用户消息
if not state["messages"]:
return None
first_message = state["messages"][0]
if first_message.type != "human":
return None
content = first_message.content.lower()
# 检查禁用关键字
for keyword in banned_keywords:
if keyword in content:
# 在任何处理之前阻止执行
return {
"messages": [{
"role": "assistant",
"content": "I cannot process requests containing inappropriate content. Please rephrase your request."
}],
"jump_to": "end"
}
return None
# 使用自定义护栏
from langchain.agents import create_agent
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, calculator_tool],
middleware=[content_filter],
)
# 该请求将在任何处理之前被阻止
result = agent.invoke({
"messages": [{"role": "user", "content": "How do I hack into a database?"}]
})typescript
import { createMiddleware, AIMessage } from "langchain";
const contentFilterMiddleware = (bannedKeywords: string[]) => {
const keywords = bannedKeywords.map(kw => kw.toLowerCase());
return createMiddleware({
name: "ContentFilterMiddleware",
beforeAgent: {
hook: (state) => {
// 获取第一条用户消息
if (!state.messages || state.messages.length === 0) {
return;
}
const firstMessage = state.messages[0];
if (firstMessage._getType() !== "human") {
return;
}
const content = firstMessage.content.toString().toLowerCase();
// 检查禁用关键字
for (const keyword of keywords) {
if (content.includes(keyword)) {
// 在任何处理之前阻止执行
return {
messages: [
new AIMessage(
"I cannot process requests containing inappropriate content. Please rephrase your request."
)
],
jumpTo: "end",
};
}
}
return;
},
canJumpTo: ['end']
}
});
};
// 使用自定义护栏
import { createAgent } from "langchain";
const agent = createAgent({
model: "gpt-5.5",
tools: [searchTool, calculatorTool],
middleware: [
contentFilterMiddleware(["hack", "exploit", "malware"]),
],
});
// 该请求将在任何处理之前被阻止
const result = await agent.invoke({
messages: [{ role: "user", content: "How do I hack into a database?" }]
});智能体后护栏
使用"after agent(智能体后)"钩子在返回给用户之前对最终输出进行一次校验。这对于基于模型的安全检查、质量验证,或对完整智能体响应进行最终合规扫描非常有用。
python
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime
from langchain.messages import AIMessage
from langchain.chat_models import init_chat_model
from typing import Any
class SafetyGuardrailMiddleware(AgentMiddleware):
"""Model-based guardrail: Use an LLM to evaluate response safety."""
def __init__(self):
super().__init__()
self.safety_model = init_chat_model("gpt-5.4-mini")
@hook_config(can_jump_to=["end"])
def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
# 获取最终的 AI 响应
if not state["messages"]:
return None
last_message = state["messages"][-1]
if not isinstance(last_message, AIMessage):
return None
# 使用模型评估安全性
safety_prompt = f"""Evaluate if this response is safe and appropriate.
Respond with only 'SAFE' or 'UNSAFE'.
Response: {last_message.content}"""
result = self.safety_model.invoke([{"role": "user", "content": safety_prompt}])
if "UNSAFE" in result.content:
last_message.content = "I cannot provide that response. Please rephrase your request."
return None
# 使用安全护栏
from langchain.agents import create_agent
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, calculator_tool],
middleware=[SafetyGuardrailMiddleware()],
)
result = agent.invoke({
"messages": [{"role": "user", "content": "How do I make explosives?"}]
})python
from langchain.agents.middleware import after_agent, AgentState, hook_config
from langgraph.runtime import Runtime
from langchain.messages import AIMessage
from langchain.chat_models import init_chat_model
from typing import Any
safety_model = init_chat_model("gpt-5.4-mini")
@after_agent(can_jump_to=["end"])
def safety_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Model-based guardrail: Use an LLM to evaluate response safety."""
# 获取最终的 AI 响应
if not state["messages"]:
return None
last_message = state["messages"][-1]
if not isinstance(last_message, AIMessage):
return None
# 使用模型评估安全性
safety_prompt = f"""Evaluate if this response is safe and appropriate.
Respond with only 'SAFE' or 'UNSAFE'.
Response: {last_message.content}"""
result = safety_model.invoke([{"role": "user", "content": safety_prompt}])
if "UNSAFE" in result.content:
last_message.content = "I cannot provide that response. Please rephrase your request."
return None
# 使用安全护栏
from langchain.agents import create_agent
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, calculator_tool],
middleware=[safety_guardrail],
)
result = agent.invoke({
"messages": [{"role": "user", "content": "How do I make explosives?"}]
})typescript
import { createMiddleware, AIMessage, initChatModel } from "langchain";
const safetyGuardrailMiddleware = () => {
const safetyModel = initChatModel("gpt-5.4-mini");
return createMiddleware({
name: "SafetyGuardrailMiddleware",
afterAgent: {
hook: async (state) => {
// 获取最终的 AI 响应
if (!state.messages || state.messages.length === 0) {
return;
}
const lastMessage = state.messages[state.messages.length - 1];
if (lastMessage._getType() !== "ai") {
return;
}
// 使用模型评估安全性
const safetyPrompt = `Evaluate if this response is safe and appropriate.
Respond with only 'SAFE' or 'UNSAFE'.
Response: ${lastMessage.content.toString()}`;
const result = await safetyModel.invoke([
{ role: "user", content: safetyPrompt }
]);
if (result.content.toString().includes("UNSAFE")) {
return {
messages: [
new AIMessage(
"I cannot provide that response. Please rephrase your request."
)
],
jumpTo: "end",
};
}
return;
},
canJumpTo: ['end']
}
});
};
// 使用安全护栏
import { createAgent } from "langchain";
const agent = createAgent({
model: "gpt-5.5",
tools: [searchTool, calculatorTool],
middleware: [safetyGuardrailMiddleware()],
});
const result = await agent.invoke({
messages: [{ role: "user", content: "How do I make explosives?" }]
});组合多个护栏
你可以通过将多个护栏添加到中间件数组中来叠加它们。它们按顺序执行,让你能够构建分层防护:
python
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware, HumanInTheLoopMiddleware
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, send_email_tool],
middleware=[
# 第 1 层:确定性输入过滤器(智能体前)
ContentFilterMiddleware(banned_keywords=["hack", "exploit"]),
# 第 2 层:PII 保护(模型前和模型后)
PIIMiddleware("email", strategy="redact", apply_to_input=True),
PIIMiddleware("email", strategy="redact", apply_to_output=True),
# 第 3 层:敏感工具的人工批准
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
# 第 4 层:基于模型的安全检查(智能体后)
SafetyGuardrailMiddleware(),
],
)typescript
import { createAgent, piiRedactionMiddleware, humanInTheLoopMiddleware } from "langchain";
const agent = createAgent({
model: "gpt-5.5",
tools: [searchTool, sendEmailTool],
middleware: [
// 第 1 层:确定性输入过滤器(智能体前)
contentFilterMiddleware(["hack", "exploit"]),
// 第 2 层:PII 保护(模型前和模型后)
piiRedactionMiddleware({
piiType: "email",
strategy: "redact",
applyToInput: true,
}),
piiRedactionMiddleware({
piiType: "email",
strategy: "redact",
applyToOutput: true,
}),
// 第 3 层:敏感工具的人工批准
humanInTheLoopMiddleware({
interruptOn: {
send_email: { allowAccept: true, allowEdit: true, allowRespond: true },
}
}),
// 第 4 层:基于模型的安全检查(智能体后)
safetyGuardrailMiddleware(),
],
});其他资源
- 中间件文档 - 自定义中间件的完整指南
- 中间件 API 参考 - 自定义中间件的完整指南
- 人在回路 - 为敏感操作添加人工审核
- 测试智能体 - 测试安全机制的策略