tee_peer() — langchain Function Reference

Architecture documentation for the tee_peer() function in iter.py from the langchain codebase.

Function python CoreAbstractions RunnableInterface calls 1 called by 1

Entity Profile

CoreAbstractions→ RunnableInterface→ tee_peer() — langchain Function Reference

Dependency Diagram

graph TD
  16ec1910_820e_5f2d_8846_3a0c78f077de["tee_peer()"]
  b289db43_7cd8_e404_08a4_ad932b851c13["iter.py"]
  16ec1910_820e_5f2d_8846_3a0c78f077de -->|defined in| b289db43_7cd8_e404_08a4_ad932b851c13
  260deecc_1a31_b67a_f8f2_9d251be84e64["__init__()"]
  260deecc_1a31_b67a_f8f2_9d251be84e64 -->|calls| 16ec1910_820e_5f2d_8846_3a0c78f077de
  884c0eba_4881_a0c7_628a_a68b059c4131["close()"]
  16ec1910_820e_5f2d_8846_3a0c78f077de -->|calls| 884c0eba_4881_a0c7_628a_a68b059c4131
  style 16ec1910_820e_5f2d_8846_3a0c78f077de fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/utils/iter.py lines 35–88

def tee_peer(
    iterator: Iterator[T],
    # the buffer specific to this peer
    buffer: deque[T],
    # the buffers of all peers, including our own
    peers: list[deque[T]],
    lock: AbstractContextManager[Any],
) -> Generator[T, None, None]:
    """An individual iterator of a `.tee`.

    This function is a generator that yields items from the shared iterator `iterator`.
    It buffers items until the least advanced iterator has yielded them as well. The
    buffer is shared with all other peers.

    Args:
        iterator: The shared iterator.
        buffer: The buffer for this peer.
        peers: The buffers of all peers.
        lock: The lock to synchronise access to the shared buffers.

    Yields:
        The next item from the shared iterator.
    """
    try:
        while True:
            if not buffer:
                with lock:
                    # Another peer produced an item while we were waiting for the lock.
                    # Proceed with the next loop iteration to yield the item.
                    if buffer:
                        continue
                    try:
                        item = next(iterator)
                    except StopIteration:
                        break
                    else:
                        # Append to all buffers, including our own. We'll fetch our
                        # item from the buffer again, instead of yielding it directly.
                        # This ensures the proper item ordering if any of our peers
                        # are fetching items concurrently. They may have buffered their
                        # item already.
                        for peer_buffer in peers:
                            peer_buffer.append(item)
            yield buffer.popleft()
    finally:
        with lock:
            # this peer is done - remove its buffer
            for idx, peer_buffer in enumerate(peers):  # pragma: no branch
                if peer_buffer is buffer:
                    peers.pop(idx)
                    break
            # if we are the last peer, try and close the iterator
            if not peers and hasattr(iterator, "close"):
                iterator.close()