Cloud Monitor:データセット

SDK のインストール

pip install alibabacloud-cms20240330==6.4.0

認証情報の構成

環境変数または .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"

スキーマのフィールド型

データセットはユーザー定義のスキーマをサポートします。以下のフィールド型が利用可能です：

ベクターインデックス：embedding を text 型フィールドに設定すると、ベクターインデックスが有効化され、意味検索が可能になります。

組み込みの ID プライマリキー：データセットは各データ入力に対して自動的に一意の 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 Q&A データセット",
    schema=schema,
)
client.create_dataset(workspace, request)

データの挿入（INSERT）

ExecuteQuery API を使用した標準 SQL ステートメントにより、データを挿入できます。

単一挿入：

from alibabacloud_cms20240330.models import ExecuteQueryRequest

sql = """
INSERT INTO my_dataset (input, output, model, score)
VALUES (
    '直近 1 時間のエラーログを表示する方法は？',
    'クエリを使用：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
    ('本日の各インターフェイスに対する API 呼び出し数をカウント', '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 を取得し、その後その ID を使用してレコードを更新または削除できます。

全文検索

全文検索は、SLS クエリ構文と互換性のあるキーワード一致構文を使用します：

キーワード一致構文を単独で使用することも、パイプ | を使用して SQL と組み合わせることもできます：

from alibabacloud_cms20240330.models import ExecuteQueryRequest

# 全文検索
request = ExecuteQueryRequest(query="input:error", type="SQL")
response = client.execute_query(workspace, "my_dataset", request)

# 全文検索 ＋ SQL
query = "input:error | 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)

意味検索

意味検索は、対応するフィールドにスキーマで embedding ベクターインデックスが構成されている場合に、ベクター類似度に基づく取得を実行します。

異なるクエリおよび分析シナリオに応じて、2 種類の形式が利用可能です：

形式 1：検索構文 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",
)

形式 2：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 分析

データセットは、GROUP BY、HAVING、ORDER BY、LIMIT、CTE（共通テーブル式）、サブクエリ、ウィンドウ関数を含む標準 SQL クエリおよび統計分析をサポートします。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)

ハイブリッドクエリ

4 種類のクエリモードは、パイプ | を使用して自由に組み合わせられます。| の左側に検索条件を、右側に 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 機能（関数、句、制限事項を含む）の完全な一覧については、「データセット製品ドキュメント」をご参照ください。

データセットの一覧表示

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)

データセットの表示

response = client.get_dataset(workspace, "my_dataset")
print(response.body.to_map())  # スキーマ、作成時刻なども含まれます。

データセットの更新

from alibabacloud_cms20240330.models import UpdateDatasetRequest

request = UpdateDatasetRequest(description="更新後の説明")
client.update_dataset(workspace, "my_dataset", request)

データセットの削除

# 注意：この操作は取り消せません。
client.delete_dataset(workspace, "my_dataset")

ユースケース 1：不良ケース管理

本番データからスコアが低いサンプルをフィルタリングし、手動ラベル付け後に更新します：

# 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 = 'ハルシネーション', fix_suggestion = '事実確認が必要'
WHERE id = 'bad-case-id'
"""
client.execute_query(workspace, "my_dataset", ExecuteQueryRequest(query=sql, type="SQL"))

ユースケース 2：回帰テスト

異なるモデルバージョンのパフォーマンスを比較するためのベンチマークデータセットを構築します：

# ベンチマークデータセットを取得
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"])
    # 評価結果を記録...

ユースケース 3：トレーニングデータのエクスポート

高品質なサンプルをフィルタリングし、モデルの 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")

スキーマ設計

• 意味検索で頻繁に使用されるフィールドには、ベクターインデックスを有効化するために embedding の構成を推奨します。

• メタデータやカスタム多次元タグ列など、動的に拡張可能なデータ列には、json データ型の使用を推奨します。

データ管理

• id をクエリで取得し、その後更新または削除を行ってください。

• ノイズ比率を低く保つために、定期的に低品質データをクリーンアップしてください。

• データクリーンアップおよびインジェストを自動化するパイプラインを活用してください。

クエリ最適化

• 意味検索と条件付きフィルタリングを組み合わせて、結果セットを絞り込んでください。

• 大規模な結果セットの場合、ページネーションを使用してデータをバッチ処理で取得してください。

• 複雑な集約には、SQL 分析をご活用ください。

データセットの管理

データ操作

すべてのデータ操作は ExecuteQuery API を使用します：

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": "..."
}