_invoke() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1 called by 1

Entity Profile

LangChainCore→ Runnables→ _invoke() — langchain Function Reference

Dependency Diagram

graph TD
  15bcbb64_a12d_e5ab_35ce_209867a93c74["_invoke()"]
  c6a370e4_a02a_c56a_38eb_7ca734dcbfea["RunnableLambda"]
  15bcbb64_a12d_e5ab_35ce_209867a93c74 -->|defined in| c6a370e4_a02a_c56a_38eb_7ca734dcbfea
  d7c8b78b_bd8b_c15c_ce21_97525a7a5764["_invoke()"]
  d7c8b78b_bd8b_c15c_ce21_97525a7a5764 -->|calls| 15bcbb64_a12d_e5ab_35ce_209867a93c74
  d7c8b78b_bd8b_c15c_ce21_97525a7a5764["_invoke()"]
  15bcbb64_a12d_e5ab_35ce_209867a93c74 -->|calls| d7c8b78b_bd8b_c15c_ce21_97525a7a5764
  style 15bcbb64_a12d_e5ab_35ce_209867a93c74 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 4851–4894

    def _invoke(
        self,
        input_: Input,
        run_manager: CallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> Output:
        if inspect.isgeneratorfunction(self.func):
            output: Output | None = None
            for chunk in call_func_with_variable_args(
                cast("Callable[[Input], Iterator[Output]]", self.func),
                input_,
                config,
                run_manager,
                **kwargs,
            ):
                if output is None:
                    output = chunk
                else:
                    try:
                        output = output + chunk  # type: ignore[operator]
                    except TypeError:
                        output = chunk
        else:
            output = call_func_with_variable_args(
                self.func, input_, config, run_manager, **kwargs
            )
        # If the output is a Runnable, invoke it
        if isinstance(output, Runnable):
            recursion_limit = config["recursion_limit"]
            if recursion_limit <= 0:
                msg = (
                    f"Recursion limit reached when invoking {self} with input {input_}."
                )
                raise RecursionError(msg)
            output = output.invoke(
                input_,
                patch_config(
                    config,
                    callbacks=run_manager.get_child(),
                    recursion_limit=recursion_limit - 1,
                ),
            )
        return cast("Output", output)