**kwargs: Any,
    ) -> AsyncIterator[Dict[str, Any]]:
        async for chunk in self._atransform_stream_with_config(
            input, self._atransform, config, **kwargs
        ):
            yield chunk
[docs]    async def astream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> AsyncIterator[Dict[str, Any]]:
        async def input_aiter() -> AsyncIterator[Input]:
            yield input
        async for chunk in self.atransform(input_aiter(), config):
            yield chunk
# We support both names
RunnableMap = RunnableParallel
[docs]class RunnableGenerator(Runnable[Input, Output]):
    """
    A runnable that runs a generator function.
    """
[docs]    def __init__(
        self,
        transform: Union[
            Callable[[Iterator[Input]], Iterator[Output]],
            Callable[[AsyncIterator[Input]], AsyncIterator[Output]],
        ],
        atransform: Optional[
            Callable[[AsyncIterator[Input]], AsyncIterator[Output]]
        ] = None,
    ) -> None:
        if atransform is not None:
            self._atransform = atransform
        if inspect.isasyncgenfunction(transform):
            self._atransform = transform
        elif inspect.isgeneratorfunction(transform):
            self._transform = transform
        else:
            raise TypeError(
                "Expected a generator function type for `transform`."
                f"Instead got an unsupported type: {type(transform)}"
            )
    @property
    def InputType(self) -> Any: