Skip to content

Write an Algorithm

The bundled LoadBalancingAssignmentAlgorithm is a short reference. Copy it as a starting point for your own:

from beekeeper import (
    Algorithm,
    AnyEntity,
    AnyRequest,
    Assignment,
    AssignmentState,
)


class MyAlgorithm[TEntity: AnyEntity, TAllocReq: AnyRequest](
    Algorithm[TEntity, TAllocReq],
):
    def run(self, allocations, entities, candidates, rules):
        state: AssignmentState[TEntity, TAllocReq] = AssignmentState()
        rules_list = list(rules)

        for allocation in allocations:
            # The bundled load-balancing reference sorts by score / (1 + load) so
            # already-assigned entities are scored down. A pure-greedy variant
            # would just sort by `c.score` instead.
            ranked = sorted(
                candidates.get(id(allocation), []),
                key=lambda c: c.score / (1 + len(state.get_assignments_done_by(c.entity))),
                reverse=True,
            )
            chosen = []
            for c in ranked:
                if len(chosen) >= allocation.required_count:
                    break
                if all(r.evaluate(c.entity, allocation, state).compatible for r in rules_list):
                    chosen.append(c.entity)
            if len(chosen) == allocation.required_count:
                state.add_assignment(
                    Assignment(allocation=allocation, assigned_entities=tuple(chosen))
                )
        return state

Bundled implementations

Three reference implementations live under beekeeper.algorithm.implementations.*. Pick the one closest to what your domain needs and copy or wrap; or write a fresh implementation against the same Algorithm contract.

Module Class Use when
load_balancing LoadBalancingAssignmentAlgorithm Default reference. Picks the highest-scored compatible candidates with a per-entity load penalty (score / (1 + load)) so work disperses across the pool. Deterministic. Never raises.
backtracking BacktrackingAssignmentAlgorithm Stateful rules + constrained candidate pools. Tries alternative orderings to find a complete assignment; raises IncompleteSolutionError if it can't, so a chain like [backtracking, load_balancing] falls back gracefully. Has a configurable top-K cap and iteration budget.
or_tools OrToolsAssignmentAlgorithm Globally optimal under the modeled constraints. Heaviest dep (~50 MB); requires the optional ortools extra (uv sync --extra ortools in the repo, uv add 'beekeeper[ortools]' from another project, or pip install 'beekeeper[ortools]'). Stateful rules are not encoded into the CP-SAT formulation — use backtracking if you need them. Raises IncompleteSolutionError on INFEASIBLE / MODEL_INVALID.

See The Algorithm Contract for what your run(...) is allowed to assume and required to return.