base.py — langchain Source File

Source Code

"""Base classes for chain routing."""

from __future__ import annotations

from abc import ABC
from collections.abc import Mapping
from typing import Any, NamedTuple

from langchain_core.callbacks import (
    AsyncCallbackManagerForChainRun,
    CallbackManagerForChainRun,
    Callbacks,
)
from pydantic import ConfigDict
from typing_extensions import override

from langchain_classic.chains.base import Chain


class Route(NamedTuple):
    """A route to a destination chain."""

    destination: str | None
    next_inputs: dict[str, Any]


class RouterChain(Chain, ABC):
    """Chain that outputs the name of a destination chain and the inputs to it."""

    @property
    @override
    def output_keys(self) -> list[str]:
        return ["destination", "next_inputs"]

    def route(self, inputs: dict[str, Any], callbacks: Callbacks = None) -> Route:
        """Route inputs to a destination chain.

        Args:
            inputs: inputs to the chain
            callbacks: callbacks to use for the chain

        Returns:
            a Route object
        """
        result = self(inputs, callbacks=callbacks)
        return Route(result["destination"], result["next_inputs"])

    async def aroute(
        self,
        inputs: dict[str, Any],
        callbacks: Callbacks = None,
    ) -> Route:
        """Route inputs to a destination chain.

        Args:
            inputs: inputs to the chain
            callbacks: callbacks to use for the chain

        Returns:
            a Route object
// ... (88 more lines)