LogStreamCallbackHandler Class — langchain Architecture

Source Code

libs/core/langchain_core/tracers/log_stream.py lines 232–561

class LogStreamCallbackHandler(BaseTracer, _StreamingCallbackHandler):
    """Tracer that streams run logs to a stream."""

    def __init__(
        self,
        *,
        auto_close: bool = True,
        include_names: Sequence[str] | None = None,
        include_types: Sequence[str] | None = None,
        include_tags: Sequence[str] | None = None,
        exclude_names: Sequence[str] | None = None,
        exclude_types: Sequence[str] | None = None,
        exclude_tags: Sequence[str] | None = None,
        # Schema format is for internal use only.
        _schema_format: Literal["original", "streaming_events"] = "streaming_events",
    ) -> None:
        """A tracer that streams run logs to a stream.

        Args:
            auto_close: Whether to close the stream when the root run finishes.
            include_names: Only include runs from `Runnable` objects with matching
                names.
            include_types: Only include runs from `Runnable` objects with matching
                types.
            include_tags: Only include runs from `Runnable` objects with matching tags.
            exclude_names: Exclude runs from `Runnable` objects with matching names.
            exclude_types: Exclude runs from `Runnable` objects with matching types.
            exclude_tags: Exclude runs from `Runnable` objects with matching tags.
            _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 deprecated
                    entirely in favor of a dedicated async tracer for streaming events.

        Raises:
            ValueError: If an invalid schema format is provided (internal use only).
        """
        if _schema_format not in {"original", "streaming_events"}:
            msg = (
                f"Invalid schema format: {_schema_format}. "
                f"Expected one of 'original', 'streaming_events'."
            )
            raise ValueError(msg)
        super().__init__(_schema_format=_schema_format)

        self.auto_close = auto_close
        self.include_names = include_names
        self.include_types = include_types
        self.include_tags = include_tags
        self.exclude_names = exclude_names
        self.exclude_types = exclude_types
        self.exclude_tags = exclude_tags

        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
        memory_stream = _MemoryStream[RunLogPatch](loop)
        self.lock = threading.Lock()
        self.send_stream = memory_stream.get_send_stream()
        self.receive_stream = memory_stream.get_receive_stream()
        self._key_map_by_run_id: dict[UUID, str] = {}
        self._counter_map_by_name: dict[str, int] = defaultdict(int)
        self.root_id: UUID | None = None

    def __aiter__(self) -> AsyncIterator[RunLogPatch]:
        """Iterate over the stream of run logs.

        Returns:
            An async iterator over the run log patches.
        """
        return self.receive_stream.__aiter__()

    def send(self, *ops: dict[str, Any]) -> bool:
        """Send a patch to the stream, return `False` if the stream is closed.

        Args: