Source code for langchain_core.runnables.router
from __future__ import annotations
from typing import (
    Any,
    AsyncIterator,
    Callable,
    Iterator,
    List,
    Mapping,
    Optional,
    Union,
    cast,
)
from typing_extensions import TypedDict
from langchain_core.runnables.base import (
    Input,
    Output,
    Runnable,
    RunnableSerializable,
    coerce_to_runnable,
)
from langchain_core.runnables.config import (
    RunnableConfig,
    get_config_list,
    get_executor_for_config,
)
from langchain_core.runnables.utils import (
    ConfigurableFieldSpec,
    gather_with_concurrency,
    get_unique_config_specs,
)
[docs]class RouterInput(TypedDict):
    """A Router input.
    Attributes:
        key: The key to route on.
        input: The input to pass to the selected runnable.
    """
    key: str
    input: Any
[docs]class RouterRunnable(RunnableSerializable[RouterInput, Output]):
    """
    A runnable that routes to a set of runnables based on Input['key'].
    Returns the output of the selected runnable.
    """
    runnables: Mapping[str, Runnable[Any, Output]]
    @property
    def config_specs(self) -> List[ConfigurableFieldSpec]:
        return get_unique_config_specs(
            spec for step in self.runnables.values() for spec in step.config_specs
        )
    def __init__(
        self,
        runnables: Mapping[str, Union[Runnable[Any, Output], Callable[[Any], Output]]],
    ) -> None:
        super().__init__(