Cloud Monitor：資料集（Dataset）

安裝 SDK

pip install alibabacloud-cms20240330-inner==6.0.8

配置憑證

通過環境變數或 .env 檔案設定：

export ALIBABA_CLOUD_ACCESS_KEY_ID="your-access-key-id"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="your-access-key-secret"
export ALIBABA_CLOUD_ENDPOINT="cms.cn-shanghai.aliyuncs.com"
export ALIBABA_CLOUD_CMS_WORKSPACE="your-workspace"

初始化用戶端

from alibabacloud_cms20240330.client import Client
from alibabacloud_tea_openapi.models import Config

config = Config(
    access_key_id="your-access-key-id",
    access_key_secret="your-access-key-secret",
    endpoint="cms.cn-shanghai.aliyuncs.com",
)
client = Client(config)
workspace = "your-workspace"

Schema 欄位類型

Dataset 支援使用者自訂 Schema，可選欄位類型：

向量索引：對 text 類型欄位設定 embedding 即開啟向量索引，支援語義檢索。

內建 ID 主鍵：Dataset 自動為每條資料產生唯一的 id 主鍵。UPDATE 和 DELETE 必須通過該主鍵進行。

SDK 樣本

from alibabacloud_cms20240330.models import (
    CreateDatasetRequest,
    IndexJsonKey,
    IndexKey,
)

schema = {
    "input": IndexKey(
        type="text",
        chn=True,                          # 開啟中文分詞
        embedding="text-embedding-v4",     # 開啟向量索引
    ),
    "output": IndexKey(
        type="text",
        chn=True,
        embedding="text-embedding-v4",
    ),
    "model": IndexKey(type="text"),
    "score": IndexKey(type="double"),
    "metadata": IndexKey(
        type="json",
        json_keys={
            "input_tokens": IndexJsonKey(type="long"),
            "output_tokens": IndexJsonKey(type="long"),
        },
    ),
}

request = CreateDatasetRequest(
    dataset_name="my_dataset",
    description="AI 問答資料集",
    schema=schema,
)
client.create_dataset(workspace, request)

插入資料（INSERT）

資料寫入通過 ExecuteQuery 介面執行標準 SQL。

單條插入：

from alibabacloud_cms20240330.models import ExecuteQueryRequest

sql = """
INSERT INTO my_dataset (input, output, model, score)
VALUES (
    '如何查看最近一小時的錯誤記錄檔？',
    '使用查詢語句: level:ERROR',
    'qwen-plus',
    0.95
)
"""
request = ExecuteQueryRequest(query=sql, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)
print(f"affected_rows: {response.body.meta.affected_rows}")

批量插入：

sql = """
INSERT INTO my_dataset (input, output, model, score)
VALUES
    ('統計今天各介面調用次數', 'SELECT api, count(*) ...', 'gpt-4o', 0.88),
    ('尋找響應逾時的請求', 'latency > 5000 | SELECT ...', 'claude-3.5-sonnet', 0.92)
"""
request = ExecuteQueryRequest(query=sql, type="SQL")
client.execute_query(workspace, "my_dataset", request)

更新資料（UPDATE）

更新操作必須通過 id 主鍵進行，暫不支援批次更新：

sql = """
UPDATE my_dataset
SET score = 0.98, output = '最佳化後的回答...'
WHERE id = 'your-doc-id'
"""
request = ExecuteQueryRequest(query=sql, type="SQL")
client.execute_query(workspace, "my_dataset", request)

刪除資料（DELETE）

刪除操作同樣必須通過 id 主鍵進行：

sql = "DELETE FROM my_dataset WHERE id = 'your-doc-id'"
request = ExecuteQueryRequest(query=sql, type="SQL")
client.execute_query(workspace, "my_dataset", request)

提示：可先通過查詢擷取目標資料的 id，再執行更新或刪除。

全文檢索索引

全文檢索索引使用關鍵詞匹配文法，與 SLS 查詢文法一致：

使用關鍵詞匹配文法，可單獨使用或通過管道 | 與 SQL 組合：

from alibabacloud_cms20240330.models import ExecuteQueryRequest

