Weaviate Pythonクライアント徹底解説：ベクトル検索をもっと身近に 🚀

Weaviate Pythonクライアントを利用するには、まずPython環境が必要です。公式にはPython 3.8以上がサポートされています (v4.11.3時点)。インストールはpipコマンドを使って簡単に行えます。

最新の安定版 (v4系) をインストールするには、以下のコマンドを実行します。v4クライアントはgRPCを内部的に利用することで、v3クライアントと比較して大幅なパフォーマンス向上が図られています。特にバッチ処理やクエリ速度が改善されています。

pip install -U weaviate-client

注意点: v4クライアントはWeaviateサーバーのバージョン 1.23.7 以降が必要です。また、v4クライアントはgRPCを使用するため、WeaviateサーバーのgRPCポート（デフォルト: 50051）への接続が許可されている必要があります。Docker ComposeでWeaviateを起動している場合は、`ports` の設定で `50051:50051` が公開されていることを確認してください。

Pythonクライアントを使ってWeaviateインスタンスに接続する方法はいくつかあります。v4クライアントでは、接続方法に応じた便利なヘルパー関数が用意されています。

1. ローカルインスタンスへの接続

Docker Composeなどでローカルマシン上でWeaviateを実行している場合に最も一般的な方法です。

import weaviate
import os

# Weaviateがデフォルト設定 (http://localhost:8080, grpc://localhost:50051) で動作している場合
# 認証が無効な場合
client = weaviate.connect_to_local()

# Weaviateが異なるポートや認証設定で動作している場合
# client = weaviate.connect_to_local(
#     host="localhost",
#     port=8080,
#     grpc_port=50051,
#     auth_client_secret=weaviate.AuthApiKey(api_key="YOUR-WEAVIATE-API-KEY"), # 認証が必要な場合
#     headers={
#         "X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY"), # 外部APIキー(例: OpenAI)
#         "X-Cohere-Api-Key": os.getenv("COHERE_APIKEY"),
#     }
# )

try:
    # 接続を確認
    print(f"Client is ready: {client.is_ready()}")

    # ここにWeaviate操作のコードを記述

finally:
    # 接続を閉じる (v4クライアントでは重要！)
    client.close()

print("Connection closed.")

v4クライアントでは、client.close() を呼び出して接続を明示的に閉じることが推奨されています。`try…finally` ブロックを使用すると、エラーが発生した場合でも確実に接続が閉じられます。

2. Weaviate Cloud Services (WCS) への接続

Weaviateが提供するマネージドサービスであるWCS上のインスタンスに接続します。

import weaviate
import os

# WCSコンソールから取得した情報を設定
wcs_url = "YOUR-WCS-INSTANCE-URL" # 例: "https://my-sandbox-cluster.weaviate.network"
wcs_api_key = os.getenv("WCS_API_KEY")
openai_api_key = os.getenv("OPENAI_APIKEY") # 例としてOpenAIのAPIキー

client = weaviate.connect_to_wcs(
    cluster_url=wcs_url,
    auth_client_secret=weaviate.AuthApiKey(api_key=wcs_api_key),
    headers={
        "X-OpenAI-Api-Key": openai_api_key, # WCSで使用するモジュールに応じたAPIキー
    }
)

try:
    print(f"Client is ready: {client.is_ready()}")
    # WCSインスタンスに対する操作
finally:
    client.close()

print("Connection closed.")

WCSの場合、RESTのエンドポイントURLを指定するだけで、クライアントが自動的にgRPC接続も設定してくれます。

3. Embedded Weaviate への接続

アプリケーション内にWeaviateデータベースを直接組み込む方法です。小規模なプロジェクトやテスト環境で便利です。

import weaviate
import os

# 環境変数で永続化ディレクトリやバージョンを指定可能
# persistence_data_path = "./weaviate_data"
# version = "1.24.1" # 特定のバージョンを指定

client = weaviate.connect_to_embedded(
    # version=version,
    # persistence_data_path=persistence_data_path,
    headers={
        "X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY"),
    }
)

try:
    print(f"Client is ready: {client.is_ready()}")
    # Embedded Weaviateに対する操作
finally:
    client.close()

print("Connection closed.")

Embedded Weaviate は、初回実行時にDockerイメージをダウンロードするため、少し時間がかかる場合があります。

4. 直接的な接続 (高度な設定)

