追踪

Trace

基础: ABC

一个完整的端到端工作流，包含相关的跨度（spans）和元数据。

追踪代表一个逻辑工作流或操作（例如，“客户服务查询”或“代码生成”），并包含在该工作流期间发生的所有跨度（单个操作）。

示例

# Basic trace usage
with trace("Order Processing") as t:
    validation_result = await Runner.run(validator, order_data)
    if validation_result.approved:
        await Runner.run(processor, order_data)

# Trace with metadata and grouping
with trace(
    "Customer Service",
    group_id="chat_123",
    metadata={"customer": "user_456"}
) as t:
    result = await Runner.run(support_agent, query)

注意事项

• 使用描述性的工作流名称
• 使用一致的 group_ids 对相关的追踪进行分组
• 添加相关的元数据用于过滤/分析
• 使用上下文管理器进行可靠的清理
• 在添加追踪数据时考虑隐私

源代码位于 src/agents/tracing/traces.py

class Trace(abc.ABC):
    """A complete end-to-end workflow containing related spans and metadata.

    A trace represents a logical workflow or operation (e.g., "Customer Service Query"
    or "Code Generation") and contains all the spans (individual operations) that occur
    during that workflow.

    Example:
        ```python
        # Basic trace usage
        with trace("Order Processing") as t:
            validation_result = await Runner.run(validator, order_data)
            if validation_result.approved:
                await Runner.run(processor, order_data)

        # Trace with metadata and grouping
        with trace(
            "Customer Service",
            group_id="chat_123",
            metadata={"customer": "user_456"}
        ) as t:
            result = await Runner.run(support_agent, query)
        ```

    Notes:
        - Use descriptive workflow names
        - Group related traces with consistent group_ids
        - Add relevant metadata for filtering/analysis
        - Use context managers for reliable cleanup
        - Consider privacy when adding trace data
    """

    @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):
        """Start the trace and optionally mark it as the current trace.

        Args:
            mark_as_current: If true, marks this trace as the current trace
                in the execution context.

        Notes:
            - Must be called before any spans can be added
            - Only one trace can be current at a time
            - Thread-safe when using mark_as_current
        """
        pass

    @abc.abstractmethod
    def finish(self, reset_current: bool = False):
        """Finish the trace and optionally reset the current trace.

        Args:
            reset_current: If true, resets the current trace to the previous
                trace in the execution context.

        Notes:
            - Must be called to complete the trace
            - Finalizes all open spans
            - Thread-safe when using reset_current
        """
        pass

    @property
    @abc.abstractmethod
    def trace_id(self) -> str:
        """Get the unique identifier for this trace.

        Returns:
            str: The trace's unique ID in the format 'trace_<32_alphanumeric>'

        Notes:
            - IDs are globally unique
            - Used to link spans to their parent trace
            - Can be used to look up traces in the dashboard
        """
        pass

    @property
    @abc.abstractmethod
    def name(self) -> str:
        """Get the human-readable name of this workflow trace.

        Returns:
            str: The workflow name (e.g., "Customer Service", "Data Processing")

        Notes:
            - Should be descriptive and meaningful
            - Used for grouping and filtering in the dashboard
            - Helps identify the purpose of the trace
        """
        pass

    @abc.abstractmethod
    def export(self) -> dict[str, Any] | None:
        """Export the trace data as a serializable dictionary.

        Returns:
            dict | None: Dictionary containing trace data, or None if tracing is disabled.

        Notes:
            - Includes all spans and their data
            - Used for sending traces to backends
            - May include metadata and group ID
        """
        pass

    @property
    @abc.abstractmethod
    def tracing_api_key(self) -> str | None:
        """The API key to use when exporting this trace and its spans."""
        pass

trace_id abstractmethod property

trace_id: str

获取此追踪的唯一标识符。

返回值

名称	类型	描述
str	str	格式为 'trace_<32_alphanumeric>' 的追踪的唯一 ID

注意事项

• ID 在全局范围内是唯一的
• 用于将跨度链接到其父追踪
• 可用于在仪表板中查找追踪

name abstractmethod property

name: str

获取此工作流追踪的可读名称。

返回值

名称	类型	描述
str	str	工作流名称（例如，“客户服务”、“数据处理”）

注意事项

• 应具有描述性和意义
• 用于在仪表板中进行分组和过滤
• 有助于识别追踪的目的

tracing_api_key abstractmethod property

tracing_api_key: str | None

导出此追踪及其跨度时使用的 API 密钥。

start abstractmethod

start(mark_as_current: bool = False)

启动追踪并可选地将其标记为当前追踪。

参数

名称	类型	描述	默认
mark_as_current	bool	如果为 true，则将此追踪标记为执行上下文中的当前追踪。	False

注意事项

• 必须在添加任何跨度之前调用
• 一次只能有一个追踪为当前追踪
• 使用 mark_as_current 时线程安全

源代码位于 src/agents/tracing/traces.py

@abc.abstractmethod
def start(self, mark_as_current: bool = False):
    """Start the trace and optionally mark it as the current trace.

    Args:
        mark_as_current: If true, marks this trace as the current trace
            in the execution context.

    Notes:
        - Must be called before any spans can be added
        - Only one trace can be current at a time
        - Thread-safe when using mark_as_current
    """
    pass

finish abstractmethod

finish(reset_current: bool = False)

完成追踪并可选地重置当前追踪。

参数

名称	类型	描述	默认
reset_current	bool	如果为 true，则将当前追踪重置为执行上下文中之前的追踪。	False

注意事项

• 必须调用以完成追踪
• 完成所有打开的跨度
• 使用 reset_current 时线程安全