# 全文檢索索引
request = ExecuteQueryRequest(query="input:錯誤", type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

# 全文檢索索引 + SQL
query = "input:錯誤 | SELECT input, score FROM my_dataset ORDER BY score DESC LIMIT 5"
request = ExecuteQueryRequest(query=query, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

語義搜尋

語義搜尋基於向量相似性進行檢索，前提是對應欄位已在 Schema 中配置 embedding 向量索引。

提供兩種使用形式，滿足不同的查詢和分析情境：

形式一：檢索文法 similarity()

用於管道 | 左側的檢索條件中，可與全文檢索索引文法通過 AND / OR混合使用：

# 語義搜尋
request = ExecuteQueryRequest(query="similarity(input, '日誌分析') < 0.3", type="SQL")

# 語義搜尋 + 全文檢索索引
request = ExecuteQueryRequest(
    query="similarity(input, '日誌分析') < 0.3 AND model:qwen-plus",
    type="SQL",
)

# 語義搜尋 + SQL
request = ExecuteQueryRequest(
    query="similarity(input, '日誌分析') < 0.3 | SELECT input, score FROM my_dataset ORDER BY score DESC",
    type="SQL",
)

形式二：SQL 函數 semantic_distance()

用於管道 | 右側的 SQL 陳述式中，可在 SELECT、WHERE、ORDER BY 中靈活使用：

sql = """
SELECT input, semantic_distance(input, '日誌查詢統計') AS similarity
FROM my_dataset
WHERE semantic_distance(input, '日誌查詢統計') < 0.3
ORDER BY semantic_distance(input, '日誌查詢統計') ASC
"""
request = ExecuteQueryRequest(query=sql, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

SQL 分析

Dataset 支援標準 SQL 查詢與統計分析，包括 GROUP BY / HAVING / ORDER BY / LIMIT / CTE / 子查詢 / 視窗函數等。SQL 引擎基於 PrestoSQL 文法，在服務端轉換為 PostgreSQL 執行。

sql = """
SELECT model, count(*) AS total, avg(score) AS avg_score
FROM my_dataset
GROUP BY model
ORDER BY total DESC
"""
request = ExecuteQueryRequest(query=sql, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

分頁使用 LIMIT offset, count 文法：

page_size = 10
page = 2
skip = (page - 1) * page_size
# 第 2 頁（跳過 10 條，取 10 條）
sql = f"SELECT * FROM my_dataset ORDER BY score DESC LIMIT {skip}, {page_size}"
request = ExecuteQueryRequest(query=sql, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

組合查詢

四種查詢模式可通過管道 | 自由組合，檢索條件放 | 左側，SQL 放右側：

# 全文 + SQL + semantic_distance
query = """
model:qwen-plus
| SELECT input, output, score
  FROM my_dataset
  WHERE semantic_distance(input, '資料分析') < 0.4
  ORDER BY score DESC
  LIMIT 10
"""
request = ExecuteQueryRequest(query=query, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

關於 SQL 支援的完整能力（函數、子句、限制等），請參閱 Dataset 產品功能文檔。

列舉 Dataset

from alibabacloud_cms20240330.models import ListDatasetsRequest

request = ListDatasetsRequest(max_results=100)
response = client.list_datasets(workspace, request)
for ds in response.body.datasets:
    print(f"{ds.dataset_name}: {ds.description}")

# 按名稱過濾
request = ListDatasetsRequest(max_results=100, dataset_name="my_dataset")
response = client.list_datasets(workspace, request)

擷取 Dataset 詳情

response = client.get_dataset(workspace, "my_dataset")
print(response.body.to_map())  # 包含 Schema、建立時間等

更新 Dataset

from alibabacloud_cms20240330.models import UpdateDatasetRequest

request = UpdateDatasetRequest(description="更新後的描述")
client.update_dataset(workspace, "my_dataset", request)

刪除 Dataset

# 謹慎操作，資料不可恢複
client.delete_dataset(workspace, "my_dataset")

情境一：Bad Case 管理

從線上資料中篩選低分樣本，人工標註後更新：

# 1. 篩選低分樣本
sql = "SELECT id, input, output, score FROM my_dataset WHERE score < 0.3 ORDER BY score ASC LIMIT 50"
request = ExecuteQueryRequest(query=sql, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

# 2. 人工審核後更新標註
sql = """
UPDATE my_dataset
SET human_label = 'hallucination', fix_suggestion = '需要增加事實核查'
WHERE id = 'bad-case-id'
"""
client.execute_query(workspace, "my_dataset", ExecuteQueryRequest(query=sql, type="SQL"))

情境二：版本迴歸測試

基於 Dataset 構建評測基準集，對比不同模型版本效果：

# 擷取基準測試集
sql = "SELECT id, input, expected_output FROM my_dataset WHERE is_baseline = 'true'"
request = ExecuteQueryRequest(query=sql, type="SQL")
baseline = client.execute_query(workspace, "my_dataset", request)

# 運行新版本模型，對比結果
for sample in baseline.body.data:
    new_output = run_new_model(sample["input"])
    score = evaluate(new_output, sample["expected_output"])
    # 記錄評測結果...

情境三：訓練資料匯出

篩選高品質樣本匯出用於模型 SFT 微調或 RL 後訓練：

import json

# 查詢高品質樣本
sql = """
SELECT input, output FROM my_dataset
WHERE human_label = 'correct' AND score >= 0.9
LIMIT 10000
"""
request = ExecuteQueryRequest(query=sql, type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

# 匯出為 JSONL 格式
with open("sft_data.jsonl", "w") as f:
    for item in response.body.data:
        f.write(json.dumps({
            "input": item["input"],
            "output": item["output"],
        }, ensure_ascii=False) + "\n")

Schema 設計

• 有高頻語義搜尋需求的欄位，建議配置 embedding 開啟向量索引。

• 使用 json 類型支援動態擴充的資料列，如中繼資料、自訂多維打標列等。

資料管理

• 通過查詢擷取 id 後再執行更新 / 刪除。

• 定期清理低品質資料，保持信噪比。

• 結合 Pipeline 實現自動化資料清洗入庫。

查詢最佳化

• 語義搜尋配合條件過濾縮小範圍。

• 巨量資料量情境建議使用分頁方式分批擷取資料。

• 複雜統計優先使用 SQL 分析。

Dataset 管理

資料操作

所有資料操作統一通過 ExecuteQuery 介面：

from alibabacloud_cms20240330.models import ExecuteQueryRequest

request = ExecuteQueryRequest(query="...", type="SQL")
response = client.execute_query(workspace, dataset_name, request)

響應結構

{
  "data": [{"id": "...", "input": "...", "score": 0.95, ...}],
  "meta": {
    "count": 10,
    "affectedRows": 0,
    "elapsedMillisecond": 42,
    "progress": "Complete"
  },
  "requestId": "..."
}