ヘルパー関数を使わず、より詳細な接続パラメータを指定したい場合は、`WeaviateClient` クラスを直接インスタンス化します。

import weaviate
from weaviate.connect import ConnectionParams
from weaviate.auth import AuthApiKey
from weaviate.classes.init import AdditionalConfig, Timeout
import os

client = weaviate.WeaviateClient(
    connection_params=ConnectionParams.from_params(
        http_host="localhost",
        http_port=8080,
        http_secure=False,
        grpc_host="localhost",
        grpc_port=50051,
        grpc_secure=False,
    ),
    auth_client_secret=AuthApiKey(api_key="YOUR-WEAVIATE-API-KEY"), # 認証が必要な場合
    additional_headers={
        "X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")
    },
    additional_config=AdditionalConfig(
        timeout=Timeout(init=30, query=60, insert=120), # タイムアウト設定 (秒)
        trust_env=False, # 環境変数からのSSL証明書読み込みを制御
    ),
    skip_init_checks=False # 起動時の接続チェックをスキップするか
)

try:
    # 直接インスタンス化した場合、手動でconnect()を呼び出す必要がある
    client.connect()
    print(f"Client is ready: {client.is_ready()}")
    # 操作...
finally:
    client.close()

print("Connection closed.")

Weaviateにデータを格納する前に、「コレクション (Collection)」（旧バージョンではクラス (Class)）のスキーマを定義する必要があります。スキーマは、格納するデータの構造（プロパティとそのデータ型）、使用するベクトル化モジュール（Vectorizer）、インデックス設定などを指定します。

v4クライアントでは、スキーマ定義がよりPythonicになり、型安全性が向上しました。`weaviate.classes.config` モジュール以下のクラスを使って定義します。

import weaviate
import weaviate.classes as wvc # weaviate.classesのエイリアスとしてwvcを使用
import os

client = weaviate.connect_to_local(
    headers={
        "X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY"),
    }
)

try:
    # コレクション名の指定
    collection_name = "MyNewCollection"

    # 既存のコレクションがあれば削除 (開発・テスト用)
    if client.collections.exists(collection_name):
        print(f"Deleting existing collection: {collection_name}")
        client.collections.delete(collection_name)

    print(f"Creating collection: {collection_name}")

    # コレクションの作成
    my_collection = client.collections.create(
        name=collection_name,
        # Vectorizerの設定: ここではOpenAIのtext-embedding-ada-002を使用
        vectorizer_config=wvc.config.Configure.Vectorizer.text2vec_openai(),
        # 生成モジュール(RAG用)の設定: ここではOpenAI (GPT-3.5 Turbo)を使用
        generative_config=wvc.config.Configure.Generative.openai(),
        # プロパティの定義
        properties=[
            wvc.config.Property(
                name="title",
                data_type=wvc.config.DataType.TEXT,
                description="記事のタイトル",
                # このプロパティをベクトル化に含めるか (デフォルト: True)
                # skip_vectorization=False,
                # このプロパティにベクトルインデックスを作成するか (デフォルト: True)
                # index_filterable=True,
                # index_searchable=True,
            ),
            wvc.config.Property(
                name="content",
                data_type=wvc.config.DataType.TEXT,
                description="記事の本文",
            ),
            wvc.config.Property(
                name="word_count",
                data_type=wvc.config.DataType.INT,
                description="記事の単語数",
                skip_vectorization=True, # 単語数はベクトル化しない
                index_filterable=True, # フィルタリング可能にする
                index_searchable=False, # キーワード検索対象外にする
            ),
            wvc.config.Property(
                name="publish_date",
                data_type=wvc.config.DataType.DATE,
                description="公開日",
            ),
        ],
        # 他のオプション:
        # vector_index_config=wvc.config.Configure.VectorIndex.hnsw( ... ), # ANNインデックスの設定
        # sharding_config=wvc.config.Configure.Sharding.multi_tenant( ... ), #シャーディング設定
        # replication_config=wvc.config.Configure.Replication.simple(factor=3), # レプリケーション設定
    )

    print(f"Collection '{collection_name}' created successfully.")

    # 作成されたコレクションのスキーマ情報を取得
    schema = client.collections.get(collection_name).config.get()
    print("\nCollection Schema:")
    print(schema)


finally:
    client.close()

