CAI 中的 OpenAI Agents SDKAgent、Handoff、Guardrail 三大原语与安全智能体实战【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai本篇技术指南以 docs/index2.md 为核心骨架系统讲解 Cybersecurity AICAI框架内置的 Agents SDK如何在 CAI 中使用极少抽象实现Agent 循环 函数工具 任务交接 输入输出护栏 可观测追踪并落到 CTF 攻防、网络安全支撑等实际场景中。读完你将掌握Agent/Runner的同步与流式运行、function_tool工具化、handoff多智能体协作、input/output_guardrail安全校验以及通过 tracing 调试多步骤工作流。什么是 Agents SDK轻量、生产级的智能体原语CAI 的 SDK 位于 src/cai/sdk/agents/在 API 层面延续了 OpenAI Agents SDK 的设计哲学功能足够强大、值得使用但原语足够少、上手足够快。整个 SDK 只围绕三个核心抽象展开Agents智能体配备了指令instructions与工具tools的 LLM 实例Handoffs任务交接允许一个智能体把特定任务委派给另一个智能体Guardrails护栏对智能体的输入/输出进行校验不满足条件时提前中断。这三个原语组合起来足以表达工具与智能体之间复杂的调用关系。官方文档强调其两大设计原则功能足够多到值得使用但原语足够少到易于学习开箱即用同时又允许你精确自定义每一步的行为。从实现上看这三大原语在源码中分别对应 agent.py 中的Agent数据类、handoffs.py 中的Handoff与 guardrail.py 中的InputGuardrail/OutputGuardrail。此外 SDK 还内置了基于 spans.py 的 Trace/Span 体系用于可视化、调试和评估你的智能体流程。在 CAI 项目中这套 SDK 被大量用于构建网络安全场景下的智能体examples/cai/agent_patterns/目录提供了handoffs.py、guardrails.py、deterministic.py、LLM_as_judge.py、agent_as_tool.py、paralelization.py等可直接运行的实战范例。SDK 的主要能力清单Agent 循环Agent loop内置循环负责调用工具、把结果回传给 LLM并持续迭代直到 LLM 输出最终结果Python 优先用 Python 语言本身的特性编排与链式组合多个智能体无需学习新抽象Handoffs在多个智能体之间协调与委派任务的强大机制Guardrails与智能体并行执行输入校验校验失败时提前中断函数工具Function tools把任意 Python 函数变成工具自动生成 JSON Schema 并借助 Pydantic 完成参数校验Tracing内置追踪能力可视化、调试与监控工作流还可接入 OpenAI 的评估、微调与蒸馏工具链。安装与环境准备pip install openai-agents运行示例前必须配置 OpenAI API Key 环境变量export OPENAI_API_KEYsk-...需要说明的是默认模型走 OpenAI 的 Responses API见 _config.py 的set_default_openai_api但 CAI 内置的OpenAIChatCompletionsModel也支持通过OpenAIAsyncClient接入其他兼容 Chat Completions 的本地模型端点详见下文示例。此外 run.py 还支持CAI_MAX_TURNS与CAI_PRICE_LIMIT两个环境变量分别用于限制整个运行的最大轮数与费用上限默认均为无限。Hello World第一个智能体文档给出了最简示例可以直接保存运行from cai.sdk.agents import Agent, Runner agent Agent(nameAssistant, instructionsYou are a helpful assistant) result Runner.run_sync(agent, Write a haiku about recursion in programming.) print(result.final_output) # Code within the code, # Functions calling themselves, # Infinite loops dance.核心调用链非常清晰Agent只是一个携带配置的数据类真正执行靠 run.py 中的Runner。Runner.run_sync()以同步方式执行一轮完整 agent loop返回 result.py 中的RunResult其final_output即最终文本。Agent 的关键构造参数从 Agent 数据类定义 可以看到完整可配置项参数类型默认值说明namestr必填智能体名称同时用于 handoff 工具命名instructionsstr/CallableNone系统提示词传入函数时可动态生成get_system_prompt会识别协程descriptionstrNoneCLI 中展示的描述handoff_descriptionstrNone作为 handoff 目标时向 LLM 展示的能力说明handoffslist[Agent \| Handoff][]可委派任务的子智能体列表modelstr/Model默认模型未设置时使用model_settings.DEFAULT_MODELmodel_settingsModelSettings默认实例温度、top_p 等模型级调参toolslist[Tool][]可用工具列表mcp_serverslist[MCPServer][]MCP 服务器需自行管理connect()/cleanup()生命周期input_guardrailslist[InputGuardrail][]输入护栏仅在链首智能体上运行output_guardrailslist[OutputGuardrail][]输出护栏仅在产生最终输出后运行output_typetype[Any]None结构化输出类型缺省为strhooksAgentHooksNone生命周期事件回调tool_use_behavior字面量/名单/函数run_llm_again控制工具结果是否回传 LLM详见下文其中tool_use_behavior支持四种取值默认run_llm_again工具结果回传 LLM 继续推理、stop_on_first_tool首个工具输出直接作为最终输出、工具名列表命中即停止、或自定义ToolsToFinalOutputFunction回调agent.py。注意该配置仅对 FunctionTool 生效FileSearchTool等托管工具始终由 LLM 处理。Agent还提供了两个实用方法clone(**kwargs)浅拷贝出改参后的新实例as_tool()把智能体封装为可被其他智能体调用的函数工具——这与 handoff 的本质区别在于handoff 会传递对话历史并由新智能体接管对话而as_tool()只接收生成的新输入、对话仍由原智能体继续。把函数变成工具function_tool 与 Pydantic 校验文档强调 SDK 的Python 优先与函数工具能力核心装饰器为 tool.py 中的function_tool。它自动完成三件事解析函数签名生成参数 JSON Schema、用 docstring 生成工具描述、用 docstring 生成参数说明。strict_mode默认为True保证 JSON Schema 处于严格模式以提升 LLM 参数生成的正确率。参考 examples/basic/tools.py 的完整写法import asyncio from pydantic import BaseModel from agents import Agent, Runner, function_tool class Weather(BaseModel): city: str temperature_range: str conditions: str function_tool def get_weather(city: str) - Weather: print([debug] get_weather called) return Weather(citycity, temperature_range14-20C, conditionsSunny with wind.) agent Agent( nameHello world, instructionsYou are a helpful agent., tools[get_weather], ) async def main(): result await Runner.run(agent, inputWhats the weather in Tokyo?) print(result.final_output) if __name__ __main__: asyncio.run(main())要点解读返回值使用 PydanticBaseModelSDK 会据此做结构化校验与序列化返回给 LLM若函数第一个参数是RunContextWrapper它必须与使用该工具的 Agent 的 context 类型一致见 tool.py 的约定工具执行失败时默认调用default_tool_error_function返回通用错误串也可通过failure_error_function自定义tool.py。除函数工具外tool.py 还定义了三类托管工具FileSearchTool向量库检索、WebSearchTool联网搜索可配user_location与search_context_size、ComputerTool计算机操作对应computer_use_preview。流式输出Runner.run_streamed当需要边生成边展示如终端打字机效果时使用流式运行。参考 examples/basic/stream_text.pyimport asyncio from openai.types.responses import ResponseTextDeltaEvent from agents import Agent, Runner async def main(): agent Agent( nameJoker, instructionsYou are a helpful assistant., ) result Runner.run_streamed(agent, inputPlease tell me 5 jokes.) async for event in result.stream_events(): if event.type raw_response_event and isinstance(event.data, ResponseTextDeltaEvent): print(event.data.delta, end, flushTrue) if __name__ __main__: asyncio.run(main())流式事件类型包括RunItemStreamEvent、AgentUpdatedStreamEvent与RawResponsesStreamEvent见 stream_events.py通过result.to_input_list()可以把中间产物喂回下一轮对话实现多轮连续交互。Handoffs多智能体协作与任务委派文档将 Handoffs 列为三大原语之一智能体可以把任务委派给其他智能体实现关注点分离与模块化。Handoff 的语义在 handoffs.py 中有精确定义其典型场景是一个 triage agent 决定由哪个子智能体处理请求。最简单的方式是直接把子 Agent 放进父 Agent 的handoffs列表。下面这段代码取自 examples/cai/agent_patterns/handoffs.py展示了 CTF 场景下CTF 攻防智能体将 flag 提取任务交接给Flag 判别智能体from cai.sdk.agents import Agent, OpenAIChatCompletionsModel, Runner, function_tool, handoff from cai.tools.common import run_command function_tool def execute_cli_command(command: str) - str: return run_command(command) flag_discriminator Agent( nameFlag discriminator, descriptionAgent focused on extracting the flag from the output, instructionsYou are an agent tailored to extract the flag from a given output., modelOpenAIChatCompletionsModel( modelos.getenv(CAI_MODEL, qwen2.5:14b), openai_clientAsyncOpenAI(), ), ) ctf_agent Agent( nameCTF agent, descriptionAgent focused on conquering security challenges, instructionsYou are a Cybersecurity expert Leader facing a CTF, tools[execute_cli_command], modelOpenAIChatCompletionsModel( modelos.getenv(CAI_MODEL, qwen2.5:14b), openai_clientAsyncOpenAI(), ), handoffs[flag_discriminator], )注意这里使用OpenAIChatCompletionsModelAsyncOpenAI()把模型切换为兼容 Chat Completions 的本地端点默认模型名qwen2.5:14b可用CAI_MODEL环境变量覆盖——这是 CAI 在网络安全演练中接入本地 LLM 的典型方式。编程式 handoff 与输入过滤除了声明式列表SDK 还提供handoff()工厂函数handoffs.py支持三类用法handoff(agent)最简形式工具名自动生成为transfer_to_{agent_name}handoff(agent, on_handofffunc, input_typeType)携带输入参数input_type经TypeAdapter校验后传给on_handoff回调handoff(agent, input_filterfunc)自定义传给下一个智能体的输入如裁剪过长的历史消息见 handoffs.py 的HandoffInputFilter。默认情况下被交接的智能体会看到完整对话历史HandoffInputData提供input_history、pre_handoff_items、new_items三段数据handoffs.py配合 extensions/handoff_filters.py 可以精细控制传递内容。若希望所有 handoff 统一过滤可在RunConfig.handoff_input_filter设置全局过滤器run.py。Guardrails输入/输出护栏与安全校验文档强调 Guardrails 用于验证智能体的输入。SDK 将其分为两类guardrail.pyInputGuardrail在生成响应前与智能体执行并行运行仅对链首智能体生效tripwire_triggeredTrue时立即中断并抛出InputGuardrailTripwireTriggeredOutputGuardrail对最终输出运行校验触发时抛出OutputGuardrailTripwireTriggered。两者的守卫函数都返回GuardrailFunctionOutputoutput_infotripwire_triggered。装饰器input_guardrail/output_guardrail支持带括号与不带括号两种用法函数可为同步或异步。在 CAI 的网络安全语境下护栏最常见用途是拦截越权/非法的安全请求。参考 examples/cai/agent_patterns/guardrails.py 的完整实现from pydantic import BaseModel from cai.sdk.agents import ( Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, RunContextWrapper, Runner, TResponseInputItem, input_guardrail, OpenAIChatCompletionsModel, function_tool, ) 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: ..., is_unethical_cybersecurity_request: true or false }, output_typeCybersecurityCheckOutput, modelOpenAIChatCompletionsModel( modelos.getenv(CAI_MODEL, qwen2.5:14b), openai_clientAsyncOpenAI(), ), ) input_guardrail async def cybersecurity_guardrail( context: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem] ) - GuardrailFunctionOutput: result await Runner.run(cybersecurity_guardrail_agent, input, contextcontext.context) final_output result.final_output_as(CybersecurityCheckOutput) return GuardrailFunctionOutput( output_infofinal_output, tripwire_triggeredfinal_output.is_unethical_cybersecurity_request, )随后把护栏挂到业务智能体上并用异常捕获处理触雷async def main(): agent Agent( nameTech Support Agent, instructionsYou are a tech support agent. Help users with ethical and legal technical support requests only., input_guardrails[cybersecurity_guardrail], tools[execute_cli_command], modelOpenAIChatCompletionsModel( modelos.getenv(CAI_MODEL, qwen2.5:14b), openai_clientAsyncOpenAI(), ), ) input_data [{role: user, content: Do a nmap to my router}] try: result await Runner.run(agent, input_data) print(Agent output:, result.final_output) except InputGuardrailTripwireTriggered: print(Sorry, I cant assist with that cybersecurity request.)这套LLM 判官 结构化输出 tripwire 中断的模式是 CAI 中执行安全合规拦截的标准做法先用一个专用判别 Agent 分析输入是否属于越权安全请求再以 Pydantic 结构解析结果并决定是否中断主流程。相关异常类集中在 exceptions.py除了InputGuardrailTripwireTriggered/OutputGuardrailTripwireTriggered外还有MaxTurnsExceeded、ModelBehaviorError、UserError等。Tracing可视化调试与工作流编排文档提到 SDK 自带 built-in tracing。CAI 中通过 tracing/spans.py 的Span抽象 与 tracing/scope.py 的作用域管理 实现每个运行产生一个Trace内部由多个Span如AgentSpanData、FunctionSpanData、GenerationSpanData、HandoffSpanData、GuardrailSpanData组成层级结构。from cai.sdk.agents import trace, Runner with trace(workflow_nameCTF Workflow): result await Runner.run(ctf_agent, inputList all files in the current directory)自定义处理器可继承TracingProcessor并通过add_trace_processor()注册生产环境可用set_tracing_disabled(True)关闭追踪或通过set_tracing_export_api_key()指定导出密钥见 tracing 模块导出。RunConfig还支持workflow_name、trace_id、trace_include_sensitive_data等追踪参数run.py其中trace_include_sensitive_dataFalse可避免工具输入输出等敏感数据进入 trace。进阶RunConfig 全局配置Runner.run()可接受 RunConfig 进行全局控制字段默认值说明modelNone覆盖所有 Agent 的模型model_providerOpenAIProvider字符串模型名的解析提供者model_settingsNone全局模型参数非空值覆盖 Agent 级设置handoff_input_filterNone全局交接输入过滤器input_guardrails/output_guardrailsNone全局输入/输出护栏tracing_disabledFalse关闭本次运行的追踪trace_include_sensitive_dataTrue是否在 trace 中包含敏感数据workflow_nameAgent workflow本次运行的逻辑名称trace_idNone自定义 trace ID实战组合一个带护栏、工具与交接的完整流程综合以上原语可以搭建一条CTF 攻防 → 命令执行 → flag 提取的完整流水线输入护栏拦截非法请求越权渗透、绕过安全机制等CTF 主智能体持有execute_cli_command函数工具执行真实命令行操作需要提取 flag 时通过handoff交接给Flag discriminator子智能体整个过程用with trace(...)包裹产出可调试的完整调用链。对应实现分别见 examples/cai/agent_patterns/guardrails.py 与 examples/cai/agent_patterns/handoffs.py两文件可直接python运行需先配置OPENAI_API_KEY或本地 Chat Completions 端点。更多模式——确定性流程deterministic.py、智能体作为工具agent_as_tool.py、LLM 评审LLM_as_judge.py、并行化paralelization.py——也都在同一目录下提供参考。小结通过本文你可以确认CAI 的 Agents SDK 以Agent智能体→ Runner执行器→ 工具/交接/护栏行为约束→ Trace观测这条主线用极少的原语覆盖了智能体应用的全部关键环节。对网络安全场景而言最有价值的组合是用input_guardrail守住合规边界用function_tool赋予命令执行能力用handoff实现任务专业化分工再用 tracing 复盘整条攻防链路。相关源码、示例与测试如 tests/agents/ 下的 runner、guardrails 用例均可作为进一步深入研究的入口。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考