1. 先搞清楚“AI Agent安全”到底在防什么如果你正在用LangChain、LangGraph这类框架做AI应用开发尤其是涉及金融、政务、客服等有合规要求的场景那么“安全”这个词就绝对不能停留在概念层面。它不是一个可选的附加功能而是决定你的Agent能否上线的第一道门槛。很多人一提到Agent安全第一反应是“别让AI胡说八道”于是去折腾提示词工程试图用“System Prompt”把模型锁死。这思路对了一半但远远不够。真正的生产级安全防的是两类更具体、更致命的风险信息泄露用户或系统在对话中无意或有意输入了身份证号、手机号、银行卡号、地址等敏感信息PII你的Agent在处理、调用工具或生成回答时把这些信息原封不动地输出或记录到了日志、数据库里。越权操作Agent根据用户指令调用了某个工具Tool比如“发送邮件”、“查询数据库”、“调用内部API”。如果缺乏校验一个普通用户可能通过精心构造的对话让Agent执行管理员才能做的操作或者访问不该他看的数据。所以一个安全的AI Agent核心能力不是“会说话”而是“能闭嘴”和“能请示”。“闭嘴”指的是对流入流出的信息进行自动化的敏感信息识别与脱敏“请示”指的是在执行关键操作前有能力触发人工审核流程。LangChain提供的“护栏”Guardrails机制就是用来系统化地构建这两层防御的。这篇文章我会以一个接近银行客服的场景为例拆解如何用LangChain的四层护栏输入、输出、工具调用、人工审批搭建一个安全闭环。你会看到从单点防护到流程串联的全过程而不仅仅是几个API的调用。2. 环境与核心概念LangChain的“护栏”到底是什么在开始写代码之前得先把环境搭对概念理清。LangChain的版本迭代很快安全相关的模块也在不断整合。2.1 环境准备与依赖安装我建议创建一个干净的Python虚拟环境来操作避免包冲突。核心依赖就两个langchain和langchain-community社区工具集。如果你需要用到特定的LLM比如OpenAI再额外安装openai。# 创建并激活虚拟环境以conda为例 conda create -n langchain-safety python3.10 conda activate langchain-safety # 安装核心包 pip install langchain langchain-community # 按需安装LLM包这里以OpenAI为例 pip install openai关键点务必检查你的langchain版本。护栏相关功能在较新的版本中如0.1.0才比较稳定。可以用pip list | grep langchain查看。2.2 理解“四层护栏”的职责LangChain的护栏Guardrails不是一个大而化之的开关而是针对Agent运行流程中不同环节的钩子Hooks。我们说的“四层”通常指输入护栏Input Guardrails在用户输入User Input被传递给LLM或Agent之前进行拦截和清洗。主要做敏感信息检测与脱敏。例如把“我的身份证是110101199003077832”变成“我的身份证是[ID_CARD]”。输出护栏Output Guardrails在LLM或Agent生成回复Response之后返回给用户之前进行拦截和清洗。同样用于敏感信息检测与脱敏防止模型在生成内容时泄露信息。同时也可以做内容合规性检查如是否包含不当言论。工具调用护栏Tool Call Guardrails在Agent决定要调用某个工具Tool时进行拦截。这是权限控制和操作审计的关键。可以检查当前用户是否有权调用此工具或者这个调用请求是否合理。人工审批护栏Human-in-the-Loop这不是一个独立的护栏类型而是一种特殊的处理策略。通常集成在工具调用护栏或关键决策点当检测到高风险操作如“转账”、“修改权限”时暂停自动化流程将决策权交给真人审批。这四层构成了一个纵深防御体系。输入输出护栏管“数据”工具调用护栏管“行为”人工审批是最后的“安全阀”。3. 第一层实战输入输出敏感信息自动脱敏这是最基础也最常用的一层。目标很简单不让明文敏感信息进入LLM的上下文也不让它们从LLM的输出中泄露。3.1 选择与配置脱敏工具LangChain社区提供了现成的脱敏工具比如langchain_community.document_transformers里的BeautifulSoupTransformer可以处理HTML但对于PII我们更常用NerTransformer基于NER模型或使用第三方服务。这里我演示一个更直接、可控的方案使用正则表达式匹配和替换。我们先自己实现一个简单的脱敏函数然后将其嵌入到LangChain的流程中。import re from typing import Dict, Any from langchain_core.messages import HumanMessage from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 1. 定义敏感信息模式与替换规则 SENSITIVE_PATTERNS { r\b\d{17}[\dXx]\b: [ID_CARD], # 身份证号 r\b1[3-9]\d{9}\b: [PHONE], # 手机号 r\b\d{16,19}\b: [BANK_CARD], # 银行卡号 # 可以继续添加邮箱、地址等模式 } def sanitize_text(text: str) - str: 对文本进行脱敏处理 sanitized text for pattern, replacement in SENSITIVE_PATTERNS.items(): sanitized re.sub(pattern, replacement, sanitized) return sanitized # 2. 创建带脱敏功能的提示词模板和链 prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手请根据用户问题回答。注意所有涉及个人身份的信息都已用占位符如[ID_CARD]代替。), (human, {user_input}) ]) llm ChatOpenAI(modelgpt-3.5-turbo) # 请设置你的OPENAI_API_KEY chain prompt | llm | StrOutputParser() # 3. 在调用链之前处理输入 def safe_invoke(user_input: str): sanitized_input sanitize_text(user_input) print(f原始输入: {user_input}) print(f脱敏后输入: {sanitized_input}) response chain.invoke({user_input: sanitized_input}) # 理论上输出也应该再脱敏一次防止LLM“编造”出敏感信息 sanitized_response sanitize_text(response) print(f原始输出: {response}) print(f脱敏后输出: {sanitized_response}) return sanitized_response # 测试 test_input 用户说我的身份证是110101199003077832手机号是13912345678请帮我查一下余额。 result safe_invoke(test_input)运行后你会看到身份证号和手机号在输入LLM前就被替换成了[ID_CARD]和[PHONE]。LLM永远看不到真实号码从根本上避免了信息泄露。输出环节再做一次脱敏是双保险防止模型在不知情的情况下生成类似格式的信息。关键点正则的局限性正则简单快速但覆盖不全容易误判或漏判。生产环境建议使用专业的NLP服务或模型如Presidio、Microsoft的PII检测服务。脱敏策略直接替换为占位符是最简单的。更复杂的策略可以是对称加密或标记化Tokenization以便在后续授权环节能还原。位置这个脱敏函数应该放在LangChain的RunnableLambda或自定义的Runnable中以便更好地集成到链里。3.2 集成到LangChain的Runnable流程上面的例子是手动调用。更好的方式是利用LangChain的Runnable接口创建可组合的脱敏环节。from langchain_core.runnables import RunnableLambda # 创建输入脱敏的Runnable input_sanitizer RunnableLambda(lambda x: {user_input: sanitize_text(x[user_input])}) # 创建输出脱敏的Runnable output_sanitizer RunnableLambda(lambda x: sanitize_text(x)) # 构建安全的链 safe_chain ( input_sanitizer # 第一步输入脱敏 | prompt # 第二步填充提示词 | llm # 第三步调用模型 | StrOutputParser() # 第四步解析输出 | output_sanitizer # 第五步输出脱敏 ) # 现在可以像普通链一样调用 response safe_chain.invoke({user_input: test_input}) print(response)这样脱敏就成了链中一个透明的环节逻辑更清晰也更容易测试和复用。4. 第二层实战工具调用的权限拦截与审计Agent的强大在于能调用工具。但能力越大风险越大。工具调用护栏的核心是在工具执行前判断“能不能做”和“该不该做”。4.1 定义有风险的工具假设我们有一个银行客服Agent它有两个工具query_balance: 查询账户余额低风险但需验证身份。transfer_money: 发起转账高风险必须严格管控。from langchain.tools import tool from typing import Optional tool def query_balance(account_id: str, user_token: str) - str: 根据账户ID和用户令牌查询余额。这是一个低风险操作。 # 模拟查询逻辑 # 在实际应用中这里会调用内部API或数据库 if user_token valid_token_123: return f账户 {account_id} 的余额是 10000.00 元。 else: return 身份验证失败无法查询余额。 tool def transfer_money(from_account: str, to_account: str, amount: float, approval_code: Optional[str] None) - str: 从一个账户转账到另一个账户。这是一个高风险操作需要人工审批码。 # 模拟转账逻辑 if approval_code APPROVE_2024: return f成功从 {from_account} 向 {to_account} 转账 {amount} 元。 else: return 转账失败缺少或审批码无效。需要人工审批。4.2 实现工具调用护栏我们需要在Agent调用工具前插入检查逻辑。在LangChain中可以通过创建自定义的BaseTool类重写_run方法或者在AgentExecutor层面通过callbacks和middleware来实现。这里展示一个通过自定义Tool包装器实现的清晰方案。from langchain.tools import BaseTool from pydantic import BaseModel, Field import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class ToolGuardrailWrapper(BaseTool): 工具包装器增加权限检查和审计日志 original_tool: BaseTool required_permission: str user # 默认需要用户权限 def _run(self, *args, **kwargs): # 1. 权限检查这里简化为例从上下文获取用户角色 # 假设我们从某个上下文如会话状态中能拿到当前用户信息 current_user_role kwargs.pop(_user_role, guest) # 默认为访客 if self.required_permission admin and current_user_role ! admin: logger.warning(f权限拒绝用户 {current_user_role} 试图调用管理员工具 {self.name}) return f错误您没有权限执行此操作{self.name}。 # 2. 审计日志记录谁在什么时候调用了什么工具参数是什么 # 注意真实参数中可能包含敏感信息记录前需要脱敏 sanitized_args {k: (v if k not in [from_account, to_account] else [ACCOUNT]) for k, v in kwargs.items()} logger.info(f工具调用审计 - 用户: {current_user_role}, 工具: {self.name}, 参数: {sanitized_args}) # 3. 高风险工具特殊处理例如转账需要审批码 if self.name transfer_money: if kwargs.get(approval_code) is None: logger.critical(f高风险操作拦截转账请求缺少审批码。参数: {sanitized_args}) return 此操作需要人工审批。请提供有效的审批码。 # 4. 一切检查通过调用原始工具 logger.info(f工具调用放行{self.name}) return self.original_tool.run(*args, **kwargs) async def _arun(self, *args, **kwargs): # 异步版本逻辑同上 return self._run(*args, **kwargs) # 包装原始工具 guarded_query_balance ToolGuardrailWrapper( original_toolquery_balance, namequery_balance, descriptionquery_balance.description, required_permissionuser # 查询余额需要用户权限 ) guarded_transfer_money ToolGuardrailWrapper( original_tooltransfer_money, nametransfer_money, descriptiontransfer_money.description, required_permissionadmin # 转账需要管理员权限 )现在guarded_transfer_money在被调用时会先检查用户角色是否为admin然后检查是否有approval_code最后才会执行真正的转账逻辑。所有调用尝试都会被记录到日志中。关键点上下文传递如何将用户身份如_user_role从Agent的会话传递到工具层是设计的关键。这通常需要借助LangChain的RunnableConfig或自定义回调函数来传递上下文。参数脱敏审计日志绝不能记录明文敏感信息。在记录前必须对工具参数进行脱敏处理如上例中对账号的掩码。粒度控制权限检查可以非常细致比如基于用户、基于时间、基于额度等。5. 第三层实战串联人工审批流程对于最高风险的操作如大额转账自动化的权限检查可能还不够需要引入真人裁决。这就是“人在回路”Human-in-the-Loop。5.1 设计审批触发机制我们修改ToolGuardrailWrapper当检测到特定条件如转账金额超过阈值时不直接执行或拒绝而是抛出一个特殊事件将流程挂起等待外部审批。class HumanApprovalRequired(Exception): 自定义异常表示需要人工审批 def __init__(self, tool_name: str, request_id: str, context: dict): self.tool_name tool_name self.request_id request_id self.context context # 包含审批所需信息已脱敏 super().__init__(fHuman approval required for {tool_name}. Request ID: {request_id}) class ToolGuardrailWrapperWithApproval(BaseTool): original_tool: BaseTool approval_threshold: float 5000.0 # 超过5000元需要审批 def _run(self, *args, **kwargs): current_user_role kwargs.pop(_user_role, guest) # 1. 基础权限检查 if self.required_permission admin and current_user_role ! admin: return f错误权限不足。 # 2. 高风险操作检查是否需要人工审批 if self.name transfer_money: amount kwargs.get(amount, 0) if amount self.approval_threshold: # 生成一个唯一的审批请求ID import uuid request_id str(uuid.uuid4())[:8] # 准备审批上下文务必脱敏 approval_context { request_id: request_id, tool: self.name, user: current_user_role, from_account: [ACCOUNT_MASKED], to_account: [ACCOUNT_MASKED], amount: amount, timestamp: datetime.now().isoformat() } logger.warning(f触发人工审批: {approval_context}) # 抛出异常中断自动化流程 raise HumanApprovalRequired( tool_nameself.name, request_idrequest_id, contextapproval_context ) # 3. 低风险或已审批操作继续执行 logger.info(f工具执行: {self.name}) return self.original_tool.run(*args, **kwargs)5.2 构建审批处理与流程恢复当HumanApprovalRequired异常被抛出时你的主程序需要捕获它并将审批上下文request_id,context存储到数据库或消息队列中同时通知审批人通过邮件、钉钉、内部系统等。审批人在一个管理后台看到待审批任务做出“通过”或“拒绝”的决定。# 模拟一个简单的审批处理函数 pending_approvals {} # 用字典模拟存储生产环境用DB def process_agent_request(user_input: str, user_role: str): 处理Agent请求的主函数 try: # ... 这里是构建和调用Agent链的代码 ... # 假设链的调用最终会用到我们的 guarded tool # 在调用链时传入用户角色 result agent_chain.invoke( {input: user_input}, config{callbacks: [], _user_role: user_role} # 传递用户角色 ) return result except HumanApprovalRequired as e: # 捕获审批异常 pending_approvals[e.request_id] { context: e.context, original_kwargs: kwargs, # 注意这里需要保存原始参数以便恢复但生产环境需加密存储 status: pending } # 通知审批人这里打印日志模拟 logger.info(f已创建审批请求 {e.request_id}。请管理员处理。) # 返回等待信息给用户 return f您的操作请求ID: {e.request_id}已提交需要人工审批。请等待通知。 # 模拟审批通过后的恢复执行 def approve_and_resume(request_id: str, approval_code: str): 审批通过后恢复执行被挂起的工具调用 if request_id not in pending_approvals: return 审批请求不存在或已处理。 request_data pending_approvals[request_id] if request_data[status] ! pending: return 该请求已处理。 # 恢复原始调用参数并附上审批码 original_kwargs request_data[original_kwargs] original_kwargs[approval_code] approval_code # 重新执行工具调用这里需要能定位到原始的工具对象 # 注意生产环境需要更严谨的状态恢复机制 tool_to_resume guarded_transfer_money # 假设我们知道是这个工具 try: result tool_to_resume.run(**original_kwargs) request_data[status] approved logger.info(f审批请求 {request_id} 已执行结果: {result}) return f审批通过操作已执行{result} except Exception as ex: logger.error(f恢复执行失败 {request_id}: {ex}) return f操作执行失败{ex}这样一个完整的人工审批闭环就建立了触发 - 挂起 - 通知 - 审批 - 恢复执行。关键点状态管理被挂起的请求状态参数、上下文必须持久化存储并能通过request_id准确恢复。安全性存储的原始参数必须加密审批后台必须有严格的权限控制。用户体验需要给最终用户清晰的等待状态提示并可能在审批后通过异步方式如WebSocket、轮询通知结果。6. 第四层整合与生产级考量前三层是点状的防护我们需要把它们串起来形成一个在LangChain Agent中自洽运行的流程。同时还要考虑生产环境下的稳定性、监控和迭代。6.1 构建带多层护栏的Agent执行链我们将输入脱敏、工具调用护栏整合到一个标准的ReAct Agent中。from langchain.agents import AgentExecutor, create_react_agent from langchain.memory import ConversationBufferMemory from langchain import hub # 1. 准备工具列表使用我们加了护栏的工具 tools [guarded_query_balance, guarded_transfer_money_with_approval] # 假设第二个工具是带审批的版本 # 2. 从LangChain Hub拉取一个ReAct风格的提示词可自定义 prompt hub.pull(hwchase17/react-chat) # 3. 创建Agent llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) agent create_react_agent(llm, tools, prompt) # 4. 创建Agent执行器并注入内存和护栏逻辑 # 注意这里的高阶护栏如整个Agent的输入输出过滤可以通过callbacks或自定义Runnable实现。 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, # 开启详细日志方便调试 handle_parsing_errorsTrue, # 处理解析错误 # 可以在这里传入 max_iterations, early_stopping_method 等控制流 ) # 5. 定义一个最外层的安全调用包装函数 def safe_agent_invoke(user_input: str, user_role: str): 安全的Agent调用入口 # a. 输入脱敏 sanitized_input sanitize_text(user_input) print(f[安全日志] 输入已脱敏: {user_input} - {sanitized_input}) # b. 调用Agent传入用户角色需要通过config或自定义方式传递到工具层 # 这里演示通过修改agent_executor的临时配置传递。更健壮的做法是使用自定义Callback或修改工具绑定。 try: # 注意实际项目中需要设计机制将user_role传递到每个tool的调用中。 # 一种方法是在invoke时将user_role放入config的“run_name”或自定义字段在Tool的_run方法中读取。 # 此处为演示我们简化处理假设有一个全局状态或上下文管理器。 result agent_executor.invoke( {input: sanitized_input}, # config 可以用于传递一些上下文但标准AgentExecutor不一定直接传给tool ) # c. 输出脱敏 sanitized_output sanitize_text(result[output]) print(f[安全日志] 输出已脱敏) return sanitized_output except HumanApprovalRequired as e: # 处理人工审批中断 return process_approval_interruption(e) except Exception as e: # 处理其他异常 logger.error(fAgent执行异常: {e}) return 系统处理您的请求时出现错误请稍后再试或联系管理员。6.2 生产环境必须考虑的五个问题当你把带护栏的Agent推向生产时光有功能还不够必须解决以下问题性能与延迟脱敏正则匹配很快但复杂的NLP模型如用于精准PII识别的会显著增加延迟。考虑异步处理或缓存热点数据。工具调用检查权限查询如果是远程调用如查询权限中心网络延迟是关键。需要设置合理的超时和降级策略如“默认拒绝”或“缓存权限”。人工审批这是最大的延迟源。必须设定超时如24小时未审批则自动拒绝并给用户明确的预期。错误处理与降级当脱敏服务不可用时是阻断请求Fail Closed还是放行明文Fail Open在金融场景通常选择阻断并记录告警。当权限服务超时时是默认拒绝还是允许一个低权限集安全优先默认拒绝。Agent执行过程中任何一层护栏抛出未捕获异常都应有统一的错误处理返回友好提示并触发告警。审计日志的完备性必须记录时间戳、会话ID、用户ID、输入脱敏后、输出脱敏后、调用的工具、工具参数脱敏后、权限检查结果、审批状态、最终结果。日志需要集中收集如ELK并设置关键风险操作如调用transfer_money的实时告警。护栏规则的动态更新敏感词库、权限规则、审批阈值不可能一成不变。需要设计管理界面或API支持在不重启服务的情况下热更新这些规则。例如将正则模式、审批阈值存储在数据库或配置中心护栏组件定期拉取或监听变更。测试与验证单元测试对每个脱敏函数、权限检查函数进行充分测试包括边界情况如超长字符串、特殊字符、混淆的敏感信息。集成测试模拟完整对话流测试护栏是否在正确环节触发。例如输入带手机号的问题查看日志中是否被脱敏模拟低权限用户请求转账查看是否被拦截。渗透测试尝试绕过护栏例如使用Unicode变体、图片OCR如果支持上传、提示词注入Prompt Injection让模型忽略系统指令等。6.3 监控与可观测性上线后必须监控以下指标护栏触发率每天有多少请求被脱敏有多少工具调用被权限拦截有多少触发了人工审批这能帮你了解风险概况。平均审批时间从触发审批到人工处理完成的平均耗时。如果时间过长需要优化审批流程。误报/漏报率脱敏是否把正常信息误判了误报是否有没有识别出的真实敏感信息漏报需要定期抽样审计。Agent性能基线加入护栏后平均响应时间增加了多少这关系到用户体验和资源规划。7. 总结从功能实现到安全思维搭建一个具备四层护栏的AI Agent技术实现只是第一步。更关键的是建立一套安全优先的开发与运维思维。默认拒绝在权限不明确或检查失败时Agent的默认行为应该是拒绝执行而不是冒险尝试。最小权限给Agent工具授权时遵循最小权限原则。一个客服Agent不需要“删除用户”的权限。纵深防御不要依赖单一护栏。输入脱敏可能失效所以输出也要脱敏自动权限检查可能被绕过所以对核心操作加人工审批。可审计所有决策、所有操作都必须有迹可循。日志是你的最后一道防线当出现问题时可以用来复盘和定责。持续迭代攻击手段在进化业务规则在变化。安全护栏的规则和策略也需要定期回顾和更新。回到开头的问题LangChain的护栏框架提供了很好的钩子但真正的安全来自于你如何利用这些钩子构建贴合业务、考虑周全的防护逻辑。从识别敏感信息开始到控制工具调用再到引入人工监督每一步都需要你把Agent当作一个可能出错的“新员工”为它划定清晰的行动边界。这样构建出来的AI应用才敢在真实的业务场景里放手去用。