跳转至

Tracing module

TracingProcessor

Bases: ABC

处理 span 的接口。

Source code in src/agents/tracing/processor_interface.py
class TracingProcessor(abc.ABC):
    """处理 span 的接口。"""

    @abc.abstractmethod
    def on_trace_start(self, trace: "Trace") -> None:
        """当 trace 开始时调用。

        参数:
            trace: 已开始的 trace。
        """
        pass

    @abc.abstractmethod
    def on_trace_end(self, trace: "Trace") -> None:
        """当 trace 结束时调用。

        参数:
            trace: 已结束的 trace。
        """
        pass

    @abc.abstractmethod
    def on_span_start(self, span: "Span[Any]") -> None:
        """当 span 开始时调用。

        参数:
            span: 已开始的 span。
        """
        pass

    @abc.abstractmethod
    def on_span_end(self, span: "Span[Any]") -> None:
        """当 span 结束时调用。此方法不应阻塞或抛出异常。

        参数:
            span: 已结束的 span。
        """
        pass

    @abc.abstractmethod
    def shutdown(self) -> None:
        """当应用程序停止时调用。"""
        pass

    @abc.abstractmethod
    def force_flush(self) -> None:
        """强制立即刷新所有排队的 span/trace。"""
        pass

on_trace_start abstractmethod

on_trace_start(trace: Trace) -> None

当 trace 开始时调用。

参数

trace: 已开始的 trace。

Source code in src/agents/tracing/processor_interface.py
@abc.abstractmethod
def on_trace_start(self, trace: "Trace") -> None:
    """当 trace 开始时调用。

    参数:
        trace: 已开始的 trace。
    """
    pass

on_trace_end abstractmethod

on_trace_end(trace: Trace) -> None

当 trace 结束时调用。

参数

trace: 已结束的 trace。

Source code in src/agents/tracing/processor_interface.py
@abc.abstractmethod
def on_trace_end(self, trace: "Trace") -> None:
    """当 trace 结束时调用。

    参数:
        trace: 已结束的 trace。
    """
    pass

on_span_start abstractmethod

on_span_start(span: Span[Any]) -> None

当 span 开始时调用。

参数

span: 已开始的 span。

Source code in src/agents/tracing/processor_interface.py
@abc.abstractmethod
def on_span_start(self, span: "Span[Any]") -> None:
    """当 span 开始时调用。

    参数:
        span: 已开始的 span。
    """
    pass

on_span_end abstractmethod

on_span_end(span: Span[Any]) -> None

当 span 结束时调用。此方法不应阻塞或抛出异常。

参数

span: 已结束的 span。

Source code in src/agents/tracing/processor_interface.py
@abc.abstractmethod
def on_span_end(self, span: "Span[Any]") -> None:
    """当 span 结束时调用。此方法不应阻塞或抛出异常。

    参数:
        span: 已结束的 span。
    """
    pass

shutdown abstractmethod

shutdown() -> None

当应用程序停止时调用。

Source code in src/agents/tracing/processor_interface.py
@abc.abstractmethod
def shutdown(self) -> None:
    """当应用程序停止时调用。"""
    pass

force_flush abstractmethod

force_flush() -> None

强制立即刷新所有排队的 span/trace。

Source code in src/agents/tracing/processor_interface.py
@abc.abstractmethod
def force_flush(self) -> None:
    """强制立即刷新所有排队的 span/trace。"""
    pass

AgentSpanData

Bases: SpanData

表示 trace 中的 Agent Span。 包含 name、handoffs、tools 和 output_type。

Source code in src/agents/tracing/span_data.py
class AgentSpanData(SpanData):
    """
    表示 trace 中的 Agent Span。
    包含 name、handoffs、tools 和 output_type。
    """

    __slots__ = ("name", "handoffs", "tools", "output_type")

    def __init__(
        self,
        name: str,
        handoffs: list[str] | None = None,
        tools: list[str] | None = None,
        output_type: str | None = None,
    ):
        self.name = name
        self.handoffs: list[str] | None = handoffs
        self.tools: list[str] | None = tools
        self.output_type: str | None = output_type

    @property
    def type(self) -> str:
        return "agent"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "name": self.name,
            "handoffs": self.handoffs,
            "tools": self.tools,
            "output_type": self.output_type,
        }

CustomSpanData

Bases: SpanData

表示 trace 中的 Custom Span。 包含 name 和 data 属性包。

Source code in src/agents/tracing/span_data.py
class CustomSpanData(SpanData):
    """
    表示 trace 中的 Custom Span。
    包含 name 和 data 属性包。
    """

    __slots__ = ("name", "data")

    def __init__(self, name: str, data: dict[str, Any]):
        self.name = name
        self.data = data

    @property
    def type(self) -> str:
        return "custom"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "name": self.name,
            "data": self.data,
        }

FunctionSpanData

Bases: SpanData

表示 trace 中的 Function Span。 包含 input、output 以及 MCP 数据(如适用)。

Source code in src/agents/tracing/span_data.py
class FunctionSpanData(SpanData):
    """
    表示 trace 中的 Function Span。
    包含 input、output 以及 MCP 数据(如适用)。
    """

    __slots__ = ("name", "input", "output", "mcp_data")

    def __init__(
        self,
        name: str,
        input: str | None,
        output: Any | None,
        mcp_data: dict[str, Any] | None = None,
    ):
        self.name = name
        self.input = input
        self.output = output
        self.mcp_data = mcp_data

    @property
    def type(self) -> str:
        return "function"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "name": self.name,
            "input": self.input,
            "output": str(self.output) if self.output else None,
            "mcp_data": self.mcp_data,
        }

GenerationSpanData

Bases: SpanData

表示 trace 中的 Generation Span。 包含 input、output、model、model_config 和 usage。

Source code in src/agents/tracing/span_data.py
class GenerationSpanData(SpanData):
    """
    表示 trace 中的 Generation Span。
    包含 input、output、model、model_config 和 usage。
    """

    __slots__ = (
        "input",
        "output",
        "model",
        "model_config",
        "usage",
    )

    def __init__(
        self,
        input: Sequence[Mapping[str, Any]] | None = None,
        output: Sequence[Mapping[str, Any]] | None = None,
        model: str | None = None,
        model_config: Mapping[str, Any] | None = None,
        usage: dict[str, Any] | None = None,
    ):
        self.input = input
        self.output = output
        self.model = model
        self.model_config = model_config
        self.usage = usage

    @property
    def type(self) -> str:
        return "generation"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "input": self.input,
            "output": self.output,
            "model": self.model,
            "model_config": self.model_config,
            "usage": self.usage,
        }

GuardrailSpanData

Bases: SpanData

表示 trace 中的 Guardrail Span。 包含 name 和 triggered 状态。

Source code in src/agents/tracing/span_data.py
class GuardrailSpanData(SpanData):
    """
    表示 trace 中的 Guardrail Span。
    包含 name 和 triggered 状态。
    """

    __slots__ = ("name", "triggered")

    def __init__(self, name: str, triggered: bool = False):
        self.name = name
        self.triggered = triggered

    @property
    def type(self) -> str:
        return "guardrail"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "name": self.name,
            "triggered": self.triggered,
        }

HandoffSpanData

Bases: SpanData

表示 trace 中的 Handoff Span。 包含源 agent 和目标 agent。

Source code in src/agents/tracing/span_data.py
class HandoffSpanData(SpanData):
    """
    表示 trace 中的 Handoff Span。
    包含源 agent 和目标 agent。
    """

    __slots__ = ("from_agent", "to_agent")

    def __init__(self, from_agent: str | None, to_agent: str | None):
        self.from_agent = from_agent
        self.to_agent = to_agent

    @property
    def type(self) -> str:
        return "handoff"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "from_agent": self.from_agent,
            "to_agent": self.to_agent,
        }

MCPListToolsSpanData

Bases: SpanData

表示 trace 中的 MCP List Tools Span。 包含 server 和 result。

Source code in src/agents/tracing/span_data.py
class MCPListToolsSpanData(SpanData):
    """
    表示 trace 中的 MCP List Tools Span。
    包含 server 和 result。
    """

    __slots__ = (
        "server",
        "result",
    )

    def __init__(self, server: str | None = None, result: list[str] | None = None):
        self.server = server
        self.result = result

    @property
    def type(self) -> str:
        return "mcp_tools"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "server": self.server,
            "result": self.result,
        }

ResponseSpanData

Bases: SpanData

表示 trace 中的 Response Span。 包含 response 和 input。