上記の例では、`MyNewCollection` という名前のコレクションを作成しています。

• `vectorizer_config` で `text2vec_openai` を指定し、OpenAIのモデルを使ってテキストプロパティ (`title`, `content`) からベクトルを自動生成するように設定しています。APIキーは接続時の `headers` で渡します。
• `generative_config` で `openai` を指定し、後述する生成検索（RAG）でOpenAIのモデルを利用できるようにしています。
• `properties` で、格納するデータの各フィールド（プロパティ）の名前、データ型 (`TEXT`, `INT`, `DATE` など）、説明、ベクトル化やインデックス作成に関する設定を行っています。
• `skip_vectorization=True` とすると、そのプロパティはベクトル計算の対象外となります。
• `index_filterable=True` とすると、そのプロパティを後述するフィルタリング（`where`句）の条件として使用できます。
• `index_searchable=True` とすると、そのプロパティをキーワード検索（BM25F）の対象に含めることができます。

データ型には `TEXT`, `INT`, `NUMBER` (浮動小数点), `DATE`, `BOOL`, `UUID`, `GEO_COORDINATES`, `PHONE_NUMBER`, `BLOB`, そして他のコレクションへの参照を表す `CROSS_REFERENCE` などがあります。 Vectorizerには `text2vec-openai`, `text2vec-cohere`, `text2vec-huggingface`, `multi2vec-clip` (画像+テキスト), `img2vec-neural` など、様々なモデルやモジュールが用意されています。`none` を指定すれば、ベクトル化をWeaviate側で行わず、自前で計算したベクトルを登録することも可能です。

スキーマを定義したら、データの挿入（Create）、取得（Read）、更新（Update）、削除（Delete）といった操作を行います。v4クライアントでは、コレクションオブジェクト（`client.collections.get(“CollectionName”)` で取得）を通してこれらの操作を実行します。

1. データ挿入 (Insert)

単一のオブジェクトを挿入するには、コレクションオブジェクトの `data.insert()` メソッドを使用します。

import weaviate
import weaviate.classes as wvc
import os
import uuid
from datetime import datetime, timezone

client = weaviate.connect_to_local(
    headers={"X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")}
)
collection_name = "MyNewCollection"

try:
    my_collection = client.collections.get(collection_name)

    # 挿入するデータ (辞書形式)
    data_properties = {
        "title": "Weaviate Python Client v4 解説",
        "content": "新しいv4クライアントはgRPCを採用し、大幅に高速化されました...",
        "word_count": 550,
        "publish_date": datetime.now(timezone.utc) # UTCで日時を指定
    }

    # 単一オブジェクトの挿入
    inserted_uuid = my_collection.data.insert(
        properties=data_properties,
        # uuid=uuid.uuid4() # UUIDを自分で指定することも可能
        # vector=[0.1, 0.2, ...] # vectorizer=Noneの場合、ベクトルを直接指定
    )

    print(f"Object inserted with UUID: {inserted_uuid}")

    # 複数オブジェクトを一括挿入 (insert_many)
    data_objects_to_insert = [
        wvc.data.DataObject(
            properties={
                "title": "ベクトル検索の基本",
                "content": "ベクトル検索は意味的な類似性に基づいてデータを検索します。",
                "word_count": 300,
                "publish_date": datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
            },
            # uuid=uuid.uuid5(uuid.NAMESPACE_DNS, "vector-search-basics") # UUIDを指定
        ),
        wvc.data.DataObject(
            properties={
                "title": "RAGとは何か？",
                "content": "Retrieval-Augmented Generationは、検索と生成を組み合わせた技術です。",
                "word_count": 420,
                "publish_date": datetime(2024, 2, 15, 14, 30, 0, tzinfo=timezone.utc)
            }
            # vector=... # vectorizer=noneの場合
        )
    ]

    insert_result = my_collection.data.insert_many(data_objects_to_insert)

    print("\nMultiple objects inserted:")
    if insert_result.has_errors:
        print(f"  Errors occurred: {insert_result.errors}")
    else:
        print(f"  Successfully inserted {len(insert_result.all_responses)} objects.")
        for i, resp in enumerate(insert_result.all_responses):
             print(f"  - Object {i+1} UUID: {resp}")


finally:
    client.close()

