mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(kg): make k-shortest path search side-effect free (#1000)

Browse Source 
* fix(kg): make k-shortest path search side-effect free

* fix(kg): respect traversal direction for edge exclusion
This commit is contained in:
Guofang.Tang
2026-08-15 15:34:26 +05:00
committed by GitHub
parent 557e29ee14
commit b8175ea801
2 changed files with 207 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
semantica/kg/path_finder.py
+162 -129
View File
@@ -127,66 +127,100 @@ class PathFinder:
try: try:
self.logger.info(f"Finding Dijkstra shortest path from {source} to {target}") self.logger.info(f"Finding Dijkstra shortest path from {source} to {target}")
# Validate nodes exist path = self._dijkstra_shortest_path(
if not self._node_exists(graph, source): graph,
raise ValueError(f"Source node {source} not found") source,
if not self._node_exists(graph, target): target,
raise ValueError(f"Target node {target} not found") weight_attribute,
default_weight,
traversal_graph = graph if directed else self._make_undirected_view(graph) directed,
)
# Dijkstra's algorithm
distances = {source: 0.0}
previous = {}
priority_queue = [(0.0, source)]
visited = set()
while priority_queue:
current_distance, current_node = heapq.heappop(priority_queue)
if current_node in visited:
continue
visited.add(current_node)
if current_node == target:
break
# Explore neighbors
for neighbor, edge_data in self._get_neighbors(traversal_graph, current_node):
if neighbor in visited:
continue
# Get edge weight
weight = self._get_edge_weight(edge_data, weight_attribute, default_weight)
distance = current_distance + weight
if neighbor not in distances or distance < distances[neighbor]:
distances[neighbor] = distance
previous[neighbor] = current_node
heapq.heappush(priority_queue, (distance, neighbor))
# Reconstruct path
if target not in previous and source != target:
return [] # No path found
path = []
current = target
while current is not None:
path.append(current)
current = previous.get(current)
path.reverse()
self.logger.info(f"Found path of length {len(path)}") self.logger.info(f"Found path of length {len(path)}")
return path return path
except ValueError: except ValueError:
# Re-raise ValueError for invalid nodes # Re-raise ValueError for invalid nodes
raise raise
except Exception as e: except Exception as e:
self.logger.error(f"Dijkstra path finding failed: {str(e)}") self.logger.error(f"Dijkstra path finding failed: {str(e)}")
raise RuntimeError(f"Path finding failed: {str(e)}") raise RuntimeError(f"Path finding failed: {str(e)}")
def _dijkstra_shortest_path(
self,
graph: Any,
source: str,
target: str,
weight_attribute: str = "weight",
default_weight: float = 1.0,
directed: bool = True,
excluded_nodes: Optional[Set[str]] = None,
excluded_edges: Optional[Set[Tuple[str, str]]] = None,
) -> List[str]:
"""Find a shortest path without mutating the graph.
``excluded_nodes`` and ``excluded_edges`` are used internally by
Yen's algorithm to model its temporary graph modifications.
"""
excluded_nodes = excluded_nodes or set()
excluded_edges = excluded_edges or set()
# Validate nodes exist before applying the temporary exclusions.
if not self._node_exists(graph, source):
raise ValueError(f"Source node {source} not found")
if not self._node_exists(graph, target):
raise ValueError(f"Target node {target} not found")
if source in excluded_nodes or target in excluded_nodes:
return []
traversal_graph = graph if directed else self._make_undirected_view(graph)
# Dijkstra's algorithm
distances = {source: 0.0}
previous = {}
priority_queue = [(0.0, source)]
visited = set()
while priority_queue:
current_distance, current_node = heapq.heappop(priority_queue)
if current_node in visited or current_node in excluded_nodes:
continue
visited.add(current_node)
if current_node == target:
break
# Explore neighbors
for neighbor, edge_data in self._get_neighbors(traversal_graph, current_node):
if neighbor in visited or neighbor in excluded_nodes:
continue
if self._edge_is_excluded(
traversal_graph, current_node, neighbor, excluded_edges
):
continue
# Get edge weight
weight = self._get_edge_weight(edge_data, weight_attribute, default_weight)
distance = current_distance + weight
if neighbor not in distances or distance < distances[neighbor]:
distances[neighbor] = distance
previous[neighbor] = current_node
heapq.heappush(priority_queue, (distance, neighbor))
# Reconstruct path
if target not in previous and source != target:
return [] # No path found
path = []
current = target
while current is not None:
path.append(current)
current = previous.get(current)
path.reverse()
return path
def a_star_search( def a_star_search(
self, self,
@@ -493,69 +527,89 @@ class PathFinder:
raise ValueError("k must be positive") raise ValueError("k must be positive")
# Find first shortest path # Find first shortest path
first_path = self.dijkstra_shortest_path(graph, source, target, weight_attribute, default_weight) first_path = self.dijkstra_shortest_path(
graph, source, target, weight_attribute, default_weight
)
if not first_path: if not first_path:
return [] return []
paths = [first_path] paths = [first_path]
candidates = [] candidates = []
candidate_paths = {tuple(first_path)}
for i in range(1, k): candidate_order = 0
# Generate candidate paths
for j in range(len(paths[-1]) - 1): while len(paths) < k:
spur_node = paths[-1][j] previous_path = paths[-1]
root_path = paths[-1][:j + 1]
# Generate candidate paths from every spur node in the last path.
# Temporarily remove edges for j in range(len(previous_path) - 1):
removed_edges = [] spur_node = previous_path[j]
root_path = previous_path[:j + 1]
# Block the next edge of every accepted path sharing this root.
excluded_edges = set()
for path in paths: for path in paths:
if len(path) > j and path[:j + 1] == root_path: if len(path) > j + 1 and path[:j + 1] == root_path:
if j + 1 < len(path): excluded_edges.add((path[j], path[j + 1]))
edge_data = self._get_edge_data(graph, path[j], path[j + 1])
if edge_data is not None: # Block root nodes so the combined path remains loopless.
removed_edges.append((path[j], path[j + 1], edge_data)) excluded_nodes = set(root_path[:-1])
self._remove_edge(graph, path[j], path[j + 1]) spur_path = self._dijkstra_shortest_path(
graph,
# Temporarily remove nodes (except spur node and nodes that don't exist) spur_node,
removed_nodes = [] target,
for node in root_path[:-1]: weight_attribute,
if node != spur_node and node != source and self._node_exists(graph, node): default_weight,
removed_nodes.append(node) excluded_nodes=excluded_nodes,
self._remove_node(graph, node) excluded_edges=excluded_edges,
)
# Find spur path
spur_path = self.dijkstra_shortest_path(graph, spur_node, target, weight_attribute, default_weight) if not spur_path:
continue
# Restore graph
for node in removed_nodes: candidate_path = root_path[:-1] + spur_path
self._restore_node(graph, node) if len(candidate_path) != len(set(candidate_path)):
for u, v, data in removed_edges: continue
self._restore_edge(graph, u, v, data)
candidate_key = tuple(candidate_path)
# Combine root and spur paths if candidate_key in candidate_paths:
if spur_path: continue
candidate_path = root_path[:-1] + spur_path
if candidate_path not in candidates and candidate_path not in paths: try:
candidates.append(candidate_path) length = self.path_length(
graph, candidate_path, weight_attribute, default_weight
# Calculate path lengths and sort )
candidates_with_lengths = [] except ValueError:
for path in candidates: continue
try:
length = self.path_length(graph, path, weight_attribute, default_weight) candidate_paths.add(candidate_key)
candidates_with_lengths.append((path, length)) heapq.heappush(candidates, (length, candidate_order, candidate_path))
except ValueError: candidate_order += 1
# Skip invalid paths
continue if not candidates:
break
candidates_with_lengths.sort(key=lambda x: x[1])
_, _, next_path = heapq.heappop(candidates)
# Add shortest unique paths paths.append(next_path)
for path, length in candidates_with_lengths:
if len(paths) < k and path not in paths:
paths.append(path)
return paths return paths
def _edge_is_excluded(
self,
graph: Any,
source: str,
target: str,
excluded_edges: Set[Tuple[str, str]],
) -> bool:
"""Check whether an edge is excluded for the current traversal."""
if (source, target) in excluded_edges:
return True
is_directed = getattr(graph, "is_directed", None)
if callable(is_directed) and not is_directed():
return (target, source) in excluded_edges
return False
def _node_exists(self, graph: Any, node: str) -> bool: def _node_exists(self, graph: Any, node: str) -> bool:
"""Check if node exists in graph.""" """Check if node exists in graph."""
@@ -614,27 +668,6 @@ class PathFinder:
return edge_data.get(weight_attribute, default_weight) return edge_data.get(weight_attribute, default_weight)
return default_weight return default_weight
def _remove_edge(self, graph: Any, u: str, v: str) -> None:
"""Remove edge from graph."""
if hasattr(graph, 'remove_edge'):
graph.remove_edge(u, v)
def _restore_edge(self, graph: Any, u: str, v: str, data: Any) -> None:
"""Restore edge to graph."""
if hasattr(graph, 'add_edge'):
graph.add_edge(u, v, **data)
def _remove_node(self, graph: Any, node: str) -> None:
"""Remove node from graph."""
if hasattr(graph, 'remove_node'):
graph.remove_node(node)
def _restore_node(self, graph: Any, node: str) -> None:
"""Restore node to graph (implementation depends on graph type)."""
# This is a simplified implementation
# In practice, you'd need to restore the node and its connections
pass
def _reconstruct_all_paths( def _reconstruct_all_paths(
self, self,
previous: Dict[str, List[str]], previous: Dict[str, List[str]],
tests/kg/test_path_finder.py
+45 -1
View File
@@ -241,7 +241,51 @@ class TestPathFinder:
if len(paths) > 1: if len(paths) > 1:
lengths = [self.finder.path_length(multi_path_graph, path) for path in paths] lengths = [self.finder.path_length(multi_path_graph, path) for path in paths]
assert all(lengths[i] <= lengths[i+1] for i in range(len(lengths)-1)) assert all(lengths[i] <= lengths[i+1] for i in range(len(lengths)-1))
def test_find_k_shortest_paths_preserves_graph(self):
"""Test k-shortest path search does not mutate the input graph."""
graph = nx.Graph()
graph.add_edges_from([
("A", "X"), ("X", "Y"), ("Y", "E"),
("A", "B"), ("B", "C"), ("C", "E"),
])
original_nodes = set(graph.nodes)
original_edges = set(graph.edges)
paths = self.finder.find_k_shortest_paths(graph, "A", "E", k=5)
assert len(paths) == 2
assert set(graph.nodes) == original_nodes
assert set(graph.edges) == original_edges
def test_find_k_shortest_paths_returns_loopless_paths(self):
"""Test k-shortest paths do not repeat nodes."""
graph = nx.Graph()
graph.add_edges_from([
("A", "D"), ("A", "E"), ("A", "C"),
("B", "D"), ("B", "C"),
])
paths = self.finder.find_k_shortest_paths(graph, "A", "B", k=5)
assert paths == [["A", "C", "B"], ["A", "D", "B"]]
assert all(len(path) == len(set(path)) for path in paths)
def test_dijkstra_exclusion_respects_undirected_traversal(self):
"""Test exclusions apply in both directions for undirected traversal."""
graph = nx.DiGraph()
graph.add_edge("A", "B")
path = self.finder._dijkstra_shortest_path(
graph,
"B",
"A",
directed=False,
excluded_edges={("A", "B")},
)
assert path == []
def test_find_k_shortest_paths_no_path(self): def test_find_k_shortest_paths_no_path(self):
"""Test finding k shortest paths with no path available.""" """Test finding k shortest paths with no path available."""
paths = self.finder.find_k_shortest_paths(self.disconnected_graph, "A", "D", k=3) paths = self.finder.find_k_shortest_paths(self.disconnected_graph, "A", "D", k=3)