Vector Retrieval Service for Milvus:Alibaba Cloud Milvus のデータ検査機能で AI 質疑応答のコンテンツ安全ガードレールを構築

ステップ 1: 共有コードの準備

以下のコードには、接続設定、REST ラッパー、TEXTTRANSFORM 型の互換性フォールバック、およびガードレール付きのコレクションを作成するヘルパー関数が含まれています。ガードレールスイッチは、params 内の data_inspection の 1 行です。

from __future__ import annotations

import json
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from pymilvus import DataType, Function, FunctionType, MilvusClient
from pymilvus.exceptions import MilvusException

# ==================== 接続設定 ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # ポートは 19530 である必要があります
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = MILVUS_URI
MODEL_NAME = "<your-configured-text-model>"  # プロバイダーで設定済みのテキストモデルに置き換えてください

TEXTTRANSFORM_FUNCTION_TYPE = 9

# 安全ポリシーがトリガーされたときにプロバイダーから返される契約識別子。
# コンテンツがブロックされたかどうかを判断するための唯一の信頼できる基準です。
BLOCK_MARKERS = ("DataInspectionFailed", "inappropriate content")

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

def texttransform_function_type() -> Any:
    """TEXTTRANSFORM の FunctionType を取得します。古いバージョンで enum が欠落している場合は、メンバーを動的に追加します。"""
    for type_name in ("TEXTTRANSFORM", "TEXT_TRANSFORM", "TextTransform"):
        ft = getattr(FunctionType, type_name, None)
        if ft is not None:
            return ft
    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 post_json(path: str, body: dict[str, Any], timeout: int = 120) -> tuple[int, dict[str, Any]]:
    """REST ラッパー: (http_status, data) を返し、HTTP が 2xx でない場合でもレスポンスボディの解析を試みます。"""
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        return exc.code, json.loads(exc.read().decode("utf-8"))

def build_guard_collection(name: str, func_name: str, in_field: str, out_field: str,
                           prompt: str, data_inspection: str) -> None:
    """TextTransform ガードレールを持つ書き込み指向のコレクションを作成します。"""
    if client.has_collection(name):
        client.drop_collection(name)

    schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
    schema.add_field("id", DataType.INT64, is_primary=True)
    schema.add_field(in_field, DataType.VARCHAR, max_length=1024)
    schema.add_field(out_field, DataType.VARCHAR, max_length=4096)
    # コレクションには少なくとも 1 つのベクトルフィールドが必要です。この例ではベクトル検索を実行しないため、
    # 2 次元のプレースホルダーフィールドで制約を満たします。
    # nullable=True と宣言されているため、挿入時にフィールドを渡す必要はありません。
    schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
    schema.add_function(
        Function(
            name=func_name,
            function_type=texttransform_function_type(),
            input_field_names=[in_field],
            output_field_names=[out_field],
            params={
                "provider": "aliyun_milvus",
                "model_name": MODEL_NAME,
                "task": "ai_text_generate",
                "prompt": prompt,
                "data_inspection": data_inspection,   # <- ガードレールスイッチ
                "temperature": "0.2",
                "enable_thinking": "false",
                "timeout_sec": "45",
            },
        )
    )
    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=name, schema=schema, index_params=index_params)

Collection には少なくとも1つのベクトルフィールドを含める必要があり、そうでない場合は schema does not contain vector field というエラーで作成に失敗します。この例ではベクトル検索を実行しないため、2次元のプレースホルダーのベクトルフィールドで制約を満たし、書き込みごとに値を渡すのを避けるために nullable=True と宣言されています。

ステップ 2: 傍受結果を 3 つの判定結果に収束させる

これは、ガードレールを本番稼働させる際、最も過小評価されがちな技術的な難点です。安全ポリシーがトリガーされると、サーバーは HTTP エラー、ゼロ以外のビジネスコード、または HTTP 200 内のテキストによる明示的な拒否を返すことがあります。そのため、クライアントは観測されたシグナルを、相互に排他的な 3 つの判定結果に収束させる必要があります。

# ==================== 3 状態の判定結果分類 ====================
# 両方の呼び出しパス (gRPC 例外 / REST レスポンスボディ) は、まずプロバイダーの契約識別子をチェックし、
# 次に HTTP ステータスとビジネスコードにフォールバックするため、同じ傍受でも一貫した
# 判定結果が両方のパスで得られます。

def classify_grpc(exc: MilvusException) -> str:
    """gRPC 例外を 3 状態の判定結果に収束させます。"""
    msg = str(exc)
    if any(marker in msg for marker in BLOCK_MARKERS):
        return "policy_blocked"      # 安全ポリシーがトリガーされました。これは期待されるビジネス結果です
    return "operational_error"       # それ以外はすべてサービス障害と見なします