`data.insert()` は挿入されたオブジェクトのUUIDを返します。`data.insert_many()` は複数の `DataObject` をリストで受け取り、一括で挿入します。`DataObject` を使うことで、プロパティだけでなくUUIDやベクトルも個別に指定できます。`insert_many` の戻り値 (`BatchObjectReturn`) で、エラーの有無や各オブジェクトの挿入結果を確認できます。

バッチ挿入 (Batching)

大量のデータを効率的に挿入するには、バッチ処理が不可欠です。v4クライアントでは、より柔軟なバッチ設定が可能になりました。コレクションオブジェクトの `batch.dynamic()` や `batch.fixed_size()`, `batch.rate_limit()` をコンテキストマネージャ (`with`) と組み合わせて使用します。

import weaviate
import weaviate.classes as wvc
import os
import time

client = weaviate.connect_to_local(
    headers={"X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")}
)
collection_name = "MyNewCollection"

try:
    my_collection = client.collections.get(collection_name)

    # 大量のサンプルデータ (実際にはファイルから読み込むなど)
    articles_data = [
        {
            "title": f"Article Title {i}",
            "content": f"This is the content for article {i}. " * 10,
            "word_count": 100 + i,
            "publish_date": datetime(2024, 3, i % 30 + 1, tzinfo=timezone.utc)
        } for i in range(1, 101) # 100件のデータ
    ]

    start_time = time.time()
    print("\nStarting batch import...")

    # Dynamic Batching: Weaviateがバッチサイズを自動調整 (推奨)
    with my_collection.batch.dynamic() as batch:
        for data in articles_data:
            batch.add_object(
                properties=data
                # uuid=...
                # vector=...
            )
        # コンテキストマネージャを抜ける際に自動的にフラッシュされる

    # Fixed Size Batching: バッチサイズを固定 (例: 10件ごと)
    # with my_collection.batch.fixed_size(batch_size=10) as batch:
    #     for data in articles_data:
    #         batch.add_object(properties=data)

    # Rate Limit Batching: 1分あたりのオブジェクト数を制限 (例: 1分あたり50件)
    # 外部APIのレート制限対策に有効
    # with my_collection.batch.rate_limit(requests_per_minute=50) as batch:
    #     for data in articles_data:
    #         batch.add_object(properties=data)

    end_time = time.time()
    print(f"Batch import finished in {end_time - start_time:.2f} seconds.")

    # バッチ処理結果の確認 (エラーハンドリング)
    if batch.number_errors > 0:
        print(f"Errors occurred during batch import: {batch.errors}")

finally:
    client.close()

`batch.dynamic()` は、Weaviateサーバーがネットワーク状況などを考慮して最適なバッチサイズを動的に決定するため、多くの場合で推奨されます。 `batch.fixed_size()` は、指定した件数ごとにデータを送信します。 `batch.rate_limit()` は、外部API（特にVectorizerとして利用する場合）のレート制限を超えないように、1分あたりのリクエスト数を制御したい場合に便利です。 バッチ処理中に発生したエラーは `batch.number_errors` や `batch.errors` で確認できます。

2. データ取得 (Read/Fetch)

UUIDを指定して特定のオブジェクトを取得するには `data.fetch_object_by_id()` を使用します。複数のUUIDを指定する場合は `data.fetch_objects()` を使います。

import weaviate
import weaviate.classes as wvc
import os

client = weaviate.connect_to_local()
collection_name = "MyNewCollection"

try:
    my_collection = client.collections.get(collection_name)

    # 挿入時に取得したUUID、または検索で見つけたUUIDを使う
    target_uuid = inserted_uuid # 上の挿入例で取得したUUID

    # UUIDを指定して単一オブジェクトを取得
    fetched_object = my_collection.data.fetch_object_by_id(uuid=target_uuid)

    if fetched_object:
        print("\nFetched Object by UUID:")
        print(f"  UUID: {fetched_object.uuid}")
        print(f"  Properties: {fetched_object.properties}")
        # print(f"  Vector: {fetched_object.vector}") # ベクトルも取得可能 (別途指定が必要な場合あり)
    else:
        print(f"\nObject with UUID {target_uuid} not found.")

    # 複数のUUIDで取得
    uuids_to_fetch = [inserted_uuid, insert_result.all_responses[0]] # insert_manyの結果など
    fetched_objects = my_collection.data.fetch_objects(limit=len(uuids_to_fetch)) # limitで最大数を指定

    print("\nFetched Multiple Objects by UUIDs:")
    for obj in fetched_objects.objects:
        print(f"  UUID: {obj.uuid}, Title: {obj.properties.get('title')}")


finally:
    client.close()

検索については、次のセクションで詳しく説明します。

3. データ更新 (Update)

既存のオブジェクトのプロパティを更新するには `data.update()` を使用します。更新したいプロパティのみを指定します。

import weaviate
import weaviate.classes as wvc
import os

client = weaviate.connect_to_local()
collection_name = "MyNewCollection"
target_uuid = inserted_uuid # 更新対象のUUID

try:
    my_collection = client.collections.get(collection_name)

    # 更新するプロパティ
    updated_properties = {
        "title": "Weaviate Python Client v4 完全解説 (更新版)",
        "word_count": 600
    }

    # オブジェクトの更新
    my_collection.data.update(
        uuid=target_uuid,
        properties=updated_properties,
        # vector=[...] # ベクトルを更新する場合
    )

    print(f"\nObject {target_uuid} updated.")

    # 更新後のオブジェクトを確認
    updated_object = my_collection.data.fetch_object_by_id(uuid=target_uuid)
    if updated_object:
        print("Updated Properties:")
        print(updated_object.properties)

finally:
    client.close()

`data.replace()` を使うと、指定したプロパティ以外は削除され、指定したプロパティで完全に置き換えられます。

4. データ削除 (Delete)

UUIDを指定して単一のオブジェクトを削除するには `data.delete_by_id()` を使用します。複数のオブジェクトを条件で削除する場合は `data.delete_many()` を使用します（フィルタリングの知識が必要）。

import weaviate
import weaviate.classes as wvc
import os

client = weaviate.connect_to_local()
collection_name = "MyNewCollection"
uuid_to_delete = insert_result.all_responses[1] # insert_manyで追加した2番目のオブジェクトのUUID

try:
    my_collection = client.collections.get(collection_name)

    # UUIDを指定して単一オブジェクトを削除
    my_collection.data.delete_by_id(uuid=uuid_to_delete)
    print(f"\nObject {uuid_to_delete} deleted.")

    # 削除されたか確認 (fetch_object_by_id は None を返すはず)
    deleted_obj_check = my_collection.data.fetch_object_by_id(uuid=uuid_to_delete)
    if not deleted_obj_check:
        print(f"Confirmed: Object {uuid_to_delete} no longer exists.")

    # 条件に一致する複数オブジェクトを削除 (例: word_countが150未満のものを削除)
    # delete_many には Filter が必要
    delete_filter = wvc.query.Filter.by_property("word_count").less_than(150)

    delete_result = my_collection.data.delete_many(
        where=delete_filter
    )

    print(f"\nDeleted {delete_result.successful} objects matching the filter.")
    print(f"Failed to delete {delete_result.failed} objects.")
    if delete_result.failed > 0:
        print(f"Errors: {delete_result.errors}")

finally:
    client.close()

`delete_many()` では `where` パラメータにフィルタ条件を指定します。フィルタリングについては次の検索セクションで詳しく触れます。

Weaviateの真価は、その強力な検索機能にあります。Pythonクライアント (v4) では、コレクションオブジェクトの `query` サブモジュールを使って様々な種類の検索を実行できます。v4ではクエリAPIも刷新され、より直感的で表現力豊かになりました。

1. ベクトル検索 (Vector Search / Semantic Search)

データの内容や意味に基づいて類似したオブジェクトを検索します。テキストによる検索 (`near_text`) や、ベクトルそのものによる検索 (`near_vector`) が可能です。

import weaviate
import weaviate.classes as wvc
import os

client = weaviate.connect_to_local(
    headers={"X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")}
)
collection_name = "MyNewCollection"

