CUBE SUGAR CONTAINER

技術系のこと書きます。

Python: Annoy の近似最近傍探索 (ANN) を試す

今回は Spotify の作った近似最近傍探索 (ANN: Approximate Nearest Neighbor algorithms search) ライブラリの Annoy を試してみる。 ANN は k-NN (k-Nearest Neighbor algorithms search) の一種で、厳密な解を追い求めない代わりに高いスループットが得られる。

ちなみに ANN のライブラリごとのベンチマークを公開している Web サイトがある。 その中でいうと Annoy は大体のベンチマークで真ん中くらいの位置にある。 その代わり Annoy はインストールが簡単という利点がある。

ANN-Benchmarks

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.14
BuildVersion:   18A391
$ python -V                     
Python 3.7.1

下準備

まずは Annoy 本体と、データセットの読み込みなどに使うために scikit-learn をインストールしておく。

$ pip install annoy scikit-learn

上記のように pip から一発でインストールできるのは Annoy の良いところだと思う。

Annoy を試してみる

早速だけど Annoy のサンプルコードを以下に示す。 データセットには Iris データセットを使って、分類精度を 3-fold CV を使って計測した。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from sklearn import datasets
from annoy import AnnoyIndex
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import accuracy_score


def main():
    # Iris データセットを読み込む
    dataset = datasets.load_iris()
    X, y = dataset.data, dataset.target

    # 3-fold CV
    skf = StratifiedKFold(n_splits=3, shuffle=True, random_state=42)
    for train_index, test_index in skf.split(X, y):
        # データを学習用と検証用に分ける
        X_train, X_test = X[train_index], X[test_index]
        y_train, y_test = y[train_index], y[test_index]

        # ユークリッド距離で探索するモデルを用意する
        annoy_index = AnnoyIndex(X.shape[1], metric='euclidean')

        # モデルを学習させる
        for i, x in enumerate(X_train):
            annoy_index.add_item(i, x)

        # k-d tree をビルドする
        annoy_index.build(n_trees=10)

        # 近傍 1 点を探索して、そのクラスを返す
        y_pred = [y_train[annoy_index.get_nns_by_vector(v, 1)[0]] for v in X_test]

        # 精度 (Accuracy) を計算する
        score = accuracy_score(y_test, y_pred)
        print('acc:', score)


if __name__ == '__main__':
    main()

上記の実行結果は次の通り。 だいたい 96% 前後の精度が出ている。

$ python helloannoy.py 
acc: 0.9607843137254902
acc: 0.9607843137254902
acc: 0.9583333333333334

もう少し詳しく見ていく

ここからはもう少し詳しく Annoy の API を眺めてみる。 まずは Python のインタプリタを起動する。

$ python

とりあえず動作確認用に Iris データセットを読み込んで、それをホールドアウト検証用に分割しておく。

>>> from sklearn import datasets
>>> dataset = datasets.load_iris()
>>> X, y = dataset.data, dataset.target
>>> from sklearn.model_selection import train_test_split
>>> X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)

続いて肝心の Annoy のモデルを用意する。 これには AnnoyIndex というクラスを用いる。 インスタンス化するときは、データの次元数と、距離の計算に用いるメトリックを指定する。 先ほどはユークリッド距離 (euclidean) を使っていたけど、ここではコサイン距離 (angular) を使っている。 これは正規化したユークリッド距離を表しているらしい。

>>> from annoy import AnnoyIndex
>>> annoy_index = AnnoyIndex(X.shape[1], metric='angular')

続いてモデルに学習データを登録していく。 ここで指定したインデックスが後ほど探索結果として得られる。

>>> for i, x in enumerate(X_train):
...     annoy_index.add_item(i, x)
... 

Annoy は k-d tree を元に探索を高速化している。 そこで続いては学習データを元に木をビルドする。

>>> annoy_index.build(n_trees=10)
True

これで分類の準備が整った。 あとは AnnoyIndex#get_nns_by_vector() メソッドで探索結果が得られる。 これには、近傍を見つけたいデータと、見つけたい近傍数を与える。 例えば近傍数が 1 ならこんな感じ。

>>> annoy_index.get_nns_by_vector(X_test[0], 1)
[85]

近傍数が 3 なら次のようになる。。

>>> annoy_index.get_nns_by_vector(X_test[0], 3)
[85, 71, 26]

