Platform For AI:Kohya トレーニングサービスのデプロイ

前提条件

トレーニング中に生成されるモデルおよび構成ファイルを保存するための Object Storage Service (OSS) バケットを作成済みである必要があります。詳細については、「バケットの作成」をご参照ください。

スケーラブルな Kohya トレーニングサービスのデプロイ

このセクションでは、PAI が提供する kohya_ss 事前設定イメージを例として使用します。

1. PAI コンソールにログインします。ページ上部からリージョンを選択し、目的のワークスペースを選択して、Elastic Algorithm Service (EAS) をクリックします。

2. トレーニングサービスをデプロイします。

   以下のデプロイ方法が利用可能です：

   統合デプロイメント

   統合デプロイメントでは、キュー サービス、永続フロントエンド サービス、および スケーラブル ジョブ サービス を単一ユニットとしてデプロイします。

   1. Deploy Service をクリックし、Custom Model Deployment エリアで JSON Deployment をクリックします。

   2. 以下の JSON 構成を入力します。

      {
        "cloud": {
          "computing": {
            "instance_type": "ecs.gn6i-c4g1.xlarge"
          }
        },
        "containers": [
          {
            "image": "eas-registry-vpc.cn-hangzhou.cr.aliyuncs.com/pai-eas/kohya_ss:2.2"
          }
        ],
        "features": {
          "eas.aliyun.com/extra-ephemeral-storage": "30Gi"
        },
        "front_end": {
          "image": "eas-registry-vpc.cn-hangzhou.cr.aliyuncs.com/pai-eas/kohya_ss:2.2",
          "port": 8001,
          "script": "python -u kohya_gui.py --listen 0.0.0.0 --server_port 8001 --data-dir /workspace --headless --just-ui --job-service"
        },
        "metadata": {
          "cpu": 4,
          "enable_webservice": true,
          "gpu": 1,
          "instance": 1,
          "memory": 15000,
          "name": "kohya_job",
          "type": "ScalableJobService"
        },
        "name": "kohya_job",
        "storage": [
          {
            "mount_path": "/workspace",
            "oss": {
              "path": "oss://examplebucket/kohya/",
              "readOnly": false
            },
            "properties": {
              "resource_type": "model"
            }
          }
        ]
      }

      主なパラメーター：

      パラメーター		説明
      metadata	name	カスタムサービス名。リージョン内で一意である必要があります。
      	type	サービスタイプ。統合デプロイメントの場合は ScalableJobService を指定します。
      	enable_webservice	フロントエンドの Web アプリケーション をデプロイするには、true を指定します。
      front_end	image	フロントエンドインスタンスのイメージ。バージョン 2.2 の kohya_ss イメージを選択します。 説明 イメージのバージョンは頻繁に更新されます。デプロイ時には最新バージョンを選択してください。
      	script	フロントエンドインスタンスの起動コマンド：python -u kohya_gui.py --listen 0.0.0.0 --server_port 8000 --headless --just-ui --job-service。パラメーターの説明は以下のとおりです： --listen：外部リクエスト用に指定された IP アドレスにバインドします。 --server_port：リスニングポート番号です。 --just-ui：UI 専用フロントエンドモードでサービスを起動します。 --job-service：トレーニングタスクをスケーラブルなジョブサービスにルーティングします。
      	port	ポート番号。これは front_end.script 内の server_port 値と一致させる必要があります。
      containers	image	未指定の場合、フロントエンドインスタンスのイメージがデフォルト値として使用されます。
      	instance_type	スケーラブルなジョブサービスのインスタンスタイプ。必ず GPU タイプを指定してください。未指定の場合は、cloud.computing.instance_type の値がデフォルトとなります。
      storage	path	トレーニング出力を保存する同一リージョン内の OSS パス。例：oss://examplebucket/kohya/。
      	readOnly	OSS へのモデルファイルの保存を許可するには、false を指定します。
      	mount_path	カスタマイズ可能なマウントパス。この例では /workspace を指定します。
      cloud	instance_type	サービスのインスタンスタイプ。統合デプロイメントでは、フロントエンドおよびジョブインスタンスの両方に適用されます。Kohya トレーニングには GPU タイプが必要です。

   3. Deploy をクリックします。

   独立デプロイメント

   独立デプロイメントでは、スケーラブルなジョブサービスとフロントエンドサービスを別々にデプロイします。これにより、単一のスケーラブルなジョブサービスが複数のフロントエンドサービスからのリクエストを受け付けられるようになります。

   1. スケーラブルなジョブサービスをデプロイします。

      1. Deploy Service をクリックし、Custom Model Deployment セクションで JSON Deployment をクリックします。

      2. スケーラブルなジョブサービス用の以下の JSON 構成を入力します。

         {
           "cloud": {
             "computing": {
               "instance_type": "ecs.gn6i-c4g1.xlarge"
             }
           },
           "containers": [
             {
               "image": "eas-registry-vpc.cn-hangzhou.cr.aliyuncs.com/pai-eas/kohya_ss:2.2"
             }
           ],
           "features": {
             "eas.aliyun.com/extra-ephemeral-storage": "30Gi"
           },
           "metadata": {
             "instance": 1,
             "name": "kohya_scalable_job",
             "type": "ScalableJob"
           },
           "storage": [
             {
               "mount_path": "/workspace",
               "oss": {
                 "path": "oss://examplebucket/kohya/",
                 "readOnly": false
               },
               "properties": {
                 "resource_type": "model"
               }
             }
           ]
         }

         主なパラメーター：

         パラメーター		説明
         metadata	name	カスタムサービス名。リージョン内で一意である必要があります。
         	type	サービスタイプ。独立デプロイメントの場合は ScalableJob を指定します。
         containers	image	スケーラブルなジョブサービスのイメージ。バージョン 2.2 の kohya_ss イメージを選択します。 説明 イメージのバージョンは頻繁に更新されます。デプロイ時には最新バージョンを選択してください。
         storage	path	トレーニング出力を保存する同一リージョン内の OSS パス。例：oss://examplebucket/kohya/。
         	readOnly	OSS へのモデルファイルの保存を許可するには、false を指定します。
         	mount_path	カスタマイズ可能なマウントパス。この例では /workspace を指定します。
         cloud	instance_type	スケーラブルなジョブサービスのインスタンスタイプ。Kohya トレーニングには GPU タイプが必要です。

      3. Deploy をクリックします。

      4. デプロイ後に、Invocation Information を Service Type 列からクリックします。Public Endpoint タブで、エンドポイントとトークンをコピーして保存します。

   2. （任意）フロントエンドサービスをデプロイします。

      1. Deploy Service をクリックし、Custom Model Deployment セクションで JSON Deployment をクリックします。

      2. フロントエンドサービス用の以下の JSON 構成を入力します。

         {
           "cloud": {
             "computing": {
               "instance_type": "ecs.g6.large"
             }
           },
           "containers": [
             {
               "image": "eas-registry-vpc.cn-hangzhou.cr.aliyuncs.com/pai-eas/kohya_ss:2.2",
               "port": 8000,
               "script": "python kohya_gui.py --listen 0.0.0.0 --server_port 8000 --headless --just-ui --job-service --job-service-endpoint 166233998075****.vpc.cn-hangzhou.pai-eas.aliyuncs.com --job-service-token test-token --job-service-inputname kohya_scalable_job"
             }
           ],
           "metadata": {
             "enable_webservice": true,
             "instance": 1,
             "name": "kohya_scalable_job_front"
           },
           "storage": [
             {
               "mount_path": "/workspace",
               "oss": {
                 "path": "oss://examplebucket/kohya/",
                 "readOnly": false
               },
               "properties": {
                 "resource_type": "model"
               }
             }
           ]
         }

         主なパラメーター：

         パラメーター		説明
         metadata	name	フロントエンドサービスのカスタム名。
         	enable_webservice	フロントエンドの Web アプリケーション をデプロイするには、true を指定します。
         containers	image	フロントエンドサービスのイメージ。バージョン 2.2 の kohya_ss イメージを選択します。 説明 イメージのバージョンは頻繁に更新されます。デプロイ時には最新バージョンを選択してください。
         	script	フロントエンドサービスの起動コマンド：python kohya_gui.py --listen 0.0.0.0 --server_port 8000 --headless --just-ui --job-service --job-service-endpoint 166233998075****.vpc.cn-hangzhou.pai-eas.aliyuncs.com --job-service-token test-token --job-service-inputname kohya_scaled_job。パラメーターの説明は以下のとおりです： --listen：外部リクエスト用に指定された IP アドレスにバインドします。 --server_port：リスニングポート番号です。 --just-ui：UI 専用フロントエンドモードでサービスを起動します。 --job-service：トレーニングタスクをスケーラブルなジョブサービスにルーティングします。 --job-service-endpoint：スケーラブルなジョブサービスのエンドポイントです。 --job-service-token：スケーラブルなジョブサービスのトークンです。 --job-service-inputname：スケーラブルなジョブサービスのサービス名です。
         	port	ポート番号。これは containers.script 内の server_port 値と一致させる必要があります。
         storage	path	トレーニング出力を保存する同一リージョン内の OSS パス。例：oss://examplebucket/kohya/。
         	readOnly	OSS へのモデルファイルの保存を許可するには、false を指定します。
         	mount_path	カスタマイズ可能なマウントパス。この例では /workspace を指定します。
         cloud	instance_type	フロントエンドサービスのインスタンスタイプ。CPU タイプで十分です。

   3. Deploy をクリックします。

