Source code for langchain_core.runnables.configurable
from __future__ import annotations
import enum
import threading
from abc import abstractmethod
from typing import (
    Any,
    AsyncIterator,
    Callable,
    Dict,
    Iterator,
    List,
    Optional,
    Sequence,
    Tuple,
    Type,
    Union,
    cast,
)
from weakref import WeakValueDictionary
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.runnables.base import Runnable, RunnableSerializable
from langchain_core.runnables.config import (
    RunnableConfig,
    get_config_list,
    get_executor_for_config,
)
from langchain_core.runnables.utils import (
    AnyConfigurableField,
    ConfigurableField,
    ConfigurableFieldMultiOption,
    ConfigurableFieldSingleOption,
    ConfigurableFieldSpec,
    Input,
    Output,
    gather_with_concurrency,
    get_unique_config_specs,
)
[docs]class DynamicRunnable(RunnableSerializable[Input, Output]):
    """A Serializable Runnable that can be dynamically configured."""
    default: RunnableSerializable[Input, Output]
    class Config:
        arbitrary_types_allowed = True
[docs]    @classmethod
    def is_lc_serializable(cls) -> bool:
        return True
[docs]    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        """Get the namespace of the langchain object."""
        return ["langchain", "schema", "runnable"]
    @property
    def InputType(self) -> Type[Input]:
        return self.default.InputType
    @property
    def OutputType(self) -> Type[Output]:
        return self.default.OutputType
[docs]    def get_input_schema(