ainvoke() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  ceb9be43_8736_64b5_bd2d_654b803a8bed["ainvoke()"]
  c4bf8d59_69f9_f8e5_efba_837c20df2360["RunnableSequence"]
  ceb9be43_8736_64b5_bd2d_654b803a8bed -->|defined in| c4bf8d59_69f9_f8e5_efba_837c20df2360
  255c479b_b9fa_44d8_4de5_2562051e06b5["get_name()"]
  ceb9be43_8736_64b5_bd2d_654b803a8bed -->|calls| 255c479b_b9fa_44d8_4de5_2562051e06b5
  style ceb9be43_8736_64b5_bd2d_654b803a8bed fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 3167–3204

    async def ainvoke(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Output:
        # setup callbacks and context
        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),
        )
        input_ = input

        # invoke all steps in sequence
        try:
            for i, step in enumerate(self.steps):
                # mark each step as a child run
                config = patch_config(
                    config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
                )
                with set_config_context(config) as context:
                    if i == 0:
                        part = functools.partial(step.ainvoke, input_, config, **kwargs)
                    else:
                        part = functools.partial(step.ainvoke, input_, config)
                    input_ = await coro_with_context(part(), context, create_task=True)
            # finish the root run
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise
        else:
            await run_manager.on_chain_end(input_)
            return cast("Output", input_)