openai-agents-python 追踪体系深度解析Span 生命周期、实现原理与自定义扩展【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python在 openai-agents-pythonOpenAI Agents SDK的分布式追踪体系中Span跨度是描述单次可观测操作的最小单元——一次 LLM 生成、一次工具调用、一次 Agent 运行或一次 Guardrail 检查都会被封装成一个 Span 并挂载到 Trace端到端工作流之下。本文基于仓库中 docs/ref/tracing/spans.md 对应的 API 参考深入源码实现src/agents/tracing/spans.py与配套模块完整讲解Span抽象基类、SpanImpl与NoOpSpan两种实现、Span 的生命周期管理、父子嵌套机制、错误记录与导出协议并给出自定义 Span 的实战写法。读完本文你将掌握如何为多 Agent 工作流埋点、如何在追踪关闭时优雅降级、以及如何编写自定义 Tracing Processor 消费 Span 数据。一、先建立整体认知Trace 与 Span 的关系在深入 Span 之前先明确它在整个追踪体系中的位置。参考 docs/tracing.md 的说明Trace代表一次端到端的工作流操作如代码生成客户服务包含workflow_name、trace_id、可选的group_id与metadataSpan代表有明确开始和结束时间的操作隶属于某个 Trace并可通过parent_id与父 Span 组成树状层级。一次典型的Runner.run()会默认产生如下嵌套结构各环节分别由task_span、turn_span、agent_span、generation_span、function_span、guardrail_span、handoff_span等工厂函数创建Trace(Agent workflow) └── TaskSpan一次顶层 Runner 调用 └── TurnSpan一次 Agent 循环轮次 └── AgentSpan某个 Agent 的运行 ├── GenerationSpanLLM 生成 ├── FunctionSpan函数工具调用 └── HandoffSpanAgent 间交接这些工厂函数全部位于 src/agents/tracing/create.py最终都汇聚到同一个入口get_trace_provider().create_span(...)。也就是说无论创建哪种 Span底层的对象构造、父级推断、禁用降级逻辑都是统一的。二、Span 抽象基类可观测操作的契约Span定义于 src/agents/tracing/spans.py是一个泛型抽象基类class Span(abc.ABC, Generic[TSpanData])。泛型参数TSpanData约束为该 Span 携带的操作数据SpanData的子类如GenerationSpanData、AgentSpanData。从源码看Span暴露了以下核心抽象属性与方法这也是所有实现SpanImpl、NoOpSpan必须遵循的契约成员类型含义trace_idstr该 Span 所属 Trace 的唯一 IDspan_idstrSpan 自身的唯一 ID同一 Trace 内唯一span_dataTSpanData操作专属数据如 LLM 生成的输入/输出/用量parent_idstr \| None父 Span 的 ID为None表示根 Spanstarted_at/ended_atstr \| NoneISO 8601 格式的开始/结束时间戳未开始/未结束时为NoneerrorSpanError \| None执行期间发生的错误详情tracing_api_keystr \| None导出该 Span 时使用的 API Keytrace_metadatadict \| None继承自 Trace 的元数据基类默认返回Nonestart(mark_as_current)—启动 Span可选地将其标记为当前 Spanfinish(reset_current)—结束 Span可选地恢复之前的当前 Spanset_error(error)—记录错误SpanError为 TypedDictmessage: str 可选data: dictexport()dict \| None导出为可序列化的字典供 Processor 消费__enter__/__exit__—支持with上下文管理器用法SpanError错误记录的数据结构SpanError是一个TypedDictsrc/agents/tracing/spans.pyclass SpanError(TypedDict): message: str # 人类可读的错误描述 data: dict[str, Any] | None # 可选的附加错误上下文它通过span.set_error(...)挂到 Span 上并在export()时随载荷一并输出供后端仪表盘或自定义 Processor 分析失败原因。三、三种 Span 对象的角色分工虽然Span是抽象基类但开发者实际接触到的只有两种具象类型——SpanImpl与NoOpSpan二者都实现于 src/agents/tracing/spans.py。理解为什么需要两种实现是理解整个追踪开关机制的关键。3.1 SpanImpl真正记录的 SpanSpanImpl是标准实现负责真实数据的采集与上报。它的构造参数src/agents/tracing/spans.py包括trace_id所属 Trace 的 IDspan_id可选未传时由util.gen_span_id()生成parent_id父 Span IDNone表示根 SpanprocessorTracingProcessorSpan 生命周期事件的消费者span_data操作数据tracing_api_key导出用 API Keytrace_metadata可选从 Trace 继承的元数据。其关键行为src/agents/tracing/spans.pystart()设置_started_at util.time_iso()然后调用self._processor.on_span_start(self)通知处理器若mark_as_currentTrue通过Scope.set_current_span(self)将自身登记为当前 Span。finish()设置_ended_at调用self._processor.on_span_end(self)若reset_currentTrue且有先前 token则恢复之前的当前 Span。幂等保护重复start()会打印Span already started警告并直接返回重复finish()同理Span already finished避免状态被二次覆盖。set_error()将错误保存到_error随导出输出。3.2 NoOpSpan追踪关闭时的优雅降级NoOpSpansrc/agents/tracing/spans.py是不记录任何数据的空实现用于追踪被禁用但代码仍需正常运行的场景。它仍然维护span_id/trace_id固定返回字符串no-op支持上下文管理器语义与mark_as_current/reset_current的 Scope 登记/恢复但export()返回Noneset_error()为空操作started_at/ended_at均为None。为什么需要它追踪体系贯穿 Agent 运行的核心路径如果禁用追踪时直接抛错或返回None会让业务代码被迫写满判空逻辑。NoOpSpan保证了接口兼容即使OPENAI_AGENTS_DISABLE_TRACING1所有埋点代码依然可执行只是静默丢弃数据。这正是接口隔离Interface Segregation与空对象模式Null Object Pattern的典型应用。从 src/agents/tracing/provider.py 的create_span可以看到NoOpSpan的触发条件全局或单次调用禁用self._disabled or disabled传入的span_id是no-op没有活动 Trace 时No active trace. Make sure to start a trace withtrace()first父 Trace 或父 Span 本身是 NoOp_is_noop_trace/_is_noop_span判定。也就是说NoOp 具有传染性一旦父级是空操作子 Span 也会自动降级为空操作从而避免在禁用状态下产生半真半假的追踪数据。3.3 工厂函数面向业务的标准入口开发者通常不需要直接new SpanImpl(...)而是调用 src/agents/tracing/create.py 中的工厂函数。它们共享同一组参数约定span_id可选推荐用util.gen_span_id()生成以保证格式正确parent可选指定父 Trace 或 Span不传时自动以当前 Trace/当前 Span为父级disabled为True时返回NoOpSpan。常用工厂及其对应的SpanData类型工厂函数SpanData典型用途agent_span(name, handoffs, tools, output_type)AgentSpanData单个 Agent 的运行task_span(name)TaskSpanData一次顶层 Runner 调用turn_span(turn, agent_name)TurnSpanData一次 Agent 循环轮次generation_span(input, output, model, model_config, usage)GenerationSpanDataLLM 生成function_span(name, input, output)FunctionSpanData函数工具调用handoff_span(from_agent, to_agent)HandoffSpanDataAgent 交接guardrail_span(name, triggered)GuardrailSpanData护栏检查custom_span(name, data)CustomSpanData自定义埋点response_span(response)ResponseSpanData仅捕获 Response 标识transcription_span(...)/speech_span(...)/speech_group_span(...)语音相关语音管线STT/TTSmcp_tools_span(server, result)MCPListToolsSpanDataMCP 服务器工具列表四、Span 生命周期从 start 到 finish 的全过程4.1 推荐的上下文管理器用法源码文档src/agents/tracing/spans.py明确推荐使用with语法以保证 start/finish 的可靠配对from agents.tracing import custom_span # 基础用法with 块进入时自动 start(mark_as_currentTrue)退出时自动 finish(reset_currentTrue) 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)} # 错误处理在 with 块内捕获异常并 set_error然后重新抛出 with custom_span(risky_operation) as span: try: result perform_risky_operation() except Exception as e: span.set_error({ message: str(e), data: {operation: risky_operation}, }) raise注意span.span_data是可变对象CustomSpanData.data是属性包可以在 Span 存活期间动态补充字段——这在记录查询结果、中间状态时非常实用。4.2 手动管理start/finish 显式调用工厂函数创建出的 Span不会自动启动。若不便使用with需手动配对调用span custom_span(manual_work) span.start(mark_as_currentTrue) # 标记为当前 Span子 Span 会自动挂到它下面 try: # ... 业务逻辑 ... pass finally: span.finish(reset_currentTrue) # 恢复之前的当前 Span4.3 底层机制contextvars 与 ScopeSpan 的当前性与自动父子嵌套依赖 src/agents/tracing/scope.py 中基于contextvars实现的Scope类_current_span: contextvars.ContextVar[Span[Any] | None] contextvars.ContextVar( current_span, defaultNone )Scope.set_current_span(span)返回一个TokenScope.reset_current_span(token)用于恢复之前的值。这正是start(mark_as_currentTrue)/finish(reset_currentTrue)的内部实现——通过 token 精确恢复上下文保证嵌套正确且线程/协程安全contextvars天然隔离异步任务之间的上下文。DefaultTraceProvider.create_span在未显式传入parent时正是通过Scope.get_current_span()与Scope.get_current_trace()推断父级src/agents/tracing/provider.py有当前 Span 则挂到其下否则挂到当前 Trace 下parent_id取当前 Span 的span_id。4.4 边界情况GeneratorExit 的容错处理Span的__exit__对GeneratorExit有专门的容错路径src/agents/tracing/spans.py。当异步生成器被aclose()关闭时with块会以GeneratorExit解退。若生成器是在其他任务中被aclose终结恢复代码运行的上下文可能从未设置过该 token直接ContextVar.reset会抛ValueError。为此_finish_on_generator_exitsrc/agents/tracing/spans.py会以finish(reset_currentFalse)正常结束 Span不强制重置尝试Scope.reset_current_span(token)捕获ValueError并降级为 debug 日志Skipping span context reset, token belongs to another context。也就是说该容错只覆盖 GeneratorExit 这一条路径显式从错误上下文调用finish仍会抛出属于上下文所有权违规。仓库的 tests/tracing/test_spans_impl.py 对SpanImpl与NoOpSpan两种实现都参数化验证了同任务内关闭生成器必须恢复调用方 Span这一行为test_generator_close_in_the_same_task_releases_the_span_scope。五、SpanDataSpan 携带的操作数据span_data是 Span 的灵魂。抽象基类SpanDatasrc/agents/tracing/span_data.py只要求两个成员export() - dict将数据导出为字典type - strSpan 类型标识。具体子类按操作类型区分例如AgentSpanDataname、handoffs可交接的 Agent 列表、tools可用工具列表、output_type导出typeagentGenerationSpanDatainput输入消息序列、output输出消息序列、model、model_config超参数、usagetoken 用量导出typegenerationFunctionSpanDataname、input、output以及可选的mcp_data导出typefunction输出会str()序列化HandoffSpanDatafrom_agent、to_agent导出typehandoffGuardrailSpanDataname、triggered是否触发护栏CustomSpanDataname 任意data属性包导出typecustom——这是自定义埋点的主力语音相关TranscriptionSpanDataSTTinput_format默认pcm、SpeechSpanDataTTSoutput_format默认pcm含first_content_at首字节时间、SpeechGroupSpanDataTaskSpanData/TurnSpanData内部以typecustomsdk_span_type字段导出TaskSpanData代表顶层 Runner 调用TurnSpanData代表单次 Agent 循环轮次MCPListToolsSpanDataserverresult工具列表导出typemcp_tools。这些数据类的字段与工厂函数参数一一对应在设计自定义 Span 时可直接参照它们的export()实现来保证序列化一致性。六、导出协议Span 如何离开进程SpanImpl.export()src/agents/tracing/spans.py产出的载荷结构为{ object: trace.span, id: span_id, trace_id: trace_id, parent_id: parent_id | null, started_at: ISO 时间戳, ended_at: ISO 时间戳, span_data: { ...: 由 span_data.export() 产生 }, error: { message: ..., data: null } }此外export()还会做元数据路由合并从 Trace 级trace_metadata中挑选_SPAN_METADATA_ROUTING_KEYS当前为(agent_harness_id,)对应的键再合并span_data.metadata若存在且为 dict后者不覆盖前者。合并结果非空时才写入载荷的metadata字段。这为多租户路由如按 harness ID 分发保留了扩展点。export()的结果由TracingProcessor消费。处理器接口定义于 src/agents/tracing/processor_interface.py包含四个生命周期回调on_trace_start、on_trace_end、on_span_start、on_span_end外加shutdown()与force_flush()。仓库默认的SynchronousMultiTracingProcessorsrc/agents/tracing/provider.py会按注册顺序把事件转发给所有处理器且每个处理器异常都会被捕获并记录不影响 Agent 主流程的执行——这是追踪系统的容错设计底线。若要实现自定义处理器只需继承TracingProcessor并实现上述六个方法通过set_tracing_processors([...])或add_tracing_processor(...)注册后即可在on_span_end中拿到完整的span.export()载荷实现日志输出、指标采集或推送到自建后端。七、实践组合使用 trace custom_span 的完整示例结合 docs/tracing.md 的高层 Trace 用法与自定义 Span一个典型的业务埋点如下import asyncio from agents import Agent, Runner, trace from agents.tracing import custom_span, flush_traces async def main(): agent Agent(nameJoke generator, instructionsTell funny jokes.) # 用 trace 包裹多次 run让它们属于同一个工作流 with trace(Joke workflow, group_idchat_42, metadata{user: user_123}) as t: first_result await Runner.run(agent, Tell me a joke) # 业务自定义埋点自动嵌套在当前的 agent_span 之下 with custom_span(post_process, {source: joke_rating}) as span: second_result await Runner.run(agent, fRate this joke: {first_result.final_output}) span.span_data.data[rating] second_result.final_output print(fJoke: {first_result.final_output}) print(fRating: {second_result.final_output}) # 长驻进程Celery/FastAPI 后台任务中需要即时导出时 flush_traces() asyncio.run(main())要点回顾custom_span不传parent时会自动挂到当前Scope中的活动 Span/Trace 之下无需手动指定父子关系with语法保证异常路径下 Span 依然被正确结束__exit__总会执行finish需要立即上报时调用flush_traces()参考 docs/tracing.md 的Long-running workers and immediate exports一节若整个服务需要关闭追踪可设置环境变量OPENAI_AGENTS_DISABLE_TRACING1此时所有 Span 自动降级为NoOpSpan业务代码无需任何改动。八、小结与扩展阅读Span 是 openai-agents-python 追踪体系的核心抽象Span基类定义契约SpanImpl负责真实采集NoOpSpan保证禁用时的优雅降级Scope基于contextvars维护当前上下文实现自动嵌套SpanData子类刻画每种操作的数据形态export()则与TracingProcessor配合把数据送出进程。这套设计让追踪既有默认开箱即用的完整埋点又允许开发者用custom_span与自定义 Processor 深度定制。源码与测试Span实现见 src/agents/tracing/spans.py工厂函数见 src/agents/tracing/create.py数据类见 src/agents/tracing/span_data.pyProvider 与降级逻辑见 src/agents/tracing/provider.py生成器退出边界测试见 tests/tracing/test_spans_impl.py相关参考页Trace 契约见 docs/ref/tracing/traces.mdSpan 数据见 docs/ref/tracing/span_data.md处理器接口见 src/agents/tracing/processor_interface.py使用指南完整追踪功能说明见 docs/tracing.md。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考