Kohya トレーニングサービスの呼び出し

Web UI からの呼び出し

フロントエンドサービスが kohya プリセットイメージ (バージョン 2.2 以降) を使用する場合、スケーラブルなジョブサービス機能がサポートされています。デプロイ後、Service Type 列の View Web App をクリックして LoRA トレーニングパラメーターを設定し、Kohya モデルをトレーニングします。詳細については、「LoRA モデルをトレーニングする」をご参照ください。

• トレーニングの開始 をクリックしてトレーニングジョブを送信します。現在のジョブが完了または終了するまで、このボタンは無効化されます。スケーラブルなジョブサービスのインスタンスは、トレーニングジョブの数に応じて自動的にスケールアップ／スケールダウンします。

• トレーニングの停止 をクリックして、現在実行中のトレーニングタスクを停止します。

カスタムフロントエンドからの呼び出し

Python 向け SDK を使用してスケーラブルなジョブサービスを呼び出し、コマンドタスクリクエストをキューに送信し、実行ログを取得します。カスタムフロントエンドイメージを使用する場合は、Web UI からスケーラブルなジョブサービスを呼び出すために、イメージ内に以下の API オペレーションを実装します。

1. スケーラブルなジョブサービスのエンドポイントおよびトークンを取得します。

   統合デプロイメント

   Elastic Algorithm Service (EAS) ページで、サービス名をクリックしてサービス詳細ページを開きます。Basic Information セクションで、View Endpoint Information をクリックします。Public Endpoint タブで、エンドポイントおよびトークンを取得します：

   • サービスエンドポイント：形式は <queue_name>.<service_name>.<uid>.<region>.pai-eas.aliyuncs.com です。例：kohya-job-queue-b****4f0.kohya-job.175805416243****.cn-beijing.pai-eas.aliyuncs.com。<queue_name> は、キュー サービスインスタンス名の「-0」より前の部分です。この名前は、Service Details ページのサービスインスタンス一覧で確認できます。

   • トークン：例：OGZlNzQwM2VlMWUyM2E2ZTAyMGRjOGQ5MWMyOTFjZGExNDgwMT****==

   独立デプロイメント

   Elastic Algorithm Service (EAS) ページで、スケーラブルなジョブサービスの Invocation Information を Service Type 列からクリックして、エンドポイントおよびトークンを取得します：

   • エンドポイント：例：175805416243****.cn-beijing.pai-eas.aliyuncs.com

   • トークン：例：Njk5NDU5MGYzNmRlZWQ3ND****QyMDIzMGM4MjExNmQ1NjE1NzY5Mw==

2. Python 向け SDK をインストールします。

   pip install -U eas-prediction --user

   SDK の API オペレーションの詳細については、「Python 向け SDK 使用ガイド」をご参照ください。

3. 入力キューおよび出力キューのクライアントを作成します。

   統合デプロイメント

   from eas_prediction import QueueClient
   
   if __name__ == '__main__':
       token = 'OGZlNzQwM2VlMWUyM2E2ZTAyMGRjOGQ5MWMyOTFjZGExNDgwMT****=='
       input_url = 'kohya-job-queue-bf****f0.kohya-job.175805416243****.cn-hangzhou.pai-eas.aliyuncs.com'
       sink_url = input_url + '/sink'
   
       # コマンドタスクのトレーニングおよび終了リクエストを送信するための入力キューを作成します。
       inputQueue = QueueClient(custom_url = input_url)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # タスク状態およびログを取得するための出力キューを作成します。
       sinkQueue = QueueClient(custom_url = sink_url)
       sinkQueue.set_token(token)
       sinkQueue.init()
   

   パラメーター：

   • token：前述の手順で取得したトークンに置き換えます。

   • input_url：前述の手順で取得したエンドポイントに置き換えます。

   独立デプロイメント

   from eas_prediction import QueueClient
   
   if __name__ == '__main__':
       endpoint = '166233998075****.cn-hangzhou.pai-eas.aliyuncs.com'
       token = 'YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OT****c1MTUxNg=='
       input_name = 'kohya_scalable_job'
       sink_name = input_name + '/sink'
   
       # コマンドタスクのトレーニングおよび終了リクエストを送信するための入力キューを作成します。
       inputQueue = QueueClient(endpoint, input_name)
       inputQueue.set_token(token)
       inputQueue.init()
   
       # タスク状態およびログを取得するための出力キューを作成します。
       sinkQueue = QueueClient(endpoint, sink_name)
       sinkQueue.set_token(token)
       sinkQueue.init()
   

   パラメーター：

   • endpoint：前述の手順で取得したエンドポイントに置き換えます。

   • token：前述の手順で取得したサービストークンに置き換えます。

   • input_name：スケーラブルなジョブサービスの名前です。

