OpenAI Agents SDK #34：自定义 TracingProcessor——把 Agent 链路数据接进你自己的可观测性系统

默认链路的样子

Agent Run
   ↓
TraceProvider（全局单例）
   ↓
BatchTraceProcessor（后台线程 + 内存队列）
   ↓
BackendSpanExporter（HTTP → api.openai.com/v1/traces/ingest）

两个定制入口

自己实现一个 TracingExporter

from agents.tracing.processor_interface import TracingExporter
from agents import Trace
from agents.tracing.spans import Span
from typing import Any

class MyExporter(TracingExporter):
    def export(self, items: list[Trace | Span[Any]]) -> None:
        for item in items:
            exported = item.export()   # 返回 dict 或 None
            if exported:
                # 发往你的系统：写数据库、调 HTTP API……
                send_to_my_backend(exported)

{
    "object": "trace.span",
    "id": "span_xxxx",
    "trace_id": "trace_xxxx",
    "parent_id": "span_yyyy",   # None 表示根 span
    "started_at": "2025-01-01T00:00:00.000Z",
    "ended_at":   "2025-01-01T00:00:01.234Z",
    "span_data":  { "type": "agent", ... },   # 类型特定字段
    "error":      None,                        # 或 SpanError dict
    "metadata":   { ... }                      # 可选
}

用 BatchTraceProcessor 封装自定义 Exporter

from agents import add_trace_processor
from agents.tracing import BatchTraceProcessor

my_processor = BatchTraceProcessor(
    exporter=MyExporter(),
    max_queue_size=4096,    # 队列容量，超出则丢弃
    max_batch_size=64,      # 单批最多导出数量
    schedule_delay=3.0,     # 后台定时导出间隔（秒）
    export_trigger_ratio=0.6,  # 队列达到 60% 时触发立即导出
)

add_trace_processor(my_processor)

flush_traces：长驻进程的关键

from agents import Runner, flush_traces, trace

# Celery 任务
@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 立即上报

# FastAPI 后台任务
from fastapi import BackgroundTasks, FastAPI

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"}

custom_span：业务埋点

from agents import custom_span

async def fetch_user_data(user_id: str):
    with custom_span("db_query", data={"table": "users", "user_id": user_id}) as span:
        result = await db.get_user(user_id)
        span.span_data.data["row_count"] = len(result)
        return result

with custom_span("sub_task", parent=current_span) as span:
    ...

同时对接多个系统

from agents import add_trace_processor, set_tracing_export_api_key
from agents.tracing import BatchTraceProcessor, BackendSpanExporter

# 场景：既要发 OpenAI Traces，又要同步给内部 Langfuse

# 1. 保留默认的 OpenAI 后台（add_trace_processor 不覆盖默认）
# 2. 额外挂 Langfuse
from langfuse.openai import LangfuseTracingExporter  # 社区实现示例

add_trace_processor(
    BatchTraceProcessor(exporter=LangfuseTracingExporter())
)

from agents import set_trace_processors
from agents.tracing import BatchTraceProcessor

set_trace_processors([
    BatchTraceProcessor(exporter=MyPrimaryExporter()),
    BatchTraceProcessor(exporter=MySecondaryExporter()),
])

非 OpenAI 模型的 tracing key

import os
from agents import set_tracing_export_api_key, Agent, Runner
from agents.extensions.models.any_llm_model import AnyLLMModel

# 单独设置 tracing 专用 key
set_tracing_export_api_key(os.environ["OPENAI_API_KEY"])

model = AnyLLMModel(
    model="anthropic/claude-3-5-sonnet",
    api_key=os.environ["ANTHROPIC_API_KEY"],
)
agent = Agent(name="Claude Agent", model=model)

from agents import Runner, RunConfig

await Runner.run(
    agent,
    input="Hello",
    run_config=RunConfig(tracing={"api_key": "sk-tracing-only-key"}),
)

生产级接入清单

1. 长驻进程必须调 flush_traces()：在每个任务的 finally 块里，不要依赖定时批量导出的时间窗口。
2. add_trace_processor vs set_trace_processors 区分清楚：只想旁路抄送就用 add；需要替换（ZDR、完全私有部署）才用 set。
3. 敏感数据控制：RunConfig(trace_include_sensitive_data=False) 可以禁止 LLM 输入输出进入 trace，避免把用户 PII 带进可观测性系统。
4. 队列容量 vs 导出延迟：高并发场景适当调大 max_queue_size，同时调小 schedule_delay 避免内存积压；突发流量时注意队列满会丢 span（只有 warning log，不抛异常）。
5. ecosystem 直接用现成集成：Datadog、Langfuse、Braintrust、Pydantic Logfire 等 20+ 个社区集成已经实现好了 TracingProcessor，直接 add_trace_processor 就能用，不必自己实现 Exporter。1

与 #7 基础篇的定位差异