Algorithm Search⌘K

Karger-Stein algorithm in Python

Recursive random contraction improving Karger's running time to near O(n^2).

File
karger_stein.py, 116 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
karger_stein.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/karger-stein/. The authors are listed in prov/QPM3H8.jsonl.

import itertools
import math
import random


def karger_stein(
    n: int,
    edges: list[tuple[int, int, float]],
    runs: int | None = None,
    rng: random.Random | None = None,
) -> tuple[float, set[int]]:
    """Global minimum cut of an undirected graph by recursive contraction.

    Vertices are 0 .. n-1 with n >= 2, and edges (u, v, w) are undirected with
    weight w >= 0, an integer or a real number. Parallel edges add up and
    self-loops are ignored. Returns (value, S): S is a nonempty proper subset of
    the vertices and value is the exact weight of its cut. The value is a
    minimum with high probability; it is never below the minimum. The default
    number of runs is ceil(log2(n))^2 + 1.
    """
    if n < 2:
        raise ValueError("need at least two vertices")
    rng = rng or random.Random()
    w = [[0] * n for _ in range(n)]
    for u, v, c in edges:
        if c < 0:
            raise ValueError("weights must be nonnegative")
        if u != v:
            w[u][v] += c
            w[v][u] += c
    if runs is None:
        runs = math.ceil(math.log2(n)) ** 2 + 1
    groups = [frozenset([v]) for v in range(n)]
    best, best_set = None, set()  # @line L1
    for _ in range(runs):  # @line L2
        value, side = recursive_contract([row[:] for row in w], list(groups), rng)  # @line L3
        if best is None or value < best:
            best, best_set = value, set(side)
    return best, best_set  # @line L4


def recursive_contract(w: list[list[float]], groups: list[frozenset[int]], rng: random.Random) -> tuple[float, frozenset[int]]:
    """One run on the graph with weight matrix w; vertex i stands for groups[i]."""
    k = len(groups)  # @line L5
    if k <= 6:  # @line L6
        return brute_force_cut(w, groups)
    t = math.ceil(1 + k / math.sqrt(2))  # @line L7
    results = []
    for _ in range(2):  # @line L8
        w2, g2 = contract(w, groups, t, rng)
        if w2 is None:  # no edge left: g2 is a side whose cut has weight 0
            results.append((0, g2))
        else:
            results.append(recursive_contract(w2, g2, rng))
    return min(results, key=lambda r: r[0])  # @line L9


def contract(w, groups, t, rng):
    """Contract random edges, chosen with probability proportional to weight, down to t vertices.

    Returns a new (matrix, groups). If no edge of positive weight is left while
    more than t vertices remain, the graph is disconnected and the group of any
    vertex is a cut of weight 0; then (None, that group) is returned.
    """
    w = [row[:] for row in w]
    groups = list(groups)
    alive = list(range(len(groups)))
    deg = [sum(row) for row in w]
    while len(alive) > t:  # @line L10
        total = sum(deg[v] for v in alive)
        if total == 0:
            return None, groups[alive[0]]
        u = pick(alive, [deg[v] for v in alive], total, rng)  # @line L11
        v = pick(alive, [w[u][x] for x in alive], deg[u], rng)
        groups[u] = groups[u] | groups[v]
        for x in alive:
            w[u][x] += w[v][x]
            w[x][u] = w[u][x]
        deg[u] += deg[v] - 2 * w[u][u]
        w[u][u] = 0
        alive.remove(v)
    w2 = [[w[a][b] for b in alive] for a in alive]  # @line L12
    return w2, [groups[a] for a in alive]


def pick(items, weights, total, rng):
    """An item chosen with probability weight / total; total is positive."""
    if all(isinstance(x, int) for x in weights):
        r = rng.randrange(total)
    else:
        r = rng.random() * total
    acc = 0
    last = None
    for item, x in zip(items, weights):
        if x > 0:
            acc += x
            last = item
            if r < acc:
                return item
    return last


