CAI Agents SDK 内置追踪Tracing体系详解Trace、Span 与自定义导出链路【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/caiCAICybersecurity AI的 Agents SDK 在 src/cai/sdk/agents/tracing 模块中内置了一套完整的可观测性Observability能力能在一次 Agent 运行期间自动记录 LLM 生成、工具调用、Handoff、Guardrail 以及自定义事件的全量执行轨迹。本文将带你理解 Trace 与 Span 的数据模型、默认追踪行为、手动埋点方式、敏感数据脱敏配置以及如何通过自定义 Trace Processor 将追踪数据导出到任意后端为安全 Agent 的调试与线上监控提供完整方案。注意根据 docs/tracing.md 顶部声明的当前状态本仓库中的 tracing 功能目前处于禁用状态项目正在按 OpenTelemetry 标准重新实现该功能新的实现将在未来版本中提供。本文所述内容与源码结构反映的是当前仓库的真实实现与设计意图使用前请确认你所安装版本的实际情况。图为 CAI 的 Trace 可视化面板左侧树状结构展示了从 CLI 入口到 Red Team Agent、generic_linux_command、search_github、ChatCompletion等完整调用链及每步耗时右侧展示各角色的输入输出消息与整体状态。核心概念Trace 与 Span追踪系统的两个基础抽象是 Trace 与 Span其定义可在 traces.py 与 spans.py 中找到。Trace一条完整的工作流Trace代表一次端到端的工作流workflow操作由多个 Span 组合而成对应源码中Trace抽象类的定义。Trace 拥有以下属性属性含义说明workflow_name逻辑工作流或应用的名称例如 Code generation、Customer service 或安全场景下的 Red team operationtrace_idTrace 的唯一 ID不传时自动生成格式必须为trace_32位字母数字源码中通过gen_trace_id()见 util.py用 UUID 生成group_id可选的组 ID用于把同一会话的多个 Trace 关联起来例如聊天线程 IDdisabled是否禁用记录为True时该 Trace 不会被记录metadata可选的附加元数据任意字典可携带用户自定义信息从源码看TraceImpl的export()方法会将 Trace 导出为{object: trace, id, workflow_name, group_id, metadata}结构而NoOpTrace则代表一个不会被记录的空操作 Trace禁用时的占位对象。Span一个有起止时间的操作Span表示一个具有开始与结束时间的操作拥有以下属性started_at与ended_at时间戳ISO 8601 格式由util.time_iso()生成使用 UTC 时区trace_id标识它所属的 Traceparent_id指向其父 Span如果有由此形成嵌套层级span_dataSpan 的具体信息载体。例如AgentSpanData包含 Agent 信息、GenerationSpanData包含 LLM 生成信息等。SpanImpl在start()时记录started_at并通知处理器在finish()时记录ended_at它还支持set_error(SpanError)记录错误信息SpanError为{message, data}结构。所有 SpanData 类型都定义在 span_data.py 中每个类型都有对应的type标识与export()导出方法包括AgentSpanDatatype: agentAgent 名称、可用 Handoff 列表、工具列表、输出类型FunctionSpanDatatype: function函数名、输入、输出以及可选的mcp_dataGenerationSpanDatatype: generationLLM 输入消息序列、输出序列、模型名、模型配置、usage 用量ResponseSpanDatatype: responseOpenAI Response 对象及其 IDHandoffSpanDatatype: handofffrom_agent/to_agentGuardrailSpanDatatype: guardrail守卫名称与是否触发triggeredCustomSpanDatatype: custom自定义名称与任意结构化数据TranscriptionSpanDatatype: transcription语音转文本的输入base64 PCM、输入格式、输出、模型SpeechSpanDatatype: speech文本转语音的输入、输出音频base64 PCM、输出格式、模型、首字节时间first_content_atSpeechGroupSpanDatatype: speech-group关联语音 Span 的分组容器MCPListToolsSpanDatatype: mcp_toolsMCP 服务器的 list tools 调用记录。默认追踪行为开箱即用的全链路记录默认情况下 SDK 会自动在以下节点包裹 Span你无需编写任何埋点代码触发点自动生成的 SpanRunner.run()/Runner.run_sync()/Runner.run_streamed()整体一个trace()每次 Agent 运行agent_span()LLM 生成generation_span()每次 Function 工具调用function_span()Guardrails 执行guardrail_span()Handoffs 切换handoff_span()语音输入语音转文本transcription_span()语音输出文本转语音speech_span()相关的语音 Span父级speech_group_span()默认情况下 Trace 名称为 Agent trace。若使用trace()手动创建可自定义该名称也可以通过在RunConfig见 run.py中配置名称与其他属性来定制。另外SDK 默认开启了追踪但有两种方式可以关闭全局禁用设置环境变量OPENAI_AGENTS_DISABLE_TRACING1。从 setup.py 的源码看TraceProvider在初始化时会读取该环境变量true/1均视为禁用。单次运行禁用将RunConfig.tracing_disabled设为True。源码中TraceProvider.create_trace()/create_span()会先检查全局_disabled与本次传入的disabled标志命中则直接返回NoOpTrace/NoOpSpan从而零开销跳过记录。需要特别说明的是对于在 OpenAI API 上采用零数据保留Zero Data Retention, ZDR策略的组织tracing 不可用因为追踪数据默认会导出到 OpenAI 后端。高层级 Trace把多次 run() 合并为一条完整工作流某些场景下你希望多次Runner.run()调用归属于同一条 Trace例如先生成笑话、再评价笑话的两段式流程。此时只需把整段代码包在with trace(...)中即可from cai.sdk.agents import Agent, Runner, trace async def main(): agent Agent(nameJoke generator, instructionsTell funny jokes.) with trace(Joke workflow): # (1)! first_result await Runner.run(agent, Tell me a joke) second_result await Runner.run(agent, fRate this joke: {first_result.final_output}) print(fJoke: {first_result.final_output}) print(fRating: {second_result.final_output})由于两次Runner.run调用被with trace()包裹两次独立运行将成为整体 Trace 的一部分而不是各自创建一条 Trace。这种模式在安全攻防演练中非常实用例如一次完整的渗透测试流程信息收集 → 漏洞分析 → 利用 → 报告可以被组织为一条端到端 Trace便于在可视化面板中整体回放。创建 Trace上下文管理器与手动生命周期使用trace()函数创建 Trace。Trace 需要显式开始和结束官方提供了两种方式推荐作为上下文管理器使用即with trace(...) as my_trace会在进入时自动 start、退出时自动 finish。从TraceImpl.__enter__/__exit__的实现看__enter__调用start(mark_as_currentTrue)__exit__调用finish(reset_currentTrue)并且对GeneratorExit做了特殊处理生成器场景下不重置当前上下文。手动调用trace.start()与trace.finish()。当前 Trace 通过 Python 的contextvars中Scope类用两个ContextVar分别保存当前 Trace与当前 Span并提供set/get/reset方法。如果你手动 start/end 一个 Trace需要通过start(mark_as_currentTrue)与finish(reset_currentTrue)来更新当前上下文中的 Trace 指针。trace()函数签名来自 create.py为def trace( workflow_name: str, trace_id: str | None None, group_id: str | None None, metadata: dict[str, Any] | None None, disabled: bool False, ) - Trace其中trace_id不传时自动生成官方建议使用util.gen_trace_id()生成以确保格式正确metadata可附加任意用户自定义信息disabledTrue时返回一个不会被记录的 Trace。另外需要注意的是如果当前已存在一个活跃 Trace 再创建新 TraceSDK 会打出一条 warning 日志Trace already exists...提示这可能是误操作。创建 Span各种*_span()与自定义埋点你可以使用各种*_span()方法创建 Span它们集中定义在 create.py 中。通常情况下你不需要手动创建 Span——SDK 会自动包裹上文表格中的各类操作。但当需要跟踪自定义逻辑例如一次内部子流程、一次数据库查询、一次渗透测试中的特定步骤时可以使用custom_span()from cai.sdk.agents.tracing import custom_span with custom_span(Recon scan, {target: 10.0.0.8, ports: 1-1024}) as span: # 你的自定义逻辑 passSpan 会自动归属于当前 Trace并嵌套在最近的当前 Span 之下同样通过contextvars的Scope跟踪。所有*_span()方法都遵循同一套签名约定共同参数包括span_idSpan ID不传时自动生成推荐用util.gen_span_id()parent父 Span 或 Trace不传时自动取当前 Trace/Span 作为父级disabled为True时返回不会被记录的 Span。各方法特有参数如下均可从 create.py 源码确认agent_span(name, handoffs, tools, output_type)记录 Agent 名称、可用 Handoff、工具列表与输出类型function_span(name, input, output)记录函数调用名称、输入与输出generation_span(input, output, model, model_config, usage)记录 LLM 生成的输入消息序列、输出、模型与用量若只需记录模型响应 ID可用response_span(response)handoff_span(from_agent, to_agent)记录交接来源与目标 Agentguardrail_span(name, triggered)记录守卫名称与触发状态transcription_span(model, input, input_formatpcm, output, model_config)语音转文本speech_span(model, input, output, output_formatpcm, model_config, first_content_at)文本转语音speech_group_span(input)语音分组容器mcp_tools_span(server, result)MCP 服务器 list tools 调用记录该能力还体现在 tests/mcp/test_mcp_tracing.py 等测试中。与 Trace 相同Span 也支持with span()上下文管理器用法或手动start()/finish()。敏感数据与隐私控制部分 Span 可能捕获潜在敏感数据需要格外注意generation_span()会存储 LLM 生成的输入/输出function_span()会存储函数调用的输入/输出——这些内容可能包含敏感信息。可以通过RunConfig.trace_include_sensitive_data关闭该类数据的捕获。从 run.py 源码可见其默认值为True默认捕获将其设为False后_run_impl.py见 src/cai/sdk/agents/_run_impl.py中对应的生成 Span 与函数 Span 将不再携带输入输出内容。语音类 Span 默认包含 base64 编码的 PCM 音频数据输入与输出。可通过配置VoicePipelineConfig.trace_include_sensitive_audio_data关闭对音频数据的捕获相关参数在语音管线的模型层src/cai/sdk/agents/voice/model.py与 OpenAI STT 实现src/cai/sdk/agents/voice/models/openai_stt.py中均有体现。此外前面提到的两种禁用方式环境变量OPENAI_AGENTS_DISABLE_TRACING1与RunConfig.tracing_disabledTrue也属于隐私控制手段你还可以在代码中调用set_tracing_disabled(True)动态地全局关闭追踪。自定义 Tracing Processors把追踪数据导出到任意后端默认架构追踪系统的高层架构如下对应 setup.py 与 processors.py 的实现初始化时创建全局唯一的TraceProviderGLOBAL_TRACE_PROVIDER负责创建 Trace 与 Span用BatchTraceProcessor配置TraceProvider它会将 Trace/Span分批发送给BackendSpanExporterBackendSpanExporter负责把 Trace 与 Span 批量导出到 OpenAI 后端默认端点https://api.openai.com/v1/traces/ingest。从 processors.py 源码可以看到几个关键实现细节BatchTraceProcessor使用线程安全的queue.Queue默认容量 8192由一个守护线程后台消费并分批导出以最大限度降低对主流程的性能影响。可通过max_batch_size默认 128、schedule_delay默认 5.0 秒、export_trigger_ratio默认 0.7即队列达到 70% 容量时立即触发导出等参数调优shutdown()时会先排空队列再做最终导出force_flush()可强制立即导出。BackendSpanExporter默认使用环境变量OPENAI_API_KEY、OPENAI_ORG_ID、OPENAI_PROJECT_ID作为鉴权信息通过httpx保持连接池超时 60s。上传失败时采用指数退避 10% 抖动的重试策略max_retries默认 3、base_delay默认 1.0s、max_delay默认 30.0s4xx 客户端错误不重试5xx 与网络错误视为瞬时错误重试。若未设置OPENAI_API_KEY会跳过导出并给出 warning。ConsoleSpanExporter一个开箱即用的调试型导出器直接把 Trace/Span 打印到控制台。两种定制方式模块根入口 tracing/init.py 提供了两个关键 APIadd_trace_processor()向全局TraceProvider追加一个额外的 Trace Processor它会在 Trace/Span 就绪时收到全部数据。适用于在发送到 OpenAI 后端之外再做一份自有处理的场景。set_trace_processors()用你自己的 Processors替换默认的处理器列表。替换后除非你包含一个会上报到 OpenAI 后端的TracingProcessor否则数据将不再发送到 OpenAI 后端。此外set_tracing_export_api_key(api_key)可动态设置后端导出器使用的 OpenAI API Key。模块加载时会自动注册默认处理器default_processor()BatchTraceProcessor并通过atexit注册GLOBAL_TRACE_PROVIDER.shutdown()保证进程退出时排空缓冲见 tracing/init.py。自定义 Processor 的实现接口要实现自己的 Processor 或 Exporter只需实现 processor_interface.py 中的抽象基类TracingProcessor需实现on_trace_start(trace)、on_trace_end(trace)、on_span_start(span)、on_span_end(span)、shutdown()、force_flush()六个方法其中on_span_end要求不应阻塞或抛异常。TracingExporter只需实现export(items)方法接收一批Trace | Span列表进行导出例如写入日志、上报自建平台等。一个基于控制台导出的最小示例如下from cai.sdk.agents.tracing import add_trace_processor from cai.sdk.agents.tracing.processors import BatchTraceProcessor, ConsoleSpanExporter # 追加一个额外的处理器除了默认上报 OpenAI 后端外 # 同时把 Trace/Span 打印到控制台便于本地调试。 add_trace_processor(BatchTraceProcessor(ConsoleSpanExporter()))多处理器的转发由SynchronousMultiTracingProcessor完成见 setup.py它按注册顺序把每个 Trace/Span 事件依次转发给所有已注册的 Processor并使用元组 锁保证遍历时的线程安全。外部追踪处理器生态官方文档中列出了可接入的第三方追踪平台均提供与 OpenAI Agents SDK 的集成方案可结合add_trace_processor()实现数据上报Weights BiasesWB Weave、Arize-Phoenix、MLflow自托管 OSS 与 Databricks 托管、Braintrust、Pydantic Logfire、AgentOps、Scorecard、Keywords AI、LangSmith、Maxim AI、Comet Opik、Langfuse、Langtrace。小结CAI Agents SDK 的追踪体系围绕Trace → Span → SpanData三级模型构建Runner自动埋点让零配置即可获得 Agent 运行全链路记录trace()/ 各*_span()函数支持手动组织高层级工作流与自定义埋点RunConfig.tracing_disabled、trace_include_sensitive_data与语音管线的trace_include_sensitive_audio_data提供了精细的隐私控制而add_trace_processor()/set_trace_processors()配合TracingProcessor、TracingExporter接口则让追踪数据可以灵活地导出到 OpenAI 后端或任意自建/第三方平台。对于安全攻防 Agent 而言这套追踪能力尤其有价值一次渗透测试中 Agent 的决策链、工具调用、模型推理与守卫触发情况都能被完整复现为结果审计、误报排查与攻击路径复盘提供了可回放的第一手证据。需要深入了解各模块细节时可继续阅读 docs/ref/tracing 下的参考文档create、spans、traces、span_data、processors、setup、scope、processor_interface、util或直接研读 src/cai/sdk/agents/tracing 的源码实现与 tests/tracing 中的测试用例。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考