runnables = [self.runnables[key] for key in keys]
        configs = self._get_config_list(config, len(inputs))
        return await _gather_with_concurrency(
            max_concurrency,
            *(
                runnable.ainvoke(input, config)
                for runnable, input, config in zip(runnables, actual_inputs, configs)
            ),
        )
[docs]    def stream(
        self, input: RouterInput, config: Optional[RunnableConfig] = None
    ) -> Iterator[Output]:
        key = input["key"]
        actual_input = input["input"]
        if key not in self.runnables:
            raise ValueError(f"No runnable associated with key '{key}'")
        runnable = self.runnables[key]
        yield from runnable.stream(actual_input, config)
[docs]    async def astream(
        self, input: RouterInput, config: Optional[RunnableConfig] = None
    ) -> AsyncIterator[Output]:
        key = input["key"]
        actual_input = input["input"]
        if key not in self.runnables:
            raise ValueError(f"No runnable associated with key '{key}'")
        runnable = self.runnables[key]
        async for output in runnable.astream(actual_input, config):
            yield output
def _patch_config(
    config: RunnableConfig, callback_manager: BaseCallbackManager
) -> RunnableConfig:
    config = config.copy()
    config["callbacks"] = callback_manager
    return config
def _coerce_to_runnable(
    thing: Union[
        Runnable[Input, Output],
        Callable[[Input], Output],
        Mapping[str, Union[Runnable[Input, Output], Callable[[Input], Output]]],
    ]
) -> Runnable[Input, Output]: