OpenAI Agents SDK #8：为 Agent 装上「双保险」——Guardrails 防护栏全解析

一、为什么需要 Guardrails？

• 提示注入：用户通过构造输入覆盖系统提示，劫持 Agent 行为
• 有害内容输出：Agent 在没有拦截的情况下生成违规、品牌危机或不合规内容
• PII 泄露：Agent 在回复中意外暴露用户姓名、手机号、身份证等隐私数据3
• 越权工具调用：Agent 调用工具时携带了不合法的参数，如含 API Key 的字符串

二、三种 Guardrail 类型：防护的位置决定防护的边界

三、@input_guardrail 装饰器：写法与执行逻辑

from pydantic import BaseModel
from agents import (
    Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
    RunContextWrapper, Runner, TResponseInputItem, input_guardrail,
)

# ① 定义 guardrail 内部 Agent 的输出结构
class HomeworkOutput(BaseModel):
    is_math_homework: bool
    reasoning: str

# ② 专门用于判断输入的「内卫 Agent」——小模型、便宜
guardrail_agent = Agent(
    name="Homework Guardrail",
    instructions="判断用户是否在要求帮做数学作业。",
    output_type=HomeworkOutput,
)

# ③ 用 @input_guardrail 装饰器定义检查函数
@input_guardrail
async def math_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    # 运行内卫 Agent，得到结构化判断
    result = await Runner.run(guardrail_agent, input, context=ctx.context)

return GuardrailFunctionOutput(
        output_info=result.final_output,          # 可选：将判断结果存入 output_info，供后续审计
        tripwire_triggered=result.final_output.is_math_homework,  # 关键布尔值
    )

# ④ 主 Agent 绑定 guardrail
main_agent = Agent(
    name="Customer Agent",
    instructions="你是一个客服助手，只回答产品相关问题。",
    input_guardrails=[math_guardrail],
)

# ⑤ 运行时捕获异常
async def main():
    try:
        result = await Runner.run(main_agent, "帮我解一下这道方程：2x+5=13")
        print(result.final_output)
    except InputGuardrailTripwireTriggered as e:
        # tripwire 被触发时，Runner 立即抛出此异常，不会继续执行主 Agent
        print(f"输入被拦截：{e.guardrail_result.output.output_info.reasoning}")

• @input_guardrail 同步和异步函数都支持，装饰器自动包装成 InputGuardrail 实例5
• tripwire_triggered=True 是唯一的拉闸信号——Runner 一旦看到这个值为 True，立刻抛出 InputGuardrailTripwireTriggered 异常，主 Agent 连一个 token 都不会生成
• output_info 是 Any 类型，放什么都行：判断原因、置信度分数、完整的 Pydantic 对象，都可以，异常触发时会随 guardrail_result 一起传递出来

四、@output_guardrail 装饰器：守住最后一道门

from agents import (
    Agent, GuardrailFunctionOutput, OutputGuardrailTripwireTriggered,
    RunContextWrapper, Runner, output_guardrail,
)
import re

# 代码护栏示例：检测输出中是否含有 PII（手机号）
@output_guardrail
async def pii_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    output: str,          # 这里是主 Agent 的原始输出文本
) -> GuardrailFunctionOutput:
    # 简单正则：检测 11 位手机号
    phone_pattern = re.compile(r"1[3-9]\d{9}")
    contains_pii = bool(phone_pattern.search(output))

return GuardrailFunctionOutput(
        output_info={"detected_pii": contains_pii},
        tripwire_triggered=contains_pii,
    )

main_agent = Agent(
    name="Customer Agent",
    instructions="你是一个客服助手。",
    output_guardrails=[pii_guardrail],
)

async def main():
    try:
        result = await Runner.run(main_agent, "帮我查一下张三的联系方式")
        print(result.final_output)
    except OutputGuardrailTripwireTriggered as e:
        print("输出包含敏感信息，已被拦截。")

五、GuardrailFunctionOutput 数据结构：只有两个字段，但缺一不可

@dataclass
class GuardrailFunctionOutput:
    tripwire_triggered: bool   # 是否拉闸。True = 立即中断执行
    output_info: Any = None    # 任意附加信息（审计、原因、置信度……）

• InputGuardrailTripwireTriggered：其 .guardrail_result 是 InputGuardrailResult，包含 guardrail（InputGuardrail 对象）和 output（GuardrailFunctionOutput）
• OutputGuardrailTripwireTriggered：其 .guardrail_result 是 OutputGuardrailResult，额外携带 agent_output（代理的原始输出）和 agent（Agent 对象）

六、并发检查机制：InputGuardrail 默认和主 Agent「赛跑」

@dataclass
class InputGuardrail(Generic[TContext]):
    guardrail_function: ...
    name: str | None = None
    run_in_parallel: bool = True   # 默认 True：与主 Agent 并发运行！

@input_guardrail(run_in_parallel=False)
async def blocking_guardrail(ctx, agent, input):
    # 此 guardrail 会在主 Agent 启动前完成，确保检查结果可靠
    ...

七、Tool Guardrail：精细化到每次工具调用

from agents import function_tool, tool_input_guardrail, ToolGuardrailFunctionOutput

# 工具输入护栏：检查参数中是否含有 API Key
@tool_input_guardrail
async def check_no_api_key(ctx, tool_call):
    if "sk-" in str(tool_call.arguments):
        # reject_content() 拒绝工具调用，返回错误消息给主 Agent
        return ToolGuardrailFunctionOutput.reject_content(
            "参数中检测到 API Key，已拒绝执行此工具调用。"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool(tool_input_guardrails=[check_no_api_key])
def call_external_api(endpoint: str, payload: str) -> str:
    """调用外部 API"""
    ...

1. ToolGuardrailFunctionOutput.allow() 和 .reject_content(message) 是互斥的返回——不是 tripwire/bool 那套，而是更直接的「放行/拒绝」语义
2. Tool Guardrail 只适用于 @function_tool 创建的工具，不覆盖 WebSearchTool、FileSearchTool、ComputerTool 等内置工具，也不覆盖 Handoffs4

八、LLM 护栏 vs 代码护栏：别什么都交给 LLM 判断

• 「这条消息是否有害？」——需要理解语义、上下文、隐含意图
• 「这个回复是否偏离了品牌调性？」——主观判断，规则写不完
• 「用户是否在绕过系统提示？」——需要理解意图

• 参数中是否含有 sk- 前缀（API Key 检测）——字符串匹配
• 输出是否含手机号/身份证（PII 检测）——正则表达式
• 输入 token 数是否超限——计数逻辑

九、全局 Guardrails：用 RunConfig 一次性覆盖所有 Agent

from agents import RunConfig, Runner

# 全局输入护栏：对所有 Agent（包括 Handoff 链的第一个）生效
config = RunConfig(
    input_guardrails=[pii_input_guardrail, injection_guardrail],
    output_guardrails=[pii_output_guardrail],
)

result = await Runner.run(
    starting_agent=my_agent,
    input="用户消息",
    run_config=config,
)

十、落地建议：三条可以直接执行的策略

小结