Python案例如何用Networkx做二分图匹配

wen python案例 2026-07-15 192

本文目录导读：

基础二分图匹配
带权重的二分图匹配 (任务分配问题)
实际应用：任务分配系统
可视化匹配结果
评估匹配质量

我来详细讲解如何使用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}")

这些示例展示了：