Source code for langchain_core.runnables.utils
from __future__ import annotations
import ast
import asyncio
import inspect
import textwrap
from inspect import signature
from itertools import groupby
from typing import (
    Any,
    AsyncIterable,
    Callable,
    Coroutine,
    Dict,
    Iterable,
    List,
    Mapping,
    NamedTuple,
    Optional,
    Protocol,
    Sequence,
    Set,
    TypeVar,
    Union,
)
Input = TypeVar("Input", contravariant=True)
# Output type should implement __concat__, as eg str, list, dict do
Output = TypeVar("Output", covariant=True)
[docs]async def gated_coro(semaphore: asyncio.Semaphore, coro: Coroutine) -> Any:
    """Run a coroutine with a semaphore.
    Args:
        semaphore: The semaphore to use.
        coro: The coroutine to run.
    Returns:
        The result of the coroutine.
    """
    async with semaphore:
        return await coro
[docs]async def gather_with_concurrency(n: Union[int, None], *coros: Coroutine) -> list:
    """Gather coroutines with a limit on the number of concurrent coroutines."""
    if n is None:
        return await asyncio.gather(*coros)
    semaphore = asyncio.Semaphore(n)
    return await asyncio.gather(*(gated_coro(semaphore, c) for c in coros))
[docs]def accepts_run_manager(callable: Callable[..., Any]) -> bool:
    """Check if a callable accepts a run_manager argument."""
    try:
        return signature(callable).parameters.get("run_manager") is not None
    except ValueError:
        return False