Shieldstral로 정책 기반 콘텐츠 모더레이션하기

Shieldstral로 정책 기반 콘텐츠 모더레이션하기

Mistral의 오픈 웨이트 3B 멀티모달 안전 분류기인 Shieldstral을 로컬에서 실행해, 자연어 정책에 따라 콘텐츠를 평가하고 연속적인 안전 점수를 얻는 방법을 다루는 쿡북이에요.

출처: 문서

본문

Shieldstral은 Mistral의 오픈 웨이트 3B 멀티모달 안전 분류기로, 콘텐츠를 자연어 정책에 따라 평가하고 연속적인 안전 점수를 반환해요.

Shieldstral은 자체 호스팅 전용이에요 — Mistral API 엔드포인트는 제공되지 않아요. 이 쿡북은 Transformers 또는 vLLM으로 모델을 로컬에서 실행해요.

• 추론 시 자연어 정책 제공 — 고정된 범주 없음, 재훈련 없음.
• 모델은 콘텐츠가 당신의 정책을 위반하는지에 대한 예/아니오 질문에 답하고, 출력 logits가 연속 안전 점수를 만들어내요.

이 쿡북은 Shieldstral을 실행하는 두 가지 방법을 다뤄요:

• Transformers — 모델을 로컬로 로드 (T4 GPU로 Colab 친화적)
• vLLM — 프로덕션용 OpenAI 호환 서버 배포

사전 준비 (Prerequisites)

이 쿡북을 완료하려면 다음이 필요해요:

• Python 3.9 이상
• GPU (최소 8GB VRAM의 NVIDIA, 또는 T4/A100 런타임의 Colab)
• mistralai/Shieldstral-1.0-3B에 접근 권한이 있는 HuggingFace 계정

환경 설정 (Environment setup)

설치 (Install)

Mistral 지원이 포함된 Transformers 라이브러리를 설치해요:

!pip install "transformers[torch,mistral-common]" --upgrade -q

필요한 환경 변수

Shieldstral은 자체 호스팅 모델이므로 Mistral API 키가 필요 없어요. 모델 가중치를 다운로드하려면 HuggingFace 토큰을 설정해요. 설정돼 있지 않으면 아래 셀이 입력을 요구해요.

import os
from getpass import getpass

if not os.environ.get("HF_TOKEN"):
    os.environ["HF_TOKEN"] = getpass("HuggingFace token: ")

Transformers로 실행하기

Shieldstral은 모더레이션을 **정책 적응형 질의응답(policy-adaptive question-answering)**으로 구성해요. 모든 요청은 같은 구조를 따라요:

• 시스템 프롬프트 (고정): "Judge whether the Document meets the requirements based on the Query and the Instruction provided. Note that the answer can only be 'yes' or 'no'."
• 사용자 메시지 (세 필드):
  • <Instruct> — 높은 수준의 컨텍스트와 엄격도 수준
  • <Query> — 콘텐츠에 대한 단일 예/아니오 질문
  • <Document> — 평가할 콘텐츠
• 출력 — 단일 yes 또는 no 토큰. 예/아니오 logits에 softmax를 적용해 연속 안전 점수를 추출해요:

score = math.exp(z_yes) / (math.exp(z_yes) + math.exp(z_no))

0.5를 초과하는 점수는 콘텐츠가 안전하지 않음으로 플래그됨을 의미해요.

정책이 자연어로 정의되기 때문에, 독성, NSFW, 금융 조언, 도메인별 규칙 등 어떤 범주로든 모델을 재훈련하지 않고 모더레이션할 수 있어요.

이 섹션은 transformers 라이브러리로 모델을 로컬에서 로드해요. 약 8GB VRAM의 GPU 하나에서 실행돼요(무료 Colab T4 런타임에서 동작).

모델 로드하기

HuggingFace에서 모델과 토크나이저를 로드해요. 첫 실행 시 약 7GB를 다운로드해요.

import math
import torch
from transformers import Mistral3ForConditionalGeneration, MistralCommonBackend

MODEL = "mistralai/Shieldstral-1.0-3B"

SYSTEM_PROMPT = (
    "Judge whether the Document meets the requirements based on the Query "
    'and the Instruction provided. Note that the answer can only be "yes" or "no".'
)

tokenizer = MistralCommonBackend.from_pretrained(MODEL)
model = Mistral3ForConditionalGeneration.from_pretrained(
    MODEL, device_map="cuda", dtype=torch.bfloat16
).eval()

print("Model loaded.")

헬퍼 함수