try:
    my_collection = client.collections.get(collection_name)

    # 検索クエリテキスト
    query_text = "機械学習モデルのスケーリングについて"

    print(f"\nPerforming near_text search for: '{query_text}'")

    # near_text検索の実行
    response = my_collection.query.near_text(
        query=query_text,
        limit=3, # 返却する結果の最大数
        # certainty=0.7, # 類似度の閾値 (0から1)。distanceと排他的。
        distance=0.3, # 距離の閾値 (小さいほど類似)。certaintyと排他的。
        # return_metadata=wvc.query.MetadataQuery(uuid=True, distance=True, certainty=True), # UUIDや距離/類似度をメタデータとして取得
        return_properties=["title", "publish_date", "word_count"] # 返却するプロパティを指定 (指定しない場合は全プロパティ)
    )

    print("\nNear Text Search Results:")
    if not response.objects:
        print("  No results found.")
    else:
        for obj in response.objects:
            print(f"  UUID: {obj.uuid}")
            print(f"  Title: {obj.properties['title']}")
            print(f"  Publish Date: {obj.properties['publish_date']}")
            print(f"  Word Count: {obj.properties['word_count']}")
            # メタデータを要求した場合
            if obj.metadata:
                print(f"  Distance: {obj.metadata.distance:.4f}")
                # print(f"  Certainty: {obj.metadata.certainty:.4f}")
            print("-" * 10)

    # near_vector検索 (ベクトルが分かっている場合)
    # target_vector = [0.1, 0.2, ..., 0.9] # 検索に使用するベクトル
    # response_vec = my_collection.query.near_vector(
    #     near_vector=target_vector,
    #     limit=3,
    #     distance=0.3,
    #     return_metadata=wvc.query.MetadataQuery(distance=True),
    #     return_properties=["title"]
    # )
    # print("\nNear Vector Search Results:")
    # for obj in response_vec.objects:
    #     print(f"  Title: {obj.properties['title']}, Distance: {obj.metadata.distance:.4f}")


finally:
    client.close()

`near_text` は、指定された `query` テキストを、コレクション作成時に設定したVectorizer（この例では `text2vec-openai`）を使ってベクトル化し、そのベクトルに近いオブジェクトを探します。 `limit` で取得件数を制限し、`certainty` または `distance` で類似度の閾値を設定できます（両方は指定できません）。`certainty` は類似度（1に近いほど類似）、`distance` はベクトル間の距離（0に近いほど類似）を表します。どちらを使うかは、使用するベクトルインデックスの種類や好みによります。 `return_metadata` で、UUID、距離、類似度などのメタデータを取得できます。 `return_properties` で取得するプロパティを絞り込むと、レスポンスサイズを小さくできます。

2. ハイブリッド検索 (Hybrid Search)

ベクトル検索（意味的な類似性）と従来のキーワード検索（BM25Fアルゴリズムによるテキストマッチング）を組み合わせて、両方の長所を活かした検索を行います。

import weaviate
import weaviate.classes as wvc
import os

client = weaviate.connect_to_local(
    headers={"X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")}
)
collection_name = "MyNewCollection"

try:
    my_collection = client.collections.get(collection_name)

    # 検索クエリテキスト
    query_text = "python クライアント"
    # キーワード検索用のクエリ (ベクトル検索と同じでも良い)
    keyword_query = "python client"

    print(f"\nPerforming hybrid search for: '{query_text}' (keyword: '{keyword_query}')")

    # hybrid検索の実行
    response = my_collection.query.hybrid(
        query=query_text, # ベクトル検索用のクエリ
        query_properties=["title", "content"], # キーワード検索の対象プロパティ (指定しなければ全TEXTプロパティ)
        # alpha=0.75, # ベクトル検索の重み (0から1、デフォルト: 0.5)。1で純粋なベクトル検索、0で純粋なキーワード検索。
        limit=5,
        return_metadata=wvc.query.MetadataQuery(score=True), # ハイブリッド検索では score (統合スコア) が返される
        return_properties=["title", "word_count"]
    )

    print("\nHybrid Search Results:")
    if not response.objects:
        print("  No results found.")
    else:
        for obj in response.objects:
            print(f"  UUID: {obj.uuid}")
            print(f"  Title: {obj.properties['title']}")
            print(f"  Word Count: {obj.properties['word_count']}")
            if obj.metadata:
                print(f"  Score: {obj.metadata.score:.4f}") # 統合スコア
            print("-" * 10)

finally:
    client.close()

`hybrid` メソッドを使用します。`query` はベクトル検索部分に使われ、`query_properties` で指定されたプロパティに対してキーワード検索（BM25F）が行われます。 `alpha` パラメータでベクトル検索とキーワード検索の重み付けを調整できます。`alpha=0.5` がデフォルトで、両者を均等に扱います。`alpha=1.0` ならベクトル検索のみ、`alpha=0.0` ならキーワード検索のみと同じ結果になります。 メタデータとして `score` を要求すると、ベクトル検索とキーワード検索の結果を統合したスコアが返されます。

