openai-agents-python 追踪系统核心深入解析 Trace 类与端到端工作流追踪【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python导读Trace追踪是 openai-agents-python 内建可观测性体系的顶层抽象代表一次完整端到端的工作流如客户服务查询、代码生成内部承载该工作流期间产生的所有 Span 与元数据。本文以 docs/ref/tracing/traces.md 对应的agents.tracing.traces模块为骨架结合源码逐层拆解Trace抽象类、TraceImpl真实实现、NoOpTrace降级实现、TraceState状态序列化与ReattachedTrace上下文重挂载机制并串联trace()工厂函数、TraceProvider与Scope的调用关系。读完本文你将掌握如何正确创建、启停、导出与持久化 Trace理解追踪禁用时的行为并能在多工作流、长时运行、会话恢复等场景中合理运用 Trace API。一、Trace 是什么工作流的全景容器在 agents 的追踪模型里Trace 与 Span 是两层核心抽象对应 docs/tracing.md 的说明Trace表示一个逻辑工作流或一次完整操作例如Customer Service、Code Generation它由多个 Span 组合而成Span表示有明确开始与结束时间的单个操作例如一次 LLM 生成、一次函数工具调用、一次 handoff。Trace类在 src/agents/tracing/traces.py 中定义为抽象基类abc.ABC其 docstring 给出了定位A trace represents a logical workflow or operation and contains all the spans (individual operations) that occur during that workflow.作为抽象类Trace定义了所有追踪实现必须遵循的统一契约具体实现包括实现类位置用途TraceImplsrc/agents/tracing/traces.py真正被追踪库记录的 Trace接入TracingProcessorNoOpTracesrc/agents/tracing/traces.py追踪被禁用时的空实现保持上下文管理但不记录数据ReattachedTracesrc/agents/tracing/traces.py从持久化状态重建的 Trace 上下文不重新发出 trace start 事件Trace 的核心属性契约抽象类通过属性与抽象方法约定每个 Trace 必须具备的能力trace_id全局唯一标识符格式为trace_32位字母数字用于把 Span 关联到所属 Trace也可在追踪看板中按此 ID 检索src/agents/tracing/traces.pyname人类可读的工作流名称如Customer Service用于看板分组与过滤src/agents/tracing/traces.pyexport()把 Trace 数据导出为可序列化字典当追踪被禁用时返回Nonesrc/agents/tracing/traces.pytracing_api_key导出该 Trace 及其 Span 时使用的 API Keysrc/agents/tracing/traces.py。此外Trace.to_json(include_tracing_api_keyFalse)提供序列化封装src/agents/tracing/traces.py它先调用export()若返回None则整体返回None否则在副本中加入可选的tracing_api_key。include_tracing_api_key默认False正是为了避免把密钥意外持久化。二、Trace 生命周期上下文管理器优先手动启停兜底官方文档明确给出两种启停方式docs/tracing.md推荐上下文管理器——with trace(...) as my_trace:自动在正确时机 start 与 finish手动调用——trace.start()与trace.finish()。Trace抽象类为此分别定义了__enter__/__exit__src/agents/tracing/traces.py与start/finish抽象方法src/agents/tracing/traces.pystart(mark_as_currentFalse)启动 Trace必须早于任何 Span 的创建mark_as_currentTrue时把自身标记为执行上下文中的当前 Tracefinish(reset_currentFalse)结束 Trace会收尾所有未关闭的 Spanreset_currentTrue时把当前 Trace 恢复为上下文中的上一个 Trace。这两个方法都被注明是线程安全的在mark_as_current/reset_current场景下这一点对并发 Agent 运行至关重要。上下文管理器中的嵌套与恢复以TraceImpl.__enter__src/agents/tracing/traces.py为例def __enter__(self) - Trace: if self._started: if not self._prev_context_token: logger.error(Trace already started but no context token set) return self self.start(mark_as_currentTrue) return self进入with块时自动调用start(mark_as_currentTrue)__exit__src/agents/tracing/traces.py则调用finish(reset_currentTrue)。也就是说嵌套的with trace()可以安全地保存并恢复上一个 Trace退出内层块后外层 Trace 依然是当前 Trace。生成器异常路径的容错模块顶部还实现了一个精巧的边界处理函数_finish_on_generator_exitsrc/agents/tracing/traces.py当 Trace 的with块因GeneratorExit展开例如异步生成器被其他任务aclose()收尾时负责 finish 的协程并非创建该 Trace 上下文的那个任务直接ContextVar.reset会抛出ValueError。因此该函数在finally中执行 reset并捕获令牌属于其他上下文的ValueError后仅记录 debug 日志——明确把容错范围限定在这一不可避免的路径而显式从错误上下文调用finish仍视为上下文所有权违规照常抛错。这是理解 SDK 在异常/并发场景下不会因追踪而崩溃的关键细节。三、如何创建 Tracetrace()工厂与TraceProvider日常使用中你不会直接实例化TraceImpl而是通过agents.tracing.trace()工厂函数src/agents/tracing/create.pydef 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_namestr必填逻辑应用/工作流名称如code_bot、customer_support_agenttrace_idstr \| None自动生成Trace 唯一 ID推荐用util.gen_trace_id()生成以保证格式正确group_idstr \| NoneNone分组标识把同一会话的多个 Trace 关联起来例如聊天线程 IDmetadatadict \| NoneNone附加到 Trace 上的任意用户自定义元数据tracingTracingConfig \| NoneNone该 Trace 的导出配置可携带独立 API KeydisabledboolFalse为True时返回 Trace 但不记录任何数据两个值得注意的行为重复创建告警如果当前上下文中已存在一个 Trace再调用trace()会记录一条 warningTrace already exists. Creating a new trace, but this is probably a mistake.——这提示你把多次Runner.run放进同一个with trace()以形成高层级 Trace而不是嵌套创建新 Trace参考 docs/tracing.md 的Joke workflow示例创建 ≠ 启动工厂函数返回的 Trace不会自动启动必须用with或手动start()。底层TraceProvider.create_tracetrace()最终委托给全局TraceProvider.create_trace()src/agents/tracing/provider.py其流程是_refresh_disabled_flag()刷新全局禁用标志若全局禁用或传入了disabledTrue返回NoOpTrace()不记录否则生成trace_id缺省时构造TraceImpl并把TracingConfig中的api_key作为该 Trace 的tracing_api_key传入。return TraceImpl( namename, trace_idtrace_id, group_idgroup_id, metadatametadata, processorself._multi_processor, tracing_api_keytracing.get(api_key) if tracing else None, )TraceImpl.start()src/agents/tracing/traces.py会调用self._processor.on_trace_start(self)通知所有已注册的TracingProcessorfinish()则调用on_trace_end(self)——这是追踪数据流入导出管线的入口。追踪禁用与NoOpTrace全局禁用有三种常见方式docs/tracing.md 顶部说明环境变量OPENAI_AGENTS_DISABLE_TRACING1代码中set_tracing_disabled(True)定义于 src/agents/tracing/init.py单次运行通过RunConfig.tracing_disabled关闭。禁用后create_trace返回NoOpTracesrc/agents/tracing/traces.py。它保持完整的上下文管理语义with照样进出、照样设置/恢复 current trace但trace_id与name恒为字符串no-opexport()恒返回Noneto_json也随之返回Nonetracing_api_key恒为None。NoOpTrace还提供了模块级单例NO_OP_TRACEsrc/agents/tracing/traces.py。这种接口照常可用、数据一概不记的设计让上层 Agent 运行逻辑完全无需感知追踪是否开启。四、当前 Trace 的跟踪机制Scope与 contextvars当前 Trace是如何随并发自动传播的答案在 src/agents/tracing/scope.py。模块用两个contextvars.ContextVar分别保存当前 Span 与当前 Trace_current_span: contextvars.ContextVar[Span[Any] | None] contextvars.ContextVar(current_span, defaultNone) _current_trace: contextvars.ContextVar[Trace | None] contextvars.ContextVar(current_trace, defaultNone)Scope类提供四个关键操作get_current_trace()/set_current_trace(trace)/reset_current_trace(token)get_current_span()/set_current_span(span)/reset_current_span(token)由于ContextVar天然按上下文隔离并发的 asyncio 任务各自持有独立的当前 Trace视图——这正是 docs/tracing.md 所说current trace is tracked via a Python contextvar, meaning it works with concurrency automatically的底层实现。TraceImpl在start(mark_as_currentTrue)时通过Scope.set_current_trace(self)拿到Token存入_prev_context_tokenfinish(reset_currentTrue)时用该 Token 恢复——Token只在创建它的那个Context中有效这也是上一节生成器容错逻辑存在的原因。同理TraceProvider.create_spansrc/agents/tracing/provider.py在未显式传parent时会读取当前 Span/Tracespan 自动挂到最近的当前 Span之下parent_id current_span.span_idtrace 归属取current_trace.trace_id并继承当前 Trace 的tracing_api_key与 metadata。若当前没有任何 Trace则返回NoOpSpan并给出调试提示No active trace. Make sure to start a trace withtrace()first。五、Trace 的序列化TraceState与to_json为了让 Trace 信息能够跨进程持久化例如存入 RunState 快照、会话恢复后重挂载模块提供了两个关键能力。Trace.to_json与export的载荷结构TraceImpl.export()src/agents/tracing/traces.py导出的结构为{ object: trace, id: self.trace_id, workflow_name: self.name, group_id: self.group_id, metadata: self.metadata, }to_json(include_tracing_api_keyFalse)在其基础上按需追加tracing_api_key字段src/agents/tracing/traces.py。注意export()返回的 metadata 是原引用to_json会做一次dict(exported)浅拷贝避免污染内部状态。TraceState可序列化的 Trace 元数据TraceState是一个 dataclasssrc/agents/tracing/traces.py字段包括字段含义trace_idTrace IDworkflow_name工作流名称group_id分组 IDmetadata元数据字典tracing_api_key显式追踪密钥原始值仅内存态tracing_api_key_hash密钥的 SHA-256 指纹object_type对象类型标记extra其他未识别字段的兜底容器关键方法是TraceState.from_trace(trace)src/agents/tracing/traces.py对 Trace 调用to_json(include_tracing_api_keyTrue)后转成状态对象TraceState.from_json(payload)src/agents/tracing/traces.py从字典还原兼容id/trace_id两种键名TraceState.to_json(include_tracing_api_keyFalse)src/agents/tracing/traces.py反向序列化若所有核心字段为空则返回None。密钥指纹而非明文_hash_tracing_api_key模块专门实现了_hash_tracing_api_keysrc/agents/tracing/traces.pydef _hash_tracing_api_key(tracing_api_key: str | None) - str | None: if tracing_api_key is None: return None return hashlib.sha256(tracing_api_key.encode(utf-8)).hexdigest()其设计意图源码注释明确说明持久化时只保存指纹以便恢复运行可以校验是否使用了相同的显式追踪密钥而无需存储明文密钥。from_json中若快照被安全化剥离了原始密钥tracing_api_key_hash已存在而 raw key 缺失会保留已存储的指纹用于恢复期匹配to_json则总是输出tracing_api_key_hash保证默认的 RunState 快照依然能够校验显式恢复密钥。这一机制与 src/agents/run_state.py 中TraceState的存取相互配合。六、会话恢复与 Trace 重挂载reattach_trace与ReattachedTrace当 Agent 运行被打断并需要从持久化的 RunState 恢复时SDK 需要重建Trace 上下文——但不能重新触发一次on_trace_start否则会在看板上产生一条新 Trace 记录。这正是ReattachedTrace与reattach_trace()的职责src/agents/tracing/traces.py。reattach_trace(trace_state, *, tracing_api_keyNone)的逻辑def reattach_trace(trace_state: TraceState, *, tracing_api_key: str | None None) - Trace | None: if trace_state.trace_id is None: return None return ReattachedTrace( nametrace_state.workflow_name or Agent workflow, trace_idtrace_state.trace_id, group_idtrace_state.group_id, metadatadict(trace_state.metadata) if trace_state.metadata is not None else None, tracing_api_key( trace_state.tracing_api_key if trace_state.tracing_api_key is not None else tracing_api_key ), )要点若状态中没有trace_id返回None无法重挂载工作流名缺省时回退为Agent workflow追踪密钥优先取状态中保存的原始值否则用调用方显式传入的密钥兜底。ReattachedTrace.start()src/agents/tracing/traces.py与TraceImpl的关键差异是它只把 trace_id 登记进已启动 ID集合并设置 current trace绝不调用processor.on_trace_start因此不会重复上报 start 事件。其export()载荷结构与TraceImpl完全一致保证恢复后的 Trace 与原始 Trace 在看板上合并为同一条。已启动 Trace ID 的登记与上限模块还维护了一个带锁的OrderedDict_started_trace_ids上限_MAX_STARTED_TRACE_IDS 4096见 src/agents/tracing/traces.py用于去重与 LRU 淘汰_mark_trace_id_started登记重复则移到队尾超出上限时弹出最旧条目_trace_id_was_started查询。NoOpTrace的no-opID 与空值会被直接跳过避免无意义登记。这为同一 trace_id 只应启动一次提供了进程内保障。七、实战把 Trace API 组合起来基础用法单次工作流追踪from agents import Agent, Runner, trace agent Agent(nameJoke generator, instructionsTell funny jokes.) # 两次 run 合并在同一条 Trace 中高层级 trace参考 docs/tracing.md#higher-level-traces with trace(Joke workflow) as t: 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})携带分组与元数据with trace( Customer Service, group_idchat_123, # 同一会话多条 Trace 归组 metadata{customer: user_456}, # 供看板过滤/分析 ) as t: result await Runner.run(support_agent, query)手动启停含并发语义from agents.tracing import trace t trace(Manual Workflow) t.start(mark_as_currentTrue) # 更新当前 Trace线程安全 try: # ... 执行 Agent 运行Span 自动挂到当前 Trace 下 ... pass finally: t.finish(reset_currentTrue) # 恢复上一个 Trace长时运行任务中确保立即导出默认BatchTraceProcessor每隔数秒在后台批量导出进程退出时还会最终 flush在 Celery/RQ/FastAPI 后台任务等长时运行场景若需要任务结束即刻可见可在with trace()退出后调用flush_traces()定义于 src/agents/tracing/init.py它会force_flush所有已注册处理器docs/tracing.mdfrom 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()自定义处理器的接入点Trace的生命周期事件正是所有可观测性扩展的挂载点TracingProcessor接口src/agents/tracing/processor_interface.py定义了on_trace_start、on_trace_end、on_span_start、on_span_end、shutdown、force_flush六个抽象方法通过add_trace_processor()追加处理器或set_trace_processors()整体替换默认处理器src/agents/tracing/init.py即可把 Trace/Span 数据推送到自建后端。Trace 在这里扮演的正是一次工作流的完整快照这一数据单元。八、使用建议与注意事项结合Trace类 docstring 与源码实现可以沉淀以下最佳实践使用描述性的工作流名称name直接用于看板分组与过滤Agent workflow这类默认名会让所有任务难以区分用一致的group_id归组相关 Trace多轮对话场景把同一聊天线程 ID 传入能串起完整会话脉络合理使用 metadata附加客户 ID、请求来源等可过滤信息但需注意隐私——docstring 明确提示Consider privacy when adding trace data优先上下文管理器with trace()自动处理 start/finish 与 current-trace 恢复规避手动启停时遗漏reset_current导致的上下文串扰不要在已存在的 Trace 内无谓地再建 Tracetrace()会发出 warning多个Runner.run应共享同一个外层 Trace敏感数据控制generation_span/function_span会记录 LLM 输入输出等潜在敏感内容可通过RunConfig.trace_include_sensitive_data关闭Trace.to_json默认不携带tracing_api_key持久化时优先使用指纹tracing_api_key_hash而非明文参考 docs/tracing.md。相关资源导航模块文档docs/ref/tracing/traces.mdagents.tracing.traces全量 API 参考追踪总览docs/tracing.mdTraces/Span 概念、默认追踪、导出与处理器核心实现src/agents/tracing/traces.py、src/agents/tracing/provider.py、src/agents/tracing/scope.py工厂与工具src/agents/tracing/create.py、src/agents/tracing/util.py处理器接口与导出src/agents/tracing/processor_interface.py、src/agents/tracing/processors.py【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考