正常化DLC/DSW/EAS服務打標 - Platform For AI

標籤規則用於在工作空間內統一定義 DLC、DSW、EAS 等服務的標籤填寫規範。工作空間管理員或負責人可配置必填標籤 Key、Value 取值範圍、備忘及適用產品範圍等規則。規則生效後，演算法開發人員在建立 DLC 任務、DSW 執行個體、EAS 服務時，需按規則填寫標籤，並進行合理性校正，從而推動團隊成員正常化打標，為後續按標籤歸檔、分析任務分布和資源消耗等提供基礎。

定義標籤規則

說明

工作空間負責人和管理員有權編輯標籤規則。

進入工作空間詳情頁，單擊右上方工作空间配置 > 通用配置。
下滑至标签配置地區，單擊修改配置，進入編輯模式。
按需添加標籤規則，配置完成後單擊保存。注意，標籤規則按配置的適用產品範圍生效；在對應子產品頁面中，這些標籤預設為必填項，未按規則填寫將無法提交。演算法開發人員也可在此基礎上，按照標籤Key與Value取值要求添加自訂標籤。
- 标签Key 取值要求：不能以 aliyun 或 acs: 開頭，不能包含 http:// 和 https://，長度不超過 128 個字元。Key 只能包含字母、數字、虛線（-）、中文、點（.），不允許使用底線（_）。
- Value 取值要求：不能包含 http:// 和 https://，長度不超過 256 個字元。多個枚舉值以英文逗號分隔。
- 限制取值：開啟後，使用者只能從預置枚舉值中選擇，無法自訂輸入。建議保持預設開啟，確保團隊打標一致性。
- 适用产品：選擇當前規則對哪些子產品（DLC、DSW、EAS）生效，至少勾選一項。

規範打標

提交 DLC 任務或建立 DSW、EAS 執行個體時，需按工作空間定義的標籤規則打標。以 DLC 任務提交為例：

進入 DLC 控制台，單擊新建任务。
在任務配置頁面，找到标签配置地區，按工作空間定義的標籤規則依次填寫各標籤的索引值。
在建立任務時您也可以添加額外的自訂標籤。
填寫所有必選標籤後，提交任務。

重要

標籤規則在提交時自動校正。標籤缺失或值不符合規範，提交將被拒絕並提示錯誤原因。

按標籤篩選分析

以DLC任務為例，工作清單頁支援按標籤過濾，方便按情境、模型類型等維度統計任務分布和用量。

進入 DLC 控制台工作清單頁。
在搜尋欄選擇标签篩選條件，輸入標籤 Key 和 Value 後單擊搜尋，列表將僅顯示合格任務。

附錄

API 參考

如需通過 API 大量操作或對接自動化流程，參考以下範例程式碼，替換地區、工作空間 ID 和 AK/SK 後直接使用。

大量設定標籤規則

UpdateConfigs 介面支援大量設定預置標籤規則，適用於多個工作空間統一配置標籤的情境。

樣本請求體結構如下：

{
  "Configs": [
    {
      "ConfigKey": "predefinedTags",
      "ConfigValue": "[{\"Type\":\"Tag\",\"Key\":\"情境\",\"Value\":\"{\\\"Products\\\":\\\"DLC,DSW,EAS\\\",\\\"Values\\\":\\\"訓練,評估,推理\\\",\\\"ValueFlexible\\\":false,\\\"Description\\\":\\\"請選擇當前使用情境\\\"}\"}]",
      "Labels": [
        {
          "Key": "system.categoryName",
          "Value": "CommonTagConfig"
        }
      ]
    }
  ]
}

