batch() — langchain Function Reference

Architecture documentation for the batch() function in configurable.py from the langchain codebase.

Function python LangChainCore Runnables calls 2

Entity Profile

LangChainCore→ Runnables→ batch() — langchain Function Reference

Dependency Diagram

graph TD
  07a943d3_3320_1d24_5872_3ef47a0d5f77["batch()"]
  018f968f_f1d1_6f85_05f3_0ff4abc12150["DynamicRunnable"]
  07a943d3_3320_1d24_5872_3ef47a0d5f77 -->|defined in| 018f968f_f1d1_6f85_05f3_0ff4abc12150
  e71e4d88_aaa7_d3d0_06ac_43f60d5cd48a["prepare()"]
  07a943d3_3320_1d24_5872_3ef47a0d5f77 -->|calls| e71e4d88_aaa7_d3d0_06ac_43f60d5cd48a
  f91bd2c5_5084_1b3b_a187_2f3ccbca8f0b["invoke()"]
  07a943d3_3320_1d24_5872_3ef47a0d5f77 -->|calls| f91bd2c5_5084_1b3b_a187_2f3ccbca8f0b
  style 07a943d3_3320_1d24_5872_3ef47a0d5f77 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/configurable.py lines 156–196

    def batch(
        self,
        inputs: list[Input],
        config: RunnableConfig | list[RunnableConfig] | None = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Any | None,
    ) -> list[Output]:
        configs = get_config_list(config, len(inputs))
        prepared = [self.prepare(c) for c in configs]

        if all(p is self.default for p, _ in prepared):
            return self.default.batch(
                inputs,
                [c for _, c in prepared],
                return_exceptions=return_exceptions,
                **kwargs,
            )

        if not inputs:
            return []

        def invoke(
            prepared: tuple[Runnable[Input, Output], RunnableConfig],
            input_: Input,
        ) -> Output | Exception:
            bound, config = prepared
            if return_exceptions:
                try:
                    return bound.invoke(input_, config, **kwargs)
                except Exception as e:
                    return e
            else:
                return bound.invoke(input_, config, **kwargs)

        # If there's only one input, don't bother with the executor
        if len(inputs) == 1:
            return cast("list[Output]", [invoke(prepared[0], inputs[0])])

        with get_executor_for_config(configs[0]) as executor:
            return cast("list[Output]", list(executor.map(invoke, prepared, inputs)))