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:
Trace与Span的抽象基类及默认实现; - processors.py:
BatchTraceProcessor、BackendSpanExporter等导出链路; - setup.py:全局
TraceProvider的懒初始化与进程退出时的自动关闭。
三种常见的禁用方式
如果出于隐私、成本或其他原因需要关闭追踪,可以按需选择以下三种方式之一:
- 全局禁用(环境变量):设置环境变量
OPENAI_AGENTS_DISABLE_TRACING=1; - 全局禁用(代码):调用
set_tracing_disabled(True),对应实现见init.py 中的set_tracing_disabled(),它会将该开关写入全局TraceProvider; - 单次运行禁用:将
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_at与ended_at时间戳(ISO 8601 格式,由 util.py 中的time_iso()生成);trace_id:所属追踪记录的 ID;parent_id:父跨度的 ID(若存在);span_data:跨度承载的业务信息。例如AgentSpanData包含智能体信息,GenerationSpanData包含 LLM 生成信息,依此类推。
SpanImpl在start()时记录开始时间并通知处理器,在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.0、max_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}")- 由于对
Runner.run的两次调用都被封装在with trace()中,两次运行会成为同一条整体追踪记录的一部分,而不是各自创建一条单独的追踪记录。
之所以可行,正是因为"当前追踪"通过contextvar跟踪:trace()进入上下文时成为当前追踪,内部的多次运行器调用及其所有跨度都自动挂载到它之下,形成一棵完整的端到端调用树。
创建追踪记录(Trace)
你可以使用trace()函数(create.py)创建追踪记录。追踪记录需要启动和结束,有两种方式:
- 推荐:将追踪记录用作上下文管理器,即
with trace(...) as my_trace:。这会自动在正确的时机启动和结束追踪(__enter__调用start(mark_as_current=True),__exit__调用finish(reset_current=True))。 - 手动:调用
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, ) -> Traceworkflow_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() | 记录一次智能体运行 | name、handoffs、tools、output_type |
task_span() | 记录一次顶层 Runner 调用 | name |
turn_span() | 记录一轮智能体循环 | turn、agent_name |
generation_span() | 记录一次 LLM 生成 | input、output、model、model_config、usage |
function_span() | 记录一次函数工具调用 | name、input、output |
guardrail_span() | 记录一次防护措施 | name、triggered |
handoff_span() | 记录一次任务转移 | from_agent、to_agent |
custom_span() | 记录自定义操作 | name、data |
transcription_span() | 记录语音转文本 | model、input、input_format(默认pcm)、output |
speech_span() | 记录文本转语音 | model、input、output、output_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_data为True。若不想编写代码,可以在运行应用前导出环境变量来设置默认值:
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_ID、OPENAI_PROJECT_ID指定组织与项目)。
从 setup.py 可以看到,全局 provider 与默认处理器采用懒初始化:首次访问追踪功能时才创建(避免模块导入阶段就建立网络连接或线程),并通过atexit注册进程退出时的关闭刷新。
两种自定义方式
若要改变默认行为——把追踪发送到其他/额外的后端,或修改导出器的行为——有两种方式:
add_trace_processor():添加一个额外的追踪处理器,它会在追踪记录和跨度准备就绪时接收它们。这样,除了发送到 OpenAI 后端之外,你还可以自行处理(例如写入本地日志、推送到自建可观测平台)。实现上,它会调用get_trace_provider().register_processor(processor)(见init.py)。set_trace_processors():用你自己的处理器列表替换默认处理器。这意味着,除非你的处理器列表中包含一个执行发送的TracingProcessor,否则追踪记录不会发送到 OpenAI 后端。实现上对应get_trace_provider().set_processors(processors)。
TracingProcessor是处理器需要实现的接口(processor_interface.py),核心回调包括on_trace_start、on_trace_end、on_span_start、on_span_end、force_flush与shutdown等。仓库还提供了一个将追踪打印到控制台的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_key与include_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),仅供参考