支持单订单分配2名工人的生鲜电商订单调度指派问题求解问询

我们目前正在解决一家食品科技初创企业（生鲜电商）的派单问题，涉及任务（待配送订单）和工作人员（骑手/拣货员/全能工）两类主体，核心需求是高效完成订单到工作人员的分配，第一阶段我们优先优化CTE（click-to-eat，即用户下单到订单送达的总时长）。

我们发现部分场景下给单个订单分配2名工作人员比单人处理效率更高：拣货员熟悉门店布局可快速完成拣货，骑手配备配送工具可快速完成配送，即便考虑订单交接成本，整体效率也远高于单人全程处理。
我们调研后发现该问题和指派问题高度相似，可通过匈牙利算法求解，但经典指派问题要求「每个任务仅分配给1名工人，每名工人仅承接1个任务」，不符合我们可分配2人处理单订单的需求。

• 将拣货员A+全能工B的组合作为单独条目加入成本矩阵，但该方案无法再单独将全能工B加入矩阵，否则可能出现全能工B同时被分配到2个订单（既作为单独个体承接订单，又作为和拣货员A的组合承接另一订单）的冲突。
• 先后运行两次匈牙利算法：第一次分配拣货任务，第二次分配配送任务。该方案在绝大多数场景下可行，但偶尔会产出效率较低的分配结果，如有需要可补充具体示例。

我们检索了大量资料，暂未找到可直接适配该场景的解决方案方向，恳请各位提供相关参考资料或解决思路，非常感谢。

我们已编写该问题的暴力求解代码，可帮助理解问题逻辑，代码如下：

# constants
delivery_speed = np.array([5, 13]) # km per hour
delivery_distance = np.array([300, 2700])  # meters
flight_distance = np.array([500, 1900]) # meters время подлета
positions_amount = np.array([4, 8]) # number of positions in one order
assembly_speed = np.array([2, 3]) # minutes per position
transit_time = 5 * 60 # sec to transfer order
number_of_orders = 3 # number of orders in a batch
number_of_workers = 3 

# size of optimization matrix
matrix_size = max(number_of_workers, number_of_orders)


# maximum diagonal length for delivery and flight
max_length = np.sqrt(max(delivery_distance)**2/2)
max_flight_length = np.sqrt(max(flight_distance)**2/2)


# store positions
A = np.array([delivery_distance[1], delivery_distance[1]])
B = np.array([A[0] + max_length / 2, A[1]])

possible_order_position_x = np.array([-max_length/2, max_length]) + A[0]
possible_order_position_y = np.array([-max_length, max_length]) + A[1]
possible_courier_position_x = np.array([-max_flight_length/2, max_flight_length]) + A[0]
possible_courier_position_y = np.array([-max_flight_length, max_flight_length]) + A[1]


# generate random data 
def random_speed(speed_array):
    return np.random.randint(speed_array[0], speed_array[1]+1)

def location(possible_x, possible_y):
    return np.random.randint([possible_x[0], possible_y[0]],
                             [possible_x[1], possible_y[1]],
                             size=2)


def generate_couriers():
    # generate couriers 
    couriers = {}
    for courier in range(number_of_workers):
        couriers[courier] = {
            'position': location(possible_courier_position_x, possible_courier_position_y),
            'delivery_speed': random_speed(delivery_speed),
            'assembly_speed': random_speed(assembly_speed),
        }
    return couriers
    
couriers = generate_couriers()

store_location = {0: A, 1:B}

def generate_orders():
    # generate orders
    orders = {}
    for order in range(number_of_orders):
        orders[order] = {
            'number_of_positions': random_speed(positions_amount),
            'store_position': store_location[np.random.randint(2)],
            'customer_position': location(possible_order_position_x, possible_order_position_y)
        }
    return orders

orders = generate_orders()
orders


# functions to calculate assembly and delivery speed 
def travel_time(location_1, location_2, speed):
    # time to get from current location to store
    flight_distance = np.linalg.norm(location_1 - location_2)
    delivery_speed = 1000 / (60 * 60) * speed # meters per second
    return flight_distance / delivery_speed # seconds


def assembly_time(courier, order):
    flight_time = travel_time(courier['position'], order['store_position'], courier['delivery_speed'])
    assembly_time = courier['assembly_speed'] * order['number_of_positions'] * 60

    
    return int(flight_time + assembly_time)
assembly_time(couriers[0], orders[0])

def brute_force_solution():
    best_cte = np.inf
    best_combination = [[],[]]

    for first_phase in itertools.permutations(range(number_of_workers), number_of_orders):
        assembly_time_s = pd.Series(index = range(number_of_orders), dtype=float)
        for order, courier in enumerate(first_phase):
            assembly_time_s[order] = assembly_time(couriers[courier], orders[order]) 


        # start to work with delivery
        for second_phase in itertools.permutations(range(number_of_workers), number_of_orders):
            delivery_time_s = pd.Series(index = range(number_of_orders), dtype=float)
            for order, courier in enumerate(second_phase):
                delivery_time = travel_time(orders[order]['store_position'],
                                            orders[order]['customer_position'],
                                            couriers[courier]['delivery_speed'])
                # different cases for different deliveries
                if courier == first_phase[order]:
                    # if courier assemblied order, then deliver immidietely 
                    delivery_time_s[order] = delivery_time
                elif courier not in first_phase:
                    # flight during assembly, wait if needed, transfer time, delivery
                    flight_time = travel_time(orders[order]['store_position'],
                                              couriers[courier]['position'],
                                              couriers[courier]['delivery_speed'])
                    wait_time = max(flight_time - assembly_time_s[order], 0)
                    delivery_time_s[order] = transit_time + wait_time + delivery_time
                else: 
                    # case when shopper transfers her order and moves deliver other 
                    # check if second order is in the same store
                    first_phase_order = first_phase.index(courier)
                    if (orders[first_phase_order]['store_position'] == orders[order]['store_position']).all():
                        # transit time - fixed and happens only once! 
                        # wait, if second order has not been assemblied yet
                        # time to assembly assigned order
                        assembly_own = assembly_time_s[first_phase_order]
                        # time to wait, if order to deliver is assemblied slower
                        wait_time = max(assembly_time_s[order] - assembly_own, 0)
                        # delivery time is calculated as loop start
                        delivery_time_s[order] = transit_time + wait_time + delivery_time
                    else:
                        # transit own order - flight to the other store - wait if needed - tansit order - delivery_time
                        flight_time = travel_time(orders[first_phase_order]['store_position'],
                                                  orders[order]['store_position'],
                                                  couriers[courier]['delivery_speed'])
                        arrival_own = (assembly_time_s[first_phase_order] + transit_time + flight_time)
                        wait_time = max(assembly_time_s[order] - arrival_own, 0)
                        delivery_time_s[order] = ((transit_time * 2) + flight_time + wait_time + delivery_time)
            
            delivery_time_s = delivery_time_s.astype(int)

            # calculate and update best result, if needed
            cte = (assembly_time_s + delivery_time_s).sum()
            if cte < best_cte:
                best_cte = cte
                best_combination = [list(first_phase), list(second_phase)]
                
    return best_cte, best_combination


best_cte, best_combination = brute_force_solution()

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