MultiRouteChain Class — langchain Architecture

Architecture documentation for the MultiRouteChain class in base.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  741a7f73_f662_9d17_6b91_e7e8709f4338["MultiRouteChain"]
  f3cef70e_11b0_61c9_7ec0_7308f4b45056["Chain"]
  741a7f73_f662_9d17_6b91_e7e8709f4338 -->|extends| f3cef70e_11b0_61c9_7ec0_7308f4b45056
  3d5ca688_b19a_893f_4a6b_4798a6f7566d["base.py"]
  741a7f73_f662_9d17_6b91_e7e8709f4338 -->|defined in| 3d5ca688_b19a_893f_4a6b_4798a6f7566d
  1279be43_06ff_3af8_82af_05ce61c47c82["input_keys()"]
  741a7f73_f662_9d17_6b91_e7e8709f4338 -->|method| 1279be43_06ff_3af8_82af_05ce61c47c82
  75d9ee0c_d912_078a_f9b9_d13bdf90be1d["output_keys()"]
  741a7f73_f662_9d17_6b91_e7e8709f4338 -->|method| 75d9ee0c_d912_078a_f9b9_d13bdf90be1d
  895503b7_0854_2cce_f272_84c2606acbb1["_call()"]
  741a7f73_f662_9d17_6b91_e7e8709f4338 -->|method| 895503b7_0854_2cce_f272_84c2606acbb1
  447ae41b_89ff_2a5a_8300_64cddd946b97["_acall()"]
  741a7f73_f662_9d17_6b91_e7e8709f4338 -->|method| 447ae41b_89ff_2a5a_8300_64cddd946b97

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/router/base.py lines 66–147

class MultiRouteChain(Chain):
    """Use a single chain to route an input to one of multiple candidate chains."""

    router_chain: RouterChain
    """Chain that routes inputs to destination chains."""
    destination_chains: Mapping[str, Chain]
    """Chains that return final answer to inputs."""
    default_chain: Chain
    """Default chain to use when none of the destination chains are suitable."""
    silent_errors: bool = False
    """If `True`, use default_chain when an invalid destination name is provided."""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        extra="forbid",
    )

    @property
    def input_keys(self) -> list[str]:
        """Will be whatever keys the router chain prompt expects."""
        return self.router_chain.input_keys

    @property
    def output_keys(self) -> list[str]:
        """Will always return text key."""
        return []

    def _call(
        self,
        inputs: dict[str, Any],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        callbacks = _run_manager.get_child()
        route = self.router_chain.route(inputs, callbacks=callbacks)

        _run_manager.on_text(
            str(route.destination) + ": " + str(route.next_inputs),
            verbose=self.verbose,
        )
        if not route.destination:
            return self.default_chain(route.next_inputs, callbacks=callbacks)
        if route.destination in self.destination_chains:
            return self.destination_chains[route.destination](
                route.next_inputs,
                callbacks=callbacks,
            )
        if self.silent_errors:
            return self.default_chain(route.next_inputs, callbacks=callbacks)
        msg = f"Received invalid destination chain name '{route.destination}'"
        raise ValueError(msg)

    async def _acall(
        self,
        inputs: dict[str, Any],
        run_manager: AsyncCallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _run_manager = run_manager or AsyncCallbackManagerForChainRun.get_noop_manager()
        callbacks = _run_manager.get_child()
        route = await self.router_chain.aroute(inputs, callbacks=callbacks)

        await _run_manager.on_text(
            str(route.destination) + ": " + str(route.next_inputs),
            verbose=self.verbose,
        )
        if not route.destination:
            return await self.default_chain.acall(
                route.next_inputs,
                callbacks=callbacks,
            )
        if route.destination in self.destination_chains:
            return await self.destination_chains[route.destination].acall(
                route.next_inputs,
                callbacks=callbacks,
            )
        if self.silent_errors:
            return await self.default_chain.acall(
                route.next_inputs,
                callbacks=callbacks,
            )
        msg = f"Received invalid destination chain name '{route.destination}'"