invoke() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ invoke() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 3131–3164

    def invoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any
    ) -> Output:
        # setup callbacks and context
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)
        # start the root run
        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),
        )
        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:
                        input_ = context.run(step.invoke, input_, config, **kwargs)
                    else:
                        input_ = context.run(step.invoke, input_, config)
        # finish the root run
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise
        else:
            run_manager.on_chain_end(input_)
            return cast("Output", input_)