mcpField
T
task-assignment-problem

task-assignment-problem

github.com/kishorkukrejaデータ・データベース8448 回インストール3 週間前に更新
リポジトリ

task-assignment-problem について

オペレーションやサプライチェーンの資源配分を最適化するスキルです。作業者や資源をタスクに割り当て、一対一、容量制約、多目的、動的な問題を扱います。

主な機能

  1. 01線形・一般化・ボトルネック型の識別
  2. 02ハンガリー法による一対一割り当て
  3. 03スキル・容量・除外制約の組み込み
  4. 04多目的および不均衡問題への対応
  5. 05動的な割り当て方法の検討

活用シーン

  1. 01コストを抑える作業者の仕事割り当て
  2. 02倉庫ピッカーのゾーン配置
  3. 03ドライバーと配送ルートの割り当て
  4. 04タスクの到着に応じた再割り当て

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/kishorkukreja/awesome-supply-chain --skill task-assignment-problem

インストール後、Claude Code で/task-assignment-problemを入力して読み込まれたか確認します。

詳細

割り当て問題

オペレーションおよびサプライチェーンにおける割り当て最適化とリソース配分の専門家です。作業者とタスク、リソースとジョブなどを最適にマッチングする割り当て問題を解決し、コストの最小化、効率の最大化、その他の目的の最適化を支援します。1 対 1 または 1 対多のあらゆる割り当て問題に対応します。

初期評価

割り当て問題を解く前に、次の点を把握します。

  1. 問題の構造

    • 何を割り当てる必要がありますか?(作業者、機械、トラック、枠)
    • 何に割り当てますか?(タスク、注文、ルート、場所)
    • 1 対 1 の割り当てですか、それとも 1 対多ですか?
    • 割り当て数は固定ですか、それとも可変ですか?
    • 割り当て期間はどのくらいですか?(一度限り、定期的、恒久的)
  2. 目的

    • 総コストを最小化しますか?
    • 効率またはスループットを最大化しますか?
    • リソース間で作業負荷のバランスを取りますか?
    • 完了時間(メイクスパン)を最小化しますか?
    • 複数の目的がありますか?
  3. 制約

    • 容量制限(作業者が処理できるタスクは最大 N 件)がありますか?
    • 必要なスキルや資格はありますか?
    • 順序制約(ある割り当てを別の割り当てより先に行う必要がある)がありますか?
    • 除外条件(許可されない割り当て)がありますか?
    • 予算またはリソースの制限がありますか?
  4. データの有無

    • コストまたは便益の行列はありますか?
    • 作業者のスキルと能力に関する情報はありますか?
    • タスクの要件と優先度はありますか?
    • 過去の実績データはありますか?

割り当て問題のフレームワーク

問題の種類

1. 線形割り当て問題(LAP)

  • n 人の作業者、n 件のタスク
  • 各作業者を必ず 1 件のタスクに割り当てる
  • 各タスクを必ず 1 人の作業者に割り当てる
  • 目的:総コストを最小化する
  • 解法:ハンガリアン法(O(n³))

2. ボトルネック割り当て問題

  • 最大コスト(総コストではなく)を最小化する
  • 最悪の割り当てを最小化する
  • 解法:修正ハンガリアン法

3. 不均衡割り当て問題

  • m ≠ n となる m 人の作業者と n 件のタスク
  • ダミーの作業者またはタスクを追加する
  • 解法:パディングを行ったハンガリアン法

4. 一般化割り当て問題(GAP)

  • 1 人の作業者に複数のタスクを割り当て可能
  • 作業者の容量制約がある
  • NP 困難な問題
  • 解法:分枝限定法、ヒューリスティクス、近似法

5. 二次割り当て問題(QAP)

  • コストが割り当てのペアに依存する(例:施設配置)
  • NP 困難な問題
  • 解法:メタヒューリスティクス(シミュレーテッドアニーリング、遺伝的アルゴリズム)

