**kwargs: Optional[Any],
    ) -> Output:
        """Invoke this runnable asynchronously."""
        the_func = self.afunc if hasattr(self, "afunc") else self.func
        return await self._acall_with_config(
            self._ainvoke,
            input,
            self._config(config, the_func),
            **kwargs,
        )
[docs]class RunnableEachBase(RunnableSerializable[List[Input], List[Output]]):
    """
    A runnable that delegates calls to another runnable
    with each element of the input sequence.
    Use only if creating a new RunnableEach subclass with different __init__ args.
    """
    bound: Runnable[Input, Output]
    class Config:
        arbitrary_types_allowed = True
    @property
    def InputType(self) -> Any:
        return List[self.bound.InputType]  # type: ignore[name-defined]
[docs]    def get_input_schema(
        self, config: Optional[RunnableConfig] = None
    ) -> Type[BaseModel]:
        return create_model(
            "RunnableEachInput",
            __root__=(
                List[self.bound.get_input_schema(config)],  # type: ignore
                None,
            ),
        )
    @property
    def OutputType(self) -> Type[List[Output]]:
        return List[self.bound.OutputType]  # type: ignore[name-defined]
[docs]    def get_output_schema(
        self, config: Optional[RunnableConfig] = None
    ) -> Type[BaseModel]:
        schema = self.bound.get_output_schema(config)
        return create_model(
            "RunnableEachOutput",
            __root__=(
                List[schema],  # type: ignore
                None,
            ),
        )
    @property