def classify_rest(status: int, data: dict[str, Any]) -> str:
    """REST レスポンスを 3 状態の判定結果に収束させます。"""
    raw = json.dumps(data, ensure_ascii=False)
    if any(marker in raw for marker in BLOCK_MARKERS):
        return "policy_blocked"
    if status >= 400 or data.get("code", 0) != 0:
        return "operational_error"
    # HTTP 200 で構造が正常な場合: モデルがボディ内で明示的な拒否を返した可能性があります。
    # プロトコルレイヤーでは判断できないため、安全なパスとして扱うのではなく、手動レビューにルーティングします。
    return "manual_review"

classify_rest は意図的に保守的に動作します。応答が正常な構造の HTTP 200 であっても、モデルが本文に明示的な拒否を埋め込んでいる可能性があるため、プロトコル レイヤーだけでは本文が安全であると保証することはできません。そのため、この関数は、このような判定不能なケースに対して、サイレントにパスさせるのではなく、manual_review を返します。classify_rest を独自の回答検証と組み合わせ、正当なビジネス上の回答として確認できる応答をパスとして扱い、判定不能なままの応答のみがレビュー キューに送られるようにします。これにより、シナリオ テーブルの通常の入力に対する「Passed through」は、ごく少数のあいまいなサンプルのみを手動レビューにルーティングするという目標と整合性が保たれます。

なぜ判断が HTTP ステータスコードだけでなく、プロバイダーの契約識別子に基づいている必要があるのか。テストにより、「ガードレールによってブロックされた非準拠の入力」と「存在しないモデル名によって引き起こされたサービス障害」が、まったく同じステータスコードの組み合わせを返すことが確認されました。

最初の 2 行にある HTTP ステータスコードとビジネスコードは同一であるため、「コンテンツのブロック」と「サービス障害」を区別できません。この 2 つは運用上の意味が全く異なります。ポリシーによるブロックをサービス障害として誤って分類すると、すべてのコンテンツ安全性のインターセプトが誤ったサービス障害アラートに変わり、時間とともに実際の障害が埋もれてしまいます。したがって、判断はレスポンスボディ内の DataInspectionFailed コントラクト識別子に依拠する必要があります。同じ理由から、特定の HTTP ステータスコードの値には依存しないでください。ゲートウェイのバージョンによって変わる可能性があるためです。

「リスク」、「拒否」、「できない」などのキーワードを使用してコンテンツがブロックされたかどうかを判断しないでください。通常のビジネステキストにもこれらの単語が含まれている可能性があり、誤ブロックを引き起こす原因となります。キーワードはせいぜい補助的なシグナルであり、最終的な判定結果を変更してはなりません。

ステップ 3: input モードと output モードで準拠コンテンツを検査する

input モードは、モデルが呼び出される前にユーザー入力を検査し、output モードは、結果が返される前にモデルの出力を検査します。準拠したコンテンツは通常どおり通過します。

# ==================== ステップ 3a: input モード、リクエスト前の検査 ====================
build_guard_collection(
    "guard_input", "inspect_customer_request", "request", "response",
    "Answer in one sentence in a customer service tone: ${request}", "input",
)
client.insert("guard_input", [{"request": "How long does a refund take to arrive after it is approved?"}])
client.flush("guard_input")
for row in client.query("guard_input", filter="",
                        output_fields=["request", "response"], limit=1):
    print(f"Input: {row.get('request')}")
    print(f"Output: {row.get('response')}")
# ガードレールはトリガーされなかったため、コンテンツは通過し、モデルは正常に応答を返します。
# 代わりにジェイルブレイクプロンプトを使用した場合、モデルは呼び出されず、ガードレールが事前にブロックします。

# ==================== ステップ 3b: output モード、モデル出力の公開前の検査 ====================
build_guard_collection(
    "guard_output", "inspect_generated_copy", "draft_request", "publish_copy",
    "Generate a membership campaign summary suitable for publishing in an app: ${draft_request}", "output",
)
client.insert("guard_output",
              [{"draft_request": "Generate a benefits summary for Members Day in 30 words or fewer, with no exaggerated promises."}])
client.flush("guard_output")
for row in client.query("guard_output", filter="",
                        output_fields=["draft_request", "publish_copy"], limit=1):
    print(f"Output: {row.get('publish_copy')}")
# 出力は安全検査に合格し、公開待ちキューに入ります。

テストでは以下の結果が検証されました。

