Vector Retrieval Service for Milvus:Bangun guardrail keamanan konten untuk tanya jawab AI dengan Data inspection di Alibaba Cloud Milvus

Langkah 1: Siapkan kode bersama

Kode berikut berisi pengaturan koneksi, wrapper REST, fallback kompatibilitas untuk tipe TEXTTRANSFORM, dan fungsi pembantu yang membuat Koleksi dengan guardrail. Sakelar guardrail adalah satu baris data_inspection di dalam params.

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

# ==================== Pengaturan koneksi ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # Port harus 19530
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = MILVUS_URI
MODEL_NAME = "<your-configured-text-model>"  # Ganti dengan model teks yang sudah dikonfigurasi di Provider

TEXTTRANSFORM_FUNCTION_TYPE = 9

# Identifier kontrak yang dikembalikan oleh Provider saat kebijakan keamanan dipicu.
# Ini adalah satu-satunya dasar andal untuk menentukan apakah konten diblokir.
BLOCK_MARKERS = ("DataInspectionFailed", "inappropriate content")

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

def texttransform_function_type() -> Any:
    """Dapatkan FunctionType untuk TEXTTRANSFORM; tambahkan anggota secara dinamis saat enum tidak ada di versi lama."""
    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]]:
    """Wrapper REST: mengembalikan (http_status, data), dan tetap mencoba mengurai badan respons saat HTTP bukan 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:
    """Buat Koleksi berorientasi tulis dengan guardrail 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)
    # Koleksi harus memiliki setidaknya satu bidang vektor. Contoh ini tidak menjalankan pencarian vektor,
    # sehingga bidang placeholder 2 dimensi memenuhi batasan.
    # Dengan nullable=True dideklarasikan, bidang ini tidak perlu dilewatkan saat insert.
    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,   # <- sakelar guardrail
                "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)

Koleksi harus berisi setidaknya satu bidang vektor, jika tidak pembuatan gagal dengan schema does not contain vector field. Contoh ini tidak menjalankan pencarian vektor, sehingga bidang vektor placeholder 2 dimensi memenuhi batasan dan dideklarasikan nullable=True untuk menghindari keharusan menyediakan nilai pada setiap penulisan.

Langkah 2: Gabungkan hasil intersepsi menjadi tiga disposisi

Ini adalah tantangan teknis yang paling sering diremehkan saat guardrail diterapkan di produksi. Saat kebijakan keamanan dipicu, server dapat mengembalikan error HTTP, kode bisnis non-nol, atau penolakan eksplisit sebagai teks dalam HTTP 200. Oleh karena itu, klien perlu menggabungkan sinyal yang diamati menjadi tiga disposisi yang saling eksklusif:

# ==================== Klasifikasi disposisi tiga keadaan ====================
# Kedua jalur panggilan (eksepsi gRPC / badan respons REST) pertama-tama memeriksa identifier kontrak Provider,
# lalu fallback ke status HTTP dan kode bisnis, sehingga intersepsi yang sama menghasilkan disposisi konsisten
# di kedua jalur.

def classify_grpc(exc: MilvusException) -> str:
    """Gabungkan eksepsi gRPC menjadi disposisi tiga keadaan."""
    msg = str(exc)
    if any(marker in msg for marker in BLOCK_MARKERS):
        return "policy_blocked"      # Kebijakan keamanan dipicu, yang merupakan hasil bisnis yang diharapkan
    return "operational_error"       # Semua hal lain dihitung sebagai kesalahan layanan

def classify_rest(status: int, data: dict[str, Any]) -> str:
    """Gabungkan respons REST menjadi disposisi tiga keadaan."""
    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 dengan struktur normal: model mungkin telah mengembalikan penolakan eksplisit dalam badan.
    # Lapisan protokol tidak dapat memutuskan, sehingga arahkan ke tinjauan manual alih-alih menganggapnya lolos.
    return "manual_review"