6. 多目的割り当て

  • コストの最小化と品質の最大化を同時に行う
  • 目的間にトレードオフがある
  • 解法:加重和、パレート最適化

数学的定式化

線形割り当て問題

決定変数:

  • x[i,j] = 作業者 i がタスク j に割り当てられる場合は 1、それ以外は 0

パラメーター:

  • c[i,j] = 作業者 i をタスク j に割り当てるコスト
  • n = 作業者数 = タスク数

目的関数:

Minimize: Σ Σ c[i,j] × x[i,j]  for i,j in 1..n

制約条件:

# 1. Each worker assigned to exactly one task
for i in workers:
    Σ x[i,j] = 1  for all j

# 2. Each task assigned to exactly one worker
for j in tasks:
    Σ x[i,j] = 1  for all i

# 3. Binary assignment
for i in workers:
    for j in tasks:
        x[i,j] ∈ {0, 1}
一般化割り当て問題(GAP)

決定変数:

  • x[i,j] = タスク j が作業者 i に割り当てられる場合は 1、それ以外は 0

パラメーター:

  • c[i,j] = タスク j を作業者 i に割り当てるコスト
  • r[i,j] = タスク j を作業者 i に割り当てた場合のリソース消費量(例:時間)
  • R[i] = 作業者 i のリソース容量
  • m = 作業者数
  • n = タスク数

目的関数:

Minimize: Σ Σ c[i,j] × x[i,j]  for i in 1..m, j in 1..n

制約条件:

# 1. Each task assigned to exactly one worker
for j in tasks:
    Σ x[i,j] = 1  for all i

# 2. Worker capacity constraint
for i in workers:
    Σ (r[i,j] × x[i,j]) ≤ R[i]  for all j

# 3. Binary assignment
x[i,j] ∈ {0, 1}

割り当てアルゴリズム

ハンガリアン法(LAP に最適)
import numpy as np
from scipy.optimize import linear_sum_assignment

def hungarian_assignment(cost_matrix):
    """
    Solve assignment problem using Hungarian algorithm

    Parameters:
    -----------
    cost_matrix : 2D numpy array
        cost[i,j] = cost of assigning worker i to task j

    Returns:
    --------
    Optimal assignment
    """

    # Solve using scipy's implementation
    row_ind, col_ind = linear_sum_assignment(cost_matrix)

    # Calculate total cost
    total_cost = cost_matrix[row_ind, col_ind].sum()

    assignments = list(zip(row_ind, col_ind))

    return {
        'assignments': assignments,
        'total_cost': total_cost,
        'row_indices': row_ind,
        'col_indices': col_ind
    }


# Example: Assign 5 workers to 5 tasks
workers = ['Worker_A', 'Worker_B', 'Worker_C', 'Worker_D', 'Worker_E']
tasks = ['Task_1', 'Task_2', 'Task_3', 'Task_4', 'Task_5']

# Cost matrix (worker × task)
cost_matrix = np.array([
    [9, 2, 7, 8, 5],   # Worker_A costs
    [6, 4, 3, 7, 9],   # Worker_B costs
    [5, 8, 1, 8, 3],   # Worker_C costs
    [7, 6, 9, 4, 2],   # Worker_D costs
    [3, 7, 8, 2, 6]    # Worker_E costs
])

result = hungarian_assignment(cost_matrix)

print("Hungarian Algorithm Solution:")
print(f"Total Cost: {result['total_cost']}")
print("\nAssignments:")
for worker_idx, task_idx in result['assignments']:
    print(f"  {workers[worker_idx]} → {tasks[task_idx]} "
          f"(cost: {cost_matrix[worker_idx, task_idx]})")
貪欲割り当て
import pandas as pd

