OpenAI Agents SDK Python Tracing 模块完全指南:从 Trace/Span 原理到自定义处理器 OpenAI Agents SDK Python Tracing 模块完全指南从 Trace/Span 原理到自定义处理器【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本文是 OpenAI Agents SDKopenai-agents-python内置 Tracing追踪/可观测性模块的完整技术指南。该模块负责在 Agent 运行期间自动采集 LLM 生成、工具调用、Handoff、Guardrail 与自定义事件等全量记录支撑开发与生产环境下的调试、可视化与监控。读完本文你将掌握 Trace/Span 的核心数据模型、SDK 默认自动埋点的完整清单、三种禁用方式、敏感数据控制策略以及通过自定义 Tracing Processor 对接第三方观测平台的进阶方案。模块定位一次 Agent 运行的可观测性骨架在 src/agents/tracing/ 目录下SDK 以独立子包的形式实现了完整的追踪体系。agents.tracing对外暴露的 API 分为几类Trace 创建trace()表示一次端到端的工作流Span 创建agent_span()、task_span()、turn_span()、generation_span()、function_span()、guardrail_span()、handoff_span()、custom_span()、response_span()、transcription_span()、speech_span()、speech_group_span()、mcp_tools_span()上下文查询get_current_trace()、get_current_span()全局配置set_tracing_disabled()、add_trace_processor()、set_trace_processors()、set_trace_provider()、set_tracing_export_api_key()、flush_traces()数据模型Trace、Span、SpanError以及各类*SpanData。所有符号的完整导出清单见 src/agents/tracing/init.py对应的 API 参考页位于 docs/ref/tracing/其中index.md是模块总览入口。模块级详细文档可参考用户指南 docs/tracing.md。注意Tracing 默认是开启的。对于在 Zero Data RetentionZDR政策下使用 OpenAI API 的组织Tracing 不可用。Trace 与 Span两个核心数据模型整个追踪体系建立在两个抽象概念之上Trace追踪Trace 表示一次逻辑工作流的端到端完整操作由若干 Span 组合而成。其属性包括属性说明workflow_name逻辑工作流或应用的名称例如 Code generation、Customer servicetrace_idTrace 的唯一 ID不传则自动生成格式必须为trace_32位字母数字group_id可选分组 ID用于把同一会话的多个 Trace 关联起来例如聊天线程 IDdisabled若为True该 Trace 不会被记录metadata可选的 Trace 元数据字典从源码实现看src/agents/tracing/traces.py 中的TraceImpl是真实记录 Trace 的实现类其trace_id未提供时由util.gen_trace_id()生成格式为trace_前缀加 32 位十六进制见 src/agents/tracing/provider.py 中DefaultTraceProvider.gen_trace_id()即ftrace_{uuid.uuid4().hex}。此外还有两个特殊实现NoOpTrace当 Tracing 被禁用时返回的空实现保持上下文管理语义但不记录任何数据trace_id恒为no-opReattachedTrace从持久化的运行状态TraceState重建的 Trace 上下文用于恢复场景而不重复发送 trace start 事件相关序列化逻辑见 src/agents/tracing/traces.py 中的TraceState。Span跨度Span 表示一次有开始和结束时间的具体操作。其属性包括started_at/ended_at开始与结束的时间戳trace_id所属 Trace 的 IDparent_id父 Span 的 ID若有用于构建嵌套层级span_dataSpan 的类型化信息例如AgentSpanData保存 Agent 信息、GenerationSpanData保存 LLM 生成信息。在源码中src/agents/tracing/spans.py 定义了抽象基类Span与实现类SpanImpl。SpanImpl.start()会记录开始时间并调用处理器的on_span_start()finish()则记录结束时间并调用on_span_end()。Span 还支持通过set_error(SpanError)记录执行错误其中SpanError是一个 TypedDict包含message与可选的data字段。export()方法产出的载荷结构为{object: trace.span, id, trace_id, parent_id, started_at, ended_at, span_data, error}。Span 的类型化数据span_data的类型决定了 Span 的语义。定义在 src/agents/tracing/span_data.py 中的数据类型包括SpanData 类型语义AgentSpanDataAgent 运行信息name、handoffs、tools、output_typeTaskSpanData一次顶层Runner调用TurnSpanData一次 Agent 循环轮次turn、agent_nameGenerationSpanDataLLM 生成input、output、model、model_config、usageFunctionSpanData函数工具调用name、input、outputResponseSpanDataOpenAI Response 对象HandoffSpanDataHandoff 事件from_agent、to_agentGuardrailSpanDataGuardrail 执行name、triggeredCustomSpanData自定义数据name、data 字典TranscriptionSpanData语音转文本model、input、input_format、outputSpeechSpanData文本转语音model、input、output、output_format、first_content_atSpeechGroupSpanData音频 Span 分组MCPListToolsSpanDataMCP 服务器工具列表拉取server、result默认 TracingRunner 自动埋点清单SDK 在默认情况下会对一次 Agent 运行自动记录以下事件对应官方文档 docs/tracing.md 的 Default tracing 一节整个Runner.{run, run_sync, run_streamed}()调用被一个trace()包裹每次 Runner 调用被task_span()包裹每个模型轮次turn被turn_span()包裹每次 Agent 运行被agent_span()包裹LLM 生成被generation_span()包裹函数工具调用被function_span()包裹Guardrail 执行被guardrail_span()包裹Handoff 被handoff_span()包裹音频输入语音转文本被transcription_span()包裹音频输出文本转语音被speech_span()包裹SDK 可能将相关音频 Span 挂到同一个speech_group_span()之下。默认的 Trace 名称是字面量字符串Agent workflow。如果你使用trace()手动创建可以自行命名也可以通过RunConfig配置名称与其他属性。紧凑层级关闭 task/turn Span如果你希望得到更紧凑的层级结构可以禁用自动生成的 task 与 turn Spanagent、generation、function、guardrail、handoff 以及自定义 Span 仍会记录from agents import RunConfig, Runner result await Runner.run( agent, Hello, run_configRunConfig(tracing{include_task_and_turn_spans: False}), )这里的tracing参数对应 src/agents/tracing/config.py 中的TracingConfigTypedDict它支持两个可选键api_key导出 Trace 时使用的 API Keyinclude_task_and_turn_spans是否由 Runner 自动创建 task/turn Span省略时默认为True。控制 Tracing三种禁用方式Tracing 默认开启若需关闭有以下三种途径对应 docs/tracing.md全局环境变量设置OPENAI_AGENTS_DISABLE_TRACING1代码全局禁用调用set_tracing_disabled(True)单次运行禁用将RunConfig.tracing_disabled设为True。从源码看DefaultTraceProvider对禁用状态的处理很有意思src/agents/tracing/provider.py环境变量OPENAI_AGENTS_DISABLE_TRACING在首次使用时惰性读取取值为true或1时视为禁用从而允许在 import 之后、首次创建 Trace 之前再设置环境变量一旦通过set_disabled()设置了手动标志手动标志优先于环境变量禁用状态下create_trace()返回NoOpTracecreate_span()返回NoOpSpan它们不产生任何导出数据但with上下文管理仍正常工作因此业务代码无需分支判断。手动创建 Trace 与 Span创建 Tracetrace()函数定义于 src/agents/tracing/create.py用于创建 Trace。Trace 需要被显式启动与结束有两种方式推荐上下文管理器。with trace(...) as my_trace会在恰当的时机自动 start 与 end手动管理调用trace.start()与trace.finish()。当前 Trace 通过 Python 的contextvars机制跟踪因此天然支持并发。如果手动 start/finish需要通过start(mark_as_currentTrue)和finish(reset_currentTrue)来维护当前 Trace状态。trace()的完整签名与参数如下trace( workflow_name: str, # 工作流名称如 code_bot trace_id: str | None None, # 不传则自动生成 trace_32位 group_id: str | None None, # 会话分组 ID如聊天线程 ID metadata: dict[str, Any] | None None, # 附加元数据 tracing: TracingConfig | None None, # 导出配置api_key 等 disabled: bool False, # True 则创建但不记录 ) - Trace注意如果在已有当前 Trace 的情况下再次调用trace()SDK 会打印警告Trace already exists提示这多半是误用嵌套创建新 Trace 会割裂上下文。创建 Span各种*_span()工厂函数用于创建 Span。一般而言无需手动创建 Span——SDK 的 Runner 会自动完成。只有在你需要记录自定义业务事件时才使用custom_span()from agents import custom_span with custom_span(database_query, {operation: SELECT, table: users}) as span: results await db.query(SELECT * FROM users) span.span_data.data[output] {count: len(results)}Span 会自动成为当前 Trace 的一部分并嵌套在最近的当前 Span 之下同样通过contextvars跟踪。custom_span(name, dataNone)允许你附加任意结构化字典数据。若 Span 执行出错可用span.set_error({message: ..., data: ...})记录错误。高层级 Trace把多次 run 合并为一条 Trace有时你希望多次Runner.run()调用属于同一条 Trace例如先生成笑话、再给它打分的两步流程。做法是把整段代码包进一个trace()上下文from agents import Agent, Runner, trace async def main(): agent Agent(nameJoke generator, instructionsTell funny jokes.) with trace(Joke workflow): 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。长时运行 Worker 与立即导出默认的BatchTraceProcessor实现见 src/agents/tracing/processors.py在后台线程中每隔数秒导出一次 Trace或者在内存队列达到大小阈值时提前导出进程退出时还会执行最终 flush。因此对于 Celery、RQ、Dramatiq、FastAPI 后台任务这类长时运行 WorkerTrace 通常无需额外代码即可自动导出——只是每个任务结束后可能不会立即出现在 Traces 面板中。如果需要在某个工作单元结束时获得立即投递的保证请在 Trace 上下文退出后调用flush_traces()。flush_traces()会阻塞直到当前缓冲的 Trace/Span 全部导出因此应放在trace()关闭之后调用避免 flush 到半成品 Trace。官方文档给出了两个典型示例docs/tracing.mdCelery 任务from agents import Runner, flush_traces, trace celery_app.task def run_agent_task(prompt: str): try: with trace(celery_task): result Runner.run_sync(agent, prompt) return result.final_output finally: flush_traces()FastAPI 后台任务from fastapi import BackgroundTasks, FastAPI from agents import Runner, flush_traces, trace app FastAPI() def process_in_background(prompt: str) - None: try: with trace(background_job): Runner.run_sync(agent, prompt) finally: flush_traces() app.post(/run) async def run(prompt: str, background_tasks: BackgroundTasks): background_tasks.add_task(process_in_background, prompt) return {status: queued}当默认导出延迟可接受时可以省略flush_traces()调用。敏感数据处理部分 Span 会捕获潜在的敏感数据对应官方文档 docs/tracing.md 的 Sensitive data 一节generation_span()存储 LLM 生成的输入/输出function_span()存储函数调用的输入/输出。这些数据可能包含敏感内容可通过RunConfig.trace_include_sensitive_data关闭采集。默认值为True若不想改代码可以在启动应用前导出环境变量OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA值为true/1或false/0来设置默认值。同理音频类 Span 默认包含 base64 编码的 PCM 输入/输出音频数据可通过VoicePipelineConfig.trace_include_sensitive_audio_data配置关闭相关配置位于 src/agents/voice/pipeline_config.py。从源码看后端导出器还做了额外的数据卫生处理BackendSpanExporter在上传到 OpenAI traces ingest 接口前会按 100,000 字节的字段上限对大字段做截断追加... [truncated]后缀并对非 generation 类型 Span 丢弃usage之外的键、过滤不可 JSON 序列化的值避免 ingest 接口拒绝整批数据见 src/agents/tracing/processors.py 中的_sanitize_for_openai_tracing_api系列方法。自定义 Tracing Processors高层架构Tracing 的整体架构为对应 docs/tracing.md 的 Custom tracing processors 一节初始化时创建一个全局TraceProvider负责创建 Trace/Span用BatchTraceProcessor配置该TraceProvider后者将 Trace/Span 分批发送给BackendSpanExporter由它批量导出到 OpenAI 后端。关键实现细节全部可在 src/agents/tracing/ 中验证TraceProvider是抽象接口src/agents/tracing/provider.py默认实现为DefaultTraceProvider内部持有一个SynchronousMultiTracingProcessor按注册顺序把事件转发给所有处理器TracingProcessor接口src/agents/tracing/processor_interface.py要求实现五个方法on_trace_start、on_trace_end、on_span_start、on_span_end、shutdown、force_flush。所有方法应保证线程安全、快速返回、内部捕获异常避免干扰 Agent 执行TracingExporter接口只有一个方法export(items)负责把一批 Trace/Span 发送出去ConsoleSpanExporter打印到控制台与BackendSpanExporterHTTP 上传都是其实现BatchTraceProcessorsrc/agents/tracing/processors.py使用线程安全的queue.Queue、后台守护线程导出参数包括max_queue_size8192队列满后丢弃新 Span、max_batch_size128单批最多导出条数、schedule_delay5.0调度检查间隔秒数、export_trigger_ratio0.7队列容量达到 70% 时提前触发导出BackendSpanExporter默认端点为https://api.openai.com/v1/traces/ingest支持max_retries3与带 10% jitter 的指数退避基础延迟 1.0s最大 30s4xx 客户端错误不重试5xx 与网络错误重试API Key 缺省取os.environ[OPENAI_API_KEY]还支持OPENAI_ORG_ID、OPENAI_PROJECT_ID全局 Provider 与默认处理器均为惰性初始化src/agents/tracing/setup.py首次访问时才创建避免 import SDK 时就建立网络客户端与线程进程退出时通过atexit注册的钩子以 5 秒超时执行 shutdown flush。定制方式添加或替换处理器要自定义默认配置发送到其他/额外后端或修改导出行为有两种方式add_trace_processor(processor)追加一个额外的处理器它会在 Trace/Span 就绪时收到事件。这让你在继续发送到 OpenAI 后端之外还能做自己的处理例如写入本地日志或第三方观测平台set_trace_processors(processors)替换默认的处理器列表。这意味除非你提供的TracingProcessor显式上传否则 Trace 不会再发往 OpenAI 后端。一个自定义处理器的骨架接口签名来自 src/agents/tracing/processor_interface.pyfrom agents.tracing import TracingProcessor class CustomProcessor(TracingProcessor): def on_trace_start(self, trace): pass def on_trace_end(self, trace): pass def on_span_start(self, span): pass def on_span_end(self, span): pass def shutdown(self): pass def force_flush(self): pass注册方式from agents import add_trace_processor add_trace_processor(CustomProcessor())项目中SynchronousMultiTracingProcessor会对每个处理器包裹异常捕获与日志诊断含处理器身份标识保证单个处理器故障不会拖垮整个运行。非 OpenAI 模型下的 Tracing使用非 OpenAI 模型时可以给 Tracing 导出器单独提供一个 OpenAI API Key从而在不关闭 Tracing 的前提下免费在 OpenAI Traces 面板查看数据。模型适配器的选择与注意事项参见 docs/models/index.md 的 Third-party adapters 一节。全局设置导出 Keyimport os from agents import set_tracing_export_api_key, Agent from agents.extensions.models.any_llm_model import AnyLLMModel tracing_api_key os.environ[OPENAI_API_KEY] set_tracing_export_api_key(tracing_api_key) model AnyLLMModel( modelyour-provider/your-model-name, api_keyyour-api-key, ) agent Agent( nameAssistant, modelmodel, )仅对单次运行使用不同 Key如果只需为某一次运行指定不同的 Tracing Key通过RunConfig传入即可不必改动全局导出器from agents import Runner, RunConfig await Runner.run( agent, inputHello, run_configRunConfig(tracing{api_key: sk-tracing-123}), )注意TracingConfig.api_key的传递链路RunConfig.tracing中的api_key会经DefaultTraceProvider.create_trace()写入TraceImpl再透传到其下所有 SpanSpanImpl通过tracing_api_key属性继承导出器在 src/agents/tracing/processors.py 中按tracing_api_key对条目分组后分别上传。此外Trace.to_json(include_tracing_api_keyTrue)与TraceState支持将 Trace 上下文含 Key 的 SHA-256 指纹持久化供恢复运行场景复用相关实现见 src/agents/tracing/traces.py。模块 API 一览agents.tracing子包的完整公开 API来自 src/agents/tracing/init.py 的__all__Trace/Span 创建trace、agent_span、task_span、turn_span、generation_span、function_span、guardrail_span、handoff_span、custom_span、response_span、transcription_span、speech_span、speech_group_span、mcp_tools_span上下文与配置get_current_trace、get_current_span、add_trace_processor、set_trace_processors、set_trace_provider、set_tracing_disabled、set_tracing_export_api_key、flush_traces、TracingConfig、TraceCtxManager数据模型Trace、Span、SpanError、SpanData、AgentSpanData、CustomSpanData、FunctionSpanData、GenerationSpanData、GuardrailSpanData、HandoffSpanData、MCPListToolsSpanData、ResponseSpanData、SpeechGroupSpanData、SpeechSpanData、TaskSpanData、TranscriptionSpanData、TurnSpanData、TracingProcessor、TraceProvider工具函数gen_trace_id、gen_span_id对应的逐项 API 参考文档位于 docs/ref/tracing/含config.md、create.md、spans.md、traces.md、provider.md、processors.md、processor_interface.md、setup.md、span_data.md、context.md、scope.md、util.md、model_tracing.md、logger.md等。环境变量速查环境变量作用取值OPENAI_AGENTS_DISABLE_TRACING全局禁用 Tracing1/trueOPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA控制 LLM 生成/函数调用的输入输出是否采集true/1或false/0OPENAI_API_KEY后端导出器的默认 API Key也可用set_tracing_export_api_key()设置API Key 字符串OPENAI_ORG_ID导出请求的组织 ID组织 ID 字符串OPENAI_PROJECT_ID导出请求的项目 ID项目 ID 字符串小结openai-agents-python 的 Tracing 模块以TraceSpan两级模型覆盖了一次多 Agent 工作流的完整生命周期Runner 自动埋点让你零成本获得 LLM 生成、工具调用、Handoff、Guardrail 乃至语音链路的全量记录trace()上下文与custom_span()让你在框架之上叠加自定义业务事件BatchTraceProcessorBackendSpanExporter的后台批量导出兼顾了性能与可靠性而add_trace_processor()/set_trace_processors()则为你对接任意第三方观测后端提供了标准扩展点。对于长时运行 Workerflush_traces()是保证可观测性即时性的关键一招。模块相关测试如 tests/tracing/ 下的环境变量禁用、Span 排序、Trace 上下文等用例可进一步佐证上述行为。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考