Pythonライブラリ「qdrant-client」徹底解説：ベクトル検索エンジンQdrantを使いこなす 🚀

はじめに：Qdrantとベクトル検索の世界へようこそ ✨

近年、AI技術の進化に伴い、テキスト、画像、音声などの非構造化データを効率的に扱う必要性が高まっています。特に、これらのデータを「意味」に基づいて検索するセマンティック検索（意味検索）が注目されています。

従来のキーワード検索では、「スポーツ」に関連する記事を探す場合、「野球」や「サッカー」といった具体的なキーワードを列挙する必要がありましたが、セマンティック検索では、「スポーツ」という抽象的な概念に近い文書を自動的に見つけ出すことができます。

このセマンティック検索を実現する中核技術がベクトル検索です。ベクトル検索では、データ（テキスト、画像など）をベクトルと呼ばれる数値の配列に変換（埋め込み/Embedding）し、このベクトル間の距離（類似度）を計算することで、意味的に近いデータを探し出します。

そして、このベクトル検索を高速かつ効率的に行うために開発されたのがベクトルデータベースです。今回ご紹介するQdrant（クアドラントと読みます）は、Rust言語で書かれたオープンソースの高性能ベクトルデータベース（ベクトル検索エンジン）です。

Qdrantは、高次元のベクトルデータを効率的に処理し、類似性検索だけでなく、追加情報（ペイロード）に基づいた高度なフィルタリング機能も提供します。本番環境での利用を想定した堅牢なサービスであり、ローカル、オンプレミス、クラウド（Qdrant Cloud）など様々な環境で利用可能です。

この記事では、QdrantをPythonから操作するための公式ライブラリである qdrant-client について、インストール方法から基本的な使い方、応用的な機能まで、網羅的に解説していきます。

インストール 💻

qdrant-client はPyPIで公開されており、pipを使って簡単にインストールできます。Python 3.8以上が推奨されます。仮想環境を作成してインストールすることをお勧めします。

# 仮想環境を作成 (例: venv)
python3 -m venv qdrant-env
source qdrant-env/bin/activate # Linux/macOS
# qdrant-env\Scripts\activate # Windows

# qdrant-client をインストール
pip install qdrant-client

qdrant-client は最小限の依存関係で動作しますが、テキストデータを直接Qdrantに追加し、内部で自動的にベクトル化（Embedding）を行いたい場合は、fastembed というオプションの依存関係を追加でインストールできます。fastembed は、軽量かつ高速な埋め込み生成ライブラリです。

# fastembed を含む qdrant-client をインストール
pip install 'qdrant-client[fastembed]'

基本的な使い方 🛠️

ここでは、qdrant-client の基本的な操作方法を見ていきましょう。

1. Qdrantへの接続

まず、Qdrantのインスタンスに接続するためのクライアントオブジェクトを作成します。接続方法はいくつかあります。

ローカルモード（インメモリ）

Qdrantサーバーを別途起動せずに、テストや簡単な実験のためにメモリ上でQdrantを実行できます。データは永続化されません。

from qdrant_client import QdrantClient

# インメモリモードでクライアントを初期化
client = QdrantClient(":memory:")

print("Qdrantクライアント (インメモリモード) が初期化されました。")

ローカルモード（ディスク永続化）

ローカルファイルシステムにデータを永続化するモードです。プロトタイピングに適しています。

from qdrant_client import QdrantClient

# ディスク永続化モードでクライアントを初期化 (指定したパスにデータが保存される)
client = QdrantClient(path="./qdrant_data")

print("Qdrantクライアント (ディスク永続化モード) が初期化されました。")

サーバーモード（Docker、オンプレミス、クラウド）

Dockerコンテナや自前のサーバー、またはQdrant Cloudで稼働しているQdrantインスタンスに接続します。本番環境では通常このモードを使用します。

from qdrant_client import QdrantClient

# Dockerでローカル実行しているQdrantに接続 (デフォルトポート)
# client = QdrantClient(host="localhost", port=6333)
# またはURLで指定
client = QdrantClient(url="http://localhost:6333")

# Qdrant Cloudに接続する場合 (APIキーが必要)
# client = QdrantClient(
#     url="YOUR_QDRANT_CLOUD_URL",
#     api_key="YOUR_API_KEY",
# )

# gRPCを使用する場合 (デフォルトポート: 6334)
# client = QdrantClient(host="localhost", grpc_port=6334, prefer_grpc=True)

print("Qdrantクライアント (サーバーモード) が初期化されました。")

2. コレクションの作成

Qdrantでは、ベクトルデータをコレクションという単位で管理します。これはリレーショナルデータベースのテーブルに似ています。コレクションを作成するには、ベクトル（Vector）の次元数（size）と距離計算方法（distance）を指定する必要があります。

from qdrant_client import QdrantClient, models

# 例: ローカルサーバーに接続
client = QdrantClient(host="localhost", port=6333)

COLLECTION_NAME = "my_first_collection"
VECTOR_SIZE = 768 # 例: BERTモデルの次元数

# コレクションが存在しない場合のみ作成する
if not client.collection_exists(collection_name=COLLECTION_NAME):
    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=models.VectorParams(size=VECTOR_SIZE, distance=models.Distance.COSINE)
        # vectors_config=models.VectorParams(size=VECTOR_SIZE, distance=models.Distance.EUCLID) # ユークリッド距離
        # vectors_config=models.VectorParams(size=VECTOR_SIZE, distance=models.Distance.DOT)    # 内積
    )
    print(f"コレクション '{COLLECTION_NAME}' を作成しました。")
else:
    print(f"コレクション '{COLLECTION_NAME}' は既に存在します。")

# コレクション情報を取得
collection_info = client.get_collection(collection_name=COLLECTION_NAME)
print("\nコレクション情報:")
print(collection_info)

利用可能な距離計算方法（models.Distance）:

• COSINE: コサイン類似度 (ベクトル間の角度を基にする。方向性が近いほど類似度が高い)
• EUCLID: ユークリッド距離 (ベクトル間の直線距離)
• DOT: 内積 (Dot Product)

使用する埋め込みモデル（Embedding Model）やタスクに応じて適切な距離尺度を選択することが重要です。

3. ポイントの追加・更新 (Upsert)

コレクションには、ポイント (Point) を追加します。各ポイントは、一意のID、ベクトル、そしてオプションでペイロード (Payload) と呼ばれる追加データ（JSON形式）を持ちます。

upsert メソッドを使うと、指定したIDのポイントが存在しない場合は新規に追加し、存在する場合は更新します。

import numpy as np
from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"
VECTOR_SIZE = 768

# 追加/更新するポイントのリストを作成
points_to_upsert = [
    models.PointStruct(
        id=1,
        vector=np.random.rand(VECTOR_SIZE).tolist(), # 768次元のランダムなベクトル
        payload={"document_title": "最初のドキュメント", "year": 2024, "tags": ["tech", "ai"]}
    ),
    models.PointStruct(
        id=2,
        vector=np.random.rand(VECTOR_SIZE).tolist(),
        payload={"document_title": "２番目の記事", "year": 2023, "tags": ["python", "qdrant"]}
    ),
    # IDにUUIDを使用することも可能
    models.PointStruct(
        id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        vector=np.random.rand(VECTOR_SIZE).tolist(),
        payload={"document_title": "UUIDを使ったポイント", "year": 2024}
    ),
]

# ポイントをUpsert (バッチ処理)
operation_info = client.upsert(
    collection_name=COLLECTION_NAME,
    points=points_to_upsert,
    wait=True # 操作が完了するまで待機する (デフォルトはFalse)
)

print("\nUpsert操作の結果:")
print(operation_info)

# コレクション内のポイント数をカウント
count_result = client.count(collection_name=COLLECTION_NAME, exact=True)
print(f"\nコレクション '{COLLECTION_NAME}' の現在のポイント数: {count_result.count}")

# fastembed がインストールされている場合
docs = ["これは最初の文書です。", "これは二番目の文書です。"]
metadata = [
    {"source": "doc1.txt"},
    {"source": "doc2.txt"}
]
ids = [10, 11] # IDを指定、省略するとUUIDが自動生成される

client.add(
    collection_name=COLLECTION_NAME,
    documents=docs,
    metadata=metadata,
    ids=ids
)
print("FastEmbedを使ってドキュメントを追加しました。")

4. ベクトル検索 (Similarity Search)

Qdrantの最も重要な機能の一つが、指定したベクトルに類似するベクトル（ポイント）を検索することです。

import numpy as np
from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"
VECTOR_SIZE = 768

# 検索に使用するクエリベクトル (例: ランダムベクトル)
query_vector = np.random.rand(VECTOR_SIZE).tolist()

# 類似ベクトルを検索
search_result = client.search(
    collection_name=COLLECTION_NAME,
    query_vector=query_vector,
    limit=3 # 上位3件を取得
)

print("\n類似ベクトル検索の結果:")
for hit in search_result:
    print(f"  ID: {hit.id}, Score: {hit.score:.4f}, Payload: {hit.payload}")

# fastembed が有効な場合はテキストで検索も可能
# search_result_text = client.query(
#     collection_name=COLLECTION_NAME,
#     query_text="検索したいテキスト",
#     limit=3
# )
# print("\nテキストによる検索結果:")
# for hit in search_result_text:
#     print(f"  ID: {hit.id}, Document: {hit.document}") # addメソッドで追加した場合

score はクエリベクトルとの類似度（または距離）を表します。models.Distance.COSINE の場合は値が大きいほど類似度が高く（最大1.0）、EUCLID の場合は値が小さいほど距離が近く類似度が高いことを示します。

ペイロードによるフィルタリング検索

ベクトル検索と同時に、ペイロードの条件で結果を絞り込むことができます。これにより、例えば「2024年以降に作成され、’ai’タグが付いているドキュメントの中から、クエリベクトルに最も類似するもの」といった検索が可能になります。

from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"
VECTOR_SIZE = 768
query_vector = np.random.rand(VECTOR_SIZE).tolist()

# フィルター条件を作成: yearが2024以上 かつ tagsに"ai"が含まれる
query_filter = models.Filter(
    must=[
        models.FieldCondition(
            key="year", # ペイロードのキー
            range=models.Range(gte=2024) # gte: Greater Than or Equal (以上)
        ),
        models.FieldCondition(
            key="tags", # ペイロードのキー
            match=models.MatchValue(value="ai") # 特定の値にマッチ
        )
    ]
    # must_not=[ ... ] や should=[ ... ] も指定可能
)

# フィルタリング検索を実行
search_result_filtered = client.search(
    collection_name=COLLECTION_NAME,
    query_vector=query_vector,
    query_filter=query_filter,
    limit=3
)

print("\nフィルタリング検索の結果:")
if search_result_filtered:
    for hit in search_result_filtered:
        print(f"  ID: {hit.id}, Score: {hit.score:.4f}, Payload: {hit.payload}")
else:
    print("  条件に一致するポイントは見つかりませんでした。")

フィルター条件は非常に柔軟で、範囲指定（range）、完全一致（match）、地理空間検索（geo_bounding_box, geo_radius）など、様々な条件を組み合わせることができます。

5. ポイントの取得

IDを指定して特定のポイントを取得できます。

from qdrant_client import QdrantClient

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"

# 取得したいポイントのIDリスト
point_ids = [1, "a1b2c3d4-e5f6-7890-abcd-ef1234567890"]

# ポイントを取得 (with_payload=Trueでペイロードも取得, with_vector=Trueでベクトルも取得)
retrieved_points = client.retrieve(
    collection_name=COLLECTION_NAME,
    ids=point_ids,
    with_payload=True,
    with_vectors=False # 通常、ベクトル自体は不要なことが多い
)

print("\nID指定で取得したポイント:")
for point in retrieved_points:
    print(f"  ID: {point.id}, Payload: {point.payload}")

6. ポイントの削除

IDを指定してポイントを削除します。

from qdrant_client import QdrantClient

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"

# 削除したいポイントのIDリスト
point_ids_to_delete = [2]

# ポイントを削除
operation_info = client.delete(
    collection_name=COLLECTION_NAME,
    points_selector=models.PointIdsList(points=point_ids_to_delete),
    wait=True
)

print("\n削除操作の結果:")
print(operation_info)

# フィルター条件に一致するポイントを削除することも可能
# delete_filter = models.Filter(...)
# client.delete(
#     collection_name=COLLECTION_NAME,
#     points_selector=models.FilterSelector(filter=delete_filter)
# )

7. コレクションの削除

コレクション全体を削除します。この操作は元に戻せないので注意が必要です。

from qdrant_client import QdrantClient

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"

# コレクションを削除
try:
    result = client.delete_collection(collection_name=COLLECTION_NAME)
    if result:
        print(f"\nコレクション '{COLLECTION_NAME}' を削除しました。")
    else:
        print(f"\nコレクション '{COLLECTION_NAME}' の削除に失敗しました。")
except Exception as e:
    print(f"\nコレクション '{COLLECTION_NAME}' の削除中にエラーが発生しました: {e}")

応用的な機能 ⚙️

qdrant-client は基本的なCRUD操作以外にも、高度な機能を提供しています。

非同期クライアント (Async Client)