Source code in src/agents/tracing/span_data.py
class ResponseSpanData(SpanData):
    """
    表示 trace 中的 Response Span。
    包含 response 和 input。
    """

    __slots__ = ("response", "input")

    def __init__(
        self,
        response: Response | None = None,
        input: str | list[ResponseInputItemParam] | None = None,
    ) -> None:
        self.response = response
        # 这不会被 OpenAI 的 trace 处理器使用,但对其他追踪处理器实现很有用
        self.input = input

    @property
    def type(self) -> str:
        return "response"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "response_id": self.response.id if self.response else None,
        }

SpanData

Bases: ABC

表示 trace 中的 span 数据。

Source code in src/agents/tracing/span_data.py
class SpanData(abc.ABC):
    """
    表示 trace 中的 span 数据。
    """

    @abc.abstractmethod
    def export(self) -> dict[str, Any]:
        """以字典形式导出 span 数据。"""
        pass

    @property
    @abc.abstractmethod
    def type(self) -> str:
        """返回 span 的类型。"""
        pass

type abstractmethod property

type: str

返回 span 的类型。

export abstractmethod

export() -> dict[str, Any]

以字典形式导出 span 数据。

Source code in src/agents/tracing/span_data.py
@abc.abstractmethod
def export(self) -> dict[str, Any]:
    """以字典形式导出 span 数据。"""
    pass

SpeechGroupSpanData

Bases: SpanData

表示 trace 中的 Speech Group Span。

Source code in src/agents/tracing/span_data.py
class SpeechGroupSpanData(SpanData):
    """
    表示 trace 中的 Speech Group Span。
    """

    __slots__ = "input"

    def __init__(
        self,
        input: str | None = None,
    ):
        self.input = input

    @property
    def type(self) -> str:
        return "speech_group"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "input": self.input,
        }

SpeechSpanData

Bases: SpanData

表示 trace 中的 Speech Span。 包含 input、output、model、model_config 和 first_content_at。

Source code in src/agents/tracing/span_data.py
class SpeechSpanData(SpanData):
    """
    表示 trace 中的 Speech Span。
    包含 input、output、model、model_config 和 first_content_at。
    """

    __slots__ = ("input", "output", "model", "model_config", "first_content_at")

    def __init__(
        self,
        input: str | None = None,
        output: str | None = None,
        output_format: str | None = "pcm",
        model: str | None = None,
        model_config: Mapping[str, Any] | None = None,
        first_content_at: str | None = None,
    ):
        self.input = input
        self.output = output
        self.output_format = output_format
        self.model = model
        self.model_config = model_config
        self.first_content_at = first_content_at

    @property
    def type(self) -> str:
        return "speech"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "input": self.input,
            "output": {
                "data": self.output or "",
                "format": self.output_format,
            },
            "model": self.model,
            "model_config": self.model_config,
            "first_content_at": self.first_content_at,
        }

TranscriptionSpanData

Bases: SpanData

表示 trace 中的 Transcription Span。 包含 input、output、model 和 model_config。

Source code in src/agents/tracing/span_data.py
class TranscriptionSpanData(SpanData):
    """
    表示 trace 中的 Transcription Span。
    包含 input、output、model 和 model_config。
    """

    __slots__ = (
        "input",
        "output",
        "model",
        "model_config",
    )

    def __init__(
        self,
        input: str | None = None,
        input_format: str | None = "pcm",
        output: str | None = None,
        model: str | None = None,
        model_config: Mapping[str, Any] | None = None,
    ):
        self.input = input
        self.input_format = input_format
        self.output = output
        self.model = model
        self.model_config = model_config

    @property
    def type(self) -> str:
        return "transcription"

    def export(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "input": {
                "data": self.input or "",
                "format": self.input_format,
            },
            "output": self.output,
            "model": self.model,
            "model_config": self.model_config,
        }

Span

Bases: ABC, Generic[TSpanData]

