feat: add NeuG as parallel graph storage engine with Cypher query support

[BingqingLyu]

Summary

• Add NeuG as an optional parallel graph storage engine alongside NetworkX
• When installed, NeuG automatically generates a graph.db during extraction, enabling Cypher queries via CLI (graphify cypher) and MCP server (cypher_query tool)
• Native incremental update via Cypher MERGE — O(delta) vs NetworkX's O(full graph) rebuild
• Fix pre-existing id_remap bug in incremental extraction that caused unstable file node IDs

Motivation

Graphify currently uses NetworkX + graph.json as its core graph storage. This architecture has bottlenecks:

• Limited query capability: No declarative graph query language — only Python API traversal
• Inefficient incremental updates: Every update requires loading full graph.json → merge → rebuild → re-serialize (O(full graph) even for single-file changes)
• Performance ceiling at scale: Entire graph must be loaded into memory; NetworkX's pure-Python execution becomes a bottleneck on large graphs
• Limited graph algorithm extensibility: Adding custom graph algorithms requires Python-level implementation with no native acceleration path

Why NeuG?

NeuG is a lightweight embedded graph database (C++ core, Python bindings):

1. Native Cypher support — Declarative graph query language; AI agents can query the knowledge graph directly without custom Python code
2. Native incremental updates — Cypher MERGE enables O(delta) upserts in-place, no full-graph reload needed; for 10K+ node graphs, single-file updates are near-instantaneous
3. Battle-tested performance — LDBC benchmark world record holder; lightweight & embeddable (no standalone server, pip install neug is all it takes)
4. Extensible graph algorithms — Native C++ extension framework for custom graph algorithms; Louvain community detection already available, with more algorithms (Leiden, PageRank, etc.) in development — can replace the current Python-based algorithm layer with significant performance gains

Architecture

Dual-engine coexistence, each independently consuming extraction data:

extraction dict ──┬──> NetworkX (build.py)  → graph.json  (existing)
                  └──> NeuG (storage.py)    → graph.db    (new)

Changes

File	Description
graphify/storage.py	New — NeuG adapter layer (init, schema, ingest via MERGE, query, close)
graphify/__main__.py	NeuG ingest during extract + graphify cypher CLI command
graphify/serve.py	cypher_query MCP tool for AI agents
graphify/extract.py	Fix id_remap bug in incremental extraction
pyproject.toml	Add neug>=0.1.2 optional dependency
tests/	Unit tests (13 cases) + e2e integration script

Usage

# Install
pip install graphify[neug]

# Extract (automatically generates graph.db)
graphify extract /path/to/project

# Cypher query
graphify cypher "MATCH (n:code) RETURN n.label, n.source_file LIMIT 10"
graphify cypher "MATCH (a:code)-[e:edge_code_code_calls]->(b:code) RETURN a.label, b.label LIMIT 10"

# MCP server (AI agents query via cypher_query tool)
python -m graphify.serve graphify-out/graph.json

Bugfix: incremental extraction id_remap

The id_remap step uses an auto-inferred root (resolves to path.parent for single-file extraction), inconsistent with the project root used during full extraction. This causes file node IDs to be unstable, producing duplicate nodes on each incremental update.

Fix: use cache_root (the project target directory) for relative_to() in the id_remap step.

Note: deduplicate_entities() incorrectly merges AST nodes

During testing, we found that deduplicate_entities() merges functions from different files that share similar names (e.g., hooks.py:install() and __main__.py:install()). These functions have distinct IDs and different source_files — only their labels are similar. For pure AST extraction, node IDs are inherently unique, making fuzzy dedup harmful.

The NeuG engine writes raw extraction data directly (skipping dedup), preserving full precision. We suggest discussing dedup strategy optimization separately (e.g., applying fuzzy dedup only to LLM-extracted concept nodes).

Test Plan

• pytest tests/test_storage.py tests/test_cypher_cli.py -v — 13 tests passed
• MCP server cypher_query tool end-to-end verified
• Full extract → cypher count matches expected
• Incremental extract (add one function) → MERGE upsert correct
• Uninstall neug → graphify extract . runs normally (silent skip)

🤖 Generated with Claude Code

[safishamsi]

Great idea and architecturally sound — soft-import pattern is correct, the neug optional extra is right, and the bundled id_remap bugfix is a nice bonus. Six things to fix before merge:

1. Module-level hard import: import neug at the top of storage.py means import graphify.storage raises ImportError if neug is missing. Move it inside init_db() or the first function that uses it.

2. Cypher injection: relation, label, source_file, and source_location come from extraction dicts (including LLM output) and are interpolated directly into Cypher strings. Use parameterised queries (conn.execute(query, params)) if NeuG supports them, or at minimum document the trust boundary explicitly.

3. Process-global _created_rel_tables: This module-level set means a second database opened in the same process won't re-issue CREATE statements. Move it into a per-connection registry.

4. ingest_communities is O(nodes × 6 tables): On a 10k-node graph that's 60k Cypher round-trips. Pass the node_types dict (already populated in ingest_extraction) through so you can look up each node's label directly instead of probing all 6 tables.

5. No upper version pin: neug>=0.1.2 — 0.x packages can break wire formats between minor versions. Pin <0.2 or equivalent.

6. Bash e2e test: tests/test_neug_e2e.sh won't be picked up by CI (pytest). Convert to a pytest test or remove — the existing tests/test_storage.py is the right place.

Fix those and this is ready to land.

[BingqingLyu]

Thanks for the detailed review! All 6 items are addressed in the latest push:

1. Module-level hard import — Fixed. import neug is now inside init_db() only. The rest of storage.py imports nothing from neug at module level.
2. Cypher injection / parameterised queries — Fixed. Replaced all string interpolation (_cesc()) with NeuG's native parameterised queries ($param syntax with parameters={} dict). _cesc() has been removed entirely.
3. Process-global _created_rel_tables — Fixed. ensure_schema() now returns a set[str] (per-connection registry), which is threaded through to ingest_extraction() and _ensure_rel_table(). The module-level global has been removed.
4. ingest_communities O(nodes × 6) — Fixed. ingest_extraction() now returns a node_types: dict[str, str] (node ID → file_type), which is passed to ingest_communities() for direct table lookup. Falls back to probing all 6 tables only when node_types is not provided.
5. No upper version pin — Fixed. Changed to neug>=0.1.2,<0.2 in both the neug extra and the all extra.
6. Bash e2e test — Removed. tests/test_neug_e2e.sh has been deleted. Coverage is handled by tests/test_storage.py and tests/test_cypher_cli.py.

---

Looking ahead, we're happy to keep contributing on the NeuG integration. A few directions we have in mind:

• Community detection on NeuG: NeuG has an extensible extension architecture that allows plugging in custom graph algorithms. We're planning to develop community detection algorithms (e.g., Louvain/Leiden) as NeuG extensions, which would enable running community detection directly on NeuG instead of the current NetworkX/graspologic path — this could bring significant performance gains on large graphs.
• More graph algorithm extensions: Beyond community detection, NeuG's C++ extension framework opens the door for PageRank, centrality, and other graph analytics to run natively, which could benefit graphify's analysis pipeline.

We're very excited about this collaboration and would love to keep working on it together.

[BingqingLyu]

Hi @safishamsi , All 6 review items are addressed and I've merged the latest v8 to resolve conflicts. Let me know if there's anything else you'd like us to adjust — happy to iterate. Otherwise this should be ready to merge whenever you get a chance to take another look.

[BingqingLyu]

Benchmark: NeuG vs NetworkX end-to-end performance

Ran an end-to-end comparison of the two storage backends (NetworkX graph.json vs NeuG graph.db) on real-world repos at different scales. Each test averaged over 3 rounds on Apple M1.

Repo	Nodes	Edges	Operation	NetworkX	NeuG	Speedup
graphify	8,540	15,658	Full build	464ms	774ms	0.6x
graphify	8,540	15,658	Incremental (502 nodes)	738ms	532ms	1.4x
graphify	8,540	15,658	Query (calls)	29ms	4ms	7.6x
transformers	120,377	314,344	Full build	6.11s	5.47s	1.1x
transformers	120,377	314,344	Incremental (1318 nodes)	7.17s	953ms	7.5x
transformers	120,377	314,344	Query (calls)	483ms	120ms	4.0x
kubernetes	364,085	777,189	Full build	17.72s	12.96s	1.4x
kubernetes	364,085	777,189	Incremental (4921 nodes)	20.41s	2.23s	9.2x
kubernetes	364,085	777,189	Query (calls)	1.27s	536ms	2.4x

Key takeaways:

