1. 项目概述为什么我们需要为AI应用链加上“安全锁”最近在折腾LangChain构建AI应用链的朋友估计都遇到过类似的头疼事你精心设计的Agent用户一个刁钻的Prompt就可能让它“口无遮拦”要么泄露了不该说的系统提示词要么从你连接的数据库里“顺嘴”说出了敏感信息。这还不是最糟的万一用户输入里夹带了恶意指令诱导模型去执行删除文件、调用危险API的操作那可就真成安全事故了。我自己在开发一个金融问答机器人时就深有体会RAG系统检索出的内部文档片段如果未经检查就直接喂给模型模型很可能会把诸如“内部测试账号admin/123456”这样的信息原封不动地吐出来。这就是为什么“安全”不再是大型语言模型应用开发中一个可选项而是必须内置的底层能力。LLM Guard这个库的出现正好切中了这个痛点。它不是另一个花哨的AI框架而是一个专注的“安全卫士”专门负责在数据流入流出LLM的各个环节进行扫描和过滤。而LangChain作为当前最流行的AI应用编排框架如何将LLM Guard无缝集成到其应用链中构建起从输入到输出的全方位防御体系就是我们今天要深入探讨的核心。这不仅仅是加几个过滤函数那么简单它关乎整个应用链的架构设计、性能权衡与安全纵深。接下来我会结合一个具体的金融风控场景案例带你一步步拆解集成的思路、核心模块的实战配置以及那些只有踩过坑才知道的避雷技巧。2. 核心思路构建纵深防御的AI应用链安全架构直接给LLM套上一个安全过滤器是最简单的想法但效果往往最差。因为安全威胁可能出现在链条的任何一个环节用户输入、模型输出、工具调用结果、甚至是记忆存储。因此我们的集成思路必须是系统性的、纵深防御的。2.1 理解LLM Guard的核心能力模块在动手集成前得先摸清LLM Guard手里有哪些“武器”。它主要提供了几大类扫描器Scanner我们可以像搭积木一样按需组合输入扫描器Input Scanners在用户Prompt进入LLM之前进行拦截。毒性Toxicity检测辱骂、仇恨言论。提示注入Prompt Injection识别试图覆盖系统指令的恶意提示。语言Language限制只处理特定语言如仅中文避免模型因不熟悉语言产生乱码或错误。拒绝服务Denial of Service检测过长的输入防止资源耗尽攻击。敏感信息Secrets尝试检测输入中是否包含API密钥、密码等硬编码秘密。输出扫描器Output Scanners在LLM生成回复后、返回给用户前进行过滤。毒性Toxicity同上确保模型输出文明。敏感话题Sensitive Topics可配置过滤如暴力、政治等特定话题。正则匹配Regex使用自定义正则表达式屏蔽电话号码、身份证号、银行卡号等特定模式信息。代码Code防止模型输出可执行代码片段除非你的应用场景需要。事实一致性Factual Consistency在RAG场景下检查模型输出是否与提供的上下文证据相悖。工具扫描器Tool/Agent Scanners这是一个关键但常被忽略的环节。当Agent调用外部工具如数据库查询、API请求时需要对工具的输入参数和返回结果进行安全检查。例如一个“文件读取”工具其输入路径参数需要被扫描防止../../../etc/passwd这样的路径遍历攻击。工具返回的结果可能包含敏感数据需要在拼接到最终提示前进行脱敏。2.2 LangChain集成策略在关键节点嵌入安全钩子LangChain的核心抽象是Runnable。每个链Chain、工具Tool、模型LLM都是一个Runnable。我们的目标就是在这些Runnable的组合流水线中插入作为安全审查节点的Runnable。具体来说有三个核心集成点在RunnableSequence的起始端嵌入输入扫描器将用户原始输入首先通过一个由LLM Guard输入扫描器构成的“安检门”。包装LLM模型在invoke或stream方法前后嵌入扫描逻辑这是最核心的集成。我们可以创建一个自定义的Runnable它内部封装了真正的LLM如ChatOpenAI并在调用LLM前执行输入扫描在得到响应后执行输出扫描。在工具调用Tool/Agent流程中嵌入扫描对工具的args进行输入扫描对工具的output进行输出扫描。这可以通过自定义Tool类或使用LangChain的RunnableLambda包装工具执行逻辑来实现。这种设计的好处是非侵入性和可组合性。你不需要重写现有的业务链只需要像添加中间件一样把安全Runnable插入到合适的位置。同时你可以根据不同的链例如一个对内的管理链和一个对外的客服链配置不同严格程度的安全扫描器组合。3. 实战集成一步步构建安全的RAG问答链理论说再多不如一行代码。我们以一个典型的、基于FastAPI和Qwen的金融知识RAG问答链为例演示如何将LLM Guard深度集成进去。假设我们的链流程是用户提问 - 输入安全检查 - 向量库检索 - 上下文组装 - 调用Qwen模型 - 输出安全检查 - 返回答案。3.1 环境准备与依赖安装首先确保你的环境包含以下核心包。我强烈建议使用虚拟环境。# 核心框架与模型 pip install langchain langchain-community langchain-openai # 向量数据库与嵌入模型这里以Chroma为例 pip install chromadb sentence-transformers # 安全核心 pip install llm-guard # Web框架 pip install fastapi uvicorn # 可选用于异步流式响应 pip install sse-starlette注意llm-guard的一些扫描器如事实一致性可能需要额外的模型依赖例如NLI模型请根据其官方文档按需安装。对于生产环境建议将所有依赖及其版本号写入requirements.txt并严格锁定。3.2 构建自定义的安全LLM Runnable这是集成的核心。我们将创建一个SecureQwenLLM类它继承自Runnable内部封装了真正的Qwen调用客户端和安全扫描器。from typing import Any, Dict, List, Optional, Union from langchain_core.runnables import Runnable, RunnableConfig from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from llm_guard import scan_output, scan_prompt from llm_guard.input_scanners import Toxicity, PromptInjection, Language from llm_guard.output_scanners import Toxicity as OutputToxicity, Regex, SensitiveTopics from llm_guard.vault import Vault import asyncio from openai import AsyncOpenAI # 假设使用OpenAI兼容的API调用Qwen class SecureQwenLLM(Runnable): def __init__(self, base_url: str, api_key: str, model: str qwen-max, input_scanners: Optional[List] None, output_scanners: Optional[List] None): 初始化安全LLM封装器。 Args: base_url: Qwen API的基础URL。 api_key: API密钥。 model: 使用的模型名称。 input_scanners: 输入扫描器列表如果为None则使用默认组合。 output_scanners: 输出扫描器列表如果为None则使用默认组合。 self.client AsyncOpenAI(base_urlbase_url, api_keyapi_key) self.model model # 初始化默认输入扫描器 if input_scanners is None: self.input_scanners [ Toxicity(threshold0.7), # 毒性检测阈值0.7 PromptInjection(threshold0.8), # 提示注入检测 Language(allowed_languages[zh]) # 只允许中文输入 ] else: self.input_scanners input_scanners # 初始化默认输出扫描器 if output_scanners is None: # 初始化一个虚拟保险库用于存储检测到的敏感信息可替换为真实存储 vault Vault() self.output_scanners [ OutputToxicity(threshold0.7), SensitiveTopics(topics[violence, financial_fraud], threshold0.7), # 屏蔽暴力和金融欺诈话题 Regex(bad_patterns[ r\b\d{17}[\dXx]\b, # 身份证号 r\b\d{16}\b, # 银行卡号 r\b1[3-9]\d{9}\b, # 手机号 ], redact_modepartial), # 部分脱敏如显示前3后4位 ] else: self.output_scanners output_scanners async def _scan_input(self, prompt: str) - tuple[bool, str, Dict]: 执行输入扫描。 sanitized_prompt prompt results_valid True scan_results {} for scanner in self.input_scanners: try: sanitized_prompt, is_valid, risk_score scanner.scan(sanitized_prompt) scan_results[scanner.__class__.__name__] { is_valid: is_valid, risk_score: risk_score, sanitized_prompt: sanitized_prompt } if not is_valid: results_valid False # 可以根据风险等级决定是阻断还是仅记录日志 # 这里简单处理一旦无效就停止后续扫描并返回 break except Exception as e: # 扫描器本身可能出错记录错误但不应阻断主流程根据安全策略决定 print(fInput scanner {scanner.__class__.__name__} error: {e}) scan_results[scanner.__class__.__name__] {error: str(e)} return results_valid, sanitized_prompt, scan_results async def _scan_output(self, text: str) - tuple[bool, str, Dict]: 执行输出扫描。 sanitized_text text results_valid True scan_results {} for scanner in self.output_scanners: try: sanitized_text, is_valid, risk_score scanner.scan(sanitized_text) scan_results[scanner.__class__.__name__] { is_valid: is_valid, risk_score: risk_score, sanitized_text: sanitized_text } if not is_valid: results_valid False # 输出无效时通常直接替换为安全回复而非传递原始文本 sanitized_text [内容因违反安全策略而被屏蔽] break except Exception as e: print(fOutput scanner {scanner.__class__.__name__} error: {e}) scan_results[scanner.__class__.__name__] {error: str(e)} # 输出扫描出错时出于安全考虑应阻断可能不安全的输出 results_valid False sanitized_text [安全检查过程发生错误] return results_valid, sanitized_text, scan_results async def _acall(self, input_data: Union[str, List[BaseMessage]], config: Optional[RunnableConfig] None) - str: 核心的异步调用方法。 # 1. 准备Prompt if isinstance(input_data, list): # 处理LangChain的Message列表 prompt \n.join([msg.content for msg in input_data if isinstance(msg, HumanMessage)]) else: prompt str(input_data) # 2. 输入扫描 input_valid, sanitized_prompt, input_scan_results await self._scan_input(prompt) if not input_valid: # 记录安全日志 print(fInput blocked. Scan results: {input_scan_results}) return 您的输入包含不合规内容请重新提问。 # 3. 调用底层LLM try: response await self.client.chat.completions.create( modelself.model, messages[{role: user, content: sanitized_prompt}], streamFalse # 简化示例非流式 ) raw_output response.choices[0].message.content except Exception as e: return f模型调用失败: {e} # 4. 输出扫描 output_valid, final_output, output_scan_results await self._scan_output(raw_output) # 记录输出扫描日志用于审计和分析 if not output_valid or output_scan_results: print(fOutput scanned. Results: {output_scan_results}. Final output: {final_output}) return final_output # 为了兼容LangChain的Runnable接口需要同步方法内部调用异步 def invoke(self, input_data: Union[str, List[BaseMessage]], config: Optional[RunnableConfig] None) - str: return asyncio.run(self._acall(input_data, config)) # 实现stream方法以支持流式输出更复杂需要逐块扫描 # 此处省略流式实现下文会单独讨论关键点解析扫描器组合我们在__init__中初始化了默认的输入输出扫描器。在实际项目中这个配置应该通过配置文件如YAML来管理方便对不同环境测试/生产和不同场景客服/审核设置不同的安全策略。扫描流程输入扫描是串联的一个扫描器的输出净化后的文本作为下一个扫描器的输入。一旦某个扫描器判定无效is_validFalse我们可以选择立即阻断并返回安全回复。输出扫描逻辑类似但通常一旦发现违规就会用预定义的安全文本来替换原始输出。错误处理扫描器本身可能抛出异常。对于输入扫描如果扫描器故障一个保守的策略是阻断请求Fail Closed因为无法确认其安全性。对于输出扫描故障时也应阻断输出防止潜在风险泄露。审计日志所有扫描结果无论是否违规都应该被详细记录。这对于事后追溯、调整扫描阈值、分析攻击模式至关重要。上面的print语句应替换为结构化的日志输出。3.3 将安全LLM集成到LangChain链中现在我们可以像使用普通LLM一样在LangChain的LCELLangChain Expression Language表达式中使用我们的SecureQwenLLM。from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader # 1. 准备知识库简化示例 loader TextLoader(financial_knowledge.txt) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) texts text_splitter.split_documents(documents) embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-small-zh-v1.5) vectorstore Chroma.from_documents(texts, embeddings, collection_namefinancial_kb) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 2. 创建安全LLM实例 secure_llm SecureQwenLLM( base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, # 假设的Qwen兼容API端点 api_keyyour-api-key-here, modelqwen-max ) # 3. 构建安全的RAG链 template 你是一个专业的金融助手请根据以下上下文信息回答问题。如果信息不足请明确告知。 上下文 {context} 问题{question} 请提供专业、准确的回答 prompt ChatPromptTemplate.from_template(template) # 定义检索并格式化上下文的Runnable def format_docs(docs): return \n\n.join([d.page_content for d in docs]) # 使用LCEL组合链 rag_chain ( {context: retriever | format_docs, question: lambda x: x[question]} | prompt | secure_llm # 关键这里使用了我们封装的安全LLM | StrOutputParser() ) # 4. 调用链 async def ask_question(question: str): result await rag_chain.ainvoke({question: question}) return result # 测试 import asyncio test_question 请问个人信用卡套现的常见手法有哪些 # 一个可能涉及敏感/违规话题的问题 answer asyncio.run(ask_question(test_question)) print(f问题{test_question}) print(f回答{answer})在这个链中secure_llm这个Runnable接管了原本ChatOpenAI或ChatQwen的位置。用户的问题经过检索、组装成最终提示词后会先经过secure_llm内部的输入扫描再调用真实的Qwen API返回的结果再经过输出扫描最后才被StrOutputParser处理。整个安全过程对上游的prompt和下游的parser都是透明的。3.4 处理流式输出Streaming的安全挑战上面的例子处理的是非流式响应。对于流式响应安全扫描会变得复杂因为我们需要在文本块chunk生成的过程中进行实时或近实时的扫描。LLM Guard的扫描器通常针对完整文本设计直接扫描每个chunk可能效果不佳例如一个敏感词被拆分成两个chunk就无法识别。一种实用的折中方案是“缓冲扫描”设置一个合理的缓冲区大小例如每积累50个字符或一个句子结束符。将积累的缓冲区内容送入输出扫描器。如果扫描通过则将缓冲区的字符逐个流出如果发现违规则立即停止流式传输并发送一个终止消息和替换文本。这需要在自定义的SecureQwenLLM中实现astream方法并可能涉及更复杂的状态管理。一个更简单但体验稍差的方法是先让LLM完整生成在内存中完成扫描然后再以流式方式发送已通过安全检查的文本。但这失去了真正的低延迟流式体验。实操心得对于大多数对实时性要求不极致的场景我推荐先使用非流式接口确保安全待技术成熟后再考虑安全的流式方案。或者可以将流式分为两段第一段是“安全生成”的流式内容一旦后台扫描发现问题立即通过一个独立的通道如WebSocket消息发送中断指令和替换内容。4. 高级场景为Agent和工具调用加上安全锁在Agent场景中安全风险更高因为模型可以自主决定调用工具。我们需要在两个地方加强控制4.1 工具输入参数扫描当Agent决定调用一个“执行SQL查询”的工具时我们需要检查它生成的SQL语句是否包含DROP TABLE、DELETE等高危操作。这可以通过在自定义Tool的_run方法开始时插入扫描逻辑来实现。from langchain.tools import BaseTool from llm_guard.input_scanners import Anonymize, Secrets # 示例使用匿名化和秘密检测 class SafeDatabaseTool(BaseTool): name query_financial_db description 执行只读的金融数据库查询。输入应为安全的SQL SELECT语句。 def _run(self, query: str) - str: # 1. 输入扫描检查SQL注入和敏感信息 sql_scanner PromptInjection(threshold0.9) # 高阈值严格检测注入 secret_scanner Secrets() sanitized_query, is_valid, _ sql_scanner.scan(query) _, secret_found, _ secret_scanner.scan(query) if not is_valid: return 错误检测到潜在的不安全查询语句已阻止执行。 if secret_found: return 错误查询中可能包含敏感信息已阻止执行。 # 2. 进一步白名单校验确保只是SELECT查询简单示例生产环境需更严谨 if not sanitized_query.strip().upper().startswith(SELECT): return 错误只允许执行SELECT查询。 # 3. 执行安全查询假设有一个安全的数据库连接池 # result safe_db_execute(sanitized_query) # return str(result) return f安全地执行了查询: {sanitized_query} async def _arun(self, query: str) - str: return self._run(query)4.2 工具输出结果扫描工具返回的数据可能包含用户不应看到的详细信息如内部错误信息、系统路径、部分敏感数据。需要在工具返回结果给Agent之前进行扫描和脱敏。class SafeFileReadTool(BaseTool): name read_file description 读取指定路径的文件内容。路径必须是相对路径。 def __init__(self, allowed_base_path: str): super().__init__() self.allowed_base_path allowed_base_path self.output_scanner Regex( bad_patterns[r\b(?:password|token|key|secret)\s*[:]\s*[\]?[\w\-\.][\]?, r\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b], # 简单密码模式和IP地址 redact_modefull # 完全替换 ) def _run(self, file_path: str) - str: # 1. 路径遍历攻击防护 import os requested_path os.path.join(self.allowed_base_path, file_path) canonical_path os.path.realpath(requested_path) if not canonical_path.startswith(os.path.realpath(self.allowed_base_path)): return 错误禁止访问指定路径之外的文件。 # 2. 读取文件 try: with open(canonical_path, r, encodingutf-8) as f: content f.read() except Exception as e: return f读取文件失败: {str(e)} # 3. 对读取的内容进行输出扫描和脱敏 sanitized_content, is_valid, _ self.output_scanner.scan(content) if not is_valid: # 即使无效我们也返回脱敏后的内容但可以加警告 sanitized_content [部分内容因包含敏感模式已被脱敏]\n sanitized_content return sanitized_content4.3 在Agent执行流程中集成安全使用LangGraph或自定义的AgentExecutor时我们可以通过RunnableLambda或中间件Middleware在Agent的每一步决策生成工具调用、解析工具输出前后插入安全钩子。from langchain.agents import AgentExecutor, create_react_agent from langchain_core.agents import AgentAction, AgentFinish # 假设我们已经有了一个基础的agent和tools列表tools中包含上述SafeTool # 创建一个安全包装器用于包装Agent的执行步骤 def safe_agent_step(state): 在Agent决策后、执行动作前进行安全检查 # state中包含agent决定的下一个动作AgentAction if isinstance(state[next], AgentAction): action state[next] # 检查要调用的工具是否在安全工具白名单内 if action.tool not in [t.name for t in safe_tools]: return {next: AgentFinish(return_values{output: f错误不允许调用工具 {action.tool}。}, log)} # 可以对action.tool_input进行输入扫描这里简化 # scanned_input, is_valid scan_input(str(action.tool_input)) # if not is_valid: ... return state # 在构建Agent执行流程时将这个安全步骤插入进去 # 使用LangGraph可以更优雅地实现这里展示概念5. 性能、配置与监控让安全策略真正落地集成安全组件不是一劳永逸的它引入了一定的复杂性和性能开销需要精心调优和监控。5.1 性能考量与优化扫描器开销像Toxicity、PromptInjection这类基于神经网络的扫描器推理耗时可能不亚于一次小型LLM调用。Regex扫描则很快。优化策略分层扫描先进行快速、轻量的规则扫描如Regex、长度限制过滤掉大部分明显问题再进行耗时的模型扫描。异步扫描利用asyncio并发执行多个不依赖的扫描器。缓存对常见的、安全的查询和回复进行缓存避免重复扫描。阈值调优不要一味追求“零误报”。过高的阈值如毒性检测设为0.9可能导致大量正常对话被拦截影响用户体验。需要通过日志分析在安全性和可用性之间找到平衡点。可以从较宽松的阈值开始逐步收紧。5.2 动态配置与策略管理安全策略不应硬编码在代码中。建议使用配置文件如YAML来定义不同环境、不同用户角色、不同功能模块的安全策略。# security_policies.yaml policies: customer_service: input_scanners: - name: Toxicity threshold: 0.7 - name: PromptInjection threshold: 0.75 - name: Language allowed_languages: [zh, en] output_scanners: - name: Toxicity threshold: 0.7 - name: Regex patterns: - pattern: \b\d{17}[\dXx]\b action: redact_partial # 部分脱敏 - pattern: \b(?:密码|口令|密钥)\s*[:]\s*\S action: block # 直接阻断 internal_admin: input_scanners: - name: Secrets threshold: 0.5 output_scanners: [] # 内部管理员可能不需要输出过滤在应用启动时加载配置并根据请求的上下文如用户角色动态创建对应的SecureQwenLLM实例。5.3 监控、审计与持续改进结构化日志记录每一次扫描的详细信息扫描器名称、输入/输出片段、风险分数、是否通过、处理动作放行/阻断/脱敏。这些日志应发送到集中的日志系统如ELK Stack。指标收集监控扫描器的调用延迟、阻断率、误报率。这些指标能帮助你发现性能瓶颈和策略问题。定期审计定期审查被阻断的请求日志分析攻击模式。是用户在尝试恶意注入还是你的扫描器过于敏感根据审计结果调整扫描器阈值或规则。红队演练定期对你的AI应用进行模拟攻击测试安全防护的有效性。尝试各种已知的提示注入、越狱Jailbreak技术看你的防护体系能否有效识别和阻断。6. 常见陷阱与疑难排查在实际集成中我踩过不少坑这里分享几个最典型的扫描器顺序导致的性能浪费把最耗时的扫描器如基于BERT的模型放在最前面。正确做法先进行快速、高召回率的规则扫描如Regex、关键词快速阻断明显违规内容避免不必要的复杂计算。流式响应与扫描的冲突如前所述流式输出和逐块扫描存在矛盾。临时方案对于强安全要求的场景暂时关闭流式或采用“缓冲扫描”方案。同时关注LLM Guard等社区是否推出支持流式或增量扫描的版本。过度脱敏影响可读性正则表达式过于激进把正常的数字序列如产品代码“SN12345678”也脱敏了。解决方案精心设计正则模式使用更精确的上下文感知方法如命名实体识别NER来区分敏感信息。对于金融场景可以集成专门的金融数据脱敏库。Agent工具安全的白名单漏洞只检查了工具名但工具内部可能根据输入参数执行不同分支代码。解决方案安全策略必须下沉到工具内部。像上面的SafeDatabaseTool示例不仅在工具调用入口检查还要在内部业务逻辑的关键点如SQL拼接处、文件路径解析处进行校验。误报False Positive处理用户正常提问“如何防范钓鱼邮件攻击”被敏感话题扫描器阻断。处理流程首先在安全回复中给用户一个友好的提示如“您的问题可能涉及敏感话题如果您确认是正当咨询请尝试换一种方式提问或联系人工客服。”其次后台必须有便捷的“误报申诉”通道让审核人员可以快速查看并放行合法请求同时用于优化扫描器模型。依赖库版本冲突llm-guard可能依赖特定版本的transformers或torch与你项目中其他组件冲突。建议使用poetry或uv这类能进行更严格依赖解析的包管理工具。在Docker容器中固化依赖环境。将LLM Guard集成到LangChain应用链中本质上是将安全思维“左移”到了AI应用开发的架构设计阶段。它不是一个可以事后附加的插件而是一套需要贯穿数据流始终的防护体系。通过本文的指南你应当能够构建起一个具备输入输出过滤、工具调用检查的基本安全框架。但请记住安全是一个持续的过程而非一个静态的目标。你需要结合具体的业务场景不断调整你的扫描策略、优化性能、分析日志才能让你的AI应用在充满挑战的环境中稳健运行。