OpenAI Agents SDK #7：Tracing——让 Agent 的每一步都「可见」

为什么 Agent 调试这么难？

Tracing 的本质：两层结构

Agent Loop 的全链路 Span

Trace: customer-service-run
├── AgentSpan: CustomerServiceAgent
│   ├── LLMSpan: gpt-4o (理解用户问题)
│   ├── ToolSpan: search_knowledge_base (查询知识库)
│   ├── LLMSpan: gpt-4o (生成回答)
│   └── HandoffSpan: → HumanAgentHandoff (转人工)

• AgentSpanData：记录 Agent 名称、工具列表、输出类型
• LLMSpanData：记录模型名称、请求 token、响应内容（可选过滤敏感数据）
• ToolSpanData：记录工具名、输入参数、输出结果
• HandoffSpanData：记录源 Agent、目标 Agent、触发原因
• GuardrailSpanData：记录护栏名称、触发与否、过滤结果
• CustomSpanData：你自定义事件的任意数据

create_trace 和 create_span：手动控制追踪

1. 多次 Runner.run() 属于同一个逻辑工作流，想把它们归在一个 Trace 下
2. 需要追踪 Agent Loop 以外的自定义操作，比如数据预处理步骤

from agents import Runner, trace, custom_span
from agents.tracing import create_trace, create_span

async def run_research_workflow(query: str):
    # 用 trace() 创建顶层 Trace，覆盖整个工作流
    # workflow_name 会显示在 Dashboard 的 Trace 列表里
    with trace("research-workflow", metadata={"query": query}):

# 第一阶段：数据采集（自定义 Span，追踪非 Agent 操作）
        with custom_span("data-collection"):
            raw_data = fetch_external_data(query)   # 你自己的函数

# 第二阶段：Agent 分析（自动产生子 Span）
        result = await Runner.run(
            research_agent,
            input=f"分析以下数据：{raw_data}"
        )

# 第三阶段：结果存储（另一个自定义 Span）
        with custom_span("result-storage"):
            save_to_db(result.final_output)

return result

内置 Processor vs 自定义 Processor

from agents import set_trace_processors
from langfuse.openai_agents import LangfuseTraceProcessor

# 替换默认 Processor，追踪数据发往 Langfuse
set_trace_processors([LangfuseTraceProcessor()])

# 或者追加（同时发往多个平台）
# from agents.tracing import add_trace_processor
# add_trace_processor(LangfuseTraceProcessor())

from agents.tracing import set_tracing_export_api_key
set_tracing_export_api_key("your-openai-api-key")  # 仅用于追踪导出，不影响模型调用

自定义 Processor：完全接管追踪流程

from agents.tracing import TracingProcessor, Trace, Span
from agents import add_trace_processor

class MyCustomProcessor(TracingProcessor):
    """把 Trace 数据写入自己的数据库"""

def on_trace_start(self, trace: Trace) -> None:
        # Trace 开始时触发：初始化记录
        db.insert_trace(trace.trace_id, trace.name, started_at=now())

def on_trace_end(self, trace: Trace) -> None:
        # Trace 结束时触发：更新状态
        db.update_trace(trace.trace_id, ended_at=now())

def on_span_start(self, span: Span) -> None:
        # 每个 Span 开始时触发
        db.insert_span(span.span_id, span.trace_id, span.started_at)

def on_span_end(self, span: Span) -> None:
        # Span 结束时触发：这里能拿到完整的 span.span_data
        # span.span_data 是 AgentSpanData / LLMSpanData / ToolSpanData 等
        db.update_span(span.span_id, span.span_data, ended_at=now())

def shutdown(self) -> None:
        # 进程退出时调用，做清理工作
        db.flush()

def force_flush(self) -> None:
        # 长运行应用中立即导出缓冲区
        db.flush()

# 注册自定义 Processor
add_trace_processor(MyCustomProcessor())

三个实用控制开关

from agents import RunConfig, Runner

result = await Runner.run(
    agent,
    input="...",
    run_config=RunConfig(tracing_disabled=True)
)

from agents.tracing import set_tracing_disabled
set_tracing_disabled(disabled=True)

run_config=RunConfig(trace_include_sensitive_data=False)

run_config=RunConfig(
    workflow_name="customer-service",     # Dashboard 里显示的工作流名
    trace_id="my-custom-id-001",          # 自定义 Trace ID，便于关联外部系统
    group_id="session-user-123",          # 关联同一用户/会话的多次 Run
    trace_metadata={"env": "production", "version": "v2.1"}
)

实际调试场景：快速定位 Agent 异常

Trace: research-agent-run (耗时 45s ⚠️)
├── AgentSpan: ResearchAgent (45s)
│   ├── LLMSpan: gpt-4o (2s) ✓
│   ├── ToolSpan: web_search (40s ⚠️)  ← 就是这里
│   └── LLMSpan: gpt-4o (3s) ✓

4 条可立即落地的实践建议

写在最后