如何在有向无环图中有效地找到由 k 个节点形成的所有路径？答案

【问题标题】：How to efficiently find all paths formed by k number of nodes in a directed acyclic graph?如何在有向无环图中有效地找到由 k 个节点形成的所有路径？
【发布时间】：2016-08-27 19:22:31
【问题描述】：

我有一个如下所示的 DAG： Example DAG

我想提取这张图中4个节点构成的所有路径。

我的预期结果应该是这样的：

N1 -> N2 -> N3 -> N4

N1 -> N2 -> N3 -> N5

N1 -> N3 -> N4 -> N5

N2 -> N3 -> N4 -> N5

我目前的尝试是这样的

def path_finder(n1):
    paths = []
    if DAG.has_node(n1):
        for n2 in DAG.successors(n1):
            for n3 in DAG.successors(n2):
                for n4 in DAG.successors(n3):
                    paths.append([n1, n2, n3, n4])
    return paths

我正在为每个节点调用这个函数。 DAG 是一个全局变量，更具体地说，它是一个 networkx 对象 (DAG = networkx.DiGraph() ) 这个幼稚的函数慢得可怜。有没有更有效的策略来做到这一点？

我看过问题20262712，但问题的作者以相当模糊的方式自行解决了。

谢谢

更新：

由于我找不到任何令人满意的算法来解决这个问题，我最终使用我的幼稚函数作为工作人员并行化了这项工作，同时将所有数据转储到队列中。我使用pool.imap_unordered 启动工作函数并聚合队列中的结果。它仍然很慢（5M 节点需要几个小时）。我还应该提供我正在处理的节点的平均程度的数据，因为这会影响我的工人运行的速度。但是，我暂时不提这个。

【问题讨论】：

注意 - 您链接的问题的答案中描述的回溯基本上是利用这样一个事实，即一旦您计算了节点的所有路径，如果您不需要再次这样做您再次遇到该节点（如果您已保存该数据）。我的回答以不同的方式使用它。
你能说一下你需要这个做什么吗？您确定需要列表而不是生成器吗？
这是我正在尝试开发的更大算法的一部分，用于在人类基因组中寻找特定的重复序列（基本上是一个由四个字母 A、T、G、C 组成的大字符串） .这里的每个节点都标记了特定重复的位置并确定了它们的距离。只有当它们的距离小于定义值时，节点才会连接。现在我想确定这个重复的块，因为它们在四个重复的任何组合中都是有意义的。
我想将所有路径转储到 HDF5 文件中。我希望这不会是一个快速的过程，因为我可能有多达 1 亿个节点。因此，我需要在所有昂贵的图形遍历之后转储。
我没有看到好的解决方案。你应该检查一下，但我怀疑你的运行时间是由paths.append([n1, n2, n3, n4]) 控制的。如果是这样，您将无能为力。

标签： python algorithm graph networkx graph-traversal

【解决方案1】：

这是一个函数，它返回图中所有节点之间给定长度的路径。它在所有节点集之间迭代，并使用networkx.all_simple_paths 来获取路径。

import networkx as nx

g = nx.DiGraph()

g.add_nodes_from(['A','B','C','D','E'])

g.add_path(['A','B','C','D'])
g.add_path(['A','B','C','E'])
g.add_path(['A','C','D','E'])
g.add_path(['B','C','D','D'])

def find_paths(graph, number_nodes=4):
    paths = []
    for source in graph.nodes_iter():
        for target in graph.nodes_iter():
            if not source==target:
                p_source_target = nx.all_simple_paths(graph, 
                                                      source, 
                                                      target, 
                                                      cutoff=number_nodes-1)
                paths.extend([p for p in p_source_target if len(p)==number_nodes])
    return paths

find_paths(g)
# output:
[['B', 'C', 'D', 'E'],
 ['A', 'C', 'D', 'E'],
 ['A', 'B', 'C', 'E'],
 ['A', 'B', 'C', 'D']]

【讨论】：