Python 範例程式碼：大量設定標籤規則

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PAI 工作空間 - 大量建立/更新預置標籤規則 (CommonTagConfig)
介面：PUT /api/v1/workspaces/{WorkspaceId}/configs
使用前請填寫 ACCESS_KEY_ID、ACCESS_KEY_SECRET、REGION_ID、WORKSPACE_ID，
並在 TAGS 列表中定義標籤規則。
"""

import json, sys, hashlib, hmac, datetime, requests

ACCESS_KEY_ID     = "你的AccessKeyId"
ACCESS_KEY_SECRET = "你的AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "你的工作空間ID"

ENDPOINT = f"aiworkspace.{REGION_ID}.aliyuncs.com"
API_PATH = f"/api/v1/workspaces/{WORKSPACE_ID}/configs"

# 標籤定義：key-標籤名，products-生效產品，values-可選值，flexible-是否允許自訂，desc-描述
TAGS = [
    {"key": "情境",   "products": "DLC,DSW,EAS", "values": "訓練,評估,推理",       "flexible": False, "desc": "請選擇當前使用情境"},
    {"key": "模型類型","products": "DLC,DSW",     "values": "NLP文本,CV映像,多模態", "flexible": False, "desc": "請選擇模型類型"},
    {"key": "實驗方向","products": "DLC",          "values": "學習率策略,資料增強,模型蒸餾", "flexible": True, "desc": "請自訂實驗方向"},
]

def build_config_value(tags):
    arr = []
    for t in tags:
        inner = {"Products": t["products"], "Values": t["values"],
                 "ValueFlexible": t.get("flexible", False), "Description": t.get("desc", "")}
        arr.append({"Type": "Tag", "Key": t["key"], "Value": json.dumps(inner, ensure_ascii=False)})
    return json.dumps(arr, ensure_ascii=False)

def sign_request(method, path, headers, body):
    now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    headers.update({"x-acs-date": now, "x-acs-version": "2021-02-04",
                    "x-acs-action": "UpdateConfigs",
                    "x-acs-signature-nonce": hashlib.md5((now + body).encode()).hexdigest()})
    signed_keys = sorted(k for k in headers if k.startswith("x-acs-") or k in ("host", "content-type"))
    canonical_headers = "".join(f"{k}:{headers[k]}\n" for k in signed_keys)
    hashed_body = hashlib.sha256(body.encode("utf-8")).hexdigest()
    canonical_request = f"{method}\n{path}\n\n{canonical_headers}\n{';'.join(signed_keys)}\n{hashed_body}"
    string_to_sign = f"ACS3-HMAC-SHA256\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
    signature = hmac.new(ACCESS_KEY_SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest()
    headers["Authorization"] = (f"ACS3-HMAC-SHA256 Credential={ACCESS_KEY_ID},"
                                 f"SignedHeaders={';'.join(signed_keys)},Signature={signature}")
    return headers

def main():
    config_value = build_config_value(TAGS)
    request_body = json.dumps({"Configs": [{"ConfigKey": "predefinedTags", "ConfigValue": config_value,
                                            "Labels": [{"Key": "system.categoryName", "Value": "CommonTagConfig"}]}]},
                              ensure_ascii=False)
    url = f"https://{ENDPOINT}{API_PATH}"
    headers = sign_request("PUT", API_PATH, {"host": ENDPOINT, "content-type": "application/json"}, request_body)
    resp = requests.put(url, headers=headers, data=request_body.encode("utf-8"), timeout=30)
    print(f"HTTP {resp.status_code}: {resp.text}")
    if resp.status_code != 200:
        sys.exit(1)

if __name__ == "__main__":
    main()

DLC 提交任務時指定標籤

通過 CreateJob 介面提交 DLC 任務時，在 Settings.Tags 欄位填寫標籤。預置標籤必須全部填寫，值須從預設選項中選擇；ValueFlexible 為 true 的標籤可自訂輸入。

Python 範例程式碼：DLC 提交任務時指定標籤

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PAI-DLC 提交訓練任務（含預置標籤）
介面：POST /api/v1/jobs（API 版本 2020-12-03）
標籤放在 Settings.Tags 中，格式為字典 {"標籤名": "標籤值"}。
"""

import json, sys, hashlib, hmac, datetime, time, requests

ACCESS_KEY_ID     = "你的AccessKeyId"
ACCESS_KEY_SECRET = "你的AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "你的工作空間ID"
RESOURCE_ID       = ""  # 資源配額 ID，留空使用公用資源

DLC_ENDPOINT = f"pai-dlc.{REGION_ID}.aliyuncs.com"
DLC_API_PATH = "/api/v1/jobs"

# 根據工作空間預置標籤填寫，所有預置標籤必須全部填上
TAGS = {
    "情境":   "訓練",
    "模型類型": "NLP文本",
    "實驗方向": "精細化資料處理",
}

