SelfQueryRetriever Class — langchain Architecture

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

Class python

Entity Profile

Dependency Diagram

graph TD
  6a182800_8459_6acf_626f_f203297062ee["SelfQueryRetriever"]
  2a401977_bd56_ea94_9c8f_d0b77072baae["BaseRetriever"]
  6a182800_8459_6acf_626f_f203297062ee -->|extends| 2a401977_bd56_ea94_9c8f_d0b77072baae
  78f4cfa8_800d_7358_1d94_fbcb78a17327["base.py"]
  6a182800_8459_6acf_626f_f203297062ee -->|defined in| 78f4cfa8_800d_7358_1d94_fbcb78a17327
  3b9abb98_3b01_3c02_b145_53f4c9e5730a["validate_translator()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| 3b9abb98_3b01_3c02_b145_53f4c9e5730a
  a27d2a8e_4179_fdbc_2caa_e6d89553add0["llm_chain()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| a27d2a8e_4179_fdbc_2caa_e6d89553add0
  97e98348_d8ff_40f3_78ae_a8e9b56bb8b8["_prepare_query()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| 97e98348_d8ff_40f3_78ae_a8e9b56bb8b8
  340efb6a_9c6f_53a0_973f_988be54c5a2f["_get_docs_with_query()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| 340efb6a_9c6f_53a0_973f_988be54c5a2f
  4c67a05f_f2a3_d7f6_8e71_18fa6dbfb021["_aget_docs_with_query()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| 4c67a05f_f2a3_d7f6_8e71_18fa6dbfb021
  6b1749c7_60d9_33e1_f5a5_2e8d91544a1a["_get_relevant_documents()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| 6b1749c7_60d9_33e1_f5a5_2e8d91544a1a
  d3d9414c_ca11_a2a6_1127_f35fc0ffcea1["_aget_relevant_documents()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| d3d9414c_ca11_a2a6_1127_f35fc0ffcea1
  6840f078_21af_3411_ed45_36262f0aa505["from_llm()"]
  6a182800_8459_6acf_626f_f203297062ee -->|method| 6840f078_21af_3411_ed45_36262f0aa505

Relationship Graph

Source Code

libs/langchain/langchain_classic/retrievers/self_query/base.py lines 237–407

class SelfQueryRetriever(BaseRetriever):
    """Self Query Retriever.

    Retriever that uses a vector store and an LLM to generate the vector store queries.
    """

    vectorstore: VectorStore
    """The underlying vector store from which documents will be retrieved."""
    query_constructor: Runnable[dict, StructuredQuery] = Field(alias="llm_chain")
    """The query constructor chain for generating the vector store queries.

    llm_chain is legacy name kept for backwards compatibility."""
    search_type: str = "similarity"
    """The search type to perform on the vector store."""
    search_kwargs: dict = Field(default_factory=dict)
    """Keyword arguments to pass in to the vector store search."""
    structured_query_translator: Visitor
    """Translator for turning internal query language into `VectorStore` search params."""  # noqa: E501
    verbose: bool = False

    use_original_query: bool = False
    """Use original query instead of the revised new query from LLM"""

    model_config = ConfigDict(
        populate_by_name=True,
        arbitrary_types_allowed=True,
    )

    @model_validator(mode="before")
    @classmethod
    def validate_translator(cls, values: dict) -> Any:
        """Validate translator."""
        if "structured_query_translator" not in values:
            values["structured_query_translator"] = _get_builtin_translator(
                values["vectorstore"],
            )
        return values

    @property
    def llm_chain(self) -> Runnable:
        """llm_chain is legacy name kept for backwards compatibility."""
        return self.query_constructor

    def _prepare_query(
        self,
        query: str,
        structured_query: StructuredQuery,
    ) -> tuple[str, dict[str, Any]]:
        new_query, new_kwargs = self.structured_query_translator.visit_structured_query(
            structured_query,
        )
        if structured_query.limit is not None:
            new_kwargs["k"] = structured_query.limit
        if self.use_original_query:
            new_query = query
        search_kwargs = {**self.search_kwargs, **new_kwargs}
        return new_query, search_kwargs

    def _get_docs_with_query(
        self,
        query: str,
        search_kwargs: dict[str, Any],
    ) -> list[Document]:
        return self.vectorstore.search(query, self.search_type, **search_kwargs)

    async def _aget_docs_with_query(
        self,
        query: str,
        search_kwargs: dict[str, Any],
    ) -> list[Document]:
        return await self.vectorstore.asearch(query, self.search_type, **search_kwargs)

    @override
    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> list[Document]:
        structured_query = self.query_constructor.invoke(
            {"query": query},