classify_rest sengaja konservatif. Saat respons adalah HTTP 200 dengan struktur normal, lapisan protokol saja tidak dapat menjamin bahwa badannya aman, karena model mungkin telah menyematkan penolakan eksplisit di dalamnya. Fungsi ini karenanya mengembalikan manual_review untuk kasus yang tidak dapat diputuskan ini alih-alih melewatinya diam-diam. Pasangkan classify_rest dengan validasi jawaban Anda sendiri sehingga respons yang dapat Anda konfirmasi sebagai jawaban bisnis asli diperlakukan sebagai lolos, dan hanya respons yang tetap tidak dapat diputuskan yang mencapai antrian tinjauan. Inilah yang menjaga konsistensi "Lolos" untuk input normal dalam tabel skenario dengan tujuan hanya mengarahkan sejumlah kecil sampel ambigu ke tinjauan manual.

Mengapa keputusan harus didasarkan pada identifier kontrak Provider alih-alih hanya kode status HTTP. Pengujian memverifikasi bahwa "input tidak sesuai yang diblokir oleh guardrail" dan "kesalahan layanan akibat nama model tidak ada" mengembalikan kombinasi kode status yang sama persis:

Kode status HTTP dan kode bisnis pada dua baris pertama identik, sehingga tidak dapat membedakan "konten diblokir" dari "layanan gagal". Makna operasional keduanya benar-benar berbeda: salah mengklasifikasikan pemblokiran kebijakan sebagai kesalahan layanan mengubah setiap intersepsi keamanan konten menjadi alert kesalahan layanan palsu, yang seiring waktu mengubur kesalahan nyata. Keputusan karenanya harus mengandalkan identifier kontrak DataInspectionFailed dalam badan respons. Untuk alasan yang sama, jangan bergantung pada nilai kode status HTTP tertentu, karena nilainya dapat berubah di berbagai versi gateway.

Jangan gunakan kata kunci seperti "risk", "reject", atau "cannot" untuk menentukan apakah konten diblokir, karena teks bisnis normal juga dapat mengandung kata-kata ini dan menyebabkan pemblokiran palsu. Kata kunci paling-paling hanya sinyal tambahan dan tidak boleh mengubah disposisi akhir.

Langkah 3: Periksa konten sesuai dalam mode input dan output

Mode input memeriksa input pengguna sebelum model dipanggil, dan mode output memeriksa output model sebelum hasil dikembalikan. Konten yang sesuai lolos secara normal.

# ==================== Langkah 3a: mode input, pemeriksaan sebelum permintaan ====================
build_guard_collection(
    "guard_input", "inspect_customer_request", "request", "response",
    "Jawab dalam satu kalimat dengan nada layanan pelanggan: ${request}", "input",
)
client.insert("guard_input", [{"request": "Berapa lama waktu yang dibutuhkan refund untuk tiba setelah disetujui?"}])
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')}")
# Guardrail tidak dipicu, sehingga konten lolos dan model mengembalikan respons secara normal.
# Dengan prompt jailbreak, model tidak pernah dipanggil dan guardrail memblokirnya di awal.

# ==================== Langkah 3b: mode output, pemeriksaan sebelum menerbitkan output model ====================
build_guard_collection(
    "guard_output", "inspect_generated_copy", "draft_request", "publish_copy",
    "Buat ringkasan kampanye keanggotaan yang cocok untuk diterbitkan di aplikasi: ${draft_request}", "output",
)
client.insert("guard_output",
              [{"draft_request": "Buat ringkasan manfaat untuk Hari Anggota dalam 30 kata atau kurang, tanpa janji berlebihan."}])
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')}")
# Output lolos pemeriksaan keamanan dan masuk ke antrian penerbitan tertunda.

Hasil berikut telah diverifikasi dalam pengujian:

Saat mode input dipicu, model tidak dipanggil, sehingga risiko diblokir dan biaya pemanggilan model juga dihemat. Mode output cocok untuk skenario di mana input dipercaya dan satu-satunya kekhawatiran adalah improvisasi model.

Langkah 4: Blokir input tidak sesuai dalam mode both

