From b37af27b4520a8d864563f2b3037593140815bd4 Mon Sep 17 00:00:00 2001 From: Hao Liang Date: Thu, 17 Sep 2026 17:40:54 +0800 Subject: [PATCH] fix: preserve edge-distinct paths in graph traversal --- .../graph/providers/networkx_store.py | 23 +++-- datamind/tests/test_graph_parallel_paths.py | 94 +++++++++++++++++++ docs/STABLE_API.md | 15 +++ 3 files changed, 125 insertions(+), 7 deletions(-) create mode 100644 datamind/tests/test_graph_parallel_paths.py diff --git a/datamind/capabilities/graph/providers/networkx_store.py b/datamind/capabilities/graph/providers/networkx_store.py index d19b631..b136dff 100644 --- a/datamind/capabilities/graph/providers/networkx_store.py +++ b/datamind/capabilities/graph/providers/networkx_store.py @@ -298,19 +298,29 @@ async def traverse( relation_filter: list[str] | None = None, max_results: int = 100, ) -> list[GraphPath]: - if not self._g.has_node(start): + """Return edge-distinct simple paths, including parallel relations. + + Traverse breadth-first in stable target/relation/edge-key order and + stop after ``max_results`` paths. Sort that bounded selection by score; + this is not an exhaustive global top-k search. Each path excludes + repeated nodes, while different edge identities may share its nodes. + """ + if max_hops <= 0 or max_results <= 0 or not self._g.has_node(start): return [] allowed = set(relation_filter) if relation_filter else None # BFS over (node, path_edges) up to max_hops. paths: list[GraphPath] = [] - visited: set[tuple[str, ...]] = set() frontier: list[tuple[str, list[Edge], set[str]]] = [(start, [], {start})] depth = 0 while frontier and depth < max_hops: next_frontier: list[tuple[str, list[Edge], set[str]]] = [] for node, edges_so_far, seen in frontier: - for u, v, d in self._g.out_edges(node, data=True): + outgoing = sorted( + self._g.out_edges(node, keys=True, data=True), + key=lambda e: (e[1], str(e[3].get("relation", "related")), str(e[2])), + ) + for u, v, _edge_key, d in outgoing: rel = d.get("relation", "related") if allowed is not None and rel not in allowed: continue @@ -319,10 +329,9 @@ async def traverse( edge_obj = self._edge(u, v, d) path_edges = edges_so_far + [edge_obj] nodes = [start] + [e.target for e in path_edges] - key = tuple(nodes) - if key in visited: - continue - visited.add(key) + # Each frontier entry is an exact edge-sequence prefix. + # Extending it once per keyed edge already enumerates + # distinct paths; node-only dedup loses parallel evidence. paths.append( GraphPath( nodes=nodes, diff --git a/datamind/tests/test_graph_parallel_paths.py b/datamind/tests/test_graph_parallel_paths.py new file mode 100644 index 0000000..8072bbd --- /dev/null +++ b/datamind/tests/test_graph_parallel_paths.py @@ -0,0 +1,94 @@ +"""Parallel edges are distinct evidence paths, including after persistence.""" +import pytest + +from datamind.capabilities.graph.providers.networkx_store import NetworkXGraphStore +from datamind.core.protocols import GraphTriple + + +def triple(src, rel, dst, source=None, confidence=1.0, **props): + return GraphTriple(subject=src, relation=rel, object=dst, source=source, + confidence=confidence, properties=props) + + +@pytest.mark.asyncio +async def test_parallel_relations_extend_into_distinct_multihop_paths(tmp_path): + store = NetworkXGraphStore(persist_path=tmp_path / 'g.json') + await store.upsert_triples([ + triple('A', 'authored', 'B', confidence=0.8), + triple('A', 'reviewed', 'B', confidence=0.6), + triple('B', 'published_in', 'C'), + ]) + paths = await store.traverse('A', max_hops=2) + assert {tuple(e.relation for e in p.edges) for p in paths} == { + ('authored',), ('reviewed',), + ('authored', 'published_in'), ('reviewed', 'published_in'), + } + scores = {tuple(e.relation for e in p.edges): p.score for p in paths} + assert scores[('authored', 'published_in')] == pytest.approx(0.9) + assert [p.score for p in paths] == sorted(scores.values(), reverse=True) + + +@pytest.mark.asyncio +async def test_same_relation_different_origins_survive_reload_and_upsert(tmp_path): + path = tmp_path / 'g.json' + store = NetworkXGraphStore(persist_path=path) + triples = [triple('A', 'cites', 'B', source='doc1'), + triple('A', 'cites', 'B', source='doc2'), + triple('A', 'cites', 'B', source='doc1', _profile_managed=True)] + await store.upsert_triples(triples) + await store.upsert_triples(triples) # Exact edge identities still overwrite. + before = await store.traverse('A') + assert len(before) == 3 + assert {(p.edges[0].properties['source'], + p.edges[0].properties.get('_profile_managed', False)) for p in before} == { + ('doc1', False), ('doc2', False), ('doc1', True), + } + await store.persist() + after = await NetworkXGraphStore(persist_path=path).traverse('A') + assert [p.model_dump() for p in before] == [p.model_dump() for p in after] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('cap', [1, 3, 100]) +async def test_insertion_order_does_not_change_results_even_when_capped(tmp_path, cap): + triples = [triple('A', 'reviewed', 'B', source='z'), + triple('A', 'authored', 'B', source='b'), + triple('A', 'authored', 'B', source='a'), + triple('B', 'next', 'C')] + outputs = [] + for i, ordered in enumerate([triples, list(reversed(triples))]): + store = NetworkXGraphStore(persist_path=tmp_path / f'{i}.json') + await store.upsert_triples(ordered) + paths = await store.traverse('A', max_results=cap) + assert len(paths) <= cap + outputs.append([p.model_dump() for p in paths]) + assert outputs[0] == outputs[1] + + +@pytest.mark.asyncio +async def test_filters_hops_and_cycles_remain_bounded(tmp_path): + store = NetworkXGraphStore(persist_path=tmp_path / 'g.json') + await store.upsert_triples([ + triple('A', 'skip', 'B'), triple('A', 'keep', 'B'), + triple('B', 'keep', 'C'), triple('C', 'keep', 'A'), + triple('A', 'keep', 'A'), + ]) + paths = await store.traverse('A', max_hops=10, relation_filter=['keep']) + assert [p.nodes for p in paths] == [['A', 'B'], ['A', 'B', 'C']] + assert all(e.relation == 'keep' for p in paths for e in p.edges) + assert len(await store.traverse('A', max_hops=1, relation_filter=['keep'])) == 1 + assert await store.traverse('A', relation_filter=['missing']) == [] + assert await store.traverse('missing') == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('cap', [0, -1, 1, 7]) +async def test_dense_parallel_graph_obeys_result_budget(tmp_path, cap): + store = NetworkXGraphStore(persist_path=tmp_path / 'g.json') + await store.upsert_triples([ + triple(str(i), f'relation{j}', str(i+1)) + for i in range(8) for j in range(8) + ]) + paths = await store.traverse('0', max_hops=8, max_results=cap) + assert len(paths) == max(0, cap) + assert await store.traverse('0', max_hops=0) == [] diff --git a/docs/STABLE_API.md b/docs/STABLE_API.md index 2576ced..75922ff 100644 --- a/docs/STABLE_API.md +++ b/docs/STABLE_API.md @@ -112,6 +112,21 @@ Receipt status is one of `stored`, `unchanged`, or `failed`. A receipt records what DataMind observed and attempted; it is not a two-phase commit with an external database or sink. +## NetworkX graph traversal + +`NetworkXGraphStore.traverse` returns paths distinguished by their exact stored +edge sequences, not just their node sequences. Parallel relations and the same +relation from distinct origins remain separate evidence paths; relation, +weight, and source properties are preserved. Upserting an existing edge identity +still replaces that edge rather than creating another path. + +Traversal uses breadth-first expansion in deterministic target/relation/edge-key +order. `relation_filter` applies at every hop, no path repeats a node, and +`max_hops` and `max_results` bound expansion. Non-positive limits return no paths. +Each emitted prefix counts toward `max_results`, so reaching the cap stops +expansion rather than enumerating all combinations of parallel edges. Selected +paths are sorted by descending score; this is not a global top-k guarantee. + ## HTTP API contract The bundled FastAPI app is a thin transport over the same facade. Request body