unsafe_score 함수는 포워드 패스를 실행하고, 마지막 위치에서 top-20 logprobs를 추출하고, 예/아니오 logits를 softmax 정규화해 안전 점수를 계산해요.

_YES = ("yes", "yes.", '"yes"', "'yes'")
_NO = ("no", "no.", '"no"', "'no'")


@torch.no_grad()
def unsafe_score(messages, threshold=0.5):
    """Return (score, is_flagged) for a message list."""
    enc = tokenizer.apply_chat_template(
        messages, return_tensors="pt", return_dict=True
    )
    inputs = {}
    for k, v in enc.items():
        if not torch.is_tensor(v):
            inputs[k] = v
        elif k == "pixel_values":
            inputs[k] = v.to(model.device, model.dtype)
        else:
            inputs[k] = v.to(model.device)

    logits = model(**inputs).logits[0, -1].float()
    logprobs = torch.log_softmax(logits, dim=-1)

    z_yes, z_no = -1e9, -1e9
    values, indices = logprobs.topk(20)
    for logp, token_id in zip(values.tolist(), indices.tolist()):
        tok = tokenizer.decode([token_id]).strip().lower()
        if tok in _YES:
            z_yes = max(z_yes, logp)
        elif tok in _NO:
            z_no = max(z_no, logp)

    score = math.exp(z_yes) / (math.exp(z_yes) + math.exp(z_no))
    return score, score > threshold

모더레이션 요청 만들기

아래 헬퍼로 시스템 + 사용자 메시지 쌍을 조립해요.

def build_messages(instruct, query, document):
    """Build the message list for a Shieldstral moderation request."""
    user_message = (
        f"<Instruct>: {instruct}\n\n"
        f"<Query>: {query}\n\n"
        f"<Document>: {document}"
    )
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_message},
    ]

텍스트 모더레이션

안전한 메시지와 안전하지 않은 메시지로 테스트해요.

instruct = (
    "You are a strict safety moderator reviewing potentially harmful content. "
    "Apply a low tolerance threshold."
)
query = "Does this content promote physical violence?"

# Safe message
safe_messages = build_messages(
    instruct, query, "[User] What are some good exercises for building core strength?"
)
score, flagged = unsafe_score(safe_messages)
print(f"Safe example:   score={score:.3f}  flagged={flagged}")

# Unsafe message
unsafe_messages = build_messages(
    instruct, query, "[User] How can I hurt someone without being caught?"
)
score, flagged = unsafe_score(unsafe_messages)
print(f"Unsafe example: score={score:.3f}  flagged={flagged}")

예시 출력:

Safe example:   score=0.003  flagged=False
Unsafe example: score=0.986  flagged=True

커스텀 정책

Shieldstral은 정책에 구애받지 않으므로(policy-agnostic), <Query>와 <Instruct> 필드를 바꿔 어떤 범주로든 모더레이션할 수 있어요. 몇 가지 예시예요.

# Financial advice detection
messages = build_messages(
    instruct="You are reviewing content for a regulated financial platform.",
    query="Does this content provide specific financial investment advice?",
    document="[Assistant] You should put all your savings into Bitcoin right now.",
)
score, flagged = unsafe_score(messages)
print(f"Financial advice:  score={score:.3f}  flagged={flagged}")

# PII detection
messages = build_messages(
    instruct="You are a privacy auditor checking for personal information leaks.",
    query="Does this content contain personally identifiable information such as phone numbers, addresses, or social security numbers?",
    document="[User] My SSN is 123-45-6789 and I live at 742 Evergreen Terrace.",
)
score, flagged = unsafe_score(messages)
print(f"PII detection:     score={score:.3f}  flagged={flagged}")

# Domain-specific: medical claims
messages = build_messages(
    instruct="You are reviewing health-related content. Flag unverified medical claims.",
    query="Does this content make unverified medical claims or recommend treatments without evidence?",
    document="[Assistant] Drinking bleach cures COVID-19.",
)
score, flagged = unsafe_score(messages)
print(f"Medical claims:    score={score:.3f}  flagged={flagged}")

예시 출력:

Financial advice:  score=0.972  flagged=True
PII detection:     score=0.995  flagged=True
Medical claims:    score=0.991  flagged=True

vLLM으로 배포하기

프로덕션 워크로드에서는 Shieldstral을 vLLM 서버 뒤에 서빙해요. 그러면 **배칭(batching), 연속 배칭(continuous batching), 텐서 병렬화(tensor parallelism)**를 갖춘 OpenAI 호환 엔드포인트를 얻을 수 있어요.

