在AI应用开发中如何让大语言模型LLM不仅“聪明”地完成任务还能在安全、可控的边界内运行是每个开发者从Demo走向生产环境时必须面对的挑战。你是否遇到过Agent自行调用危险工具、生成不当内容或陷入死循环的情况本文将基于LangChain 1.3手把手带你构建一个集成了**Guardrail安全护栏和HITL人在回路**机制的安全可控AI Agent。从核心概念到项目实战你将掌握一套让AI既强大又“听话”的工程化方案。1. 背景与核心概念为什么AI Agent需要安全护栏在深入代码之前我们首先要厘清几个关键概念理解为什么单纯依赖大模型的“自由发挥”存在风险。AI Agent通常指能够感知环境、进行决策并执行行动以达成目标的智能体。在LangChain的语境下一个典型的Agent由大语言模型LLM、**工具Tools和记忆Memory**等核心组件构成。LLM作为“大脑”负责规划和推理工具则是其可以调用的“手脚”。然而一个不受约束的Agent可能带来以下问题工具滥用例如一个拥有文件读写工具的Agent可能被诱导删除系统关键文件。内容安全风险生成包含偏见、虚假信息或不当言论的内容。资源消耗失控陷入无意义的循环调用消耗大量API Token和计算资源。决策不可解释做出关键决策时其推理过程对开发者而言是个黑箱。为了解决这些问题我们需要引入两个核心机制Guardrail安全护栏这是一套在Agent执行动作之前、之中或之后进行干预的规则和检查系统。它像交通规则一样为Agent的行为设定边界。例如在调用工具前检查参数是否安全在生成最终答案前过滤敏感词。HITLHuman-in-the-Loop人在回路指在AI系统的关键决策点引入人工审核或确认。当Agent遇到高不确定性、高风险操作或超出其权限的任务时会暂停执行并向人类用户请求指导。这结合了AI的效率与人类的判断力是实现高可靠性系统的有效手段。LangChain作为一个强大的AI应用开发框架提供了丰富的底层抽象和接口使得集成Guardrail和HITL变得可行且灵活。本次实战将聚焦于LangChain 1.3版本展示如何利用其新特性与现有组件构建安全体系。2. 环境准备与版本说明本实战项目基于Python环境使用LangChain作为核心框架。请确保你的环境满足以下要求操作系统Windows 10/11, macOS 或 Linux (Ubuntu 20.04)Python版本3.8 或 3.9推荐3.9兼容性最佳核心框架LangChain 1.3.xLLM接入OpenAI API (或兼容OpenAI API的本地模型)其他工具需要用到requests库进行网络调用pydantic用于数据验证。项目初始化与依赖安装创建并进入项目目录mkdir safe_ai_agent cd safe_ai_agent创建虚拟环境推荐python -m venv venv # Windows激活 venv\Scripts\activate # macOS/Linux激活 source venv/bin/activate安装核心依赖。我们使用pip安装指定版本的LangChain及其他必要库。pip install langchain0.1.3 openai requests pydantic注意LangChain版本迭代较快0.1.3是1.x系列的一个稳定版本。请根据官方文档和你的实际需求调整版本号。设置OpenAI API密钥如果你使用OpenAI的模型。你可以将其设置为环境变量或在代码中直接配置不推荐将密钥硬编码在代码中。# 在Linux/macOS的终端中 export OPENAI_API_KEYyour-api-key-here # 在Windows的PowerShell中 $env:OPENAI_API_KEYyour-api-key-here3. 核心组件拆解LangChain中的Agent、Tool与Guardrail3.1 Agent与Tool的基础工作流在LangChain中一个简单的Agent工作流如下用户输入问题input。Agent基于LLM分析问题决定是否需要使用工具以及使用哪个工具。如果使用工具Agent会生成工具调用请求包含工具名和参数。框架执行具体的工具函数。工具返回结果给Agent。Agent根据工具结果进行下一步思考可能继续调用工具或生成最终答案给用户。下面是一个最简单的自定义工具和Agent创建示例# file: basic_agent.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate # 1. 定义自定义工具 def get_weather(city: str) - str: 根据城市名获取天气信息。这是一个模拟函数。 # 这里模拟一个API调用 weather_data { 北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C } return weather_data.get(city, f未找到{city}的天气信息) # 将函数包装成LangChain Tool对象 weather_tool Tool( nameGetWeather, funcget_weather, description当用户询问某个城市的天气时使用此工具。输入应为城市名称。 ) # 2. 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 创建Agent提示词模板 prompt PromptTemplate.from_template( 你是一个有帮助的助手。你可以使用以下工具 {tools} 请严格按照以下格式回答 问题用户的输入 思考你需要思考做什么 行动要使用的工具名 行动输入工具的输入 观察工具返回的结果 ... (这个思考/行动/观察循环可以重复多次) 最终答案根据观察得出的最终答案 开始 问题{input} 思考{agent_scratchpad} ) # 4. 创建Agent并执行 tools [weather_tool] agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 5. 运行Agent if __name__ __main__: result agent_executor.invoke({input: 上海今天天气怎么样}) print(\n--- 最终结果 ---) print(result[output])运行上述代码你会看到Agent的完整思考链Chain of Thought它决定调用GetWeather工具并成功返回了模拟的天气信息。这是无护栏状态下的基础Agent。3.2 构建Guardrail输入/输出验证与工具调用拦截Guardrail可以在多个层面实施。我们首先构建一个工具调用拦截器在Agent决定调用工具时进行检查。# file: guardrail_agent.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.schema import AgentAction, AgentFinish from typing import List, Union, Tuple, Any import re # --- 自定义工具定义 (同上) --- def get_weather(city: str) - str: weather_data {北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C} return weather_data.get(city, f未找到{city}的天气信息) weather_tool Tool(nameGetWeather, funcget_weather, description获取城市天气。输入应为城市名称。) # --- 安全规则定义 --- class SecurityGuardrail: 一个简单的安全护栏类用于检查工具调用和输入/输出。 forbidden_cities [华盛顿, 莫斯科] # 示例禁止查询某些敏感城市的天气 suspicious_patterns [r删除, rrm -rf, rformat] # 可疑命令模式 classmethod def check_tool_call(cls, tool_name: str, tool_input: str) - Tuple[bool, str]: 检查工具调用是否被允许。返回 (是否允许, 拒绝原因) # 规则1检查工具名是否合法 allowed_tools [GetWeather] if tool_name not in allowed_tools: return False, f工具 {tool_name} 不在允许列表内。 # 规则2针对特定工具的输入检查 if tool_name GetWeather: # 检查是否查询了禁止的城市 for city in cls.forbidden_cities: if city in tool_input: return False, f不允许查询城市 {city} 的天气信息。 # 检查输入是否仅为城市名简单的正则检查 if not re.match(r^[\u4e00-\u9fa5A-Za-z\s]$, tool_input.strip()): return False, f工具输入 {tool_input} 格式可疑应仅为城市名称。 # 规则3通用输入内容安全检查 for pattern in cls.suspicious_patterns: if re.search(pattern, tool_input, re.IGNORECASE): return False, f工具输入包含可疑指令: {pattern} return True, classmethod def check_output(cls, output: str) - Tuple[bool, str]: 检查Agent最终输出是否安全。 # 示例检查输出中是否包含敏感词 sensitive_words [机密, 攻击, 漏洞] for word in sensitive_words: if word in output: # 可以选择替换或直接拒绝 return False, f输出中包含敏感词 {word} return True, # --- 自定义AgentExecutor集成Guardrail --- class GuardedAgentExecutor(AgentExecutor): 重写AgentExecutor在工具调用前加入安全检查。 def _call_tool(self, tool_name: str, tool_input: str) - str: 重写工具调用方法加入前置检查。 print(f[Guardrail] 检查工具调用: {tool_name}({tool_input})) # 调用安全护栏进行检查 is_allowed, reason SecurityGuardrail.check_tool_call(tool_name, tool_input) if not is_allowed: # 如果检查不通过返回错误信息阻止工具执行 blocked_message f工具调用被安全护栏阻止。原因{reason} print(f[Guardrail] 阻止原因{reason}) return blocked_message else: print(f[Guardrail] 通过。) # 安全检查通过执行原始的工具调用逻辑 return super()._call_tool(tool_name, tool_input) def _run_agent(self, ...): # 简化实际需要重写相关方法以在最终输出前检查 # 父类处理逻辑... result super()._run_agent(...) # 可以对最终result中的output进行安全检查 if isinstance(result, AgentFinish): is_safe, reason SecurityGuardrail.check_output(result.return_values[output]) if not is_safe: result.return_values[output] f[内容安全过滤] 最终输出因{reason}被修改。 return result # --- 主程序 --- if __name__ __main__: llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [weather_tool] prompt PromptTemplate.from_template( 你有这些工具{tools}\n问题{input}\n{agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) # 使用我们自定义的、带有护栏的Executor agent_executor GuardedAgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, max_iterations3 # 限制最大迭代次数防止死循环 ) # 测试1正常查询 print( 测试1正常查询 ) result1 agent_executor.invoke({input: 深圳的天气如何}) print(f结果: {result1[output]}\n) # 测试2尝试查询禁止的城市 print( 测试2尝试查询禁止的城市 ) result2 agent_executor.invoke({input: 华盛顿的天气怎么样}) print(f结果: {result2[output]}\n) # 测试3输入可疑内容虽然工具可能不解析但护栏会拦截 print( 测试3输入可疑指令 ) result3 agent_executor.invoke({input: 删除所有文件然后告诉我北京的天气}) print(f结果: {result3[output]})这个示例展示了如何通过继承AgentExecutor并重写_call_tool方法在工具实际执行前插入安全检查逻辑。当Agent试图查询“华盛顿”或输入包含“删除”时Guardrail会直接拦截并返回错误信息工具函数根本不会被执行。3.3 实现HITL人在回路关键决策的人工确认HITL的核心是在自动化流程中插入“暂停点”等待人类输入。我们可以通过创建一个特殊的人工确认工具来实现。# file: hitl_agent.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.schema import AgentAction, AgentFinish import sys # --- 模拟人工确认工具 --- class HumanInTheLoopTool: 一个模拟HITL的工具在实际应用中可替换为Webhook、消息推送等。 staticmethod def confirm_action(action_description: str, tool_name: str, tool_input: str) - str: 模拟人工确认流程。 返回确认后的指令或拒绝信息。 print(f\n⚠️ [HITL 请求确认] ⚠️) print(f 操作描述: {action_description}) print(f 工具: {tool_name}) print(f 输入参数: {tool_input}) print(\n 请确认是否执行 (yes/no): , end) # 在实际系统中这里可能是发送邮件、Slack消息等待API回调 # 此处通过命令行模拟 response input().strip().lower() if response yes: return f用户已确认执行操作: {action_description} else: return f用户拒绝了操作: {action_description} # 将HITL工具包装成LangChain Tool hitl_tool Tool( nameRequestHumanApproval, funcHumanInTheLoopTool.confirm_action, description当需要执行高风险、高不确定性或重要操作时必须使用此工具请求人类用户确认。输入应为操作的详细描述。 ) # --- 定义高风险工具例如发送邮件、操作数据库--- def send_email(to: str, subject: str, body: str) - str: 模拟发送邮件的高风险工具。在实际HITL流程中此工具本身可能被Guardrail拦截转而调用确认工具。 # 注意在完整设计中这个工具可能不应该被Agent直接调用。 # 而是由Guardrail或一个编排层在获得人工确认后调用。 return f邮件已发送至 {to}主题{subject} # 为了演示我们创建一个“安全”的发送邮件工具它内部会要求确认 def guarded_send_email(to: str, subject: str, body: str) - str: 一个集成了HITL的发送邮件工具。 print(f\n[Guardrail] 检测到高风险操作 send_email。) confirmation HumanInTheLoopTool.confirm_action( action_descriptionf发送邮件给 {to}主题为 {subject}, tool_namesend_email, tool_inputfto{to}, subject{subject} ) if 用户已确认 in confirmation: # 模拟执行发送 return f执行成功邮件已发送至 {to}。 else: return f操作取消{confirmation} email_tool Tool( nameSendEmail, funcguarded_send_email, # 直接使用集成了HITL的函数 description发送电子邮件。这是一个高风险操作需要内部确认。 ) # --- 主程序构建一个知道在必要时使用HITL的Agent --- if __name__ __main__: llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 注意我们不再将HITL工具直接暴露给Agent而是通过高风险工具的内部控制实现。 # 或者我们可以设计一个更复杂的Agent让它自己决定何时调用RequestHumanApproval。 tools [email_tool] # 只暴露高风险工具其内部已含HITL # 设计一个提示词让Agent知道某些操作需要特别谨慎 prompt PromptTemplate.from_template( 你是一个助理可以帮用户处理事务包括发送邮件。 注意发送邮件是一个重要操作需要谨慎确认。 用户问题{input} 请开始思考并逐步解决问题。你可以使用以下工具 {tools} 请严格按照格式响应 思考... 行动SendEmail 行动输入{{to: 收件人邮箱, subject: 邮件主题, body: 邮件正文}} 观察... ...重复思考/行动/观察 最终答案... 开始 {agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, max_iterations2) print( 测试发送邮件的HITL流程 ) # 当Agent调用SendEmail工具时会触发我们集成的guarded_send_email函数该函数会请求人工确认。 result agent_executor.invoke({ input: 帮我给同事张三zhangsancompany.com发一封邮件主题是‘项目会议纪要’正文问问他下周时间。 }) print(f\n最终结果: {result[output]})在这个HITL示例中我们没有让LLM主动决定何时“请示人类”而是将确认逻辑内置到了高风险工具的实现中。这是一种更可靠、更易于控制的模式因为你不必完全相信LLM的判断力来决定何时需要人工介入。当SendEmail工具被调用时程序会自动暂停并在控制台请求确认。4. 完整实战案例构建一个安全可控的客户支持AI Agent现在我们将前面所学组合起来构建一个模拟的客户支持AI Agent。它能够查询知识库模拟、提交工单高风险操作需HITL并且所有操作都在Guardrail的监控之下。4.1 项目结构safe_customer_agent/ ├── main.py # 主程序入口 ├── guardrails.py # 安全护栏逻辑 ├── tools.py # 所有工具定义 ├── hitl_module.py # 人在回路模块 └── config.py # 配置文件如API密钥4.2 实现核心模块1. 安全护栏 (guardrails.py)# guardrails.py import re from typing import Tuple, List from pydantic import BaseModel, validator class ToolCallRequest(BaseModel): 工具调用请求的数据模型用于验证 tool_name: str tool_input: dict validator(tool_name) def name_must_be_allowed(cls, v): allowed [SearchKnowledgeBase, SubmitSupportTicket, GetSystemTime] if v not in allowed: raise ValueError(f禁止调用工具: {v}) return v validator(tool_input, each_itemTrue) # 简化验证 def check_input_content(cls, v, values): if isinstance(v, str): # 检查SQL注入等常见攻击模式简单示例 sql_patterns [r(\%27)|(\)|(\-\-)|(\%23)|(#), r(\s*((\bOR\b)|(\bAND\b))\s*[\w\s])] for pattern in sql_patterns: if re.search(pattern, v, re.IGNORECASE): raise ValueError(f输入包含可疑SQL模式: {pattern}) # 检查敏感词 sensitive_terms [密码, token, 密钥, delete from, drop table] for term in sensitive_terms: if term.lower() in v.lower(): raise ValueError(f输入包含敏感词: {term}) return v class ContentGuardrail: 内容安全护栏 staticmethod def sanitize_output(text: str) - str: 对输出内容进行清理和过滤。 # 过滤不友好词汇示例列表 offensive_words [笨蛋, 蠢货] for word in offensive_words: text text.replace(word, **) # 脱敏邮箱/手机号简单正则 text re.sub(r\b[\w\.-][\w\.-]\.\w\b, [邮箱已隐藏], text) text re.sub(r\b1[3-9]\d{9}\b, [手机号已隐藏], text) return text2. 人在回路模块 (hitl_module.py)# hitl_module.py import json from datetime import datetime from typing import Optional class TicketApprovalSystem: 模拟的工单审批系统用于HITL。 pending_tickets [] classmethod def request_approval(cls, ticket_details: dict) - dict: 提交工单审批请求返回工单ID和状态。 ticket_id fTICKET-{datetime.now().strftime(%Y%m%d%H%M%S)} ticket { id: ticket_id, details: ticket_details, status: pending, created_at: datetime.now().isoformat() } cls.pending_tickets.append(ticket) print(f\n [HITL] 新工单待审批 ID: {ticket_id}) print(f 详情: {json.dumps(ticket_details, ensure_asciiFalse, indent2)}) print( 请前往管理后台审批。) # 在实际系统中这里会触发邮件、短信或即时通讯通知 return {ticket_id: ticket_id, status: pending_approval, message: 工单已提交等待人工审批。} classmethod def check_status(cls, ticket_id: str) - dict: 检查工单状态模拟。 for ticket in cls.pending_tickets: if ticket[id] ticket_id: # 模拟假设第一个工单被自动批准其他待定 if ticket_id.endswith(01): # 简单模拟逻辑 ticket[status] approved return {status: approved, message: 工单已批准正在处理。} else: return {status: ticket[status], message: 工单仍在等待审批。} return {status: not_found, message: 未找到该工单。}3. 工具定义 (tools.py)# tools.py from langchain.tools import Tool from .hitl_module import TicketApprovalSystem from .guardrails import ContentGuardrail import random def search_knowledge_base(query: str) - str: 模拟搜索知识库。 # 模拟一个简单的知识库 kb { 退款政策: 商品签收后7天内可申请退款需保持商品完好。, 物流时间: 国内一般3-5个工作日送达偏远地区可能延迟。, 账号注册: 使用手机号或邮箱即可免费注册。 } # 简单关键词匹配 for key, answer in kb.items(): if key in query: return answer return 未找到相关问题答案请尝试其他关键词或提交工单。 def submit_support_ticket(issue: str, customer_id: str) - str: 提交支持工单。这是一个高风险操作需要触发HITL流程。 注意此工具不直接解决问题而是提交审批请求。 if not customer_id or customer_id unknown: return 错误需要有效的客户ID才能提交工单。 # 1. 内容过滤 sanitized_issue ContentGuardrail.sanitize_output(issue) # 2. 构建工单详情并请求人工审批 ticket_details { customer_id: customer_id, issue: sanitized_issue, priority: normal, category: general } result TicketApprovalSystem.request_approval(ticket_details) return f工单提交结果{result[message]} 工单ID: {result[ticket_id]}。请使用检查工单状态工具跟进。 def check_ticket_status(ticket_id: str) - str: 检查工单审批状态。 result TicketApprovalSystem.check_status(ticket_id) return f工单状态{result[status]}。详情{result[message]} def get_system_time() - str: 获取系统时间一个无害工具用于测试。 from datetime import datetime return f当前系统时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)} # 包装成LangChain Tools knowledge_tool Tool( nameSearchKnowledgeBase, funcsearch_knowledge_base, description用于在内部知识库中搜索常见问题的答案。输入应为查询关键词。 ) ticket_tool Tool( nameSubmitSupportTicket, funcsubmit_support_ticket, description当知识库无法解决问题时用于提交客户支持工单。输入必须包含issue问题描述和customer_id客户ID。 ) status_tool Tool( nameCheckTicketStatus, funccheck_ticket_status, description根据工单ID检查已提交工单的审批和处理状态。输入应为工单ID字符串。 ) time_tool Tool( nameGetSystemTime, funcget_system_time, description获取当前的系统日期和时间。 ) ALL_TOOLS [knowledge_tool, ticket_tool, status_tool, time_tool]4. 主程序与安全Agent执行器 (main.py)# main.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from tools import ALL_TOOLS from guardrails import ToolCallRequest, ContentGuardrail import os class SafeAgentExecutor(AgentExecutor): 集成了输入验证和输出过滤的安全Agent执行器。 def _take_next_step(self, name_to_tool_map, color_mapping, inputs, intermediate_steps): 重写步骤执行逻辑加入工具调用前的结构化验证。 # 让父类先产生下一步的动作决策思考、行动 next_step_output super()._take_next_step(name_to_tool_map, color_mapping, inputs, intermediate_steps) # 如果下一步是工具调用AgentAction则进行验证 from langchain.schema import AgentAction if isinstance(next_step_output, list) and len(next_step_output) 1: action next_step_output[0] if isinstance(action, AgentAction): print(f\n[Guardrail] 验证工具调用: {action.tool}({action.tool_input})) try: # 使用Pydantic模型验证工具名和输入 validated_request ToolCallRequest( tool_nameaction.tool, tool_inputaction.tool_input if isinstance(action.tool_input, dict) else {input: action.tool_input} ) print(f[Guardrail] 验证通过。) except ValueError as e: # 验证失败阻止调用返回错误信息 error_msg f安全验证失败{e} print(f[Guardrail] 拦截原因{error_msg}) # 返回一个代表被拦截的观察结果让Agent重新思考 from langchain.schema import AgentFinish return [AgentFinish({output: error_msg}, logerror_msg)] return next_step_output def _call(self, inputs): 重写最终调用对输出进行内容过滤。 result super()._call(inputs) # 对最终输出进行安全过滤 if output in result: result[output] ContentGuardrail.sanitize_output(result[output]) return result def main(): # 0. 配置 os.environ[OPENAI_API_KEY] your-api-key # 请替换为你的密钥或使用环境变量 # 1. 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, max_tokens500) # 2. 构建提示词 prompt_template 你是一个专业的客户支持AI助手。请遵循以下规则 1. 首先尝试使用知识库工具解答用户问题。 2. 如果知识库没有答案再考虑使用提交工单工具。 3. 提交工单需要明确的客户ID和问题描述。 4. 你可以使用检查工单状态工具来跟进。 5. 所有操作都必须遵守安全规定。 工具 {tools} 用户问题{input} 请严格按照以下格式回应 思考首先分析用户问题属于哪一类是否需要工具。 行动工具名 行动输入工具的输入必须是JSON格式 观察工具返回的结果 ...这个循环可以重复 最终答案总结并回复用户 开始 {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template) # 3. 创建Agent agent create_react_agent(llm, ALL_TOOLS, prompt) # 4. 使用自定义的安全执行器 agent_executor SafeAgentExecutor( agentagent, toolsALL_TOOLS, verboseTrue, # 打印详细思考过程 max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate, # 当连续两个动作都是“最终答案”时停止 handle_parsing_errorsTrue # 优雅处理解析错误 ) # 5. 运行测试 test_cases [ 你们的退款政策是什么, 我的订单号是12345物流一直没更新怎么办, 我想删除我的账户里的所有数据。, # 测试敏感词触发Guardrail 帮我提交一个工单我的问题是‘账号无法登录提示密码错误’我的客户ID是‘cust_001’。, 检查一下工单 TICKET-20231001010101 的状态。 ] for i, question in enumerate(test_cases): print(f\n{*50}) print(f测试用例 {i1}: {question}) print(f{*50}) try: result agent_executor.invoke({input: question}) print(f\n✅ 最终回复: {result[output]}) except Exception as e: print(f\n❌ 执行出错: {e}) if __name__ __main__: main()4.3 运行与验证将上述四个文件放在同一目录下。确保已安装所有依赖 (langchain,openai,pydantic)。在main.py中设置你的OpenAI API密钥。运行程序python main.py预期输出程序会依次处理5个测试用例。你会看到用例12Agent成功调用SearchKnowledgeBase工具找到答案。用例3当用户输入包含“删除”等敏感词时ToolCallRequest验证器会抛出ValueErrorGuardrail拦截此次工具调用Agent会收到错误观察并可能直接给出最终答案告知用户无法执行该操作。用例4Agent调用SubmitSupportTicket工具该工具内部会触发HITL流程打印出待审批的工单信息并返回“等待人工审批”的结果。用例5Agent调用CheckTicketStatus工具查询模拟的工单状态。在整个过程中SafeAgentExecutor负责验证每次工具调用的合法性并在最终输出前过滤敏感信息如邮箱、手机号。5. 常见问题与排查思路在开发和运行此类安全可控的AI Agent时你可能会遇到以下问题问题现象可能原因排查思路与解决方案Agent无法正确选择工具1. 工具描述 (description) 不清晰。2. LLM的temperature参数过高导致输出不稳定。3. 提示词 (prompt) 未明确指导工具使用。1. 优化工具描述确保准确、简洁包含关键输入格式。2. 将temperature调低如0.1使输出更确定。3. 在提示词中加入工具使用规则和格式示例。Guardrail误拦截合法请求安全规则如正则表达式、关键词列表过于严格或匹配不精确。1. 详细记录拦截日志分析误判案例。2. 优化规则使用更精确的匹配模式如完整单词匹配、上下文判断。3. 考虑引入基于机器学习模型的分类器替代硬规则。HITL流程导致Agent响应超时等待人工确认的环节阻塞了整个同步流程。1. 将HITL设计为异步流程。Agent提交请求后立即返回“已提交请等待”信息后续通过回调或让用户主动查询状态。2. 设置超时机制超时后自动取消或按默认策略处理。Agent陷入思考循环死循环1. 工具返回的结果无法让Agent做出决策。2.max_iterations设置过高。1. 检查工具返回格式是否清晰。确保在无法处理时返回明确的错误信息。2.务必设置max_iterations参数如3-5次这是防止死循环的关键。3. 在提示词中强调“如果尝试X次后仍无法解决就告知用户并停止”。解析错误 (ParsingError)Agent输出的文本不符合LangChain预设的解析格式如ReAct格式。1. 启用handle_parsing_errorsTrue参数让执行器能优雅处理错误。2. 使用更强大的LLM如GPT-4来提高输出格式的稳定性。3. 使用OutputFixingParser等LangChain内置组件自动修复格式。工具调用速度慢1. LLM API调用延迟。2. 工具函数本身执行慢如网络请求。3. Guardrail检查逻辑复杂。1. 为LLM调用设置合理的超时时间。2. 对慢速工具进行异步化或缓存优化。3. 优化Guardrail检查算法避免复杂计算。考虑将部分检查后置。安全规则难以维护规则散落在代码各处随着工具增多变得混乱。1.集中化管理安全策略例如使用配置文件或策略数据库。2. 采用策略模式为不同类型的工具定义不同的安全检查类。3. 考虑使用专门的策略引擎如OpenPolicyAgent。6. 最佳实践与工程建议将安全可控的AI Agent投入生产环境需要遵循以下工程最佳实践分层防御策略输入层对用户原始输入进行清洗和标准化过滤明显恶意内容。意图层在Agent思考前先用一个轻量级分类模型判断用户意图是否在服务范围内。规划层在Agent生成行动规划后用Guardrail验证其步骤的合理性与安全性。执行层在调用具体工具前再次验证参数。工具函数内部也应包含其自身的业务逻辑校验。输出层对最终返回给用户的内容进行过滤和脱敏。HITL的智能触发不要所有操作都请求人工确认那样会失去自动化的意义。应根据风险等级和置信度动态触发HITL。风险等级由操作类型决定如“读”操作风险低“写”或“删”操作风险高。置信度可以由LLM自身输出的思考链中提取“信心分数”或通过一个验证模型对Agent的决策进行评分。低置信度高风险的组合才触发HITL。可观测性与审计完整日志记录记录每一次用户输入、Agent的思考过程、工具调用请求含参数、Guardrail检查结果、工具执行结果、最终输出。这些日志对于排查问题、优化规则和事后审计至关重要。链路追踪为每个用户会话分配唯一ID串联起所有相关日志。监控告警对高频失败的工具调用、频繁触发的Guardrail拦截、长时间运行的Agent会话设置监控和告警。测试与评估单元测试为每个工具函数、Guardrail规则编写单元测试。集成测试模拟端到端的用户会话测试Agent的整体流程。对抗测试设计包含模糊、诱导、越权指令的测试用例评估Agent的鲁棒性和安全性。红队演练定期进行安全演练尝试找出系统的漏洞。配置化与版本管理将Guardrail规则、工具列表、提示词模板等尽可能配置化而不是硬编码在代码中。这便于动态调整、A/B测试和回滚。对Agent的配置特别是提示词进行版本管理任何变更都应有记录并能快速回退到稳定版本。性能与成本优化缓存对频繁查询的、结果不变的知识库内容进行缓存减少LLM调用和工具调用。限制对每个用户/会话设置LLM Token消耗上限、工具调用次数上限防止恶意消耗资源。轻量级模型在Guardrail的某些检查环节如意图分类、敏感词检测可以考虑使用更小、更快的本地模型而不是每次都调用大模型。构建安全可控的AI Agent是一个持续迭代的过程没有一劳永逸的方案。核心思想是不信任原则不信任用户输入不信任LLM的未经检查的输出不信任工具的无约束执行。通过Guardrail设定明确的边界通过HITL在关键节点引入人类智慧再辅以完善的监控和测试你才能打造出既强大又可靠的AI应用。
基于LangChain构建安全可控AI Agent：Guardrail与HITL实战指南
在AI应用开发中如何让大语言模型LLM不仅“聪明”地完成任务还能在安全、可控的边界内运行是每个开发者从Demo走向生产环境时必须面对的挑战。你是否遇到过Agent自行调用危险工具、生成不当内容或陷入死循环的情况本文将基于LangChain 1.3手把手带你构建一个集成了**Guardrail安全护栏和HITL人在回路**机制的安全可控AI Agent。从核心概念到项目实战你将掌握一套让AI既强大又“听话”的工程化方案。1. 背景与核心概念为什么AI Agent需要安全护栏在深入代码之前我们首先要厘清几个关键概念理解为什么单纯依赖大模型的“自由发挥”存在风险。AI Agent通常指能够感知环境、进行决策并执行行动以达成目标的智能体。在LangChain的语境下一个典型的Agent由大语言模型LLM、**工具Tools和记忆Memory**等核心组件构成。LLM作为“大脑”负责规划和推理工具则是其可以调用的“手脚”。然而一个不受约束的Agent可能带来以下问题工具滥用例如一个拥有文件读写工具的Agent可能被诱导删除系统关键文件。内容安全风险生成包含偏见、虚假信息或不当言论的内容。资源消耗失控陷入无意义的循环调用消耗大量API Token和计算资源。决策不可解释做出关键决策时其推理过程对开发者而言是个黑箱。为了解决这些问题我们需要引入两个核心机制Guardrail安全护栏这是一套在Agent执行动作之前、之中或之后进行干预的规则和检查系统。它像交通规则一样为Agent的行为设定边界。例如在调用工具前检查参数是否安全在生成最终答案前过滤敏感词。HITLHuman-in-the-Loop人在回路指在AI系统的关键决策点引入人工审核或确认。当Agent遇到高不确定性、高风险操作或超出其权限的任务时会暂停执行并向人类用户请求指导。这结合了AI的效率与人类的判断力是实现高可靠性系统的有效手段。LangChain作为一个强大的AI应用开发框架提供了丰富的底层抽象和接口使得集成Guardrail和HITL变得可行且灵活。本次实战将聚焦于LangChain 1.3版本展示如何利用其新特性与现有组件构建安全体系。2. 环境准备与版本说明本实战项目基于Python环境使用LangChain作为核心框架。请确保你的环境满足以下要求操作系统Windows 10/11, macOS 或 Linux (Ubuntu 20.04)Python版本3.8 或 3.9推荐3.9兼容性最佳核心框架LangChain 1.3.xLLM接入OpenAI API (或兼容OpenAI API的本地模型)其他工具需要用到requests库进行网络调用pydantic用于数据验证。项目初始化与依赖安装创建并进入项目目录mkdir safe_ai_agent cd safe_ai_agent创建虚拟环境推荐python -m venv venv # Windows激活 venv\Scripts\activate # macOS/Linux激活 source venv/bin/activate安装核心依赖。我们使用pip安装指定版本的LangChain及其他必要库。pip install langchain0.1.3 openai requests pydantic注意LangChain版本迭代较快0.1.3是1.x系列的一个稳定版本。请根据官方文档和你的实际需求调整版本号。设置OpenAI API密钥如果你使用OpenAI的模型。你可以将其设置为环境变量或在代码中直接配置不推荐将密钥硬编码在代码中。# 在Linux/macOS的终端中 export OPENAI_API_KEYyour-api-key-here # 在Windows的PowerShell中 $env:OPENAI_API_KEYyour-api-key-here3. 核心组件拆解LangChain中的Agent、Tool与Guardrail3.1 Agent与Tool的基础工作流在LangChain中一个简单的Agent工作流如下用户输入问题input。Agent基于LLM分析问题决定是否需要使用工具以及使用哪个工具。如果使用工具Agent会生成工具调用请求包含工具名和参数。框架执行具体的工具函数。工具返回结果给Agent。Agent根据工具结果进行下一步思考可能继续调用工具或生成最终答案给用户。下面是一个最简单的自定义工具和Agent创建示例# file: basic_agent.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate # 1. 定义自定义工具 def get_weather(city: str) - str: 根据城市名获取天气信息。这是一个模拟函数。 # 这里模拟一个API调用 weather_data { 北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C } return weather_data.get(city, f未找到{city}的天气信息) # 将函数包装成LangChain Tool对象 weather_tool Tool( nameGetWeather, funcget_weather, description当用户询问某个城市的天气时使用此工具。输入应为城市名称。 ) # 2. 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 创建Agent提示词模板 prompt PromptTemplate.from_template( 你是一个有帮助的助手。你可以使用以下工具 {tools} 请严格按照以下格式回答 问题用户的输入 思考你需要思考做什么 行动要使用的工具名 行动输入工具的输入 观察工具返回的结果 ... (这个思考/行动/观察循环可以重复多次) 最终答案根据观察得出的最终答案 开始 问题{input} 思考{agent_scratchpad} ) # 4. 创建Agent并执行 tools [weather_tool] agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 5. 运行Agent if __name__ __main__: result agent_executor.invoke({input: 上海今天天气怎么样}) print(\n--- 最终结果 ---) print(result[output])运行上述代码你会看到Agent的完整思考链Chain of Thought它决定调用GetWeather工具并成功返回了模拟的天气信息。这是无护栏状态下的基础Agent。3.2 构建Guardrail输入/输出验证与工具调用拦截Guardrail可以在多个层面实施。我们首先构建一个工具调用拦截器在Agent决定调用工具时进行检查。# file: guardrail_agent.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.schema import AgentAction, AgentFinish from typing import List, Union, Tuple, Any import re # --- 自定义工具定义 (同上) --- def get_weather(city: str) - str: weather_data {北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C} return weather_data.get(city, f未找到{city}的天气信息) weather_tool Tool(nameGetWeather, funcget_weather, description获取城市天气。输入应为城市名称。) # --- 安全规则定义 --- class SecurityGuardrail: 一个简单的安全护栏类用于检查工具调用和输入/输出。 forbidden_cities [华盛顿, 莫斯科] # 示例禁止查询某些敏感城市的天气 suspicious_patterns [r删除, rrm -rf, rformat] # 可疑命令模式 classmethod def check_tool_call(cls, tool_name: str, tool_input: str) - Tuple[bool, str]: 检查工具调用是否被允许。返回 (是否允许, 拒绝原因) # 规则1检查工具名是否合法 allowed_tools [GetWeather] if tool_name not in allowed_tools: return False, f工具 {tool_name} 不在允许列表内。 # 规则2针对特定工具的输入检查 if tool_name GetWeather: # 检查是否查询了禁止的城市 for city in cls.forbidden_cities: if city in tool_input: return False, f不允许查询城市 {city} 的天气信息。 # 检查输入是否仅为城市名简单的正则检查 if not re.match(r^[\u4e00-\u9fa5A-Za-z\s]$, tool_input.strip()): return False, f工具输入 {tool_input} 格式可疑应仅为城市名称。 # 规则3通用输入内容安全检查 for pattern in cls.suspicious_patterns: if re.search(pattern, tool_input, re.IGNORECASE): return False, f工具输入包含可疑指令: {pattern} return True, classmethod def check_output(cls, output: str) - Tuple[bool, str]: 检查Agent最终输出是否安全。 # 示例检查输出中是否包含敏感词 sensitive_words [机密, 攻击, 漏洞] for word in sensitive_words: if word in output: # 可以选择替换或直接拒绝 return False, f输出中包含敏感词 {word} return True, # --- 自定义AgentExecutor集成Guardrail --- class GuardedAgentExecutor(AgentExecutor): 重写AgentExecutor在工具调用前加入安全检查。 def _call_tool(self, tool_name: str, tool_input: str) - str: 重写工具调用方法加入前置检查。 print(f[Guardrail] 检查工具调用: {tool_name}({tool_input})) # 调用安全护栏进行检查 is_allowed, reason SecurityGuardrail.check_tool_call(tool_name, tool_input) if not is_allowed: # 如果检查不通过返回错误信息阻止工具执行 blocked_message f工具调用被安全护栏阻止。原因{reason} print(f[Guardrail] 阻止原因{reason}) return blocked_message else: print(f[Guardrail] 通过。) # 安全检查通过执行原始的工具调用逻辑 return super()._call_tool(tool_name, tool_input) def _run_agent(self, ...): # 简化实际需要重写相关方法以在最终输出前检查 # 父类处理逻辑... result super()._run_agent(...) # 可以对最终result中的output进行安全检查 if isinstance(result, AgentFinish): is_safe, reason SecurityGuardrail.check_output(result.return_values[output]) if not is_safe: result.return_values[output] f[内容安全过滤] 最终输出因{reason}被修改。 return result # --- 主程序 --- if __name__ __main__: llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [weather_tool] prompt PromptTemplate.from_template( 你有这些工具{tools}\n问题{input}\n{agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) # 使用我们自定义的、带有护栏的Executor agent_executor GuardedAgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, max_iterations3 # 限制最大迭代次数防止死循环 ) # 测试1正常查询 print( 测试1正常查询 ) result1 agent_executor.invoke({input: 深圳的天气如何}) print(f结果: {result1[output]}\n) # 测试2尝试查询禁止的城市 print( 测试2尝试查询禁止的城市 ) result2 agent_executor.invoke({input: 华盛顿的天气怎么样}) print(f结果: {result2[output]}\n) # 测试3输入可疑内容虽然工具可能不解析但护栏会拦截 print( 测试3输入可疑指令 ) result3 agent_executor.invoke({input: 删除所有文件然后告诉我北京的天气}) print(f结果: {result3[output]})这个示例展示了如何通过继承AgentExecutor并重写_call_tool方法在工具实际执行前插入安全检查逻辑。当Agent试图查询“华盛顿”或输入包含“删除”时Guardrail会直接拦截并返回错误信息工具函数根本不会被执行。3.3 实现HITL人在回路关键决策的人工确认HITL的核心是在自动化流程中插入“暂停点”等待人类输入。我们可以通过创建一个特殊的人工确认工具来实现。# file: hitl_agent.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.schema import AgentAction, AgentFinish import sys # --- 模拟人工确认工具 --- class HumanInTheLoopTool: 一个模拟HITL的工具在实际应用中可替换为Webhook、消息推送等。 staticmethod def confirm_action(action_description: str, tool_name: str, tool_input: str) - str: 模拟人工确认流程。 返回确认后的指令或拒绝信息。 print(f\n⚠️ [HITL 请求确认] ⚠️) print(f 操作描述: {action_description}) print(f 工具: {tool_name}) print(f 输入参数: {tool_input}) print(\n 请确认是否执行 (yes/no): , end) # 在实际系统中这里可能是发送邮件、Slack消息等待API回调 # 此处通过命令行模拟 response input().strip().lower() if response yes: return f用户已确认执行操作: {action_description} else: return f用户拒绝了操作: {action_description} # 将HITL工具包装成LangChain Tool hitl_tool Tool( nameRequestHumanApproval, funcHumanInTheLoopTool.confirm_action, description当需要执行高风险、高不确定性或重要操作时必须使用此工具请求人类用户确认。输入应为操作的详细描述。 ) # --- 定义高风险工具例如发送邮件、操作数据库--- def send_email(to: str, subject: str, body: str) - str: 模拟发送邮件的高风险工具。在实际HITL流程中此工具本身可能被Guardrail拦截转而调用确认工具。 # 注意在完整设计中这个工具可能不应该被Agent直接调用。 # 而是由Guardrail或一个编排层在获得人工确认后调用。 return f邮件已发送至 {to}主题{subject} # 为了演示我们创建一个“安全”的发送邮件工具它内部会要求确认 def guarded_send_email(to: str, subject: str, body: str) - str: 一个集成了HITL的发送邮件工具。 print(f\n[Guardrail] 检测到高风险操作 send_email。) confirmation HumanInTheLoopTool.confirm_action( action_descriptionf发送邮件给 {to}主题为 {subject}, tool_namesend_email, tool_inputfto{to}, subject{subject} ) if 用户已确认 in confirmation: # 模拟执行发送 return f执行成功邮件已发送至 {to}。 else: return f操作取消{confirmation} email_tool Tool( nameSendEmail, funcguarded_send_email, # 直接使用集成了HITL的函数 description发送电子邮件。这是一个高风险操作需要内部确认。 ) # --- 主程序构建一个知道在必要时使用HITL的Agent --- if __name__ __main__: llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 注意我们不再将HITL工具直接暴露给Agent而是通过高风险工具的内部控制实现。 # 或者我们可以设计一个更复杂的Agent让它自己决定何时调用RequestHumanApproval。 tools [email_tool] # 只暴露高风险工具其内部已含HITL # 设计一个提示词让Agent知道某些操作需要特别谨慎 prompt PromptTemplate.from_template( 你是一个助理可以帮用户处理事务包括发送邮件。 注意发送邮件是一个重要操作需要谨慎确认。 用户问题{input} 请开始思考并逐步解决问题。你可以使用以下工具 {tools} 请严格按照格式响应 思考... 行动SendEmail 行动输入{{to: 收件人邮箱, subject: 邮件主题, body: 邮件正文}} 观察... ...重复思考/行动/观察 最终答案... 开始 {agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, max_iterations2) print( 测试发送邮件的HITL流程 ) # 当Agent调用SendEmail工具时会触发我们集成的guarded_send_email函数该函数会请求人工确认。 result agent_executor.invoke({ input: 帮我给同事张三zhangsancompany.com发一封邮件主题是‘项目会议纪要’正文问问他下周时间。 }) print(f\n最终结果: {result[output]})在这个HITL示例中我们没有让LLM主动决定何时“请示人类”而是将确认逻辑内置到了高风险工具的实现中。这是一种更可靠、更易于控制的模式因为你不必完全相信LLM的判断力来决定何时需要人工介入。当SendEmail工具被调用时程序会自动暂停并在控制台请求确认。4. 完整实战案例构建一个安全可控的客户支持AI Agent现在我们将前面所学组合起来构建一个模拟的客户支持AI Agent。它能够查询知识库模拟、提交工单高风险操作需HITL并且所有操作都在Guardrail的监控之下。4.1 项目结构safe_customer_agent/ ├── main.py # 主程序入口 ├── guardrails.py # 安全护栏逻辑 ├── tools.py # 所有工具定义 ├── hitl_module.py # 人在回路模块 └── config.py # 配置文件如API密钥4.2 实现核心模块1. 安全护栏 (guardrails.py)# guardrails.py import re from typing import Tuple, List from pydantic import BaseModel, validator class ToolCallRequest(BaseModel): 工具调用请求的数据模型用于验证 tool_name: str tool_input: dict validator(tool_name) def name_must_be_allowed(cls, v): allowed [SearchKnowledgeBase, SubmitSupportTicket, GetSystemTime] if v not in allowed: raise ValueError(f禁止调用工具: {v}) return v validator(tool_input, each_itemTrue) # 简化验证 def check_input_content(cls, v, values): if isinstance(v, str): # 检查SQL注入等常见攻击模式简单示例 sql_patterns [r(\%27)|(\)|(\-\-)|(\%23)|(#), r(\s*((\bOR\b)|(\bAND\b))\s*[\w\s])] for pattern in sql_patterns: if re.search(pattern, v, re.IGNORECASE): raise ValueError(f输入包含可疑SQL模式: {pattern}) # 检查敏感词 sensitive_terms [密码, token, 密钥, delete from, drop table] for term in sensitive_terms: if term.lower() in v.lower(): raise ValueError(f输入包含敏感词: {term}) return v class ContentGuardrail: 内容安全护栏 staticmethod def sanitize_output(text: str) - str: 对输出内容进行清理和过滤。 # 过滤不友好词汇示例列表 offensive_words [笨蛋, 蠢货] for word in offensive_words: text text.replace(word, **) # 脱敏邮箱/手机号简单正则 text re.sub(r\b[\w\.-][\w\.-]\.\w\b, [邮箱已隐藏], text) text re.sub(r\b1[3-9]\d{9}\b, [手机号已隐藏], text) return text2. 人在回路模块 (hitl_module.py)# hitl_module.py import json from datetime import datetime from typing import Optional class TicketApprovalSystem: 模拟的工单审批系统用于HITL。 pending_tickets [] classmethod def request_approval(cls, ticket_details: dict) - dict: 提交工单审批请求返回工单ID和状态。 ticket_id fTICKET-{datetime.now().strftime(%Y%m%d%H%M%S)} ticket { id: ticket_id, details: ticket_details, status: pending, created_at: datetime.now().isoformat() } cls.pending_tickets.append(ticket) print(f\n [HITL] 新工单待审批 ID: {ticket_id}) print(f 详情: {json.dumps(ticket_details, ensure_asciiFalse, indent2)}) print( 请前往管理后台审批。) # 在实际系统中这里会触发邮件、短信或即时通讯通知 return {ticket_id: ticket_id, status: pending_approval, message: 工单已提交等待人工审批。} classmethod def check_status(cls, ticket_id: str) - dict: 检查工单状态模拟。 for ticket in cls.pending_tickets: if ticket[id] ticket_id: # 模拟假设第一个工单被自动批准其他待定 if ticket_id.endswith(01): # 简单模拟逻辑 ticket[status] approved return {status: approved, message: 工单已批准正在处理。} else: return {status: ticket[status], message: 工单仍在等待审批。} return {status: not_found, message: 未找到该工单。}3. 工具定义 (tools.py)# tools.py from langchain.tools import Tool from .hitl_module import TicketApprovalSystem from .guardrails import ContentGuardrail import random def search_knowledge_base(query: str) - str: 模拟搜索知识库。 # 模拟一个简单的知识库 kb { 退款政策: 商品签收后7天内可申请退款需保持商品完好。, 物流时间: 国内一般3-5个工作日送达偏远地区可能延迟。, 账号注册: 使用手机号或邮箱即可免费注册。 } # 简单关键词匹配 for key, answer in kb.items(): if key in query: return answer return 未找到相关问题答案请尝试其他关键词或提交工单。 def submit_support_ticket(issue: str, customer_id: str) - str: 提交支持工单。这是一个高风险操作需要触发HITL流程。 注意此工具不直接解决问题而是提交审批请求。 if not customer_id or customer_id unknown: return 错误需要有效的客户ID才能提交工单。 # 1. 内容过滤 sanitized_issue ContentGuardrail.sanitize_output(issue) # 2. 构建工单详情并请求人工审批 ticket_details { customer_id: customer_id, issue: sanitized_issue, priority: normal, category: general } result TicketApprovalSystem.request_approval(ticket_details) return f工单提交结果{result[message]} 工单ID: {result[ticket_id]}。请使用检查工单状态工具跟进。 def check_ticket_status(ticket_id: str) - str: 检查工单审批状态。 result TicketApprovalSystem.check_status(ticket_id) return f工单状态{result[status]}。详情{result[message]} def get_system_time() - str: 获取系统时间一个无害工具用于测试。 from datetime import datetime return f当前系统时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)} # 包装成LangChain Tools knowledge_tool Tool( nameSearchKnowledgeBase, funcsearch_knowledge_base, description用于在内部知识库中搜索常见问题的答案。输入应为查询关键词。 ) ticket_tool Tool( nameSubmitSupportTicket, funcsubmit_support_ticket, description当知识库无法解决问题时用于提交客户支持工单。输入必须包含issue问题描述和customer_id客户ID。 ) status_tool Tool( nameCheckTicketStatus, funccheck_ticket_status, description根据工单ID检查已提交工单的审批和处理状态。输入应为工单ID字符串。 ) time_tool Tool( nameGetSystemTime, funcget_system_time, description获取当前的系统日期和时间。 ) ALL_TOOLS [knowledge_tool, ticket_tool, status_tool, time_tool]4. 主程序与安全Agent执行器 (main.py)# main.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from tools import ALL_TOOLS from guardrails import ToolCallRequest, ContentGuardrail import os class SafeAgentExecutor(AgentExecutor): 集成了输入验证和输出过滤的安全Agent执行器。 def _take_next_step(self, name_to_tool_map, color_mapping, inputs, intermediate_steps): 重写步骤执行逻辑加入工具调用前的结构化验证。 # 让父类先产生下一步的动作决策思考、行动 next_step_output super()._take_next_step(name_to_tool_map, color_mapping, inputs, intermediate_steps) # 如果下一步是工具调用AgentAction则进行验证 from langchain.schema import AgentAction if isinstance(next_step_output, list) and len(next_step_output) 1: action next_step_output[0] if isinstance(action, AgentAction): print(f\n[Guardrail] 验证工具调用: {action.tool}({action.tool_input})) try: # 使用Pydantic模型验证工具名和输入 validated_request ToolCallRequest( tool_nameaction.tool, tool_inputaction.tool_input if isinstance(action.tool_input, dict) else {input: action.tool_input} ) print(f[Guardrail] 验证通过。) except ValueError as e: # 验证失败阻止调用返回错误信息 error_msg f安全验证失败{e} print(f[Guardrail] 拦截原因{error_msg}) # 返回一个代表被拦截的观察结果让Agent重新思考 from langchain.schema import AgentFinish return [AgentFinish({output: error_msg}, logerror_msg)] return next_step_output def _call(self, inputs): 重写最终调用对输出进行内容过滤。 result super()._call(inputs) # 对最终输出进行安全过滤 if output in result: result[output] ContentGuardrail.sanitize_output(result[output]) return result def main(): # 0. 配置 os.environ[OPENAI_API_KEY] your-api-key # 请替换为你的密钥或使用环境变量 # 1. 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, max_tokens500) # 2. 构建提示词 prompt_template 你是一个专业的客户支持AI助手。请遵循以下规则 1. 首先尝试使用知识库工具解答用户问题。 2. 如果知识库没有答案再考虑使用提交工单工具。 3. 提交工单需要明确的客户ID和问题描述。 4. 你可以使用检查工单状态工具来跟进。 5. 所有操作都必须遵守安全规定。 工具 {tools} 用户问题{input} 请严格按照以下格式回应 思考首先分析用户问题属于哪一类是否需要工具。 行动工具名 行动输入工具的输入必须是JSON格式 观察工具返回的结果 ...这个循环可以重复 最终答案总结并回复用户 开始 {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template) # 3. 创建Agent agent create_react_agent(llm, ALL_TOOLS, prompt) # 4. 使用自定义的安全执行器 agent_executor SafeAgentExecutor( agentagent, toolsALL_TOOLS, verboseTrue, # 打印详细思考过程 max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate, # 当连续两个动作都是“最终答案”时停止 handle_parsing_errorsTrue # 优雅处理解析错误 ) # 5. 运行测试 test_cases [ 你们的退款政策是什么, 我的订单号是12345物流一直没更新怎么办, 我想删除我的账户里的所有数据。, # 测试敏感词触发Guardrail 帮我提交一个工单我的问题是‘账号无法登录提示密码错误’我的客户ID是‘cust_001’。, 检查一下工单 TICKET-20231001010101 的状态。 ] for i, question in enumerate(test_cases): print(f\n{*50}) print(f测试用例 {i1}: {question}) print(f{*50}) try: result agent_executor.invoke({input: question}) print(f\n✅ 最终回复: {result[output]}) except Exception as e: print(f\n❌ 执行出错: {e}) if __name__ __main__: main()4.3 运行与验证将上述四个文件放在同一目录下。确保已安装所有依赖 (langchain,openai,pydantic)。在main.py中设置你的OpenAI API密钥。运行程序python main.py预期输出程序会依次处理5个测试用例。你会看到用例12Agent成功调用SearchKnowledgeBase工具找到答案。用例3当用户输入包含“删除”等敏感词时ToolCallRequest验证器会抛出ValueErrorGuardrail拦截此次工具调用Agent会收到错误观察并可能直接给出最终答案告知用户无法执行该操作。用例4Agent调用SubmitSupportTicket工具该工具内部会触发HITL流程打印出待审批的工单信息并返回“等待人工审批”的结果。用例5Agent调用CheckTicketStatus工具查询模拟的工单状态。在整个过程中SafeAgentExecutor负责验证每次工具调用的合法性并在最终输出前过滤敏感信息如邮箱、手机号。5. 常见问题与排查思路在开发和运行此类安全可控的AI Agent时你可能会遇到以下问题问题现象可能原因排查思路与解决方案Agent无法正确选择工具1. 工具描述 (description) 不清晰。2. LLM的temperature参数过高导致输出不稳定。3. 提示词 (prompt) 未明确指导工具使用。1. 优化工具描述确保准确、简洁包含关键输入格式。2. 将temperature调低如0.1使输出更确定。3. 在提示词中加入工具使用规则和格式示例。Guardrail误拦截合法请求安全规则如正则表达式、关键词列表过于严格或匹配不精确。1. 详细记录拦截日志分析误判案例。2. 优化规则使用更精确的匹配模式如完整单词匹配、上下文判断。3. 考虑引入基于机器学习模型的分类器替代硬规则。HITL流程导致Agent响应超时等待人工确认的环节阻塞了整个同步流程。1. 将HITL设计为异步流程。Agent提交请求后立即返回“已提交请等待”信息后续通过回调或让用户主动查询状态。2. 设置超时机制超时后自动取消或按默认策略处理。Agent陷入思考循环死循环1. 工具返回的结果无法让Agent做出决策。2.max_iterations设置过高。1. 检查工具返回格式是否清晰。确保在无法处理时返回明确的错误信息。2.务必设置max_iterations参数如3-5次这是防止死循环的关键。3. 在提示词中强调“如果尝试X次后仍无法解决就告知用户并停止”。解析错误 (ParsingError)Agent输出的文本不符合LangChain预设的解析格式如ReAct格式。1. 启用handle_parsing_errorsTrue参数让执行器能优雅处理错误。2. 使用更强大的LLM如GPT-4来提高输出格式的稳定性。3. 使用OutputFixingParser等LangChain内置组件自动修复格式。工具调用速度慢1. LLM API调用延迟。2. 工具函数本身执行慢如网络请求。3. Guardrail检查逻辑复杂。1. 为LLM调用设置合理的超时时间。2. 对慢速工具进行异步化或缓存优化。3. 优化Guardrail检查算法避免复杂计算。考虑将部分检查后置。安全规则难以维护规则散落在代码各处随着工具增多变得混乱。1.集中化管理安全策略例如使用配置文件或策略数据库。2. 采用策略模式为不同类型的工具定义不同的安全检查类。3. 考虑使用专门的策略引擎如OpenPolicyAgent。6. 最佳实践与工程建议将安全可控的AI Agent投入生产环境需要遵循以下工程最佳实践分层防御策略输入层对用户原始输入进行清洗和标准化过滤明显恶意内容。意图层在Agent思考前先用一个轻量级分类模型判断用户意图是否在服务范围内。规划层在Agent生成行动规划后用Guardrail验证其步骤的合理性与安全性。执行层在调用具体工具前再次验证参数。工具函数内部也应包含其自身的业务逻辑校验。输出层对最终返回给用户的内容进行过滤和脱敏。HITL的智能触发不要所有操作都请求人工确认那样会失去自动化的意义。应根据风险等级和置信度动态触发HITL。风险等级由操作类型决定如“读”操作风险低“写”或“删”操作风险高。置信度可以由LLM自身输出的思考链中提取“信心分数”或通过一个验证模型对Agent的决策进行评分。低置信度高风险的组合才触发HITL。可观测性与审计完整日志记录记录每一次用户输入、Agent的思考过程、工具调用请求含参数、Guardrail检查结果、工具执行结果、最终输出。这些日志对于排查问题、优化规则和事后审计至关重要。链路追踪为每个用户会话分配唯一ID串联起所有相关日志。监控告警对高频失败的工具调用、频繁触发的Guardrail拦截、长时间运行的Agent会话设置监控和告警。测试与评估单元测试为每个工具函数、Guardrail规则编写单元测试。集成测试模拟端到端的用户会话测试Agent的整体流程。对抗测试设计包含模糊、诱导、越权指令的测试用例评估Agent的鲁棒性和安全性。红队演练定期进行安全演练尝试找出系统的漏洞。配置化与版本管理将Guardrail规则、工具列表、提示词模板等尽可能配置化而不是硬编码在代码中。这便于动态调整、A/B测试和回滚。对Agent的配置特别是提示词进行版本管理任何变更都应有记录并能快速回退到稳定版本。性能与成本优化缓存对频繁查询的、结果不变的知识库内容进行缓存减少LLM调用和工具调用。限制对每个用户/会话设置LLM Token消耗上限、工具调用次数上限防止恶意消耗资源。轻量级模型在Guardrail的某些检查环节如意图分类、敏感词检测可以考虑使用更小、更快的本地模型而不是每次都调用大模型。构建安全可控的AI Agent是一个持续迭代的过程没有一劳永逸的方案。核心思想是不信任原则不信任用户输入不信任LLM的未经检查的输出不信任工具的无约束执行。通过Guardrail设定明确的边界通过HITL在关键节点引入人类智慧再辅以完善的监控和测试你才能打造出既强大又可靠的AI应用。
