_TracerCore Class — langchain Architecture

Source Code

libs/core/langchain_core/tracers/core.py lines 40–705

class _TracerCore(ABC):
    """Abstract base class for tracers.

    This class provides common methods, and reusable methods for tracers.
    """

    log_missing_parent: bool = True

    def __init__(
        self,
        *,
        _schema_format: Literal[
            "original", "streaming_events", "original+chat"
        ] = "original",
        **kwargs: Any,
    ) -> None:
        """Initialize the tracer.

        Args:
            _schema_format: Primarily changes how the inputs and outputs are handled.

                For internal use only. This API will change.

                - `'original'` is the format used by all current tracers.

                    This format is slightly inconsistent with respect to inputs and
                    outputs.
                - `'streaming_events'` is used for supporting streaming events, for
                    internal usage. It will likely change in the future, or be
                    deprecated entirely in favor of a dedicated async tracer for
                    streaming events.
                - `'original+chat'` is a format that is the same as `'original'` except
                    it does NOT raise an attribute error `on_chat_model_start`
            **kwargs: Additional keyword arguments that will be passed to the
                superclass.
        """
        super().__init__(**kwargs)

        self._schema_format = _schema_format  # For internal use only API will change.

        self.run_map: dict[str, Run] = {}
        """Map of run ID to run. Cleared on run end."""

        self.order_map: dict[UUID, tuple[UUID, str]] = {}
        """Map of run ID to (trace_id, dotted_order). Cleared when tracer GCed."""

    @abstractmethod
    def _persist_run(self, run: Run) -> Coroutine[Any, Any, None] | None:
        """Persist a run."""

    @staticmethod
    def _add_child_run(
        parent_run: Run,
        child_run: Run,
    ) -> None:
        """Add child run to a chain run or tool run."""
        parent_run.child_runs.append(child_run)

    @staticmethod
    def _get_stacktrace(error: BaseException) -> str:
        """Get the stacktrace of the parent error."""
        msg = repr(error)
        try:
            tb = traceback.format_exception(error)
            return (msg + "\n\n".join(tb)).strip()
        except Exception:
            return msg

    def _start_trace(self, run: Run) -> Coroutine[Any, Any, None] | None:  # type: ignore[return]
        current_dotted_order = run.start_time.strftime("%Y%m%dT%H%M%S%fZ") + str(run.id)
        if run.parent_run_id:
            if parent := self.order_map.get(run.parent_run_id):
                run.trace_id, run.dotted_order = parent
                run.dotted_order += "." + current_dotted_order
                if parent_run := self.run_map.get(str(run.parent_run_id)):
                    self._add_child_run(parent_run, run)
            else:
                if self.log_missing_parent:
                    logger.debug(
                        "Parent run %s not found for run %s. Treating as a root run.",
                        run.parent_run_id,