跨度

Span

基础类: ABC, Generic[TSpanData]

表示带有时间和上下文的可追踪操作的基类。

Span 代表一个 trace 中的单个操作（例如，LLM 调用、工具执行或 agent 运行）。Span 跟踪时间、操作之间的关系以及特定于操作的数据。

示例

# Creating a custom span
with custom_span("database_query", {
    "operation": "SELECT",
    "table": "users"
}) as span:
    results = await db.query("SELECT * FROM users")
    span.set_output({"count": len(results)})

# Handling errors in spans
with custom_span("risky_operation") as span:
    try:
        result = perform_risky_operation()
    except Exception as e:
        span.set_error({
            "message": str(e),
            "data": {"operation": "risky_operation"}
        })
        raise

注意: - Span 会自动嵌套在当前 trace 下 - 使用上下文管理器以可靠地开始/结束 - 包含相关数据，但避免敏感信息 - 使用 set_error() 正确处理错误

源代码在 src/agents/tracing/spans.py

class Span(abc.ABC, Generic[TSpanData]):
    """Base class for representing traceable operations with timing and context.

    A span represents a single operation within a trace (e.g., an LLM call, tool execution,
    or agent run). Spans track timing, relationships between operations, and operation-specific
    data.

    Type Args:
        TSpanData: The type of span-specific data this span contains.

    Example:
        ```python
        # Creating a custom span
        with custom_span("database_query", {
            "operation": "SELECT",
            "table": "users"
        }) as span:
            results = await db.query("SELECT * FROM users")
            span.set_output({"count": len(results)})

        # Handling errors in spans
        with custom_span("risky_operation") as span:
            try:
                result = perform_risky_operation()
            except Exception as e:
                span.set_error({
                    "message": str(e),
                    "data": {"operation": "risky_operation"}
                })
                raise
        ```

        Notes:
        - Spans automatically nest under the current trace
        - Use context managers for reliable start/finish
        - Include relevant data but avoid sensitive information
        - Handle errors properly using set_error()
    """

    @property
    @abc.abstractmethod
    def trace_id(self) -> str:
        """The ID of the trace this span belongs to.

        Returns:
            str: Unique identifier of the parent trace.
        """
        pass

    @property
    @abc.abstractmethod
    def span_id(self) -> str:
        """Unique identifier for this span.

        Returns:
            str: The span's unique ID within its trace.
        """
        pass

    @property
    @abc.abstractmethod
    def span_data(self) -> TSpanData:
        """Operation-specific data for this span.

        Returns:
            TSpanData: Data specific to this type of span (e.g., LLM generation data).
        """
        pass

    @abc.abstractmethod
    def start(self, mark_as_current: bool = False):
        """
        Start the span.

        Args:
            mark_as_current: If true, the span will be marked as the current span.
        """
        pass

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

        Args:
            reset_current: If true, the span will be reset as the current 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:
        """ID of the parent span, if any.

        Returns:
            str | None: The parent span's ID, or None if this is a root span.
        """
        pass

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

    @property
    @abc.abstractmethod
    def error(self) -> SpanError | None:
        """Any error that occurred during span execution.

        Returns:
            SpanError | None: Error details if an error occurred, None otherwise.
        """
        pass

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

    @property
    @abc.abstractmethod
    def started_at(self) -> str | None:
        """When the span started execution.

        Returns:
            str | None: ISO format timestamp of span start, None if not started.
        """
        pass

    @property
    @abc.abstractmethod
    def ended_at(self) -> str | None:
        """When the span finished execution.

        Returns:
            str | None: ISO format timestamp of span end, None if not finished.
        """
        pass

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

trace_id abstractmethod property

trace_id: str

此 span 属于的 trace 的 ID。

返回值

名称	类型	描述
str	str	父 trace 的唯一标识符。

span_id abstractmethod property

span_id: str

此 span 的唯一标识符。

返回值

名称	类型	描述
str	str	此 span 在其 trace 中的唯一 ID。

span_data abstractmethod property

span_data: TSpanData

此 span 的特定于操作的数据。

返回值

名称	类型	描述
TSpanData	TSpanData	特定于此类型 span 的数据（例如，LLM 生成数据）。

parent_id abstractmethod property

parent_id: str | None

父 span 的 ID（如果有）。

返回值

类型	描述
str | None	str | None: 父 span 的 ID，如果此 span 是根 span，则为 None。

error abstractmethod property

error: SpanError | None

span 执行期间发生的任何错误。

返回值

类型	描述
SpanError | None	SpanError | None: 如果发生错误，则为错误详细信息，否则为 None。

started_at abstractmethod property

started_at: str | None

span 开始执行的时间。

返回值

类型	描述
str | None	str | None: span 开始的 ISO 格式时间戳，如果尚未开始则为 None。

ended_at abstractmethod property

ended_at: str | None

span 完成执行的时间。

返回值

类型	描述
str | None	str | None: span 结束的 ISO 格式时间戳，如果尚未结束则为 None。

tracing_api_key abstractmethod property

tracing_api_key: str | None

导出此 span 时使用的 API 密钥。

start abstractmethod

start(mark_as_current: bool = False)

开始 span。

参数

名称	类型	描述	默认
mark_as_current	bool	如果为 true，则 span 将被标记为当前 span。	False

源代码在 src/agents/tracing/spans.py

@abc.abstractmethod
def start(self, mark_as_current: bool = False):
    """
    Start the span.

    Args:
        mark_as_current: If true, the span will be marked as the current span.
    """
    pass

finish abstractmethod

finish(reset_current: bool = False) -> None

结束 span。

参数

名称	类型	描述	默认
reset_current	bool	如果为 true，则 span 将被重置为当前 span。	False

源代码在 src/agents/tracing/spans.py

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

    Args:
        reset_current: If true, the span will be reset as the current span.
    """
    pass

NoOpSpan

基础类: Span[TSpanData]

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

当禁用 tracing 但仍然需要 span 操作时使用。

参数

名称	类型	描述	默认
span_data	TSpanData	此 span 的特定于操作的数据。	required

源代码在 src/agents/tracing/spans.py

class NoOpSpan(Span[TSpanData]):
    """A no-op implementation of Span that doesn't record any data.

    Used when tracing is disabled but span operations still need to work.

    Args:
        span_data: The operation-specific data for this span.
    """

    __slots__ = ("_span_data", "_prev_span_token")

    def __init__(self, span_data: TSpanData):
        self._span_data = span_data
        self._prev_span_token: contextvars.Token[Span[TSpanData] | None] | None = None

    @property
    def trace_id(self) -> str:
        return "no-op"

    @property
    def span_id(self) -> str:
        return "no-op"

    @property
    def span_data(self) -> TSpanData:
        return self._span_data

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

    def start(self, mark_as_current: bool = False):
        if mark_as_current:
            self._prev_span_token = Scope.set_current_span(self)

    def finish(self, reset_current: bool = False) -> None:
        if reset_current and self._prev_span_token is not None:
            Scope.reset_current_span(self._prev_span_token)
            self._prev_span_token = None

    def __enter__(self) -> Span[TSpanData]:
        self.start(mark_as_current=True)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        reset_current = True
        if exc_type is GeneratorExit:
            logger.debug("GeneratorExit, skipping span reset")
            reset_current = False

        self.finish(reset_current=reset_current)

    def set_error(self, error: SpanError) -> None:
        pass

    @property
    def error(self) -> SpanError | None:
        return None

    def export(self) -> dict[str, Any] | None:
        return None

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

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

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

SpanImpl

基础类: Span[TSpanData]

源代码在 src/agents/tracing/spans.py

class SpanImpl(Span[TSpanData]):
    __slots__ = (
        "_trace_id",
        "_span_id",
        "_parent_id",
        "_started_at",
        "_ended_at",
        "_error",
        "_prev_span_token",
        "_processor",
        "_span_data",
        "_tracing_api_key",
    )

    def __init__(
        self,
        trace_id: str,
        span_id: str | None,
        parent_id: str | None,
        processor: TracingProcessor,
        span_data: TSpanData,
        tracing_api_key: str | None,
    ):
        self._trace_id = trace_id
        self._span_id = span_id or util.gen_span_id()
        self._parent_id = parent_id
        self._started_at: str | None = None
        self._ended_at: str | None = None
        self._processor = processor
        self._error: SpanError | None = None
        self._prev_span_token: contextvars.Token[Span[TSpanData] | None] | None = None
        self._span_data = span_data
        self._tracing_api_key = tracing_api_key

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

    @property
    def span_id(self) -> str:
        return self._span_id

    @property
    def span_data(self) -> TSpanData:
        return self._span_data

    @property
    def parent_id(self) -> str | None:
        return self._parent_id

    def start(self, mark_as_current: bool = False):
        if self.started_at is not None:
            logger.warning("Span already started")
            return

        self._started_at = util.time_iso()
        self._processor.on_span_start(self)
        if mark_as_current:
            self._prev_span_token = Scope.set_current_span(self)

    def finish(self, reset_current: bool = False) -> None:
        if self.ended_at is not None:
            logger.warning("Span already finished")
            return

        self._ended_at = util.time_iso()
        self._processor.on_span_end(self)
        if reset_current and self._prev_span_token is not None:
            Scope.reset_current_span(self._prev_span_token)
            self._prev_span_token = None

    def __enter__(self) -> Span[TSpanData]:
        self.start(mark_as_current=True)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        reset_current = True
        if exc_type is GeneratorExit:
            logger.debug("GeneratorExit, skipping span reset")
            reset_current = False

        self.finish(reset_current=reset_current)

    def set_error(self, error: SpanError) -> None:
        self._error = error

    @property
    def error(self) -> SpanError | None:
        return self._error

    @property
    def started_at(self) -> str | None:
        return self._started_at

    @property
    def ended_at(self) -> str | None:
        return self._ended_at

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

    def export(self) -> dict[str, Any] | None:
        return {
            "object": "trace.span",
            "id": self.span_id,
            "trace_id": self.trace_id,
            "parent_id": self._parent_id,
            "started_at": self._started_at,
            "ended_at": self._ended_at,
            "span_data": self.span_data.export(),
            "error": self._error,
        }