Source code for langchain.schema.runnable.passthrough
from __future__ import annotations
import asyncio
import threading
from typing import (
    Any,
    AsyncIterator,
    Callable,
    Dict,
    Iterator,
    List,
    Mapping,
    Optional,
    Type,
    Union,
    cast,
)
from langchain.load.serializable import Serializable
from langchain.pydantic_v1 import BaseModel, create_model
from langchain.schema.runnable.base import Input, Runnable, RunnableMap
from langchain.schema.runnable.config import RunnableConfig, get_executor_for_config
from langchain.schema.runnable.utils import AddableDict
from langchain.utils.aiter import atee, py_anext
from langchain.utils.iter import safetee
[docs]def identity(x: Input) -> Input:
    return x
[docs]async def aidentity(x: Input) -> Input:
    return x
[docs]class RunnablePassthrough(Serializable, Runnable[Input, Input]):
    """
    A runnable that passes through the input.
    """
    input_type: Optional[Type[Input]] = None
[docs]    @classmethod
    def is_lc_serializable(cls) -> bool:
        return True
[docs]    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        return cls.__module__.split(".")[:-1]
    @property
    def InputType(self) -> Any:
        return self.input_type or Any
    @property
    def OutputType(self) -> Any:
        return self.input_type or Any
[docs]    @classmethod
    def assign(
        cls,
        **kwargs: Union[
            Runnable[Dict[str, Any], Any],
            Callable[[Dict[str, Any]], Any],
            Mapping[