ainvoke() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  3a58a06d_fe7f_61d1_c630_542cb9f85c27["ainvoke()"]
  17599172_8889_afc6_2237_4429f3439071["RunnableParallel"]
  3a58a06d_fe7f_61d1_c630_542cb9f85c27 -->|defined in| 17599172_8889_afc6_2237_4429f3439071
  432a5d2e_5b75_7d42_3ab9_d76375f00ff9["get_name()"]
  3a58a06d_fe7f_61d1_c630_542cb9f85c27 -->|calls| 432a5d2e_5b75_7d42_3ab9_d76375f00ff9
  255c479b_b9fa_44d8_4de5_2562051e06b5["get_name()"]
  3a58a06d_fe7f_61d1_c630_542cb9f85c27 -->|calls| 255c479b_b9fa_44d8_4de5_2562051e06b5
  style 3a58a06d_fe7f_61d1_c630_542cb9f85c27 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 3894–3946

    async def ainvoke(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> dict[str, Any]:
        # setup callbacks
        config = ensure_config(config)
        callback_manager = get_async_callback_manager_for_config(config)
        # start the root run
        run_manager = await callback_manager.on_chain_start(
            None,
            input,
            name=config.get("run_name") or self.get_name(),
            run_id=config.pop("run_id", None),
        )

        async def _ainvoke_step(
            step: Runnable[Input, Any], input_: Input, config: RunnableConfig, key: str
        ) -> Any:
            child_config = patch_config(
                config,
                callbacks=run_manager.get_child(f"map:key:{key}"),
            )
            with set_config_context(child_config) as context:
                return await coro_with_context(
                    step.ainvoke(input_, child_config), context, create_task=True
                )

        # gather results from all steps
        try:
            # copy to avoid issues from the caller mutating the steps during invoke()
            steps = dict(self.steps__)
            results = await asyncio.gather(
                *(
                    _ainvoke_step(
                        step,
                        input,
                        # mark each step as a child run
                        config,
                        key,
                    )
                    for key, step in steps.items()
                )
            )
            output = dict(zip(steps, results, strict=False))
        # finish the root run
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise
        else:
            await run_manager.on_chain_end(output)
            return output