)
        if not inputs:
            return []
        async def ainvoke(
            prepared: Tuple[Runnable[Input, Output], RunnableConfig],
            input: Input,
        ) -> Union[Output, Exception]:
            bound, config = prepared
            if return_exceptions:
                try:
                    return await bound.ainvoke(input, config, **kwargs)
                except Exception as e:
                    return e
            else:
                return await bound.ainvoke(input, config, **kwargs)
        coros = map(ainvoke, prepared, inputs)
        return await gather_with_concurrency(configs[0].get("max_concurrency"), *coros)
[docs]    def stream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> Iterator[Output]:
        runnable, config = self._prepare(config)
        return runnable.stream(input, config, **kwargs)
[docs]    async def astream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> AsyncIterator[Output]:
        runnable, config = self._prepare(config)
        async for chunk in runnable.astream(input, config, **kwargs):
            yield chunk
[docs]    def transform(
        self,
        input: Iterator[Input],
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> Iterator[Output]:
        runnable, config = self._prepare(config)
        return runnable.transform(input, config, **kwargs)
[docs]    async def atransform(
        self,
        input: AsyncIterator[Input],