Fixstars Amplifyによる割り当て問題の最適化

Fixstars Amplifyって？

量子コンピュータ・もしくはFPGAによる疑似的な量子コンピュータを、実行するためのクラウド基盤です。 量子コンピュータに関する専門性の高い部分は自動化されており、簡単に量子コンピュータを用いた 実験、サービスの開発を開始することができます。 実際Google Colaboratoyで

$ pip install amplify

を実行するだけでAmplifyの実行環境は整います。

最適化したい題材

Fusicには一週間に一度1on1on1なる1対1ではなく、1対1対1の3人で行う面談があります。この面談の割り当てを探索していきますが、 この面談の割り当てには以下のような要望がありました。

• 面談の人数は3~4人(人数の関係で3人で組めない場合がある)
• ひとつ前の面談のメンバーとは絶対被らせたくない
• ひとつ前の面談を除いた過去の面談のメンバーとはなるべく被らせたくない
• なるべくかかわりのない人(違う部署・チーム)の人と面談を組ませたい

割り当て最適化

データの準備

今回使用するのはFusicメンバーの所属データと、過去の1on1on1の組みわけのデータ3回分を使用します。

メンバーと所属データ(member_df)

	name	team	team2	team3
0	name0	採用広報	NaN	NaN
1	name1	sigfy	360	NaN
2	name2	Kind	sigfy	NaN
3	name3	機械学習	NaN	NaN
4	name4	LIGHT	NaN	NaN
.
.
.

ライブラリのインポート

from amplify import BinaryPoly, gen_symbols, sum_poly, Solver, decode_solution
from amplify.constraint import equal_to, less_equal, greater_equal, clamp, penalty
from amplify.client import FixstarsClient

symbolの定義

n_members = len(member_df)
n_groups = (n_members) // 3
n_members = n_members

q = gen_symbols(BinaryPoly, n_members, n_groups)

変数'q'には量子コンピュータが扱えるbitの配列を定義をしています。bitは必用に応じて、定数として値を1か0で固定することができます。 今回のタスクでは固定化する必要が無いので値の固定はしていません。

制約の定義

def get_member_constraints():
    '''
    メンバーは1つの面談にしか出席できないようにする制約
    '''
    member_constraints = []
    for i in range(n_members):
        member_constraints.append(
            equal_to(sum_poly([q[i][j] for j in range(n_groups)]), 1)
        )
    return member_constraints

def get_number_of_people_constraints():
    '''
    面談の人数を3~4人に制限する制約
    '''
    number_of_people_constraints = []
    for j in range(n_groups):
        number_of_people_constraints.append(
            clamp(sum_poly([q[i][j] for i in range(n_members)]), 3, 4)
        )
    return number_of_people_constraints

def get_duplicate_constraints(last_time_groups):
    '''
    一つ前の面談とメンバーが被らないようにする制約
    '''
    duplicate_constraints = []
    for j in range(n_groups):
        for last_time_group in last_time_groups:
            duplicate_constraints.append(
                less_equal(sum_poly([q[i][j] for i in last_time_group]), 1)
            )
    return duplicate_constraints

制約は必ず満たさなければならない条件であり、量子コンピュータはこの制約を満たす解を探索します。 今回の割り当て問題では'面談の人数は3~4人', 'ひとつ前の面談のメンバーとは絶対被らせたくない'に該当します。 余談ですが、量子コンピュータはこの制約を満たす解を出しますが、制約をそもそも満たすことのできない場合は実行はできますが 解が帰ってきません。

コスト

def get_team_cost(teams):
    '''
    なるべく同じチームにならないようにするコスト
    '''
    team_cost = sum(
        [sum(q[i][j] for i in team) for team in teams for j in range(n_groups)]
    )
    return team_cost

def get_duplicate_cost(past_groups):
    '''
    過去の面談となるべくメンバーが被らないようにするコスト
    '''
    duplicate_list = flatten_list(past_groups)

    past_groups = sum(past_groups, [])
    duplicate_cost = []
    for member in duplicate_list:
        for j in range(n_groups):
            duplicate = list(set(sum([groups for groups in past_groups if member in groups], [])))
            duplicate_cost.append(
                sum_poly([q[i][j] for i in duplicate])
            )
    return sum(duplicate_cost)


def flatten_list(l):
    for el in l:
        if isinstance(el, list):
            yield from flatten_list(el)
        else:
            yield el

コストは制約を満たしつつ最適解を探すうえで最小化したい値のことを指します。 今回の割り当て問題では'ひとつ前の面談を除いた過去の面談のメンバーとはなるべく被らせたくない', 'なるべくかかわりのない人(違う部署・チーム)の人と面談を組ませたい' に該当します。過去の面談のメンバーが一人被るごとにコストが増加する、同じ部署・チームの人と同じ面談になるとコストが増加する、のようにコストを設定します。 量子コンピュータは、制約を満たしつつコストが最小になるような解を返してきます。

modelの定義と実行

team_cost_k = 0.5
duplicate_cost_k = 1.0

member_constraints = get_member_constraints()
number_of_people_constraints = get_number_of_people_constraints()
model = sum(member_constraints) + sum(number_of_people_constraints)

duplicate_constraints = get_duplicate_constraints(last_time_groups[-1])
model += sum(duplicate_constraints)

past_groups = [last_time_groups[:-1]]

duplicate_cost = get_duplicate_cost(past_groups)
team_cost = get_team_cost(teams)
model = model + (duplicate_cost * duplicate_cost_k + team_cost * team_cost_k)

client = FixstarsClient()
client.token = 'Fixstars_Amplify_token'
client.parameters.timeout = 5_000
solver = Solver(client)

result = self.solver.solve(model)
energy, values = result[0].energy, result[0].values
q_values = decode_solution(q, values)

変数のself.team_cost_kやself.duplicate_cost_kの変数はコストに対するバイアスで、 これを調整する事でコストの影響の大きさを調整することができます。今回は、チームのコストに0.5を設定して、過去の面談となるべくメンバーが被らないようにすることを優先しています。

テスト

制約が満たされていることを確認します。まずは、面談の人数を3~4人に制限する制約です。 これは上記の結果を確認すると、満たされていることがわかります。 一つ前の面談とメンバーが被らないようにする制約については人の手で数えるのは意外とめんどくさいのでテスト用のスクリプトを書きました。

def check_duplicate(group_list_1: list, group_list_2: list):
    score = 0
    for group in group_list_1:
        for number in group:
            for target_group in group_list_2:
                if number in target_group:
                    score += get_score(group, target_group)
                else:
                    continue
    return score

def get_score(group, target_group):
    score = -1
    for number in group:
        for target_number in target_group:
            if number == target_number:
                score += 1 
    print(f'duplicate score: {score}')

これに上の結果とひとつ前の面談のリストを入力します。
そしてテスト実行

duplicate score: 0

一つ前の面談とメンバーが被らないようにする制約は満たされているのが確認できました。 次はコストがどの程度小さくなっているかを確認します。 コストの値については、実行したときに結果とともに返してくるenergy変数に数値として返してくれます。 この実行では34.0でした、これは、チームコストと過去の面談のコストの合計を返してくれているようです。

こんな感じで1on1on1の自動割り当てをさせてみましたが、意外と使い勝手がよく応用もきくので、活躍できる 場面は多いのではないかと感じます。