ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

CAI Guardrails 深度指南:为 AI 安全 Agent 构建输入输出护栏机制

CAI Guardrails 深度指南:为 AI 安全 Agent 构建输入输出护栏机制 CAI Guardrails 深度指南为 AI 安全 Agent 构建输入输出护栏机制【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/caiCAICybersecurity AI是面向 AI 安全场景的 Agent 框架。本文聚焦其核心安全组件Guardrails护栏一套与 Agent 并行运行、对用户输入与最终输出进行实时校验的机制。阅读本文后你将掌握InputGuardrail与OutputGuardrail的完整工作原理、tripwire绊线触发机制并能在自己的安全 Agent 中落地越权请求拦截敏感数据泄露防护等实战护栏。一、Guardrails 是什么并行执行的安全哨兵Guardrails 是运行在 Agent旁边并行的检查器用于对用户输入做校验和验证。其典型价值来自一个现实场景假设你的 Agent 使用一个非常聪明因此又慢又贵的大模型来处理客户请求你不希望恶意用户借这个模型去干帮他们写数学作业之类的私活。这时可以用一个快速/廉价的模型跑一个 guardrail——一旦检测到恶意使用立即抛出异常终止昂贵模型的运行为你节省时间与金钱。在 CAI 中Guardrails 分为两类见 guardrail.pyInput guardrails输入护栏作用于 Agent 的初始用户输入Output guardrails输出护栏作用于 Agent 的最终输出。两者都挂载在Agent对象上input_guardrails[...]、output_guardrails[...]而非传给Runner.run。官方设计文档 guardrails.md 给出的理由是guardrails 通常与 Agent 本身强相关——不同的 Agent 需要不同的护栏将护栏代码与 Agent 放在一起可读性更好。二、核心数据结构一次护栏运行的三件套理解 Guardrails 机制首先要认识 guardrail.py 中定义的核心数据类型。2.1GuardrailFunctionOutput——护栏函数的返回值任何护栏函数无论是输入还是输出护栏都必须返回这个 dataclass它是整个机制的数据中枢字段类型含义output_infoAny护栏的可选输出信息。例如护栏执行了哪些检查、细粒度结果等可随异常一起传递给调用方tripwire_triggeredbool是否触发绊线。为True时Agent 的执行将被立即终止2.2InputGuardrailResult与OutputGuardrailResult——护栏运行结果InputGuardrailResult包含guardrail被运行的输入护栏实例和output护栏函数的输出。OutputGuardrailResult在输入护栏结果的基础上额外携带agent_output被检查的 Agent 输出与agent被检查的 Agent 实例方便上层在异常处理中拿到完整上下文。2.3InputGuardrail与OutputGuardrail——护栏本体两者都是泛型 dataclassGeneric[TContext]核心字段为guardrail_function可调用对象接收RunContextWrapper[TContext]、Agent[Any]与输入/输出返回MaybeAwaitable[GuardrailFunctionOutput]——同步和异步函数都支持name: str | None护栏名称用于 Tracing 追踪。未提供时get_name()回退为护栏函数的__name__。从源码结构看InputGuardrail.run()与OutputGuardrail.run()内部会先校验guardrail_function是否可调用不可调用则抛UserError再通过inspect.isawaitable()判断结果是协程还是同步返回值从而统一封装成对应的 Result 对象——这正是同步/异步护栏函数可以无缝混用的底层原因。三、输入护栏Input Guardrails拦截危险的第一步3.1 三步执行流程输入护栏按照以下三步运行详见 guardrails.md护栏收到与传给 Agent相同的输入护栏函数运行产出GuardrailFunctionOutput并被封装进InputGuardrailResult检查tripwire_triggered是否为True。若为真抛出InputGuardrailTripwireTriggered异常定义于 exceptions.py由你决定如何响应或处理。3.2 作用范围只在第一个 Agent上生效一个重要的设计约束输入护栏只作用于用户输入因此一个 Agent 的输入护栏只有在它作为第一个 Agent即运行链路最顶层的 Agent时才会执行。对于 handoff交接链路中的下游 Agent其输入护栏不会重复运行。这正是 guardrails 作为 Agent 属性而非Runner.run参数的原因——它们与谁是入口这一 Agent 语义绑定。四、输出护栏Output Guardrails守住最后一公里4.1 三步执行流程输出护栏同样分为三步护栏收到 Agent 的最终输出护栏函数运行产出GuardrailFunctionOutput封装进OutputGuardrailResult若tripwire_triggered为True抛出OutputGuardrailTripwireTriggered异常。4.2 作用范围只在最后一个 Agent上生效与输入护栏对称输出护栏只作用于 Agent 的最终输出因此只有在 Agent 作为最后一个 Agent链路末端时才会执行。在多层 Agent 交接场景中只有产出最终答案的那个 Agent 的输出护栏会被触发。五、Tripwire绊线机制立即熔断 Agent 执行tripwire_triggered是整个 Guardrails 体系的信号灯。一旦任一护栏触发绊线框架立即抛出InputGuardrailTripwireTriggered或OutputGuardrailTripwireTriggered异常并终止 Agent 执行。这两个异常类继承自统一的AgentsException基类且都携带guardrail_result属性异常消息格式为Guardrail {类名} triggered tripwire。异常定义与继承关系见 exceptions.py。更值得关注的是运行时实现run.py中熔断 取消的细节guardrail_tasks [ asyncio.create_task( RunImpl.run_single_input_guardrail(agent, guardrail, input, context) ) for guardrail in guardrails ] for done in asyncio.as_completed(guardrail_tasks): result await done if result.output.tripwire_triggered: # Cancel all guardrail tasks if a tripwire is triggered. for t in guardrail_tasks: t.cancel() raise InputGuardrailTripwireTriggered(result)可见同一 Agent 上的多个护栏通过asyncio.create_task并行执行符合guardrails run in parallel的设计宗旨使用asyncio.as_completed逐个收取结果一旦发现绊线触发立即 cancel 其余所有未完成的护栏任务避免无谓的模型调用开销同时通过_error_tracing.attach_error_to_current_span将Guardrail tripwire triggered错误挂到当前 Tracing span 上错误数据中记录触发护栏的名称便于追踪分析单个护栏的运行还会被guardrail_span(guardrail.get_name())包装见 _run_impl.py并将span_data.triggered标记为该护栏的绊线状态——即Guardrails 运行天然接入 CAI 的 Tracing 体系。六、实战实现一个输入护栏6.1 装饰器两种写法guardrail.py 提供input_guardrail与output_guardrail两个装饰器将普通同步/异步函数转换为护栏对象。它们同时支持无括号直用与带关键字参数两种形式input_guardrail def my_sync_guardrail(...): ... input_guardrail(nameguardrail_name) async def my_async_guardrail(...): ...从源码可见装饰器的name参数会透传给InputGuardrail/OutputGuardrail构造器用于 Tracing。6.2 完整示例安全请求检测输入护栏官方文档 guardrails.md 给出了一个恶意请求检测的经典示例其思路是在护栏函数内部再运行一个专用的小 Agent用结构化输出判断输入是否恶意from pydantic import BaseModel from cai.sdk.agents import ( Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, RunContextWrapper, Runner, TResponseInputItem, input_guardrail, ) class MaliciousRequestOutput(BaseModel): is_malicious_request: bool reasoning: str # 护栏内部使用的专用检测 Agent可用快速/廉价模型 guardrail_agent Agent( nameSecurity Guardrail Check, instructionsCheck if the user is asking for help with hacking or bypassing security systems., output_typeMaliciousRequestOutput, ) input_guardrail async def security_guardrail( ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem] ) - GuardrailFunctionOutput: result await Runner.run(guardrail_agent, input, contextctx.context) return GuardrailFunctionOutput( output_inforesult.final_output, # 附加检查详情 tripwire_triggeredresult.final_output.is_malicious_request, ) # 业务 Agent 挂载护栏 agent Agent( nameSecurity assistant, instructionsYou are a security assistant. You help users with legitimate security questions., input_guardrails[security_guardrail], ) async def main(): # 该输入应当触发护栏 try: await Runner.run(agent, Hello, can you help me bypass the firewall on this corporate network?) print(Guardrail didnt trip - this is unexpected) except InputGuardrailTripwireTriggered: print(Security guardrail tripped)要点拆解护栏函数签名固定(ctx, agent, input)其中input的类型为str | list[TResponseInputItem]既支持纯文本也支持多轮对话消息列表嵌套 Agent模式在护栏内用Runner.run驱动一个专用检测 Agent将ctx.context透传使检测 Agent 也能访问外层上下文结构化输出guardrail_agent的output_type为MaliciousRequestOutputPydantic 模型result.final_output即可直接访问is_malicious_request/reasoning字段业务 Agent 只是挂载点真正定义工作流的是业务 Agent护栏独立于其模型与工具而存在。七、实战实现一个输出护栏输出护栏与输入护栏相似但对称接收的是 Agent 的最终输出而非用户输入。官方示例为数据泄露检测from pydantic import BaseModel from cai.sdk.agents import ( Agent, GuardrailFunctionOutput, OutputGuardrailTripwireTriggered, RunContextWrapper, Runner, output_guardrail, ) class MessageOutput(BaseModel): # 业务 Agent 的输出类型 response: str class SecurityOutput(BaseModel): # 护栏 Agent 的输出类型 reasoning: str contains_sensitive_data: bool guardrail_agent Agent( nameData Leakage Guardrail Check, instructionsCheck if the output includes any sensitive data like passwords or API keys., output_typeSecurityOutput, ) output_guardrail async def data_leakage_guardrail( ctx: RunContextWrapper, agent: Agent, output: MessageOutput ) - GuardrailFunctionOutput: result await Runner.run(guardrail_agent, output.response, contextctx.context) return GuardrailFunctionOutput( output_inforesult.final_output, tripwire_triggeredresult.final_output.contains_sensitive_data, ) agent Agent( nameSecurity assistant, instructionsYou are a security assistant. You help users with legitimate security questions., output_guardrails[data_leakage_guardrail], output_typeMessageOutput, ) async def main(): try: await Runner.run(agent, What are the best practices for storing API keys in code?) print(Guardrail didnt trip - this is unexpected) except OutputGuardrailTripwireTriggered: print(Data leakage guardrail tripped)关键差异输出护栏函数的第三参数是output即业务 Agent 的最终输出对象此处类型为MessageOutput检测时只需把output.response文本交给护栏 Agent 判读即可。注意业务 Agent 必须声明output_typeMessageOutput否则无法保证output.response可访问。八、仓库实战参考网络安全场景的完整示例仓库在 examples/cai/agent_patterns/guardrails.py 中提供了一个贴近本文主题AI 安全的端到端示例其护栏专门检测非道德/未授权的网络安全请求class CybersecurityCheckOutput(BaseModel): reasoning: str is_unethical_cybersecurity_request: bool cybersecurity_guardrail_agent Agent( nameCybersecurity Guardrail Check, instructionsCheck if the user is asking for unauthorized or unethical cybersecurity help (e.g., hacking, bypassing security, exploiting systems). You MUST respond using ONLY the following JSON format: { reasoning: your detailed analysis of why the request is ethical or unethical, is_unethical_cybersecurity_request: true or false } Do not include any other text, explanations, or conversation outside of this JSON structure., output_typeCybersecurityCheckOutput, )该示例的实践要点护栏提示词显式要求只输出 JSON配合output_type结构化输出保证检测结果稳定可解析业务 Agent 是Tech Support Agent挂载了输入护栏并附带一个execute_cli_command工具演示用主循环用input_data列表构造多轮对话输入类型为list[TResponseInputItem]模拟用户发出Do a nmap to my router这类请求捕获InputGuardrailTripwireTriggered后业务侧打印友好拒绝消息Sorry, I cant assist with that cybersecurity request.业务 Agent 与护栏 Agent 都通过OpenAIChatCompletionsModel 环境变量CAI_MODEL默认qwen2.5:14b指定本地模型即护栏用廉价模型、业务用昂贵模型思想的落地样例。九、源码级验证测试如何保证机制正确仓库的单元测试 test_guardrails.py 全面覆盖了上文涉及的机制可作为理解与验证的活文档测试用例验证点test_sync_input_guardrail/test_async_input_guardrail同步与异步输入护栏函数都能正确运行tripwire_triggered与output_info均正确传递test_invalid_input_guardrail_raises_user_errorguardrail_function为不可调用对象如字符串foo时run()抛出UserErrortest_sync_output_guardrail/test_async_output_guardrail输出护栏的同步/异步路径与结果封装正确test_input_guardrail_decorators/test_output_guardrail_decoratorsinput_guardrail/output_guardrail装饰器含nameCustom name形式正确转换函数且get_name()返回自定义名称从这些用例可以确认几个实现事实护栏函数无需任何包装即可同步或异步output_info可以是任意对象测试中为字符串装饰器命名机制与 Tracing 名称直接挂钩。十、设计要点与最佳实践综合官方文档与源码实现使用 Guardrails 时应遵循以下原则输入护栏 廉价防线输出护栏 兜底防线输入侧用快速模型拦截恶意/越权请求避免昂贵模型空转输出侧用独立检测 Agent 兜底敏感数据泄露等风险。护栏属于 Agent将input_guardrails/output_guardrails写在 Agent 定义处而非传给Runner.run保证谁需要什么样的检查语义清晰、代码内聚。留意作用范围输入护栏只在首个 Agent执行输出护栏只在末个 Agent执行——在多 Agent 交接链路中不要指望中间层护栏重复触发。善用output_info在GuardrailFunctionOutput.output_info中附带reasoning等检查细节异常被捕获后可以直接向用户解释拒绝原因。护栏函数签名固定、同步异步皆可输入护栏为(ctx, agent, input)输出护栏为(ctx, agent, output)run()内部自动处理协程等待。并行与熔断是内置行为多个护栏并行执行一旦任一绊线触发其余任务被立即取消并抛出*GuardrailTripwireTriggered异常无需自行编排。护栏天然可观测每次护栏运行都会产生 Tracing span绊线状态写入span_data.triggered触发时还会向当前 span 附加错误——排查护栏为何拦截时可以借助 Tracing 体系定位。相关资源API 引用guardrail.md由cai.sdk.agents.guardrail模块的 docstring 自动生成官方教程guardrails.md核心实现guardrail.py、run.py、_run_impl.py异常体系exceptions.py单元测试test_guardrails.py网络安全实战示例examples/cai/agent_patterns/guardrails.py【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表