CAI 示例代码库全解析从基础用法、Agent 设计模式到完整业务系统的实战指南【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai本文以 CAICybersecurity AI仓库的 examples 目录 为主线系统梳理仓库内置的全部示例分类从basic基础能力、agent_patterns设计模式到tools、model providers、handoffs再到customer_service与research_bot两个完整业务系统并补充本仓库特有的cai网络安全场景示例。读完本文你将掌握 CAI SDK 的核心 APIAgent、Runner、trace、handoff、function_tool等在实际代码中的组合方式能够直接以这些示例为模板搭建自己的多 Agent 应用。示例仓库概览一份可直接运行的多场景代码库原文档docs/examples.md将示例按能力划分为多个类别本仓库在examples/下实际提供了远超文档列举的更丰富实现。整体布局如下examples/ ├── agent_patterns/ # 常见 Agent 设计模式确定性流程、Agent 即工具、并行执行等 ├── basic/ # SDK 基础能力Hello World、动态系统提示、流式输出、生命周期 ├── cai/ # CAI 网络安全专属示例Agent 模式 提示注入 PoC 基础用法 ├── customer_service/ # 完整业务示例航空公司客服系统 ├── research_bot/ # 完整业务示例深度研究机器人 ├── handoffs/ # Agent 交接handoff与消息过滤 ├── mcp/ # MCP 协议集成文件系统、Git、SSE ├── model_providers/ # 非 OpenAI 模型接入 ├── tools/ # OAI 托管工具Web 搜索、文件搜索、计算机使用 ├── voice/ # 语音静态/流式工作流示例 └── financial_research_agent/ # 多 Agent 金融研究系统运行示例前需先完成 SDK 安装与环境变量配置API Key 等安装方式见 docs/cai/getting-started/installation.md 与 docs/cai/getting-started/configuration.md。示例基本都遵循同一结构async def main()asyncio.run(main())可以直接python examples/xxx.py运行。basic掌握 SDK 基础能力examples/basic/用于演示 SDK 最基础、最常用的能力对应文档所述的三类核心能力动态系统提示、流式输出、生命周期事件。最小可用示例Hello Worldexamples/basic/hello_world.py 展示了创建一个 Agent 并运行的最小闭环import asyncio from agents import Agent, Runner async def main(): agent Agent( nameAssistant, instructionsYou only respond in haikus., ) result await Runner.run(agent, Tell me about recursion in programming.) print(result.final_output) if __name__ __main__: asyncio.run(main())核心只有三个 APIAgent(...)定义智能体nameinstructions为必填关键项、Runner.run(agent, input)同步执行一次对话、result.final_output取最终输出文本。这是理解后续所有复杂示例的基石。仓库中另有 examples/basic/hello_world_jupyter.py 面向 Jupyter 环境的版本。动态系统提示随上下文变化的 instructionsexamples/basic/dynamic_system_prompt.py 演示了instructions参数不传字符串、而是传入一个函数的用法from agents import Agent, RunContextWrapper, Runner class CustomContext: def __init__(self, style: Literal[haiku, pirate, robot]): self.style style def custom_instructions( run_context: RunContextWrapper[CustomContext], agent: Agent[CustomContext] ) - str: context run_context.context if context.style haiku: return Only respond in haikus. elif context.style pirate: return Respond as a pirate. else: return Respond as a robot and say beep boop a lot. agent Agent(nameChat agent, instructionscustom_instructions) result await Runner.run(agent, user_message, contextCustomContext(stylechoice))要点instructions可以是(RunContextWrapper[TContext], Agent[TContext]) - str的函数Runner.run(..., context...)传入的自定义上下文对象会在每次运行前被读取从而实现「同一 Agent、不同上下文、不同行为」。这在实际业务中常用于按用户画像、租户、会话状态动态生成提示词。流式输出与生命周期事件examples/basic/stream_text.py通过Runner.run_streamed()逐 token 流式接收文本输出适合聊天类 UI。examples/basic/stream_items.py按 item 粒度流式消费消息、工具调用等结构化事件。examples/basic/lifecycle_example.py 与 examples/basic/agent_lifecycle_example.py演示 Agent 生命周期钩子hook可在运行开始/结束、工具调用前后等时机插入自定义逻辑与 SDK 中 src/cai/sdk/agents/lifecycle.py 的生命周期机制对应。examples/basic/usage_tracking_example.py展示用量统计token 消耗等相关实现见 src/cai/sdk/agents/global_usage_tracker.py。agent_patterns常见 Agent 设计模式文档明确列出了本类别的三大核心模式仓库中均有对应实现且都使用了with trace(...)将整个工作流包进一次追踪tracing 详见 docs/ref/tracing/index.md。确定性工作流Deterministic workflowsexamples/agent_patterns/deterministic.py 演示「步骤化流水线」每个步骤由专门 Agent 完成前一步输出作为后一步输入中间可加校验门gate。story_outline_agent Agent( namestory_outline_agent, instructionsGenerate a very short story outline based on the users input., ) class OutlineCheckerOutput(BaseModel): good_quality: bool is_scifi: bool outline_checker_agent Agent( nameoutline_checker_agent, instructionsRead the given story outline, and judge the quality. ..., output_typeOutlineCheckerOutput, # 结构化输出 ) async def main(): with trace(Deterministic story flow): outline_result await Runner.run(story_outline_agent, input_prompt) outline_checker_result await Runner.run(outline_checker_agent, outline_result.final_output) # 门控质量不达标或非科幻题材则终止 if not outline_checker_result.final_output.good_quality: exit(0) if not outline_checker_result.final_output.is_scifi: exit(0) story_result await Runner.run(story_agent, outline_result.final_output)这个示例还演示了output_type结构化输出让 Agent 输出 Pydantic 模型OutlineCheckerOutputfinal_output即可直接以类型安全的方式访问good_quality/is_scifi字段是「Agent 产出可编程数据」的标准写法。Agent 即工具Agents as toolsexamples/agent_patterns/agents_as_tools.py 演示编排模式一个「前台」Agent 把多个翻译子 Agent 注册为工具由 LLM 自行决定调用哪个。spanish_agent Agent( namespanish_agent, instructionsYou translate the users message to Spanish, handoff_descriptionAn english to spanish translator, ) orchestrator_agent Agent( nameorchestrator_agent, instructions(You are a translation agent. You use the tools given to you to translate. ...), tools[ spanish_agent.as_tool(tool_nametranslate_to_spanish, tool_descriptionTranslate the users message to Spanish), french_agent.as_tool(tool_nametranslate_to_french, ...), italian_agent.as_tool(tool_nametranslate_to_italian, ...), ], )关键 API 是agent.as_tool(tool_name..., tool_description...)把子 Agent 包装成可供 LLM 调用的工具让大模型在推理过程中动态选择子 Agent实现动态路由。示例还演示了用synthesizer_agentorchestrator_result.to_input_list()把前序运行结果作为新 Agent 的输入以及遍历new_items用ItemHelpers.text_message_output(item)提取各步骤文本。并行执行Parallel agent executionexamples/agent_patterns/parallelization.py 演示并行化模式用asyncio.gather同时跑三个西班牙语翻译 Agent再用一个translation_pickerAgent 选出最佳结果——即经典的「并行采样 LLM 评选」res_1, res_2, res_3 await asyncio.gather( Runner.run(spanish_agent, msg), Runner.run(spanish_agent, msg), Runner.run(spanish_agent, msg), ) outputs [ItemHelpers.text_message_outputs(r.new_items) for r in (res_1, res_2, res_3)] best_translation await Runner.run(translation_picker, fInput: {msg}\n\nTranslations:\n{translations})更多模式examples/agent_patterns/forcing_tool_use.py强制 Agent 使用指定工具。examples/agent_patterns/routing.py路由模式按输入分发到不同 Agent。examples/agent_patterns/input_guardrails.py 与 examples/agent_patterns/output_guardrails.py输入/输出护栏相关机制见 src/cai/sdk/agents/guardrail.py。examples/agent_patterns/llm_as_a_judge.pyLLM 作为裁判评估输出质量。对应测试可参考 tests/agents/test_agent_one_tool.py 与 tests/tools/ 下的工具行为测试帮助理解各模式的边界行为。toolsOAI 托管工具集成文档指出本类别讲解如何实现并集成 OAI 托管的工具如 web search 与 file search仓库中examples/tools/提供三个对应示例examples/tools/web_search.py通过WebSearchTool让 Agent 具备联网搜索能力from agents import Agent, Runner, WebSearchTool agent Agent( nameWeb searcher, instructionsYou are a helpful agent., tools[WebSearchTool(user_location{type: approximate, city: New York})], ) result await Runner.run(agent, search the web for local sports news ...)examples/tools/file_search.py文件搜索工具用于在指定的文件集合中检索信息适合 RAG 场景。examples/tools/computer_use.py计算机使用computer use能力示例相关实现见 src/cai/sdk/agents/computer.py。托管工具以「普通工具列表」的形式注入Agent(tools[...])对 Agent 而言与自定义function_tool完全一致切换成本低。model providers接入非 OpenAI 模型SDK 通过模型 Provider 抽象支持非 OpenAI 模型examples/model_providers/提供了完整接入样例对应文档「Explore how to use non-OpenAI models with the SDK」examples/model_providers/custom_example_agent.py在单 Agent 上使用自定义 Provider。examples/model_providers/custom_example_global.py通过全局配置为所有 Agent 设置自定义模型。examples/model_providers/custom_example_provider.py实现一个自定义 Provider 类的完整流程。examples/model_providers/litellm.py 与 examples/model_providers/litellm_config.yaml通过 LiteLLM 接入数百种模型。模型层实现位于 src/cai/sdk/agents/models/其中interface.py定义了 Provider 必须实现的接口openai_provider.py、openai_chatcompletions.py、openai_responses.py是内置实现。CAI 场景下还可在 examples/cai/basic_usage.py 中看到通过OpenAIChatCompletionsModel直接指定本地模型如qwen2.5:14b的用法。更多部署细节见 docs/providers/如 docs/providers/ollama.md、docs/providers/azure.md、docs/providers/openrouter.md。handoffsAgent 交接实战examples/handoffs/演示多 Agent 之间如何交接handoff并展示交接时如何过滤消息历史。examples/handoffs/message_filter.py 的核心是一个自定义交接过滤器from agents import Agent, HandoffInputData, Runner, function_tool, handoff, trace from agents.extensions import handoff_filters def spanish_handoff_message_filter(handoff_message_data: HandoffInputData) - HandoffInputData: # 1) 移除历史中的所有工具消息 handoff_message_data handoff_filters.remove_all_tools(handoff_message_data) # 2) 再丢弃前两条历史演示用 history tuple(handoff_message_data.input_history[2:]) return HandoffInputData( input_historyhistory, pre_handoff_itemstuple(handoff_message_data.pre_handoff_items), new_itemstuple(handoff_message_data.new_items), ) second_agent Agent( nameAssistant, instructionsBe a helpful assistant. If the user speaks Spanish, handoff to the Spanish assistant., handoffs[handoff(spanish_agent, input_filterspanish_handoff_message_filter)], )示例先用普通 Agent 完成两轮对话其中包含工具调用再触发一次西班牙语交接最终打印to_input_list()确认过滤效果工具消息被移除、前两条历史被丢弃。交接机制的底层实现见 src/cai/sdk/agents/handoffs.py 与 src/cai/sdk/agents/extensions/handoff_filters.py。流式版本参见 examples/handoffs/message_filter_streaming.py。customer_service完整的航空公司客服系统文档将其定位为「更完整的业务示例」examples/customer_service/main.py 实现了一个可实际运行的航空公司客服系统是理解「多 Agent 工具 交接 钩子 上下文」组合的最佳范本包含四大要素会话上下文AirlineAgentContext(BaseModel)保存乘客名、确认号、座位号、航班号由RunContextWrapper在工具间共享。工具function_tool装饰器定义faq_lookup_toolFAQ 查询支持name_override/description_override与update_seat接收RunContextWrapper可读写上下文。交接与钩子triage_agent通过handoffs[faq_agent, handoff(agentseat_booking_agent, on_handoffon_seat_booking_handoff)]分派任务on_seat_booking_handoff钩子在交接发生时自动为乘客生成航班号写入上下文。推荐提示前缀各 Agent 的 instructions 都拼接了from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX实现见 src/cai/sdk/agents/extensions/handoff_prompt.py让模型更可靠地执行交接。主循环展示了真实聊天应用的标准写法维护current_agent与input_items每次用户输入用trace(Customer service, group_idconversation_id)包裹遍历result.new_items按MessageOutputItem/HandoffOutputItem/ToolCallItem/ToolCallOutputItem分类打印最后input_items result.to_input_list()、current_agent result.last_agent进入下一轮。research_bot深度研究机器人examples/research_bot/是一个「简单深度研究克隆」采用「规划 → 并行搜索 → 流式写报告」三段式流水线是多 Agent 协作 流式 追踪的完整样板examples/research_bot/agents/planner_agent.py规划 Agent以WebSearchPlanPydantic 模型含 520 条带理由的搜索词为output_type输出结构化搜索计划。examples/research_bot/agents/search_agent.py搜索 Agent负责执行单条 Web 搜索。examples/research_bot/agents/writer_agent.py写作 Agent输出ReportData含 markdown 报告、摘要、追问问题。examples/research_bot/manager.py编排层ResearchManager用trace(Research trace, trace_idgen_trace_id())包裹全程_perform_searches用asyncio.create_taskasyncio.as_completed并行执行所有搜索并用custom_span(Search the web)打点_write_report使用Runner.run_streamed边流式生成边更新进度文案。examples/research_bot/main.py入口接收查询并调用ResearchManager().run(query)。printer.py基于 rich 提供实时进度展示sample_outputs/中保留了该系统的真实输出样例如vacation.md、product_recs.md可作为观察最终报告形态的参考。cai网络安全场景专属示例本仓库在examples/cai/下额外提供了网络安全AI Security场景的示例是区别于通用 SDK 的项目特色也是agent_patterns与basic的 CAI 化落地examples/cai/agent_patterns/确定性流程、Agent 即工具、并行化、护栏、交接、交接工具组合在安全场景下的复刻版本。examples/cai/prompt_injections/16 个提示注入 PoCpoc1.txt~poc16.txt与配套server.py可用于验证 Agent 对注入攻击的防御能力与 docs/cai_prompt_injection.md 呼应。examples/cai/basic_usage.py网络安全版基础用法——定义带 CLI 命令工具的「CTF Agent」使用OpenAIChatCompletionsModel接入本地模型默认qwen2.5:14b可用环境变量CAI_MODEL覆盖并同时演示普通运行与流式运行两种模式。examples/cai/simple_one_tool_test.py 与 examples/cai/simple_one_tool_test_streamed.py单工具最小测试普通 流式。examples/cai/test_guardrails.py 与 examples/cai/test_guardrails_enhanced.py安全场景护栏验证。注意examples/cai/下的示例直接使用from cai.sdk.agents import ...的显式导入而其他目录示例多使用from agents import ...的短名导入——两者指向同一套 SDK 接口src/cai/sdk/agents/可按项目约定任选其一。延伸更多可用示例与配套资料除上述文档列举的分类外examples/还包含examples/mcp/MCP 协议集成文件系统、Git、SSE 三种服务端示例配套文档见 docs/ref/mcp/ 与 docs/cai/getting-started/MCP.md。examples/voice/语音静态与流式工作流示例配套文档见 docs/voice/。examples/financial_research_agent/规划、搜索、风控、财务分析、验证、写作六 Agent 协作的金融研究系统。examples/continue_mode_jokes.py 与 examples/continue_mode_security_audit.py连续模式多轮续跑示例与 docs/continue_mode.md 对应。如需验证示例背后的 SDK 行为tests/目录提供了覆盖 Agent 运行tests/agents/、工具tests/tools/、追踪tests/tracing/、MCPtests/mcp/、语音tests/voice/等维度的测试用例可作为理解各 API 边界行为的权威参考。综上从三行 Hello World 到完整的客服系统与深度研究机器人examples/构成了一个由浅入深、可直接复制改造的 CAI 实战知识库。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考