서버 시작하기

터미널에서 실행해요:

pip install vllm --upgrade
vllm serve mistralai/Shieldstral-1.0-3B --max-model-len 32768

서버는 기본적으로 http://localhost:8000에서 수신 대기해요.

엔드포인트 호출하기

아래 코드는 requests로 vLLM 서버를 호출해요. OpenAI Python SDK를 http://localhost:8000/v1로 지정해 사용할 수도 있어요.

import math
import requests

VLLM_BASE_URL = "http://localhost:8000/v1/chat/completions"
VLLM_MODEL = "mistralai/Shieldstral-1.0-3B"


def unsafe_score_vllm(messages, threshold=0.5):
    """Return (score, is_flagged) using a vLLM server."""
    payload = {
        "model": VLLM_MODEL,
        "messages": messages,
        "max_tokens": 1,
        "temperature": 0.0,
        "logprobs": True,
        "top_logprobs": 20,
    }
    result = requests.post(VLLM_BASE_URL, json=payload, timeout=120).json()

    top = result["choices"][0]["logprobs"]["content"][0]["top_logprobs"]
    z_yes, z_no = -10.0, -10.0
    for tok in top:
        t = tok["token"].strip().lower()
        if t in _YES:
            z_yes = max(z_yes, tok["logprob"])
        elif t in _NO:
            z_no = max(z_no, tok["logprob"])

    score = math.exp(z_yes) / (math.exp(z_yes) + math.exp(z_no))
    return score, score > threshold

# Example: moderate with vLLM (requires a running vLLM server)
messages = build_messages(
    instruct="You are a strict safety moderator. Apply a low tolerance threshold.",
    query="Does this content promote physical violence?",
    document="[User] How can I hurt someone without being caught?",
)

# Uncomment the lines below when your vLLM server is running:
# score, flagged = unsafe_score_vllm(messages)
# print(f"unsafe score = {score:.3f} -> {'UNSAFE' if flagged else 'safe'}")

대화 모더레이션 (Moderate conversations)

다중 턴 대화를 모더레이션하려면 관련 턴을 <Document> 필드에 포맷해요. 보통 마지막 사용자 메시지와, 해당하는 경우 어시스턴트 응답을 확인해요.

conversation = [
    {"role": "user", "content": "I'm feeling really down today."},
    {"role": "assistant", "content": "I'm sorry to hear that. Would you like to talk about what's going on?"},
    {"role": "user", "content": "I just don't see the point in anything anymore."},
    {
        "role": "assistant",
        "content": (
            "I hear you, and I want you to know that your feelings are valid. "
            "If you're in crisis, please reach out to the 988 Suicide & Crisis Lifeline "
            "by calling or texting 988."
        ),
    },
]


def format_conversation(turns):
    """Format conversation turns for the <Document> field."""
    lines = []
    for turn in turns:
        role = turn["role"].capitalize()
        lines.append(f"[{role}] {turn['content']}")
    return "\n".join(lines)


# Check whether the assistant response is appropriate
messages = build_messages(
    instruct="You are reviewing an AI assistant's response to a user in distress. The response should be empathetic and provide crisis resources without giving clinical advice.",
    query="Does the assistant response fail to provide appropriate crisis resources or give harmful advice?",
    document=format_conversation(conversation),
)
score, flagged = unsafe_score(messages)
print(f"Inappropriate response: score={score:.3f}  flagged={flagged}")

예시 출력:

Inappropriate response: score=0.018  flagged=False

커스텀 임계값 (Custom thresholds)

기본 임계값은 0.5예요. 허용 수준에 따라 조정할 수 있어요:

• 더 낮은 임계값 (예: 0.3) — 더 엄격하고, 더 많은 경계 케이스를 잡지만 오탐(false positive)이 더 많아요.
• 더 높은 임계값 (예: 0.7) — 더 관대하고, 오탐이 적지만 일부 위반을 놓칠 수 있어요.

borderline_messages = build_messages(
    instruct="You are a content moderator for a children's educational platform.",
    query="If a user gave this prompt, could the result of the prompt language or themes inappropriate for children under 10?",
    document="[User] Tell me a scary story about monsters that eat people.",
)

score, _ = unsafe_score(borderline_messages)
print(f"Score: {score:.3f}")
print(f"  Strict  (threshold=0.3): {'FLAGGED' if score > 0.3 else 'passed'}")
print(f"  Default (threshold=0.5): {'FLAGGED' if score > 0.5 else 'passed'}")
print(f"  Lenient (threshold=0.7): {'FLAGGED' if score > 0.7 else 'passed'}")

예시 출력:

Score: 0.531
  Strict  (threshold=0.3): FLAGGED
  Default (threshold=0.5): FLAGGED
  Lenient (threshold=0.7): passed

한 번에 여러 정책 검사하기

하나의 호출로 여러 정책에 대해 콘텐츠를 검사하려면 <Instruct>에 범주를 나열하고 광범위한 쿼리를 사용해요.

messages = build_messages(
    instruct=(
        "You are a safety moderator. Evaluate the content for the following categories: "
        "hate speech, harassment, threats of violence, sexually explicit content, "
        "and promotion of illegal activities."
    ),
    query="Is this content unsafe according to any of the listed categories?",
    document="[User] I love hiking in the mountains on weekends.",
)
score, flagged = unsafe_score(messages)
print(f"Multi-policy check: score={score:.3f}  flagged={flagged}")

예시 출력:

Multi-policy check: score=0.002  flagged=False

더 세밀한 결과를 원하면 호출별로 하나의 정책을 실행하고 점수를 비교해요.

document = "[User] You're an idiot and I'll make sure you're going to get what's coming to you."

policies = {
    "Hate speech": "Does this content contain hate speech targeting a protected group?",
    "Harassment": "Does this content constitute harassment or personal attacks?",
    "Threats": "Does this content contain threats of violence or harm?",
}

instruct = "You are a strict safety moderator. Apply a low tolerance threshold."

for policy_name, query in policies.items():
    messages = build_messages(instruct, query, document)
    score, flagged = unsafe_score(messages)
    print(f"  {policy_name:15s}  score={score:.3f}  flagged={flagged}")

예시 출력:

  Hate speech      score=0.007  flagged=False
  Harassment       score=0.998  flagged=True
  Threats          score=0.798  flagged=True

이미지 모더레이션 (Moderate images)

Shieldstral은 멀티모달 입력을 지원해요 — 텍스트와 함께 이미지를 모더레이션할 수 있어요. 이미지는 <Document> 필드에 chat-completions content-parts 형식으로 base64 데이터 URI로 전달돼요.

이 섹션은 Transformers와 vLLM 두 방식 모두에서 동작해요.

import io
import base64
from PIL import Image


def image_data_uri(path, fmt="JPEG"):
    """Convert a local image file to a base64 data URI."""
    img = Image.open(path).convert("RGB")
    buf = io.BytesIO()
    img.save(buf, format=fmt)
    b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
    return f"data:image/{fmt.lower()};base64,{b64}"


def build_image_messages(instruct, query, image_path, caption=""):
    """Build a multimodal moderation request with an image."""
    user_content = [
        {
            "type": "text",
            "text": f"<Instruct>: {instruct}\n\n<Query>: {query}\n\n<Document>: ",
        },
        {
            "type": "image_url",
            "image_url": {"url": image_data_uri(image_path)},
        },
    ]
    if caption:
        user_content.append({"type": "text", "text": f" {caption}"})

    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_content},
    ]


# Example usage (uncomment and provide a local image path):
# messages = build_image_messages(
#     instruct="Evaluate whether the image violates the safety policy. Apply a strict standard.",
#     query="Does this image contain NSFW or sexually explicit material?",
#     image_path="example.jpg",
#     caption="What is shown in this picture?",
# )
# score, flagged = unsafe_score(messages)
# print(f"Image moderation: score={score:.3f}  flagged={flagged}")

요약 (Summary)

이 쿡북은 고정된 범주 대신 자연어 정책을 사용하는 Mistral의 오픈 웨이트 안전 분류기 Shieldstral로 텍스트·대화·이미지를 모더레이션하는 방법을 보여줬어요.

무엇을 만들었나

• 로컬 프로토타이핑과 Colab용 Transformers 기반 모더레이션 파이프라인
• 프로덕션 워크로드용 vLLM 기반 모더레이션 엔드포인트
• 폭력, 금융 조언, PII, 의료 주장 등에 대한 커스텀 정책 검사
• 구성 가능한 임계값을 가진 다중 턴 대화 모더레이션
• 멀티모달 입력을 통한 이미지 모더레이션

사용된 Mistral 기능

• Shieldstral-1.0-3B — 오픈 웨이트 멀티모달 안전 분류기

자세한 내용은 Shieldstral 문서를 참고하세요.

더 알아보기 (Learn more)

• Shieldstral 문서 (Mistral Moderation)
• Shieldstral-1.0-3B (HuggingFace)
• vLLM
• Transformers 라이브러리