asyncio を使用した非同期アプリケーション向けに、非同期クライアント AsyncQdrantClient も提供されています。基本的なAPIは同期クライアントと同じですが、メソッド呼び出しに await を使用します。

import asyncio
import numpy as np
from qdrant_client import AsyncQdrantClient, models

async def main():
    COLLECTION_NAME = "async_collection"
    VECTOR_SIZE = 128

    # 非同期クライアントを初期化
    async_client = AsyncQdrantClient(host="localhost", port=6333)

    # 非同期でコレクションを作成
    await async_client.recreate_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=models.VectorParams(size=VECTOR_SIZE, distance=models.Distance.COSINE)
    )
    print(f"非同期でコレクション '{COLLECTION_NAME}' を作成/再作成しました。")

    # 非同期でポイントをUpsert
    await async_client.upsert(
        collection_name=COLLECTION_NAME,
        points=[
            models.PointStruct(id=i, vector=np.random.rand(VECTOR_SIZE).tolist(), payload={"num": i})
            for i in range(100) # 100個のポイント
        ],
        wait=True
    )
    print("非同期でポイントを追加しました。")

    # 非同期で検索
    query_vector = np.random.rand(VECTOR_SIZE).tolist()
    search_result = await async_client.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_vector,
        limit=5
    )
    print("\n非同期検索の結果:")
    for hit in search_result:
        print(f"  ID: {hit.id}, Score: {hit.score:.4f}")

    # 忘れずにクライアントを閉じる (非同期コンテキストマネージャを使うのが良い)
    await async_client.close() # Alternatively use 'async with AsyncQdrantClient(...) as client:'

if __name__ == "__main__":
    asyncio.run(main())

ペイロードインデックス

ペイロードの特定のキーに対してインデックスを作成することで、フィルタリング検索のパフォーマンスを向上させることができます。

from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "indexed_collection"

# "year" フィールド (integer) にインデックスを作成
client.create_payload_index(
    collection_name=COLLECTION_NAME,
    field_name="year",
    field_schema=models.PayloadSchemaType.INTEGER
)
print("ペイロードインデックス 'year' を作成しました。")

# "tags" フィールド (keyword) にインデックスを作成
client.create_payload_index(
    collection_name=COLLECTION_NAME,
    field_name="tags",
    field_schema=models.PayloadSchemaType.KEYWORD
)
print("ペイロードインデックス 'tags' を作成しました。")

# インデックス情報を取得
collection_info = client.get_collection(collection_name=COLLECTION_NAME)
print("\n更新後のコレクション情報 (ペイロードインデックス):")
print(collection_info.payload_schema)

# インデックスを削除する場合
# client.delete_payload_index(collection_name=COLLECTION_NAME, field_name="year")

レコメンデーション API

指定したポイントID（複数可）に類似する他のポイントを検索するレコメンデーション機能があります。協調フィルタリングのような使い方が可能です。

from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection" # 事前にデータが入っているとする

# ポジティブな例 (これらのポイントに似たものを探す)
positive_example_ids = [1]
# ネガティブな例 (これらのポイントに似ていないものを探す)
negative_example_ids = [] # Optional

# レコメンデーション検索を実行
recommendations = client.recommend(
    collection_name=COLLECTION_NAME,
    positive=positive_example_ids,
    negative=negative_example_ids,
    limit=5
)

print(f"\nID {positive_example_ids} に基づくレコメンデーション:")
if recommendations:
    for hit in recommendations:
        # positive/negativeで指定したID自身は結果から除外される
        print(f"  ID: {hit.id}, Score: {hit.score:.4f}, Payload: {hit.payload}")
else:
    print("  レコメンド結果が見つかりませんでした。")

バッチ処理とスクロール

大量のデータを扱う場合、upsert や delete をバッチで実行する方が効率的です。また、コレクション内の全ポイントを効率的に取得するために scroll API が用意されています。

from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"

# Scroll API を使って全ポイントを取得 (バッチ単位で取得)
print("\nScroll API によるポイント取得:")
next_page_offset = None
all_points_count = 0
while True:
    scroll_result, next_page_offset = client.scroll(
        collection_name=COLLECTION_NAME,
        limit=10, # 1回のscrollで取得する最大件数
        offset=next_page_offset,
        with_payload=True,
        with_vectors=False
    )
    if not scroll_result:
        break # ポイントがなくなったらループ終了

    print(f"  取得したポイント数: {len(scroll_result)}")
    # for point in scroll_result:
    #     print(f"    ID: {point.id}")
    all_points_count += len(scroll_result)

    if next_page_offset is None:
        break # 次のページがない場合は終了

