CAI 追踪处理器接口TracingProcessor / TracingExporter深度解析与自定义实现指南【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai导读本指南围绕 Cybersecurity AICAI开源仓库中 docs/ref/tracing/processor_interface.md 所引用的核心模块cai.sdk.agents.tracing.processor_interface展开该模块定义了 CAI 可观测性Tracing管线的两个抽象契约处理器接口TracingProcessor与导出器接口TracingExporter。读完本文你将掌握 CAI 追踪系统的事件流trace 开始/结束、span 开始/结束如何流转到处理器、内置BatchTraceProcessor与ConsoleSpanExporter/BackendSpanExporter的底层实现原理并能基于接口编写、注册自己的自定义追踪处理器把 Agent 运行轨迹接入日志、监控或自建后端。定位processor_interface 是 CAI 追踪体系的「插槽层」原文档是使用 mkdocstrings 指令::: cai.sdk.agents.tracing.processor_interface生成的 API 引用页其内容直接来自源码模块 src/cai/sdk/agents/tracing/processor_interface.py。在整个追踪模块入口见 docs/ref/tracing/index.md中该文件扮演「插槽层」角色事件产生方Trace一次逻辑工作流与Span工作流内的一段操作在生命周期切换时发出事件事件消费方TracingProcessor接收这些事件TracingExporter负责把 trace/span 真正写出去日志、后端 HTTP 等中间调度方BatchTraceProcessor等处理器内部再调用 exporter实现缓冲、批量与异步导出。因此理解这两个接口就等于理解了 CAI 追踪从「产生事件」到「落盘/上报」的整条链路也才能安全地扩展自己的追踪后端。一、TracingProcessor追踪生命周期钩子的完整契约TracingProcessor是一个抽象基类abc.ABC共定义6 个抽象方法对应追踪对象生命周期中的关键时点定义于 processor_interface.py方法触发时点关键约定on_trace_start(trace)一个 Trace 被创建并 start 时参数为Trace实例on_trace_end(trace)一个 Trace 结束时参数为Trace实例on_span_start(span)一个 Span 被 start 时参数为Span[Any]携带任意SpanDataon_span_end(span)一个 Span 结束时不应阻塞也不应抛出异常源码 docstring 明确约定shutdown()应用停止时用于清理后台线程、关闭连接force_flush()任意时刻手动调用强制立即冲刷所有排队中的 spans/traces接口签名要点见 processor_interface.pyabc.abstractmethod def on_span_start(self, span: Span[Any]) - None: ... abc.abstractmethod def on_span_end(self, span: Span[Any]) - None: Called when a span is finished. Should not block or raise exceptions. ...on_span_end的「不阻塞、不抛异常」约定至关重要它是追踪热路径Agent 每次函数调用、每次模型生成都会结束 span任何在此处的阻塞或异常都会反向拖慢甚至中断业务执行。仓库内置实现均遵循这一约定入队失败时仅logger.warning后丢弃见下文 BatchTraceProcessor。二、TracingExportertrace/span 的最终目的地抽象与处理器配套的是TracingExporterprocessor_interface.pyclass TracingExporter(abc.ABC): abc.abstractmethod def export(self, items: list[Trace | Span[Any]]) - None: Exports a list of traces and spans.它只定义一个export(items)方法接收一批trace/span 混合列表。把「接收事件」与「写出数据」拆成两个接口是 CAI 分层设计的核心处理器负责缓冲/调度导出器负责具体落地两者可自由组合。三、钩子如何被触发Span 与 Trace 内部的调用链处理器的方法不会凭空被调用它们由Span/Trace的实现类在生命周期切换时主动驱动。以标准实现SpanImpl为例spans.pydef start(self, mark_as_current: bool False): if self.started_at is not None: logger.warning(Span already started) return self._started_at util.time_iso() self._processor.on_span_start(self) # 触发钩子 if mark_as_current: self._prev_span_token Scope.set_current_span(self) def finish(self, reset_current: bool False) - None: if self.ended_at is not None: logger.warning(Span already finished) return self._ended_at util.time_iso() self._processor.on_span_end(self) # 触发钩子 ...TraceImpl同理traces.pystart()里调用self._processor.on_trace_start(self)finish()里调用on_trace_end(self)。两点实现细节值得注意幂等保护start/finish通过started_at/ended_at/_started标志做重复调用拦截重复触发仅打印 warning不会重复上报时间戳前置钩子在记录完started_at/ended_atISO 8601 UTC见 util.py之后才调用保证处理器拿到的 span 时间信息完整。此外Span的export()方法spans.py产出的序列化结构是 exporter 的标准数据形态{ object: trace.span, id: self.span_id, trace_id: self.trace_id, parent_id: self._parent_id, started_at: self._started_at, ended_at: self._ended_at, span_data: self.span_data.export(), error: self._error, }Trace.export()则输出object: trace、workflow_name、group_id、metadatatraces.py。四、处理器如何被组装多路转发与全局 Provider接口之上CAI 提供了把多个处理器「叠起来」的装配层SynchronousMultiTracingProcessorsetup.py。它本身也实现TracingProcessor内部以元组保存处理器列表并按注册顺序同步转发所有事件class SynchronousMultiTracingProcessor(TracingProcessor): def __init__(self): self._processors: tuple[TracingProcessor, ...] () self._lock threading.Lock() def add_tracing_processor(self, tracing_processor: TracingProcessor): with self._lock: self._processors (tracing_processor,) def on_trace_start(self, trace: Trace) - None: for processor in self._processors: processor.on_trace_start(trace) ...这里用「锁 元组」而非列表是为了避免遍历处理器时被并发修改导致迭代异常源码注释明确说明。全局单例GLOBAL_TRACE_PROVIDER TraceProvider()setup.py在此基础上提供对外的公共操作模块级便捷函数定义在 tracing/init.pyadd_trace_processor(processor)追加一个处理器所有新处理器都会收到全部事件set_trace_processors(processors)替换整个处理器列表可用来移除默认后端set_tracing_disabled(disabled)全局开关也可通过环境变量OPENAI_AGENTS_DISABLE_TRACINGtrue生效见 setup.pyset_tracing_export_api_key(api_key)为默认后端导出器设置 API Key。值得注意的是__init__.py在导入时即调用add_trace_processor(default_processor())注册默认处理器并atexit.register(GLOBAL_TRACE_PROVIDER.shutdown)保证进程退出时排空缓冲init.py。禁用追踪或disabledTrue时TraceProvider会返回NoOpTrace/NoOpSpan空实现避免无谓开销。五、开箱即用实现BatchTraceProcessor 与两个内置 Exporter处理器接口的「官方参考实现」是BatchTraceProcessorprocessors.py其设计目标是「以最小性能代价完成导出」源码注释总结了三点线程安全的queue.Queue、后台线程导出、spans 在内存中暂存直至导出。5.1 构造参数均有默认值可直接覆盖参数默认值含义exporter必填实际执行导出的TracingExportermax_queue_size8192队列最大容量写满后新事件直接丢弃记 warningmax_batch_size128单次导出批次最大条数schedule_delay5.0定时导出周期秒后台线程按此间隔检查export_trigger_ratio0.7队列水位触发比队列长度达到max_queue_size * 0.7时立即导出5.2 事件流转与调度逻辑on_trace_startTrace 立即put_nowait入队on_trace_end不做事trace 已在开始时上报on_span_start不做事on_span_endSpanput_nowait入队后台守护线程_run每 0.2s 检查一次到达schedule_delay定时点或队列超过水位阈值时调用_export_batches()批量取数并交给 exporterforce_flush()忽略批次大小限制一次排空全部队列shutdown(timeout)置位停止事件并join后台线程线程退出前会做一次强制排空避免进程退出丢数据。5.3 两个内置 ExporterConsoleSpanExporterprocessors.py把 trace/span 打印到控制台适合本地调试与快速验证BackendSpanExporterprocessors.py通过httpx.Client把{data: [...]}载荷 POST 到追踪后端。构造参数包括endpoint默认https://api.openai.com/v1/traces/ingest、max_retries默认 3、base_delay/max_delay退避上下限默认 1.0s/30.0s。请求头携带Authorization: Bearer key与OpenAI-Beta: tracesv1。其重试策略是有选择的4xx 客户端错误直接放弃重试无意义5xx 与httpx.RequestError才按「指数退避 10% 随机抖动」重试API Key 缺失时静默跳过并记 warning。全局默认实例通过default_processor()/default_exporter()暴露processors.py。六、实战基于接口编写并注册自定义处理器下面基于接口实现一个「内存版」追踪处理器完整参考实现即仓库测试组件 tests/testing_processor.py 中的SpanProcessorForTests线程安全、适合测试或轻量使用import threading from typing import Any, Literal from cai.sdk.agents.tracing import Span, Trace, TracingProcessor TestSpanProcessorEvent Literal[trace_start, trace_end, span_start, span_end] class SpanProcessorForTests(TracingProcessor): def __init__(self) - None: self._lock threading.Lock() self._spans: list[Span[Any]] [] self._traces: list[Trace] [] self._events: list[TestSpanProcessorEvent] [] def on_trace_start(self, trace: Trace) - None: with self._lock: self._traces.append(trace) self._events.append(trace_start) def on_trace_end(self, trace: Trace) - None: with self._lock: self._events.append(trace_end) def on_span_start(self, span: Span[Any]) - None: with self._lock: self._events.append(span_start) def on_span_end(self, span: Span[Any]) - None: with self._lock: self._events.append(span_end) self._spans.append(span) # 只保留已结束的 span def shutdown(self) - None: pass def force_flush(self) - None: pass注册到全局管线from cai.sdk.agents.tracing import add_trace_processor add_trace_processor(SpanProcessorForTests()) # 追加默认后端仍会同时工作 # 若想完全替换默认后端 # from cai.sdk.agents.tracing import set_trace_processors # set_trace_processors([SpanProcessorForTests()])此后无论通过trace()、agent_span()、custom_span()还是function_span()创建的任何追踪对象相关辅助函数见 tracing/init.pyspan 数据类型见 span_data.py其生命周期事件都会同步转发到你的处理器。注意add_trace_processor是追加语义默认的BatchTraceProcessor依旧在运行如需彻底接管例如禁用默认后端上报应使用set_trace_processors。七、接口契约的行为保证来自测试用例的验证仓库测试 tests/others/test_trace_processor.py 系统性地验证了接口与实现的各项契约可作为自定义实现时的行为基准入队规则on_trace_start入队 trace、on_span_end入队 span而on_trace_end/on_span_start不应入队test_batch_processor_doesnt_enqueue_on_trace_end_or_span_start队列上限max_queue_size耗尽后继续入队不会撑破队列多余事件被丢弃test_batch_trace_processor_queue_full定时导出超过schedule_delay后事件会被自动导出test_batch_trace_processor_scheduled_export强制冲刷force_flush会无视批次上限导出全部缓存test_batch_trace_processor_force_flush优雅关闭shutdown会排空队列中所有剩余事件test_batch_trace_processor_shutdown_flushes导出器语义空列表不发请求、无 API Key 静默跳过、2xx 只发一次、4xx 不重试、5xx 与网络错误按max_retries重试、close()关闭底层连接test_backend_span_exporter_*系列。此外 tests/testing_processor.py 的fetch_normalized_spans还展示了消费端如何利用export()结果做断言trace/span 的id分别以trace_/span_前缀生成见 util.pystarted_at/ended_at必须是合法的 ISO 8601 时间span_data以type字段区分如agent、function、generation、custom、guardrail等。八、实现自定义处理器时的设计建议on_span_end保持轻量接口注释明确「不应阻塞或抛异常」可参考BatchTraceProcessor的做法——把工作放入线程安全队列或派发到后台线程热点路径只做入队考虑并发安全Agent 可能并行执行多个分支见仓库并行相关模块处理器回调可能来自不同线程务必用锁或原子结构保护共享状态shutdown要可重入且能排空参考BatchTraceProcessor先置停止标志、再join工作线程、最后强制排空避免进程退出时丢事件善用SpanData的类型字段不同业务语义的 span 用type区分span_data.py导出或过滤时可按类型分流无痛降级拿不到鉴权信息或后端不可用时应像BackendSpanExporter一样「记日志 跳过」绝不能让追踪故障反噬业务。相关文档导航接口原始 API 引用processor_interface.md追踪模块总览docs/ref/tracing/index.md装配与 Providerdocs/ref/tracing/setup.mdSpan/Trace 数据模型docs/ref/tracing/spans.md、docs/ref/tracing/traces.md内置实现与默认实例processors.py测试基线tests/others/test_trace_processor.py、tests/testing_processor.py【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考