ElasticSearchをPythonで使う方法

概要

この記事では、オープンソースの分散型検索エンジンである ElasticSearch を Python で使う方法について解説します。

前提

ElasticSearch を使うにあたり、そのセットアップを終わらせていることを前提としています。

1. ElasticSearch のインストール

以下のシェルコマンドを使って、Python クライアントをインストールします。Python のバージョンは 3.10 以降を想定しています。

pip install elasticsearch

Jupyter ノートブックを使って ElasticSearch をインストールすることもできます。

!pip install elasticsearch

2. 使用法

ElasticSearch の import と初期化

ローカル構築の場合

from elasticsearch import Elasticsearch
ELASTIC_HOST_ID = ''
ELASTIC_PASSWORD = ''
es = Elasticsearch(
    hosts=ELASTIC_HOST_ID,
    basic_auth=(
        'elastic',
        ELASTIC_PASSWORD,
    ),
)
es.info() # 接続チェック

クラウド利用の場合

from elasticsearch import Elasticsearch
ELASTIC_CLOUD_ID = '' # Cloud IDは ElasticCloud管理画面に見れます
ELASTIC_PASSWORD = ''
client = Elasticsearch(
    cloud_id=ELASTIC_CLOUD_ID,
    basic_auth=(
        'elastic',
        ELASTIC_PASSWORD,
    ),
)
client.info() # 接続チェック

インデックスの作成

Step 1 インデックス作成に必要な設定を以下のPythonのスクリプトで定義する。

indexSettings = {
    "analysis": {
        "analyzer": {
            "custome_ja_analyzer": {
                "type": "custom",
                "char_filter": ["icu_normalizer", "html_strip"], # 文字フィルターとHTMLタグフィルタ
                "tokenizer": "kuromoji_tokenizer",
                "filter": [
                    "kuromoji_baseform",
                    "kuromoji_part_of_speech",
                    "ja_stop",
                    "cjk_width",
                    "kuromoji_number",
                    "kuromoji_stemmer",
                    "lowercase",
                ],
            },
        }
    }
}

ICU analysis pluginについては、こちらを参照してください。

Kuromoji analysis pluginにつきまして、こちらを参照してください。

Step 2 インデックスに必要なフィールドを以下のPythonのスクリプトで定義する。

indexMappings = {
    "properties": {
        "document_id": {"type": "keyword"}, #　他のタイプについてelastic search HPに参照してください。
        "content": {
            "type": "keyword",
            "fields": {
                "search": {
                    "type": "text",
                    "analyzer": "custome_ja_analyzer", #　Step 1に定義された日本語アナライザー
                },
            },
        },
        "embedding": {
            "type": "dense_vector", # ベクトルタイプ
            "dims": 1024, # ベクトルの次元数
            "index": True,
            "similarity": "cosine", #　類似度計算方法
            "index_options": {
                "type": "hnsw", # Hierarchical Navigable Small World, ANN検索アルゴリズムの一種
                "m": 16, # HNSWパラメーターの一つ、新たなedgeを入れる時にスコア計算式に影響する対象ノードに直接繋げているノードの数
                "ef_construction": 64, # HNSWパラメーターの一つ、対象ノース周りのネットワークを構築する時にアクセス必要なノード数
            },
        },
        "metadata": {"type": "object", "enabled": False}, # メタデータ
    }
}

** Step 3** インデックスの作成

下記のpythonスクリプトでインデックスを作成

INDEX_NAME = ''
client.indices.create(
    index=INDEX_NAME,
    settings=indexSettings,
    mappings=indexMappings,
)

ベクトルの挿入

以下では、ベクトルを挿入します。

from tqdm import tqdm
INDEX_NAME = '' # index name
for i,doc in tqdm(enumerate(docs)): # `docs` is the list of Langchain Documents
    # insert data
    es_doc = {
        'doc_id':doc.metadata['doc_id'],
        'text':doc.page_content,
        'embedding':dense_vec,
        'metadata': doc.metadata
    }
    resp = client.index(index=INDEX_NAME, id=doc.metadata['doc_id'], document=es_doc)

インデックスにあるドキュメントの取得

以下では、あるドキュメントを取得します。

client.get(index=INDEX_NAME, id="my_document_id")

インデックスのクエリ

以下では、インデックスをクエリします。

# 下記はハイブリッド検索の場合に使うコード。単なるベクトル検索の場合には、`query`の部分を除いて大丈夫です。
client.search(
    index=INDEX_NAME,
    size = 10, # top_10ドキュメントを持ってくる
    query={ # 全文検索の部分
        "match": {
            "content": {
                "query": "foo"
            }
        }
   },
   knn={ #　ベクトル検索の部分
        "field": "embedding",
        "query_vector": dense_vec, # クエリとembedding modelにより生成されたdense vector
        "k": 30, # top kの結果を返す
        "num_candidates": 500, # HNSWアルゴリズムにより探索に必要な近傍ノード数
  },
)

ドキュメントの更新

以下では、ドキュメントのIDによりドキュメントを更新します。

client.update(
    index=INDEX_NAME,
    id="my_document_id",
    doc={ #　更新したいところだけを埋めればいい
        "content": "bar",
    },
)

ドキュメントの削除

以下では、ドキュメントのIDによりドキュメントを削除します。

client.delete(index=INDEX_NAME, id="my_document_id")

インデックスの削除

以下では、インデックスを削除します。

if client.indices.exists(index=INDEX_NAME): # インデックスが存在しているかどうか
    client.indices.delete(index=INDEX_NAME)

3. RAG開発向けのLLMとの接続

Elastic SearchのPython APIに直接LLMとの接続用のメソッドが提供されていません。 Elastic Searchからドキュメントを取ってきて、プロンプトに必要な部分を組み込み、LLMに投げるのは一般的なやり方です。

参照

備考

Hakky ではエンジニアを募集中です！まずは話してみたいなどでも構いませんので、ぜひお気軽に採用ページからお問い合わせくださいませ。