这将找到所有节点对之间的所有路径。然后它选择长度为 4 的路径。您可以通过将截止设置为 4 来显着加快此速度，因此一旦路径长于 4，它就会停止。
谢谢詹姆斯。我担心这种方法的复杂性可能是 O^2 或更糟，因为您要对所有节点进行双重迭代。我对您的代码进行了基准测试，它比我的幼稚策略和上面 Joel 建议的递归策略要慢得多。我很欣赏使用all_simple_paths 的想法。思考如何以更好的方式构建它。
更具体地说，我尝试在具有 1K 个节点的 Graph 上运行您的代码，大约需要 4 分钟。在同一张图上，Joel 的上述策略耗时约 2.7 秒，而我的策略耗时约 3.5 秒。

【解决方案2】：

您的部分问题可能是，如果您遇到一个节点 u 作为路径中的第二个节点，那么您将执行所有计算以找到所有长度为 3 的路径。但是如果您遇到 @987654322 @再次作为第二个节点，你重复所有这些计算。

所以尽量避免这种情况。我们将首先递归计算所有长度为 3 的路径（这需要计算长度为 2 的路径）

def get_paths(G, n):
    '''returns a dict, paths, such that paths[u] is a list of all paths 
       of length n that start from u'''
    if n == 1: #base case, return a dict so that D[u] is a
               #list of all length 1 paths starting from u.
               #it's a boring list.
        return {u: [[u]] for u in G.nodes()}
    #if we get to here n>1 (unless input was bad)
    subpath_dict = get_paths(G,n-1)  #contains all length n-1 paths, 
                                     #indexed by first node
    path_dict = {}
    for u in G:
        path_dict[u] = []  
        for v in G.successors(u):
            path_dict[u].extend([[u]+subpath for subpath in subpath_dict[v]])
    return(path_dict)

G=nx.DiGraph()
G.add_path([1,2,3,4,5,6])
G.add_path([1,3,6,8,10])

path_dict = get_paths(G,4)
path_list = []
for paths in path_dict.values():
    path_list.extend(paths)

【讨论】：

谢谢乔尔。在这里使用递归是非常周到和贴切的。但是，当我对这段代码进行基准测试时，我发现我的天真的策略没有任何性能提升。此外，我拥有拥有数百万个节点的大型网络。我想跟踪路径搜索进度，递归使它变得难以处理。我们可以进一步改进吗？
这让我很吃惊，但仔细观察它开始变得有意义。我必须为u 的每个后继者的每条路径做[u]+subpath。您将进入并为ni 的每个后继者调用一个 for 循环。那些可能有类似的成本。我将在 cmets 中针对您的问题进行进一步的澄清，但我不知道还有很多可以改进的地方。

【解决方案3】：

序列的数量为|V|*d^3，其中d是平均节点输出度。从图的创建方式来看，d 是有界的。我想 d 不是很小（比如 1G 路径。

由于找到一条路径很快（它们很短），因此不确定类似 DP 的算法是否可以提供帮助。类似 DP 的算法试图利用部分计算的数据，因此存在存储和检索该数据的开销，并且可能比仅计算所需的部分数据的开销更大。

一个想法是算法以反向拓扑顺序遍历 DAG 并做两件事：

对于节点，保留从该节点开始的长度为 3 的所有路径，
使用长度为 3 的后继路径打印所有长度为 4 的路径。

此方法会占用大量内存，但可以为不是任何遍历边界节点的后继节点释放部分内存。

其他想法只是让简单的算法更优化。在您的解决方案中，每个节点都有三个 for 循环。这意味着所有路径都有四个 for 循环。请注意，每个循环都是通过节点。有可能的通过迭代边缘来加入前两个循环。这是因为每条路径都必须从一个边缘开始。算法是这样的：

for n1, n2 in DAG.edges():
  for n3 in DAG.successors(n2):
    for n4 in DAG.successors(n3):
      paths.append([n1, n2, n3, n4])

或者更简单，首先选择中间边缘：

for n2, n3 in DAG.edges():
  for n1, n4 in itertools.product(DAG.predecessors(n2), DAG.successors(n3)):
    paths.append([n1, n2, n3, n4])

可以通过不选择从源节点开始或在目标节点结束的中间边缘来优化外循环。但这在 product() 方法中检测得非常快。也许这种优化可以通过不将不需要的数据发送到其他进程来提供帮助。

【讨论】：