在 AI 应用开发中构建一个既智能又安全的 AI Agent 已经成为工程实践的核心挑战。很多团队在初步尝试 LangChain 或类似框架时往往把重点放在功能实现上而忽略了安全护栏的重要性。等到实际部署后才发现未经检查的 AI 输出可能导致数据泄露、违规内容或系统异常。真正可靠的 AI Agent 需要在架构层面就考虑安全可控而不是事后补救。本文将围绕 LangChain 的分层架构从底层组件到高层 Agent 编排逐步构建一个具备多层防护的 AI 应用。重点会放在 Guardrails 护栏机制的集成、手写 PII个人可识别信息检测代码以及如何在实际项目中平衡灵活性与安全性。通过完整的代码示例和配置说明你可以掌握构建生产级 AI Agent 的关键技术。1. 理解 LangChain 的分层架构与安全挑战LangChain 的核心价值在于将大语言模型LLM的调用、工具使用、记忆管理和决策逻辑组织成可维护的工程结构。但在深入安全机制前需要先理解其分层设计因为安全防护需要对应到每一层。1.1 LangChain 的六层架构典型的 LangChain 应用可以分为六个层次模型层Models直接与 LLM 交互处理输入输出标准化提示层Prompts管理模板和少量示例控制 LLM 的输入格式链层Chains将多个组件组合成可复用的业务流程检索层Retrieval处理外部知识库的查询和匹配代理层Agents让 LLM 决定何时以及如何使用工具记忆层Memory维护对话历史和上下文状态安全风险分布在每一层模型层可能泄露敏感信息提示层可能被注入恶意指令代理层可能执行危险操作。有效的防护需要在各层设置检查点。1.2 AI Agent 的典型安全风险在实际项目中AI Agent 可能面临以下几类安全挑战数据泄露LLM 可能在响应中意外暴露训练数据中的个人信息提示注入用户输入可能包含恶意指令绕过系统设定的行为规则工具滥用Agent 可能调用不该调用的工具执行删除、修改等危险操作内容违规生成的内容可能包含不当、偏见或违规信息资源耗尽恶意用户可能通过复杂查询消耗大量计算资源针对这些风险需要建立纵深防御体系而 Guardrails 正是专门为此设计的框架。2. 环境准备与依赖配置构建安全可控的 AI Agent 需要精心选择工具版本特别是 LangChain 生态更新频繁版本兼容性至关重要。2.1 环境要求与版本选择基于当前项目需求推荐以下版本组合# requirements.txt langchain0.1.11 langchain-community0.0.28 langchain-core0.1.33 guardrails-ai0.4.1 pydantic2.6.4 python-dotenv1.0.0这个组合确保了 LangChain 核心组件与社区工具的兼容性。特别注意langchain 0.1.x 版本与早期的 0.0.x 版本在 API 设计上有较大变化新项目建议直接使用 0.1.x 系列。2.2 项目结构设计良好的项目结构是安全实践的基础ai-agent-project/ ├── config/ │ ├── __init__.py │ ├── settings.py # 应用配置 │ └── guards.py # Guardrails 配置 ├── core/ │ ├── __init__.py │ ├── security.py # 安全检测模块 │ └── agents.py # Agent 定义 ├── tools/ │ ├── __init__.py │ └── custom_tools.py # 自定义工具 ├── scripts/ │ └── setup_environment.py ├── tests/ │ └── test_security.py ├── .env.example └── main.py这种结构将配置、核心逻辑、工具实现分离便于维护和安全审计。2.3 环境变量与密钥管理AI 项目必须妥善管理 API 密钥和配置# config/settings.py import os from dotenv import load_dotenv from pydantic_settings import BaseSettings load_dotenv() class Settings(BaseSettings): # LLM 配置 openai_api_key: str os.getenv(OPENAI_API_KEY, ) openai_base_url: str os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) # 安全配置 pii_detection_enabled: bool True max_response_length: int 1000 allowed_tools: list [search, calculator, time] class Config: env_file .env settings Settings()对应的.env文件模板# .env.example OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 PII_DETECTION_ENABLEDtrue MAX_RESPONSE_LENGTH1000注意永远不要将真实的 API 密钥提交到版本控制系统。使用.env文件管理敏感配置并通过.gitignore确保其不会被意外提交。3. 构建基础 Guardrails 护栏系统Guardrails 是一个专门为 AI 应用设计的安全框架可以在 LLM 输入输出上施加约束和验证。与简单的后处理不同Guardrails 能够与 LangChain 深度集成在多个环节提供保护。3.1 Guardrails 核心概念Guardrails 基于三个核心概念Rail Spec用 XML 格式定义预期的输出结构和约束Validator自定义的验证逻辑检查内容是否符合要求Guard将 Rail Spec 和 Validator 组合成的可执行防护单元3.2 配置基础护栏首先定义基本的 Rail Spec限制输出格式和内容!-- config/guards/basic_spec.xml -- rail version0.1 output string nameresponse descriptionAI assistants response to the user query formatvalid-json on-fail-valid-jsonreask / /output prompt Given the following user query, provide a helpful and appropriate response. User query: ${user_query} ${gr.complete_json_suffix} /prompt /rail对应的 Python 配置代码# config/guards.py from guardrails import Guard from guardrails.validators import ValidLength, TwoWords, OneLine def create_basic_guard(): 创建基础输出防护 guard Guard.from_rail_string( rail version0.1 output string nameresponse descriptionAI assistants response formatlength: 1 1000 on-fail-lengthfix / /output /rail ) return guard3.3 集成到 LangChain 调用链将 Guardrails 集成到 LangChain 的 LLM 调用中# core/security.py from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from guardrails import Guard class GuardrailsWrapper: def __init__(self, guard: Guard): self.guard guard def __call__(self, llm_response: str) - str: 包装 LLM 响应应用 Guardrails 验证 try: # 应用护栏验证 validated_output self.guard.parse(llm_response) return validated_output except Exception as e: # 验证失败时的处理 return f安全检查未通过: {str(e)} def create_secured_chain(llm, guard): 创建带有安全防护的调用链 guard_wrapper GuardrailsWrapper(guard) chain ( llm | StrOutputParser() | RunnableLambda(guard_wrapper) ) return chain这种设计确保了所有 LLM 输出都经过护栏检查防止意外内容泄露。4. 手写 PII 检测与数据脱敏虽然可以使用现成的 PII 检测服务但在某些对数据隐私要求极高的场景手写检测逻辑可以提供更好的可控性和透明度。4.1 PII 类型识别首先定义需要检测的 PII 类型# core/security.py import re from typing import List, Dict, Tuple from dataclasses import dataclass dataclass class PIIType: name: str patterns: List[re.Pattern] description: str class PIIDetector: def __init__(self): self.pii_types self._initialize_pii_types() def _initialize_pii_types(self) - List[PIIType]: 初始化 PII 检测规则 return [ PIIType( nameemail, patterns[re.compile(r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b)], description电子邮件地址 ), PIIType( namephone, patterns[re.compile(r\b\d{3}[-.]?\d{3}[-.]?\d{4}\b)], description手机号码 ), PIIType( nameid_card, patterns[re.compile(r\b\d{17}[\dXx]\b)], description身份证号 ), PIIType( namebank_card, patterns[re.compile(r\b\d{16,19}\b)], description银行卡号 ) ]4.2 检测与脱敏实现实现具体的检测和脱敏逻辑# core/security.py dataclass class DetectionResult: pii_type: str original_text: str start_pos: int end_pos: int confidence: float class PIIDetector: # ... 初始化代码 ... def detect(self, text: str) - List[DetectionResult]: 检测文本中的 PII results [] for pii_type in self.pii_types: for pattern in pii_type.patterns: for match in pattern.finditer(text): result DetectionResult( pii_typepii_type.name, original_textmatch.group(), start_posmatch.start(), end_posmatch.end(), confidenceself._calculate_confidence(match.group(), pii_type.name) ) results.append(result) return sorted(results, keylambda x: x.start_pos) def anonymize(self, text: str, detection_results: List[DetectionResult]) - str: 对检测到的 PII 进行脱敏处理 if not detection_results: return text # 从后往前替换避免位置偏移 segments [] last_pos 0 for result in sorted(detection_results, keylambda x: x.start_pos, reverseTrue): # 添加脱敏后的片段 segments.append(text[result.end_pos:]) segments.append(self._get_anonymized_text(result.original_text, result.pii_type)) last_pos result.start_pos segments.append(text[:last_pos]) return .join(reversed(segments)) def _get_anonymized_text(self, original: str, pii_type: str) - str: 生成脱敏后的文本 if pii_type email: local, domain original.split(, 1) return f{local[0]}***{domain} elif pii_type phone: return original[:3] **** original[-4:] elif pii_type id_card: return original[:6] ******** original[-4:] elif pii_type bank_card: return **** original[-4:] else: return *** # 通用脱敏4.3 集成到数据处理流水线将 PII 检测集成到 LangChain 的文档处理流程中# core/security.py from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter class SecureDocumentProcessor: def __init__(self, pii_detector: PIIDetector): self.detector pii_detector self.splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) def process_document(self, document: Document) - Document: 处理单个文档检测并脱敏 PII content document.page_content # 检测 PII detections self.detector.detect(content) if detections: # 脱敏处理 anonymized_content self.detector.anonymize(content, detections) # 创建新文档保留元数据 return Document( page_contentanonymized_content, metadata{ **document.metadata, pii_detected: True, pii_types: list(set(d.pii_type for d in detections)), original_length: len(content), anonymized_length: len(anonymized_content) } ) return document5. 构建多层防护的 AI Agent现在将安全组件整合到完整的 AI Agent 中实现端到端的防护。5.1 Agent 工具的安全包装首先确保所有工具调用都经过安全检查# tools/custom_tools.py from langchain.tools import BaseTool from typing import Type from pydantic import BaseModel, Field class SecureBaseTool(BaseTool): 安全基础工具类 def _run(self, *args, **kwargs): # 前置安全检查 self._pre_execution_check(*args, **kwargs) # 执行工具逻辑 result self._secure_run(*args, **kwargs) # 后置结果检查 return self._post_execution_check(result) def _pre_execution_check(self, *args, **kwargs): 执行前检查 # 检查参数是否包含敏感信息 from core.security import PIIDetector detector PIIDetector() all_inputs .join([str(arg) for arg in args] [str(kwargs)]) detections detector.detect(all_inputs) if detections: raise ValueError(f工具输入包含敏感信息: {[d.pii_type for d in detections]}) def _post_execution_check(self, result: str) - str: 执行后结果检查 from core.security import PIIDetector detector PIIDetector() detections detector.detect(str(result)) if detections: return detector.anonymize(str(result), detections) return result def _secure_run(self, *args, **kwargs): 子类需要实现的安全执行逻辑 raise NotImplementedError class CalculatorTool(SecureBaseTool): name: str calculator description: str 用于执行数学计算 class ArgsSchema(BaseModel): expression: str Field(description数学表达式如 22) def _secure_run(self, expression: str) - str: 安全执行计算 # 简单的表达式评估生产环境应使用更安全的评估方式 try: # 限制可用的数学操作 allowed_chars set(0123456789-*/.() ) if not all(c in allowed_chars for c in expression): return 表达式包含不允许的字符 result eval(expression) # 注意生产环境应使用更安全的方式 return f计算结果: {result} except Exception as e: return f计算错误: {str(e)}5.2 安全 Agent 的完整实现构建整合所有安全组件的 AI Agent# core/agents.py from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from core.security import PIIDetector, GuardrailsWrapper from tools.custom_tools import CalculatorTool class SecureAIAgent: def __init__(self, config): self.config config self.llm self._initialize_llm() self.tools self._initialize_tools() self.pii_detector PIIDetector() self.agent_executor self._create_agent() def _initialize_llm(self): 初始化安全的 LLM 实例 return ChatOpenAI( modelgpt-3.5-turbo, temperature0.1, # 降低随机性提高确定性 max_tokensself.config.max_response_length ) def _initialize_tools(self): 初始化安全工具集 return [ CalculatorTool() ] def _create_agent(self): 创建安全 Agent prompt ChatPromptTemplate.from_messages([ (system, 你是一个安全的AI助手。在回答问题时需要遵循以下规则 1. 不泄露任何个人信息 2. 不执行危险操作 3. 遇到不确定的问题时明确说明 4. 所有工具调用都需要经过安全检查 当前用户查询: {input}), (human, {input}) ]) agent create_tool_calling_agent(self.llm, self.tools, prompt) return AgentExecutor(agentagent, toolsself.tools, verboseTrue) def invoke(self, user_input: str) - str: 安全地执行用户查询 # 输入检查 input_detections self.pii_detector.detect(user_input) if input_detections: return 输入包含敏感信息请重新表述您的问题 # 执行 Agent try: response self.agent_executor.invoke({input: user_input}) output response.get(output, ) # 输出检查 output_detections self.pii_detector.detect(output) if output_detections: output self.pii_detector.anonymize(output, output_detections) return output except Exception as e: return f处理请求时出错: {str(e)}6. 测试验证与问题排查构建完整的安全防护体系后需要通过系统化测试验证其有效性。6.1 安全功能测试用例编写针对性的测试用例# tests/test_security.py import pytest from core.security import PIIDetector, SecureDocumentProcessor from langchain_core.documents import Document class TestPIIDetection: def setup_method(self): self.detector PIIDetector() def test_email_detection(self): text 请联系 testexample.com 获取更多信息 results self.detector.detect(text) assert len(results) 1 assert results[0].pii_type email def test_anonymization(self): text 我的电话是 13800138000邮箱是 userdomain.com results self.detector.detect(text) anonymized self.detector.anonymize(text, results) assert 13800138000 not in anonymized assert userdomain.com not in anonymized assert 138****8000 in anonymized class TestDocumentProcessing: def test_document_anonymization(self): detector PIIDetector() processor SecureDocumentProcessor(detector) doc Document( page_content用户信息张三电话 13800138000, metadata{source: test} ) processed processor.process_document(doc) assert 13800138000 not in processed.page_content assert processed.metadata[pii_detected] is True6.2 常见问题排查指南在实际部署中可能会遇到以下典型问题问题现象可能原因检查方式解决方案Guardrails 验证失败输出格式不符合预期检查 Rail Spec 定义调整提示词或放宽约束PII 检测误报正则表达式过于严格测试边缘案例优化检测规则Agent 执行超时工具调用过于复杂检查工具实现添加超时限制内存使用过高文档处理未分块监控内存使用优化文本分割策略6.3 性能优化建议安全防护可能带来性能开销以下优化策略值得考虑分层检测先进行快速规则检测再进行复杂的模型检测缓存机制对重复内容应用检测结果缓存异步处理将检测任务异步化避免阻塞主流程采样检查对低风险内容进行采样检查而非全量检查7. 生产环境部署建议将安全 AI Agent 部署到生产环境时还需要考虑以下关键因素7.1 监控与日志建立完善的可观测性体系# core/monitoring.py import logging from datetime import datetime class SecurityLogger: def __init__(self): self.logger logging.getLogger(security) def log_pii_detection(self, detection_results, context): 记录 PII 检测事件 for result in detection_results: self.logger.warning( fPII检测 - 类型:{result.pii_type} f上下文:{context} 时间:{datetime.now()} ) def log_guardrail_violation(self, violation_type, original_output): 记录护栏违规事件 self.logger.error( f护栏违规 - 类型:{violation_type} f原始输出:{original_output[:100]}... )7.2 安全配置清单部署前检查以下安全配置[ ] API 密钥是否正确配置且具有最小必要权限[ ] 网络访问限制是否到位防火墙、VPC 等[ ] 日志是否包含敏感信息是否已脱敏[ ] 错误信息是否不会泄露内部实现细节[ ] 速率限制是否可防止滥用[ ] 数据存储是否加密访问是否受控7.3 持续安全维护AI 安全不是一次性的工作需要持续维护定期更新及时更新 LangChain、Guardrails 等依赖版本规则优化根据实际使用情况调整 PII 检测规则威胁建模定期进行安全评估识别新的威胁场景审计跟踪保留足够的安全事件日志供审计使用构建安全可控的 AI Agent 是一个系统工程需要在功能实现之初就考虑安全防护。通过 LangChain 的分层架构设计结合 Guardrails 的验证机制和自定义的 PII 检测可以建立纵深防御体系。实际项目中还需要根据具体业务需求调整安全策略在安全性和用户体验之间找到合适的平衡点。最重要的是建立安全第一的开发文化每个新功能都要经过安全评估每次迭代都要考虑安全影响。只有这样才能构建出真正可靠、可信的 AI 应用系统。