def greedy_assignment(cost_matrix, workers, tasks):
    """
    Greedy heuristic for assignment problem

    Algorithm:
    1. Find minimum cost assignment
    2. Assign it, remove worker and task from pool
    3. Repeat until all assigned

    Not optimal, but fast O(n²)

    Parameters:
    -----------
    cost_matrix : 2D numpy array
    workers : list
    tasks : list

    Returns:
    --------
    Assignment (not optimal, but fast)
    """

    assignments = []
    total_cost = 0

    available_workers = set(range(len(workers)))
    available_tasks = set(range(len(tasks)))

    while available_workers and available_tasks:
        # Find minimum cost among available assignments
        min_cost = float('inf')
        best_assignment = None

        for i in available_workers:
            for j in available_tasks:
                if cost_matrix[i, j] < min_cost:
                    min_cost = cost_matrix[i, j]
                    best_assignment = (i, j)

        # Make assignment
        worker_idx, task_idx = best_assignment
        assignments.append((worker_idx, task_idx))
        total_cost += min_cost

        # Remove from available
        available_workers.remove(worker_idx)
        available_tasks.remove(task_idx)

    return {
        'assignments': assignments,
        'total_cost': total_cost
    }


result_greedy = greedy_assignment(cost_matrix, workers, tasks)

print("\nGreedy Assignment Solution:")
print(f"Total Cost: {result_greedy['total_cost']}")
print(f"Optimal Cost: {result['total_cost']}")
print(f"Greedy vs Optimal: {(result_greedy['total_cost'] - result['total_cost']) / result['total_cost'] * 100:.1f}% worse")
一般化割り当て問題(GAP) - ヒューリスティック
from pulp import *

def solve_gap_heuristic(tasks, workers, costs, resource_usage, capacities):
    """
    Solve Generalized Assignment Problem using MIP

    Parameters:
    -----------
    tasks : list
        Task identifiers
    workers : list
        Worker identifiers
    costs : dict
        {(worker, task): cost}
    resource_usage : dict
        {(worker, task): time_hours}
    capacities : dict
        {worker: max_hours}

    Returns:
    --------
    Assignments (may be suboptimal for large problems)
    """

    prob = LpProblem("Generalized_Assignment", LpMinimize)

    # Decision variables
    x = LpVariable.dicts("assign",
                        [(w, t) for w in workers for t in tasks],
                        cat='Binary')

    # Objective: minimize total cost
    prob += lpSum([
        costs.get((w, t), 1000) * x[w, t]
        for w in workers for t in tasks
    ]), "Total_Cost"

    # Constraints

    # 1. Each task assigned to exactly one worker
    for t in tasks:
        prob += lpSum([x[w, t] for w in workers]) == 1, f"Task_{t}"

    # 2. Worker capacity
    for w in workers:
        prob += lpSum([
            resource_usage.get((w, t), 0) * x[w, t]
            for t in tasks
        ]) <= capacities.get(w, 40), f"Worker_{w}_Capacity"

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    # Extract solution
    assignments = []
    total_cost = 0

    for w in workers:
        worker_tasks = []
        for t in tasks:
            if x[w, t].varValue > 0.5:
                worker_tasks.append(t)
                total_cost += costs.get((w, t), 0)

        if worker_tasks:
            assignments.append({
                'worker': w,
                'tasks': worker_tasks,
                'num_tasks': len(worker_tasks),
                'utilization': sum(resource_usage.get((w, t), 0) for t in worker_tasks)
            })

    return {
        'status': LpStatus[prob.status],
        'assignments': pd.DataFrame(assignments),
        'total_cost': total_cost
    }


# Example: Assign 10 tasks to 3 workers
tasks = [f'Task_{i}' for i in range(1, 11)]
workers = ['Worker_A', 'Worker_B', 'Worker_C']

# Different workers have different costs and speeds for tasks
costs = {
    ('Worker_A', t): np.random.randint(10, 50) for t in tasks
}
costs.update({
    ('Worker_B', t): np.random.randint(15, 60) for t in tasks
})
costs.update({
    ('Worker_C', t): np.random.randint(12, 55) for t in tasks
})

