如何用Python按流程图逻辑遍历NetworkX生成的GraphML图？

Alright, let's walk through how to traverse your flowchart (exported as GraphML via NetworkX) exactly as per its intended flow. Since flowcharts are directed graphs with clear start/end points and decision branches, here's a step-by-step approach tailored to your use case:

First, we need to pull the flowchart data into a NetworkX graph object. Make sure you're working with a directed graph (since flowcharts have one-way flow) — if your original GraphML was generated as undirected, we can convert it on the fly:

import networkx as nx

# Load the GraphML file
G = nx.read_graphml("your_flowchart.graphml")

# Convert to directed graph if needed (skip if you generated a DiGraph initially)
if not isinstance(G, nx.DiGraph):
    G = nx.DiGraph(G)

Flowcharts rely on specific node types, so we need to pinpoint these first. If you added custom attributes to nodes when generating the GraphML (like type="start" or type="decision"), use those to filter:

# Find your single start node (adjust the attribute key/value to match your graph)
start_node = next(node for node, attrs in G.nodes(data=True) if attrs.get("type") == "start")

# Find all end nodes (flow termination points)
end_nodes = [node for node, attrs in G.nodes(data=True) if attrs.get("type") == "end"]

# Flag decision nodes (points where the flow branches)
decision_nodes = [node for node, attrs in G.nodes(data=True) if attrs.get("type") == "decision"]

If you didn't add attributes, you can infer these nodes by their connectivity:

# Start node: has 0 incoming edges, at least 1 outgoing edge
start_node = next(node for node in G.nodes() if G.in_degree(node) == 0)

# End nodes: have at least 1 incoming edge, 0 outgoing edges
end_nodes = [node for node in G.nodes() if G.out_degree(node) == 0]

# Decision nodes: have more than 1 outgoing edge (since they branch)
decision_nodes = [node for node in G.nodes() if G.out_degree(node) > 1]

Now for the core part: traversing the graph in line with how the flowchart is supposed to run. We'll cover two common traversal styles, plus how to handle decision rules if your edges have rule attributes.

3.1 Breadth-First Traversal (Level-by-Level Flow) #

This is great if you want to process steps in the order they appear, handling all parallel branches at the same level before moving on:

from collections import deque

def bfs_flow_traversal(G, start_node, end_nodes):
    visited = set()
    # Queue stores tuples of (current node, path taken to reach it)
    queue = deque([(start_node, [start_node])])
    
    while queue:
        current_node, path = queue.popleft()
        
        # If we hit an end node, print the completed path
        if current_node in end_nodes:
            print(f"Completed flow path: {' -> '.join(path)}")
            continue
        
        if current_node in visited:
            continue
        visited.add(current_node)
        
        # Follow all outgoing edges (the direction of the flow)
        for neighbor in G.successors(current_node):
            new_path = path.copy()
            new_path.append(neighbor)
            queue.append((neighbor, new_path))

# Run the traversal
bfs_flow_traversal(G, start_node, end_nodes)

3.2 Depth-First Traversal (Full Branch Completion) #

Use this if you want to follow one branch all the way to an end node before backtracking to the last decision point and exploring the next branch:

def dfs_flow_traversal(G, current_node, path, end_nodes, visited):
    path.append(current_node)
    
    # Print the full path when we reach an end node
    if current_node in end_nodes:
        print(f"Completed flow path: {' -> '.join(path)}")
        path.pop()
        return
    
    if current_node in visited:
        path.pop()
        return
    visited.add(current_node)
    
    # Explore each outgoing edge one by one
    for neighbor in G.successors(current_node):
        # Pass a copy of visited to avoid cross-branch contamination
        dfs_flow_traversal(G, neighbor, path, end_nodes, visited.copy())
    
    path.pop()

# Run the traversal
dfs_flow_traversal(G, start_node, [], end_nodes, set())

3.3 Handling Decision Rules (If Your Edges Have Rule Attributes) #

If you added rule/condition attributes to edges (e.g., condition="Customer is Premium"), you can include those in the traversal output to make it clear which rule is triggering each branch:

def dfs_with_decisions(G, current_node, path, end_nodes, visited):
    path.append(current_node)
    
    if current_node in end_nodes:
        print(f"Completed flow path: {' -> '.join(path)}")
        path.pop()
        return
    
    if current_node in visited:
        path.pop()
        return
    visited.add(current_node)
    
    for neighbor in G.successors(current_node):
        # Pull the rule/condition from the edge attributes
        edge_attrs = G.get_edge_data(current_node, neighbor)
        if current_node in decision_nodes and edge_attrs:
            rule = edge_attrs.get("condition", "Unspecified Rule")
            print(f"At decision node {current_node}: Applying rule '{rule}' to proceed to {neighbor}")
        
        dfs_with_decisions(G, neighbor, path, end_nodes, visited.copy())
    
    path.pop()

# Run this version if your edges have rule metadata
dfs_with_decisions(G, start_node, [], end_nodes, set())

• If your flowchart has loops, you'll want to adjust the visited logic to handle cycles without infinite recursion/loops.
• Sharing a snippet of your GraphML or the flowchart image would help refine this to match your exact node/edge naming/attributes!

内容的提问来源于stack exchange，提问作者Syed Shahrukh Ali Gellani