向量检索服务 Milvus 版：通过阿里云Milvus的DATA_INSPECTION为AI问答构建内容安全护栏

准备公共代码

以下代码包含连接配置、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"
}