deepset-ai/haystack · error · ValueError

concurrency_limit must be greater than or equal to 1.

Error message

concurrency_limit must be greater than or equal to 1.

What it means

run_async_generator (pipeline.py:889) validates its concurrency_limit argument before running the pipeline. A value below 1 would deadlock or misbehave in the asyncio task scheduling, so a plain ValueError is raised. run_async forwards its concurrency_limit here, so both entry points can trigger it.

Source

Thrown at haystack/core/pipeline/pipeline.py:889

        :param concurrency_limit: The maximum number of components that are allowed to run concurrently.
        :param include_outputs_from:
            Set of component names whose individual outputs are to be
            included in the pipeline's output. For components that are
            invoked multiple times (in a loop), only the last-produced
            output is included.
        :return: An async iterator containing partial (and final) outputs.

        :raises ValueError:
            If invalid inputs are provided to the pipeline, or if `concurrency_limit` is less than 1.
        :raises PipelineMaxComponentRuns:
            If a component exceeds the maximum number of allowed executions within the pipeline.
        :raises PipelineRuntimeError:
            If the Pipeline contains cycles with unsupported connections that would cause
            it to get stuck and fail running.
            Or if a Component fails or returns output in an unsupported type.
        """
        if concurrency_limit < 1:
            raise ValueError("concurrency_limit must be greater than or equal to 1.")

        pipeline_running(self)  # telemetry

        # warm up the pipeline by running each component's warm_up_async (or warm_up) method
        await self.warm_up_async()

        if include_outputs_from is None:
            include_outputs_from = set()

        pipeline_outputs: dict[str, Any] = {}

        # Normalize `data` and raise ValueError if the input is malformed in some way.
        data = self._prepare_component_input_data(data)

        # Raise ValueError if input is malformed in some way
        self.validate_input(data)

        # We create a list of components in the pipeline sorted by name, so that the algorithm runs

View on GitHub (pinned to e318778c9b)

Solutions

  1. Pass concurrency_limit >= 1 (e.g. 1 for sequential execution).
  2. If the value comes from config, clamp it: max(1, configured_value).
  3. If 'unlimited' was intended, omit the argument and use the default instead of 0.

Example fix

# before
await pipe.run_async(data, concurrency_limit=0)
# after
await pipe.run_async(data, concurrency_limit=max(1, configured_limit))
Defensive patterns

Strategy: validation

Validate before calling

if concurrency_limit is not None and concurrency_limit < 1:
    raise ValueError('concurrency_limit must be >= 1')

Try / catch

try:
    agen = pipe.run_async(data, concurrency_limit=cfg_limit)
except ValueError as e:
    if 'concurrency_limit' in str(e):
        agen = pipe.run_async(data, concurrency_limit=max(1, cfg_limit))

Prevention

When it happens

Trigger: pipe.run_async(data, concurrency_limit=0) or concurrency_limit=-1, often from a config value or computed value like max(len(workers), 0) when a list is empty.

Common situations: Config file where concurrency is set to 0 to mean 'unlimited' (it does not); dynamic sizing from an empty collection; typos in defaults.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

AI-assisted analysis of deepset-ai/haystack@e318778c9b (2026-08-30). Data as JSON: /api/errors/afb9701a6051878b. Report an issue: GitHub.