4. 入力キューにトレーニングタスクリクエストを送信します。

   統合デプロイメント

   from eas_prediction import QueueClient
   import uuid
   
   if __name__ == '__main__':
       token = 'OGZlNzQwM2VlMWUyM2E2ZTAyMGRjOGQ5MWMyOTFjZGExNDgwMT****=='
       input_url = 'kohya-job-queue-bf****f0.kohya-job.175805416243****.cn-hangzhou.pai-eas.aliyuncs.com'
       sink_url = input_url + '/sink'
   
       # コマンドリクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(custom_url = input_url)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # 各タスクリクエストに対して一意の taskId を生成します。
       task_id = uuid.uuid1().hex
       # コマンド文字列を作成します。
       cmd = "for i in {1..10}; do date; sleep 1; done;"
       # taskType を command として指定し、taskId を指定します。
       tags = {"taskType": "command", "taskId": task_id}
       # 入力キューにコマンドトレーニングタスクリクエストを送信します。
       index, request_id = inputQueue.put(cmd, tags)
       print(f'send index: {index}, request id: {request_id}')
   

   主なパラメーター：

   パラメーター	説明
   token	前述の手順で取得したトークンに置き換えます。
   input_url	前述の手順で取得したエンドポイントに置き換えます。
   cmd	実行するコマンドです。Python コマンドの場合は、リアルタイムログ出力のために -u を追加します。
   tags	トレーニングタスクリクエストのタグ： taskType：必ず command を指定します。 taskId：トレーニングタスクの一意の識別子です。

   独立デプロイメント

   from eas_prediction import QueueClient
   import uuid
   
   if __name__ == '__main__':
       endpoint = '166233998075****.cn-hangzhou.pai-eas.aliyuncs.com'
       token = 'M2EyNWYzNDJmNjQ5ZmUzMmM0OTMyMzgzYj****djN2IyODc1MTM5ZQ=='
       input_name = 'kohya_scalable_job'
   
       # コマンドリクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(endpoint, input_name)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # 各タスクリクエストに対して一意の taskId を生成します。
       task_id = uuid.uuid1().hex
       # コマンド文字列を作成します。
       cmd = "for i in {1..10}; do date; sleep 1; done;"
       # taskType を command として指定し、taskId を指定します。
       tags = {"taskType": "command", "taskId": task_id}
       # 入力キューにコマンドトレーニングタスクリクエストを送信します。
       index, request_id = inputQueue.put(cmd, tags)
       print(f'send index: {index}, request id: {request_id}')
   

   主なパラメーター：

   パラメーター	説明
   endpoint	前述の手順で取得したエンドポイントに置き換えます。
   token	前述の手順で取得したサービストークンに置き換えます。
   cmd	実行するコマンドです。Python コマンドの場合は、リアルタイムログ出力のために -u を追加します。
   tags	トレーニングタスクリクエストのタグ： taskType：必ず command を指定します。 taskId：トレーニングタスクの一意の識別子です。

