Document

オンプレミス環境でベクトル検索をしてみる（OpenSearch）

昨今、生成AIやLLMが話題ですが、より回答精度を上げる方法としてRAGという言葉をよく聞きます。
RAG（Retrieval-Augmented Generation）とは、ざっくりいうと、FAQなどのデータをデータベースに格納しておき、ユーザーからのインプットに対して最適な回答をデータベースから抽出し、その回答をもとにLLMに文章を生成させることで、より正確に回答を生成できるようにする方法になります。

RAG用のデータベースでは、ベクトル検索という手法が主に使われており、入力されたキーワードを全文検索するのではなく、類似度で検索をする形式が多いようです。
今回は、このベクトル検索について触れていきたいと思います。ただ、AzureやAWSなどのクラウドサービスを使えば容易に実現できる思いますが、少し勉強しながら構築したいのでオンプレミス環境に導入していこうと思います。

なお、私はデータベースにそこまで詳しくなく、見様見真似で進めているので、間違った解釈があるかもしれませんがご容赦ください。
ただベクトル検索に興味のある方の参考になれば幸いです。

データベースの準備

今回ベクトル検索可能なデータベースとして、OpenSearchを使ってみたいと思います。
クラウド環境では、ElasticsearchやOpenSearchの名前をよく見ます。調べてもこのあたりのデータベースが良さそうという理由で採用します。
ちなみに、このElasticsearchはOSSであったものの、色々商用問題などがあったようですね。。。

さっそく、OpenSearchを導入していきたいと思います。
環境は、Ubuntu22.04です。

Javaをインストールする

OpenSearchの利用にはJavaが必要なため、JDKをインストールします。
利用するOpenSearchのバージョンにもよりますが、今回はJDK17をインストールしておきます。

sudo apt update
sudo apt install -y openjdk-17-jdk

OpenSearchの資材をダウンロードする

続いて、OpenSearchの資材をダウンロードします。

wget https://artifacts.opensearch.org/releases/bundle/opensearch/2.18.0/opensearch-2.18.0-linux-x64.tar.gz

展開して配置する

先程ダウンロードした資材を展開して、展開したものを/usr/share配下に移動させていきます。

