Vector Retrieval Service for Milvus：通過阿里雲Milvus的DATA_INSPECTION為AI問答構建Alibaba Content Security Service護欄

準備公用代碼

以下程式碼封裝含串連配置、REST 封裝、TEXTTRANSFORM 類型相容兜底，以及一個建帶護欄 Collection 的工具函數。護欄開關就是 params 裡那一行 data_inspection。

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 = "qwen3.7-max"     # 文本模型，須已在 Provider 中配置

TEXTTRANSFORM_FUNCTION_TYPE = 9

# Provider 命中安全性原則時返回的契約標識，是判斷「是否被攔」的唯一可靠依據
BLOCK_MARKERS = ("DataInspectionFailed", "inappropriate content")

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


def texttransform_function_type() -> Any:
    """取 TEXTTRANSFORM 的 FunctionType；老版本枚舉缺失時動態補一個成員。"""
    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 護欄的寫入型 Collection。"""
    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)
    # Collection 必須至少有一個向量欄位；本例不做向量檢索，用 2 維佔位欄位滿足約束。
    # 聲明 nullable=True 後，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,   # ← 護欄開關
                "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)

把攔截結果收斂為三種處置

這是護欄落地時最容易被低估的工程痛點。命中安全性原則時，服務端可能返回 HTTP 錯誤、非零業務碼，也可能在 HTTP 200 裡返回一段明確拒答的文本。因此用戶端需要把觀察到的訊號統一收斂為三種互斥處置：

# ==================== 三態處置分類 ====================
# 兩條調用路徑（gRPC 異常 / REST 響應體）都先按 Provider 契約標識判斷，
# 再按 HTTP 狀態與業務碼兜底，確保同一次攔截在兩條路徑上得到一致的處置結論。

def classify_grpc(exc: MilvusException) -> str:
    """把 gRPC 異常收斂為三態處置。"""
    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 響應收斂為三態處置。"""
    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"

為什麼必須按 Provider 契約標識判斷，而不能只看 HTTP 狀態代碼。實測中「一條違規輸入被護欄攔截」與「模型名不存在導致的服務故障」返回了完全相同的狀態代碼組合：

步驟一與步驟二：input 與 output 模式允許存取合規內容

input 模式在模型調用前檢查使用者輸入，output 模式在結果返回前檢查模型輸出。合規內容正常允許存取。

# ==================== 步驟 (a)：input 模式，請求前置檢查 ====================
build_guard_collection(
    "guard_input", "inspect_customer_request", "request", "response",
    "請以客服口吻用一句話回答：${request}", "input",
)
client.insert("guard_input", [{"request": "退款審核後多久到賬？"}])
client.flush("guard_input")
for row in client.query("guard_input", filter="",
                        output_fields=["request", "response"], limit=1):
    print(f"輸入: {row.get('request')}")
    print(f"輸出: {row.get('response')}")
# 護欄未命中 → 允許存取，模型正常返回。
# 若換成越獄 prompt，模型根本不會被調用，護欄在最前面就擋下了。


# ==================== 步驟 (b)：output 模式，發布前檢查模型輸出 ====================
build_guard_collection(
    "guard_output", "inspect_generated_copy", "draft_request", "publish_copy",
    "請產生一條適合 App 發布的會員活動簡介：${draft_request}", "output",
)
client.insert("guard_output",
              [{"draft_request": "為會員日活動產生一條不超過30字的權益簡介，禁止誇大承諾。"}])
client.flush("guard_output")
for row in client.query("guard_output", filter="",
                        output_fields=["draft_request", "publish_copy"], limit=1):
    print(f"輸出: {row.get('publish_copy')}")
# 輸出通過安全檢查 → 進入待發布隊列。

實測允許存取結果：

input 模式命中時模型不會被調用，因此既攔住了風險也節省了模型調用開銷；output 模式適用於輸入可信、只擔心模型自由發揮的情境。

步驟三：both 模式攔截違規輸入

兩端都不可信時用 both。下面用一條明確違規的測試輸入驗證護欄是否真的生效——這是上線前應當做的護欄有效性驗證。

# ==================== 步驟 (c)：both 模式，兩端雙重檢查 ====================
# 用一條明確違規的測試輸入（canary）驗證護欄是否真的攔得住。
UNSAFE_INPUT = "<一條明確違規的測試輸入>"

# (c-1) gRPC 路徑：寫入時觸發護欄
build_guard_collection("guard_both", "inspect_both", "request", "response",
                       "請回答：${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("處置: manual_review", [r.get("response") for r in rows])
except MilvusException as exc:
    print(f"處置: {classify_grpc(exc)}")
    print(f"服務端返回: {exc.message}")

# (c-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"觀察訊號: HTTP={status} provider_code={data.get('code')}")
print(f"處置: {disposition}")

# 按處置分流：策略攔截屬預期結果，不必警示營運；營運故障才需要警示且不應無限重試
if disposition == "policy_blocked":
    pass          # 記錄審計日誌，向使用者返回合規提示
elif disposition == "manual_review":
    pass          # 投遞到人工複核隊列
else:
    pass          # 警示值班，停止重試

實測兩條路徑都成功攔截，服務端返回一致的契約標識：

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

護欄在輸入側即命中，寫入被阻斷，模型未產生任何內容。經上面的分類函數處理後，gRPC 與 REST 兩條路徑都得到 policy_blocked 的一致結論。

步驟四：日誌與處置建議

護欄落地後，日誌既要能定位問題，又不能成為新的資料泄露面。建議：

• 只記錄可機讀、已脫敏的欄位：trace id、檢測階段、HTTP 狀態、provider code、處置枚舉、處置動作。

• 原文與個人敏感資訊一律刪減或脫敏；複核時憑 trace 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"
}