Azure AI Servicesのモデルデプロイからデプロイ削除までのPythonのコードです。ファインチューニング済モデルに対して実行していますが、基本モデルに対してでも同じはずです(未確認)。
以下の2つのモチベーションで、Python Script作りました。
- ファインチューニング済モデルをデプロイすると、そこそこ課金されるためデプロイからデプロイ削除までを一連の流れで自動化したかった
- ファインチューニングを複数パターンでやってそのモデル評価をやるので、手作業を少なくしたかった
こちらの内容をベースに作成しています。
前提
Azureリソース
Azure AI Serviceがあれば問題ないかと。画面で確認するのにAzure AI Foundryを使っています。
やってから少し時間が経ってしまい記憶があいまいなのですが、デプロイのためにこちらを参考にロール割当追加しました。
あと、ファインチューニングが済んでいることも前提です。
Python実行環境
最近WSLの調子が悪いのでWindowsでやっています。
| 種類 | バージョン | 備考 |
|---|---|---|
| Python(Pyenv) | 3.13.1 | |
| OS | Windows11 24H2 | |
| Poetry | 2.0.1 |
Python 追加パッケージ
| 種類 | バージョン | 備考 |
|---|---|---|
| azure-ai-inference | 1.0.0b9 | |
| azure-identity | 1.21.0 | |
| azure-mgmt-cognitiveservices | 13.6.0 | デプロイで使用 |
| jupyterlab | 4.3.5 | Jupyterで実施 |
| openai | 1.63.2 | ファインチューニングジョブの情報取得に使用 |
| pandas | 2.2.3 | デプロイ後の推論で使用 |
| python-dotenv | 1.0.1 |
Python Script
0. 環境変数
python-dotenvパッケージで環境変数をロードするので、.envを作っておきます。設定値は省略します。
AZURE_CLIENT_ID、AZURE_TENANT_ID、AZURE_CLIENT_SECRETの設定はこちらを参考にしています。
.env
AZURE_OPENAI_TRAIN_API_KEY=
AZURE_OPENAI_TRAIN_ENDPOINT=
SUBSCRIPTION_ID=
RESOURCE_GROUP_NAME=
RESOURCE_NAME=
AZURE_CLIENT_ID=
AZURE_TENANT_ID=
AZURE_CLIENT_SECRET=
1. Python Packageインポート
import os
import re
from azure.ai.inference import ChatCompletionsClient
from azure.ai.inference.models import SystemMessage, UserMessage
from azure.core.credentials import AzureKeyCredential
from azure.identity import DefaultAzureCredential
from azure.mgmt.cognitiveservices import CognitiveServicesManagementClient
from openai import AzureOpenAI
from dotenv import load_dotenv
import pandas as pd
2. 環境変数の読込とjob_id入力
job_idはinput関数で入力しています。
入力値はAI Foundry 画面で、IDの値を入れます。
load_dotenv(override=True)
endpoint = os.getenv("AZURE_OPENAI_TRAIN_ENDPOINT")
key = os.getenv("AZURE_OPENAI_TRAIN_API_KEY")
subscription = os.getenv("SUBSCRIPTION_ID")
resource_group = os.getenv("RESOURCE_GROUP_NAME")
resource_name = os.getenv("RESOURCE_NAME")
job_id = input()
print(f"{job_id=}")
3. モデル名取得
モデル名を取得し、その値が冗長なので、不要部分を切り捨ててデプロイ名を作っています。
もともと、チェックポイントのファイルをローカル保存させて作った処理なので、必須ではないです。
aoi_client = AzureOpenAI(
azure_endpoint = endpoint,
api_key = key,
api_version = "2024-08-01-preview" # This API version or later is required to access seed/events/checkpoint features
)
def get_fine_tuned_model_info(client, job_id:str) -> str:
# Fine-tuning jobの情報を取得
response = client.fine_tuning.jobs.retrieve(job_id)
print(response.model_dump_json(indent=2))
fine_tuned_model = response.fine_tuned_model
pattern = r"-[0-9a-f]{20,}-" # 20桁以上の16進数文字列にマッチする正規表現
deployment_name = re.sub(pattern, "-", fine_tuned_model)
# チェックポイント保存
response = client.fine_tuning.jobs.checkpoints.list(job_id)
print(response.model_dump_json(indent=2))
# with open(f"./data/09.checkpoints_{job_id}.json", "w", encoding="utf-8") as f:
# f.write(response.model_dump_json(indent=2))
return fine_tuned_model, deployment_name
fine_tuned_model, deployment_name = get_fine_tuned_model_info(aoi_client, job_id)
4. モデルデプロイ
モデルのデプロイです。begin_create_or_update関数のdeploymentに渡す値は、何度か変えました。skuのcapacityは「1 分あたりのトークン数レート制限」で、nameがデプロイの種類です。
実行するとだいたい8分くらいかかることが多かったです。
csm_client = CognitiveServicesManagementClient(
credential=DefaultAzureCredential(),
subscription_id=subscription,
)
def create_deployment(
client: CognitiveServicesManagementClient,
resource_group: str,
resource_name: str,
deployment_name: str,
model_name: str,
) -> None:
# デプロイメントの作成
try:
response = client.deployments.begin_create_or_update(
resource_group_name=resource_group,
account_name=resource_name,
deployment_name=deployment_name,
deployment={
"properties": {
"model": {"format": "OpenAI", "name": model_name, "version": "1"}
},
"sku": {"capacity": 200, "name": "GlobalStandard"},
},
)
response.result()
print(f"Deployment {deployment_name} created successfully.")
except Exception as e:
print(f"Error creating deployment: {e}")
create_deployment(
csm_client, resource_group, resource_name, deployment_name, fine_tuned_model
)
モデルデプロイが成功しているかを試しています。
def get_chat_client(endpoint: str, model_name: str) -> ChatCompletionsClient:
client = ChatCompletionsClient(
endpoint=endpoint+"models",
credential=AzureKeyCredential(key),
)
response = client.complete(
messages=[
SystemMessage(content="You are a helpful assistant."),
UserMessage(content="Hello")
],
model=model_name,
)
print(response.choices[0].message.content)
return client
chat_client = get_chat_client(endpoint, deployment_name)
5. 推論実施
ローカルのCSVを読み込んで、その内容で推論を1回ずつかけていきます。
システムプロンプトは適当に入れます。
df = pd.read_table("<ファイル>")
df.info()
display(df.head())
def get_answer(
system_prompt: str, question: str, client: ChatCompletionsClient, model_name: str
) -> str:
response = client.complete(
messages=[
SystemMessage(content=system_prompt),
UserMessage(content=question),
],
model=model_name,
)
return response.choices[0].message.content
df_answered = df.copy()
df_answered["infer_answer"] = None
system_prompt = "<システムプロンプト内容>"
for index, row in df_answered.iterrows():
df_answered.loc[index, "infer_answer"] = get_answer(
system_prompt, row["question"], chat_client, deployment_name
)
print(f"\rProcessing line {index}", end="")
display(df_answered)
ファイルから読み込んだデータフレームの構造です。questionとanswer以外は不要です。
df.info()結果
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 row 100 non-null int64
1 category 100 non-null object
2 question 100 non-null object
3 answer 100 non-null object
6. 結果のローカルファイル保存
結果のDataFrameをローカルファイルに保存します。
df_answered.to_csv(
f"./data/infer_result_{deployment_name}.txt", index=False, sep="\t"
)
7. デプロイ削除
課金を抑えるためにすぐにデプロイを削除します。
だいたい4分くらいかかる処理でした。
csm_client.deployments.begin_delete(
resource_group_name=resource_group,
account_name=resource_name,
deployment_name=deployment_name,
).result()