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）など、セキュリティ機能も充実しています。
• クロスリファレンス: オブジェクト間にリレーションシップ（関連）を設定し、それを辿るようなクエリを実行できます。グラフデータベース的な使い方も可能です。