reach.cluster

Modularity-based skill clustering for subagent catalog scoping.

Cluster skills into cohesive subagent scopes using modularity optimization.

ClusterPartition

Bases: BaseModel

Represent a modularity clustering partition of a skill corpus.

Source code in src/reach/cluster.py

class ClusterPartition(BaseModel):
    """Represent a modularity clustering partition of a skill corpus."""

    model_config = ConfigDict(frozen=True)

    clusters: tuple[SkillCluster, ...] = ()
    modularity: float = 0.0
    total_skills: int = 0

    @property
    def cluster_map(self) -> dict[str, str]:
        """Map each skill name directly to its assigned cluster identifier."""
        return {skill: c.id for c in self.clusters for skill in c.skills}

cluster_map property

cluster_map: dict[str, str]

Map each skill name directly to its assigned cluster identifier.

SkillCluster

Bases: BaseModel

Represent a cohesive cluster of skills for subagent scoping.

Source code in src/reach/cluster.py

class SkillCluster(BaseModel):
    """Represent a cohesive cluster of skills for subagent scoping."""

    model_config = ConfigDict(frozen=True)

    id: str
    skills: tuple[str, ...] = ()
    cohesion: float = 0.0

cluster_skills

cluster_skills(
    skills: Sequence[Skill],
    *,
    resolution: float = 1.0,
    max_clusters: int | None = None,
    target_size: int | None = None,
) -> ClusterPartition

Partition skills into cohesive communities using modularity optimization.

Build an adjacency matrix from symmetrized BM25 overlap scores between skill descriptions, then iteratively merge communities that produce the greatest gain in modularity (Q).

Parameters:

Name	Type	Description	Default
skills	Sequence[Skill]	The corpus of skills to cluster.	required
resolution	float	Resolution parameter (gamma). Higher values yield smaller, more compact clusters; lower values yield larger, merged clusters. Defaults to 1.0.	1.0
max_clusters	int | None	Optional upper bound on the number of output clusters.	None
target_size	int | None	Optional soft upper bound on the maximum number of skills per cluster.	None

Returns:

Type	Description
ClusterPartition	ClusterPartition with identified skill clusters and final modularity score.

Source code in src/reach/cluster.py

def cluster_skills(
    skills: Sequence[Skill],
    *,
    resolution: float = 1.0,
    max_clusters: int | None = None,
    target_size: int | None = None,
) -> ClusterPartition:
    """Partition skills into cohesive communities using modularity optimization.

    Build an adjacency matrix from symmetrized BM25 overlap scores between skill descriptions,
    then iteratively merge communities that produce the greatest gain in modularity (Q).

    Args:
        skills: The corpus of skills to cluster.
        resolution: Resolution parameter (gamma). Higher values yield smaller, more compact
            clusters; lower values yield larger, merged clusters. Defaults to 1.0.
        max_clusters: Optional upper bound on the number of output clusters.
        target_size: Optional soft upper bound on the maximum number of skills per cluster.

    Returns:
        ClusterPartition with identified skill clusters and final modularity score.
    """
    unique_by_name = {s.name: s for s in skills}
    unique_skills = list(unique_by_name.values())
    total_skills = len(unique_skills)
    if total_skills == 0:
        return ClusterPartition(clusters=(), modularity=0.0, total_skills=0)

    if total_skills == 1:
        cluster = SkillCluster(id="cluster-1", skills=(unique_skills[0].name,), cohesion=1.0)
        return ClusterPartition(clusters=(cluster,), modularity=0.0, total_skills=1)

    n = total_skills
    w, deg, total_weight = _build_adjacency_matrix(unique_skills)

    if total_weight < _EPSILON:
        clusters = tuple(
            SkillCluster(id=f"cluster-{i + 1}", skills=(unique_skills[i].name,), cohesion=1.0)
            for i in range(n)
        )
        return ClusterPartition(clusters=clusters, modularity=0.0, total_skills=n)

    communities: dict[int, set[int]] = {i: {i} for i in range(n)}
    deg_sum: dict[int, float] = {i: deg[i] for i in range(n)}
    internal_w: dict[int, float] = dict.fromkeys(range(n), 0.0)

    pair_w: dict[tuple[int, int], float] = {}
    for i in range(n):
        for j in range(i + 1, n):
            if w[i][j] > 0.0:
                pair_w[(i, j)] = 2.0 * w[i][j]

    initial_q = -resolution / (total_weight**2) * sum(d**2 for d in deg)
    best_q, best_communities = _optimize_communities(
        communities=communities,
        deg_sum=deg_sum,
        internal_w=internal_w,
        pair_w=pair_w,
        total_weight=total_weight,
        resolution=resolution,
        max_clusters=max_clusters,
        target_size=target_size,
        initial_q=initial_q,
    )

    final_modularity = max(0.0, round(best_q, 4)) if best_q > -1.0 else 0.0
    clusters = _build_clusters(unique_skills, best_communities, w)

    return ClusterPartition(
        clusters=clusters,
        modularity=final_modularity,
        total_skills=total_skills,
    )