はじめに
こんにちは！masa-asaです。
突然ですが、以前にこのような記事を投稿しました。
内容を簡単に説明すると、AI Searchでベクトル化したデータを含むインデックスを作成し、検索できるような状態にするまでの手順というものでした。
この記事では、インデックスの作成・データソースの作成・スキルセットの作成・インデクサーの作成をAzure Portal上で行っていますが、これらと同等のことがコードで実行可能です。そこで今回はPythonを用いてこれらを実現してみよう！というモチベーションから記事を書こうと思います。よろしくお願いします！
※リソースの構成や作成などは以前の記事と同じなのでその部分は省略します。リソースもすでに作成されている状態で進めていきます。
事前準備
pythonで仮想環境を準備する
ご自身のワークスペースで以下のコマンドを実行します。
python -m venv venv
仮想環境をアクティベートします。
source ./venv/bin/activate
コンソールに(venv)と表示されていれば仮想環境に入れています。
次に、パッケージマネージャーであるpoetryをインストールします。以下のコマンドを実行します。
pip install poetry
poetryの準備ができました！今回必要なパッケージを以下のコマンドを実行して追加します。
poetry add azure-search-documents
poetry add azure-identity
poetry add python-dotenv
また、環境変数として次のような値を事前に取得して.envに記述しています。
- AI SearchのエンドポイントURL
- AI SearchのAPI Key
- Azure OpenAIのエンドポイントURL
- Azure OpenAIのAPI Key
- データソースに用いるストレージアカウントの接続文字列
インデックスの作成
インデックスの作成でフィールドの定義には2つの関数SimpleField、SearchableFieldとSearchFieldクラスを用いています。
SimpleField、SearchableFieldは基本的にフィールドと「検索可能」な基本的なフィールドを作成するのに役立つ関数です。
また、ベクトルを格納するフィールドにはSearchFieldを用いています。こちらはデータ型の設定など設定可能な幅が広く、柔軟な定義ができるため、このクラスを用いるように統一しても良いかもしれないです。
今回はベクトルを扱うため、ベクタープロファイルの定義もここで行っています。
ベクタープロファイルとそれに付随するベクタライザの作成と、text_vectorの定義の際にプロファイルのフィールドへのアタッチを記述しています。
インデックスの作成は以下のコードで実行します。インデックスの作成含めその他のコードは、参考情報にあるリンクのコードをベースに少しの改変を加えたものです。
# coding: utf-8
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
"""
FILE: sample_index_crud_operations.py
DESCRIPTION:
This sample demonstrates how to get, create, update, or delete an index.
USAGE:
python sample_index_crud_operations.py
Set the environment variables with your own values before running the sample:
1) AZURE_SEARCH_SERVICE_ENDPOINT - the endpoint of your Azure Cognitive Search service
2) AZURE_SEARCH_API_KEY - your search API key
"""
from azure.core.credentials import AzureKeyCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
CorsOptions,
SearchIndex,
SearchFieldDataType,
SimpleField,
SearchableField,
HnswAlgorithmConfiguration,
VectorSearch,
VectorSearchProfile,
AzureOpenAIVectorizer,
AzureOpenAIVectorizerParameters,
SearchField
)
from config import index_name, vectorizer_name, vector_profile_name
import os
from dotenv import load_dotenv
load_dotenv(override=True)
service_endpoint = os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT")
key = os.getenv("AZURE_SEARCH_API_KEY")
oai_account = os.getenv("AZURE_OPENAI_ACCOUNT")
oai_key = os.getenv("AZURE_OPENAI_KEY")
client = SearchIndexClient(service_endpoint, AzureKeyCredential(key))
def create_index():
# [START create_index]
name = index_name
fields = [
SearchableField(name="id", key=True, sortable=True, analyzer_name = "keyword"),
SimpleField(name="parent_id", type=SearchFieldDataType.String ,filterable=True, sortable=True),
SearchableField(name="chunk", searchable=True, filterable=True, sortable=True, facetable=True, analyzer_name = "standard.lucene"),
SearchableField(name="file_name", searchable=True, filterable=True, sortable=True, facetable=True, analyzer_name = "standard.lucene"),
SearchField(name="text_vector", type=SearchFieldDataType.Collection(SearchFieldDataType.Single), hidden= False, vector_search_dimensions=1536, vector_search_profile_name=vector_profile_name),
]
# Configure the vector search configuration
vector_search = VectorSearch(
algorithms=[
HnswAlgorithmConfiguration(name="myHnsw"),
],
profiles=[
VectorSearchProfile(
name=vector_profile_name,
algorithm_configuration_name="myHnsw",
vectorizer_name=vectorizer_name,
)
],
vectorizers=[
AzureOpenAIVectorizer(
vectorizer_name=vectorizer_name,
kind="azureOpenAI",
parameters=AzureOpenAIVectorizerParameters(
resource_url=oai_account,
api_key=oai_key,
deployment_name="text-embedding-3-small",
model_name="text-embedding-3-small",
),
),
],
)
cors_options = CorsOptions(allowed_origins=["*"], max_age_in_seconds=60)
scoring_profiles = []
index = SearchIndex(
name=name,
fields=fields,
scoring_profiles=scoring_profiles,
cors_options=cors_options,
vector_search=vector_search
)
result = client.create_index(index)
print("Create Index Result: {}".format(result))
# [END create_index]
if __name__ == '__main__':
create_index()
完成したインデックスフィールドは図のようになります。
データソースの作成
BLOBコンテナー内の特定のフォルダをデータソースにしたい場合は、SearchIndexerDataContainerの引数にクエリを書くことで実現できます。今回はBLOBなのでtype="azureblob"を指定しています。
以下がコードです。
from azure.search.documents.indexes import SearchIndexerClient
from azure.search.documents.indexes.models import (
SearchIndexerDataContainer,
SearchIndexerDataSourceConnection
)
from azure.core.credentials import AzureKeyCredential
from config import data_source_name
import os
from dotenv import load_dotenv
load_dotenv(override=True)
service_endpoint = os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT")
key = os.getenv("AZURE_SEARCH_API_KEY")
AZURE_STORAGE_CONNECTION = os.getenv("AZURE_STORAGE_CONNECTION")
# Create a data source
indexer_client = SearchIndexerClient(endpoint=service_endpoint, credential=AzureKeyCredential(key))
container = SearchIndexerDataContainer(name="srchcontainer", query="pdfs")
data_source_connection = SearchIndexerDataSourceConnection(
name=data_source_name,
type="azureblob",
connection_string=AZURE_STORAGE_CONNECTION,
container=container
)
data_source = indexer_client.create_or_update_data_source_connection(data_source_connection)
print(f"Data source '{data_source.name}' created or updated")
スキルセットの作成
スキルセット作成で分かりにくそうな部分について説明します。
前の記事で、スキルセットのcognitiveServicesはこのように設定しました。
"cognitiveServices": {
"@odata.type": "#Microsoft.Azure.Search.DefaultCognitiveServices"
}
pythonコードで同じように設定するには、cognitive_services_accountにDefaultCognitiveServicesAccount()を指定してあげます。
ここで指定されるCognitive Services(今のAI Services)は自分で作成したリソース ではなく、Azureが内部で管理するCognitive Servicesですので、注意が必要です。開発者から見えるものではありません。
実際のpythonコードは以下です。
from azure.search.documents.indexes.models import (
SplitSkill,
InputFieldMappingEntry,
OutputFieldMappingEntry,
AzureOpenAIEmbeddingSkill,
SearchIndexerIndexProjection,
SearchIndexerIndexProjectionSelector,
SearchIndexerIndexProjectionsParameters,
IndexProjectionMode,
SearchIndexerSkillset,
DefaultCognitiveServicesAccount
)
from config import skillset_name, index_name
from azure.search.documents.indexes import SearchIndexerClient
from azure.core.credentials import AzureKeyCredential
import os
from dotenv import load_dotenv
load_dotenv(override=True)
AZURE_OPENAI_ACCOUNT = os.getenv("AZURE_OPENAI_ACCOUNT")
AZURE_SEARCH_SERVICE_ENDPOINT = os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT")
key = os.getenv("AZURE_SEARCH_API_KEY")
oai_key = os.getenv("AZURE_OPENAI_KEY")
# Create a skillset
split_skill = SplitSkill(
description="Split skill to chunk documents",
text_split_mode="pages",
context="/document",
maximum_page_length=500,
page_overlap_length=100,
inputs=[
InputFieldMappingEntry(name="text", source="/document/content"),
],
outputs=[
OutputFieldMappingEntry(name="textItems", target_name="pages")
],
)
embedding_skill = AzureOpenAIEmbeddingSkill(
description="Skill to generate embeddings via Azure OpenAI",
context="/document/pages/*",
resource_url=AZURE_OPENAI_ACCOUNT,
deployment_name="text-embedding-3-small",
model_name="text-embedding-3-small",
dimensions=1536,
inputs=[
InputFieldMappingEntry(name="text", source="/document/pages/*"),
],
outputs=[
OutputFieldMappingEntry(name="embedding", target_name="text_vector")
],
)
index_projections = SearchIndexerIndexProjection(
selectors=[
SearchIndexerIndexProjectionSelector(
target_index_name=index_name,
parent_key_field_name="parent_id",
source_context="/document/pages/*",
mappings=[
InputFieldMappingEntry(name="chunk", source="/document/pages/*"),
InputFieldMappingEntry(name="text_vector", source="/document/pages/*/text_vector"),
InputFieldMappingEntry(name="file_name", source="/document/metadata_storage_name"),
],
),
],
parameters=SearchIndexerIndexProjectionsParameters(
projection_mode=IndexProjectionMode.SKIP_INDEXING_PARENT_DOCUMENTS
),
)
cognitive_services_account = DefaultCognitiveServicesAccount()
skills = [split_skill, embedding_skill]
skillset = SearchIndexerSkillset(
name=skillset_name,
description="Skillset to chunk documents and generating embeddings",
skills=skills,
index_projection=index_projections,
cognitive_services_account=cognitive_services_account
)
client = SearchIndexerClient(endpoint=AZURE_SEARCH_SERVICE_ENDPOINT, credential=AzureKeyCredential(key))
client.create_or_update_skillset(skillset)
print(f"{skillset.name} created")
インデクサーの作成
インデクサーを作成するコードは以下です。このコードを実行すると、インデクサーの作成&実行が行われます。
スキルセット、ターゲットのインデックス、データソースを指定します。
from azure.search.documents.indexes.models import (
SearchIndexer
)
from azure.search.documents.indexes import SearchIndexerClient
from azure.core.credentials import AzureKeyCredential
from config import skillset_name, index_name, data_source_name, indexer_name
import os
from dotenv import load_dotenv
load_dotenv(override=True)
service_endpoint = os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT")
key = os.getenv("AZURE_SEARCH_API_KEY")
# Create an indexer
indexer_parameters = None
indexer = SearchIndexer(
name=indexer_name,
description="Indexer to index documents and generate embeddings",
skillset_name=skillset_name,
target_index_name=index_name,
data_source_name=data_source_name,
parameters=indexer_parameters
)
# Create and run the indexer
indexer_client = SearchIndexerClient(endpoint=service_endpoint, credential=AzureKeyCredential(key))
indexer_result = indexer_client.create_or_update_indexer(indexer)
print(f' {indexer_name} is created and running. Give the indexer a few minutes before running a query.')
インデックスの確認
作成されたインデックスで「Ephemeral Lake」と検索してみた結果です。
スコアが1位の要素は正しいpdf（検索の文言が含まれているpdf）を検索できています。
まとめ
今回の記事で行ったことは次の通りです。
言語：python
- インデックス作成
- データソース作成
- スキルセット作成
- インデクサー作成
今回は作成のみでしたが、CRUDを実現することでAzure AI Search周りの設定などはコード化できそうなので、管理もしやすくなりそうです。
お読みいただきありがとうございました！
参考文献