ainvoke() — langchain Function Reference

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

Function python LangChainCore Runnables called by 1

Entity Profile

LangChainCore→ Runnables→ ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  8fe0517c_6208_0ec9_8513_64b9f8430708["ainvoke()"]
  8c770be5_e4c6_a7e0_4227_c2085f2c1314["RunnableBranch"]
  8fe0517c_6208_0ec9_8513_64b9f8430708 -->|defined in| 8c770be5_e4c6_a7e0_4227_c2085f2c1314
  e7b52750_c385_20a6_da26_fddbf6b4f003["astream()"]
  e7b52750_c385_20a6_da26_fddbf6b4f003 -->|calls| 8fe0517c_6208_0ec9_8513_64b9f8430708
  style 8fe0517c_6208_0ec9_8513_64b9f8430708 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/branch.py lines 248–293

    async def ainvoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any
    ) -> Output:
        config = ensure_config(config)
        callback_manager = get_async_callback_manager_for_config(config)
        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),
        )
        try:
            for idx, branch in enumerate(self.branches):
                condition, runnable = branch

                expression_value = await condition.ainvoke(
                    input,
                    config=patch_config(
                        config,
                        callbacks=run_manager.get_child(tag=f"condition:{idx + 1}"),
                    ),
                )

                if expression_value:
                    output = await runnable.ainvoke(
                        input,
                        config=patch_config(
                            config,
                            callbacks=run_manager.get_child(tag=f"branch:{idx + 1}"),
                        ),
                        **kwargs,
                    )
                    break
            else:
                output = await self.default.ainvoke(
                    input,
                    config=patch_config(
                        config, callbacks=run_manager.get_child(tag="branch:default")
                    ),
                    **kwargs,
                )
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise
        await run_manager.on_chain_end(output)
        return output