ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

LangChain高级用法实战:用Guardrails与Human-in-the-loop构建安全Agent

LangChain高级用法实战:用Guardrails与Human-in-the-loop构建安全Agent 1. 为什么你的 Agent 需要 Guardrails 与 Human-in-the-loop很多人第一次用 LangChain 搭 Agent跑通 demo 就以为万事大吉结果一上真实业务就翻车用户随口一句「帮我把刚才那封邮件转发给张三」Agent 真就调了发信工具或者用户输入里夹带了信用卡号模型原样复述到日志里。这类问题不是模型不够聪明而是缺少**护栏Guardrails和人工介入Human-in-the-loop**这两层可控机制。Guardrails 说白了就是给 LLM 的输入输出加一道「安检门」。它能在调用模型前检查用户输入是否越界、是否含提示注入也能在模型输出后校验格式、业务规则、是否出现竞品名称。LangChain 把这套机制做成了中间件middleware内置了 PII 检测和人工干预两种最常用的护栏你不需要从零写校验逻辑挂上中间件就能用。Human-in-the-loop 则是高风险操作的「刹车」。金融转账、删除生产数据、向外部发送通信这类动作一旦执行就难以回滚所以要在工具真正执行前中断流程把决策权交回给人。LangGraph 的 checkpointer 会把中断状态持久化人批准或驳回后用同一个thread_id恢复执行上下文完全一致。这篇文章面向已经会写基础 Agent、想把它做「安全可控」的开发者。我会用 TaoToken 作为统一模型通道把 PII 检测、人工审批、中断恢复、本地验证完整跑一遍代码可直接复制。适合谁正在做客服机器人、内部运维 Agent、任何会碰敏感数据或敏感操作的同学。2. TaoToken 前置准备统一 Key 与 API 通道在写护栏代码之前先把模型通道理顺。LangChain 的create_agent需要一个 chat model 实例你可以用ChatOpenAI指向任意兼容 OpenAI 协议的服务。TaoToken 提供统一的 Key 和 API 通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台生成 Key 即可。为什么这里推荐用统一通道而不是每个模型单独配 Key因为护栏场景经常要切换模型做对比测试——比如 PII 检测用便宜模型、主 Agent 用强模型如果每个都单独维护 base_url 和 key配置会散落各处。统一通道的好处是 Base URL 固定换模型只改 Model ID。你需要准备三样东西我把它叫做「三件套」配置项值说明Base URLhttps://taotoken.net/api兼容 OpenAI 协议注意不带 UTMAPI Key控制台生成形如sk-...不要提交到 gitModel ID如gpt-4o/deepseek-chat按你订阅的模型填获取 Key 的入口在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。如果你还没决定用哪个模型可以先在模型对话页试一下https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。安装依赖建议用虚拟环境python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -U langchain langchain-openai langgraph环境变量建议这样管理避免 Key 硬编码进代码export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意Base URL 用https://taotoken.net/api不要在后面拼/v1LangChain 的 OpenAI 兼容层会自己处理路径。如果你拼错了最常见的报错就是 404 或local proxy failed。配置好之后先写一个最小连通性测试确认通道没问题再往上叠护栏import os from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o, openai_api_keyos.environ[TAOTOKEN_API_KEY], openai_api_baseos.environ[TAOTOKEN_BASE_URL], temperature0, ) print(llm.invoke(只回复两个字连通).content)跑通这一步说明三件套配置正确后面所有护栏代码都基于这个llm实例。如果这一步就报 401先回去检查 Key 是否复制完整、是否有多余空格。3. 可复制配置PII 检测与 Human-in-the-loop 中间件这一节是核心我把 PII 检测和人工审批的配置片段完整给出路径和参数都按 LangChain 当前版本写你直接复制到agent_guard.py就能用。先看 PII 检测。PIIMiddleware的第一个参数是pii_type支持内置类型email、credit_card、ip、mac_address、url也支持自定义类型名。strategy决定处理方式默认redact还有block、mask、hash三种。apply_to_input默认 Trueapply_to_output和apply_to_tool_results默认 False按需打开。from langchain.agents import create_agent from langchain.agents.middleware import PIIMiddleware from langchain_openai import ChatOpenAI import os llm ChatOpenAI( modelgpt-4o, openai_api_keyos.environ[TAOTOKEN_API_KEY], openai_api_baseos.environ[TAOTOKEN_BASE_URL], temperature0, ) def customer_service_tool(query: str) - str: 客服查询工具 return f已查询{query} agent create_agent( modelllm, tools[customer_service_tool], middleware[ PIIMiddleware( email, strategyredact, apply_to_inputTrue, apply_to_outputTrue, ), PIIMiddleware( credit_card, strategymask, apply_to_inputTrue, ), ], ) result agent.invoke({ messages: [{ role: user, content: 我的邮箱是 john.doeexample.com卡号 4532-1234-5678-9010 }] }) print(result[messages][-1].content)这里我挂了两个 PII 中间件一个管邮箱用redact替换成[REDACTED_EMAIL]一个管信用卡用mask只保留后四位。注意中间件是链式的第一个失败的会触发整体失败所以顺序很重要——把最容易命中的放前面。再看 Human-in-the-loop。HumanInTheLoopMiddleware的interrupt_on是一个字典key 是工具名value 是布尔值True 表示触发该工具时中断等待人工批准。必须配checkpointer否则中断状态无法持久化恢复时会报错。from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command from langchain_openai import ChatOpenAI import os llm ChatOpenAI( modelgpt-4o, openai_api_keyos.environ[TAOTOKEN_API_KEY], openai_api_baseos.environ[TAOTOKEN_BASE_URL], temperature0.7, ) def get_company_info() - str: 获取用户公司信息属于敏感操作需人工批准 return 公司信息钱多事少离家近有限责任公司 def delete_database_func(table: str) - str: 删除数据库表属于高危操作需人工批准 return f数据库表 {table} 已删除 def get_weather(city: str) - str: 获取指定城市的天气 return f{city} 天气总是晴朗 agent create_agent( modelllm, tools[get_company_info, delete_database_func, get_weather], middleware[ HumanInTheLoopMiddleware( interrupt_on{ get_company_info: True, delete_database_func: True, get_weather: False, } ) ], checkpointerInMemorySaver(), ) config {configurable: {thread_id: session_001}} first_result agent.invoke( {messages: [{role: user, content: 查一下我公司信息}]}, configconfig, ) print(中断提示, first_result)跑完这段你会看到first_result里带有中断信息说明流程在get_company_info执行前停住了。接下来就是恢复逻辑用Command(resume...)把人的决策传回去user_confirm input(是否允许执行(y/n)) if user_confirm.lower() y: command Command(resume{decisions: [{type: approve}]}) else: command Command(resume{ decisions: [{type: reject, reason: 拒绝操作}] }) final_result agent.invoke(command, configconfig) print(执行结果, final_result[messages][-1].content)关键点thread_id必须和第一次调用一致否则恢复时找不到中断状态。decisions是一个列表按中断的工具顺序对应type只能是approve或reject。提示如果你把interrupt_on里的工具名写错了比如函数名和字典 key 不一致中断不会触发Agent 会直接执行工具。这是最常见的配置坑写完先打印一下agent的工具列表核对。4. 验证请求本地跑通中断与恢复全流程配置写好了现在完整跑一遍确认中断、批准、驳回三条路径都正常。我建议把代码拆成两个文件agent_guard.py放 Agent 定义run_demo.py放交互逻辑方便反复测试。先验证 PII 检测。运行第 3 节的 PII 代码观察输出。正常情况下模型看到的输入里邮箱已经被替换成[REDACTED_EMAIL]信用卡号被遮盖成****-****-****-9010。如果你在apply_to_outputTrue的情况下让模型复述用户输入输出里的 PII 也会被同样处理。python pii_demo.py # 预期输出类似 # 已收到您的信息邮箱已脱敏处理卡号尾号 9010。再验证人工审批。运行 Human-in-the-loop 代码你会看到终端停在input等待。输入yAgent 继续执行get_company_info最终输出公司信息。输入nAgent 收到 reject 决策不会执行工具而是返回一个说明操作被拒绝的消息。这里有个细节值得注意中断发生时first_result里其实已经包含了模型决定调用哪个工具、传什么参数的信息。你可以把它打印出来人工审批时就能看到「Agent 想干什么」而不是盲目批准。生产环境里这个信息应该渲染到审批界面让审批人看到工具名和参数。import json print(json.dumps(first_result, defaultstr, ensure_asciiFalse, indent2))多轮测试时每次换一个thread_id避免状态串扰config {configurable: {thread_id: fsession_{int(time.time())}}}如果你要验证「驳回后 Agent 是否能优雅收尾」可以在 reject 的 reason 里写清楚原因模型会基于这个 reason 生成对用户的解释。实测下来reason 写得越具体模型的收尾回复越自然。最后验证一个组合场景用户输入里既有 PII 又要触发敏感工具。这时候 PII 中间件先跑把输入脱敏然后 Human-in-the-loop 中断等待审批。两个中间件的顺序决定了先脱敏还是先中断一般建议 PII 在前避免敏感信息进入审批日志。5. 常见报错排查401、local proxy failed 与中断不触发这一节把我踩过的坑集中列一下对照真实报错定位问题。报错一401 Unauthorized。最常见的原因是 Key 没读到或复制不全。先确认环境变量echo $TAOTOKEN_API_KEY | head -c 10如果输出为空说明环境变量没生效检查是否在正确的 shell 里 export或者改用.env文件加python-dotenv。另一个原因是 Key 前后有空格或换行strip()一下再传。报错二local proxy failed 或 Connection error。这类报错通常指向 Base URL 配置问题。确认你用的是https://taotoken.net/api没有多余路径也没有被系统代理拦截。如果你本地配了全局代理LangChain 的 httpx 客户端可能走错通道临时unset HTTP_PROXY HTTPS_PROXY再试。报错三reading choices 相关解析错误。这通常是返回体不是标准 OpenAI 格式或者模型 ID 写错导致服务端返回了错误页。检查model参数是否是你订阅的模型比如gpt-4o写成gpt4o就会失败。报错四中断不触发工具直接执行。三个可能interrupt_on的 key 和工具函数名不一致checkpointer没配thread_id在恢复时变了。逐个核对尤其是工具名create_agent用的是函数名作为工具名别写成中文描述。报错五恢复时报No pending interrupt。说明你恢复时用的thread_id和中断时不一致或者中间又调了一次invoke把状态消耗掉了。中断和恢复必须严格配对中间不要插入其他invoke。报错六OAuth 或鉴权相关提示。如果你用的是某些需要额外鉴权的模型确认 TaoToken 控制台里该模型已开通。鉴权失败时先回模型对话页发一条消息验证通道再排查代码。注意所有报错排查的第一步都是「最小复现」——把护栏全去掉只留llm.invoke确认通道本身没问题再逐个加回中间件。这样能快速定位是通道问题还是护栏配置问题。6. 把护栏接入你的项目从 demo 到可用跑通 demo 只是第一步真正落地还要考虑几件事。第一PII 检测的自定义规则。内置检测器覆盖常见类型但业务里可能有工号、订单号这类自定义 PII这时候用detector参数传自定义正则PIIMiddleware( employee_id, strategyhash, detectorrEMP-\d{6}, apply_to_inputTrue, )第二人工审批的异步化。demo 里用input()阻塞等待生产环境要改成审批队列把中断状态存到数据库审批人处理后回调恢复。LangGraph 的 checkpointer 可以换成SqliteSaver或PostgresSaver中断状态就能跨进程持久化。第三多 Agent 场景下的护栏。如果你用工具调用模式把子 Agent 当工具子 Agent 的中间件要单独配父 Agent 的护栏不会自动继承。交接Handoffs模式在 LangChain 里没有原生支持需要借助 LangGraph 或 AutoGen 实现这时候护栏要挂在图节点上。第四长期记忆与护栏的配合。LangGraph 把长期记忆存成 JSON 文档放在 store 里工具读写记忆时PII 中间件的apply_to_tool_results要打开否则记忆里可能残留敏感信息。如果你要把这套东西用到长期编码或 Agent 项目上Coding Plan 的额度更适合反复调试https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有各语言的完整示例。Claude Code 相关的接入配置也可以参考文档里的 Anthropic 兼容说明。最后说个实用技巧护栏的单元测试不要只测「正常路径」一定要构造恶意输入——带 PII 的、带提示注入的、触发敏感工具的把这三类用例固化成测试集每次改中间件配置都跑一遍。我试过在 CI 里加这一步能挡住大部分回归问题。
返回列表