# Time required for each worker-task combination
resource_usage = {
    ('Worker_A', t): np.random.uniform(2, 8) for t in tasks
}
resource_usage.update({
    ('Worker_B', t): np.random.uniform(3, 7) for t in tasks
})
resource_usage.update({
    ('Worker_C', t): np.random.uniform(2.5, 6) for t in tasks
})

# Worker capacities (hours available)
capacities = {
    'Worker_A': 40,
    'Worker_B': 30,  # Part-time
    'Worker_C': 40
}

result_gap = solve_gap_heuristic(tasks, workers, costs, resource_usage, capacities)

print(f"\nGAP Solution Status: {result_gap['status']}")
print(f"Total Cost: ${result_gap['total_cost']:.2f}")
print("\nAssignments:")
print(result_gap['assignments'])

高度な割り当て手法

スキルベースの割り当て
def skill_based_assignment(tasks, workers, costs, skill_requirements, worker_skills):
    """
    Assignment with skill constraints

    Workers can only be assigned to tasks matching their skills

    Parameters:
    -----------
    tasks : list
    workers : list
    costs : dict
        {(worker, task): cost}
    skill_requirements : dict
        {task: [required_skills]}
    worker_skills : dict
        {worker: [skills]}

    Returns:
    --------
    Skill-constrained assignments
    """

    prob = LpProblem("Skill_Based_Assignment", LpMinimize)

    # Filter feasible assignments (worker has required skills)
    feasible_assignments = []
    for w in workers:
        for t in tasks:
            required = set(skill_requirements.get(t, []))
            available = set(worker_skills.get(w, []))

            if required.issubset(available):
                feasible_assignments.append((w, t))

    # Decision variables (only for feasible assignments)
    x = LpVariable.dicts("assign",
                        feasible_assignments,
                        cat='Binary')

    # Objective
    prob += lpSum([
        costs.get((w, t), 1000) * x[w, t]
        for (w, t) in feasible_assignments
    ]), "Total_Cost"

    # Constraints

    # 1. Each task assigned to exactly one worker (if possible)
    for t in tasks:
        task_assignments = [(w, t) for (w, t) in feasible_assignments if t == t]
        if task_assignments:
            prob += lpSum([x[w, t] for (w, t) in task_assignments]) == 1, f"Task_{t}"

    # 2. Each worker assigned to at most one task (one-to-one assignment)
    for w in workers:
        worker_assignments = [(w, t) for (w, t) in feasible_assignments if w == w]
        if worker_assignments:
            prob += lpSum([x[w, t] for (w, t) in worker_assignments]) <= 1, f"Worker_{w}"

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    # Extract solution
    assignments = []
    for (w, t) in feasible_assignments:
        if x[w, t].varValue > 0.5:
            assignments.append({
                'worker': w,
                'task': t,
                'cost': costs.get((w, t), 0),
                'skills_used': skill_requirements.get(t, [])
            })

    return {
        'status': LpStatus[prob.status],
        'assignments': pd.DataFrame(assignments),
        'total_cost': value(prob.objective) if prob.status == 1 else None
    }


# Example
tasks = ['Task_A', 'Task_B', 'Task_C', 'Task_D']
workers = ['Worker_1', 'Worker_2', 'Worker_3', 'Worker_4']

skill_requirements = {
    'Task_A': ['forklift', 'inventory_mgmt'],
    'Task_B': ['packing', 'quality_control'],
    'Task_C': ['forklift'],
    'Task_D': ['receiving', 'inventory_mgmt']
}

worker_skills = {
    'Worker_1': ['forklift', 'inventory_mgmt', 'receiving'],
    'Worker_2': ['packing', 'quality_control'],
    'Worker_3': ['forklift', 'packing'],
    'Worker_4': ['inventory_mgmt', 'receiving', 'quality_control']
}