input モードがトリガーされると、モデルは呼び出されないため、リスクがブロックされると同時にモデルの呼び出しコストが節約されます。 output モードは、入力が信頼でき、モデルの応答の逸脱のみが懸念されるシナリオに適しています。

ステップ 4: both モードで非準拠入力をブロックする

どちらのエンドも信頼されていない場合は、both を使用してください。次の例では、明らかに非準拠のテスト入力を 1 つ送信することで、ガードレールが実際に機能することを検証します。起動前にこのガードレールの有効性チェックを実行してください。

このステップを実行する前に、<a clearly non-compliant test input> を実際の非準拠のテスト文字列 (カナリア) に置き換えます。以下に示すプレースホルダーは不活性であり、ガードレールをトリガーしません。そのため、実際の値を代入するまで、インターセプト結果を再現することはできません。

# ==================== ステップ 4: both モード、両端での二重検査 ====================
# 明らかに非準拠なテスト入力 (カナリア) を 1 つ使用して、ガードレールが実際にそれをブロックすることを確認します。
UNSAFE_INPUT = "<a clearly non-compliant test input>"

# (ステップ 4-1) gRPC パス: 書き込み時にガードレールがトリガーされる
build_guard_collection("guard_both", "inspect_both", "request", "response",
                       "Answer the following: ${request}", "both")
try:
    client.insert("guard_both", [{"request": UNSAFE_INPUT}])
    client.flush("guard_both")
    rows = client.query("guard_both", filter="",
                        output_fields=["request", "response"], limit=1)
    # エラーが発生しなかったからといって安全にパスしたとは限りません: 構造は正常だが疑わしいコンテンツは手動レビューにルーティングします
    print("Disposition: manual_review", [r.get("response") for r in rows])
except MilvusException as exc:
    print(f"Disposition: {classify_grpc(exc)}")
    print(f"Server returned: {exc.message}")

# (ステップ 4-2) REST パス: 同じ入力を同期インターフェイスに送信する
status, data = post_json("/v2/vectordb/ai/text_generate", {
    "model_name": MODEL_NAME,
    "texts": [UNSAFE_INPUT],
    "params": {"data_inspection": "both"},
})
disposition = classify_rest(status, data)
print(f"Observed signals: HTTP={status} provider_code={data.get('code')}")
print(f"Disposition: {disposition}")

# 判定結果によるルーティング: ポリシーによるブロックは期待される結果であり、運用アラートは不要です。
# サービス障害のみがアラートを必要とし、無期限にリトライしてはなりません
if disposition == "policy_blocked":
    pass          # 監査ログを記録し、ユーザーにコンプライアンス通知を返します
elif disposition == "manual_review":
    pass          # 手動レビューキューに送信します
else:
    pass          # オンコールエンジニアにアラートを出し、リトライを停止します

テストにより、両方のパスが入力のブロックに成功し、サーバーが一貫した契約識別子を返したことが確認されました。

コード: DataInspectionFailed
メッセージ: 入力データに不適切なコンテンツが含まれている可能性があります。
         詳細: https://www.alibabacloud.com/help/zh/model-studio/error-code#inappropriate-content

入力側でガードレールがトリガーされ、書き込みがブロックされ、モデルはコンテンツを生成しませんでした。分類関数が信号を処理すると、gRPC パスと REST パスの両方が同じ policy_blocked という結論に達します。

また、テストにより、both モードが通常の業務入力を誤ってブロックしないことが確認されました。同じインターフェイスに通常のカスタマーサービスの質問を送信したところ、完全な回答とともに HTTP 200 が返されました。

ステップ 5: 判定結果を安全にログに記録する

ガードレールが本番稼働した後は、ログが新たなデータ漏洩面になることなく、問題を特定できなければなりません。以下の推奨事項に従ってください。

• 機械可読で墨塗りされたフィールドのみを記録する：トレース ID、検査ステージ、HTTP ステータス、プロバイダーコード、判定結果の列挙型、判定結果のアクション。

• 生テキストと個人機密情報をトリミングまたは墨塗りする。レビュー中は、ビジネスログに生テキストを保持する代わりに、トレース ID を使用して認可されたチャネルを通じてデータを取得します。

• ポリシーがトリガーされた場合、または operational_error が発生した場合は、停止してアラートを送信し、無期限にリトライは行わないでください。

• AI_PII_MASK と組み合わせ、ログ記録の前に墨消しを行うことで、機密データの露出面を最小限に抑えます。

{
  "trace_id": "req-20260808-abc123",
  "stage": "both",
  "http_status": 500,
  "provider_code": 65535,
  "disposition": "policy_blocked",
  "note": "blocked by data inspection on input side"
}