openai-agents-python 安全护栏(Guardrails)完全指南:输入校验、输出过滤与工具调用防护
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
导读
本指南围绕 openai-agents-python 框架中的 Guardrails(安全护栏)机制展开,系统讲解如何在多智能体工作流中,对用户输入与智能体输出进行低成本、高效率的校验与拦截。通过本文,你将掌握输入护栏、输出护栏与工具护栏三类防护手段的触发边界、执行模式与绊线(Tripwire)语义,并能在真实业务中组合"便宜模型做审查、昂贵模型做主业"的成本优化方案,以及针对函数工具的敏感信息拦截实战。
为什么需要 Guardrails:用小模型为大模型把关
在多智能体应用中,负责核心业务的智能体往往绑定高性能(同时也更慢、更贵)的模型。例如一个客服智能体使用顶尖模型处理客户请求,我们不希望恶意用户诱导该模型帮忙解数学作业——这会白白消耗昂贵模型的 token 与时间。
Guardrails 解决这一问题的思路是:用快速、廉价的模型单独运行检查逻辑。当护栏检测到恶意或越界使用场景时,立即抛出错误,从而节省时间与成本。这一设计有一个重要前提,即运行模式的选择:
- 阻塞执行可以保证昂贵模型根本不会启动;
- 并行执行下,昂贵模型可能在护栏完成之前就已经开始运行。
两种模式的取舍详见下文"输入护栏的执行模式"一节。
Guardrails 的两种基本类型
框架将护栏划分为两大基础类型,分别面向智能体生命周期的两端:
- 输入护栏(Input guardrails):在最初的用户输入上执行;
- 输出护栏(Output guardrails):在最终的智能体输出上执行。
在此基础上,还有一类面向工具调用链路的工具护栏(Tool guardrails),将在后文专门展开。
工作流边界:护栏并非在任意时刻都运行
护栏虽然绑定在智能体与工具上,但在多智能体工作流(包含 manager、handoff 或委派的 specialist)中,它们并非在同一时间点全部运行:
- 输入护栏:仅对链路中的第一个智能体运行;
- 输出护栏:仅对产出最终输出的智能体运行;
- 工具护栏:在每一次自定义函数工具调用时运行——输入护栏在工具执行前、输出护栏在工具执行后。
因此,如果工作流包含 manager、handoff 或委派的 specialist,且需要在每次自定义函数工具调用的前后(或前后都要)做检查,请使用工具护栏,而不是只依赖智能体级别的输入/输出护栏。
输入护栏(Input Guardrails)
三步执行流程
输入护栏按以下 3 个步骤运行:
- 首先,护栏接收到与传入智能体相同的输入;
- 然后,护栏函数执行并产生
GuardrailFunctionOutput,它会被包装在InputGuardrailResult中; - 最后,检查
.tripwire_triggered是否为true。若为true,则抛出InputGuardrailTripwireTriggered异常,你可以据此向用户给出适当响应,或捕获处理该异常。
从源码看,这三步对应InputGuardrail.run()的实现:它调用guardrail_function(context, agent, input),通过inspect.isawaitable同时兼容同步与异步函数,最终返回包装了GuardrailFunctionOutput的InputGuardrailResult。
为什么护栏属性挂在 Agent 上,而不是 Runner.run 上?
你可能会疑惑:为什么guardrails属性配置在 Agent 上,而不是传给Runner.run?原因在于护栏通常与具体的 Agent 强相关——不同的智能体需要不同的护栏,将代码放在同一处(Agent 定义处)可读性更好。
执行模式:并行 vs 阻塞
输入护栏支持两种执行模式,由InputGuardrail的run_in_parallel字段控制:
- 并行执行(默认,
run_in_parallel=True):护栏与智能体的执行并发进行。由于二者同时启动,延迟最低;但若护栏绊线被触发,智能体在被取消前可能已经消耗了 token 并执行了工具。 - 阻塞执行(
run_in_parallel=False):护栏在智能体启动之前运行并完成。若护栏绊线被触发,智能体永远不会执行,从而杜绝 token 消耗与工具执行。这适合成本优化,或当你想避免工具调用可能带来的副作用时。
在 src/agents/guardrail.py 中,InputGuardrail的run_in_parallel: bool = True明确标注了默认行为。你既可以通过@input_guardrail装饰器的关键字参数配置(如@input_guardrail(name="guardrail_name", run_in_parallel=False)),也可以直接构造InputGuardrail实例;name参数用于 tracing,缺省时取函数名(见 get_name())。
输出护栏(Output Guardrails)
三步执行流程
输出护栏同样按 3 个步骤运行:
- 首先,护栏接收到智能体生成的输出;
- 然后,护栏函数执行并产生
GuardrailFunctionOutput,它会被包装在OutputGuardrailResult中; - 最后,检查
.tripwire_triggered是否为true。若为true,则抛出OutputGuardrailTripwireTriggered异常。
对应源码OutputGuardrail.run()同样通过inspect.isawaitable兼容同步/异步函数,并在结果中携带agent与agent_output字段,供上层排查。
边界与限制
- 输出护栏旨在对最终智能体输出运行,因此一个智能体的输出护栏仅当它是链路中最后一个智能体时才执行。原因与输入护栏相同:护栏与具体 Agent 强相关,同处存放可读性更好。
- 输出护栏始终在智能体完成后运行,因此不支持
run_in_parallel参数。
绊线与异常的会话持久化差异
输出绊线与护栏函数抛出的异常,在会话(Session)行为上截然不同:
- 绊线(Tripwire)拒绝候选最终输出。绊线触发时,runner 要求配置的会话持久化已完成的工具调用与工具输出项(连同重放这些调用所需的推理上下文),同时排除被拒绝的候选最终输出。该规则对流式与非流式运行同样适用。
- 护栏函数抛出异常(而非返回绊线结果)时,runner 将判定视为"未知",并会先要求会话持久化已完成的最终回合项,然后再把护栏异常抛给上层;若该会话写入也失败,则会话写入错误优先。
- 流式运行使用与非流式相同的持久化顺序,并从
stream_events()抛出终末异常。 - 若在输出护栏运行期间立即调用
RunResultStreaming.cancel(),会取消进行中的护栏,并且不会开始最终回合的会话写入。
终止型函数工具输出的特殊处理
当Agent.tool_use_behavior(见 src/agents/agent.py,默认"run_llm_again")使某个函数工具的结果成为最终输出,而输出绊线又拒绝了它时,工具已经先于智能体级输出护栏执行完毕,因此需要额外的处理逻辑:
- 仅当 SDK 能从已验证字段重建函数调用/输出对时,才会保留可重放的调用/输出对;
- 保留的
function_call_output载荷会被替换为固定文本:"Output withheld by an output guardrail."; - 原始工具输出载荷不会保留在会话、
RunState、流式运行结果状态或沙盒内存输入中的任何一处; - SDK 会保留重放所需的已验证函数调用元数据(包括函数参数),因此该元数据中可能包含曾出现在被拒绝输出里的数据;
- 当前响应的
OutputGuardrailResult对象中,agent_output同样被替换为固定文本,且output_info被清空; - 当前响应的
ToolOutputGuardrailResult对象保留允许/拒绝的行为类型,但承载载荷的output_info与拒绝消息会被替换为同一固定文本; - 更早被接受的回合与护栏执行结果保持不变;
- 若响应包含推理内容或其他 SDK 无法安全消毒的形状,SDK 会丢弃当前响应的完整后缀,而不是保留被拒绝的输出载荷;
- 抛出异常的护栏函数并未返回拒绝判定,因此已完成的终止型工具回合遵循上文"异常持久化行为"。
与这条链路相关的还有 run_config.py 中定义的OutputGuardrailBlockedMessageFormatter(同步格式化器):它在终止型工具输出被拒绝后、每个重放与持久化拥有者被替换为无数据占位文本之前执行,可用来自定义默认的拒绝提示文案。
工具护栏(Tool Guardrails)
设计定位
工具护栏包装FunctionTool实例,允许你在工具执行前与后校验或拦截对该工具的调用。它们配置在工具自身之上,并且每次该工具被调用时都会运行。
- 输入工具护栏:在工具执行前运行,可以跳过本次调用、将输出替换为一条消息、或触发绊线;
- 输出工具护栏:在工具执行后运行,可以替换输出或触发绊线。
与人工审批(Approval)的配合
如果某个函数工具需要审批(approval),输入工具护栏默认在审批之后、执行之前立即运行。如果你希望在发出"待审批中断"之前就执行这些输入检查,可将RunConfig.tool_execution设置为ToolExecutionConfig(pre_approval_tool_input_guardrails=True)。
源码 src/agents/run_config.py 中,ToolExecutionConfig还包含max_function_tool_concurrency(单回合内本地函数工具的最大并发数,None表示保持默认、同时启动所有工具调用)并做了参数校验:pre_approval_tool_input_guardrails必须是布尔值。注意:通过预批准检查的调用,在工具执行前仍会进行审批后的再次检查。
适用范围边界
- 工具护栏仅适用于通过
function_tool创建的函数工具; - Handoff走的是 SDK 的 handoff 管线而非普通函数工具管线,因此工具护栏不适用于 handoff 调用本身;
- 托管工具(
WebSearchTool、FileSearchTool、HostedMCPTool、CodeInterpreterTool、ImageGenerationTool)与内置执行工具(ComputerTool、ShellTool、ApplyPatchTool、LocalShellTool)同样不使用这条护栏管线; Agent.as_tool()目前不直接暴露工具护栏选项。
底层行为模型
src/agents/tool_guardrails.py 中的ToolGuardrailFunctionOutput定义了三种行为:
allow:允许工具调用/输出正常继续(默认),对应ToolGuardrailFunctionOutput.allow();reject_content:拒绝工具调用/输出,但继续执行并将消息回传给模型,对应ToolGuardrailFunctionOutput.reject_content(message);raise_exception:抛出ToolGuardrailTripwireTriggered异常以中止执行,对应ToolGuardrailFunctionOutput.raise_exception()。
输入数据方面,输入护栏收到 ToolInputGuardrailData(含context与agent);输出护栏收到 ToolOutputGuardrailData,它在输入数据基础上追加output字段。@tool_input_guardrail与@tool_output_guardrail装饰器均同时支持同步与异步函数。
绊线(Tripwires)
语义
当智能体的输入或输出未通过护栏检查时,护栏通过绊线发出信号。runner 会立即抛出异常并中止智能体执行:
- 智能体级:
InputGuardrailTripwireTriggered或OutputGuardrailTripwireTriggered; - 工具级:
ToolInputGuardrailTripwireTriggered或ToolOutputGuardrailTripwireTriggered。
对应的异常类定义在 src/agents/exceptions.py:
- 智能体级异常暴露
guardrail_result属性,用于定位是哪个护栏触发了绊线; - 工具级异常则直接暴露触发绊线的
guardrail与output。
累积结果的可观测性
- 对于 runner 抛出的输入绊线,
exception.run_data.input_guardrail_results包含运行停止前已完成的所有输入护栏执行结果,其中包括触发绊线的那一个;输出绊线则通过exception.run_data.output_guardrail_results提供等价的累积结果。 - 工具绊线异常通过
run_data.tool_input_guardrail_results与run_data.tool_output_guardrail_results保留失败前已完成回合累积的执行结果;触发绊线的那一个结果可通过异常的output取得。 - 其他 runner 管理的失败(例如
MaxTurnsExceeded)也会在同样的列表中保留已完成的工具护栏执行结果。 stream_events()抛出异常后,流式结果会暴露相同的累积智能体与工具护栏执行结果列表。- 注意:当异常发生在 runner 管理的执行路径之外时,
run_data可能为None(src/agents/exceptions.py 中这些列表均为可空字段)。
实战:实现一个输入护栏
实现护栏的核心是提供一个接收输入、返回GuardrailFunctionOutput的函数。下面的例子内部通过运行一个 Agent 来完成检查——用轻量模型判断用户是否在要求解数学作业:
from pydantic import BaseModel from agents import ( Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, RunContextWrapper, Runner, TResponseInputItem, ) from agents.decorators import input_guardrail class MathHomeworkOutput(BaseModel): is_math_homework: bool reasoning: str guardrail_agent = Agent( # (1)! name="Guardrail check", instructions="Check if the user is asking you to do their math homework.", output_type=MathHomeworkOutput, ) @input_guardrail async def math_guardrail( # (2)! ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem] ) -> GuardrailFunctionOutput: result = await Runner.run(guardrail_agent, input, context=ctx.context) return GuardrailFunctionOutput( output_info=result.final_output, # (3)! tripwire_triggered=result.final_output.is_math_homework, ) agent = Agent( # (4)! name="Customer support agent", instructions="You are a customer support agent. You help customers with their questions.", input_guardrails=[math_guardrail], ) async def main(): # This should trip the guardrail try: await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?") print("Guardrail didn't trip - this is unexpected") except InputGuardrailTripwireTriggered: print("Math homework guardrail tripped")- 该 Agent 在护栏函数内部被调用,负责输出结构化判定结果。
- 这是护栏函数:接收智能体的输入/上下文,返回执行结果。
- 可以在护栏结果中携带附加信息(
output_info)。 - 这是定义工作流的真实业务智能体,通过
input_guardrails=[math_guardrail]挂载护栏。
更多可运行变体可参考 examples/agent_patterns/input_guardrails.py。
实战:实现一个输出护栏
输出护栏与输入护栏结构类似,区别在于它接收的是智能体的最终输出:
from pydantic import BaseModel from agents import ( Agent, GuardrailFunctionOutput, OutputGuardrailTripwireTriggered, RunContextWrapper, Runner, ) from agents.decorators import output_guardrail class MessageOutput(BaseModel): # (1)! response: str class MathOutput(BaseModel): # (2)! reasoning: str is_math: bool guardrail_agent = Agent( name="Guardrail check", instructions="Check if the output includes any math.", output_type=MathOutput, ) @output_guardrail async def math_guardrail( # (3)! ctx: RunContextWrapper, agent: Agent, output: MessageOutput ) -> GuardrailFunctionOutput: result = await Runner.run(guardrail_agent, output.response, context=ctx.context) return GuardrailFunctionOutput( output_info=result.final_output, tripwire_triggered=result.final_output.is_math, ) agent = Agent( # (4)! name="Customer support agent", instructions="You are a customer support agent. You help customers with their questions.", output_guardrails=[math_guardrail], output_type=MessageOutput, ) async def main(): # This should trip the guardrail try: await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?") print("Guardrail didn't trip - this is unexpected") except OutputGuardrailTripwireTriggered: print("Math output guardrail tripped")- 这是真实业务智能体的输出类型。
- 这是护栏自身的输出类型。
- 这是护栏函数:接收智能体的输出,返回执行结果。
- 这是定义工作流的真实智能体,通过
output_guardrails与output_type完成挂载。
可参考 examples/agent_patterns/output_guardrails.py 与流式场景下的 examples/agent_patterns/streaming_guardrails.py。
实战:实现工具护栏
最后是工具护栏示例——在工具执行前拦截包含密钥的调用参数,在工具执行后对包含敏感数据的输出进行消毒:
import json from agents import ( Agent, Runner, ToolGuardrailFunctionOutput, ) from agents.decorators import tool, tool_input_guardrail, tool_output_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( "Remove secrets before calling this tool." ) return ToolGuardrailFunctionOutput.allow() @tool_output_guardrail def redact_output(data): text = str(data.output or "") if "sk-" in text: return ToolGuardrailFunctionOutput.reject_content("Output contained sensitive data.") return ToolGuardrailFunctionOutput.allow() @tool( tool_input_guardrails=[block_secrets], tool_output_guardrails=[redact_output], ) def classify_text(text: str) -> str: """Classify text for internal routing.""" return f"length:{len(text)}" agent = Agent(name="Classifier", tools=[classify_text]) result = Runner.run_sync(agent, "hello world") print(result.final_output)本例中,block_secrets从data.context.tool_arguments解析工具参数,若发现sk-前缀的疑似密钥则调用reject_content拒绝该次调用(并携带提示消息回传给模型);redact_output则在工具返回后检查输出文本。两个护栏通过@tool装饰器的tool_input_guardrails/tool_output_guardrails参数挂载到同一个函数工具上。
测试与验证:如何在仓库中确认护栏行为
如果你希望深入验证护栏的边界行为,仓库提供了完备的测试用例:
- tests/test_guardrails.py:智能体级输入/输出护栏的执行流程、绊线异常与累积结果;
- tests/test_tool_guardrails.py:工具护栏的允许/拒绝/异常三种行为及工具绊线语义;
- tests/test_output_guardrail_cancellation.py:输出护栏运行期间调用
RunResultStreaming.cancel()的取消路径; - tests/test_runner_guardrail_resume.py:绊线触发后 runner 的恢复与会话持久化行为;
- tests/test_stream_input_guardrail_timing.py:流式场景下输入护栏的时序。
这些测试直接印证了本文所述的执行模式、绊线异常与持久化规则,是理解 Guardrails 底层实现的最佳补充材料(英文原文文档见 docs/guardrails.md,日文版即本文所依据的 docs/ja/guardrails.md)。
小结
综合来看,openai-agents-python 的 Guardrails 体系围绕"检查点"思想设计:
- 输入护栏守卫工作流的起点(仅第一个智能体),支持并行/阻塞两种执行模式,是控制成本的第一道闸门;
- 输出护栏守卫工作流的终点(仅最后一个智能体),并提供绊线与异常两种差异化的会话持久化语义,保障多轮会话的一致性;
- 工具护栏为每个自定义函数工具提供执行前/执行后的细粒度拦截能力,与审批流程协同工作,是防止副作用与敏感信息泄漏的关键手段。
在落地时,建议遵循"便宜模型做审查、昂贵模型做主业"的范式:将结构化的判定逻辑(Pydantic 输出类型 + 轻量 Agent)封装为护栏函数,通过tripwire_triggered快速终止异常链路,并根据业务对延迟与成本的敏感度,为输入护栏选择合适的执行模式。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考