本文目录导读：
我来详细介绍如何使用NetworkX做关键路径分析，关键路径法(CPM)主要用于项目管理中的时间管理和进度规划。
基本实现方法
安装所需库
pip install networkx matplotlib numpy
基本关键路径实现
import networkx as nx
import matplotlib.pyplot as plt
from collections import defaultdict
def create_activity_network():
"""创建活动网络图"""
G = nx.DiGraph()
# 添加活动节点（活动名称，持续时间）
activities = {
'A': {'duration': 3},
'B': {'duration': 2},
'C': {'duration': 4},
'D': {'duration': 5},
'E': {'duration': 3},
'F': {'duration': 6},
'G': {'duration': 2}
}
# 添加节点
for act, attrs in activities.items():
G.add_node(act, **attrs)
# 添加依赖关系（前驱活动 -> 后继活动）
dependencies = [
('start', 'A', 0), # 开始节点
('start', 'B', 0),
('A', 'C', 0),
('A', 'D', 0),
('B', 'D', 0),
('B', 'E', 0),
('C', 'F', 0),
('D', 'F', 0),
('E', 'G', 0),
('F', 'end', 0),
('G', 'end', 0)
]
for pred, succ, weight in dependencies:
G.add_edge(pred, succ, weight=weight)
# 添加虚拟开始和结束节点的持续时间
G.nodes['start']['duration'] = 0
G.nodes['end']['duration'] = 0
return G
def calculate_critical_path(G):
"""计算关键路径"""
# 拓扑排序
topological_order = list(nx.topological_sort(G))
# 计算最早开始时间(ES)和最早完成时间(EF)
early_start = {node: 0 for node in G.nodes()}
early_finish = {node: 0 for node in G.nodes()}
for node in topological_order:
predecessors = list(G.predecessors(node))
if predecessors:
early_start[node] = max(early_finish[pred] for pred in predecessors)
early_finish[node] = early_start[node] + G.nodes[node]['duration']
total_duration = early_finish['end']
# 计算最晚开始时间(LS)和最晚完成时间(LF)
late_finish = {node: total_duration for node in G.nodes()}
late_start = {node: total_duration for node in G.nodes()}
for node in reversed(topological_order):
successors = list(G.successors(node))
if successors:
late_finish[node] = min(late_start[succ] for succ in successors)
late_start[node] = late_finish[node] - G.nodes[node]['duration']
# 计算总时差(TF = LS - ES = LF - EF)
total_float = {}
for node in G.nodes():
total_float[node] = late_start[node] - early_start[node]
# 关键路径上的节点（总时差为0）
critical_path_nodes = [node for node in G.nodes()
if node not in ['start', 'end'] and total_float[node] == 0]
return {
'early_start': early_start,
'early_finish': early_finish,
'late_start': late_start,
'late_finish': late_finish,
'total_float': total_float,
'critical_path_nodes': critical_path_nodes,
'total_duration': total_duration,
'topological_order': topological_order
}
def find_critical_path_sequence(G, critical_info):
"""找到完整的关键路径序列"""
critical_nodes = set(critical_info['critical_path_nodes'])
# 从start开始，沿着关键路径走
path = ['start']
current = 'start'
while current != 'end':
successors = list(G.successors(current))
# 选择关键路径上的后继节点
next_nodes = [n for n in successors if n in critical_nodes or n == 'end']
if next_nodes:
current = next_nodes[0]
path.append(current)
else:
break
return path
def visualize_critical_path(G, critical_info):
"""可视化关键路径"""
pos = nx.spring_layout(G, seed=42)
# 创建节点颜色列表
node_colors = []
for node in G.nodes():
if node in ['start', 'end']:
node_colors.append('gray')
elif node in critical_info['critical_path_nodes']:
node_colors.append('red')
else:
node_colors.append('lightblue')
# 确保节点标签显示持续时间
node_labels = {}
for node in G.nodes():
dur = G.nodes[node].get('duration', 0)
if node in ['start', 'end']:
node_labels[node] = f"{node}"
else:
node_labels[node] = f"{node}\n({dur}d)"
# 创建边颜色列表
critical_path_set = set(critical_info['critical_path_nodes'])
edge_colors = []
edge_labels = {}
for u, v, data in G.edges(data=True):
if u in critical_path_set and v in critical_path_set:
edge_colors.append('red')
else:
edge_colors.append('gray')
plt.figure(figsize=(12, 8))
# 绘制节点
nx.draw_networkx_nodes(G, pos, node_color=node_colors,
node_size=2000, alpha=0.8)
# 绘制边
nx.draw_networkx_edges(G, pos, edge_color=edge_colors,
arrows=True, arrowsize=20,
width=2)
# 绘制标签
nx.draw_networkx_labels(G, pos, node_labels, font_size=10)
plt.title(f"关键路径分析\n总工期: {critical_info['total_duration']}天\n"
f"关键路径: {' -> '.join(critical_info['critical_path_nodes'])}",
fontsize=14)
plt.axis('off')
plt.tight_layout()
plt.show()
# 使用示例
G = create_activity_network()
critical_info = calculate_critical_path(G)
print("=== 关键路径分析结果 ===")
print(f"总工期: {critical_info['total_duration']}天")
print(f"关键路径上的活动: {', '.join(critical_info['critical_path_nodes'])}")
print(f"拓扑排序顺序: {', '.join(critical_info['topological_order'])}")
print("\n=== 各活动时间参数 ===")
print(f"{'活动':<8} {'最早开始':<10} {'最早完成':<10} {'最晚开始':<10} {'最晚完成':<10} {'总时差':<8}")
print("-" * 65)
for node in critical_info['early_start']:
if node not in ['start', 'end']:
es = critical_info['early_start'][node]
ef = critical_info['early_finish'][node]
ls = critical_info['late_start'][node]
lf = critical_info['late_finish'][node]
tf = critical_info['total_float'][node]
critical_mark = " ★" if tf == 0 else ""
print(f"{node:<8} {es:<10} {ef:<10} {ls:<10} {lf:<10} {tf:<8}{critical_mark}")
# 可视化
visualize_critical_path(G, critical_info)
增强版关键路径实现
class CriticalPathAnalyzer:
"""关键路径分析器"""
def __init__(self):
self.graph = nx.DiGraph()
self.results = {}
def add_activity(self, name, duration, predecessors=None):
"""添加活动"""
if predecessors is None:
predecessors = []
self.graph.add_node(name, duration=duration)
for pred in predecessors:
self.graph.add_edge(pred, name)
def add_dummy_start_end(self):
"""添加虚拟开始和结束节点"""
# 找到没有前驱的节点
start_nodes = [n for n in self.graph.nodes()
if self.graph.in_degree(n) == 0]
# 找到没有后继的节点
end_nodes = [n for n in self.graph.nodes()
if self.graph.out_degree(n) == 0]
# 添加虚拟节点
self.graph.add_node('start', duration=0)
self.graph.add_node('end', duration=0)
# 连接
for node in start_nodes:
self.graph.add_edge('start', node)
for node in end_nodes:
self.graph.add_edge(node, 'end')
def analyze(self):
"""执行关键路径分析"""
self.add_dummy_start_end()
# 确保是有向无环图
if not nx.is_directed_acyclic_graph(self.graph):
raise ValueError("图中存在环，不能进行关键路径分析")
# 拓扑排序
topological_order = list(nx.topological_sort(self.graph))
# 正向传递：计算最早时间
early_start = {}
early_finish = {}
for node in topological_order:
predecessors = list(self.graph.predecessors(node))
if not predecessors:
early_start[node] = 0
else:
early_start[node] = max(early_finish[p] for p in predecessors)
early_finish[node] = early_start[node] + \
self.graph.nodes[node].get('duration', 0)
total_duration = early_finish['end']
# 反向传递：计算最晚时间
late_finish = {}
late_start = {}
for node in reversed(topological_order):
successors = list(self.graph.successors(node))
if not successors:
late_finish[node] = total_duration
else:
late_finish[node] = min(late_start[s] for s in successors)
late_start[node] = late_finish[node] - \
self.graph.nodes[node].get('duration', 0)
# 计算总时差和自由时差
total_float = {}
free_float = {}
for node in topological_order:
total_float[node] = late_start[node] - early_start[node]
successors = list(self.graph.successors(node))
if not successors:
free_float[node] = 0
else:
successor_es = [early_start[s] for s in successors]
free_float[node] = min(successor_es) - early_finish[node]
# 识别关键路径
critical_nodes = [n for n in self.graph.nodes()
if n not in ['start', 'end'] and total_float[n] == 0]
# 构建关键路径序列
critical_path = ['start']
current = 'start'
while current != 'end':
successors = list(self.graph.successors(current))
next_nodes = [s for s in successors
if s in critical_nodes or s == 'end']
if next_nodes:
current = next_nodes[0]
critical_path.append(current)
else:
break
# 存储结果
self.results = {
'topological_order': topological_order,
'early_start': early_start,
'early_finish': early_finish,
'late_start': late_start,
'late_finish': late_finish,
'total_float': total_float,
'free_float': free_float,
'critical_nodes': critical_nodes,
'critical_path': critical_path,
'total_duration': total_duration
}
return self.results
def print_report(self):
"""打印分析报告"""
if not self.results:
print("请先运行 analyze() 方法")
return
print("=" * 70)
print("关键路径分析报告")
print("=" * 70)
print(f"\n总工期: {self.results['total_duration']} 天")
print(f"关键路径: {' → '.join(self.results['critical_path'][1:-1])}")
print("\n" + "-" * 70)
print(f"{'活动':<8} {'工期':<6} {'ES':<6} {'EF':<6} {'LS':<6} {'LF':<6} {'TF':<6} {'FF':<6}")
print("-" * 70)
for node in self.results['topological_order']:
if node not in ['start', 'end']:
duration = self.graph.nodes[node].get('duration', 0)
es = self.results['early_start'][node]
ef = self.results['early_finish'][node]
ls = self.results['late_start'][node]
lf = self.results['late_finish'][node]
tf = self.results['total_float'][node]
ff = self.results['free_float'][node]
marker = " ★" if tf == 0 else ""
print(f"{node:<8} {duration:<6} {es:<6} {ef:<6} {ls:<6} {lf:<6} {tf:<6} {ff:<6}{marker}")
print("-" * 70)
# 使用增强版
analyzer = CriticalPathAnalyzer()
# 添加活动（模拟一个项目）
analyzer.add_activity('A', 3, []) # 起始活动
analyzer.add_activity('B', 4, ['A']) # 依赖A
analyzer.add_activity('C', 2, ['A']) # 依赖A
analyzer.add_activity('D', 5, ['B']) # 依赖B
analyzer.add_activity('E', 3, ['B', 'C']) # 依赖B和C
analyzer.add_activity('F', 6, ['D', 'E']) # 依赖D和E
analyzer.add_activity('G', 2, ['F']) # 最终活动
# 执行分析
analyzer.analyze()
analyzer.print_report()
实用功能扩展
def analyze_project_schedule(critical_info):
"""项目进度优化建议"""
print("\n=== 进度优化建议 ===")
# 分析关键路径瓶颈
critical_duration = 0
for node in critical_info['critical_path_nodes']:
critical_duration += G.nodes[node].get('duration', 0)
print(f"关键路径总工期: {critical_duration}天")
# 找出可以压缩的活动
compressible = []
for node in critical_info['critical_path_nodes']:
duration = G.nodes[node].get('duration', 0)
if duration > 1: # 可以压缩的活动
compressible.append((node, duration))
if compressible:
print("\n可压缩的关键活动:")
for node, dur in sorted(compressible, key=lambda x: x[1], reverse=True):
print(f" {node}: {dur}天 → 压缩{min(dur-1, 1)}天")
# 分析浮动时间
float_percentage = {}
for node in critical_info['total_float']:
if node not in ['start', 'end']:
duration = G.nodes[node].get('duration', 0)
tf = critical_info['total_float'][node]
if duration > 0:
float_percentage[node] = (tf / duration) * 100
print("\n资源松弛度分析:")
for node, fp in sorted(float_percentage.items()):
if fp > 0:
status = "充裕" if fp > 30 else "适中"
print(f" {node}: {fp:.1f}% ({status})")
# 运行分析
analyze_project_schedule(critical_info)
这个实现提供了完整的关键路径分析功能,包括：
- 最早/最晚开始和完成时间计算
- 总时差和自由时差计算
- 关键路径识别
- 网络图可视化
- 进度优化建议
你可以根据需要调整活动依赖关系和持续时间来适应不同的项目场景。
