_ainvoke() — langchain Function Reference

Architecture documentation for the _ainvoke() function in base.py from the langchain codebase.

Function python LangChainCore Runnables calls 1 called by 1

Entity Profile

LangChainCore→ Runnables→ _ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  4343d7c8_f1ee_8c72_b4a3_a26cae12cc7c["_ainvoke()"]
  c6a370e4_a02a_c56a_38eb_7ca734dcbfea["RunnableLambda"]
  4343d7c8_f1ee_8c72_b4a3_a26cae12cc7c -->|defined in| c6a370e4_a02a_c56a_38eb_7ca734dcbfea
  ecbf54bb_b4df_7f24_797a_3214e4425b2f["_ainvoke()"]
  ecbf54bb_b4df_7f24_797a_3214e4425b2f -->|calls| 4343d7c8_f1ee_8c72_b4a3_a26cae12cc7c
  ecbf54bb_b4df_7f24_797a_3214e4425b2f["_ainvoke()"]
  4343d7c8_f1ee_8c72_b4a3_a26cae12cc7c -->|calls| ecbf54bb_b4df_7f24_797a_3214e4425b2f
  style 4343d7c8_f1ee_8c72_b4a3_a26cae12cc7c fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 4896–4994

    async def _ainvoke(
        self,
        value: Input,
        run_manager: AsyncCallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> Output:
        if hasattr(self, "afunc"):
            afunc = self.afunc
        else:
            if inspect.isgeneratorfunction(self.func):

                def func(
                    value: Input,
                    run_manager: AsyncCallbackManagerForChainRun,
                    config: RunnableConfig,
                    **kwargs: Any,
                ) -> Output:
                    output: Output | None = None
                    for chunk in call_func_with_variable_args(
                        cast("Callable[[Input], Iterator[Output]]", self.func),
                        value,
                        config,
                        run_manager.get_sync(),
                        **kwargs,
                    ):
                        if output is None:
                            output = chunk
                        else:
                            try:
                                output = output + chunk  # type: ignore[operator]
                            except TypeError:
                                output = chunk
                    return cast("Output", output)

            else:

                def func(
                    value: Input,
                    run_manager: AsyncCallbackManagerForChainRun,
                    config: RunnableConfig,
                    **kwargs: Any,
                ) -> Output:
                    return call_func_with_variable_args(
                        self.func, value, config, run_manager.get_sync(), **kwargs
                    )

            @wraps(func)
            async def f(*args: Any, **kwargs: Any) -> Any:
                return await run_in_executor(config, func, *args, **kwargs)

            afunc = f

        if is_async_generator(afunc):
            output: Output | None = None
            async with aclosing(
                cast(
                    "AsyncGenerator[Any, Any]",
                    acall_func_with_variable_args(
                        cast("Callable", afunc),
                        value,
                        config,
                        run_manager,
                        **kwargs,
                    ),
                )
            ) as stream:
                async for chunk in cast(
                    "AsyncIterator[Output]",
                    stream,
                ):
                    if output is None:
                        output = chunk
                    else:
                        try:
                            output = output + chunk  # type: ignore[operator]
                        except TypeError:
                            output = chunk
        else:
            output = await acall_func_with_variable_args(
                cast("Callable", afunc), value, config, run_manager, **kwargs