invoke() — langchain Function Reference

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

Function python AgentOrchestration ClassicChains calls 4 called by 1

Entity Profile

AgentOrchestration→ ClassicChains→ invoke() — langchain Function Reference

Dependency Diagram

graph TD
  cbab652e_92a7_bf56_f96d_801fd5f6dacd["invoke()"]
  f3cef70e_11b0_61c9_7ec0_7308f4b45056["Chain"]
  cbab652e_92a7_bf56_f96d_801fd5f6dacd -->|defined in| f3cef70e_11b0_61c9_7ec0_7308f4b45056
  b350c2bd_ec11_d5c0_502b_65a6d1ee41a2["__call__()"]
  b350c2bd_ec11_d5c0_502b_65a6d1ee41a2 -->|calls| cbab652e_92a7_bf56_f96d_801fd5f6dacd
  d66dd3df_8d8d_1576_bf19_2f49fc27b7ec["prep_inputs()"]
  cbab652e_92a7_bf56_f96d_801fd5f6dacd -->|calls| d66dd3df_8d8d_1576_bf19_2f49fc27b7ec
  1fae3e72_380c_c210_1baa_d5fe1506cde1["_validate_inputs()"]
  cbab652e_92a7_bf56_f96d_801fd5f6dacd -->|calls| 1fae3e72_380c_c210_1baa_d5fe1506cde1
  70ced93c_1678_5151_e7bb_a3789f5e33b1["_call()"]
  cbab652e_92a7_bf56_f96d_801fd5f6dacd -->|calls| 70ced93c_1678_5151_e7bb_a3789f5e33b1
  e60bffb7_c7fe_f9f9_1671_b444e79457ef["prep_outputs()"]
  cbab652e_92a7_bf56_f96d_801fd5f6dacd -->|calls| e60bffb7_c7fe_f9f9_1671_b444e79457ef
  style cbab652e_92a7_bf56_f96d_801fd5f6dacd fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/base.py lines 131–184

    def invoke(
        self,
        input: dict[str, Any],
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> dict[str, Any]:
        config = ensure_config(config)
        callbacks = config.get("callbacks")
        tags = config.get("tags")
        metadata = config.get("metadata")
        run_name = config.get("run_name") or self.get_name()
        run_id = config.get("run_id")
        include_run_info = kwargs.get("include_run_info", False)
        return_only_outputs = kwargs.get("return_only_outputs", False)

        inputs = self.prep_inputs(input)
        callback_manager = CallbackManager.configure(
            callbacks,
            self.callbacks,
            self.verbose,
            tags,
            self.tags,
            metadata,
            self.metadata,
        )
        new_arg_supported = inspect.signature(self._call).parameters.get("run_manager")

        run_manager = callback_manager.on_chain_start(
            None,
            inputs,
            run_id,
            name=run_name,
        )
        try:
            self._validate_inputs(inputs)
            outputs = (
                self._call(inputs, run_manager=run_manager)
                if new_arg_supported
                else self._call(inputs)
            )

            final_outputs: dict[str, Any] = self.prep_outputs(
                inputs,
                outputs,
                return_only_outputs,
            )
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise
        run_manager.on_chain_end(outputs)

        if include_run_info:
            final_outputs[RUN_KEY] = RunInfo(run_id=run_manager.run_id)
        return final_outputs