init() — langchain Function Reference

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

Function python Observability TelemetryUtilities calls 1

Entity Profile

Observability→ TelemetryUtilities→ __init__() — langchain Function Reference

Dependency Diagram

graph TD
  a873a094_5a92_9243_5938_9be81c0c8843["__init__()"]
  56439126_245e_4a7c_b5a1_b9d43ab4ae02["Tee"]
  a873a094_5a92_9243_5938_9be81c0c8843 -->|defined in| 56439126_245e_4a7c_b5a1_b9d43ab4ae02
  badd2620_02d1_6899_7014_184b4f0e2001["tee_peer()"]
  a873a094_5a92_9243_5938_9be81c0c8843 -->|calls| badd2620_02d1_6899_7014_184b4f0e2001
  style a873a094_5a92_9243_5938_9be81c0c8843 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/utils/aiter.py lines 202–227

    def __init__(
        self,
        iterable: AsyncIterator[T],
        n: int = 2,
        *,
        lock: AbstractAsyncContextManager[Any] | None = None,
    ):
        """Create a `tee`.

        Args:
            iterable: The iterable to split.
            n: The number of iterators to create.
            lock: The lock to synchronise access to the shared buffers.

        """
        self._iterator = iterable.__aiter__()  # before 3.10 aiter() doesn't exist
        self._buffers: list[deque[T]] = [deque() for _ in range(n)]
        self._children = tuple(
            tee_peer(
                iterator=self._iterator,
                buffer=buffer,
                peers=self._buffers,
                lock=lock if lock is not None else NoLock(),
            )
            for buffer in self._buffers
        )