Gunakan both saat kedua sisi tidak dipercaya. Contoh berikut memverifikasi bahwa guardrail benar-benar berfungsi dengan mengirimkan satu input uji yang jelas tidak sesuai. Jalankan pemeriksaan efektivitas guardrail ini sebelum peluncuran.

Ganti <a clearly non-compliant test input> dengan string uji tidak sesuai nyata (canary) sebelum menjalankan langkah ini. Placeholder yang ditunjukkan di bawah tidak aktif dan tidak memicu guardrail, sehingga Anda tidak dapat mereproduksi hasil intersepsi hingga mengganti dengan nilai nyata.

# ==================== Langkah 4: mode both, pemeriksaan ganda di kedua sisi ====================
# Gunakan satu input uji yang jelas tidak sesuai (canary) untuk memverifikasi bahwa guardrail benar-benar memblokirnya.
UNSAFE_INPUT = "<a clearly non-compliant test input>"

# (Langkah 4-1) Jalur gRPC: guardrail dipicu saat menulis
build_guard_collection("guard_both", "inspect_both", "request", "response",
                       "Jawab pertanyaan berikut: ${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)
    # Tidak ada error yang muncul tidak berarti lolos dengan aman: arahkan konten struktur normal tetapi mencurigakan ke tinjauan manual
    print("Disposisi: manual_review", [r.get("response") for r in rows])
except MilvusException as exc:
    print(f"Disposisi: {classify_grpc(exc)}")
    print(f"Server mengembalikan: {exc.message}")

# (Langkah 4-2) Jalur REST: kirim input yang sama ke antarmuka sinkron
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"Sinyal yang diamati: HTTP={status} provider_code={data.get('code')}")
print(f"Disposisi: {disposition}")

# Arahkan berdasarkan disposisi: pemblokiran kebijakan adalah hasil yang diharapkan dan tidak memerlukan alert operasional;
# hanya kesalahan layanan yang memerlukan alert, dan tidak boleh diulang tanpa batas
if disposition == "policy_blocked":
    pass          # Catat log audit dan kembalikan pemberitahuan kepatuhan kepada pengguna
elif disposition == "manual_review":
    pass          # Kirim ke antrian tinjauan manual
else:
    pass          # Berikan alert ke engineer on-call dan hentikan retry

Pengujian memverifikasi bahwa kedua jalur berhasil memblokir input, dan server mengembalikan identifier kontrak yang konsisten:

code: DataInspectionFailed
message: Input data may contain inappropriate content.
         For details, see: https://www.alibabacloud.com/help/zh/model-studio/error-code#inappropriate-content

Guardrail dipicu di sisi input, penulisan diblokir, dan model tidak menghasilkan konten. Setelah fungsi klasifikasi memproses sinyal, baik jalur gRPC maupun REST mencapai kesimpulan policy_blocked yang sama.

Pengujian juga mengonfirmasi bahwa mode both tidak salah memblokir input bisnis normal: mengirim pertanyaan layanan pelanggan normal ke antarmuka yang sama mengembalikan HTTP 200 dengan jawaban lengkap.

Langkah 5: Catat disposisi dengan aman

Setelah guardrail berjalan di produksi, log harus dapat melokalisasi masalah tanpa menjadi permukaan kebocoran data baru. Ikuti rekomendasi berikut:

• Catat hanya bidang yang dapat dibaca mesin dan telah diredaksi: trace id, tahap pemeriksaan, status HTTP, kode provider, enum disposisi, dan aksi disposisi.

• Potong atau redaksi teks mentah dan informasi sensitif pribadi. Selama tinjauan, ambil data melalui saluran resmi menggunakan trace id alih-alih menyimpan teks mentah di log bisnis.

• Hentikan dan beri alert saat kebijakan dipicu atau terjadi operational_error, dan jangan retry tanpa batas.

• Gabungkan dengan AI_PII_MASK: redaksi sebelum mencatat log untuk meminimalkan permukaan paparan data sensitif.

{
  "trace_id": "req-20260808-abc123",
  "stage": "both",
  "http_status": 500,
  "provider_code": 65535,
  "disposition": "policy_blocked",
  "note": "diblokir oleh pemeriksaan data di sisi input"
}