[docs]    def transform(
        self,
        input: Iterator[Input],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Iterator[Output]:
        yield from self.bound.transform(
            input,
            self._merge_configs(config),
            **{**self.kwargs, **kwargs},
        )
[docs]    async def atransform(
        self,
        input: AsyncIterator[Input],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> AsyncIterator[Output]:
        async for item in self.bound.atransform(
            input,
            self._merge_configs(config),
            **{**self.kwargs, **kwargs},
        ):
            yield item
RunnableBindingBase.update_forward_refs(RunnableConfig=RunnableConfig)
[docs]class RunnableBinding(RunnableBindingBase[Input, Output]):
    """Wrap a runnable with additional functionality.
    A RunnableBinding can be thought of as a "runnable decorator" that
    preserves the essential features of Runnable; i.e., batching, streaming,
    and async support, while adding additional functionality.
    Any class that inherits from Runnable can be bound to a `RunnableBinding`.
    Runnables expose a standard set of methods for creating `RunnableBindings`
    or sub-classes of `RunnableBindings` (e.g., `RunnableRetry`,
    `RunnableWithFallbacks`) that add additional functionality.
    These methods include:
    - `bind`: Bind kwargs to pass to the underlying runnable when running it.
    - `with_config`: Bind config to pass to the underlying runnable when running it.
    - `with_listeners`:  Bind lifecycle listeners to the underlying runnable.