invoke() — langchain Function Reference

Architecture documentation for the invoke() function in retrievers.py from the langchain codebase.

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ invoke() — langchain Function Reference

Dependency Diagram

graph TD
  a0f2c9ac_e425_a622_2050_94dc89a5be6e["invoke()"]
  3a20478a_3692_141f_433b_a32429b00020["BaseRetriever"]
  a0f2c9ac_e425_a622_2050_94dc89a5be6e -->|defined in| 3a20478a_3692_141f_433b_a32429b00020
  e3fe24f4_ee5d_0e58_6571_c387f6881443["_get_ls_params()"]
  a0f2c9ac_e425_a622_2050_94dc89a5be6e -->|calls| e3fe24f4_ee5d_0e58_6571_c387f6881443
  8df94cd4_6dfb_8931_3b27_e1b97e7625b7["_get_relevant_documents()"]
  a0f2c9ac_e425_a622_2050_94dc89a5be6e -->|calls| 8df94cd4_6dfb_8931_3b27_e1b97e7625b7
  style a0f2c9ac_e425_a622_2050_94dc89a5be6e fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/retrievers.py lines 179–234

    def invoke(
        self, input: str, config: RunnableConfig | None = None, **kwargs: Any
    ) -> list[Document]:
        """Invoke the retriever to get relevant documents.

        Main entry point for synchronous retriever invocations.

        Args:
            input: The query string.
            config: Configuration for the retriever.
            **kwargs: Additional arguments to pass to the retriever.

        Returns:
            List of relevant documents.

        Examples:
        ```python
        retriever.invoke("query")
        ```
        """
        config = ensure_config(config)
        inheritable_metadata = {
            **(config.get("metadata") or {}),
            **self._get_ls_params(**kwargs),
        }
        callback_manager = CallbackManager.configure(
            config.get("callbacks"),
            None,
            verbose=kwargs.get("verbose", False),
            inheritable_tags=config.get("tags"),
            local_tags=self.tags,
            inheritable_metadata=inheritable_metadata,
            local_metadata=self.metadata,
        )
        run_manager = callback_manager.on_retriever_start(
            None,
            input,
            name=config.get("run_name") or self.get_name(),
            run_id=kwargs.pop("run_id", None),
        )
        try:
            kwargs_ = kwargs if self._expects_other_args else {}
            if self._new_arg_supported:
                result = self._get_relevant_documents(
                    input, run_manager=run_manager, **kwargs_
                )
            else:
                result = self._get_relevant_documents(input, **kwargs_)
        except Exception as e:
            run_manager.on_retriever_error(e)
            raise
        else:
            run_manager.on_retriever_end(
                result,
            )
            return result