5. リクエストのキュー状態を照会します。

   統合デプロイメント

   from eas_prediction import QueueClient
   import uuid
   
   if __name__ == '__main__':
       token = 'OGZlNzQwM2VlMWUyM2E2ZTAyMGRjOGQ5MWMyOTFjZGExNDgwMT****=='
       input_url = 'kohya-job-queue-bf****f0.kohya-job.175805416243****.cn-hangzhou.pai-eas.aliyuncs.com'
       sink_url = input_url + '/sink'
   
       # コマンドリクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(custom_url = input_url)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # 入力キューにコマンドリクエストを送信します。
       task_id = uuid.uuid1().hex
       cmd = "for i in {1..100}; do date; sleep 1; done;"
       tags = {"taskType": "command", "taskId": task_id}
       index, request_id = inputQueue.put(cmd, tags)
   
       # リクエストデータのキュー状態を照会します。
       search_info = inputQueue.search(index)
       print("index: {}, search info: {}".format(index, search_info))
   

   パラメーター：

   • token：前述の手順で取得したトークンに置き換えます。

   • input_url：前述の手順で取得したエンドポイントに置き換えます。

   独立デプロイメント

   from eas_prediction import QueueClient
   import uuid
   
   if __name__ == '__main__':
       endpoint = '166233998075****.cn-hangzhou.pai-eas.aliyuncs.com'
       token = 'M2EyNWYzNDJmNjQ5ZmUzMmM0OTMyMzgzYjBjOTdjN2I****1MTM5ZQ=='
       input_name = 'kohya_scalable_job'
   
       # コマンドリクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(endpoint, input_name)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # 入力キューにコマンドリクエストを送信します。
       task_id = uuid.uuid1().hex
       cmd = "for i in {1..100}; do date; sleep 1; done;"
       tags = {"taskType": "command", "taskId": task_id}
       index, request_id = inputQueue.put(cmd, tags)
   
       # リクエストデータのキュー状態を照会します。
       search_info = inputQueue.search(index)
       print("index: {}, search info: {}".format(index, search_info))
   

   パラメーター：

   • endpoint：前述の手順で取得したエンドポイントに置き換えます。

   • token：前述の手順で取得したサービストークンに置き換えます。

   JSON 形式のサンプル応答：

   {
   	'IsPending': False,
   	'WaitCount': 0
   }

   応答フィールド：

   パラメーター	説明
   IsPending	リクエストが処理中かどうか： True：リクエストは処理中です。 False：リクエストはキュー内にあります。
   WaitCount	キュー内におけるリクエストの位置。これは IsPending が False の場合のみ有効です。IsPending が True の場合は 0 を返します。

