如何给智能体装上输入输出双防护:openai-agents-python Guardrail 实践指南
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
一次线上复盘:客服智能体被用户追问"帮我解 2x+5=11",它竟一本正经地开始解题——更糟的是,下一位用户直接拿到了上一位用户留在上下文里的电话号码。这类失控在规模化落地时几乎必然发生,根因是模型既会"答错事",也可能"吐隐私"。openai-agents-python 框架内置的 Guardrail(防护机制)就是为此设计的:它让一个轻量的检查函数或廉价模型,在主智能体运行的同时并行裁决"这段输入能不能进、这条输出能不能发"。读完本文,你能独立写出输入/输出防护函数、选择合适的执行模式,并按业务场景搭起一套完整的智能体安全方案。
防护到底在防什么:一条安检线上的三道闸口
把一次智能体运行想象成机场安检,Guardrail 就是安在三个位置的闸口,触发时机各不相同:
- 入口闸(输入防护):只在用户原始输入进入时启动,且只挂在工作链的第一个智能体上。用户说一句"帮我写个破解脚本",检查函数立刻看到这句话并裁决。
- 工具闸(工具防护):挂在具体的
FunctionTool上,每次调用都过一遍——执行前检查参数(tool_input_guardrail),执行后检查返回值(tool_output_guardrail)。有 handoff 或多智能体协作时,只有它能覆盖"每一次工具调用",而 agent 级防护覆盖不到中间环节。 - 出口闸(输出防护):只在产出最终输出的那个智能体完成后启动,校验的是
final_output。它永远是"先出稿、后审核",所以不支持并行参数。
三道闸口共用同一套裁决语言:防护函数返回GuardrailFunctionOutput,核心字段是tripwire_triggered(是否拉响绊线)和output_info(附带证据)。tripwire_triggered=True时,Runner 立即抛出对应的InputGuardrailTripwireTriggered/OutputGuardrailTripwireTriggered/ToolInputGuardrailTripwireTriggered等异常并中止运行——你拿到异常就能接住、降级、回话。
理解这张时序图有个关键细节:输入防护默认与主智能体并行跑(run_in_parallel=True,延迟最优),但代价是触发时昂贵模型可能已消耗部分 token;若设为run_in_parallel=False,则防护先跑完、主流程才启动,能"零成本拦截"。输出防护则恒为主智能体完成后运行。
把防护网搭起来:从判定标准到降级响应
第 1 步:先定义判定标准——防护函数怎么写
防护函数本身不复杂,常见套路是"再跑一个便宜的小模型做裁判,用 Pydantic 模型约束它的结论":
from pydantic import BaseModel from agents import Agent, GuardrailFunctionOutput, Runner from agents.decorators import input_guardrail class MathHomeworkCheck(BaseModel): is_math_homework: bool reasoning: str guardrail_agent = Agent( name="Guardrail check", instructions="Check if the user is asking you to do their math homework.", output_type=MathHomeworkCheck, ) @input_guardrail async def math_guardrail(ctx, agent, input): result = await Runner.run(guardrail_agent, input, context=ctx.context) return GuardrailFunctionOutput( tripwire_triggered=result.final_output.is_math_homework, output_info=result.final_output, )要点有三:裁判模型要便宜快速(主流程用的是贵模型,防护的意义就是花小钱拦大雷);判定结果尽量结构化(bool + 理由,方便日志与审计);output_info可以塞任意证据,触发时你直接能读到"为什么拦"。纯规则场景(比如检测密钥字符串sk-)则可以完全不调模型,直接在函数里写 if 判断,更快更稳。
第 2 步:再挂到主流程——配置防护链与执行模式
防护是挂在 Agent 上的,不同 Agent 可以挂不同防护,代码就近可读:
agent = Agent( name="Customer support agent", instructions="You are a customer support agent. You help customers with their questions.", input_guardrails=[math_guardrail], # 入口闸 output_guardrails=[sensitive_data_check], # 出口闸 )工具级防护则挂在工具装饰器上,对每次调用生效,支持"放行 / 拦截并给模型一条替代消息 / 抛绊线"三种动作:
from agents import ToolGuardrailFunctionOutput from agents.decorators import tool, tool_input_guardrail @tool_input_guardrail def block_secrets(data): args = json.loads(data.context.tool_arguments or "{}") if "sk-" in json.dumps(args): return ToolGuardrailFunctionOutput.reject_content("参数含密钥,拒绝执行") return ToolGuardrailFunctionOutput.allow() @tool(tool_input_guardrails=[block_secrets]) def classify_text(text: str) -> str: """Classify text for internal routing.""" return f"length:{len(text)}"需要提醒:工具防护只作用于@tool创建的函数工具,托管工具(WebSearch 等)、handoff 调用不走这条管线。
第 3 步:处理触发后的降级——让拦截体面落地
绊线触发就是抛异常,你要做的只是"接住并说人话"。输入防护触发后,把拒绝消息作为 assistant 消息追加回上下文即可,会话能自然继续:
from agents import Runner, InputGuardrailTripwireTriggered try: result = await Runner.run(agent, user_input) except InputGuardrailTripwireTriggered as e: info = e.guardrail_result.output_info print(f"拦截原因: {info.reasoning}") # 裁判模型给出的理由 user_reply = "抱歉,这个问题我无法协助。"输出防护触发时同理捕获OutputGuardrailTripwireTriggered;区别在于被拒的候选最终输出不会写入会话持久化——已完成的工具调用会保留,违规的那版"终稿"会被丢弃。流式场景下,可以在生成过程中每 N 个 token 跑一次检查,发现越界立即cancel(),避免长篇违规内容刷完才拦截,参考 examples/agent_patterns/streaming_guardrails.py。
按场景选配置:输入防护与输出校验怎么配
| 业务场景 | 推荐防护组合 | 关键参数/做法 | 参考实现 |
|---|---|---|---|
| 客服对话 | 输入过滤(拒题/违规话题)+ 输出 PII 检测 | 裁判模型用小而快的型号;run_in_parallel=True控延迟 | examples/customer_service/main.py |
| 内容生成 | 输出主题检测 + 格式/长度校验 | output_type用严格 Pydantic 模型约束终稿结构 | examples/agent_patterns/output_guardrails.py |
| 高成本模型入口 | 输入防护前置拦截 | run_in_parallel=False阻断执行,触发时主模型零消耗 | docs/guardrails.md Execution modes 一节 |
| 工具密集流程 | 工具级入参/出参双向防护 | tool_input_guardrails+tool_output_guardrails,出参可reject_content脱敏 | examples/basic/tool_guardrails.py |
选型口诀:怕花钱、怕副作用 → 入口阻断式;怕延迟 → 并行式;多智能体/多工具链路 → 补上工具闸。
真实项目里长这样:一个"证据审计员"的合规链路
仓库自带的金融研究多智能体(examples/financial_research_agent/)值得拆着看:搜索、财报、风控、写作等智能体各司其职,而安全的关键一环藏在agents/verifier_agent.py——一个专职的 VerificationAgent。
它的做法值得借鉴:
- 输入侧:planner 只接受结构化研究请求,超出白名单的指令在入口就被挡下;
- 工具侧:联网检索类工具带输出防护,来源 URL 不在允许清单内的内容不进入上下文;
- 输出侧:最终报告交给"证据审计员"终审——它拿着原始请求、研究截止日期、报告全文和带来源 URL 的结构化证据,逐条核对数字与时间敏感声明,产出
VerificationResult:verified: bool加一份issues清单,每条 issue 标明类别(unsupported/contradicted/stale_or_unreleased)和解释。报告没过审就不发布。
这套"生成者与审核者分离"的模式,本质是把 Guardrail 的思想放大到报告粒度:审核者只看证据、不看记忆,结论结构化,可直接进 CI 或人工复核流程。
别踩这些坑:三个高频问题与对策
坑一:加了防护会不会拖慢响应?默认并行模式下,防护与主流程同时起跑,额外延迟≈ max(两者) 而非相加,通常可感知成本很小。若你的裁判函数是纯规则(正则、白名单),耗时可忽略;若裁判本身是 LLM,务必选小快模型。真正要权衡的是run_in_parallel的取舍:并行省延迟但拦不干净 token 消耗,阻断式省成本但多等一次裁判——按"入口值不值得先花一秒"来定。
坑二:误拦了正常请求怎么办?把output_info.reasoning落日志,先攒一段时间再调阈值,而不是直接改判断逻辑。实践建议:裁判输出"是否违规 + 置信理由"两个字段,低置信的只记录不拦截;确需拦截的场景用结构化拒绝话术安抚用户,并把拒绝消息回写上下文(见第 3 步示例),保证会话不断。
坑三:多智能体链路里防护"没生效"?多半是挂错了位置。输入防护只在链首智能体生效,输出防护只在链尾生效;如果危险发生在中间 handoff 或某次工具调用,只有工具级防护能覆盖。链路越长,越要把工具闸补上。
落地检查清单
从基础到进阶,逐项打勾推进:
- 最小防线:首个对用户敞开的智能体上挂 1 个输入防护(哪怕纯规则)+ 1 个输出防护
- 执行模式定调:按成本敏感度决定
run_in_parallel,并在代码里写注释说明理由 - 异常全捕获:
InputGuardrailTripwireTriggered/OutputGuardrailTripwireTriggered都有 try/except 与用户话术,拒绝消息回写上下文 - 工具闸补齐:所有会写数据、发请求的
@tool都配置入参/出参防护 - 留痕审计:
guardrail_result.output_info进日志,便于回溯"为什么拦" - 流式早停:长文本生成场景接入 examples/agent_patterns/streaming_guardrails.py 的周期性检查
- 终稿审核者:高价值输出(报告、合规文档)仿照金融研究智能体,增设独立验证 Agent 出结构化终审结论
完整 API 与边界行为说明见官方文档:docs/guardrails.md;输入、输出、工具三类防护的可运行示例分别在 examples/agent_patterns/input_guardrails.py、examples/agent_patterns/output_guardrails.py 与 examples/basic/tool_guardrails.py,跑一遍即可对照本文动手。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考