ainvoke() — langchain Function Reference

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

Function python LangChainCore ApiManagement calls 2

Entity Profile

LangChainCore→ ApiManagement→ ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  edbb3c53_e90a_2dce_9d3c_7242be164bc4["ainvoke()"]
  3a20478a_3692_141f_433b_a32429b00020["BaseRetriever"]
  edbb3c53_e90a_2dce_9d3c_7242be164bc4 -->|defined in| 3a20478a_3692_141f_433b_a32429b00020
  e3fe24f4_ee5d_0e58_6571_c387f6881443["_get_ls_params()"]
  edbb3c53_e90a_2dce_9d3c_7242be164bc4 -->|calls| e3fe24f4_ee5d_0e58_6571_c387f6881443
  97c862c5_9741_2754_a072_e4d2142e6354["_aget_relevant_documents()"]
  edbb3c53_e90a_2dce_9d3c_7242be164bc4 -->|calls| 97c862c5_9741_2754_a072_e4d2142e6354
  style edbb3c53_e90a_2dce_9d3c_7242be164bc4 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/retrievers.py lines 237–295

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

        Main entry point for asynchronous 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
        await retriever.ainvoke("query")
        ```
        """
        config = ensure_config(config)
        inheritable_metadata = {
            **(config.get("metadata") or {}),
            **self._get_ls_params(**kwargs),
        }
        callback_manager = AsyncCallbackManager.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 = await 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 = await self._aget_relevant_documents(
                    input, run_manager=run_manager, **kwargs_
                )
            else:
                result = await self._aget_relevant_documents(input, **kwargs_)
        except Exception as e:
            await run_manager.on_retriever_error(e)
            raise
        else:
            await run_manager.on_retriever_end(
                result,
            )
            return result