def sign_request(method, path, headers, body, action="CreateJob"):
    now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    headers.update({"x-acs-date": now, "x-acs-version": "2020-12-03", "x-acs-action": action,
                    "x-acs-signature-nonce": hashlib.md5((now + body).encode()).hexdigest()})
    signed_keys = sorted(k for k in headers if k.startswith("x-acs-") or k in ("host", "content-type"))
    canonical_headers = "".join(f"{k}:{headers[k]}\n" for k in signed_keys)
    hashed_body = hashlib.sha256(body.encode("utf-8")).hexdigest()
    canonical_request = f"{method}\n{path}\n\n{canonical_headers}\n{';'.join(signed_keys)}\n{hashed_body}"
    string_to_sign = f"ACS3-HMAC-SHA256\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
    signature = hmac.new(ACCESS_KEY_SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest()
    headers["Authorization"] = (f"ACS3-HMAC-SHA256 Credential={ACCESS_KEY_ID},"
                                 f"SignedHeaders={';'.join(signed_keys)},Signature={signature}")
    return headers

def main():
    job = {
        "DisplayName": f"my-training-job-{int(time.time())}",
        "JobType": "PyTorchJob",
        "WorkspaceId": WORKSPACE_ID,
        "UserCommand": "echo hello && sleep 5",
        "JobSpecs": [{"Type": "Worker",
                      "Image": "registry.cn-hangzhou.aliyuncs.com/pai-dlc/pytorch-training:1.12-gpu-py39-cu116-ubuntu20.04",
                      "PodCount": 1, "EcsSpec": "ecs.g6.large"}],
        "Settings": {"Tags": TAGS},  # 標籤放在 Settings.Tags 裡
    }
    if RESOURCE_ID:
        job["ResourceId"] = RESOURCE_ID
    request_body = json.dumps(job, ensure_ascii=False)
    url = f"https://{DLC_ENDPOINT}{DLC_API_PATH}"
    headers = sign_request("POST", DLC_API_PATH, {"host": DLC_ENDPOINT, "content-type": "application/json"}, request_body)
    resp = requests.post(url, headers=headers, data=request_body.encode("utf-8"), timeout=30)
    print(f"HTTP {resp.status_code}: {resp.text}")
    if resp.status_code != 200:
        sys.exit(1)

if __name__ == "__main__":
    main()

按標籤過濾 DLC 工作清單

ListJobs 介面支援通過 Tags 查詢參數按標籤過濾任務（JSON 字串格式）。多個標籤為 AND 關係，全部匹配才返回結果。

Python 範例程式碼：按標籤過濾 DLC 工作清單

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import sys
import hashlib
import hmac
import datetime
import requests
from urllib.parse import urlencode

ACCESS_KEY_ID     = "你的AccessKeyId"
ACCESS_KEY_SECRET = "你的AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "你的工作空間ID"

ENDPOINT = f"pai-dlc.{REGION_ID}.aliyuncs.com"
API_PATH = "/api/v1/jobs"


def build_query_params(workspace_id, filter_tags, page_size=50, page_number=1):
    query_params = {
        "WorkspaceId": workspace_id,
        "PageSize": str(page_size),
        "PageNumber": str(page_number),
        "SortBy": "GmtCreateTime",
        "Order": "desc",
    }

    if filter_tags:
        query_params["Tags"] = json.dumps(
            filter_tags, ensure_ascii=False, separators=(",", ":")
        )

    return query_params


def sign_request(method, path, query_string, headers, body=""):
    now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    headers.update({
        "x-acs-date": now,
        "x-acs-version": "2020-12-03",
        "x-acs-action": "ListJobs",
        "x-acs-signature-nonce": hashlib.md5((now + query_string + body).encode()).hexdigest(),
    })

    normalized_headers = {k.lower(): v for k, v in headers.items()}
    signed_keys = sorted(
        k for k in normalized_headers
        if k.startswith("x-acs-") or k in ("host", "content-type")
    )

    canonical_headers = "".join(f"{k}:{normalized_headers[k]}\n" for k in signed_keys)
    hashed_body = hashlib.sha256(body.encode("utf-8")).hexdigest()

    canonical_request = (
        f"{method}\n"
        f"{path}\n"
        f"{query_string}\n"
        f"{canonical_headers}\n"
        f"{';'.join(signed_keys)}\n"
        f"{hashed_body}"
    )

    string_to_sign = f"ACS3-HMAC-SHA256\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
    signature = hmac.new(
        ACCESS_KEY_SECRET.encode(),
        string_to_sign.encode(),
        hashlib.sha256
    ).hexdigest()

    headers["Authorization"] = (
        f"ACS3-HMAC-SHA256 Credential={ACCESS_KEY_ID},"
        f"SignedHeaders={';'.join(signed_keys)},"
        f"Signature={signature}"
    )
    return headers


def list_jobs_by_tags(filter_tags, page_size=50, page_number=1):
    query_params = build_query_params(WORKSPACE_ID, filter_tags, page_size, page_number)
    query_string = urlencode(sorted(query_params.items()))
    url = f"https://{ENDPOINT}{API_PATH}?{query_string}"

    headers = {
        "host": ENDPOINT,
        "content-type": "application/json",
    }
    headers = sign_request("GET", API_PATH, query_string, headers, body="")

    resp = requests.get(url, headers=headers, timeout=30)
    print(f"HTTP {resp.status_code}")
    print(resp.text)

    if resp.status_code != 200:
        sys.exit(1)

    return resp.json()


def main():
    filter_tags = {"情境": "訓練"}
    list_jobs_by_tags(filter_tags)


if __name__ == "__main__":
    main()

查看標籤修改記錄

Action Trail記錄了標籤規則的所有變更歷史，可查詢操作者、操作時間和變更內容。

登入Action Trail控制台，進入進階查詢頁面，服務名稱設為 PAI，事件名稱設為 UpdateConfigs，並尋找ConfigKey為predefinedTags的記錄，即可查看標籤規則的變更記錄。