• Performance: NeuG is faster for incremental updates (up to 9.2x at 364K nodes) and queries (up to 7.6x). Full build is comparable at scale (1.4x at 364K nodes), with NeuG slightly slower on small graphs due to fixed startup overhead.
• Functionality: The query benchmark above only tests calls edges (the common case for both backends). Beyond this, NeuG supports arbitrary Cypher queries (e.g. multi-hop traversals, pattern matching, aggregations) that are not feasible with the NetworkX JSON workflow.

@safishamsi Updated with performance improvements — benchmark data above. Happy to discuss any questions.

Benchmark script

"""Benchmark: NetworkX (graph.json) vs NeuG (graph.db) on real repos."""
from __future__ import annotations

import json
import os
import shutil
import sys
import tempfile
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import networkx as nx
from graphify.build import build_from_json
from graphify.export import to_json
from graphify.storage import (
    init_db, ensure_schema, ingest_extraction, ingest_communities,
    execute_cypher, close_db,
)

REPOS = [
    ("graphify", Path(__file__).resolve().parent.parent / "graphify-out" / "graph.json"),
]

# Add extra repos from BENCH_REPOS env var (comma-separated paths).
# Example: BENCH_REPOS=/path/to/transformers,/path/to/kubernetes
for _p in os.environ.get("BENCH_REPOS", "").split(","):
    _p = _p.strip()
    if _p:
        _gj = Path(_p) / "graphify-out" / "graph.json"
        REPOS.append((Path(_p).name, _gj))

ROUNDS = 3


def _load_extraction(graph_json_path: Path) -> dict:
    data = json.loads(graph_json_path.read_text(encoding="utf-8"))
    links_key = "edges" if "edges" in data else "links"
    return {
        "nodes": data.get("nodes", []),
        "edges": data.get(links_key, []),
    }


def _timer(fn):
    t0 = time.perf_counter()
    result = fn()
    elapsed = time.perf_counter() - t0
    return elapsed, result


def _avg_timer(fn, rounds: int = ROUNDS):
    times = []
    result = None
    for _ in range(rounds):
        t, result = _timer(fn)
        times.append(t)
    return sum(times) / len(times), result


def bench_full_build(extraction: dict, tmp: Path):
    """Full build: extraction dict -> storage."""
    nx_times = []
    nx_nodes = 0
    for i in range(ROUNDS):
        def nx_build():
            G = build_from_json(extraction, directed=True)
            out = tmp / f"nx_graph_{i}.json"
            to_json(G, {}, str(out))
            return G.number_of_nodes()
        t, nx_nodes = _timer(nx_build)
        nx_times.append(t)

    neug_times = []
    for i in range(ROUNDS):
        db_path = str(tmp / f"full_{i}.db")
        def neug_build(p=db_path):
            db, conn = init_db(p)
            known = ensure_schema(conn)
            ingest_extraction(conn, extraction, incremental=False, known_tables=known)
            close_db(db, conn)
        t, _ = _timer(neug_build)
        neug_times.append(t)

    return sum(nx_times) / ROUNDS, sum(neug_times) / ROUNDS, nx_nodes


