batch() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ batch() — langchain Function Reference

Dependency Diagram

graph TD
  c9737a6d_ee16_ffc7_0ee4_5b2b93274f59["batch()"]
  65b9aaca_53f2_847d_9ab7_024fe53cbbd2["RouterRunnable"]
  c9737a6d_ee16_ffc7_0ee4_5b2b93274f59 -->|defined in| 65b9aaca_53f2_847d_9ab7_024fe53cbbd2
  7ef8f2a4_9aa2_9baa_0143_41c8a576c151["invoke()"]
  c9737a6d_ee16_ffc7_0ee4_5b2b93274f59 -->|calls| 7ef8f2a4_9aa2_9baa_0143_41c8a576c151
  style c9737a6d_ee16_ffc7_0ee4_5b2b93274f59 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/router.py lines 136–170

    def batch(
        self,
        inputs: list[RouterInput],
        config: RunnableConfig | list[RunnableConfig] | None = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Any | None,
    ) -> list[Output]:
        if not inputs:
            return []

        keys = [input_["key"] for input_ in inputs]
        actual_inputs = [input_["input"] for input_ in inputs]
        if any(key not in self.runnables for key in keys):
            msg = "One or more keys do not have a corresponding runnable"
            raise ValueError(msg)

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

        runnables = [self.runnables[key] for key in keys]
        configs = get_config_list(config, len(inputs))
        with get_executor_for_config(configs[0]) as executor:
            return cast(
                "list[Output]",
                list(executor.map(invoke, runnables, actual_inputs, configs)),
            )