invoke() — langchain Function Reference

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

Function python LangChainCore Runnables called by 1

Entity Profile

LangChainCore→ Runnables→ invoke() — langchain Function Reference

Dependency Diagram

graph TD
  9f958f06_98f9_3ba8_4239_f2c7fb815242["invoke()"]
  8c770be5_e4c6_a7e0_4227_c2085f2c1314["RunnableBranch"]
  9f958f06_98f9_3ba8_4239_f2c7fb815242 -->|defined in| 8c770be5_e4c6_a7e0_4227_c2085f2c1314
  b69050e0_7f2d_540b_c705_85c3001d53a4["stream()"]
  b69050e0_7f2d_540b_c705_85c3001d53a4 -->|calls| 9f958f06_98f9_3ba8_4239_f2c7fb815242
  style 9f958f06_98f9_3ba8_4239_f2c7fb815242 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/branch.py lines 189–245

    def invoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any
    ) -> Output:
        """First evaluates the condition, then delegate to `True` or `False` branch.

        Args:
            input: The input to the `Runnable`.
            config: The configuration for the `Runnable`.
            **kwargs: Additional keyword arguments to pass to the `Runnable`.

        Returns:
            The output of the branch that was run.
        """
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)
        run_manager = 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 = condition.invoke(
                    input,
                    config=patch_config(
                        config,
                        callbacks=run_manager.get_child(tag=f"condition:{idx + 1}"),
                    ),
                )

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