源代码位于 src/agents/tracing/traces.py

@abc.abstractmethod
def finish(self, reset_current: bool = False):
    """Finish the trace and optionally reset the current trace.

    Args:
        reset_current: If true, resets the current trace to the previous
            trace in the execution context.

    Notes:
        - Must be called to complete the trace
        - Finalizes all open spans
        - Thread-safe when using reset_current
    """
    pass

export abstractmethod

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

将追踪数据导出为可序列化的字典。

返回值

类型	描述
dict[str, Any] | None	dict | None: 包含追踪数据的字典，或者如果禁用了追踪则为 None。

注意事项

• 包括所有跨度及其数据
• 用于将追踪发送到后端
• 可能包括元数据和 group ID

源代码位于 src/agents/tracing/traces.py

@abc.abstractmethod
def export(self) -> dict[str, Any] | None:
    """Export the trace data as a serializable dictionary.

    Returns:
        dict | None: Dictionary containing trace data, or None if tracing is disabled.

    Notes:
        - Includes all spans and their data
        - Used for sending traces to backends
        - May include metadata and group ID
    """
    pass

NoOpTrace

基类：Trace

一个 Trace 的无操作实现，不记录任何数据。

当禁用追踪但仍然需要追踪操作时使用。维护适当的上下文管理，但不存储或导出任何数据。

示例

# When tracing is disabled, traces become NoOpTrace
with trace("Disabled Workflow") as t:
    # Operations still work but nothing is recorded
    await Runner.run(agent, "query")

源代码位于 src/agents/tracing/traces.py

class NoOpTrace(Trace):
    """A no-op implementation of Trace that doesn't record any data.

    Used when tracing is disabled but trace operations still need to work.
    Maintains proper context management but doesn't store or export any data.

    Example:
        ```python
        # When tracing is disabled, traces become NoOpTrace
        with trace("Disabled Workflow") as t:
            # Operations still work but nothing is recorded
            await Runner.run(agent, "query")
        ```
    """

    def __init__(self):
        self._started = False
        self._prev_context_token: contextvars.Token[Trace | None] | None = None

    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._started = True
        self.start(mark_as_current=True)

        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.finish(reset_current=True)

    def start(self, mark_as_current: bool = False):
        if mark_as_current:
            self._prev_context_token = Scope.set_current_trace(self)

    def finish(self, reset_current: bool = False):
        if reset_current and self._prev_context_token is not None:
            Scope.reset_current_trace(self._prev_context_token)
            self._prev_context_token = None

    @property
    def trace_id(self) -> str:
        """The trace's unique identifier.

        Returns:
            str: A unique ID for this trace.
        """
        return "no-op"

    @property
    def name(self) -> str:
        """The workflow name for this trace.

        Returns:
            str: Human-readable name describing this workflow.
        """
        return "no-op"

    def export(self) -> dict[str, Any] | None:
        """Export the trace data as a dictionary.

        Returns:
            dict | None: Trace data in exportable format, or None if no data.
        """
        return None

    @property
    def tracing_api_key(self) -> str | None:
        return None

trace_id property

trace_id: str

追踪的唯一标识符。

返回值

名称	类型	描述
str	str	此追踪的唯一 ID。

name property

name: str

此追踪的工作流名称。

返回值

名称	类型	描述
str	str	描述此工作流的可读名称。

export

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

将追踪数据导出为字典。

返回值

类型	描述
dict[str, Any] | None	dict | None: 可导出格式的追踪数据，如果没有数据则为 None。

源代码位于 src/agents/tracing/traces.py

def export(self) -> dict[str, Any] | None:
    """Export the trace data as a dictionary.

    Returns:
        dict | None: Trace data in exportable format, or None if no data.
    """
    return None

TraceImpl

基类：Trace

将被追踪库记录的追踪。

源代码位于 src/agents/tracing/traces.py

class TraceImpl(Trace):
    """
    A trace that will be recorded by the tracing library.
    """

    __slots__ = (
        "_name",
        "_trace_id",
        "_tracing_api_key",
        "group_id",
        "metadata",
        "_prev_context_token",
        "_processor",
        "_started",
    )

    def __init__(
        self,
        name: str,
        trace_id: str | None,
        group_id: str | None,
        metadata: dict[str, Any] | None,
        processor: TracingProcessor,
        tracing_api_key: str | None,
    ):
        self._name = name
        self._trace_id = trace_id or util.gen_trace_id()
        self._tracing_api_key = tracing_api_key
        self.group_id = group_id
        self.metadata = metadata
        self._prev_context_token: contextvars.Token[Trace | None] | None = None
        self._processor = processor
        self._started = False

    @property
    def trace_id(self) -> str:
        return self._trace_id

    @property
    def name(self) -> str:
        return self._name

    @property
    def tracing_api_key(self) -> str | None:
        return self._tracing_api_key

    def start(self, mark_as_current: bool = False):
        if self._started:
            return

        self._started = True
        self._processor.on_trace_start(self)

        if mark_as_current:
            self._prev_context_token = Scope.set_current_trace(self)

    def finish(self, reset_current: bool = False):
        if not self._started:
            return

        self._processor.on_trace_end(self)

        if reset_current and self._prev_context_token is not None:
            Scope.reset_current_trace(self._prev_context_token)
            self._prev_context_token = None

    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_current=True)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.finish(reset_current=exc_type is not GeneratorExit)

    def export(self) -> dict[str, Any] | None:
        return {
            "object": "trace",
            "id": self.trace_id,
            "workflow_name": self.name,
            "group_id": self.group_id,
            "metadata": self.metadata,
        }