Source code for langchain_core.runnables.passthrough
"""Implementation of the RunnablePassthrough."""
from __future__ import annotations
import asyncio
import inspect
import threading
from typing import (
    TYPE_CHECKING,
    Any,
    AsyncIterator,
    Awaitable,
    Callable,
    Dict,
    Iterator,
    List,
    Mapping,
    Optional,
    Type,
    Union,
    cast,
)
from langchain_core.pydantic_v1 import BaseModel, create_model
from langchain_core.runnables.base import (
    Other,
    Runnable,
    RunnableParallel,
    RunnableSerializable,
)
from langchain_core.runnables.config import (
    RunnableConfig,
    acall_func_with_variable_args,
    call_func_with_variable_args,
    get_executor_for_config,
    patch_config,
)
from langchain_core.runnables.utils import AddableDict, ConfigurableFieldSpec
from langchain_core.utils.aiter import atee, py_anext
from langchain_core.utils.iter import safetee
if TYPE_CHECKING:
    from langchain_core.callbacks.manager import (
        AsyncCallbackManagerForChainRun,
        CallbackManagerForChainRun,
    )
[docs]def identity(x: Other) -> Other:
    """An identity function"""
    return x
[docs]async def aidentity(x: Other) -> Other:
    """An async identity function"""
    return x
[docs]class RunnablePassthrough(RunnableSerializable[Other, Other]):
    """A runnable to passthrough inputs unchanged or with additional keys.
    This runnable behaves almost like the identity function, except that it
    can be configured to add additional keys to the output, if the input is a
    dict.