OpenAI Agents SDK(Python)追踪(Tracing)完全指南:从默认埋点到自定义导出
2026/9/12 21:07:20 网站建设 项目流程

OpenAI Agents SDK(Python)追踪(Tracing)完全指南:从默认埋点到自定义导出

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

本篇指南围绕 OpenAI Agents SDK 内置的追踪(Tracing)能力展开,系统讲解追踪记录(Trace)与跨度(Span)的模型、默认埋点范围、长期运行工作进程的即时导出、敏感数据保护、自定义追踪处理器,以及非 OpenAI 模型的追踪配置。读完本文,你将掌握如何在开发与生产环境中利用追踪仪表板调试、可视化和监控多智能体工作流,并能根据实际场景定制追踪数据的采集与导出行为。

追踪是什么:开箱即用的可观测性

Agents SDK 内置追踪功能,会收集智能体运行期间各类事件的完整记录:LLM 生成、工具调用、任务转移(Handoff)、安全防护措施(Guardrail),甚至包括你主动上报的自定义事件。借助追踪仪表板(OpenAI Traces dashboard),你可以在开发和生产环境中调试、可视化并监控工作流。

追踪功能默认启用,无需任何额外配置即可开始记录。整个采集链路由 src/agents/tracing/ 目录下的若干模块支撑,其中包括:

  • create.py:trace()与各类*_span()工厂函数;
  • traces.py 与 spans.py:TraceSpan的抽象基类及默认实现;
  • processors.py:BatchTraceProcessorBackendSpanExporter等导出链路;
  • setup.py:全局TraceProvider的懒初始化与进程退出时的自动关闭。

三种常见的禁用方式

如果出于隐私、成本或其他原因需要关闭追踪,可以按需选择以下三种方式之一:

  1. 全局禁用(环境变量):设置环境变量OPENAI_AGENTS_DISABLE_TRACING=1
  2. 全局禁用(代码):调用set_tracing_disabled(True),对应实现见init.py 中的set_tracing_disabled(),它会将该开关写入全局TraceProvider
  3. 单次运行禁用:将agents.run.RunConfig.tracing_disabled设置为True,见 run_config.py。

从源码结构看,当追踪被禁用时,trace()*_span()返回的是 traces.py 中的NoOpTrace与 spans.py 中的NoOpSpan——它们仍会维护正确的上下文管理(with块、contextvar的进出栈),但不会记录、存储或导出任何数据,因此业务代码无需为“是否开启追踪”编写分支。

注意:依据零数据保留(Zero Data Retention,ZDR)政策使用 OpenAI API 的组织,追踪功能不可用。

核心概念:追踪记录(Trace)与跨度(Span)

追踪体系建立在两个抽象之上,二者均在源码中有清晰的抽象基类定义(traces.py、spans.py)。

追踪记录(Trace)

一条追踪记录表示一次"工作流"的端到端操作,由多个跨度组成。它拥有以下属性:

属性说明
workflow_name逻辑工作流或应用的名称,例如"代码生成"或"客户服务"
trace_id追踪记录的唯一 ID,未传入时自动生成;格式必须为trace_<32位字母数字>
group_id可选的组 ID,用于关联来自同一会话的多条追踪记录(例如聊天线程 ID)
disabled若为True,该条追踪不会被记录
metadata可选的追踪元数据字典

从 traces.py 的实现看,TraceImpl.start()会调用处理器的on_trace_start()并(在mark_as_current=True时)把自身设为当前追踪;finish()则调用on_trace_end()并恢复之前的上下文。此外,TraceImpl.export()输出的 JSON 结构为{"object": "trace", "id": ..., "workflow_name": ..., "group_id": ..., "metadata": ...},这正是推送到后端的数据形态。

跨度(Span)

跨度表示具有明确开始和结束时间的单个操作。跨度包含:

  • started_atended_at时间戳(ISO 8601 格式,由 util.py 中的time_iso()生成);
  • trace_id:所属追踪记录的 ID;
  • parent_id:父跨度的 ID(若存在);
  • span_data:跨度承载的业务信息。例如AgentSpanData包含智能体信息,GenerationSpanData包含 LLM 生成信息,依此类推。

SpanImplstart()时记录开始时间并通知处理器,在finish()时记录结束时间;它还提供了set_error(SpanError)方法用于标记跨度的错误信息(message+ 可选data),错误会随export()输出。

上下文追踪机制

当前追踪记录和当前跨度都是通过 Python 的contextvar进行跟踪的(相关逻辑集中在 scope.py),这意味着并发场景下各任务自动拥有正确的追踪上下文,无需手动传递。这也为后面介绍的"更高层级追踪记录"和"异步并发运行"提供了基础保障。

默认追踪:一次运行被自动埋点的内容

默认情况下,SDK 会对一次智能体运行自动埋点以下内容:

  • 整个Runner.{run, run_sync, run_streamed}()调用被封装在一个trace()中;
  • 每次运行器(runner)调用被封装在task_span()中;
  • 每轮模型交互被封装在turn_span()中;
  • 每次智能体运行被封装在agent_span()中;
  • LLM 生成被封装在generation_span()中;
  • 每次函数工具调用被封装在function_span()中;
  • 安全防护措施被封装在guardrail_span()中;
  • 任务转移被封装在handoff_span()中;
  • 音频输入(语音转文本)被封装在transcription_span()中;
  • 音频输出(文本转语音)被封装在speech_span()中;
  • SDK 可能将相关的音频跨度置于一个speech_group_span()之下。

除上述之外,create.py 还提供了response_span()(记录 OpenAI Response 对象)与mcp_tools_span()(记录 MCP 服务器列工具调用)等工厂函数,可满足更细粒度的埋点需求。

默认情况下,追踪名称为字面字符串Agent workflow(这一默认值同样体现在 traces.py 的ReattachedTrace中)。你可以通过trace()显式设置名称,也可以通过RunConfig配置名称及其他属性。

精简层级:关闭任务与交互轮次跨度

如果希望追踪记录的层级结构更紧凑,可以为单次运行禁用自动的 task span 与 turn span。此时智能体、生成、函数、安全防护、任务转移以及自定义跨度仍然会被记录

from agents import RunConfig, Runner result = await Runner.run( agent, "Hello", run_config=RunConfig(tracing={"include_task_and_turn_spans": False}), )

该选项的类型定义在 config.py 的TracingConfig(一个TypedDict)中:include_task_and_turn_spans缺省时为True,运行器据此决定是否创建 task/turn 跨度。

此外,你还可以通过自定义追踪处理器(见下文)将追踪记录推送到其他目标位置,作为默认 OpenAI 后端的替代目标或辅助目标。

长期运行的工作进程与即时导出

默认的BatchTraceProcessor会在后台每几秒批量导出一次追踪记录;当内存队列达到设定的阈值时也会提前导出;进程退出时还会执行最终刷新(shutdown(),且atexit已注册钩子,见 setup.py)。

因此,对于 Celery、RQ、Dramatiq 或 FastAPI 后台任务等长期运行的工作进程,追踪记录通常无需额外代码即可自动导出,但每项作业结束后,它们可能不会立即出现在追踪仪表板中——因为导出是批量、异步的。

如果需要保证在一个工作单元结束时立即交付,请在退出追踪上下文后调用flush_traces()。该函数会强制清空当前缓冲的追踪记录与跨度(见init.py 的实现,最终调用TraceProvider.force_flush())。

以 Celery 任务为例:

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()阻塞,直到当前已缓冲的追踪记录和跨度均导出完成。因此请务必在trace()上下文关闭之后再调用它,避免刷新尚未完全构建的追踪记录;如果默认导出延迟可以接受,则可以跳过此调用。

源码视角:BatchTraceProcessor 的工作机制

从 processors.py 可以看到BatchTraceProcessor的实现细节:

  • 使用线程安全的queue.Queue作为缓冲,默认max_queue_size=8192,队列满时丢弃新项并输出告警日志;
  • 由后台守护线程(首次入队时懒启动)按schedule_delay(默认 5.0 秒)周期检查,当队列大小达到max_queue_size * export_trigger_ratio(默认 0.7,即约 5734 项)时立即触发导出;
  • 每次导出按max_batch_size(默认 128)分批进行,force_flush()会同步排空整个队列;
  • BackendSpanExporter将跨度/追踪批量 POST 到 OpenAI 后端,默认端点、重试策略(max_retries=3、指数退避base_delay=1.0max_delay=30.0,并带 10% 抖动)均可在源码中确认。

这些默认值解释了"长期运行进程无需额外代码即可自动导出"的原因,也解释了为什么刚结束的作业可能不会立刻出现在仪表板中——存在最多约数秒的调度延迟。

更高层级的追踪记录:多次运行合并为一条 Trace

有时你希望多次调用Runner.run()成为同一条追踪记录的一部分(例如"先生成笑话,再评价笑话"这类多步骤工作流)。方法很简单:把整段代码封装在一个trace()上下文管理器中。

from agents import Agent, Runner, trace async def main(): agent = Agent(name="Joke generator", instructions="Tell funny jokes.") with trace("Joke workflow"): # (1)! first_result = await Runner.run(agent, "Tell me a joke") second_result = await Runner.run(agent, f"Rate this joke: {first_result.final_output}") print(f"Joke: {first_result.final_output}") print(f"Rating: {second_result.final_output}")
  1. 由于对Runner.run的两次调用都被封装在with trace()中,两次运行会成为同一条整体追踪记录的一部分,而不是各自创建一条单独的追踪记录。

之所以可行,正是因为"当前追踪"通过contextvar跟踪:trace()进入上下文时成为当前追踪,内部的多次运行器调用及其所有跨度都自动挂载到它之下,形成一棵完整的端到端调用树。

创建追踪记录(Trace)

你可以使用trace()函数(create.py)创建追踪记录。追踪记录需要启动和结束,有两种方式:

  1. 推荐:将追踪记录用作上下文管理器,即with trace(...) as my_trace:。这会自动在正确的时机启动和结束追踪(__enter__调用start(mark_as_current=True)__exit__调用finish(reset_current=True))。
  2. 手动:调用trace.start()trace.finish()

trace()的完整签名(含全部可选参数)为:

trace( workflow_name: str, trace_id: str | None = None, group_id: str | None = None, metadata: dict[str, Any] | None = None, tracing: TracingConfig | None = None, disabled: bool = False, ) -> Trace
  • workflow_name:逻辑应用或工作流的名称,例如"code_bot"(编码智能体)或"customer_support_agent"(客服智能体);
  • trace_id:可选,未提供时自动生成,推荐使用util.gen_trace_id()确保格式正确(trace_<32位字母数字>);
  • group_id:可选的会话分组标识,例如聊天线程 ID;
  • metadata:附加的用户自定义信息字典;
  • tracing:本次追踪的导出配置(例如单次运行的api_key);
  • disabled:为True时返回一个不会被记录的 Trace。

如果手动启动和结束追踪记录,请向start()传入mark_as_current=True,并向finish()传入reset_current=True,以正确地更新"当前追踪"上下文。从 traces.py 的TraceImpl实现可以看到,finish()还会通过on_trace_end()通知处理器完成该追踪的收尾。

创建跨度(Span)

你可以使用各种*_span()方法(create.py)创建跨度。通常无需手动创建跨度——运行器会自动为 LLM 生成、工具调用、任务转移、防护措施等创建对应跨度。当你需要跟踪自定义业务信息时,可以使用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)}

跨度会自动成为当前追踪记录的一部分,并嵌套在最近的当前跨度之下——"当前跨度"同样通过contextvar跟踪(scope.py)。因此,在任意嵌套的with块中创建的跨度都会自动挂载到正确的父节点,形成层级结构。

常用跨度工厂一览(均可作为上下文管理器使用,也可手动start()/finish()):

工厂函数用途主要参数
agent_span()记录一次智能体运行namehandoffstoolsoutput_type
task_span()记录一次顶层 Runner 调用name
turn_span()记录一轮智能体循环turnagent_name
generation_span()记录一次 LLM 生成inputoutputmodelmodel_configusage
function_span()记录一次函数工具调用nameinputoutput
guardrail_span()记录一次防护措施nametriggered
handoff_span()记录一次任务转移from_agentto_agent
custom_span()记录自定义操作namedata
transcription_span()记录语音转文本modelinputinput_format(默认pcm)、output
speech_span()记录文本转语音modelinputoutputoutput_format(默认pcm)、first_content_at

所有跨度工厂都支持span_id(缺省自动生成)、parent(缺省自动取当前跨度/追踪)与disabled参数。

敏感数据:按需关闭输入/输出采集

某些跨度可能捕获潜在的敏感数据,需要根据业务合规要求决定是否采集:

  • generation_span()会存储 LLM 生成的输入/输出function_span()会存储函数调用的输入/输出。这些内容可能包含敏感数据,可以通过RunConfig.trace_include_sensitive_data禁止捕获。
  • 默认情况下,音频跨度会包含输入和输出音频的Base64 编码 PCM 数据。可以通过配置VoicePipelineConfig.trace_include_sensitive_audio_data禁止捕获这些音频数据(该配置用于语音管线场景,参见仓库中的 voice 相关文档与示例)。

默认值方面,trace_include_sensitive_dataTrue。若不想编写代码,可以在运行应用前导出环境变量来设置默认值:

export OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA=false # 或 true/1/0

该环境变量的解析逻辑在 run_config.py 的_default_trace_include_sensitive_data()中:取OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA(缺省"true"),值的小写形式命中1/true/yes/on之一即视为开启。

自定义追踪处理器:替换或扩展导出目标

默认架构