3. 生成検索 (Generative Search / RAG) 🤖

検索結果を大規模言語モデル（LLM）に入力し、質問に対する回答や要約などを生成させる機能です。Retrieval-Augmented Generation (RAG) とも呼ばれます。コレクション作成時に `generative_config` で生成モデル（例: OpenAI, Cohere, PaLM など）を指定しておく必要があります。

import weaviate
import weaviate.classes as wvc
import os

client = weaviate.connect_to_local(
    headers={"X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")}
)
collection_name = "MyNewCollection" # generative_configでOpenAIが設定されている前提

try:
    my_collection = client.collections.get(collection_name)

    # 生成検索の質問
    question = "Weaviate Pythonクライアントv4の主な変更点は何ですか？3つ挙げてください。"

    print(f"\nPerforming generative search (single prompt) for: '{question}'")

    # 単一プロンプトによる生成検索 (検索結果全体から回答を生成)
    response_single = my_collection.generate.near_text(
        query=question, # 検索のクエリとしても使用
        limit=5, # 回答生成の元にする検索結果の数
        single_prompt=f"以下の検索結果に基づいて、質問に答えてください。質問: {question}\n\n検索結果:\n{{result_properties}}", # {result_properties}が検索結果で置換される
        # return_metadata=wvc.query.MetadataQuery(uuid=True)
    )

    print("\nGenerative Search (Single Prompt) Result:")
    if response_single.generated:
        print(f"  Generated Answer:\n{response_single.generated}")
    else:
        print("  Could not generate an answer.")
    # print("\nSource Objects:")
    # for obj in response_single.objects:
    #     print(f"  - UUID: {obj.uuid}, Title: {obj.properties['title']}")


    # グループ化プロンプトによる生成検索 (各検索結果ごとに要約などを生成)
    print(f"\nPerforming generative search (grouped task) for: '{question}'")
    response_grouped = my_collection.generate.near_text(
        query=question,
        limit=3,
        grouped_task="各記事の要点を30文字程度でまとめてください。", # 各結果に適用するタスク
        # grouped_properties=["title", "content"], # 要約に使用するプロパティを指定 (任意)
        return_metadata=wvc.query.MetadataQuery(uuid=True)
    )

    print("\nGenerative Search (Grouped Task) Results:")
    if not response_grouped.objects:
        print("  No results found.")
    else:
        for obj in response_grouped.objects:
            print(f"  UUID: {obj.uuid}")
            print(f"  Title: {obj.properties['title']}")
            if obj.generated:
                print(f"  Generated Summary: {obj.generated}")
            else:
                print("  Could not generate summary for this object.")
            print("-" * 10)


finally:
    client.close()

生成検索は `generate` サブモジュール（例: `my_collection.generate.near_text(…)`）を使用します。これは、内部的にはまず指定された検索（この例では `near_text`）を実行し、その結果をLLMに渡して処理します。

• `single_prompt`: 検索結果全体をコンテキストとして、1つの回答を生成します。プロンプト文字列内に `{result_properties}` のようなプレースホルダーを含めることで、検索結果の情報を埋め込むことができます。質問応答システムなどに適しています。
• `grouped_task`: 各検索結果オブジェクトに対して個別にタスク（例: 要約、翻訳など）を実行します。結果オブジェクトの `generated` プロパティに各々の生成結果が格納されます。

生成検索はLLMへのAPI呼び出しを伴うため、通常の検索よりも時間がかかる場合があります。必要に応じて接続時のタイムアウト設定 (`Timeout(query=…)`) を調整してください。

4. フィルタリング (Filtering)

検索結果を特定のプロパティの値に基づいて絞り込む機能です。SQLの `WHERE` 句に相当します。`wvc.query.Filter` クラスを使って条件を作成し、各種検索メソッドの `filters` パラメータに渡します。

import weaviate
import weaviate.classes as wvc
import os
from datetime import datetime, timezone

client = weaviate.connect_to_local()
collection_name = "MyNewCollection"

