xai-org/x-algorithm · error · ValueError

No generators provided

Error message

No generators provided

What it means

PriorityTaskGenerator multiplexes several weighted child generators; with zero children there is nothing to poll, so its constructor refuses an empty list immediately with ValueError.

Source

Thrown at grox/core/generators/task_generator.py:73

    @abstractmethod
    def _poll(self) -> AsyncGenerator[TaskPayload | None, None]:
        pass

    async def ack(self, result: TaskResult):
        pass

    def identify_task_origin(self, result: TaskResult) -> str | None:
        return self.TASK_GENERATOR_TYPE

    def on_terminal_failure(self, result: TaskResult) -> None:
        pass


class PriorityTaskGenerator(TaskGenerator):
    def __init__(self, generators: list[tuple[TaskGenerator, int]]):
        if not generators:
            raise ValueError("No generators provided")
        if any(weight <= 0 for _, weight in generators):
            raise ValueError("All weights must be positive")
        super().__init__(None)
        self._generators: dict[str, TaskGenerator] = {}
        self._weights: dict[str, int] = {}
        for i, (gen, weight) in enumerate(generators):
            label = f"GEN_{i}"
            self._generators[label] = gen
            self._weights[label] = weight
        self._result_cache: dict[str, str] = {}
        logger.info(
            f"Initialized priority task generator with {list(zip(self._generators.keys(), [gen.__class__.__name__ for gen in self._generators.values()], self._weights.values(), strict=True))}"
        )

    async def start(self) -> None:
        logger.info("Starting priority task generators")
        await asyncio.gather(*[gen.start() for gen in self._generators.values()])
        self._streams = {label: gen.poll() for label, gen in self._generators.items()}

View on GitHub (pinned to 24c60942c5)

Solutions

  1. Guard before construction: if the candidate list is empty, skip creating the PriorityTaskGenerator or supply a default generator.
  2. Fix the filtering/config logic that produced an empty list.
  3. Ensure config arrays contain at least one enabled generator entry.

Example fix

# before
pg = PriorityTaskGenerator([])

# after
gens = [(g, w) for g, w in configured if g.enabled]
if not gens:
    raise ValueError("config must enable at least one generator")
pg = PriorityTaskGenerator(gens)
Defensive patterns

Strategy: validation

Validate before calling

if not generators:
    raise ValueError("need at least one (generator, weight) pair")
pg = PriorityTaskGenerator(generators)

Type guard

def is_nonempty_pairs(generators: list) -> bool:
    return isinstance(generators, list) and len(generators) > 0

Try / catch

try:
    pg = PriorityTaskGenerator(gens)
except ValueError as e:
    if "No generators provided" in str(e):
        pg = PriorityTaskGenerator([(default_gen(), 1)])
    else:
        raise

Prevention

When it happens

Trigger: Constructing PriorityTaskGenerator(generators=[]) or with a list comprehension that filtered everything out; passing a programmatically-built list that happens to be empty.

Common situations: Dynamically assembling generators from config where all entries were disabled/filtered; empty config arrays; tests constructing the class without children.

Related errors


AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/80b11b805ae88759. Report an issue: GitHub.