from langchain.utils.iter import safetee
Other = TypeVar("Other")
[docs]class Runnable(Generic[Input, Output], ABC):
    """A Runnable is a unit of work that can be invoked, batched, streamed, or
    transformed."""
    @property
    def InputType(self) -> Type[Input]:
        for cls in self.__class__.__orig_bases__:  # type: ignore[attr-defined]
            type_args = get_args(cls)
            if type_args and len(type_args) == 2:
                return type_args[0]
        raise TypeError(
            f"Runnable {self.__class__.__name__} doesn't have an inferable InputType. "
            "Override the InputType property to specify the input type."
        )
    @property
    def OutputType(self) -> Type[Output]:
        for cls in self.__class__.__orig_bases__:  # type: ignore[attr-defined]
            type_args = get_args(cls)
            if type_args and len(type_args) == 2:
                return type_args[1]
        raise TypeError(
            f"Runnable {self.__class__.__name__} doesn't have an inferable OutputType. "
            "Override the OutputType property to specify the output type."
        )
    @property
    def input_schema(self) -> Type[BaseModel]:
        root_type = self.InputType
        if inspect.isclass(root_type) and issubclass(root_type, BaseModel):
            return root_type
        return create_model(
            self.__class__.__name__ + "Input", __root__=(root_type, None)
        )
    @property
    def output_schema(self) -> Type[BaseModel]:
        root_type = self.OutputType