abatch() — langchain Function Reference

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

Function python LangChainCore Runnables calls 3 called by 4

Entity Profile

LangChainCore→ Runnables→ abatch() — langchain Function Reference

Dependency Diagram

graph TD
  3e736ea3_30ef_68c3_5f32_b8a49b37d501["abatch()"]
  4a62481c_02cb_a5de_1833_50669d5351a6["Runnable"]
  3e736ea3_30ef_68c3_5f32_b8a49b37d501 -->|defined in| 4a62481c_02cb_a5de_1833_50669d5351a6
  b9fb513b_bf75_3677_fc04_41bba780611c["abatch()"]
  b9fb513b_bf75_3677_fc04_41bba780611c -->|calls| 3e736ea3_30ef_68c3_5f32_b8a49b37d501
  0eae6b52_89cc_9d0f_4747_f69145024156["abatch()"]
  0eae6b52_89cc_9d0f_4747_f69145024156 -->|calls| 3e736ea3_30ef_68c3_5f32_b8a49b37d501
  0ceb56af_9929_a187_7b84_2584155e0d7b["pipe()"]
  0ceb56af_9929_a187_7b84_2584155e0d7b -->|calls| 3e736ea3_30ef_68c3_5f32_b8a49b37d501
  ecbf54bb_b4df_7f24_797a_3214e4425b2f["_ainvoke()"]
  ecbf54bb_b4df_7f24_797a_3214e4425b2f -->|calls| 3e736ea3_30ef_68c3_5f32_b8a49b37d501
  b48e63b1_c802_57f5_1432_79f235cc1161["ainvoke()"]
  3e736ea3_30ef_68c3_5f32_b8a49b37d501 -->|calls| b48e63b1_c802_57f5_1432_79f235cc1161
  cb51eea6_993b_93b2_5bc8_da36e6fac2df["map()"]
  3e736ea3_30ef_68c3_5f32_b8a49b37d501 -->|calls| cb51eea6_993b_93b2_5bc8_da36e6fac2df
  0eae6b52_89cc_9d0f_4747_f69145024156["abatch()"]
  3e736ea3_30ef_68c3_5f32_b8a49b37d501 -->|calls| 0eae6b52_89cc_9d0f_4747_f69145024156
  style 3e736ea3_30ef_68c3_5f32_b8a49b37d501 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 1002–1048

    async def abatch(
        self,
        inputs: list[Input],
        config: RunnableConfig | list[RunnableConfig] | None = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Any | None,
    ) -> list[Output]:
        """Default implementation runs `ainvoke` in parallel using `asyncio.gather`.

        The default implementation of `batch` works well for IO bound runnables.

        Subclasses must override this method if they can batch more efficiently;
        e.g., if the underlying `Runnable` uses an API which supports a batch mode.

        Args:
            inputs: A list of inputs to the `Runnable`.
            config: A config to use when invoking the `Runnable`.

                The config supports standard keys like `'tags'`, `'metadata'` for
                tracing purposes, `'max_concurrency'` for controlling how much work to
                do in parallel, and other keys.

                Please refer to `RunnableConfig` for more details.
            return_exceptions: Whether to return exceptions instead of raising them.
            **kwargs: Additional keyword arguments to pass to the `Runnable`.

        Returns:
            A list of outputs from the `Runnable`.

        """
        if not inputs:
            return []

        configs = get_config_list(config, len(inputs))

        async def ainvoke(value: Input, config: RunnableConfig) -> Output | Exception:
            if return_exceptions:
                try:
                    return await self.ainvoke(value, config, **kwargs)
                except Exception as e:
                    return e
            else:
                return await self.ainvoke(value, config, **kwargs)

        coros = map(ainvoke, inputs, configs)
        return await gather_with_concurrency(configs[0].get("max_concurrency"), *coros)