Scope deduplication during incremental merges

(cherry picked from commit 9e65a71993)
This commit is contained in:
himanshupatro-334 authored and safishamsi committed 2026-09-11 22:46:15 +01:00
1 parent 23f2ffaa43
commit 8a89253fec
3 files changed
+540 -18

No files matched your search

+18 -1
View File
@@ -1402,6 +1402,7 @@ def build(
dedup: bool = True,
dedup_llm_backend: str | None = None,
root: str | Path | None = None,
protected_ids: "set[str] | None" = None,
) -> nx.Graph:
"""Merge multiple extraction results into one graph.
@@ -1411,6 +1412,8 @@ def build(
dedup_llm_backend: if set (e.g. "gemini", "claude", or "kimi"), uses LLM to resolve
ambiguous pairs in the 75–92 Jaro-Winkler score zone.
root: if given, absolute source_file paths are made relative to root (#932).
protected_ids: optional set of node IDs to protect from being collapsed with
other protected nodes during incremental merge (#3477).
With dedup disabled, extractions are merged in order and the last node's
attributes win (NetworkX add_node overwrites). With dedup enabled, nodes
@@ -1454,6 +1457,7 @@ def build(
# Hyperedge members reference node ids too, so they need the same
# survivor rewiring the edges get (#2805).
hyperedges=combined.get("hyperedges"),
protected_ids=protected_ids,
)
return build_from_json(combined, directed=directed, root=_root)
@@ -2027,8 +2031,21 @@ def build_merge(
if had_graph else []
)
# Untouched existing nodes must not be collapsed with each other during dedup (#3477).
_protected_ids = {
n["id"] for n in existing_nodes
if isinstance(n, dict) and n.get("id")
} if had_graph else None
all_chunks = base + list(new_chunks)
G = build(all_chunks, directed=directed, dedup=dedup, dedup_llm_backend=dedup_llm_backend, root=_eff_root)
G = build(
all_chunks,
directed=directed,
dedup=dedup,
dedup_llm_backend=dedup_llm_backend,
root=_eff_root,
protected_ids=_protected_ids,
)
# Prune nodes and edges from deleted source files
if prune_sources:
+137 -17
View File
@@ -555,6 +555,7 @@ def deduplicate_entities(
dedup_llm_backend: str | None = None,
root: str | Path | None = None,
hyperedges: "list[dict] | None" = None,
protected_ids: "set[str] | None" = None,
) -> tuple[list[dict], list[dict]]:
"""Deduplicate near-identical entities in a knowledge graph.
@@ -654,6 +655,22 @@ def deduplicate_entities(
uf = _UF()
exact_merges = 0
protected_set: set[str] = set(protected_ids) if protected_ids is not None else set()
prot_by_root: dict[str, str] = {pid: pid for pid in protected_set}
def _get_prot(nid: str) -> str | None:
return prot_by_root.get(uf.find(nid))
def _union_with_prot(x: str, y: str) -> None:
px = _get_prot(x)
py = _get_prot(y)
uf.union(x, y)
new_root = uf.find(x)
prot = px or py
if prot is not None:
prot_by_root[new_root] = prot
for key, group in norm_to_nodes.items():
if len(group) <= 1:
continue
@@ -671,10 +688,33 @@ def deduplicate_entities(
# collapsing distinct nodes that happen to share a label (#1178).
continue
if len(file_group) > 1:
winner = _pick_winner(file_group)
for node in file_group:
uf.union(winner["id"], node["id"])
exact_merges += len(file_group) - 1
if protected_set:
prot_file = [n for n in file_group if n.get("id") in protected_set]
inc_file = [n for n in file_group if n.get("id") not in protected_set]
if prot_file and not inc_file:
# All nodes belong exclusively to an untouched file — preserve all of them
continue
if prot_file and inc_file:
winner = _pick_winner(prot_file)
for node in inc_file:
px = _get_prot(winner["id"])
py = _get_prot(node["id"])
if px is not None and py is not None and px != py:
continue
if uf.find(winner["id"]) != uf.find(node["id"]):
_union_with_prot(winner["id"], node["id"])
exact_merges += 1
else:
winner = _pick_winner(file_group)
for node in file_group:
if uf.find(winner["id"]) != uf.find(node["id"]):
_union_with_prot(winner["id"], node["id"])
exact_merges += 1
else:
winner = _pick_winner(file_group)
for node in file_group:
uf.union(winner["id"], node["id"])
exact_merges += len(file_group) - 1
# Cross-file residue: union exact matches across files, but only where
# it is provably safe (#2182). `concept` is the one file_type meant to
# unify across files (#1284) — code is keyed by ID (#1205) and
@@ -703,11 +743,38 @@ def deduplicate_entities(
key=lambda n: n["id"],
)
if len(mergeable) > 1:
winner = _pick_winner(mergeable)
for node in mergeable:
if uf.find(winner["id"]) != uf.find(node["id"]):
uf.union(winner["id"], node["id"])
exact_merges += 1
if protected_set:
prot_members = [n for n in mergeable if n.get("id") in protected_set]
inc_members = [n for n in mergeable if n.get("id") not in protected_set]
if not inc_members:
# All participants belong exclusively to untouched files (#3477):
# NEVER collapse them during incremental merge.
continue
if prot_members:
# Mixed: pick AT MOST ONE protected survivor for incoming nodes to fold into.
# Multiple protected nodes must remain separate independent entities.
canonical_winner = _pick_winner(prot_members)
for inc in inc_members:
px = _get_prot(canonical_winner["id"])
py = _get_prot(inc["id"])
if px is not None and py is not None and px != py:
continue
if uf.find(canonical_winner["id"]) != uf.find(inc["id"]):
_union_with_prot(canonical_winner["id"], inc["id"])
exact_merges += 1
else:
# Incoming only: merge normally
winner = _pick_winner(inc_members)
for node in inc_members:
if uf.find(winner["id"]) != uf.find(node["id"]):
_union_with_prot(winner["id"], node["id"])
exact_merges += 1
else:
winner = _pick_winner(mergeable)
for node in mergeable:
if uf.find(winner["id"]) != uf.find(node["id"]):
uf.union(winner["id"], node["id"])
exact_merges += 1
# ── pass 2: MinHash/LSH + Jaro-Winkler (high-entropy nodes only) ─────────
candidates: list[dict] = []
@@ -757,6 +824,13 @@ def deduplicate_entities(
if uf.find(node_id) == uf.find(neighbor_id):
continue
if protected_set:
px = _get_prot(node_id)
py = _get_prot(neighbor_id)
if px is not None and py is not None and px != py:
# Prevent protected/protected unions and bridging across protected components
continue
neighbor = candidates_by_id.get(neighbor_id)
if neighbor is None:
continue
@@ -826,14 +900,31 @@ def deduplicate_entities(
# from the union of both normalized-label groups pulls
# never-compared nodes (same label, different source_file)
# into the merge, bypassing the #1046/#1178 guards.
winner = _pick_winner([node, neighbor])
uf.union(winner["id"], node_id)
uf.union(winner["id"], neighbor_id)
if protected_set:
px = _get_prot(node_id)
py = _get_prot(neighbor_id)
if px is not None and py is not None and px != py:
continue
if node_id in protected_set:
winner = node
elif neighbor_id in protected_set:
winner = neighbor
else:
winner = _pick_winner([node, neighbor])
_union_with_prot(winner["id"], node_id)
_union_with_prot(winner["id"], neighbor_id)
else:
winner = _pick_winner([node, neighbor])
uf.union(winner["id"], node_id)
uf.union(winner["id"], neighbor_id)
fuzzy_merges += 1
# ── pass 3: LLM tiebreaker for ambiguous pairs (opt-in) ──────────────────
if dedup_llm_backend is not None:
_llm_tiebreak(candidates, uf, communities, backend=dedup_llm_backend)
_llm_tiebreak(
candidates, uf, communities, backend=dedup_llm_backend,
protected_set=protected_set, get_prot=_get_prot, union_with_prot=_union_with_prot,
)
# ── build remap table from union-find components ──────────────────────────
components = uf.components()
@@ -861,7 +952,14 @@ def deduplicate_entities(
key=lambda pair: pair[0],
)
]
winner = _pick_winner(group_nodes) if group_nodes else {"id": root}
if protected_set:
prot_in_group = [n for n in group_nodes if n.get("id") in protected_set]
if prot_in_group:
winner = _pick_winner(prot_in_group)
else:
winner = _pick_winner(group_nodes) if group_nodes else {"id": root}
else:
winner = _pick_winner(group_nodes) if group_nodes else {"id": root}
winner_id = winner["id"]
# Even with the right survivor, dropping the losers wholesale loses
# whatever fields only they carried. Fill the survivor's absent
@@ -991,6 +1089,9 @@ def _llm_tiebreak(
batch_size: int = 30,
low: float = 75.0,
high: float = 92.0,
protected_set: set[str] | None = None,
get_prot=None,
union_with_prot=None,
) -> None:
"""Batch-resolve ambiguous pairs (score in [low, high)) via LLM."""
try:
@@ -1040,6 +1141,11 @@ def _llm_tiebreak(
and min(len(norm_i), len(norm_j)) >= 12):
score += _COMMUNITY_BOOST
if low <= score < high:
if protected_set and get_prot is not None:
px = get_prot(node["id"])
py = get_prot(neighbor["id"])
if px is not None and py is not None and px != py:
continue
ambiguous.append((node, neighbor, score))
if not ambiguous:
@@ -1086,8 +1192,22 @@ def _llm_tiebreak(
answer = parts[1].strip().lower()
if answer.startswith("yes"):
a, b, _ = batch[idx]
winner = _pick_winner([a, b])
uf.union(winner["id"], a["id"])
uf.union(winner["id"], b["id"])
if protected_set and get_prot is not None and union_with_prot is not None:
px = get_prot(a["id"])
py = get_prot(b["id"])
if px is not None and py is not None and px != py:
continue
if a["id"] in protected_set:
winner = a
elif b["id"] in protected_set:
winner = b
else:
winner = _pick_winner([a, b])
union_with_prot(winner["id"], a["id"])
union_with_prot(winner["id"], b["id"])
else:
winner = _pick_winner([a, b])
uf.union(winner["id"], a["id"])
uf.union(winner["id"], b["id"])
except Exception as exc:
print(f"[graphify] --dedup-llm batch failed: {exc}", flush=True)
+385
View File
@@ -0,0 +1,385 @@
"""Regression tests for Graphify issue #3477:
Scoped entity deduplication during incremental build_merge().
Ensures:
1. Untouched existing nodes are protected from collapsing with other untouched existing nodes.
2. Incoming nodes can still merge into an untouched node, with the untouched node surviving as canonical.
3. Incoming nodes that match multiple untouched nodes attach to at most one protected survivor without
transitively collapsing the protected nodes.
4. Incoming duplicates deduplicate normally among themselves.
5. Full build() with protected_ids=None retains existing global dedup behavior.
6. Fuzzy dedup respects the same protected invariants.
"""
import json
from pathlib import Path
import pytest
import networkx as nx
import graphify.build as buildmod
from graphify.build import build, build_merge
from graphify.dedup import deduplicate_entities
def _write_graph(graph_path: Path, nodes, edges=(), hyperedges=()) -> None:
graph_path.parent.mkdir(parents=True, exist_ok=True)
graph_path.write_text(
json.dumps(
{
"nodes": list(nodes),
"edges": list(edges),
"hyperedges": list(hyperedges),
}
),
encoding="utf-8",
)
def test_untouched_duplicate_nodes_survive_incremental_merge(tmp_path):
"""#3477: Two duplicate-labeled nodes in two untouched files must not be collapsed
when an unrelated third file is incrementally updated with dedup=True."""
gp = tmp_path / "graphify-out" / "graph.json"
node_a = {
"id": "a_auth_service",
"label": "Authentication Service Component",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
node_b = {
"id": "b_auth_service",
"label": "Authentication Service Component",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
_write_graph(gp, [node_a, node_b])
# Incremental update touches unrelated c.md
chunk_c = {
"nodes": [
{
"id": "c_worker",
"label": "Background Worker Job",
"file_type": "concept",
"source_file": "c.md",
"_origin": "semantic",
}
],
"edges": [],
}
G = build_merge([chunk_c], gp, dedup=True)
# Both untouched nodes must survive with their original IDs
assert "a_auth_service" in G
assert "b_auth_service" in G
assert "c_worker" in G
assert G.number_of_nodes() == 3
def test_incoming_duplicate_merges_into_untouched_node_as_canonical_survivor(tmp_path):
"""#3477: An incoming entity duplicate merges into an untouched entity, and the
untouched node MUST be the canonical survivor, with edges rewired."""
gp = tmp_path / "graphify-out" / "graph.json"
untouched_cache = {
"id": "a_cache_mgr",
"label": "Memory Cache Manager System",
"file_type": "concept",
"source_file": "a.md",
"attributes": {"tier": "primary"},
"_origin": "semantic",
}
_write_graph(gp, [untouched_cache])
# Incoming extraction from changed b.md has a richer duplicate
incoming_cache = {
"id": "b_cache_mgr", # shorter ID, would normally win on tiebreak
"label": "Memory Cache Manager System",
"file_type": "concept",
"source_file": "b.md",
"summary": "Distributed memory cache manager for session state",
"_origin": "semantic",
}
incoming_caller = {
"id": "b_client",
"label": "Cache Client Worker",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
call_edge = {
"source": "b_client",
"target": "b_cache_mgr",
"relation": "calls",
"confidence": "EXTRACTED",
"source_file": "b.md",
}
chunk_b = {
"nodes": [incoming_cache, incoming_caller],
"edges": [call_edge],
}
G = build_merge([chunk_b], gp, dedup=True)
# The untouched node MUST survive as the canonical ID
assert "a_cache_mgr" in G
assert "b_cache_mgr" not in G
assert "b_client" in G
# Missing fields from incoming loser should be merged into survivor
survivor_attrs = G.nodes["a_cache_mgr"]
assert survivor_attrs.get("summary") == "Distributed memory cache manager for session state"
# Edge must be rewired to the untouched survivor
assert G.has_edge("b_client", "a_cache_mgr")
def test_two_untouched_plus_one_incoming_duplicate_bridge_case(tmp_path):
"""#3477: When two untouched duplicate nodes exist and an incoming node shares the same label:
- Both untouched nodes survive as separate entities.
- Incoming node resolves to at most one protected survivor.
- The two protected nodes do not become transitively connected."""
gp = tmp_path / "graphify-out" / "graph.json"
node_a = {
"id": "a_auth",
"label": "Authentication Service Gateway",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
node_b = {
"id": "b_auth",
"label": "Authentication Service Gateway",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
_write_graph(gp, [node_a, node_b])
# Incoming extraction from c.md introduces another duplicate
node_c = {
"id": "c_auth",
"label": "Authentication Service Gateway",
"file_type": "concept",
"source_file": "c.md",
"_origin": "semantic",
}
edge_c = {
"source": "c_client",
"target": "c_auth",
"relation": "uses",
"confidence": "EXTRACTED",
"source_file": "c.md",
}
chunk_c = {
"nodes": [
node_c,
{
"id": "c_client",
"label": "Gateway Client",
"file_type": "concept",
"source_file": "c.md",
"_origin": "semantic",
},
],
"edges": [edge_c],
}
G = build_merge([chunk_c], gp, dedup=True)
# BOTH untouched nodes must survive
assert "a_auth" in G
assert "b_auth" in G
# Incoming node was folded into one of the protected nodes
assert "c_auth" not in G
# Total nodes: 2 protected + 1 client = 3 nodes
assert G.number_of_nodes() == 3
def test_two_incoming_duplicate_nodes_deduplicate_normally(tmp_path):
"""#3477: Multiple incoming duplicates still deduplicate normally during build_merge."""
gp = tmp_path / "graphify-out" / "graph.json"
untouched_node = {
"id": "a_existing",
"label": "Existing Untouched Component",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
_write_graph(gp, [untouched_node])
# Re-extracting / adding new files b.md and c.md that both emit a shared concept
chunk_b = {
"nodes": [
{
"id": "b_telemetry",
"label": "Telemetry Event Dispatcher",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
],
"edges": [],
}
chunk_c = {
"nodes": [
{
"id": "c_telemetry",
"label": "Telemetry Event Dispatcher",
"file_type": "concept",
"source_file": "c.md",
"_origin": "semantic",
}
],
"edges": [],
}
G = build_merge([chunk_b, chunk_c], gp, dedup=True)
# Untouched survives
assert "a_existing" in G
# Incoming nodes collapsed into one
surviving_telemetry = [n for n in G.nodes if "telemetry" in n]
assert len(surviving_telemetry) == 1
assert G.number_of_nodes() == 2
def test_full_build_protected_ids_none_retains_global_dedup():
"""#3477: When build() is called without protected_ids (cold build), normal global dedup occurs."""
chunk_a = {
"nodes": [
{
"id": "a_auth",
"label": "Authentication Service Gateway",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
],
"edges": [],
}
chunk_b = {
"nodes": [
{
"id": "b_auth",
"label": "Authentication Service Gateway",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
],
"edges": [],
}
# Full build with protected_ids=None collapses both into 1 node
G = build([chunk_a, chunk_b], dedup=True, protected_ids=None)
assert G.number_of_nodes() == 1
def test_fuzzy_dedup_protected_nodes_do_not_merge(tmp_path):
"""#3477: Pass 2 fuzzy dedup must not collapse two near-identical nodes in untouched files."""
gp = tmp_path / "graphify-out" / "graph.json"
# Near-identical labels that clear the Jaro threshold
node_a = {
"id": "a_auth_system",
"label": "Authentication Manager Processing Engine",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
node_b = {
"id": "b_auth_system",
"label": "Authentication Manager Processng Engine", # typo on token >= 6 chars
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
_write_graph(gp, [node_a, node_b])
chunk_c = {
"nodes": [
{
"id": "c_unrelated",
"label": "Unrelated Database Connector",
"file_type": "concept",
"source_file": "c.md",
"_origin": "semantic",
}
],
"edges": [],
}
G = build_merge([chunk_c], gp, dedup=True)
# Both untouched nodes must survive
assert "a_auth_system" in G
assert "b_auth_system" in G
assert G.number_of_nodes() == 3
def test_fuzzy_dedup_incoming_merges_into_protected_as_canonical(tmp_path):
"""#3477: Pass 2 fuzzy dedup merges incoming typo into untouched canonical node."""
gp = tmp_path / "graphify-out" / "graph.json"
untouched = {
"id": "a_auth_system",
"label": "Authentication Manager Processing Engine",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
_write_graph(gp, [untouched])
incoming_typo = {
"id": "b_auth_system",
"label": "Authentication Manager Processng Engine",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
chunk_b = {"nodes": [incoming_typo], "edges": []}
G = build_merge([chunk_b], gp, dedup=True)
# Untouched survives as canonical winner
assert "a_auth_system" in G
assert "b_auth_system" not in G
assert G.number_of_nodes() == 1
def test_fuzzy_dedup_incoming_bridge_case(tmp_path):
"""#3477: In fuzzy dedup, an incoming node must not bridge two untouched nodes."""
gp = tmp_path / "graphify-out" / "graph.json"
node_a = {
"id": "a_cluster",
"label": "Distributed Storage Processing Cluster Subsystem",
"file_type": "concept",
"source_file": "a.md",
"_origin": "semantic",
}
node_b = {
"id": "b_cluster",
"label": "Distributed Storage Processng Cluster Subsystem",
"file_type": "concept",
"source_file": "b.md",
"_origin": "semantic",
}
_write_graph(gp, [node_a, node_b])
# Incoming node in c.md matching a.md
node_c = {
"id": "c_cluster",
"label": "Distributed Storage Processing Cluster Subsystem",
"file_type": "concept",
"source_file": "c.md",
"_origin": "semantic",
}
chunk_c = {"nodes": [node_c], "edges": []}
G = build_merge([chunk_c], gp, dedup=True)
# Both untouched nodes must survive
assert "a_cluster" in G
assert "b_cluster" in G
assert "c_cluster" not in G
assert G.number_of_nodes() == 2