センシティブ情報のマスキング - Vector Retrieval Service for Milvus

AI_PII_MASK 関数は、テキストに含まれる個人を特定できる情報 (PII) を識別し、マスクします。この関数は、カスタマーサービスログ、チケットのアーカイブ、およびアナリティクスデータにおけるプライバシー保護を目的として設計されています。

前提条件

AI_PII_MASK を使用する前に、以下が揃っていることを確認してください。

Milvus クラスターのエンドポイント (例： http://c-xxxx.milvus.aliyuncs.com:19530)。
認証情報 (<yourUsername>:<yourPassword> 形式)。
設定済みのテキストモデル (例： qwen3.7-max)。
(Python のみ) PyMilvus がインストール済みであること。

コマンド形式

AI_PII_MASK は、2 つの呼び出し方法をサポートしています。

REST インターフェイス — API を直接呼び出して、オンデマンドでテキストをマスクします。
Python のコレクション関数 — コレクションスキーマに関数を追加することで、データ書き込み時に PII が自動的にマスクされます。

REST インターフェイス

POST リクエストを /v2/vectordb/ai/pii_mask エンドポイントに送信します：

POST /v2/vectordb/ai/pii_mask
Content-Type: application/json

{
  "model_name": "<model name>",
  "texts": ["<text>"],
  "params": {"pii_types": ["EMAIL", "PHONE"]}
}

Python (コレクション関数)

TEXTTRANSFORM 関数型を使用する Function を含むスキーマを定義します。以下で使用される texttransform_function_type() ヘルパーは、PyMilvus のバージョン間の互換性を処理します。完全な実装については、「Python の使用例」をご参照ください。

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("content", DataType.VARCHAR, max_length=4096)
schema.add_field("masked", DataType.VARCHAR, max_length=4096)
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2)
schema.add_function(
    Function(
        name="mask_pii",
        function_type=texttransform_function_type(),
        input_field_names=["content"],
        output_field_names=["masked"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "<model name>",
            "task": "ai_pii_mask",
            "pii_types": "PERSON,EMAIL,PHONE",
            "mask_char": "*",
            "preserve_length": True,
            "temperature": 0,
        },
    )
)

パラメーター

パラメーター	説明
`model_name`	必須。設定済みのテキストモデルの名前。
`texts`	REST で必須。マスクするテキストの配列。
`pii_types`	オプション。マスクする PII のタイプ。配列またはカンマ区切りの文字列が使用でき、1〜30個のタイプをサポートします。デフォルトでは、名前、メールアドレス、電話番号、ID番号、銀行カード番号、住所、URL、IPアドレス、トークン、パスワードをマスクします。
`mask_char`	オプション。単一の置換文字。デフォルトは `*` です。
`preserve_length`	オプション。元のセンシティブなセグメントの長さを維持するかどうかを指定します。デフォルトは `true` です。
`temperature`/`max_concurrency`/`timeout_sec`	オプション。モデルの安定性、同時実行性、タイムアウトの設定。
`provider`/`task`	コレクション関数でのみ必須。固定値： `aliyun_milvus` および `ai_pii_mask`。

レスポンス

data.output.outputs は、入力と同じ順序でマスクされたテキストを返します。コレクション関数を使用する場合、結果はターゲットのテキストフィールドに書き込まれます。

使用例：ストレージに保存する前のカスタマーサービスログのマスキング

運用チームは分析のために連絡記録を保持する必要がありますが、アナリティクスデータベースに名前、メールアドレス、電話番号を保存することはできません。次の例では、REST を使用してログエントリをマスクし、コレクション関数を使用して書き込み中にデータを自動的にマスクします。

REST インターフェイス

次のスクリプトを実行して、サンプルログエントリ内の PII をマスクします：

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST \
    "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

MODEL_NAME="qwen3.7-max"

BODY=$(cat <<JSON
{
  "model_name": "$MODEL_NAME",
  "texts": ["担当者の田中太郎 (メール: taro.tanaka@example.com、電話: 090-1234-5678) までお問い合わせください。"],
  "params": {
    "pii_types": ["PERSON", "EMAIL", "PHONE"],
    "mask_char": "*",
    "preserve_length": true,
    "temperature": 0
  }
}
JSON
)

RESPONSE_BODY="$(post_json "/v2/vectordb/ai/pii_mask" "$BODY")"
if command -v jq >/dev/null 2>&1; then
  echo "$RESPONSE_BODY" | jq .
  [ "$(echo "$RESPONSE_BODY" | jq -r '.code // -1')" = "0" ] || exit 1
else
  echo "$RESPONSE_BODY"
fi
# 期待される結果： masked = 「担当者の**** (メール: ***********************、電話: *************) までお問い合わせください。」

Python (コレクション関数)

PyMilvus をインストールした後、プレースホルダーの MILVUS_URI と MILVUS_TOKEN を実際のクラスターアドレスとトークンに置き換えて、スクリプトを実行します。一般的なヘルパー関数は、このサンプルコード内に直接記述されています。

from __future__ import annotations

from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "<yourUsername>:<yourPassword>"

DUMMY_VECTOR_DIM = 2
TEXTTRANSFORM_FUNCTION_TYPE = 9

def texttransform_function_type() -> Any:
    for type_name in ("TEXTTRANSFORM", "TEXT_TRANSFORM", "TextTransform"):
        function_type = getattr(FunctionType, type_name, None)
        if function_type is not None:
            return function_type
    # Alibaba Cloud Milvus は、マネージド拡張機能 (関数型値 9) として TEXTTRANSFORM を提供します。
    # 一部の pymilvus バージョンにはこの enum メンバーがネイティブに含まれていませんが、Function(...) は FunctionType(...) を通じて検証します。
    existing = getattr(FunctionType, "_value2member_map_", {}).get(TEXTTRANSFORM_FUNCTION_TYPE)
    if existing is not None:
        return existing
    extension = int.__new__(FunctionType, TEXTTRANSFORM_FUNCTION_TYPE)
    extension._name_ = "TEXTTRANSFORM"
    extension._value_ = TEXTTRANSFORM_FUNCTION_TYPE
    FunctionType._value2member_map_[TEXTTRANSFORM_FUNCTION_TYPE] = extension
    FunctionType._member_map_["TEXTTRANSFORM"] = extension
    return extension

def add_id(schema: Any) -> None:
    schema.add_field("id", DataType.INT64, is_primary=True)

def add_dummy_vector(schema: Any) -> None:
    schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=DUMMY_VECTOR_DIM)

def run_texttransform_example(*, client, collection_name, input_fields, output_field, function_name, function_params, rows) -> None:
    if client.has_collection(collection_name):
        client.drop_collection(collection_name)
    schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
    add_id(schema)
    for name, data_type, max_length in input_fields:
        field_params = {"max_length": max_length} if max_length is not None else {}
        schema.add_field(name, data_type, **field_params)
    output_name, output_data_type, output_max_length = output_field
    output_params = {"max_length": output_max_length} if output_max_length is not None else {}
    schema.add_field(output_name, output_data_type, **output_params)
    add_dummy_vector(schema)
    schema.add_function(
        Function(
            name=function_name,
            function_type=texttransform_function_type(),
            input_field_names=[name for name, _, _ in input_fields],
            output_field_names=[output_name],
            params=function_params,
        )
    )
    index_params = client.prepare_index_params()
    index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
    client.create_collection(collection_name=collection_name, schema=schema, index_params=index_params)
    client.insert(collection_name, rows)
    client.flush(collection_name)
    fields = [name for name, _, _ in input_fields] + [output_name]
    for row in client.query(collection_name, filter="", output_fields=fields, limit=len(rows)):
        print(row)

MODEL_NAME = "qwen3.7-max"
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

run_texttransform_example(
    client=client,
    collection_name="simple_ai_pii_mask",
    input_fields=[("content", DataType.VARCHAR, 4096)],
    output_field=("masked", DataType.VARCHAR, 4096),
    function_name="mask_pii",
    function_params={"provider": "aliyun_milvus", "model_name": MODEL_NAME, "task": "ai_pii_mask", "pii_types": "PERSON,EMAIL,PHONE", "mask_char": "*", "preserve_length": True, "temperature": 0},
    rows=[{"content": "担当者の田中太郎 (メール: taro.tanaka@example.com、電話: 090-1234-5678) までお問い合わせください。", "dummy_vector": [0.1, 0.2]}],
)
# 期待される結果： masked = 「担当者の**** (メール: ***********************、電話: *************) までお問い合わせください。」

期待される結果： masked は 「担当者の**** (メール: ***********************、電話: *************) までお問い合わせください。」 を返します。 — 名前、メールアドレス、電話番号が元の長さを維持したままマスクされ、データは安全にアナリティクスコレクションに書き込むことができます。