Traces

Trace

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

Source code in 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 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 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 agents/tracing/traces.py

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

NoOpTrace

Bases: Trace

一个不会被记录的 no-op（无操作）追踪。

Source code in agents/tracing/traces.py

class NoOpTrace(Trace):
    """
    一个不会被记录的 no-op（无操作）追踪。
    """

    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:
        return "no-op"

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

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

TraceImpl

Bases: Trace

一个会被 tracing 库记录的追踪。

Source code in agents/tracing/traces.py

class TraceImpl(Trace):
    """
    一个会被 tracing 库记录的追踪。
    """

    __slots__ = (
        "_name",
        "_trace_id",
        "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,
    ):
        self._name = name
        self._trace_id = trace_id or util.gen_trace_id()
        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

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