costs = {
    ('Worker_1', 'Task_A'): 10,
    ('Worker_1', 'Task_C'): 8,
    ('Worker_1', 'Task_D'): 12,
    ('Worker_2', 'Task_B'): 9,
    ('Worker_3', 'Task_C'): 7,
    ('Worker_4', 'Task_D'): 11,
    ('Worker_4', 'Task_B'): 10,
}

result_skills = skill_based_assignment(tasks, workers, costs,
                                       skill_requirements, worker_skills)

print("\nSkill-Based Assignment:")
print(result_skills['assignments'])
動的タスク割り当て
class DynamicAssignmentManager:
    """
    Manage dynamic task assignments as new tasks arrive

    Use for real-time environments (warehouse, call center, delivery)
    """

    def __init__(self, workers, initial_workloads=None):
        self.workers = workers
        self.workloads = initial_workloads if initial_workloads else {w: 0 for w in workers}
        self.assignments = []
        self.task_queue = []

    def add_task(self, task_id, priority, estimated_time):
        """Add new task to queue"""
        self.task_queue.append({
            'task_id': task_id,
            'priority': priority,
            'estimated_time': estimated_time,
            'arrival_time': datetime.now()
        })

    def assign_next_task(self, worker_capabilities=None):
        """
        Assign next task to best available worker

        Strategy:
        1. Prioritize high-priority tasks
        2. Assign to worker with lowest current workload
        3. Consider worker capabilities if provided
        """

        if not self.task_queue:
            return None

        # Sort queue by priority (highest first)
        self.task_queue.sort(key=lambda t: t['priority'], reverse=True)

        # Get next task
        next_task = self.task_queue[0]

        # Find best worker (lowest workload)
        if worker_capabilities:
            # Filter workers who can do this task
            capable_workers = [
                w for w in self.workers
                if worker_capabilities.get(w, {}).get(next_task['task_id'], False)
            ]
        else:
            capable_workers = self.workers

        if not capable_workers:
            print(f"No capable workers for task {next_task['task_id']}")
            return None

        # Assign to worker with lowest workload
        best_worker = min(capable_workers, key=lambda w: self.workloads[w])

        # Make assignment
        assignment = {
            'worker': best_worker,
            'task': next_task['task_id'],
            'assigned_time': datetime.now(),
            'estimated_completion': datetime.now() + timedelta(minutes=next_task['estimated_time'])
        }

        self.assignments.append(assignment)
        self.workloads[best_worker] += next_task['estimated_time']
        self.task_queue.pop(0)

        return assignment

    def complete_task(self, worker, actual_time):
        """Update workload when task completed"""
        self.workloads[worker] = max(0, self.workloads[worker] - actual_time)

    def rebalance_workload(self):
        """
        Rebalance tasks if workload becomes very uneven

        Reassign tasks from overloaded to underutilized workers
        """

        avg_workload = sum(self.workloads.values()) / len(self.workers)
        max_workload = max(self.workloads.values())
        min_workload = min(self.workloads.values())

        imbalance = (max_workload - min_workload) / avg_workload

        if imbalance > 0.5:  # 50% imbalance threshold
            print(f"Workload imbalance detected: {imbalance:.1%}")
            # Rebalancing logic would go here
            # Move tasks from overloaded to underutilized

        return imbalance


# Example usage
workers = ['Picker_1', 'Picker_2', 'Picker_3']
manager = DynamicAssignmentManager(workers)

# Simulate arriving tasks
for i in range(10):
    manager.add_task(
        task_id=f'Order_{i}',
        priority=np.random.choice([1, 2, 3]),
        estimated_time=np.random.randint(10, 30)
    )

# Assign tasks
print("Dynamic Task Assignment:")
for _ in range(5):
    assignment = manager.assign_next_task()
    if assignment:
        print(f"  {assignment['worker']} ← {assignment['task']}")