6. 出力キューから実行結果を取得します。

   トレーニングタスクのログはリアルタイムで出力キューに書き込まれます。queue.get(request_id=request_id, length=1, timeout='0s', tags=tags) を呼び出して、特定の task_id に対応するログを取得します。例：

   統合デプロイメント

   from eas_prediction import QueueClient
   import json
   import uuid
   
   if __name__ == '__main__':
       token = 'OGZlNzQwM2VlMWUyM2E2ZTAyMGRjOGQ5MWMyOTFjZGExNDgwMT****=='
       input_url = 'kohya-job-queue-bf****f0.kohya-job.175805416243****.cn-hangzhou.pai-eas.aliyuncs.com'
       sink_url = input_url + '/sink'
   
       # コマンドリクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(custom_url = input_url)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # コマンド実行ログを取得するための出力キュークライアントを作成します。
       sinkQueue = QueueClient(custom_url = sink_url)
       sinkQueue.set_token(token)
       sinkQueue.init()
   
       # 入力キューにコマンドリクエストを送信します。
       cmd = "for i in {1..10}; do date; sleep 1; done;"
       task_id = uuid.uuid1().hex
       tags = {"taskType": "command", "taskId": task_id}
       index, request_id = inputQueue.put(cmd, tags)
   
       # 出力キューから指定された taskId を持つトレーニングタスクの出力ログをリアルタイムで取得します。
       running = True
       while running:
           dfs = sinkQueue.get(length=1, timeout='0s', tags=tags)
           if len(dfs) == 0:
               continue
           df = dfs[0]
           data = json.loads(df.data.decode())
           state = data["state"]
           print(data.get("log", ""))
           if state in {"Exited", "Stopped", "Fatal", "Backoff"}:
               running = False
   

   パラメーター：

   • token：前述の手順で取得したトークンに置き換えます。

   • input_url：前述の手順で取得したエンドポイントに置き換えます。

   独立デプロイメント

   from eas_prediction import QueueClient
   import json
   import uuid
   
   if __name__ == '__main__':
       endpoint = '166233998075****.cn-hangzhou.pai-eas.aliyuncs.com'
       token = 'M2EyNWYzNDJmNjQ5ZmUzMmM0OTMyMzgzYjBjOTdjN2IyOD****M5ZQ=='
       input_name = 'kohya_scalable_job'
       sink_name = input_name + '/sink'
   
       # コマンドリクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(endpoint, input_name)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # コマンド実行ログを取得するための出力キュークライアントを作成します。
       sinkQueue = QueueClient(endpoint, sink_name)
       sinkQueue.set_token(token)
       sinkQueue.init()
   
       # 入力キューにコマンドリクエストを送信します。
       cmd = "for i in {1..10}; do date; sleep 1; done;"
       task_id = uuid.uuid1().hex
       tags = {"taskType": "command", "taskId": task_id}
       index, request_id = inputQueue.put(cmd, tags)
   
       # 出力キューから指定された taskId を持つトレーニングタスクの出力ログをリアルタイムで取得します。
       running = True
       while running:
           dfs = sinkQueue.get(length=1, timeout='0s', tags=tags)
           if len(dfs) == 0:
               continue
           df = dfs[0]
           data = json.loads(df.data.decode())
           state = data["state"]
           print(data.get("log", ""))
           if state in {"Exited", "Stopped", "Fatal", "Backoff"}:
               running = False
   

   パラメーター：

   • endpoint：前述の手順で取得したエンドポイントに置き換えます。

   • token：前述の手順で取得したサービストークンに置き換えます。

   バイト形式のサンプル応答：

   {
   	"taskId": "e97409eea4a111ee9cb600163e08****",
   	"command": "python3 -u test.py --args=xxx",
   	"state": "Running",
   	"log": "prepare tokenizer\\n"
   }

   応答フィールド：

   フィールド	説明
   taskId	トレーニングタスクの一意の識別子です。
   command	ジョブによって実行されたコマンドです。
   state	タスクの状態： Running：タスクは実行中です。 Exited：タスクは完了しました。 Fatal：実行例外が発生しました。 Stopping：タスクは停止中です。 Stopped：タスクは停止されました。
   log	出力ログ。完全なログは、同じ taskId のエントリのシーケンスです。

