EmbeddingRouterChain Class — langchain Architecture

Architecture documentation for the EmbeddingRouterChain class in embedding_router.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  a9ecc02f_6a92_622a_70dc_826e973e25bd["EmbeddingRouterChain"]
  8c933306_16e9_7f72_a392_80e52d7a7371["RouterChain"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|extends| 8c933306_16e9_7f72_a392_80e52d7a7371
  450e1536_9fbb_ff12_4c4e_e16cf70d26c5["embedding_router.py"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|defined in| 450e1536_9fbb_ff12_4c4e_e16cf70d26c5
  ca68184b_585d_32c0_72a1_a03f49e51e5b["input_keys()"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|method| ca68184b_585d_32c0_72a1_a03f49e51e5b
  99c33227_8d8c_86d2_ed9b_a32d7e686fc7["_call()"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|method| 99c33227_8d8c_86d2_ed9b_a32d7e686fc7
  7edf92a1_c163_8d6d_f59c_7de6fecf30ed["_acall()"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|method| 7edf92a1_c163_8d6d_f59c_7de6fecf30ed
  554892a5_c935_4b19_b6e9_8233d0752fe0["from_names_and_descriptions()"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|method| 554892a5_c935_4b19_b6e9_8233d0752fe0
  a17e7e17_0dea_42d9_3b08_2ce3197f8df1["afrom_names_and_descriptions()"]
  a9ecc02f_6a92_622a_70dc_826e973e25bd -->|method| a17e7e17_0dea_42d9_3b08_2ce3197f8df1

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/router/embedding_router.py lines 19–93

class EmbeddingRouterChain(RouterChain):
    """Chain that uses embeddings to route between options."""

    vectorstore: VectorStore
    routing_keys: list[str] = ["query"]

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

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

    @override
    def _call(
        self,
        inputs: dict[str, Any],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _input = ", ".join([inputs[k] for k in self.routing_keys])
        results = self.vectorstore.similarity_search(_input, k=1)
        return {"next_inputs": inputs, "destination": results[0].metadata["name"]}

    @override
    async def _acall(
        self,
        inputs: dict[str, Any],
        run_manager: AsyncCallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _input = ", ".join([inputs[k] for k in self.routing_keys])
        results = await self.vectorstore.asimilarity_search(_input, k=1)
        return {"next_inputs": inputs, "destination": results[0].metadata["name"]}

    @classmethod
    def from_names_and_descriptions(
        cls,
        names_and_descriptions: Sequence[tuple[str, Sequence[str]]],
        vectorstore_cls: type[VectorStore],
        embeddings: Embeddings,
        **kwargs: Any,
    ) -> EmbeddingRouterChain:
        """Convenience constructor."""
        documents = []
        for name, descriptions in names_and_descriptions:
            documents.extend(
                [
                    Document(page_content=description, metadata={"name": name})
                    for description in descriptions
                ]
            )
        vectorstore = vectorstore_cls.from_documents(documents, embeddings)
        return cls(vectorstore=vectorstore, **kwargs)

    @classmethod
    async def afrom_names_and_descriptions(
        cls,
        names_and_descriptions: Sequence[tuple[str, Sequence[str]]],
        vectorstore_cls: type[VectorStore],
        embeddings: Embeddings,
        **kwargs: Any,
    ) -> EmbeddingRouterChain:
        """Convenience constructor."""
        documents = []
        documents.extend(
            [
                Document(page_content=description, metadata={"name": name})
                for name, descriptions in names_and_descriptions
                for description in descriptions
            ]
        )
        vectorstore = await vectorstore_cls.afrom_documents(documents, embeddings)
        return cls(vectorstore=vectorstore, **kwargs)