print(f"\nCurrent Workloads: {manager.workloads}")
print(f"Remaining Tasks: {len(manager.task_queue)}")

割り当ての実践的な適用例

倉庫:ピッカーからゾーンへの割り当て
def assign_pickers_to_zones(pickers, zones, pick_volumes, picker_productivity):
    """
    Assign pickers to warehouse zones to balance workload

    Parameters:
    -----------
    pickers : list
        Available pickers
    zones : list
        Warehouse zones
    pick_volumes : dict
        {zone: lines_to_pick}
    picker_productivity : dict
        {picker: lines_per_hour}

    Returns:
    --------
    Optimal zone assignments
    """

    # Calculate time required per zone
    zone_hours = {
        zone: pick_volumes[zone] / 100  # Assume 100 lines/hour avg
        for zone in zones
    }

    # For each picker, calculate time if assigned to each zone
    costs = {}
    for picker in pickers:
        productivity = picker_productivity.get(picker, 100)
        for zone in zones:
            # Time = volume / productivity
            # Cost = time (want to minimize total time)
            costs[(picker, zone)] = pick_volumes[zone] / productivity

    # Solve assignment (GAP since multiple zones can go to one picker)
    # Simplified: use one-to-one for this example

    # If more zones than pickers, use GAP
    # If equal or fewer zones, use Hungarian

    if len(zones) <= len(pickers):
        # Pad to make square matrix
        n = max(len(pickers), len(zones))

        cost_matrix = np.full((n, n), 1000)  # High cost for dummy assignments

        for i, picker in enumerate(pickers):
            for j, zone in enumerate(zones):
                cost_matrix[i, j] = costs[(picker, zone)]

        result = hungarian_assignment(cost_matrix)

        assignments = []
        for picker_idx, zone_idx in result['assignments']:
            if picker_idx < len(pickers) and zone_idx < len(zones):
                picker = pickers[picker_idx]
                zone = zones[zone_idx]
                assignments.append({
                    'picker': picker,
                    'zone': zone,
                    'volume': pick_volumes[zone],
                    'estimated_time': costs[(picker, zone)]
                })

        return pd.DataFrame(assignments)

    else:
        # More zones than pickers - use GAP
        print("More zones than pickers - using GAP")
        # Implementation would be similar to solve_gap_heuristic


# Example
pickers = ['Picker_A', 'Picker_B', 'Picker_C', 'Picker_D']
zones = ['Zone_1', 'Zone_2', 'Zone_3', 'Zone_4']

pick_volumes = {
    'Zone_1': 450,
    'Zone_2': 380,
    'Zone_3': 520,
    'Zone_4': 290
}

picker_productivity = {
    'Picker_A': 120,  # Fast
    'Picker_B': 95,   # Average
    'Picker_C': 110,  # Above average
    'Picker_D': 85    # Slower
}

zone_assignments = assign_pickers_to_zones(pickers, zones, pick_volumes, picker_productivity)

