数据集（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": "..."
}