基于Python的物品分配算法求解：带约束的用户偏好匹配问题

现有一组用户对物品的偏好排序（示例：3个用户对3个物品的偏好：User1: Item A > Item B > Item C；User2: Item B > Item A > Item C；User3: Item A > Item C > Item B），实际场景为25个用户与25个物品。需求是找到最优分配方案：整体让每个用户匹配到尽可能高排名的物品（无需最大化首选项数量），且每个物品仅能分配一次。

尝试使用NetworkX的单纯形算法时，遇到错误：networkx.exception.NetworkXUnfeasible: total node demand is not zero，附上相关代码示例。

核心问题 #

NetworkX的最小费用流要求所有节点的需求总和为0。你的代码中存在两个潜在问题：

1. sink节点被重复添加到循环内部，虽然最终需求值正确，但写法不规范，可能引发意外问题；
2. 用户到物品的边未显式设置容量，默认无限容量可能导致逻辑隐患。

修正后的代码片段 #

# 修正节点需求设置：sink节点仅添加一次
for p in P:
    G.add_node(p[0], demand=-1)
# 仅在循环外添加sink节点，需求为用户总数
G.add_node('sink', demand=len(P))

# 修正用户到物品的边：显式设置容量为1
for p, c in itertools.product(P, C):
    G.add_edge(p[0], c[0], weight=get_cost(p[1], c[1]), capacity=1)

同时需确保用户数量与物品数量严格相等（25=25），否则即使总需求为0，也会因物品容量不足导致无法分配。

你的问题属于等规模双边匹配问题，目标是最小化整体匹配代价（等价于最大化用户满意度），以下是两种高效实现方案：

1. 最小费用最大流（NetworkX实现） #

通过构建流网络，用network_simplex算法求解，适合需要扩展复杂约束的场景。

完整修正代码：

import networkx as nx
import itertools
import random

# 生成25个用户的偏好数据（示例：每个用户的偏好是1-25的随机排列）
P = []
for i in range(1, 26):
    ranking = random.sample(range(1, 26), 25)
    P.append((f'User{i}', ranking))

# 物品列表：(名称, 索引, 最大分配数)
no_items = 25
C = [('s' + str(i), i, 1) for i in range(1, no_items + 1)]

# 计算匹配代价：排名越靠后，代价越高（平方惩罚）
def get_cost(ranking, index):
    return ranking.index(index) ** 2

# 构建最小费用流图
G = nx.DiGraph()

# 设置节点需求
for p in P:
    G.add_node(p[0], demand=-1)
G.add_node('sink', demand=len(P))

# 添加用户到物品的边
for p, c in itertools.product(P, C):
    G.add_edge(p[0], c[0], weight=get_cost(p[1], c[1]), capacity=1)

# 添加物品到sink的边
for c in C:
    G.add_edge(c[0], 'sink', capacity=c[2])

# 求解并输出结果
try:
    flow_cost, flow_dict = nx.network_simplex(G)
    print('总代价: ', flow_cost)
    print('\n分配方案：')
    for user in P:
        user_name = user[0]
        for item in C:
            item_name = item[0]
            if flow_dict[user_name].get(item_name, 0) == 1:
                print(f'{user_name} -> {item_name}')
except nx.NetworkXUnfeasible:
    print('无法找到可行分配方案，请检查用户与物品数量是否相等。')

2. 匈牙利算法（Scipy实现） #

针对等规模双边匹配，匈牙利算法（Kuhn-Munkres）时间复杂度为O(n³)，对于n=25的场景更高效直接。

代码示例：

import numpy as np
from scipy.optimize import linear_sum_assignment
import random

# 生成25个用户的偏好数据
P = []
for i in range(1, 26):
    ranking = random.sample(range(1, 26), 25)
    P.append((f'User{i}', ranking))

# 构建代价矩阵：行=用户，列=物品，值=用户对该物品的匹配代价
cost_matrix = []
for user in P:
    ranking = user[1]
    row = [ranking.index(item_idx) ** 2 for item_idx in range(1, 26)]
    cost_matrix.append(row)
cost_matrix = np.array(cost_matrix)

# 求解最小权匹配
user_indices, item_indices = linear_sum_assignment(cost_matrix)

# 输出结果
print('总代价: ', cost_matrix[user_indices, item_indices].sum())
print('\n分配方案：')
for u_idx, i_idx in zip(user_indices, item_indices):
    user_name = P[u_idx][0]
    item_name = f's{i_idx+1}'  # 对应物品名称
    print(f'{user_name} -> {item_name}')

• 代价函数：使用平方惩罚ranking.index(index)**2会放大低排名物品的代价，确保优先匹配用户偏好靠前的物品；若需强化首选项权重，可改用指数惩罚（如2**ranking.index(index)）。
• 可行性：只要用户数等于物品数，且每个用户可匹配任意物品（无论排名），就一定存在可行分配方案。

内容的提问来源于stack exchange，提问作者Johnpojohn