print("\nPicker-to-Zone Assignments:")
print(zone_assignments)
print(f"\nMax Time: {zone_assignments['estimated_time'].max():.1f} hours (bottleneck)")
輸送:ドライバーからルートへの割り当て
def assign_drivers_to_routes(drivers, routes, costs, constraints=None):
    """
    Assign drivers to delivery routes

    Consider:
    - Driver preferences
    - Route difficulty
    - Driver experience
    - Route constraints (hazmat, special equipment)

    Parameters:
    -----------
    drivers : list
    routes : list
    costs : dict
        {(driver, route): cost/preference_score}
    constraints : dict
        Special requirements

    Returns:
    --------
    Driver-route assignments
    """

    # Filter feasible assignments based on constraints
    feasible = []

    for driver in drivers:
        for route in routes:
            # Check constraints
            is_feasible = True

            if constraints:
                # Example: hazmat certification required
                if constraints.get(route, {}).get('hazmat_required', False):
                    if not constraints.get(driver, {}).get('hazmat_certified', False):
                        is_feasible = False

                # Example: commercial license required for large trucks
                if constraints.get(route, {}).get('cdl_required', False):
                    if not constraints.get(driver, {}).get('has_cdl', False):
                        is_feasible = False

            if is_feasible:
                feasible.append((driver, route))

    # Build cost matrix for feasible assignments
    n_drivers = len(drivers)
    n_routes = len(routes)
    n = max(n_drivers, n_routes)

    cost_matrix = np.full((n, n), 10000)  # High cost for infeasible

    for (driver, route) in feasible:
        i = drivers.index(driver)
        j = routes.index(route)
        cost_matrix[i, j] = costs.get((driver, route), 100)

    # Solve
    result = hungarian_assignment(cost_matrix)

    # Extract valid assignments
    assignments = []
    for driver_idx, route_idx in result['assignments']:
        if driver_idx < n_drivers and route_idx < n_routes:
            driver = drivers[driver_idx]
            route = routes[route_idx]

            if (driver, route) in feasible:
                assignments.append({
                    'driver': driver,
                    'route': route,
                    'cost': costs.get((driver, route), 0)
                })

    return pd.DataFrame(assignments)


# Example
drivers = ['Driver_1', 'Driver_2', 'Driver_3']
routes = ['Route_A', 'Route_B', 'Route_C']

# Cost based on driver preference and route difficulty
costs = {
    ('Driver_1', 'Route_A'): 10,
    ('Driver_1', 'Route_B'): 15,
    ('Driver_1', 'Route_C'): 12,
    ('Driver_2', 'Route_A'): 8,
    ('Driver_2', 'Route_B'): 20,
    ('Driver_2', 'Route_C'): 11,
    ('Driver_3', 'Route_A'): 14,
    ('Driver_3', 'Route_B'): 9,
    ('Driver_3', 'Route_C'): 16,
}

route_assignments = assign_drivers_to_routes(drivers, routes, costs)

print("\nDriver-to-Route Assignments:")
print(route_assignments)

ツールとライブラリ

割り当てソフトウェア

最適化ソルバー:

  • Gurobi:割り当てモデルに対応する商用 MIP ソルバー
  • CPLEX (IBM):エンタープライズ向け最適化ソルバー
  • Google OR-Tools:オープンソースの制約プログラミングおよびルーティングツール
  • COIN-OR CBC:オープンソースの MIP ソルバー

割り当てに特化したツール:

  • OptaPlanner (Red Hat):スケジューリング/割り当て向け AI 制約ソルバー
  • MiniZinc:制約モデリング言語
  • Satalia:AI を活用した最適化プラットフォーム
Python ライブラリ
# Assignment Algorithms
from scipy.optimize import linear_sum_assignment  # Hungarian algorithm
from pulp import *  # MIP modeling
from ortools.sat.python import cp_model  # Constraint programming
from ortools.linear_solver import pywraplp  # Linear programming

# Optimization
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans  # For grouping before assignment

# Graph Algorithms
import networkx as nx  # For bipartite matching

よくある課題と解決策

課題:不均衡な問題

問題:

  • 作業者の数がタスクより多い(またはその逆)
  • 標準的なハンガリー法を直接使用できない

解決策:

  • コストがゼロのダミータスク/作業者を追加する
  • 拡張した問題を解く
  • 解からダミーの割り当てを除外する
  • 代替案:GAP 定式化を使用する(未割り当てを許可)
課題:複数の目的

問題:

  • コストを最小化しつつ、作業負荷も均衡させたい
  • 品質を最大化しつつ、時間を最小化したい
  • 目的間にトレードオフがある

解決策:

  • 加重和アプローチ(α×コスト + β×作業負荷)
  • 辞書式最適化(主目的を最適化してから、副次目的を最適化)
  • パレートフロンティア分析(トレードオフ曲線を示す)
  • 副次目的に制約を設ける(コスト < X としたうえで均衡を図る)