7. トレーニングタスクを停止します。

   送信済みのタスクを停止するには、まず queue.search(index) を呼び出して、タスクが キュー待ち 状態か 実行中 状態かを確認し、それに応じて停止処理を行います。例：

   統合デプロイメント

   from eas_prediction.queue_client import QueueClient
   import uuid
   
   if __name__ == '__main__':
       token = 'OGZlNzQwM2VlMWUyM2E2ZTAyMGRjOGQ5MWMyOTFjZGExNDgwMT****=='
       input_url = 'kohya-job-queue-bf****f0.kohya-job.175805416243****.cn-hangzhou.pai-eas.aliyuncs.com'
       sink_url = input_url + '/sink'
   
       # コマンドおよび終了リクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(custom_url = input_url)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # 入力キューにコマンドリクエストを送信します。
       cmd = "for i in {1..10}; do date; sleep 1; done;"
       task_id = uuid.uuid1().hex  # このタスクリクエストの taskId。
       tags = {"taskType": "command", "taskId": task_id}
       index, request_id = inputQueue.put(cmd, tags)
       print(f'cmd send, index: {index}, task_id: {task_id}')
   
       job_index = index  # タスクリクエスト送信時に返されたインデックス。
   
       pending_detail = inputQueue.search(job_index)
       print(f'search info: {pending_detail}')
       if len(pending_detail) > 0 and pending_detail.get("IsPending", True) == False:
           # コマンドタスクはまだキュー内にあります。入力キューから直接削除します。
           inputQueue.delete(job_index)
           print(f'delete task index: {job_index}')
       else:
           # コマンドタスクは実行中です。入力キューに停止シグナルを送信します。
           stop_data = "stop"
           tags = {"_is_symbol_": "true", "taskId": task_id}
           inputQueue.put(stop_data, tags)
           print(f'stop task index: {job_index}')
   

   主なパラメーター：

   パラメーター	説明
   token	前述の手順で取得したトークンに置き換えます。
   input_url	前述の手順で取得したエンドポイントに置き換えます。
   stop_data	stop を指定します。
   tags	_is_symbol_：必須。終了リクエストであることを示すには true を指定します。 task_id：停止するトレーニングタスクの ID です。

   独立デプロイメント

   from eas_prediction.queue_client import QueueClient
   import uuid
   
   if __name__ == '__main__':
       endpoint = '166233998075****.cn-hangzhou.pai-eas.aliyuncs.com'
       token = 'M2EyNWYzNDJmNjQ5ZmUzMmM0OTMyMzgzYjBjOTdjN2IyODc1MTM5****'
       input_name = 'kohya_scalable_job'
   
       # コマンドおよび終了リクエストを送信するための入力キュークライアントを作成します。
       inputQueue = QueueClient(endpoint, input_name)
       inputQueue.set_token(token)
       inputQueue.init(gid="superwatcher")
   
       # 入力キューにコマンドリクエストを送信します。
       cmd = "for i in {1..10}; do date; sleep 1; done;"
   
       # このタスクリクエストの taskId。
       task_id = uuid.uuid1().hex
       tags = {"taskType": "command", "taskId": task_id}
       index, request_id = inputQueue.put(cmd, tags)
       print(f'cmd send, index: {index}, task_id: {task_id}')
   
       job_index = index  # タスクリクエスト送信時に返されたインデックス。
   
       pending_detail = inputQueue.search(job_index)
       print(f'search info: {pending_detail}')
       if len(pending_detail) > 0 and pending_detail.get("IsPending", True) == False:
           # コマンドタスクはまだキュー内にあります。入力キューから直接削除します。
           inputQueue.delete(job_index)
           print(f'delete task index: {job_index}')
       else:
           # コマンドタスクは実行中です。入力キューに停止シグナルを送信します。
           stop_data = "stop"
           tags = {"_is_symbol_": "true", "taskId": task_id}
           inputQueue.put(stop_data, tags)
           print(f'stop task index: {job_index}')
   

   主なパラメーター：

   パラメーター	説明
   endpoint	前述の手順で取得したエンドポイントに置き換えます。
   token	前述の手順で取得したサービストークンに置き換えます。
   stop_data	stop を指定します。
   tags	_is_symbol_：必須。終了リクエストであることを示すには true を指定します。 task_id：停止するトレーニングタスクの ID です。

関連ドキュメント

• スケーラブルなジョブサービスの概要については、「スケーラブルなジョブサービスの概要」をご参照ください。

• 推論用途でスケーラブルなジョブサービスを使用するには、「ID 写真生成用オンラインサービスのデプロイ」をご参照ください。