fix(graph): defer igraph rebuild to O(1) instead of O(m) per edge

MemoryGraph.add_edge() rebuilt the entire igraph after each edge
write once backend switched (at 500 nodes), making edge addition
O(m × (n+m)) — 49.81s for 5931 edges, causing 30s MCP timeout.

Defer rebuild via _ig_stale flag; rebuild once on first read
(centrality) instead of after every write. Edge addition now 0.77s.
This commit is contained in:
Apunkt 2026-07-14 16:32:15 +02:00
parent 034d72f89e
commit 2d6de461a0
No known key found for this signature in database

View file

@ -42,6 +42,7 @@ class MemoryGraph:
def __init__(self) -> None: def __init__(self) -> None:
self._nx: nx.Graph = nx.Graph() self._nx: nx.Graph = nx.Graph()
self._ig: "ig.Graph | None" = None self._ig: "ig.Graph | None" = None
self._ig_stale: bool = False
self._attrs: dict[UUID, dict[str, Any]] = {} self._attrs: dict[UUID, dict[str, Any]] = {}
self._backend: str = "networkx" self._backend: str = "networkx"
@ -79,10 +80,11 @@ class MemoryGraph:
self._nx.add_edge( self._nx.add_edge(
str(src), str(dst), weight=weight, edge_type=edge_type str(src), str(dst), weight=weight, edge_type=edge_type
) )
if self._ig is not None: self._maybe_switch_backend()
# igraph mirror is immutable by topology; rebuild after each edge if self._backend == "igraph":
# write while in igraph backend. Cheap enough at Phase-1 scale. # Lazy rebuild: igraph mirror is stale after edge writes
self._rebuild_igraph() # and will be rebuilt on next read.
self._ig_stale = True
# ------------------------------------------------------ backend switching # ------------------------------------------------------ backend switching
@ -105,11 +107,17 @@ class MemoryGraph:
weights = [ weights = [
float(self._nx[u][v].get("weight", 1.0)) for u, v in self._nx.edges() float(self._nx[u][v].get("weight", 1.0)) for u, v in self._nx.edges()
] ]
g = ig.Graph(n=len(nodes), edges=edges, directed=False) g = ig.Graph(n=len(nodes), edges=edges, directed=False) # type: ignore[unbound]
g.vs["name"] = nodes g.vs["name"] = nodes
if weights: if weights:
g.es["weight"] = weights g.es["weight"] = weights
self._ig = g self._ig = g
self._ig_stale = False
def _ensure_igraph(self) -> None:
"""Rebuild igraph mirror if stale from deferred edge writes."""
if self._ig_stale and _HAS_IGRAPH:
self._rebuild_igraph()
# ---------------------------------------------------------- graph metrics # ---------------------------------------------------------- graph metrics
@ -124,6 +132,7 @@ class MemoryGraph:
bc = nx.betweenness_centrality(self._nx, weight="weight") bc = nx.betweenness_centrality(self._nx, weight="weight")
return {UUID(n): float(c) for n, c in bc.items()} return {UUID(n): float(c) for n, c in bc.items()}
# igraph path # igraph path
self._ensure_igraph()
assert self._ig is not None assert self._ig is not None
has_weight = "weight" in self._ig.es.attributes() has_weight = "weight" in self._ig.es.attributes()
raw = self._ig.betweenness(weights="weight" if has_weight else None) raw = self._ig.betweenness(weights="weight" if has_weight else None)