課題:大規模な問題

問題:

  • 作業者が1000人以上、タスクが1000件以上
  • ハンガリー法の O(n³) の処理が遅くなる
  • MIP ソルバーの実行時間が許容できない

解決策:

  • 分解(部分問題を解く)
  • ヒューリスティクス(貪欲法、局所探索)
  • オークションアルゴリズム(大規模で疎な問題に対してより高速)
  • 並列コンピューティング
  • 制限時間を設け、見つかった最良の解を使用する
課題:動的な到着

問題:

  • タスクが時間の経過とともに到着する(最初からすべてがわかっているわけではない)
  • オンライン割り当て問題である
  • まとめて最適化するために待つことができない

解決策:

  • ローリングホライズン最適化(定期的に再計算する)
  • 貪欲なオンライン割り当て(即座に割り当てる)
  • 今後の到着に備えて容量を確保する
  • 競争比分析(オフライン最適解と比較する)
課題:ソフト制約と希望条件

問題:

  • 作業者の希望(ハード制約ではない)
  • 必須ではないが望ましいスキルの一致
  • 必須ではないが希望する割り当て

解決策:

  • 目的関数のペナルティコストとしてモデル化する
  • 二段階で行う:ハード制約を満たしてから希望条件に対応する
  • 希望条件を副次目的とする多目的最適化
  • 公平性制約(全員が希望する割り当てをいくつか得られるようにする)

出力形式

割り当てレポート

タスク割り当て結果

問題の概要:

  • 作業者:15
  • タスク:15
  • 割り当てタイプ:一対一(線形割り当て問題)
  • 目的:総コストの最小化
  • 解法:ハンガリー法

最適な割り当て:

作業者タスクコストスキル一致率推定時間
Worker_ATask_07$25100%2.5 hrs
Worker_BTask_03$18100%1.8 hrs
Worker_CTask_12$3280%3.2 hrs
Worker_DTask_01$15100%1.2 hrs
...............

パフォーマンス指標:

指標値
総コスト$345
割り当てあたりの平均コスト$23
作業負荷の均衡(標準偏差)0.8 hrs
スキル一致率92%
推定完了時間3.2 hrs(ボトルネック)

作業負荷の分布:

Worker_A: 2.5 hrs
Worker_B: 1.8 hrs
Worker_C: 3.2 hrs (bottleneck)
Worker_D: 1.2 hrs
Worker_E: 2.9 hrs
...

Most Balanced: Worker_B, Worker_D (light load - can help others)
Bottleneck: Worker_C (longest task)

推奨事項:

  1. Worker_C の負荷を均衡させるため、Task_12 の分割を検討してください
  2. Worker_B と Worker_D には追加のタスクを担当できる余力があります
  3. スキル一致率は92%です。柔軟性を高めるため、トレーニングを検討してください

確認すべき質問

より詳しい情報が必要な場合:

  1. 何を割り当てる必要がありますか(作業者、機械、リソース)?
  2. 何に割り当てますか(タスク、仕事、注文)?
  3. 一対一ですか、それとも複数のタスクを一人の作業者に割り当てられますか?
  4. 目的は何ですか(コストの最小化、効率の最大化)?
  5. 容量制約やスキル要件はありますか?
  6. 割り当てはいくつありますか(10件、100件、1000件)?
  7. 一度限りの割り当てですか、それとも定期的に行いますか?
  8. コストや希望条件のデータはありますか?

関連スキル

  • workforce-scheduling:シフトや労働力のスケジューリング向け
  • vehicle-routing-problem:車両へのルート割り当て向け
  • optimization-modeling:数理モデルの定式化向け
  • constraint-programming:複雑な制約の処理向け
  • linear-programming:LP ベースの割り当て向け
  • graph-algorithms:二部マッチング向け
  • metaheuristic-optimization:大規模なヒューリスティック解向け

コメント