elif isinstance(callbacks, BaseCallbackManager):
            callbacks = callbacks.copy()
            callbacks.inheritable_handlers.append(stream)
            config["callbacks"] = callbacks
        else:
            raise ValueError(
                f"Unexpected type for callbacks: {callbacks}."
                "Expected None, list or AsyncCallbackManager."
            )
        # Call the runnable in streaming mode,
        # add each chunk to the output stream
        async def consume_astream() -> None:
            try:
                async for chunk in self.astream(input, config, **kwargs):
                    await stream.send_stream.send(
                        RunLogPatch(
                            {
                                "op": "add",
                                "path": "/streamed_output/-",
                                "value": chunk,
                            }
                        )
                    )
            finally:
                await stream.send_stream.aclose()
        # Start the runnable in a task, so we can start consuming output
        task = asyncio.create_task(consume_astream())
        try:
            # Yield each chunk from the output stream
            async for log in stream:
                yield log
        finally:
            # Wait for the runnable to finish, if not cancelled (eg. by break)
            try:
                await task
            except asyncio.CancelledError:
                pass
[docs]    def transform(
        self,
        input: Iterator[Input],
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> Iterator[Output]:
        """
        Default implementation of transform, which buffers input and then calls stream.
        Subclasses should override this method if they can start producing output while
        input is still being generated.
        """
        final: Input
        got_first_val = False
        for chunk in input: