)
        # invoke the first steps
        try:
            for step in [self.first] + self.middle:
                input = await step.ainvoke(
                    input,
                    # mark each step as a child run
                    _patch_config(config, run_manager.get_child()),
                )
        except (KeyboardInterrupt, Exception) as e:
            await run_manager.on_chain_error(e)
            raise
        # stream the last step
        final: Union[Output, None] = None
        final_supported = True
        try:
            async for output in self.last.astream(
                input,
                # mark the last step as a child run
                _patch_config(config, run_manager.get_child()),
            ):
                yield output
                # Accumulate output if possible, otherwise disable accumulation
                if final_supported:
                    if final is None:
                        final = output
                    else:
                        try:
                            final += output  # type: ignore[operator]
                        except TypeError:
                            final = None
                            final_supported = False
                            pass
        # finish the root run
        except (KeyboardInterrupt, Exception) as e:
            await run_manager.on_chain_error(e)
            raise
        else:
            await run_manager.on_chain_end(
                final if isinstance(final, dict) else {"output": final}
            )
[docs]class RunnableMap(Serializable, Runnable[Input, Dict[str, Any]]):
    steps: Mapping[str, Runnable[Input, Any]]
    def __init__(
        self,
        steps: Mapping[
            str,
            Union[
                Runnable[Input, Any],
                Callable[[Input], Any],
                Mapping[str, Union[Runnable[Input, Any], Callable[[Input], Any]]],
            ],
        ],