近傍数と共に距離も手に入れたいときは include_distances オプションに True を渡す。 すると結果がタプルになって近傍のインデックスと共に距離が得られる。

>>> annoy_index.get_nns_by_vector(X_test[0], 3, include_distances=True)
([85, 71, 26], [0.021935366094112396, 0.025716988369822502, 0.034560393542051315])

Annoy が返すのは近傍のインデックスなので、ラベルに変換するには次のようにする。 分類させたデータのラベルを見ると 1 で、近傍 3 点のラベルも 1 になっているので、これはちゃんと分類できている。

>>> y_train[85], y_train[71], y_train[26]
(1, 1, 1)
>>> y_test[0]
1

scikit-learn の API に対応させてみる

続いて、試しに Annoy の実装を使って scikit-learn の分類器を書いてみた。 次のサンプルコードの AnnoyClassifier がそれに当たる。 scikit-learn の API に対応したことで sklearn.model_selection.cross_validate() がそのまま使えるようになっている。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from collections import Counter

from sklearn import datasets
from annoy import AnnoyIndex
from sklearn.base import BaseEstimator
from sklearn.base import ClassifierMixin
from sklearn.model_selection import StratifiedKFold
from sklearn.model_selection import cross_validate
from sklearn.utils import check_X_y
from sklearn.utils import check_array


class AnnoyClassifier(BaseEstimator, ClassifierMixin):
    """scikit-learn API に準拠した Annoy の分類器"""

    def __init__(self, n_trees, metric='angular', n_neighbors=1, search_k=-1):
        # k-d tree の数
        self.n_trees_ = n_trees
        # 計算に用いる距離
        self.metric_ = metric
        # 近傍数
        self.n_neighbors_ = n_neighbors
        # 精度とスループットのトレードオフに用いるパラメータ
        self.search_k_ = search_k
        # ビルドしたモデル
        self.clf_ = None
        # 学習データのクラスラベルを入れる
        self.train_y_ = None

    def fit(self, X, y):
        # 入力をバリデーションする
        check_X_y(X, y)

        # 学習データのクラスラベルを保存しておく
        self.train_y_ = y

        # Annoy のモデルを用意する
        self.clf_ = AnnoyIndex(X.shape[1], metric=self.metric_)

        # モデルを学習させる
        for i, x in enumerate(X):
            self.clf_.add_item(i, x)

        # k-d tree を構築する
        self.clf_.build(n_trees=self.n_trees_)

        return self

    def predict(self, X):
        # 入力をバリデーションする
        check_array(X)

        # 分類結果を返す
        y_pred = [self._predict(x) for x in X]
        return y_pred

    def _predict(self, x):
        # 近傍を見つける
        neighbors = self.clf_.get_nns_by_vector(x, self.n_neighbors_, search_k=self.search_k_)
        # インデックスをクラスラベルに変換する
        neighbor_classes = self.train_y_[neighbors]
        # 最頻値を取り出す
        counter = Counter(neighbor_classes)
        most_common = counter.most_common(1)
        # 最頻値のクラスラベルを返す
        return most_common[0][0]

    def get_params(self, deep=True):
        # 分類器のパラメータ
        return {
            'n_trees': self.n_trees_,
            'metric': self.metric_,
            'n_neighbors': self.n_neighbors_,
            'search_k': self.search_k_,
        }


def main():
    # Iris データセットを読み込む
    dataset = datasets.load_iris()
    X, y = dataset.data, dataset.target

    # Annoy を scikit-learn API でラップした分類器
    clf = AnnoyClassifier(n_trees=10)
    # 3-fold CV
    skf = StratifiedKFold(n_splits=3, shuffle=True, random_state=42)
    # 精度を評価指標にして汎化性能を計測する
    score = cross_validate(clf, X, y, cv=skf, scoring='accuracy')
    mean_test_score = score.get('test_score').mean()
    print('acc:', mean_test_score)


if __name__ == '__main__':
    main()

上記の実行結果が次の通り。

$ python scikitannoy.py  
acc: 0.9669117647058824

めでたしめでたし。

Pythonではじめる機械学習 ―scikit-learnで学ぶ特徴量エンジニアリングと機械学習の基礎

Pythonではじめる機械学習 ―scikit-learnで学ぶ特徴量エンジニアリングと機械学習の基礎