Python COS類似度計算の高速化

Question

数列同士のCOS類似度を計算する際の高速化について、質問させていただきます。

質問内容

1. 行列Aに対して、rand2で定義したデータとのコサイン類似度を求めた、配列cos_listを得たい場合、

　現在は、cos_simを内包表記で呼び出すことで、計算していますが、何らかの方法でここを高速化できないでしょうか？

import numpy as np
import time
def cos_sim(v1, v2): 
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

print("データ生成中...")
rand1 = [ np.random.rand() for i in range(5000)] #5000個の乱数データを作成する。 #[0.43839043,0.8423294420,0.2342…]
rand2 =  [ np.random.rand() for i in range(5000)] #類似度を検索する対象となる乱数データ  #[0.139043,0.9442446246420,0.231342…]
A = np.array([rand1 for j in range(10000)],dtype=np.float32)  #[[0.43839043,0.8423294420,0.2342…],[0.43839043,0.8423294420,0.2342…],[0.43839043,0.8423294420,0.2342…], …]
print("データ準備完了")
start = time.time() #時間計測用変数

###－－－－－－ーーーここを高速化したい。－－－－－－ーーー
cos_list =[ cos_sim(rand2,a) for a in A] #コサイン類似度リスト　#[0.73141,0.73141,0.73141, ...]
###－－－－－－ーーー－－－－－－ーーー－－－－－－ーーーー
#※あくまで例なのでデータは、Aの各データをrand1で固定しているので、10000個の同じコサイン類似度リストが求まります。

print("完了時間:{0}".format(time.time() - start) + "[sec]") #>> 完了時間:2.42

以上、ご教授のほどよろしくお願い致します。

実行環境：Python3.X
CPU:仮想v6コア(AWS)

考えたこと

何となく、早やくなりそうだけど、やり方がわからないのですが、
イメージ的にこんな事がやりたい気がしています。
（一回1:1の関係でリスト化して、numpyで一気に全体に対してCOS類似度を求める？）

#　target_list = [ cos_sim(rand2,a) for a in A] 
target_list = [[rand2,a] for a in A] 
#とりあえず、rand2の列とaの列を1:1の関係でリストに定義して、ここからnumpyの機能でCOS類似度を一気に求められないでしょうか・・・

Answer 1

こんな感じでどうでしょうか。

import numpy as np
import time
def cos_sim(v1, v2): 
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

print("データ生成中...")
rand1 = [ np.random.rand() for i in range(500)]
rand2 =  [ np.random.rand() for i in range(500)]
A = np.array([rand1 for j in range(1000)], dtype=np.float32)
print("データ準備完了")

# original
start = time.time() 
cos_list =[cos_sim(rand2,a) for a in A] 
print("完了時間:{0}".format(time.time() - start) + "[sec]")

# rand1とrand2をnumpy配列に
rand1 = np.array(rand1)
rand2 = np.array(rand2)
start = time.time() 
cos_list2 =[cos_sim(rand2,a) for a in A] 
assert np.allclose(cos_list, cos_list2), "error1"
print("完了時間:{0}".format(time.time() - start) + "[sec]")

# 先にスケーリングすることにする
def cos_sim2(v1, v2):
    return np.dot(v1, v2)

A = np.array([rand1 for j in range(1000)], dtype=np.float32)

start = time.time()
rand2 /= np.linalg.norm(rand2)
A /= np.linalg.norm(A, axis=1).reshape(-1, 1) 
cos_list3 =[cos_sim2(rand2,a) for a in A] 
assert np.allclose(cos_list, cos_list3), "error2"
print("完了時間:{0}".format(time.time() - start) + "[sec]")

""" =>
データ生成中...
データ準備完了
完了時間:0.059908390045166016[sec]
完了時間:0.021083831787109375[sec]
完了時間:0.003892183303833008[sec]
"""

先にスケーリングすると1桁速くなりますね（たぶんスケーリング時間込みでも）。

Comments

[KK-31]

大変参考になりました。
ありがとうございます。

Answer 2

import timeit
import numpy as np

# 入力データ
A = np.random.rand(10000, 5000)
b = np.random.rand(5000)

# 質問欄のやり方
def cos_sim(v1, v2): 
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

# Vectorized Computation
def cos_sim2(v1, v2):
    return np.inner(A, b) / (np.linalg.norm(v1, axis=1) * np.linalg.norm(v2))

# 質問欄の書き方
loop = 10
secs = timeit.timeit('[cos_sim(b, a) for a in A]', globals=globals(), number=loop)
print(secs / loop)

# Vectorized Computation
secs = timeit.timeit('cos_sim2(A, b)', globals=globals(), number=loop)
print(secs / loop)

0.1224646340124309  # 質問欄のやりかた
0.09451284902170301  # 一度に計算するやり方

numpy を使えば、基本的に for ループを使う必要性はないはずです。

一度の呼び出しで計算するやり方を記載しました。(配列の各要素ではなく、配列全体に一気に演算を適用する計算方法を vectrized computation といいます。)

約1.3倍ほど早くなりました。これ以上の劇的な高速化は難しいかもしれません。

Comments

[KK-31]

大変参考になりました。こんな書き方があるんですね・・・。
ありがとうございます。