try:
    my_collection = client.collections.get(collection_name)

    # フィルタ条件の作成
    # 例1: word_count が 500 以上
    filter1 = wvc.query.Filter.by_property("word_count").greater_or_equal(500)

    # 例2: 公開日が 2024年3月1日以降
    filter2 = wvc.query.Filter.by_property("publish_date").greater_or_equal(
        datetime(2024, 3, 1, tzinfo=timezone.utc)
    )

    # 例3: title に "Python" を含む (部分一致)
    filter3 = wvc.query.Filter.by_property("title").like("*Python*") # ワイルドカードを使用

    # フィルタ条件の組み合わせ (AND / OR)
    # word_count >= 500 AND publish_date >= 2024-03-01
    combined_filter_and = filter1 & filter2
    # combined_filter_and = wvc.query.Filter.all_of([filter1, filter2]) # これでも同じ

    # word_count >= 500 OR title contains "Python"
    combined_filter_or = filter1 | filter3
    # combined_filter_or = wvc.query.Filter.any_of([filter1, filter3]) # これでも同じ

    print("\nPerforming near_text search with filter (word_count >= 500)")

    # フィルタを適用したベクトル検索
    response = my_collection.query.near_text(
        query="Weaviate クライアント",
        limit=5,
        filters=filter1, # 作成したフィルタを適用
        return_properties=["title", "word_count"]
    )

    print("\nFiltered Search Results:")
    if not response.objects:
        print("  No results found matching the filter.")
    else:
        for obj in response.objects:
            print(f"  Title: {obj.properties['title']}, Word Count: {obj.properties['word_count']}")

finally:
    client.close()

`Filter.by_property(“プロパティ名”)` で対象プロパティを指定し、`.equals()`, `.not_equal()`, `.greater_than()`, `.less_or_equal()`, `.like()`（ワイルドカード `*`, `?` 使用可能）, `.is_none()`, `.contains_any([…])`, `.contains_all([…])` などのメソッドで条件を指定します。 複数のフィルタは `&` (AND) や `|` (OR) 演算子、または `Filter.all_of([…])`, `Filter.any_of([…])` で組み合わせることができます。 フィルタリングは、ベクトル検索、ハイブリッド検索、BM25検索など、多くの検索メソッドで利用可能です。スキーマ定義時に `index_filterable=True` が設定されているプロパティのみフィルタリングに使用できます。

これら以外にも、BM25F（キーワード検索）、オブジェクトIDによる検索、クロスリファレンス（オブジェクト間の関連）を使った検索、集計（Aggregation）など、多様なクエリが可能です。詳細は公式ドキュメントを参照してください。

Weaviate Pythonクライアントは、基本的なCRUD操作や検索以外にも、様々な高度な機能を提供しています。

• 非同期処理 (Asyncio): `WeaviateAsyncClient` クラスや `weaviate.use_async_…` ヘルパー関数を使用することで、`asyncio` を利用した非同期操作が可能です (v4.7.0以降)。これにより、特にI/Oバウンドな操作（ネットワーク通信が多い処理）において、アプリケーション全体のパフォーマンスを向上させることができます。

  import weaviate
  import asyncio
  import os
  
  async def main():
      async_client = await weaviate.use_async_with_local(
          headers={"X-OpenAI-Api-Key": os.getenv("OPENAI_APIKEY")}
      )
      try:
          collection_name = "MyNewCollection"
          async_collection = async_client.collections.get(collection_name)
  
          response = await async_collection.query.near_text(
              query="非同期処理",
              limit=2
          )
          print("Async search results:")
          for obj in response.objects:
              print(f"  - {obj.properties['title']}")
  
      finally:
          await async_client.close()
  
  # asyncio.run(main()) # イベントループで実行

• バックアップとリストア: データのバックアップを作成し、必要に応じてリストアする機能が提供されています。`client.backup.create()` や `client.backup.restore()` を使用します。バックエンドとしてS3互換ストレージやローカルファイルシステムなどを指定できます。
• マルチテナンシー: 1つのWeaviateクラスター内で、複数のテナント（顧客やプロジェクトなど）ごとにデータを分離して管理する機能です。スキーマ定義時に `sharding_config` で設定し、データ操作や検索時にテナント名を指定します。
• 認証と認可: APIキー認証やOIDC（OpenID Connect）による認証連携、ロールベースのアクセス制御（RBAC）など、セキュリティ機能も充実しています。
• クロスリファレンス: オブジェクト間にリレーションシップ（関連）を設定し、それを辿るようなクエリを実行できます。グラフデータベース的な使い方も可能です。