py_anext() — langchain Function Reference

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

Function python Observability TelemetryUtilities

Entity Profile

Observability→ TelemetryUtilities→ py_anext() — langchain Function Reference

Dependency Diagram

graph TD
  29ac23c1_5ed5_81b2_a05d_ab15cb74c190["py_anext()"]
  6c216df4_f8e5_a482_8469_57ced2643435["aiter.py"]
  29ac23c1_5ed5_81b2_a05d_ab15cb74c190 -->|defined in| 6c216df4_f8e5_a482_8469_57ced2643435
  style 29ac23c1_5ed5_81b2_a05d_ab15cb74c190 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/utils/aiter.py lines 38–84

def py_anext(
    iterator: AsyncIterator[T], default: T | Any = _no_default
) -> Awaitable[T | Any | None]:
    """Pure-Python implementation of `anext()` for testing purposes.

    Closely matches the builtin `anext()` C implementation.

    Can be used to compare the built-in implementation of the inner coroutines machinery
    to C-implementation of `__anext__()` and `send()` or `throw()` on the returned
    generator.

    Args:
        iterator: The async iterator to advance.
        default: The value to return if the iterator is exhausted.

            If not provided, a `StopAsyncIteration` exception is raised.

    Returns:
        The next value from the iterator, or the default value if the iterator is
            exhausted.

    Raises:
        TypeError: If the iterator is not an async iterator.
    """
    try:
        __anext__ = cast(
            "Callable[[AsyncIterator[T]], Awaitable[T]]", type(iterator).__anext__
        )
    except AttributeError as e:
        msg = f"{iterator!r} is not an async iterator"
        raise TypeError(msg) from e

    if default is _no_default:
        return __anext__(iterator)

    async def anext_impl() -> T | Any:
        try:
            # The C code is way more low-level than this, as it implements
            # all methods of the iterator protocol. In this implementation
            # we're relying on higher-level coroutine concepts, but that's
            # exactly what we want -- crosstest pure-Python high-level
            # implementation and low-level C anext() iterators.
            return await __anext__(iterator)
        except StopAsyncIteration:
            return default

    return anext_impl()