Source code in src/agents/tracing/spans.py
class Span(abc.ABC, Generic[TSpanData]):
    @property
    @abc.abstractmethod
    def trace_id(self) -> str:
        pass

    @property
    @abc.abstractmethod
    def span_id(self) -> str:
        pass

    @property
    @abc.abstractmethod
    def span_data(self) -> TSpanData:
        pass

    @abc.abstractmethod
    def start(self, mark_as_current: bool = False):
        """
        启动 span。

        参数:
            mark_as_current: 如果为 True,则该 span 会被标记为当前 span。
        """
        pass

    @abc.abstractmethod
    def finish(self, reset_current: bool = False) -> None:
        """
        结束 span。

        参数:
            reset_current: 如果为 True,则会将当前 span 重置为该 span。
        """
        pass

    @abc.abstractmethod
    def __enter__(self) -> Span[TSpanData]:
        pass

    @abc.abstractmethod
    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    @property
    @abc.abstractmethod
    def parent_id(self) -> str | None:
        pass

    @abc.abstractmethod
    def set_error(self, error: SpanError) -> None:
        pass

    @property
    @abc.abstractmethod
    def error(self) -> SpanError | None:
        pass

    @abc.abstractmethod
    def export(self) -> dict[str, Any] | None:
        pass

    @property
    @abc.abstractmethod
    def started_at(self) -> str | None:
        pass

    @property
    @abc.abstractmethod
    def ended_at(self) -> str | None:
        pass

start abstractmethod

start(mark_as_current: bool = False)

启动 span。

参数

mark_as_current: 如果为 True,则该 span 会被标记为当前 span。

Source code in src/agents/tracing/spans.py
@abc.abstractmethod
def start(self, mark_as_current: bool = False):
    """
    启动 span。

    参数:
        mark_as_current: 如果为 True,则该 span 会被标记为当前 span。
    """
    pass

finish abstractmethod

finish(reset_current: bool = False) -> None

结束 span。

参数

reset_current: 如果为 True,则会将当前 span 重置为该 span。

Source code in src/agents/tracing/spans.py
@abc.abstractmethod
def finish(self, reset_current: bool = False) -> None:
    """
    结束 span。

    参数:
        reset_current: 如果为 True,则会将当前 span 重置为该 span。
    """
    pass

Trace

Trace(追踪)是 tracing 创建的最顶层对象。它代表一个逻辑上的“工作流”。

Source code in src/agents/tracing/traces.py
class Trace:
    """
    Trace(追踪)是 tracing 创建的最顶层对象。它代表一个逻辑上的“工作流”。
    """

    @abc.abstractmethod
    def __enter__(self) -> Trace:
        pass

    @abc.abstractmethod
    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    @abc.abstractmethod
    def start(self, mark_as_current: bool = False):
        """
        启动追踪。

        参数:
            mark_as_current: 如果为 True,则将该 trace 标记为当前 trace。
        """
        pass

    @abc.abstractmethod
    def finish(self, reset_current: bool = False):
        """
        结束追踪。

        参数:
            reset_current: 如果为 True,则会将该 trace 重置为当前 trace。
        """
        pass

    @property
    @abc.abstractmethod
    def trace_id(self) -> str:
        """
        追踪 ID。
        """
        pass

    @property
    @abc.abstractmethod
    def name(self) -> str:
        """
        被追踪的工作流名称。
        """
        pass

    @abc.abstractmethod
    def export(self) -> dict[str, Any] | None:
        """
        将追踪导出为字典。
        """
        pass

trace_id abstractmethod property

trace_id: str

追踪 ID。

name abstractmethod property

name: str

被追踪的工作流名称。

start abstractmethod

start(mark_as_current: bool = False)

启动追踪。

参数

mark_as_current: 如果为 True,则将该 trace 标记为当前 trace。

Source code in src/agents/tracing/traces.py
@abc.abstractmethod
def start(self, mark_as_current: bool = False):
    """
    启动追踪。

    参数:
        mark_as_current: 如果为 True,则将该 trace 标记为当前 trace。
    """
    pass

finish abstractmethod

finish(reset_current: bool = False)

结束追踪。

参数

reset_current: 如果为 True,则会将该 trace 重置为当前 trace。

Source code in src/agents/tracing/traces.py
@abc.abstractmethod
def finish(self, reset_current: bool = False):
    """
    结束追踪。

    参数:
        reset_current: 如果为 True,则会将该 trace 重置为当前 trace。
    """
    pass

export abstractmethod

export() -> dict[str, Any] | None

将追踪导出为字典。

Source code in src/agents/tracing/traces.py
@abc.abstractmethod
def export(self) -> dict[str, Any] | None:
    """
    将追踪导出为字典。
    """
    pass

agent_span

agent_span(
    name: str,
    handoffs: list[str] | None = None,
    tools: list[str] | None = None,
    output_type: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[AgentSpanData]

创建一个新的 agent span。该 span 不会自动启动,你需要使用 with agent_span() ...,或者手动调用 span.start()span.finish()

参数

name: agent 的名称。 handoffs: 可选,agent 可以交接控制权的其他 agent 名称列表。 tools: 可选,该 agent 可用的工具名称列表。 output_type: 可选,agent 产生的输出类型名称。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

返回

新创建的 agent span。

Source code in src/agents/tracing/create.py
def agent_span(
    name: str,
    handoffs: list[str] | None = None,
    tools: list[str] | None = None,
    output_type: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[AgentSpanData]:
    """创建一个新的 agent span。该 span 不会自动启动,你需要使用 `with agent_span() ...`,或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        name: agent 的名称。
        handoffs: 可选,agent 可以交接控制权的其他 agent 名称列表。
        tools: 可选,该 agent 可用的工具名称列表。
        output_type: 可选,agent 产生的输出类型名称。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

    返回:
        新创建的 agent span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=AgentSpanData(name=name, handoffs=handoffs, tools=tools, output_type=output_type),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

custom_span

custom_span(
    name: str,
    data: dict[str, Any] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[CustomSpanData]

创建一个新的自定义 span,你可以为其添加自定义元数据。该 span 不会自动启动,你需要使用 with custom_span() ...,或者手动调用 span.start()span.finish()

参数

name: 自定义 span 的名称。 data: 要关联到 span 的任意结构化数据。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

返回

新创建的自定义 span。

Source code in src/agents/tracing/create.py
def custom_span(
    name: str,
    data: dict[str, Any] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[CustomSpanData]:
    """创建一个新的自定义 span,你可以为其添加自定义元数据。该 span 不会自动启动,你需要使用 `with custom_span() ...`,或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        name: 自定义 span 的名称。
        data: 要关联到 span 的任意结构化数据。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

    返回:
        新创建的自定义 span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=CustomSpanData(name=name, data=data or {}),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

function_span

function_span(
    name: str,
    input: str | None = None,
    output: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[FunctionSpanData]

创建一个新的 function span。该 span 不会自动启动,你需要使用 with function_span() ...,或者手动调用 span.start()span.finish()

参数

name: 函数的名称。 input: 传递给函数的输入。 output: 函数的输出。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

返回

新创建的 function span。

Source code in src/agents/tracing/create.py
def function_span(
    name: str,
    input: str | None = None,
    output: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[FunctionSpanData]:
    """创建一个新的 function span。该 span 不会自动启动,你需要使用 `with function_span() ...`,或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        name: 函数的名称。
        input: 传递给函数的输入。
        output: 函数的输出。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

    返回:
        新创建的 function span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=FunctionSpanData(name=name, input=input, output=output),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

generation_span

generation_span(
    input: Sequence[Mapping[str, Any]] | None = None,
    output: Sequence[Mapping[str, Any]] | None = None,
    model: str | None = None,
    model_config: Mapping[str, Any] | None = None,
    usage: dict[str, Any] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[GenerationSpanData]

创建一个新的 generation span。该 span 不会自动启动,你需要使用 with generation_span() ..., 或者手动调用 span.start()span.finish()

该 span 用于捕获模型生成的详细信息,包括输入消息序列、生成的输出、模型名称与配置,以及用量数据。 如果你只需要捕获模型响应标识符,请使用 response_span()

参数

input: 发送给模型的输入消息序列。 output: 从模型接收到的输出消息序列。 model: 用于生成的模型标识符。 model_config: 使用的模型配置(超参数)。 usage: 用量信息字典(如输入 token 数、输出 token 数等)。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

返回

新创建的 generation span。

Source code in src/agents/tracing/create.py
def generation_span(
    input: Sequence[Mapping[str, Any]] | None = None,
    output: Sequence[Mapping[str, Any]] | None = None,
    model: str | None = None,
    model_config: Mapping[str, Any] | None = None,
    usage: dict[str, Any] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[GenerationSpanData]:
    """创建一个新的 generation span。该 span 不会自动启动,你需要使用 `with generation_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    该 span 用于捕获模型生成的详细信息,包括输入消息序列、生成的输出、模型名称与配置,以及用量数据。
    如果你只需要捕获模型响应标识符,请使用 `response_span()`。

    参数:
        input: 发送给模型的输入消息序列。
        output: 从模型接收到的输出消息序列。
        model: 用于生成的模型标识符。
        model_config: 使用的模型配置(超参数)。
        usage: 用量信息字典(如输入 token 数、输出 token 数等)。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

    返回:
        新创建的 generation span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=GenerationSpanData(
            input=input,
            output=output,
            model=model,
            model_config=model_config,
            usage=usage,
        ),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

get_current_span

get_current_span() -> Span[Any] | None

Returns the currently active span, if present.

Source code in src/agents/tracing/create.py
def get_current_span() -> Span[Any] | None:
    """Returns the currently active span, if present."""
    return GLOBAL_TRACE_PROVIDER.get_current_span()

get_current_trace

get_current_trace() -> Trace | None

Returns the currently active trace, if present.

Source code in src/agents/tracing/create.py
def get_current_trace() -> Trace | None:
    """Returns the currently active trace, if present."""
    return GLOBAL_TRACE_PROVIDER.get_current_trace()

guardrail_span

guardrail_span(
    name: str,
    triggered: bool = False,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[GuardrailSpanData]

创建一个新的 guardrail span。该 span 不会自动启动,你需要使用 with guardrail_span() ..., 或者手动调用 span.start()span.finish()

参数

name: guardrail 的名称。 triggered: guardrail 是否被触发。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

Source code in src/agents/tracing/create.py
def guardrail_span(
    name: str,
    triggered: bool = False,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[GuardrailSpanData]:
    """创建一个新的 guardrail span。该 span 不会自动启动,你需要使用 `with guardrail_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        name: guardrail 的名称。
        triggered: guardrail 是否被触发。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=GuardrailSpanData(name=name, triggered=triggered),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

handoff_span

handoff_span(
    from_agent: str | None = None,
    to_agent: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[HandoffSpanData]

创建一个新的 handoff span。该 span 不会自动启动,你需要使用 with handoff_span() ..., 或者手动调用 span.start()span.finish()

参数

from_agent: 发起交接的 agent 名称。 to_agent: 接收交接的 agent 名称。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

返回

新创建的 handoff span。

Source code in src/agents/tracing/create.py
def handoff_span(
    from_agent: str | None = None,
    to_agent: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[HandoffSpanData]:
    """创建一个新的 handoff span。该 span 不会自动启动,你需要使用 `with handoff_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        from_agent: 发起交接的 agent 名称。
        to_agent: 接收交接的 agent 名称。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

    返回:
        新创建的 handoff span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=HandoffSpanData(from_agent=from_agent, to_agent=to_agent),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

mcp_tools_span

mcp_tools_span(
    server: str | None = None,
    result: list[str] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[MCPListToolsSpanData]

创建一个新的 MCP list tools span。该 span 不会自动启动,你需要使用 with mcp_tools_span() ..., 或者手动调用 span.start()span.finish()

参数

server: MCP 服务器的名称。 result: MCP list tools 调用的结果。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

Source code in src/agents/tracing/create.py
def mcp_tools_span(
    server: str | None = None,
    result: list[str] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[MCPListToolsSpanData]:
    """创建一个新的 MCP list tools span。该 span 不会自动启动,你需要使用 `with mcp_tools_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        server: MCP 服务器的名称。
        result: MCP list tools 调用的结果。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=MCPListToolsSpanData(server=server, result=result),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

response_span

response_span(
    response: Response | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[ResponseSpanData]

创建一个新的 response span。该 span 不会自动启动,你需要使用 with response_span() ..., 或者手动调用 span.start()span.finish()

参数

response: OpenAI 的 Response 对象。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

Source code in src/agents/tracing/create.py
def response_span(
    response: Response | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[ResponseSpanData]:
    """创建一个新的 response span。该 span 不会自动启动,你需要使用 `with response_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        response: OpenAI 的 Response 对象。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=ResponseSpanData(response=response),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

speech_group_span

speech_group_span(
    input: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[SpeechGroupSpanData]

创建一个新的 speech group span。该 span 不会自动启动,你需要使用 with speech_group_span() ..., 或者手动调用 span.start()span.finish()

参数

input: 用于语音请求的输入文本。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

Source code in src/agents/tracing/create.py
def speech_group_span(
    input: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[SpeechGroupSpanData]:
    """创建一个新的 speech group span。该 span 不会自动启动,你需要使用 `with speech_group_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        input: 用于语音请求的输入文本。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=SpeechGroupSpanData(input=input),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

speech_span

speech_span(
    model: str | None = None,
    input: str | None = None,
    output: str | None = None,
    output_format: str | None = "pcm",
    model_config: Mapping[str, Any] | None = None,
    first_content_at: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[SpeechSpanData]

创建一个新的 speech span。该 span 不会自动启动,你需要使用 with speech_span() ...,或者手动调用 span.start()span.finish()

参数

model: 用于文本转语音的模型名称。 input: 文本转语音的输入文本。 output: 文本转语音的音频输出,作为 base64 编码的 PCM 音频字节字符串。 output_format: 音频输出的格式(默认为 "pcm")。 model_config: 使用的模型配置(超参数)。 first_content_at: 音频输出第一个字节的时间。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

Source code in src/agents/tracing/create.py
def speech_span(
    model: str | None = None,
    input: str | None = None,
    output: str | None = None,
    output_format: str | None = "pcm",
    model_config: Mapping[str, Any] | None = None,
    first_content_at: str | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[SpeechSpanData]:
    """创建一个新的 speech span。该 span 不会自动启动,你需要使用 `with speech_span() ...`,或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        model: 用于文本转语音的模型名称。
        input: 文本转语音的输入文本。
        output: 文本转语音的音频输出,作为 base64 编码的 PCM 音频字节字符串。
        output_format: 音频输出的格式(默认为 "pcm")。
        model_config: 使用的模型配置(超参数)。
        first_content_at: 音频输出第一个字节的时间。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=SpeechSpanData(
            model=model,
            input=input,
            output=output,
            output_format=output_format,
            model_config=model_config,
            first_content_at=first_content_at,
        ),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

trace

trace(
    workflow_name: str,
    trace_id: str | None = None,
    group_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    disabled: bool = False,
) -> Trace

创建一个新的 trace(追踪)。该 trace 不会自动启动;你需要将其作为上下文管理器使用(with trace(...):),或者手动调用 trace.start()trace.finish()

除了工作流名称和可选的分组标识符外,你还可以提供一个任意的元数据字典,将用户自定义的信息附加到 trace 上。

参数

workflow_name: 逻辑应用或工作流的名称。例如,你可以为编码代理提供 "code_bot",或为客服代理提供 "customer_support_agent"。 trace_id: trace 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_trace_id() 生成 trace ID,以确保格式正确。 group_id: 可选的分组标识符,用于将同一会话或流程中的多个 trace 关联起来。例如,你可以使用聊天线程 ID。 metadata: 可选的附加元数据字典,将其附加到 trace 上。 disabled: 如果为 True,我们会返回一个 Trace,但不会记录该 Trace。如果已存在 trace 且 even_if_trace_running 为 True,则不会检查此项。

返回

新创建的 trace 对象。

Source code in src/agents/tracing/create.py
def trace(
    workflow_name: str,
    trace_id: str | None = None,
    group_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    disabled: bool = False,
) -> Trace:
    """
    创建一个新的 trace(追踪)。该 trace 不会自动启动;你需要将其作为上下文管理器使用(`with trace(...):`),或者手动调用 `trace.start()` 和 `trace.finish()`。

    除了工作流名称和可选的分组标识符外,你还可以提供一个任意的元数据字典,将用户自定义的信息附加到 trace 上。

    参数:
        workflow_name: 逻辑应用或工作流的名称。例如,你可以为编码代理提供 "code_bot",或为客服代理提供 "customer_support_agent"。
        trace_id: trace 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_trace_id()` 生成 trace ID,以确保格式正确。
        group_id: 可选的分组标识符,用于将同一会话或流程中的多个 trace 关联起来。例如,你可以使用聊天线程 ID。
        metadata: 可选的附加元数据字典,将其附加到 trace 上。
        disabled: 如果为 True,我们会返回一个 Trace,但不会记录该 Trace。如果已存在 trace 且 `even_if_trace_running` 为 True,则不会检查此项。

    返回:
        新创建的 trace 对象。
    """
    current_trace = GLOBAL_TRACE_PROVIDER.get_current_trace()
    if current_trace:
        logger.warning(
            "Trace already exists. Creating a new trace, but this is probably a mistake."
        )

    return GLOBAL_TRACE_PROVIDER.create_trace(
        name=workflow_name,
        trace_id=trace_id,
        group_id=group_id,
        metadata=metadata,
        disabled=disabled,
    )

transcription_span

transcription_span(
    model: str | None = None,
    input: str | None = None,
    input_format: str | None = "pcm",
    output: str | None = None,
    model_config: Mapping[str, Any] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[TranscriptionSpanData]

创建一个新的转录(transcription)span。该 span 不会自动启动,你需要使用 with transcription_span() ..., 或者手动调用 span.start()span.finish()

参数

model: 用于语音转文本的模型名称。 input: 语音转文本的音频输入,作为 base64 编码的音频字节字符串。 input_format: 音频输入的格式(默认为 "pcm")。 output: 语音转文本的输出文本。 model_config: 使用的模型配置(超参数)。 span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 util.gen_span_id() 生成 span ID,以确保格式正确。 parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。 disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

返回

新创建的语音转文本 span。

Source code in src/agents/tracing/create.py
def transcription_span(
    model: str | None = None,
    input: str | None = None,
    input_format: str | None = "pcm",
    output: str | None = None,
    model_config: Mapping[str, Any] | None = None,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[TranscriptionSpanData]:
    """创建一个新的转录(transcription)span。该 span 不会自动启动,你需要使用 `with transcription_span() ...`,
    或者手动调用 `span.start()` 和 `span.finish()`。

    参数:
        model: 用于语音转文本的模型名称。
        input: 语音转文本的音频输入,作为 base64 编码的音频字节字符串。
        input_format: 音频输入的格式(默认为 "pcm")。
        output: 语音转文本的输出文本。
        model_config: 使用的模型配置(超参数)。
        span_id: span 的 ID。可选。如果未提供,我们会自动生成一个 ID。建议使用 `util.gen_span_id()` 生成 span ID,以确保格式正确。
        parent: 父级 span 或 trace。如果未提供,我们会自动使用当前的 trace/span 作为父级。
        disabled: 如果为 True,我们会返回一个 Span,但不会记录该 Span。

    返回:
        新创建的语音转文本 span。
    """
    return GLOBAL_TRACE_PROVIDER.create_span(
        span_data=TranscriptionSpanData(
            input=input,
            input_format=input_format,
            output=output,
            model=model,
            model_config=model_config,
        ),
        span_id=span_id,
        parent=parent,
        disabled=disabled,
    )

gen_span_id

gen_span_id() -> str

生成一个新的 span ID。

Source code in src/agents/tracing/util.py
def gen_span_id() -> str:
    """生成一个新的 span ID。"""
    return f"span_{uuid.uuid4().hex[:24]}"

gen_trace_id

gen_trace_id() -> str

生成一个新的 trace ID。

Source code in src/agents/tracing/util.py
def gen_trace_id() -> str:
    """生成一个新的 trace ID。"""
    return f"trace_{uuid.uuid4().hex}"

add_trace_processor

add_trace_processor(
    span_processor: TracingProcessor,
) -> None

Adds a new trace processor. This processor will receive all traces/spans.

Source code in src/agents/tracing/__init__.py
def add_trace_processor(span_processor: TracingProcessor) -> None:
    """
    Adds a new trace processor. This processor will receive all traces/spans.
    """
    GLOBAL_TRACE_PROVIDER.register_processor(span_processor)

set_trace_processors

set_trace_processors(
    processors: list[TracingProcessor],
) -> None

Set the list of trace processors. This will replace the current list of processors.

Source code in src/agents/tracing/__init__.py
def set_trace_processors(processors: list[TracingProcessor]) -> None:
    """
    Set the list of trace processors. This will replace the current list of processors.
    """
    GLOBAL_TRACE_PROVIDER.set_processors(processors)

set_tracing_disabled

set_tracing_disabled(disabled: bool) -> None

Set whether tracing is globally disabled.

Source code in src/agents/tracing/__init__.py
def set_tracing_disabled(disabled: bool) -> None:
    """
    Set whether tracing is globally disabled.
    """
    GLOBAL_TRACE_PROVIDER.set_disabled(disabled)

set_tracing_export_api_key

set_tracing_export_api_key(api_key: str) -> None

Set the OpenAI API key for the backend exporter.

Source code in src/agents/tracing/__init__.py
def set_tracing_export_api_key(api_key: str) -> None:
    """
    Set the OpenAI API key for the backend exporter.
    """
    default_exporter().set_api_key(api_key)