今やっている仕事の中で、ベクトルデータベースであるQdrantの仕組みについて学ぶ必要が出てきたため、自分のローカルであるubuntu/docker環境で、実際にQdrantを動かしてクエリ検索する手順を、zennの「Qdrant ベクトル検索エンジン」という記事を参考にテストしましたのでその方法や手順をシェアしたいと思います。
Qdrantによる類似ドキュメントのクエリ検索の手順
- ベクトルデータの準備 (Livedoorニュースコーパス)
- コーパスをディクショナリ登録(json)
- コーパスのベクトル変換(Google Colabo)
- ubuntu/dockerでQdrantサーバ起動(docker)
- Qdrantでコレクションの作成
- ベクトル変換したドキュメントを登録
- クエリから類似ドキュメントの検索
1.ベクトルデータの準備 (Livedoorニュースコーパス)
ライブドアニュースコーパスをDL
https://www.rondhuit.com/download.html#ldcc
ダウンロード（通常テキスト）：ldcc-20140209.tar.gz
2.Livedoorニュースコーパスをディクショナリ登録(json)
import json
import datetime
from typing import List, Dict
from pathlib import Path
import random
CORPUS_DIR = './livedoor-corpus' # ライブドアコーパスをここにおく
QDRANT_JSON = 'livedoor.json'
SAMPLE_TEXT_LEN: int = 500 # ドキュメントを500文字でトランケート
def read_document(path: Path) -> Dict[str, str]:
"""1ドキュメントの処理"""
with open(path, 'r') as f:
lines: List[any] = f.readlines(SAMPLE_TEXT_LEN)
lines = list(map(lambda x: x.rstrip(), lines))
d = datetime.datetime.strptime(lines[1], "%Y-%m-%dT%H:%M:%S%z")
created_at = int(round(d.timestamp())) # 数値(UNIXエポックタイプ)に変換
return {
"url": lines[0],
"publisher": path.parts[1], # ['livedoor-corpus', 'it-life-hack', 'it-life-hack-12345.txt']
"created_at": created_at,
"body": ' '.join(lines[2:]) # 初めの２行をスキップし、各行をスペースで連結し、１行にする。
}
def load_dataset_from_livedoor_files() -> (List[List[float]], List[str]):
# NB. exclude LICENSE.txt, README.txt, CHANGES.txt
corpus: List[Path] = list(Path(CORPUS_DIR).rglob('*-*.txt'))
random.shuffle(corpus) # 記事をシャッフルします
with open(QDRANT_JSON, 'w') as fp:
for x in corpus:
doc: Dict[str, str] = read_document(x)
json.dump(doc, fp) # 1行分
fp.write('\n')
if __name__ == '__main__':
load_dataset_from_livedoor_files()
生成されたlivedoor.jsonをGoogle collaboで使います。
Google Colaboでコーパス(livedoor.json)をベクトル変換
!pip install -U ginza spacy
!pip install -U numpy pandas ja_ginza
colaboで辞書をベクトル化 ... 約10分ぐらいかかりました。
import numpy as np
import pandas as pd
import spacy
# from multiprocessing import Pool, cpu_count <- マルチプロセス関連は不要
# GiNZAモデルのロード (インストールが完了している前提)
try:
nlp: spacy.Language = spacy.load('ja_ginza', exclude=["tagger", "parser", "ner", "lemmatizer", "textcat", "custom"])
print("✅ GiNZAモデルのロードに成功しました。")
except OSError:
print("❌ GiNZAモデルが見つかりません。再度インストール手順を確認してください。")
# ここでエラーになる場合は、!pip install -U ginza を実行してください。
QDRANT_NPY = 'vectors-livedoor-ginza.npy' # 出力ファイル名
def f(x):
# NaNやNone値のチェック (エラー回避のため)
if pd.isna(x):
# 空のベクトルを返す、または処理をスキップ
return np.zeros(nlp.vocab.vectors_length)
doc: spacy.tokens.doc.Doc = nlp(x) # GiNZAでベクトル化
return doc.vector
def main():
try:
df = pd.read_json('livedoor.json', lines=True)
except FileNotFoundError:
print("❌ livedoor.json が見つかりません。ファイルが /content/ にアップロードされているか確認してください。")
return
print("\nデータフレームの先頭5行:")
print(df.head())
print(f"\n合計 {len(df)} 件の文書をベクトル化中... (シングルプロセス)")
# 修正箇所: df.body.apply(f) を使用してシングルプロセスでベクトル化
vectors_list = df.body.apply(f).tolist()
print("ベクトル化完了。NumPyファイルに保存中...")
# リストをNumPy配列に変換して保存
vectors_array = np.array(vectors_list)
np.save(QDRANT_NPY, vectors_array, allow_pickle=False)
print(f"\n========================================================")
print(f"✅ 保存完了: {QDRANT_NPY}")
print(f"配列の形状 (Shape): {vectors_array.shape}")
print(f"========================================================")
# 処理の実行
main()
vectors-livedoor-ginza.npy ができます。
4. ubuntu/dockerでQdrantサーバ起動(docker)
$ sudo docker pull qdrant/qdrant
$ sudo docker run -p 6333:6333
-v $(pwd)/qdrant_storage:/qdrant/storage
qdrant/qdrant
5. Qdrantでコレクションの作成
- コレクション: RDBテーブル
- ポイント: RDBレコード ポイントには、ペイロード(Payload)と呼ばれるメタ情報も一緒に登録できる。メタ情報はフィルター検索に使用する
from qdrant_client import QdrantClient
from qdrant_client.http.models import VectorParams, Distance
collection_name = 'livedoor'
qdrant_client = QdrantClient(host='localhost', port=6333)
qdrant_client.recreate_collection(
collection_name=collection_name,
vectors_config=VectorParams(size=300, distance=Distance.COSINE) # GiNZAは300次元
)
6.ベクトル変換したドキュメントを登録
import json
import numpy as np
import pandas as pd
from qdrant_client import QdrantClient
# =========================================================================
# 1. 補助関数の定義 (JSONファイルの読み込み用)
# livedoor.jsonには不要なキーが含まれている可能性があるため、
# 必要なキーだけを抽出する目的の関数です。
# =========================================================================
def hook(obj):
"""
JSONオブジェクトから必要なペイロードデータのみを抽出するフック関数。
"""
if 'body' in obj:
# 必要なキー（本文、タイトル、カテゴリ）を抽出して返す
return {
"title": obj.get("title", ""),
"body": obj.get("body", ""),
"category": obj.get("category", "")
}
return obj
# =========================================================================
# 2. メイン処理
# =========================================================================
def main():
# 接続情報
collection_name = 'livedoor'
qdrant_client = QdrantClient(host='localhost', port=6333)
# データの読み込み
try:
# ベクトルデータの読み込み
vectors = np.load('./vectors-livedoor-ginza.npy')
# JSONファイルの読み込みとペイロードの準備
print("JSONファイルを読み込んでペイロードを準備中...")
docs = []
with open('./livedoor.json', 'r', encoding='utf-8') as fd:
# 各行（一つのJSONオブジェクト）を読み込み、hook関数で必要なキーを抽出
for line in fd:
docs.append(json.loads(line, object_hook=hook))
print(f"✅ 読み込み完了。ベクトル数: {vectors.shape[0]}、文書数: {len(docs)}")
except FileNotFoundError as e:
print(f"❌ ファイルが見つかりません: {e.filename}")
print("ファイル（livedoor.json, vectors-livedoor-ginza.npy）が同じディレクトリにあるか確認してください。")
return
# コレクションへのアップロード
print("Qdrantコレクションにデータをアップロード中...")
qdrant_client.upload_collection(
collection_name=collection_name, # コレクション名
vectors=vectors, # ベクトルデータ (NumPy配列)
payload=iter(docs), # ペイロードデータ (ジェネレータまたはイテレータ)
ids=None, # IDの自動発番
batch_size=256 # バッチサイズ
)
print("✅ データアップロード完了。")
# 最終確認
collection_info = qdrant_client.get_collection(collection_name='livedoor')
print(f"最終ポイント数: {collection_info.points_count}")
# スクリプトの実行
if __name__ == "__main__":
main()
7.クエリから類似ドキュメントの検索
import numpy as np
import spacy
from qdrant_client import QdrantClient
from qdrant_client.http.models import ScoredPoint
# =========================================================================
# 1. 初期設定とモデルロード
# =========================================================================
# ベクトル化に使用したモデルと同じものをロード
# 以前のステップでインストールが完了していることを前提とします
try:
nlp: spacy.Language = spacy.load('ja_ginza', exclude=["tagger", "parser", "ner", "lemmatizer", "textcat", "custom"])
print("✅ GiNZAモデルのロードに成功しました。")
except OSError:
print("❌ GiNZAモデルが見つかりません。")
exit() # 処理を中断
# Qdrant接続情報
collection_name = 'livedoor'
qdrant_client = QdrantClient(host='localhost', port=6333)
# 検索クエリ
QUERY_TEXT = "野球情報が知りたい"
# =========================================================================
# 2. クエリテキストのベクトル化
# =========================================================================
def get_vector_from_text(text: str) -> np.ndarray:
"""
GiNZAを使用してテキストをベクトルに変換します。
"""
doc: spacy.tokens.doc.Doc = nlp(text)
# GiNZAのdoc.vectorはNumPy配列を返します
return doc.vector
# =========================================================================
# 3. Qdrantでの検索実行
# =========================================================================
def main():
print(f"\n========================================================")
print(f"🔍 検索クエリ: {QUERY_TEXT}")
print(f"========================================================")
# クエリテキストをベクトルに変換
query_vector = get_vector_from_text(QUERY_TEXT)
# Qdrantで検索を実行
hits = qdrant_client.search(
collection_name=collection_name,
query_vector=query_vector, # ベクトル化したクエリー
query_filter=None,
with_payload=True, # レスポンスにペイロードを含める
limit=5 # 上位5件を取得
)
# 検索結果の表示
print("\n[検索結果 - 上位 5件]")
if not hits:
print("類似記事は見つかりませんでした。")
return
for i, hit in enumerate(hits):
h: ScoredPoint = hit
# ペイロードからタイトルと本文を取得
title = h.payload.get('title', 'N/A')
body_snippet = h.payload.get('body', 'N/A')[:100] + '...' # 本文は先頭100文字を抜粋
print(f"--- 順位 {i+1} (スコア: {h.score:.4f}) ---")
print(f"タイトル: {title}")
print(f"本文抜粋: {body_snippet}")
# スクリプトの実行
if __name__ == "__main__":
main()
まとめ
仕事でQdrantを使っているなら、絶対に試した方がいいですね！