Source code for langchain.schema.runnable.utils
from __future__ import annotations
import ast
import asyncio
import inspect
import textwrap
from inspect import signature
from typing import (
    Any,
    AsyncIterable,
    Callable,
    Coroutine,
    Dict,
    Iterable,
    List,
    Optional,
    Protocol,
    Set,
    TypeVar,
    Union,
)
Input = TypeVar("Input")
# Output type should implement __concat__, as eg str, list, dict do
Output = TypeVar("Output")
[docs]async def gated_coro(semaphore: asyncio.Semaphore, coro: Coroutine) -> Any:
    async with semaphore:
        return await coro
[docs]async def gather_with_concurrency(n: Union[int, None], *coros: Coroutine) -> list:
    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:
    try:
        return signature(callable).parameters.get("run_manager") is not None
    except ValueError:
        return False
[docs]def accepts_config(callable: Callable[..., Any]) -> bool:
    try:
        return signature(callable).parameters.get("config") is not None
    except ValueError:
        return False
[docs]class IsLocalDict(ast.NodeVisitor):
[docs]    def __init__(self, name: str, keys: Set[str]) -> None:
        self.name = name
        self.keys = keys
[docs]    def visit_Subscript(self, node: ast.Subscript) -> Any: