本文目录导读：
我来为你详细介绍如何使用NetworkX解决各种流问题,包括最大流、最小费用流等经典案例。
安装和导入NetworkX
import networkx as nx import matplotlib.pyplot as plt import numpy as np # 设置中文显示 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False
最大流问题案例
案例1：基础网络最大流
# 创建有向图
G = nx.DiGraph()
# 添加边（源点，终点，容量）
edges = [
('S', 'A', 10), # S->A 容量10
('S', 'B', 5), # S->B 容量5
('A', 'B', 15), # A->B 容量15
('A', 'C', 10), # A->C 容量10
('A', 'D', 5), # A->D 容量5
('B', 'C', 5), # B->C 容量5
('B', 'D', 10), # B->D 容量10
('C', 'T', 10), # C->T 容量10
('D', 'T', 10) # D->T 容量10
]
# 添加边的容量属性
for u, v, capacity in edges:
G.add_edge(u, v, capacity=capacity)
# 计算最大流
flow_value, flow_dict = nx.maximum_flow(G, 'S', 'T')
print(f"最大流值: {flow_value}")
print("\n各边的流量分配:")
for u in flow_dict:
for v, flow in flow_dict[u].items():
capacity = G[u][v]['capacity']
print(f"{u} -> {v}: 流量 = {flow}, 容量 = {capacity}")
# 可视化网络
def draw_flow_network(G, flow_dict):
pos = nx.spring_layout(G, seed=42)
# 绘制节点
nx.draw_networkx_nodes(G, pos, node_color='lightblue',
node_size=500)
nx.draw_networkx_labels(G, pos)
# 绘制边
edge_labels = {}
for u, v, data in G.edges(data=True):
capacity = data['capacity']
flow = flow_dict[u][v]
edge_labels[(u, v)] = f"{flow}/{capacity}"
# 根据利用率改变颜色
utilization = flow / capacity if capacity > 0 else 0
if utilization > 0.8:
color = 'red'
elif utilization > 0.5:
color = 'orange'
else:
color = 'green'
nx.draw_networkx_edges(G, pos, edgelist=[(u, v)],
edge_color=color, width=2)
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)
plt.title(f"最大流网络 (总流量 = {sum(flow_dict['S'].values())})")
plt.show()
draw_flow_network(G, flow_dict)
案例2：多源多汇最大流
def solve_multi_source_sink_flow():
"""解决多源多汇问题"""
G = nx.DiGraph()
# 原始网络
edges = [
('A', 'C', 10), ('A', 'D', 8),
('B', 'C', 5), ('B', 'D', 12),
('C', 'E', 9), ('C', 'F', 7),
('D', 'E', 6), ('D', 'F', 8),
('E', 'G', 12), ('F', 'G', 10)
]
# 添加超源点和超汇点
G.add_edge('SuperSource', 'A', capacity=100)
G.add_edge('SuperSource', 'B', capacity=100)
G.add_edge('E', 'SuperSink', capacity=100)
G.add_edge('F', 'SuperSink', capacity=100)
# 添加原始边
for u, v, cap in edges:
G.add_edge(u, v, capacity=cap)
# 计算最大流
flow_value, flow_dict = nx.maximum_flow(G, 'SuperSource', 'SuperSink')
print(f"多源多汇最大流值: {flow_value}")
# 只显示原始网络的流量
print("\n原始网络流量分配:")
for u, v, cap in edges:
if v in flow_dict.get(u, {}):
print(f"{u} -> {v}: 流量 = {flow_dict[u][v]}, 容量 = {cap}")
return flow_value, flow_dict
solve_multi_source_sink_flow()
最小费用流问题案例
案例3：运输问题
def solve_transportation_problem():
"""解决运输问题（最小费用流）"""
G = nx.DiGraph()
# 工厂（供应节点）
factories = {
'F1': 40, # 供应量
'F2': 50 # 供应量
}
# 仓库（需求节点）
warehouses = {
'W1': 30, # 需求量
'W2': 35, # 需求量
'W3': 25 # 需求量
}
# 运输成本（工厂->仓库）
costs = {
('F1', 'W1'): 3,
('F1', 'W2'): 5,
('F1', 'W3'): 7,
('F2', 'W1'): 4,
('F2', 'W2'): 6,
('F2', 'W3'): 8
}
# 添加超源点和超汇点
G.add_edge('Source', 'F1', capacity=factories['F1'], weight=0)
G.add_edge('Source', 'F2', capacity=factories['F2'], weight=0)
G.add_edge('W1', 'Sink', capacity=warehouses['W1'], weight=0)
G.add_edge('W2', 'Sink', capacity=warehouses['W2'], weight=0)
G.add_edge('W3', 'Sink', capacity=warehouses['W3'], weight=0)
# 添加运输边
for (f, w), cost in costs.items():
min_capacity = min(factories[f], warehouses[w])
G.add_edge(f, w, capacity=min_capacity, weight=cost)
# 计算最小费用流
flow_cost, flow_dict = nx.network_simplex(G)
# 计算实际流量
total_flow = sum(flow_dict['Source'].values())
print(f"最小运输费用: {flow_cost}")
print(f"总运输量: {total_flow}")
print("\n运输方案:")
for (f, w), cost in costs.items():
flow = flow_dict.get(f, {}).get(w, 0)
if flow > 0:
print(f"{f} -> {w}: {flow}单位, 成本={cost}/单位, 总成本={flow*cost}")
return flow_cost, flow_dict
solve_transportation_problem()
实际应用案例
案例4：通信网络带宽分配
def solve_bandwidth_allocation():
"""通信网络带宽分配问题"""
G = nx.DiGraph()
# 网络拓扑（路由器之间的可用带宽）
links = [
('R1', 'R2', 100), # R1->R2 100Mbps
('R1', 'R3', 50), # R1->R3 50Mbps
('R2', 'R3', 80), # R2->R3 80Mbps
('R2', 'R4', 70), # R2->R4 70Mbps
('R3', 'R4', 60), # R3->R4 60Mbps
('R3', 'R5', 90), # R3->R5 90Mbps
('R4', 'R5', 50) # R4->R5 50Mbps
]
# 添加边（带宽作为容量）
for u, v, bandwidth in links:
G.add_edge(u, v, capacity=bandwidth)
G.add_edge(v, u, capacity=bandwidth) # 双向链路
# 假设服务器在R1，客户端在R5
# 需要分配带宽给多个应用
apps = ['视频流', '文件传输', '远程桌面']
demands = {app: {'min': 20, 'max': 40} for app in apps}
# 计算最大有效带宽
flow_value, flow_dict = nx.maximum_flow(G, 'R1', 'R5')
print(f"R1到R5的最大可用带宽: {flow_value} Mbps")
print("\n带宽分配建议:")
total_bandwidth = flow_value
total_demand_min = sum(d['min'] for d in demands.values())
total_demand_max = sum(d['max'] for d in demands.values())
for app, demand in demands.items():
if total_demand_max <= total_bandwidth:
allocated = demand['max']
elif total_demand_min <= total_bandwidth:
# 按比例分配
proportion = demand['min'] / total_demand_min
allocated = min(demand['max'],
demand['min'] + proportion * (total_bandwidth - total_demand_min))
else:
allocated = demand['min'] * (total_bandwidth / total_demand_min)
print(f" {app}: {allocated:.1f} Mbps")
return flow_value, flow_dict
solve_bandwidth_allocation()
案例5：排水系统网络
def solve_drainage_system():
"""城市排水系统容量分析"""
G = nx.DiGraph()
# 排水管道（节点，节点，最大容量m³/s）
pipes = [
('入口1', '分流1', 50),
('入口2', '分流1', 30),
('入口3', '分流2', 40),
('分流1', '分流2', 20),
('分流1', '管道1', 60),
('分流2', '管道1', 30),
('分流2', '管道2', 50),
('管道1', '处理厂1', 70),
('管道2', '处理厂2', 40),
('处理厂1', '出口', 80),
('处理厂2', '出口', 60)
]
for u, v, capacity in pipes:
G.add_edge(u, v, capacity=capacity)
# 计算从入口到出口的最大排水能力
flow_value, flow_dict = nx.maximum_flow(G, '入口1', '出口')
print("排水系统容量分析:")
print(f"最大排水能力: {flow_value} m³/s")
# 识别瓶颈
print("\n瓶颈分析（高利用率管道）:")
for u, v, data in G.edges(data=True):
if v in flow_dict.get(u, {}):
utilization = flow_dict[u][v] / data['capacity']
if utilization > 0.8:
print(f" {u} -> {v}: 利用率 {utilization:.1%} [瓶颈]")
elif utilization > 0.5:
print(f" {u} -> {v}: 利用率 {utilization:.1%}")
return flow_value, flow_dict
solve_drainage_system()
高级功能
多重流分析
def multiple_flow_analysis():
"""分析不同源汇组合的流量"""
G = nx.DiGraph()
# 构建一个中型网络
nodes = ['A', 'B', 'C', 'D', 'E', 'F']
edges = [
('A', 'B', 20), ('A', 'C', 15),
('B', 'C', 10), ('B', 'D', 25),
('C', 'D', 15), ('C', 'E', 20),
('D', 'E', 10), ('D', 'F', 20),
('E', 'F', 25)
]
for u, v, cap in edges:
G.add_edge(u, v, capacity=cap)
G.add_edge(v, u, capacity=cap) # 双向
# 分析不同的源汇对
source_sink_pairs = [('A', 'F'), ('A', 'E'), ('B', 'F'), ('C', 'F')]
print("不同源汇对的最大流:")
for source, sink in source_sink_pairs:
flow_value = nx.maximum_flow_value(G, source, sink)
print(f" {source} -> {sink}: {flow_value}")
# 找到所有节点对之间的最大流
print("\n容量矩阵（源\\汇）:")
nodes_sorted = sorted(nodes)
print(f"{'':>6}", end="")
for sink in nodes_sorted:
print(f"{sink:>6}", end="")
print()
for source in nodes_sorted:
print(f"{source:>6}", end="")
for sink in nodes_sorted:
if source != sink:
flow = nx.maximum_flow_value(G, source, sink)
print(f"{flow:>6}", end="")
else:
print(f"{' -':>6}", end="")
print()
multiple_flow_analysis()
可视化辅助函数
def visualize_flow_with_animation(G, flow_dict, pos=None):
"""动态可视化流量分配"""
if pos is None:
pos = nx.spring_layout(G, seed=42)
fig, axes = plt.subplots(1, 2, figsize=(15, 6))
# 原始网络
ax1 = axes[0]
nx.draw_networkx_nodes(G, pos, ax=ax1, node_color='lightblue',
node_size=500)
nx.draw_networkx_labels(G, pos, ax=ax1)
edge_labels_orig = {(u, v): f"cap={d['capacity']}"
for u, v, d in G.edges(data=True)}
nx.draw_networkx_edges(G, pos, ax=ax1)
nx.draw_networkx_edge_labels(G, pos, edge_labels_orig, ax=ax1)
ax1.set_title("原始网络（容量）")
# 流量网络
ax2 = axes[1]
nx.draw_networkx_nodes(G, pos, ax=ax2, node_color='lightgreen',
node_size=500)
nx.draw_networkx_labels(G, pos, ax=ax2)
# 根据流量着色
colors = []
edge_labels_flow = {}
for u, v, d in G.edges(data=True):
flow = flow_dict.get(u, {}).get(v, 0)
utilization = flow / d['capacity'] if d['capacity'] > 0 else 0
if utilization > 0.8:
colors.append('red')
elif utilization > 0.5:
colors.append('orange')
else:
colors.append('green')
edge_labels_flow[(u, v)] = f"{flow}/{d['capacity']}"
nx.draw_networkx_edges(G, pos, ax=ax2, edge_color=colors, width=3)
nx.draw_networkx_edge_labels(G, pos, edge_labels_flow, ax=ax2)
ax2.set_title("流量分配（流量/容量）")
plt.tight_layout()
plt.show()
# 使用示例
G = nx.DiGraph()
edges = [('S', 'A', 10), ('S', 'B', 5), ('A', 'B', 15),
('A', 'T', 10), ('B', 'T', 10)]
for u, v, cap in edges:
G.add_edge(u, v, capacity=cap)
flow_value, flow_dict = nx.maximum_flow(G, 'S', 'T')
visualize_flow_with_animation(G, flow_dict)
这些案例展示了NetworkX在处理流问题时的强大功能,包括：
- 最大流问题：基础网络、多源多汇
- 最小费用流：运输问题、资源分配
- 实际应用：带宽分配、排水系统
- 高级分析：多重流、瓶颈识别
- 可视化：流量分配展示
通过这些案例,你可以快速上手使用NetworkX解决各种实际的流问题。
