codekingpro/portable-devtools

1"""`Runnable` objects that can be dynamically configured."""2 3from __future__ import annotations4 5import enum6import threading7from abc import abstractmethod8from collections.abc import (9    AsyncIterator,10    Callable,11    Iterator,12    Sequence,13)14from functools import wraps15from typing import (16    TYPE_CHECKING,17    Any,18    cast,19)20from weakref import WeakValueDictionary21 22from pydantic import BaseModel, ConfigDict23from typing_extensions import override24 25from langchain_core.runnables.base import Runnable, RunnableSerializable26from langchain_core.runnables.config import (27    RunnableConfig,28    ensure_config,29    get_config_list,30    get_executor_for_config,31    merge_configs,32)33from langchain_core.runnables.utils import (34    AnyConfigurableField,35    ConfigurableField,36    ConfigurableFieldMultiOption,37    ConfigurableFieldSingleOption,38    ConfigurableFieldSpec,39    Input,40    Output,41    gather_with_concurrency,42    get_unique_config_specs,43)44 45if TYPE_CHECKING:46    from langchain_core.runnables.graph import Graph47 48 49class DynamicRunnable(RunnableSerializable[Input, Output]):50    """Serializable `Runnable` that can be dynamically configured.51 52    A `DynamicRunnable` should be initiated using the `configurable_fields` or53    `configurable_alternatives` method of a `Runnable`.54    """55 56    default: RunnableSerializable[Input, Output]57    """The default `Runnable` to use."""58 59    config: RunnableConfig | None = None60    """The configuration to use."""61 62    model_config = ConfigDict(63        arbitrary_types_allowed=True,64    )65 66    @classmethod67    @override68    def is_lc_serializable(cls) -> bool:69        """Return `True` as this class is serializable."""70        return True71 72    @classmethod73    @override74    def get_lc_namespace(cls) -> list[str]:75        """Get the namespace of the LangChain object.76 77        Returns:78            `["langchain", "schema", "runnable"]`79        """80        return ["langchain", "schema", "runnable"]81 82    @property83    @override84    def InputType(self) -> type[Input]:85        return self.default.InputType86 87    @property88    @override89    def OutputType(self) -> type[Output]:90        return self.default.OutputType91 92    @override93    def get_input_schema(self, config: RunnableConfig | None = None) -> type[BaseModel]:94        runnable, config = self.prepare(config)95        return runnable.get_input_schema(config)96 97    @override98    def get_output_schema(99        self, config: RunnableConfig | None = None100    ) -> type[BaseModel]:101        runnable, config = self.prepare(config)102        return runnable.get_output_schema(config)103 104    @override105    def get_graph(self, config: RunnableConfig | None = None) -> Graph:106        runnable, config = self.prepare(config)107        return runnable.get_graph(config)108 109    @override110    def with_config(111        self,112        config: RunnableConfig | None = None,113        # Sadly Unpack is not well supported by mypy so this will have to be untyped114        **kwargs: Any,115    ) -> Runnable[Input, Output]:116        return self.__class__(117            **{**self.__dict__, "config": ensure_config(merge_configs(config, kwargs))}  # type: ignore[arg-type]118        )119 120    def prepare(121        self, config: RunnableConfig | None = None122    ) -> tuple[Runnable[Input, Output], RunnableConfig]:123        """Prepare the `Runnable` for invocation.124 125        Args:126            config: The configuration to use.127 128        Returns:129            The prepared `Runnable` and configuration.130        """131        runnable: Runnable[Input, Output] = self132        while isinstance(runnable, DynamicRunnable):133            runnable, config = runnable._prepare(merge_configs(runnable.config, config))  # noqa: SLF001134        return runnable, cast("RunnableConfig", config)135 136    @abstractmethod137    def _prepare(138        self, config: RunnableConfig | None = None139    ) -> tuple[Runnable[Input, Output], RunnableConfig]: ...140 141    @override142    def invoke(143        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any144    ) -> Output:145        runnable, config = self.prepare(config)146        return runnable.invoke(input, config, **kwargs)147 148    @override149    async def ainvoke(150        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any151    ) -> Output:152        runnable, config = self.prepare(config)153        return await runnable.ainvoke(input, config, **kwargs)154 155    @override156    def batch(157        self,158        inputs: list[Input],159        config: RunnableConfig | list[RunnableConfig] | None = None,160        *,161        return_exceptions: bool = False,162        **kwargs: Any | None,163    ) -> list[Output]:164        configs = get_config_list(config, len(inputs))165        prepared = [self.prepare(c) for c in configs]166 167        if all(p is self.default for p, _ in prepared):168            return self.default.batch(169                inputs,170                [c for _, c in prepared],171                return_exceptions=return_exceptions,172                **kwargs,173            )174 175        if not inputs:176            return []177 178        def invoke(179            prepared: tuple[Runnable[Input, Output], RunnableConfig],180            input_: Input,181        ) -> Output | Exception:182            bound, config = prepared183            if return_exceptions:184                try:185                    return bound.invoke(input_, config, **kwargs)186                except Exception as e:187                    return e188            else:189                return bound.invoke(input_, config, **kwargs)190 191        # If there's only one input, don't bother with the executor192        if len(inputs) == 1:193            return cast("list[Output]", [invoke(prepared[0], inputs[0])])194 195        with get_executor_for_config(configs[0]) as executor:196            return cast("list[Output]", list(executor.map(invoke, prepared, inputs)))197 198    @override199    async def abatch(200        self,201        inputs: list[Input],202        config: RunnableConfig | list[RunnableConfig] | None = None,203        *,204        return_exceptions: bool = False,205        **kwargs: Any | None,206    ) -> list[Output]:207        configs = get_config_list(config, len(inputs))208        prepared = [self.prepare(c) for c in configs]209 210        if all(p is self.default for p, _ in prepared):211            return await self.default.abatch(212                inputs,213                [c for _, c in prepared],214                return_exceptions=return_exceptions,215                **kwargs,216            )217 218        if not inputs:219            return []220 221        async def ainvoke(222            prepared: tuple[Runnable[Input, Output], RunnableConfig],223            input_: Input,224        ) -> Output | Exception:225            bound, config = prepared226            if return_exceptions:227                try:228                    return await bound.ainvoke(input_, config, **kwargs)229                except Exception as e:230                    return e231            else:232                return await bound.ainvoke(input_, config, **kwargs)233 234        coros = map(ainvoke, prepared, inputs)235        return await gather_with_concurrency(configs[0].get("max_concurrency"), *coros)236 237    @override238    def stream(239        self,240        input: Input,241        config: RunnableConfig | None = None,242        **kwargs: Any | None,243    ) -> Iterator[Output]:244        runnable, config = self.prepare(config)245        return runnable.stream(input, config, **kwargs)246 247    @override248    async def astream(249        self,250        input: Input,251        config: RunnableConfig | None = None,252        **kwargs: Any | None,253    ) -> AsyncIterator[Output]:254        runnable, config = self.prepare(config)255        async for chunk in runnable.astream(input, config, **kwargs):256            yield chunk257 258    @override259    def transform(260        self,261        input: Iterator[Input],262        config: RunnableConfig | None = None,263        **kwargs: Any | None,264    ) -> Iterator[Output]:265        runnable, config = self.prepare(config)266        return runnable.transform(input, config, **kwargs)267 268    @override269    async def atransform(270        self,271        input: AsyncIterator[Input],272        config: RunnableConfig | None = None,273        **kwargs: Any | None,274    ) -> AsyncIterator[Output]:275        runnable, config = self.prepare(config)276        async for chunk in runnable.atransform(input, config, **kwargs):277            yield chunk278 279    @override280    def __getattr__(self, name: str) -> Any:  # type: ignore[misc]281        attr = getattr(self.default, name)282        if callable(attr):283 284            @wraps(attr)285            def wrapper(*args: Any, **kwargs: Any) -> Any:286                for key, arg in kwargs.items():287                    if key == "config" and (288                        isinstance(arg, dict)289                        and "configurable" in arg290                        and isinstance(arg["configurable"], dict)291                    ):292                        runnable, config = self.prepare(cast("RunnableConfig", arg))293                        kwargs = {**kwargs, "config": config}294                        return getattr(runnable, name)(*args, **kwargs)295 296                for idx, arg in enumerate(args):297                    if (298                        isinstance(arg, dict)299                        and "configurable" in arg300                        and isinstance(arg["configurable"], dict)301                    ):302                        runnable, config = self.prepare(cast("RunnableConfig", arg))303                        argsl = list(args)304                        argsl[idx] = config305                        return getattr(runnable, name)(*argsl, **kwargs)306 307                if self.config:308                    runnable, config = self.prepare()309                    return getattr(runnable, name)(*args, **kwargs)310 311                return attr(*args, **kwargs)312 313            return wrapper314 315        return attr316 317 318class RunnableConfigurableFields(DynamicRunnable[Input, Output]):319    """`Runnable` that can be dynamically configured.320 321    A `RunnableConfigurableFields` should be initiated using the322    `configurable_fields` method of a `Runnable`.323 324    Here is an example of using a `RunnableConfigurableFields` with LLMs:325 326        ```python327        from langchain_core.prompts import PromptTemplate328        from langchain_core.runnables import ConfigurableField329        from langchain_openai import ChatOpenAI330 331        model = ChatOpenAI(temperature=0).configurable_fields(332            temperature=ConfigurableField(333                id="temperature",334                name="LLM Temperature",335                description="The temperature of the LLM",336            )337        )338        # This creates a RunnableConfigurableFields for a chat model.339 340        # When invoking the created RunnableSequence, you can pass in the341        # value for your ConfigurableField's id which in this case342        # will be change in temperature343 344        prompt = PromptTemplate.from_template("Pick a random number above {x}")345        chain = prompt | model346 347        chain.invoke({"x": 0})348        chain.invoke({"x": 0}, config={"configurable": {"temperature": 0.9}})349        ```350 351    Here is an example of using a `RunnableConfigurableFields` with `HubRunnables`:352 353        ```python354        from langchain_core.prompts import PromptTemplate355        from langchain_core.runnables import ConfigurableField356        from langchain_openai import ChatOpenAI357        from langchain.runnables.hub import HubRunnable358 359        prompt = HubRunnable("rlm/rag-prompt").configurable_fields(360            owner_repo_commit=ConfigurableField(361                id="hub_commit",362                name="Hub Commit",363                description="The Hub commit to pull from",364            )365        )366 367        prompt.invoke({"question": "foo", "context": "bar"})368 369        # Invoking prompt with `with_config` method370 371        prompt.invoke(372            {"question": "foo", "context": "bar"},373            config={"configurable": {"hub_commit": "rlm/rag-prompt-llama"}},374        )375        ```376    """377 378    fields: dict[str, AnyConfigurableField]379    """The configurable fields to use."""380 381    @property382    def config_specs(self) -> list[ConfigurableFieldSpec]:383        """Get the configuration specs for the `RunnableConfigurableFields`.384 385        Returns:386            The configuration specs.387        """388        config_specs = []389 390        default_fields = type(self.default).model_fields391        for field_name, spec in self.fields.items():392            if isinstance(spec, ConfigurableField):393                config_specs.append(394                    ConfigurableFieldSpec(395                        id=spec.id,396                        name=spec.name,397                        description=spec.description398                        or default_fields[field_name].description,399                        annotation=spec.annotation400                        or default_fields[field_name].annotation,401                        default=getattr(self.default, field_name),402                        is_shared=spec.is_shared,403                    )404                )405            else:406                config_specs.append(407                    make_options_spec(spec, default_fields[field_name].description)408                )409 410        config_specs.extend(self.default.config_specs)411 412        return get_unique_config_specs(config_specs)413 414    @override415    def configurable_fields(416        self, **kwargs: AnyConfigurableField417    ) -> RunnableSerializable[Input, Output]:418        return self.default.configurable_fields(**{**self.fields, **kwargs})419 420    def _prepare(421        self, config: RunnableConfig | None = None422    ) -> tuple[Runnable[Input, Output], RunnableConfig]:423        config = ensure_config(config)424        specs_by_id = {spec.id: (key, spec) for key, spec in self.fields.items()}425        configurable_fields = {426            specs_by_id[k][0]: v427            for k, v in config.get("configurable", {}).items()428            if k in specs_by_id and isinstance(specs_by_id[k][1], ConfigurableField)429        }430        configurable_single_options = {431            k: v.options[(config.get("configurable", {}).get(v.id) or v.default)]432            for k, v in self.fields.items()433            if isinstance(v, ConfigurableFieldSingleOption)434        }435        configurable_multi_options = {436            k: [437                v.options[o]438                for o in config.get("configurable", {}).get(v.id, v.default)439            ]440            for k, v in self.fields.items()441            if isinstance(v, ConfigurableFieldMultiOption)442        }443        configurable = {444            **configurable_fields,445            **configurable_single_options,446            **configurable_multi_options,447        }448 449        if configurable:450            init_params = {451                k: v452                for k, v in self.default.__dict__.items()453                if k in type(self.default).model_fields454            }455            return (456                self.default.__class__(**{**init_params, **configurable}),457                config,458            )459        return (self.default, config)460 461 462# Before Python 3.11 native StrEnum is not available463class StrEnum(str, enum.Enum):464    """String enum."""465 466 467_enums_for_spec: WeakValueDictionary[468    ConfigurableFieldSingleOption | ConfigurableFieldMultiOption | ConfigurableField,469    type[StrEnum],470] = WeakValueDictionary()471 472_enums_for_spec_lock = threading.Lock()473 474 475class RunnableConfigurableAlternatives(DynamicRunnable[Input, Output]):476    """`Runnable` that can be dynamically configured.477 478    A `RunnableConfigurableAlternatives` should be initiated using the479    `configurable_alternatives` method of a `Runnable` or can be480    initiated directly as well.481 482    Here is an example of using a `RunnableConfigurableAlternatives` that uses483    alternative prompts to illustrate its functionality:484 485        ```python486        from langchain_core.runnables import ConfigurableField487        from langchain_openai import ChatOpenAI488 489        # This creates a RunnableConfigurableAlternatives for Prompt Runnable490        # with two alternatives.491        prompt = PromptTemplate.from_template(492            "Tell me a joke about {topic}"493        ).configurable_alternatives(494            ConfigurableField(id="prompt"),495            default_key="joke",496            poem=PromptTemplate.from_template("Write a short poem about {topic}"),497        )498 499        # When invoking the created RunnableSequence, you can pass in the500        # value for your ConfigurableField's id which in this case will either be501        # `joke` or `poem`.502        chain = prompt | ChatOpenAI(model="gpt-5.4-mini")503 504        # The `with_config` method brings in the desired Prompt Runnable in your505        # Runnable Sequence.506        chain.with_config(configurable={"prompt": "poem"}).invoke({"topic": "bears"})507        ```508 509    Equivalently, you can initialize `RunnableConfigurableAlternatives` directly510    and use in LCEL in the same way:511 512        ```python513        from langchain_core.runnables import ConfigurableField514        from langchain_core.runnables.configurable import (515            RunnableConfigurableAlternatives,516        )517        from langchain_openai import ChatOpenAI518 519        prompt = RunnableConfigurableAlternatives(520            which=ConfigurableField(id="prompt"),521            default=PromptTemplate.from_template("Tell me a joke about {topic}"),522            default_key="joke",523            prefix_keys=False,524            alternatives={525                "poem": PromptTemplate.from_template("Write a short poem about {topic}")526            },527        )528        chain = prompt | ChatOpenAI(model="gpt-5.4-mini")529        chain.with_config(configurable={"prompt": "poem"}).invoke({"topic": "bears"})530        ```531    """532 533    which: ConfigurableField534    """The `ConfigurableField` to use to choose between alternatives."""535 536    alternatives: dict[537        str,538        Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],539    ]540    """The alternatives to choose from."""541 542    default_key: str = "default"543    """The enum value to use for the default option."""544 545    prefix_keys: bool546    """Whether to prefix configurable fields of each alternative with a namespace547    of the form <which.id>==<alternative_key>, e.g. a key named "temperature" used by548    the alternative named "gpt3" becomes "model==gpt3/temperature".549    """550 551    @property552    @override553    def config_specs(self) -> list[ConfigurableFieldSpec]:554        with _enums_for_spec_lock:555            if which_enum := _enums_for_spec.get(self.which):556                pass557            else:558                which_enum = StrEnum(  # type: ignore[call-overload]559                    self.which.name or self.which.id,560                    (561                        (v, v)562                        for v in [*list(self.alternatives.keys()), self.default_key]563                    ),564                )565                _enums_for_spec[self.which] = cast("type[StrEnum]", which_enum)566        return get_unique_config_specs(567            # which alternative568            [569                ConfigurableFieldSpec(570                    id=self.which.id,571                    name=self.which.name,572                    description=self.which.description,573                    annotation=which_enum,574                    default=self.default_key,575                    is_shared=self.which.is_shared,576                ),577            ]578            # config specs of the default option579            + (580                [581                    prefix_config_spec(s, f"{self.which.id}=={self.default_key}")582                    for s in self.default.config_specs583                ]584                if self.prefix_keys585                else self.default.config_specs586            )587            # config specs of the alternatives588            + [589                (590                    prefix_config_spec(s, f"{self.which.id}=={alt_key}")591                    if self.prefix_keys592                    else s593                )594                for alt_key, alt in self.alternatives.items()595                if isinstance(alt, RunnableSerializable)596                for s in alt.config_specs597            ]598        )599 600    @override601    def configurable_fields(602        self, **kwargs: AnyConfigurableField603    ) -> RunnableSerializable[Input, Output]:604        return self.__class__(605            which=self.which,606            default=self.default.configurable_fields(**kwargs),607            alternatives=self.alternatives,608            default_key=self.default_key,609            prefix_keys=self.prefix_keys,610        )611 612    def _prepare(613        self, config: RunnableConfig | None = None614    ) -> tuple[Runnable[Input, Output], RunnableConfig]:615        config = ensure_config(config)616        which = config.get("configurable", {}).get(self.which.id, self.default_key)617        # remap configurable keys for the chosen alternative618        if self.prefix_keys:619            config = cast(620                "RunnableConfig",621                {622                    **config,623                    "configurable": {624                        _strremoveprefix(k, f"{self.which.id}=={which}/"): v625                        for k, v in config.get("configurable", {}).items()626                    },627                },628            )629        # return the chosen alternative630        if which == self.default_key:631            return (self.default, config)632        if which in self.alternatives:633            alt = self.alternatives[which]634            if isinstance(alt, Runnable):635                return (alt, config)636            return (alt(), config)637        msg = f"Unknown alternative: {which}"638        raise ValueError(msg)639 640 641def _strremoveprefix(s: str, prefix: str) -> str:642    """`str.removeprefix()` is only available in Python 3.9+."""643    return s.replace(prefix, "", 1) if s.startswith(prefix) else s644 645 646def prefix_config_spec(647    spec: ConfigurableFieldSpec, prefix: str648) -> ConfigurableFieldSpec:649    """Prefix the id of a `ConfigurableFieldSpec`.650 651    This is useful when a `RunnableConfigurableAlternatives` is used as a652    `ConfigurableField` of another `RunnableConfigurableAlternatives`.653 654    Args:655        spec: The `ConfigurableFieldSpec` to prefix.656        prefix: The prefix to add.657 658    Returns:659        The prefixed `ConfigurableFieldSpec`.660    """661    return (662        ConfigurableFieldSpec(663            id=f"{prefix}/{spec.id}",664            name=spec.name,665            description=spec.description,666            annotation=spec.annotation,667            default=spec.default,668            is_shared=spec.is_shared,669        )670        if not spec.is_shared671        else spec672    )673 674 675def make_options_spec(676    spec: ConfigurableFieldSingleOption | ConfigurableFieldMultiOption,677    description: str | None,678) -> ConfigurableFieldSpec:679    """Make options spec.680 681    Make a `ConfigurableFieldSpec` for a `ConfigurableFieldSingleOption` or682    `ConfigurableFieldMultiOption`.683 684    Args:685        spec: The `ConfigurableFieldSingleOption` or `ConfigurableFieldMultiOption`.686        description: The description to use if the spec does not have one.687 688    Returns:689        The `ConfigurableFieldSpec`.690    """691    with _enums_for_spec_lock:692        if enum := _enums_for_spec.get(spec):693            pass694        else:695            enum = StrEnum(  # type: ignore[call-overload]696                spec.name or spec.id,697                ((v, v) for v in list(spec.options.keys())),698            )699            _enums_for_spec[spec] = cast("type[StrEnum]", enum)700    if isinstance(spec, ConfigurableFieldSingleOption):701        return ConfigurableFieldSpec(702            id=spec.id,703            name=spec.name,704            description=spec.description or description,705            annotation=enum,706            default=spec.default,707            is_shared=spec.is_shared,708        )709    return ConfigurableFieldSpec(710        id=spec.id,711        name=spec.name,712        description=spec.description or description,713        annotation=Sequence[enum],  # type: ignore[valid-type]714        default=spec.default,715        is_shared=spec.is_shared,716    )717