Vector Retrieval Service for Milvus:Crie guardrails de segurança de conteúdo para Q&A de IA com DATA_INSPECTION no Alibaba Cloud Milvus

Prepare o código comum

O código a seguir inclui a configuração de conexão, um wrapper REST, um fallback de compatibilidade para o tipo TEXTTRANSFORM e uma função utilitária que cria uma Collection com um guardrail. O parâmetro do guardrail é a única linha data_inspection em 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

# ==================== Connection configuration ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # The port must be written as 19530
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = MILVUS_URI
MODEL_NAME = "qwen3.7-max"     # Text model, must already be configured in the Provider

TEXTTRANSFORM_FUNCTION_TYPE = 9

# The contract markers returned by the Provider when a safety policy is triggered;
# they are the only reliable basis for determining "whether it was intercepted"
BLOCK_MARKERS = ("DataInspectionFailed", "inappropriate content")

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

def texttransform_function_type() -> Any:
    """Get the FunctionType for TEXTTRANSFORM; dynamically add a member if the enum is missing in older versions."""
    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 interface wrapper: returns (http_status, data); still tries to parse the response body for non-2xx HTTP responses."""
    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:
    """Create a write-oriented Collection with a TextTransform guardrail."""
    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)
    # A Collection must have at least one vector field; this example does not perform vector retrieval,
    # so a 2-dimensional placeholder field is used to satisfy the constraint.
    # After nullable=True is declared, the field does not need to be passed on 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,   # ← Guardrail switch
                "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)

Converja os resultados de interceptação em três disposições

Esta é a dificuldade de engenharia mais subestimada ao implementar guardrails. Quando uma política de segurança é acionada, o servidor pode retornar um erro HTTP, um código de negócio diferente de zero ou um texto de recusa explícito dentro de uma resposta HTTP 200. Portanto, o cliente precisa unificar os sinais observados em três disposições mutuamente exclusivas:

# ==================== Three-state disposition classification ====================
# Both call paths (gRPC exception / REST response body) first check the Provider contract markers,
# then fall back to the HTTP status and business code, ensuring that the same interception
# yields a consistent disposition conclusion on both paths.

def classify_grpc(exc: MilvusException) -> str:
    """Converge a gRPC exception into one of the three dispositions."""
    msg = str(exc)
    if any(marker in msg for marker in BLOCK_MARKERS):
        return "policy_blocked"      # A safety policy was triggered; an expected business outcome
    return "operational_error"       # Everything else is treated as an operational failure

def classify_rest(status: int, data: dict[str, Any]) -> str:
    """Converge a REST response into one of the three dispositions."""
    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 with a normal structure: this may be an "explicit refusal" given by the model in the body,
    # which cannot be determined at the protocol layer, so always route to manual review
    # instead of treating it directly as having passed the safety check.
    return "manual_review"

Por que avaliar pelos marcadores de contrato do Provider em vez de apenas pelo código de status HTTP. Em testes reais, "uma entrada não conforme interceptada pelo guardrail" e "uma falha de serviço causada por um nome de modelo inexistente" retornaram exatamente a mesma combinação de códigos de status:

Etapas 1 e 2: os modos input e output aprovam conteúdos conformes

O modo input verifica a entrada do usuário antes da chamada do modelo, e o modo output verifica a saída do modelo antes do retorno do resultado. O conteúdo conforme passa normalmente.

# ==================== Step (a): input mode, pre-request check ====================
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 the review 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')}")
# The guardrail was not triggered → passed, and the model returns normally.
# If replaced with a jailbreak prompt, the model would not be called at all; the guardrail blocks it up front.

# ==================== Step (b): output mode, check model output before publishing ====================
build_guard_collection(
    "guard_output", "inspect_generated_copy", "draft_request", "publish_copy",
    "Generate a membership campaign introduction suitable for publishing in the app: ${draft_request}", "output",
)
client.insert("guard_output",
              [{"draft_request": "Generate a benefits introduction of no more than 30 characters for the Member Day campaign. Exaggerated promises are prohibited."}])
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')}")
# The output passed the safety check → it enters the pending-publication queue.

Resultados reais de aprovação:

No modo input, o modelo não é chamado quando o guardrail é acionado, o que bloqueia o risco e reduz o custo da chamada do modelo. O modo output é adequado para cenários em que a entrada é confiável e a única preocupação é a improvisação do modelo.

Etapa 3: o modo both intercepta entradas não conformes

Use both quando nenhum dos lados for confiável. O exemplo a seguir usa uma entrada de teste claramente não conforme; verifique se o guardrail realmente funciona. Esta é uma verificação de eficácia do guardrail que deve ser realizada antes da entrada em produção.

# ==================== Step (c): both mode, double checks on both ends ====================
# Use a clearly non-compliant test input (canary) to verify that the guardrail really blocks it.
UNSAFE_INPUT = "<a clearly non-compliant test input>"

# (c-1) gRPC path: the guardrail is triggered on write
build_guard_collection("guard_both", "inspect_both", "request", "response",
                       "Please answer: ${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)
    # No error thrown does not mean it passed the safety check: if the structure is normal but the content is suspicious, route to manual review
    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}")

# (c-2) REST path: the same input through the real-time interface
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}")

# Route by disposition: a policy interception is an expected outcome and does not require an operations alert;
# only operational failures need alerts, and they should not be retried indefinitely
if disposition == "policy_blocked":
    pass          # Record an audit log and return a compliance notice to the user
elif disposition == "manual_review":
    pass          # Deliver to the manual review queue
else:
    pass          # Alert the on-duty engineer and stop retrying

Em testes reais, ambos os caminhos interceptaram a entrada com sucesso, e o servidor retornou marcadores de contrato consistentes:

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

O guardrail foi acionado no lado da entrada, a gravação foi bloqueada e o modelo não gerou nenhum conteúdo. Após o processamento pelas funções de classificação acima, tanto os caminhos gRPC quanto REST chegam à mesma conclusão de policy_blocked.

Etapa 4: recomendações de logs e disposição

Após a implementação do guardrail, os logs devem ajudar a localizar problemas sem se tornarem uma nova superfície de exposição de dados. Recomendações:

• Registre apenas campos legíveis por máquina e anonimizados: trace ID, estágio de detecção, status HTTP, código do provider, enum de disposição e ação de disposição.

• Sempre oculte ou anonimize o texto original e informações pessoais sensíveis. Durante a revisão, recupere-as por meio de um canal autorizado usando o trace ID e não retenha o texto original nos logs de negócio.

• Interrompa e lance um alerta quando uma política for acionada ou ocorrer um operational_error. Não faça retries indefinidamente.

• Combine com AI_PII_MASK: anonimize antes de registrar nos logs para minimizar a superfície de exposição de dados sensíveis.

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