追踪功能的高层架构如下(均有对应源码实现):

  • 初始化时,创建一个全局TraceProvider(provider.py),负责创建追踪记录与跨度;
  • TraceProvider配置一个BatchTraceProcessor(processors.py),将追踪记录和跨度分批发送给BackendSpanExporter
  • BackendSpanExporter将跨度和追踪记录批量导出到 OpenAI 后端(默认端点为https://api.openai.com/v1/traces/ingest,鉴权头为Authorization: Bearer <api_key>,并携带OpenAI-Beta: traces=v1头;API Key 缺省从OPENAI_API_KEY环境变量读取,也可通过OPENAI_ORG_IDOPENAI_PROJECT_ID指定组织与项目)。

从 setup.py 可以看到,全局 provider 与默认处理器采用懒初始化:首次访问追踪功能时才创建(避免模块导入阶段就建立网络连接或线程),并通过atexit注册进程退出时的关闭刷新。

两种自定义方式

若要改变默认行为——把追踪发送到其他/额外的后端,或修改导出器的行为——有两种方式:

  1. add_trace_processor():添加一个额外的追踪处理器,它会在追踪记录和跨度准备就绪时接收它们。这样,除了发送到 OpenAI 后端之外,你还可以自行处理(例如写入本地日志、推送到自建可观测平台)。实现上,它会调用get_trace_provider().register_processor(processor)(见init.py)。
  2. set_trace_processors():用你自己的处理器列表替换默认处理器。这意味着,除非你的处理器列表中包含一个执行发送的TracingProcessor,否则追踪记录不会发送到 OpenAI 后端。实现上对应get_trace_provider().set_processors(processors)

TracingProcessor是处理器需要实现的接口(processor_interface.py),核心回调包括on_trace_starton_trace_endon_span_starton_span_endforce_flushshutdown等。仓库还提供了一个将追踪打印到控制台的ConsoleSpanExporter(processors.py),适合本地调试——它可以作为自定义导出器的最简参考实现。

非 OpenAI 模型的追踪

使用非 OpenAI 模型时,你可以向追踪导出器提供一个 OpenAI API 密钥,从而在不禁用追踪的情况下,在 OpenAI 追踪仪表板中启用免费追踪。适配器选择与设置注意事项请参阅模型指南中的第三方适配器部分。

全局设置示例(以AnyLLMModel为例):

import 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( model="your-provider/your-model-name", api_key="your-api-key", ) agent = Agent( name="Assistant", model=model, )

set_tracing_export_api_key()的实现见init.py,它直接作用于全局默认导出器(default_exporter().set_api_key(api_key))。

如果仅需为单次运行使用不同的追踪密钥,请通过RunConfig传入,而不要更改全局导出器:

from agents import Runner, RunConfig await Runner.run( agent, input="Hello", run_config=RunConfig(tracing={"api_key": "sk-tracing-123"}), )

RunConfig.tracing即前文提到的TracingConfig(config.py),目前支持api_keyinclude_task_and_turn_spans两个字段。从 processors.py 的导出实现可以看到,导出时会按tracing_api_key对条目分组,每条追踪/跨度携带的专属密钥会覆盖全局默认密钥,从而实现"一次运行一把钥匙"的精细控制。

附加说明

  • 可以在 OpenAI 追踪仪表板中查看免费追踪记录;即便使用非 OpenAI 模型,只要配置了导出密钥,同样可以在这里获得可视化视图。
  • 追踪数据在导出前先经内存队列缓冲,再分批上报;理解BatchTraceProcessor的调度(周期、阈值、批量大小、重试退避)有助于你判断"数据何时可见"。

生态系统集成

以下社区和供应商集成都支持 OpenAI Agents SDK 的追踪 API 接口(即上文的自定义追踪处理器机制),可用于将追踪数据接入各自的观测平台:

Weights & Biases、Arize Phoenix、Future AGI、MLflow(自托管/OSS 与 Databricks 托管)、Braintrust、Pydantic Logfire、AgentOps、Scorecard、Respan、LangSmith、Maxim AI、Comet Opik、Langfuse、Langtrace、Okahu-Monocle、Galileo、Portkey AI、LangDB AI、Agenta、PostHog、Traccia、PromptLayer、HoneyHive、Asqav、Datadog、Latitude、DProvenanceKit、Tuning Engines。

这些集成大多以add_trace_processor()set_trace_processors()为接入点,把 SDK 输出的 Trace/Span 事件流转入各自的可观测后端。你也可以参考 processors.py 中的ConsoleSpanExporter作为最小实现模板,编写自己的处理器。

小结

OpenAI Agents SDK 的追踪能力覆盖了从"零配置默认埋点"到"深度定制导出"的完整路径:默认情况下,一次Runner.run()会被自动记录为包含 task、turn、agent、generation、function、guardrail、handoff 等跨度的端到端追踪;通过trace()可以合并多次运行为一条追踪;通过flush_traces()可以在长期运行进程中即时导出;通过RunConfig可以控制敏感数据采集与单次运行的追踪密钥;通过add_trace_processor()/set_trace_processors()可以将数据接入任意后端,包括在非 OpenAI 模型场景下继续使用 OpenAI 追踪仪表板。结合 src/agents/tracing/ 的源码阅读,你可以进一步理解缓冲、批量、重试与上下文传播的底层细节,从而在生产环境中做出更精准的观测与排障决策。

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询