tar -xvf opensearch-2.18.0-linux-x64.tar.gz
sudo mkdir /usr/share/opensearch
sudo mv opensearch-2.18.0/* /usr/share/opensearch

/usr/share/opensearch

sudo chown -R xxx:xxx /usr/share/opensearch

日本語用にプラグインをインストールする

下記コマンドでプラグインをインストールします。

sudo /usr/share/opensearch/bin/opensearch-plugin install analysis-kuromoji

設定を弄る

デフォルトのままだとエラーが出て、起動しないことが多いので設定を変更します。

sudo vi /usr/share/opensearch/config/opensearch.yml

discovery.type: single-node

plugins.security.disabled: true

discovery.type: single-node
# ---------------------------------- Gateway -----------------------------------

plugins.security.disabled: true

さらに、JVMの設定も変更します。

sudo vim /usr/share/opensearch/config/jvm.options

## -Xms1g
## -Xmx1g
-Xms4g
-Xmx4g

serviceファイルを作成して、自動起動させる

設定変更ができたら起動させるために、サービスファイルを作成します。

sudo vi /etc/systemd/system/opensearch.service

[Unit]
Description=OpenSearch

[Service]
User=xxxx
Type=simple
Environment="JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64"
ExecStart=/usr/share/opensearch/bin/opensearch

[Install]
WantedBy=multi-user.target

sudo systemctl enable opensearch
sudo systemctl start opensearch

sudo systemctl status opensearch

ベクトル検索用のプラグインを確認する

ベクトル検索用にknnのプラグインもインストールするのですが、もしかすると初期からインストールされているかもしれないので、下記コマンドで確認します。

curl -X GET http://localhost:9200/_cat/plugins?v

opensearch-knn 2.18.0.0

なお、上記curlコマンドが正常に返ってくれば、OpenSearchも正常に稼働していることになります。

以上でOpenSearchのインストールは完了です。

PythonでOpenSearchを操作する

今回はPythonを使ってOpenSearchを弄っていきます。
まずは必要なライブラリをインストールします。

pip3 install sentence-transformers opensearch-py

ライブラリのインストールができたら、さっそく操作をしていきます。

indexを作成する

下記のコードでindexを作成します。
今回は、ニュース記事のタイトルと本文、ベクトルデータを格納します。

from sentence_transformers import SentenceTransformer
from opensearchpy import OpenSearch, exceptions

# 設定
opensearch_host = "localhost"  # OpenSearchのホスト名
opensearch_port = 9200  # OpenSearchのポート番号
index_name = "news_articles_v2"  # インデックス名
model_name = "intfloat/multilingual-e5-large"

# Sentence Transformerモデルのロード
model = SentenceTransformer(model_name)

# OpenSearchクライアントの初期化
client = OpenSearch(
    hosts=[{'host': opensearch_host, 'port': opensearch_port}],
    http_auth=("admin", "admin"),  # 認証情報
    use_ssl=False, # SSL設定は必要に応じて変更
    verify_certs=False, # SSL検証は必要に応じて変更
    ssl_assert_hostname=False, # SSL検証は必要に応じて変更
)

# インデックス作成（既に存在する場合はスキップ）
if not client.indices.exists(index=index_name):
    client.indices.create(
        index=index_name,
        body={
            "mappings": {
                "properties": {
                    "title": {"type": "text"},
                    "data": {"type": "text"},
                    "embedding": {"type": "knn_vector", "dimension": 1024},  # Sentence Transformerの次元数
                }
            },
            "settings": {
                "index": {
                    "knn": True
                }
            }
        }
    )
client.close()

intfloat/multilingual-e5-large

データを投入する

indexが作成できたら、データを投入していきます。
今回は、ニュース記事のタイトルと本文をそれぞれベクトル化し、8:2で平均化しています。

from sentence_transformers import SentenceTransformer
from opensearchpy import OpenSearch, exceptions
import os
import re

def preprocess_text(text):
    """テキストの前処理を行う関数"""

# 設定
opensearch_host = "localhost"  # OpenSearchのホスト名
opensearch_port = 9200  # OpenSearchのポート番号
index_name = "news_articles_v2"  # インデックス名
model_name = "intfloat/multilingual-e5-large"

# Sentence Transformerモデルのロード
model = SentenceTransformer(model_name)

# OpenSearchクライアントの初期化
client = OpenSearch(
    hosts=[{'host': opensearch_host, 'port': opensearch_port}],
    http_auth=("admin", "admin"),  # 認証情報
    use_ssl=False, # SSL設定は必要に応じて変更
    verify_certs=False, # SSL検証は必要に応じて変更
    ssl_assert_hostname=False, # SSL検証は必要に応じて変更
)

def create_combined_embedding(question, answer):
  question_embedding = model.encode(preprocess_text(question))
  answer_embedding = model.encode(preprocess_text(answer))

  # 重み付き平均化 (質問:0.7, 回答:0.3)
  combined_embedding = (0.7 * question_embedding) + (0.3 * answer_embedding)
  return combined_embedding.tolist()

def ingest_news(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            line_array = line.split(',')
            title = line_array[0].strip()
            data = line_array[1].strip()
            embedding = create_combined_embedding(preprocess_text(title), preprocess_text(data))
            doc = {
                "title": title,
                "data": data,
                "embedding": embedding
            }
            client.index(index=index_name, body=doc)
            print(f"Indexed: {title}")
    client.indices.refresh(index=index_name)

# 指定したフォルダのファイルをすべて取得する
#csv_file=os.listdir('result/test.csv')

# ニュース記事ファイルのパス
news_file = "datas/data.csv"

# ニュース記事の投入
ingest_news(news_file)

client.close()

検索する

キーワードを入力し、そのキーワードをベクトル化して近い値を取得します。

from sentence_transformers import SentenceTransformer
from opensearchpy import OpenSearch

# 設定
opensearch_host = "localhost"  # OpenSearchのホスト名
opensearch_port = 9200  # OpenSearchのポート番号
index_name = "news_articles_v2"  # インデックス名
model_name = "intfloat/multilingual-e5-large"

# Sentence Transformerモデルのロード
model = SentenceTransformer(model_name)

# OpenSearchクライアントの初期化
client = OpenSearch(
    hosts=[{'host': opensearch_host, 'port': opensearch_port}],
    http_auth=("admin", "admin"),  # 認証情報
    use_ssl=False, # SSL設定は必要に応じて変更
    verify_certs=False, # SSL検証は必要に応じて変更
    ssl_assert_hostname=False, # SSL検証は必要に応じて変更
)

def search_news(keyword):
    """キーワードに類似するニュース記事を検索"""
    keyword_embedding = model.encode(keyword).tolist()
    query = {
      "knn": {
        "embedding": {
          "vector": keyword_embedding,
          "k": 5
        }
      }
    }

    res = client.search(index=index_name, body={"query": query})
    results = []

    for hit in res['hits']['hits']:
        results.append({
            'title': hit['_source']['title'],
            'data': hit['_source']['data'],
            'score': hit['_score']
        })
    return results

while True:
  # 検索キーワードの入力と検索
  keyword = input("検索キーワードを入力してください: ")

  if keyword == 'exit':
    break

  #results = search_news(keyword)
  results = search_news(keyword)
  print("\n検索結果:")
  count = 0
  for result in results:
      count += 1
      print(count, 'つ目：')
      print(f"スコア: {result['score']}, タイトル: {result['title']}")

client.close()

簡易的ですが、これでベクトル検索ができるようになりました。
テキストの前処理や重み付けの比率調整などをすれば、より精度を上げることができるのではないでしょうか。

さいごに

今回は、OpenSearchをインストールし、ベクトル検索をできるようにしました。
2,000件であっても、ほぼ一瞬で候補が表示されるので、性能的も問題はないかなと思います。
今後はより精度を上げる方法を試してみて、他のベータベースとして使えるようにしていきたいです。
そして行く末、LLM用のRAGとして利用できるようにしていきたいと思います。

今回はこのへんで、ではまた！

Power Automateでファイル名から拡張子を取得

2月 21, 2022

このブログを検索

プログラミング研究所