def brute_force_cut(w, groups):
    """The exact minimum cut of a graph with at most six vertices."""
    k = len(groups)
    best, best_side = None, None
    for r in range(k - 1):
        for extra in itertools.combinations(range(1, k), r):
            side = {0, *extra}
            value = sum(w[a][b] for a in side for b in range(k) if b not in side)
            if best is None or value < best:
                best, best_side = value, side
    return best, frozenset().union(*(groups[a] for a in best_side))
test_karger_stein.py 92 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/karger-stein/. The authors are listed in prov/QPM3H8.jsonl.

import itertools
import random
from fractions import Fraction

import pytest

from karger_stein import karger_stein, recursive_contract

# a, b, c, d, e, f = 0 .. 5 and g, h = 6, 7: two 4-cliques of weight 2, abcg and defh, joined by cd and be
EXAMPLE = (
    [(a, b, 2) for a, b in itertools.combinations([0, 1, 2, 6], 2)]
    + [(a, b, 2) for a, b in itertools.combinations([3, 4, 5, 7], 2)]
    + [(2, 3, 1), (1, 4, 1)]
)


def cut_value(edges, side):
    return sum(c for u, v, c in edges if (u in side) != (v in side))


def brute(n, edges):
    """Independent oracle: the least cut over all nonempty proper subsets holding vertex 0."""
    best = None
    for r in range(n - 1):
        for extra in itertools.combinations(range(1, n), r):
            c = cut_value(edges, {0, *extra})
            best = c if best is None else min(best, c)
    return best


def check(n, edges, value, side):
    assert 0 < len(side) < n
    assert cut_value(edges, side) == value


def test_example() -> None:
    value, side = karger_stein(8, EXAMPLE, rng=random.Random(1))
    assert value == 2
    assert side in ({0, 1, 2, 6}, {3, 4, 5, 7})


def test_one_run_succeeds_often() -> None:
    rng = random.Random(7)
    w = [[0] * 8 for _ in range(8)]
    for u, v, c in EXAMPLE:
        w[u][v] += c
        w[v][u] += c
    groups = [frozenset([v]) for v in range(8)]
    wins = sum(recursive_contract(w, groups, rng)[0] == 2 for _ in range(2000))
    # the proof guarantees 2 / (D + 2) = 1/2 for this graph, whose recursion depth D is 2
    assert wins / 2000 > 0.5


def test_small_and_degenerate() -> None:
    assert karger_stein(2, [(0, 1, 3), (1, 0, 4)]) == (7, {0})
    assert karger_stein(2, []) == (0, {0})
    with pytest.raises(ValueError):
        karger_stein(1, [])
    with pytest.raises(ValueError):
        karger_stein(3, [(0, 1, -1)])


def test_disconnected_zero_weight_and_loops() -> None:
    edges = [(i, i + 1, 3) for i in range(5)] + [(i, i + 1, 3) for i in range(6, 11)] + [(5, 6, 0), (0, 0, 9)]
    value, side = karger_stein(12, edges, rng=random.Random(3))
    assert value == 0
    check(12, edges, value, side)


def test_cycle_and_real_weights() -> None:
    cycle = [(i, (i + 1) % 10, 1) for i in range(10)]
    assert karger_stein(10, cycle, rng=random.Random(5))[0] == 2
    rng = random.Random(11)
    edges = [(u, v, Fraction(rng.randint(1, 9), rng.randint(1, 9))) for u, v in itertools.combinations(range(8), 2)]
    value, side = karger_stein(8, edges, runs=30, rng=rng)
    assert value == brute(8, edges)
    check(8, edges, value, side)
    floats = [(u, v, c * 0.5) for u, v, c in cycle]
    assert karger_stein(10, floats, rng=rng)[0] == 1.0


def test_random_against_brute_force() -> None:
    rng = random.Random(2024)
    for _ in range(500):
        n = rng.randint(2, 10)
        edges = [(rng.randrange(n), rng.randrange(n), rng.randint(0, 4)) for _ in range(rng.randint(0, 3 * n))]
        value, side = karger_stein(n, edges, runs=25, rng=rng)
        check(n, edges, value, side)
        assert value == brute(n, edges)

What the next level needs

  • the tests did not run in this build

Read the full page about Karger-Stein algorithm