AI 助理
文档备案控制台
官方文档
首页

规范化DLC/DSW/EAS服务打标

更新时间：
复制 MD 格式
产品详情
我的收藏

标签规则用于在工作空间内统一定义 DLC、DSW、EAS 等服务的标签填写规范。工作空间管理员或负责人可配置必填标签 Key、Value 取值范围、备注及适用产品范围等规则。规则生效后，算法开发人员在创建 DLC 任务、DSW 实例、EAS 服务时，需按规则填写标签，并进行合理性校验，从而推动团队成员规范化打标，为后续按标签归档、分析任务分布和资源消耗等提供基础。

定义标签规则

说明

工作空间负责人和管理员有权编辑标签规则。

  1. 进入工作空间详情页，单击右上角工作空间配置 > 通用配置。

  2. 下滑至标签配置区域，单击修改配置，进入编辑模式。

    image.png

  3. 按需添加标签规则，配置完成后单击保存。注意，已配置的标签规则会按设置的适用产品范围生效；在对应子产品页面中，这些标签默认为必填项，未按规则填写将无法提交。算法开发人员也可在此基础上，按照标签Key与Value取值要求添加其他自定义标签。

    • 标签Key 取值要求：不能以 aliyun 或 acs: 开头，不能包含 http:// 和 https://，长度不超过 128 个字符。Key 只能包含字母、数字、短横线（-）、中文、点（.），不允许使用下划线（_）。

    • Value 取值要求：不能包含 http:// 和 https://，长度不超过 256 个字符。多个枚举值以英文逗号分隔。

    • 限制取值：开启后，用户只能从预置枚举值中选择，无法自定义输入。建议保持默认开启，确保团队打标一致性。

    • 适用产品：选择当前规则对哪些子产品（DLC、DSW、EAS）生效，至少勾选一项。

规范打标

提交 DLC 任务或创建 DSW、EAS 实例时，需按工作空间定义的标签规则打标。以 DLC 任务提交为例：

  1. 进入 DLC 控制台，单击新建任务。

  2. 在任务配置页面，找到标签配置区域，按工作空间定义的标签规则依次填写各标签的键值。

    在创建任务时您也可以添加额外的自定义标签，如图中self-define-key标签。

    image.png

  3. 填写所有必选标签后，提交任务。

重要

标签规则在提交时自动校验。标签缺失或值不符合规范，提交将被拒绝并提示错误原因。

按标签筛选分析

以DLC任务为例，任务列表页支持按标签过滤，方便按场景、模型类型等维度统计任务分布和用量。

  1. 进入 DLC 控制台任务列表页。

  2. 在搜索栏选择标签筛选条件，输入标签 Key 和 Value 后单击搜索，列表将仅显示符合条件的任务。

    image

    筛选结果：

    image

附录

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()

查看标签修改记录

操作审计记录了标签规则的所有变更历史，可查询操作者、操作时间和变更内容。

登录操作审计控制台，进入高级查询页面，服务名称设为 PAI，事件名称设为 UpdateConfigs，并查找ConfigKey为predefinedTags的记录，即可查看标签规则的变更记录。

image

上一篇：调度配置下一篇：基于RAM的Condition属性设置资源的访问权限