求有向无环图（DAG）的K条最长路径的Python/伪代码实现

Hey there, I get it—hunting for working code to find the top-K longest paths in a DAG can be frustrating even after digging through papers. Let's walk through a practical implementation based on the approach from that research you referenced, with both pseudocode and runnable Python code you can test right away.

Core Idea #

Since DAGs have no cycles, we can use topological sorting to process nodes in an order where all predecessors are handled before the node itself. For each node, we'll track the top-K longest paths ending at that node. When processing a node's neighbors, we merge the current node's paths with the neighbor's existing paths, then trim down to keep only the top-K entries. This avoids redundant calculations and ensures we only keep the most relevant paths.

Pseudocode Outline #

This breaks down the logic step-by-step for clarity:

1. Perform topological sort on the DAG to get a processing order [n₁, n₂, ..., nₘ]
2. Initialize for every node u:
   - paths[u]: List of the top-K paths ending at u (starts with just [u] with length 0 if u is a source node)
   - lengths[u]: List of path lengths corresponding to paths[u]
3. For each node u in topological order:
   - For each neighbor v of u (connected by an edge u→v with weight w):
     - Generate candidate paths: Take each path in paths[u], append v, and calculate new length (original length + w)
     - Merge these candidates with v's existing paths and lengths
     - Sort the merged entries by length in descending order, remove duplicate paths, then keep only the top-K
     - Update paths[v] and lengths[v] with this trimmed list
4. After processing all nodes, collect all paths from every node's path list, sort them by length descending, and pick the top-K unique paths

Runnable Python Code #

This implementation assumes your DAG is represented as an adjacency list, where each entry graph[u] contains tuples (v, weight) for edges from u to v.

from collections import deque

def top_k_longest_paths_dag(graph, k):
    # Step 1: Compute in-degrees and perform topological sort
    in_degree = {node: 0 for node in graph}
    for u in graph:
        for v, _ in graph[u]:
            in_degree[v] += 1
    
    # Initialize queue with source nodes (nodes with in-degree 0)
    queue = deque([node for node in in_degree if in_degree[node] == 0])
    topo_order = []
    
    while queue:
        u = queue.popleft()
        topo_order.append(u)
        for v, _ in graph[u]:
            in_degree[v] -= 1
            if in_degree[v] == 0:
                queue.append(v)
    
    # Step 2: Initialize paths and lengths for each node
    node_paths = {node: [] for node in graph}
    node_lengths = {node: [] for node in graph}
    
    # Source nodes start with their own trivial path (length 0)
    for node in [n for n in in_degree if in_degree[n] == 0]:
        node_paths[node] = [[node]]
        node_lengths[node] = [0]
    
    # Step 3: Process nodes in topological order
    for u in topo_order:
        current_paths = node_paths[u]
        current_lengths = node_lengths[u]
        
        for v, weight in graph[u]:
            # Generate candidate paths extending from u to v
            candidate_paths = [path + [v] for path in current_paths]
            candidate_lengths = [length + weight for length in current_lengths]
            
            # Merge with v's existing paths
            merged_paths = node_paths[v] + candidate_paths
            merged_lengths = node_lengths[v] + candidate_lengths
            
            # Sort by length descending, then remove duplicate paths
            sorted_pairs = sorted(zip(merged_lengths, merged_paths), key=lambda x: (-x[0], x[1]))
            unique_pairs = []
            seen_paths = set()
            
            for length, path in sorted_pairs:
                path_tuple = tuple(path)
                if path_tuple not in seen_paths:
                    seen_paths.add(path_tuple)
                    unique_pairs.append((length, path))
            
            # Keep only the top-K paths for v
            top_k_pairs = unique_pairs[:k]
            node_lengths[v] = [pair[0] for pair in top_k_pairs]
            node_paths[v] = [pair[1] for pair in top_k_pairs]
    
    # Step 4: Collect all paths and extract the final top-K
    all_paths = []
    all_lengths = []
    for node in graph:
        all_paths.extend(node_paths[node])
        all_lengths.extend(node_lengths[node])
    
    # Sort all paths by length descending, remove duplicates
    sorted_all = sorted(zip(all_lengths, all_paths), key=lambda x: (-x[0], x[1]))
    final_seen = set()
    top_k_result = []
    
    for length, path in sorted_all:
        path_tuple = tuple(path)
        if path_tuple not in final_seen:
            final_seen.add(path_tuple)
            top_k_result.append((length, path))
            if len(top_k_result) == k:
                break
    
    return top_k_result

# Example Usage
if __name__ == "__main__":
    # Sample DAG with weighted edges
    sample_dag = {
        'A': [('B', 3), ('C', 2)],
        'B': [('D', 4), ('C', 1)],
        'C': [('D', 2)],
        'D': []
    }
    k_value = 3
    longest_paths = top_k_longest_paths_dag(sample_dag, k_value)
    
    print(f"Top {k_value} longest paths in the sample DAG:")
    for i, (length, path) in enumerate(longest_paths, 1):
        print(f"{i}. Path: {' -> '.join(path)} | Length: {length}")

Notes for Optimization #

• For large-scale DAGs, instead of fully sorting merged paths, you can use a min-heap to maintain only the top-K entries. This reduces time complexity from O(m log m) to O(m log k) per merge step.
• If your DAG has unweighted edges, just set all edge weights to 1 and the code will still work.
• The code handles duplicate paths (same nodes in order) to avoid redundant entries in the final result.

内容的提问来源于stack exchange，提问作者Sina Faezi