print(f"  取得した総ポイント数: {all_points_count}")

スナップショット

コレクションやクラスター全体のスナップショット（バックアップ）を作成・復元する機能があります。データの保全や移行に役立ちます。

from qdrant_client import QdrantClient

client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "my_first_collection"

# コレクションのスナップショットを作成
snapshot_description = client.create_snapshot(collection_name=COLLECTION_NAME)
print("\nスナップショットを作成しました:")
print(snapshot_description)

# スナップショットの一覧を取得
snapshots = client.list_snapshots(collection_name=COLLECTION_NAME)
print("\n利用可能なスナップショット:")
for snapshot in snapshots:
    print(f"  Name: {snapshot.name}, Size: {snapshot.size}, Creation Time: {snapshot.creation_time}")

# スナップショットから復元する場合 (既存のコレクションは上書きされる可能性あり)
# client.recover_snapshot(
#     collection_name="restored_collection", # 新しいコレクション名 or 既存のコレクション名
#     location=f"http://localhost:6333/collections/{COLLECTION_NAME}/snapshots/{snapshot_description.name}"
# )

# スナップショットを削除する場合
# client.delete_snapshot(collection_name=COLLECTION_NAME, snapshot_name=snapshot_description.name)

その他

• 名前付きベクトル (Named Vectors): 1つのポイントに複数の名前付きベクトルを持たせることができます。異なる埋め込みモデルの結果を同時に格納するなどの用途に利用できます。
• シャーディングとレプリケーション: 大規模データや高可用性が求められる場合、Qdrantクラスターを構成し、コレクションをシャーディング（分割）したり、レプリケーション（複製）したりできます。（設定は主にQdrantサーバー側で行います）
• 量子化 (Quantization): メモリ使用量を削減し、検索速度を向上させるために、ベクトルデータを圧縮する量子化技術（バイナリ量子化、スカラー量子化など）をサポートしています。コレクション作成時に設定します。

ユースケース 🎯

Qdrantと qdrant-client は、様々なAIアプリケーションで活用できます。

• セマンティック検索エンジン: テキストドキュメントや記事をベクトル化してQdrantに格納し、自然言語のクエリに対して意味的に関連性の高い文書を検索します。
• 類似画像検索: 画像の特徴量をベクトルとして抽出し、類似する画像を検索します。
• レコメンデーションシステム: ユーザーの行動履歴やアイテムの特徴をベクトル化し、類似アイテムやユーザーへの推薦を行います。
• 質問応答システム (QA): 大量のテキストデータ（マニュアル、FAQなど）をベクトル化しておき、質問文に最も類似するテキスト断片を見つけ出し、回答生成の根拠として利用します (RAG: Retrieval-Augmented Generation)。
• 異常検知: 正常なデータのベクトル分布を学習し、分布から大きく外れたベクトルを持つデータを異常として検出します。
• 重複検出: テキストや画像のベクトルが非常に近い場合に、重複コンテンツとして検出します。

特に、LangChain や LlamaIndex といったLLM（大規模言語モデル）アプリケーション開発フレームワークとの連携が容易であり、これらのフレームワークのベクトルストア（Vector Store）としてQdrantを利用するケースが増えています。

まとめ 🏁

qdrant-client は、高性能なベクトルデータベース Qdrant を Python から簡単に操作するための強力なライブラリです。主な特徴をまとめます。

• 使いやすいAPI 同期・非同期の両方のクライアントを提供し、直感的なメソッドで操作可能。
• 豊富な機能 基本的なCRUD操作に加え、高度なフィルタリング、レコメンデーション、ペイロードインデックス、スナップショットなどをサポート。
• 柔軟な接続性 ローカルモード（インメモリ/ディスク）、サーバーモード（Docker/オンプレミス/クラウド）に対応。
• 型ヒント 全てのAPIメソッドに型ヒントが提供されており、コードの安全性と可読性を向上。
• エコシステム連携 LangChain, LlamaIndex, Haystack など、主要なAI/LLMフレームワークとの連携が容易。
• FastEmbed連携 (Optional) テキストデータのベクトル化をクライアント内部でシームレスに実行可能。

Qdrantと qdrant-client を活用することで、セマンティック検索やレコメンデーションなど、ベクトルデータが鍵となる次世代のAIアプリケーションを効率的に開発することができます。ぜひ、あなたのプロジェクトで試してみてください！🚀