def bench_incremental(extraction: dict, tmp: Path):
    """Incremental update: simulate re-extracting ~1% of source files."""
    from collections import defaultdict
    nodes_by_sf = defaultdict(list)
    for n in extraction["nodes"]:
        sf = n.get("source_file", "")
        if sf:
            nodes_by_sf[sf].append(n)

    target = max(1, len(extraction["nodes"]) // 100)
    delta_nodes = []
    picked_sfs = []
    for sf in sorted(nodes_by_sf, key=lambda k: len(nodes_by_sf[k]), reverse=True):
        delta_nodes.extend(nodes_by_sf[sf])
        picked_sfs.append(sf)
        if len(delta_nodes) >= target:
            break

    delta_ids = {n["id"] for n in delta_nodes}
    delta_edges = [
        e for e in extraction["edges"]
        if e.get("source", e.get("from", "")) in delta_ids
    ]
    delta = {"nodes": delta_nodes, "edges": delta_edges}

    from graphify.build import build_merge

    nx_times = []
    for i in range(ROUNDS):
        base_json = tmp / f"inc_base_{i}.json"
        to_json(build_from_json(extraction, directed=True), {}, str(base_json))
        def nx_incremental(p=base_json):
            G = build_merge([delta], graph_path=p, dedup=False, directed=True)
            to_json(G, {}, str(p))
        t, _ = _timer(nx_incremental)
        nx_times.append(t)

    neug_times = []
    for i in range(ROUNDS):
        db_path = str(tmp / f"inc_{i}.db")
        db, conn = init_db(db_path)
        known = ensure_schema(conn)
        ingest_extraction(conn, extraction, incremental=False, known_tables=known)
        close_db(db, conn)

        def neug_incremental(p=db_path):
            db2, conn2 = init_db(p)
            known2 = ensure_schema(conn2, create_tables=False)
            ingest_extraction(conn2, delta, incremental=True, known_tables=known2)
            close_db(db2, conn2)
        t, _ = _timer(neug_incremental)
        neug_times.append(t)

    return sum(nx_times) / ROUNDS, sum(neug_times) / ROUNDS, len(delta_nodes)


def bench_query(extraction: dict, tmp: Path):
    """Query: find all callers of a node."""
    base_json = tmp / "q_base.json"
    G = build_from_json(extraction, directed=True)
    to_json(G, {}, str(base_json))

    def nx_query():
        data = json.loads(base_json.read_text(encoding="utf-8"))
        links_key = "edges" if "edges" in data else "links"
        results = []
        for e in data.get(links_key, []):
            if e.get("relation") == "calls":
                results.append((e.get("source", e.get("from")), e.get("target", e.get("to"))))
        return len(results)

    nx_time, nx_count = _avg_timer(nx_query)

    db_path = str(tmp / "q.db")
    db, conn = init_db(db_path)
    known = ensure_schema(conn)
    ingest_extraction(conn, extraction, incremental=False, known_tables=known)

    def neug_query():
        rows = execute_cypher(
            conn,
            "MATCH (a:code)-[e:edge_code_code_calls]->(b:code) RETURN a.id, b.id",
        )
        return len(rows)

    neug_time, neug_count = _avg_timer(neug_query)
    close_db(db, conn)
    return nx_time, neug_time, nx_count


def fmt_time(t: float) -> str:
    if t < 1:
        return f"{t*1000:.0f}ms"
    return f"{t:.2f}s"


def fmt_speedup(nx_t: float, neug_t: float) -> str:
    if neug_t == 0:
        return "-"
    ratio = nx_t / neug_t
    return f"{ratio:.1f}x"


def main():
    print(f"Rounds per test: {ROUNDS}", file=sys.stderr)
    rows = []
    for repo_name, graph_path in REPOS:
        if not graph_path.exists():
            print(f"[skip] {repo_name}: {graph_path} not found", file=sys.stderr)
            continue

        print(f"\n{'='*60}", file=sys.stderr)
        print(f"Benchmarking: {repo_name}", file=sys.stderr)
        print(f"{'='*60}", file=sys.stderr)

        extraction = _load_extraction(graph_path)
        n_nodes = len(extraction["nodes"])
        n_edges = len(extraction["edges"])
        print(f"  {n_nodes:,} nodes, {n_edges:,} edges", file=sys.stderr)

        with tempfile.TemporaryDirectory() as tmp_dir:
            tmp = Path(tmp_dir)

            print("  [1/3] Full build ...", file=sys.stderr)
            fb_nx, fb_neug, _ = bench_full_build(extraction, tmp)
            rows.append((repo_name, n_nodes, n_edges, "Full build",
                         fmt_time(fb_nx), fmt_time(fb_neug), fmt_speedup(fb_nx, fb_neug)))

            print("  [2/3] Incremental update ...", file=sys.stderr)
            inc_nx, inc_neug, delta_n = bench_incremental(extraction, tmp)
            rows.append((repo_name, n_nodes, n_edges,
                         f"Incremental ({delta_n} nodes)",
                         fmt_time(inc_nx), fmt_time(inc_neug), fmt_speedup(inc_nx, inc_neug)))

            print("  [3/3] Query (all calls edges) ...", file=sys.stderr)
            q_nx, q_neug, _ = bench_query(extraction, tmp)
            rows.append((repo_name, n_nodes, n_edges, "Query (calls)",
                         fmt_time(q_nx), fmt_time(q_neug), fmt_speedup(q_nx, q_neug)))

    print()
    print("| Repo | Nodes | Edges | Operation | NetworkX | NeuG | Speedup |")
    print("|------|-------|-------|-----------|----------|------|---------|")
    for repo, nodes, edges, op, nx_t, neug_t, speedup in rows:
        print(f"| {repo} | {nodes:,} | {edges:,} | {op} | {nx_t} | {neug_t} | **{speedup}** |")


if __name__ == "__main__":
    main()

[BingqingLyu]

Superseded by #2895, which adds GDS extension support (native + incremental community detection) on top of the core NeuG integration proposed here.