invoke() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ invoke() — langchain Function Reference

Dependency Diagram

graph TD
  a9e8930f_e5d2_83a9_7c15_89d25e902358["invoke()"]
  17599172_8889_afc6_2237_4429f3439071["RunnableParallel"]
  a9e8930f_e5d2_83a9_7c15_89d25e902358 -->|defined in| 17599172_8889_afc6_2237_4429f3439071
  432a5d2e_5b75_7d42_3ab9_d76375f00ff9["get_name()"]
  a9e8930f_e5d2_83a9_7c15_89d25e902358 -->|calls| 432a5d2e_5b75_7d42_3ab9_d76375f00ff9
  255c479b_b9fa_44d8_4de5_2562051e06b5["get_name()"]
  a9e8930f_e5d2_83a9_7c15_89d25e902358 -->|calls| 255c479b_b9fa_44d8_4de5_2562051e06b5
  style a9e8930f_e5d2_83a9_7c15_89d25e902358 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 3834–3891

    def invoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any
    ) -> dict[str, Any]:
        # setup callbacks
        config = ensure_config(config)
        callback_manager = CallbackManager.configure(
            inheritable_callbacks=config.get("callbacks"),
            local_callbacks=None,
            verbose=False,
            inheritable_tags=config.get("tags"),
            local_tags=None,
            inheritable_metadata=config.get("metadata"),
            local_metadata=None,
        )
        # 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),
        )

        def _invoke_step(
            step: Runnable[Input, Any], input_: Input, config: RunnableConfig, key: str
        ) -> Any:
            child_config = patch_config(
                config,
                # mark each step as a child run
                callbacks=run_manager.get_child(f"map:key:{key}"),
            )
            with set_config_context(child_config) as context:
                return context.run(
                    step.invoke,
                    input_,
                    child_config,
                )

        # gather results from all steps
        try:
            # copy to avoid issues from the caller mutating the steps during invoke()
            steps = dict(self.steps__)

            with get_executor_for_config(config) as executor:
                futures = [
                    executor.submit(_invoke_step, step, input, config, key)
                    for key, step in steps.items()
                ]
                output = {
                    key: future.result()
                    for key, future in zip(steps, futures, strict=False)
                }
        # finish the root run
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise
        else:
            run_manager.on_chain_end(output)
            return output