init() — langchain Function Reference

Architecture documentation for the __init__() function in log_stream.py from the langchain codebase.

Function python Observability Tracers

Entity Profile

Observability→ Tracers→ __init__() — langchain Function Reference

Dependency Diagram

graph TD
  07946e14_8733_b693_5370_577c13824076["__init__()"]
  78fc514f_5439_00c6_7e00_021af9ce4e07["LogStreamCallbackHandler"]
  07946e14_8733_b693_5370_577c13824076 -->|defined in| 78fc514f_5439_00c6_7e00_021af9ce4e07
  style 07946e14_8733_b693_5370_577c13824076 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/tracers/log_stream.py lines 235–299

    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