本文目录导读：
我来详细讲解如何使用NetworkX进行二分图匹配,包含多个实例：
基础二分图匹配
import networkx as nx
import matplotlib.pyplot as plt
# 创建二分图
B = nx.Graph()
# 添加节点（使用不同前缀区分两组节点）
left_nodes = ['A1', 'A2', 'A3']
right_nodes = ['B1', 'B2', 'B3']
# 添加节点组
B.add_nodes_from(left_nodes, bipartite=0) # 第一组
B.add_nodes_from(right_nodes, bipartite=1) # 第二组
# 添加边
edges = [('A1', 'B1'), ('A1', 'B2'),
('A2', 'B1'), ('A2', 'B3'),
('A3', 'B2')]
B.add_edges_from(edges)
# 可视化二分图
def draw_bipartite_graph(G):
pos = nx.bipartite_layout(G, [n for n, d in G.nodes(data=True) if d['bipartite'] == 0])
plt.figure(figsize=(8, 4))
# 分离两组节点以便着色
left = [n for n, d in G.nodes(data=True) if d['bipartite'] == 0]
right = [n for n, d in G.nodes(data=True) if d['bipartite'] == 1]
nx.draw_networkx_nodes(G, pos, nodelist=left, node_color='lightblue',
node_size=500, label='Left Set')
nx.draw_networkx_nodes(G, pos, nodelist=right, node_color='lightgreen',
node_size=500, label='Right Set')
nx.draw_networkx_edges(G, pos, alpha=0.5)
nx.draw_networkx_labels(G, pos)
plt.title("Bipartite Graph")
plt.axis('off')
plt.show()
draw_bipartite_graph(B)
# 执行最大匹配
matching = nx.max_weight_matching(B, maxcardinality=True)
print("最大匹配:", matching)
print("匹配数量:", len(matching))
带权重的二分图匹配 (任务分配问题)
import networkx as nx
# 创建带权二分图
G = nx.Graph()
# 定义工人和任务
workers = ['W1', 'W2', 'W3']
tasks = ['T1', 'T2', 'T3']
# 添加节点
G.add_nodes_from(workers, bipartite=0)
G.add_nodes_from(tasks, bipartite=1)
# 添加带权重的边（效益矩阵）
weighted_edges = [
('W1', 'T1', 5),
('W1', 'T2', 3),
('W1', 'T3', 4),
('W2', 'T1', 2),
('W2', 'T2', 6),
('W2', 'T3', 1),
('W3', 'T1', 4),
('W3', 'T2', 2),
('W3', 'T3', 7)
]
G.add_weighted_edges_from(weighted_edges)
# 最大权重匹配（最大化总效益）
max_matching = nx.max_weight_matching(G, maxcardinality=True)
print("最大权重匹配:")
for edge in max_matching:
print(f" {edge[0]} - {edge[1]}: 权重 {G[edge[0]][edge[1]]['weight']}")
total_weight = sum(G[u][v]['weight'] for u, v in max_matching)
print(f"总权重: {total_weight}")
# 最小权重匹配（最小化成本）
min_matching = nx.min_weight_matching(G)
print("\n最小权重匹配:")
for edge in min_matching:
print(f" {edge[0]} - {edge[1]}: 权重 {G[edge[0]][edge[1]]['weight']}")
total_min_weight = sum(G[u][v]['weight'] for u, v in min_matching)
print(f"总最小权重: {total_min_weight}")
实际应用：任务分配系统
import networkx as nx
import pandas as pd
from datetime import datetime
class TaskAssignmentSystem:
def __init__(self):
self.graph = nx.Graph()
self.workers = []
self.tasks = []
def add_worker(self, worker_id, skills, availability):
"""添加工人节点"""
self.workers.append(worker_id)
self.graph.add_node(worker_id,
bipartite=0,
skills=skills,
availability=availability,
type='worker')
def add_task(self, task_id, required_skills, deadline, priority):
"""添加任务节点"""
self.tasks.append(task_id)
self.graph.add_node(task_id,
bipartite=1,
required_skills=required_skills,
deadline=deadline,
priority=priority,
type='task')
def calculate_match_score(self, worker_id, task_id):
"""计算工人与任务的匹配分数"""
worker = self.graph.nodes[worker_id]
task = self.graph.nodes[task_id]
# 技能匹配分
skills_score = len(set(worker['skills']) & set(task['required_skills']))
max_skills = max(len(worker['skills']), len(task['required_skills']))
skills_ratio = skills_score / max_skills if max_skills > 0 else 0
# 优先级加成
priority_bonus = task['priority'] * 0.1
# 截止日期紧迫度
deadline_days = (task['deadline'] - datetime.now()).days
deadline_factor = max(0, 1 - deadline_days / 30) # 30天内
# 综合分数
total_score = skills_ratio * 0.6 + priority_bonus * 0.2 + deadline_factor * 0.2
return round(total_score, 2)
def build_match_graph(self):
"""构建匹配图"""
for worker in self.workers:
for task in self.tasks:
score = self.calculate_match_score(worker, task)
if score > 0.3: # 只添加有意义的匹配
self.graph.add_edge(worker, task, weight=score, match_score=score)
def find_optimal_assignment(self):
"""找到最优分配"""
self.build_match_graph()
# 最大权重匹配
matching = nx.max_weight_matching(self.graph, maxcardinality=True)
assignments = []
for w, t in matching:
if w in self.workers and t in self.tasks:
score = self.graph[w][t]['weight']
assignments.append({
'worker': w,
'task': t,
'score': score,
'match_details': {
'worker_skills': self.graph.nodes[w]['skills'],
'task_skills': self.graph.nodes[t]['required_skills'],
'deadline': self.graph.nodes[t]['deadline'],
'priority': self.graph.nodes[t]['priority']
}
})
return assignments
# 使用示例
assignment_system = TaskAssignmentSystem()
# 添加工人
assignment_system.add_worker('W1', ['Python', 'ML', 'SQL'], True)
assignment_system.add_worker('W2', ['Java', 'Spring', 'SQL'], True)
assignment_system.add_worker('W3', ['Python', 'Django', 'React'], True)
# 添加任务
assignment_system.add_task('T1', ['Python', 'ML'],
datetime(2024, 12, 1), priority=5)
assignment_system.add_task('T2', ['Java', 'SQL'],
datetime(2024, 11, 15), priority=3)
assignment_system.add_task('T3', ['Python', 'Django'],
datetime(2024, 11, 20), priority=4)
# 找到最优分配
assignments = assignment_system.find_optimal_assignment()
print("最优任务分配方案:")
print("-" * 50)
for assign in assignments:
print(f"工人 {assign['worker']} → 任务 {assign['task']}")
print(f"匹配分数: {assign['score']}")
print(f"工人技能: {assign['match_details']['worker_skills']}")
print(f"任务需求: {assign['match_details']['task_skills']}")
print(f"截止日期: {assign['match_details']['deadline'].strftime('%Y-%m-%d')}")
print(f"优先级: {assign['match_details']['priority']}")
print("-" * 50)
# 计算总体匹配质量
total_score = sum(a['score'] for a in assignments)
avg_score = total_score / len(assignments) if assignments else 0
print(f"总匹配分数: {total_score}")
print(f"平均匹配质量: {avg_score:.2f}")
可视化匹配结果
import networkx as nx
import matplotlib.pyplot as plt
def visualize_matching(G, matching, title="Bipartite Graph Matching"):
"""可视化二分图匹配结果"""
plt.figure(figsize=(10, 6))
# 分离两组节点
left = [n for n, d in G.nodes(data=True) if d.get('bipartite') == 0 or
'bipartite' not in d and n.startswith(('W', 'A'))]
right = [n for n, d in G.nodes(data=True) if d.get('bipartite') == 1 or
'bipartite' not in d and n.startswith(('T', 'B'))]
# 布局
pos = nx.bipartite_layout(G, left)
# 绘制所有边
nx.draw_networkx_edges(G, pos, alpha=0.3, edge_color='gray')
# 突出显示匹配边
if matching:
matching_edges = [(u, v) for u, v in matching]
nx.draw_networkx_edges(G, pos, edgelist=matching_edges,
edge_color='red', width=2)
# 绘制节点
nx.draw_networkx_nodes(G, pos, nodelist=left, node_color='skyblue',
node_size=500, label='Left Set')
nx.draw_networkx_nodes(G, pos, nodelist=right, node_color='lightgreen',
node_size=500, label='Right Set')
# 绘制标签
nx.draw_networkx_labels(G, pos)
# 如果存在权重，显示权重标签
edge_labels = {(u, v): G[u][v].get('weight', '')
for u, v in G.edges() if 'weight' in G[u][v]}
if edge_labels:
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)
plt.title(title)
plt.legend()
plt.axis('off')
plt.tight_layout()
plt.show()
# 使用示例
G = nx.Graph()
G.add_nodes_from(['A1', 'A2', 'A3'], bipartite=0)
G.add_nodes_from(['B1', 'B2', 'B3'], bipartite=1)
G.add_weighted_edges_from([
('A1', 'B1', 3), ('A1', 'B2', 2),
('A2', 'B1', 4), ('A2', 'B3', 1),
('A3', 'B2', 5), ('A3', 'B3', 2)
])
matching = nx.max_weight_matching(G, maxcardinality=True)
visualize_matching(G, matching, "最大权重匹配示例")
评估匹配质量
def evaluate_matching_quality(G, matching):
"""评估匹配质量"""
if not matching:
return {"status": "无匹配"}
# 统计信息
matching_edges = [(u, v) for u, v in matching]
matching_weights = [G[u][v].get('weight', 1) for u, v in matching_edges]
# 计算指标
total_weight = sum(matching_weights)
avg_weight = total_weight / len(matching_weights) if matching_weights else 0
max_weight = max(matching_weights) if matching_weights else 0
min_weight = min(matching_weights) if matching_weights else 0
std_weight = np.std(matching_weights) if len(matching_weights) > 1 else 0
return {
"匹配数量": len(matching),
"总权重": total_weight,
"平均权重": round(avg_weight, 2),
"最大权重": max_weight,
"最小权重": min_weight,
"权重标准差": round(std_weight, 2),
"匹配覆盖度": f"{len(matching)}/{len(G.nodes()) // 2}"
}
# 使用示例
import numpy as np
quality = evaluate_matching_quality(G, matching)
print("匹配质量评估:")
for key, value in quality.items():
print(f" {key}: {value}")
这些示例展示了：
- 基本匹配：创建和可视化二分图
- 权重匹配：处理带权重的图
- 实际应用：任务分配系统
- 可视化：显示匹配结果
- 质量评估：分析匹配效果
你可以根据具体需求调整匹配算法和评估标准。
