Source code for langchain.schema.runnable.router
from __future__ import annotations
from typing import (
    Any,
    AsyncIterator,
    Callable,
    Iterator,
    List,
    Mapping,
    Optional,
    Union,
    cast,
)
from typing_extensions import TypedDict
from langchain.load.serializable import Serializable
from langchain.schema.runnable.base import Input, Output, Runnable, coerce_to_runnable
from langchain.schema.runnable.config import (
    RunnableConfig,
    get_config_list,
    get_executor_for_config,
)
from langchain.schema.runnable.utils import gather_with_concurrency
[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(Serializable, Runnable[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]]
    def __init__(
        self,
        runnables: Mapping[str, Union[Runnable[Any, Output], Callable[[Any], Output]]],
    ) -> None:
        super().__init__(
            runnables={key: coerce_to_runnable(r) for key, r in runnables.items()}
        )
    class Config:
        arbitrary_types_allowed = True
[docs]    @classmethod
    def is_lc_serializable(cls) -> bool:
        """Return whether this class is serializable."""
        return True
[docs]    @classmethod
    def get_lc_namespace(cls) -> List[str]: