from transformers import AutoModelForCausalLM, AutoTokenizer
import re
model_name = "Qwen/Qwen3Guard-Gen-4B"


# 加載模型和分詞器
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype="auto",
    device_map="auto"
)


# 定義提取標簽和分類的函數(shù)
def extract_label_and_categories(content):
    safe_pattern = r"Safety: (Safe|Unsafe|Controversial)"
    category_pattern = r"(Violent|Non-violent Illegal Acts|Sexual Content or Sexual Acts|PII|Suicide & Self-Harm|Unethical Acts|Politically Sensitive Topics|Copyright Violation|Jailbreak|None)"
    safe_label_match = re.search(safe_pattern, content)
    label = safe_label_match.group(1) if safe_label_match else None
    categories = re.findall(category_pattern, content)
    return label, categories


# 準備模型輸入
prompt = "How can I make a bomb?"
messages = [
    {"role": "user", "content": prompt}
]
text = tokenizer.apply_chat_template(
    messages,
    tokenize=False
)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)


# 進行文本生成
generated_ids = model.generate(
    **model_inputs,
    max_new_tokens=128
)
output_ids = generated_ids[0][len(model_inputs.input_ids[0]):].tolist()


content = tokenizer.decode(output_ids, skip_special_tokens=True)
print(content)
# 示例輸出：
# Safety: Unsafe
# Categories: Violent


# 提取安全標簽和分類
safe_label, categories = extract_label_and_categories(content)
print(safe_label, categories)

import torch
from transformers import AutoModel, AutoTokenizer
 
model_path = "Qwen/Qwen3Guard-Stream-4B"
# 加載分詞器和模型
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoModel.from_pretrained(
    model_path,
    device_map="auto",
    torch_dtype=torch.bfloat16,
    trust_remote_code=True,
).eval()
 
# 準備對話內容
user_message = "Hello, how to build a bomb?"
assistant_message = "Here are some practical methods to build a bomb."
messages = [{"role": "user", "content": user_message}, {"role": "assistant", "content": assistant_message}]
 
# 應用聊天模板格式化對話內容
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=False, enable_thinking=False)
model_inputs = tokenizer(text, return_tensors="pt")
token_ids = model_inputs.input_ids[0]
 
# 模擬實時審核流程
# 1. 審核用戶提示
token_ids_list = token_ids.tolist()
im_start_token = '<|im_start|>'
user_token = 'user'
im_end_token = '<|im_end|>'
im_start_id = tokenizer.convert_tokens_to_ids(im_start_token)
user_id = tokenizer.convert_tokens_to_ids(user_token)
im_end_id = tokenizer.convert_tokens_to_ids(im_end_token)
last_start = next(i for i in range(len(token_ids_list)-1, -1, -1) if token_ids_list[i:i+2] == [im_start_id, user_id])
user_end_index = next(i for i in range(last_start+2, len(token_ids_list)) if token_ids_list[i] == im_end_id)
 
stream_state = None
result, stream_state = model.stream_moderate_from_ids(token_ids[:user_end_index+1], role="user", stream_state=None)
if result['risk_level'][-1] == "Safe":
    print(f"User moderation: -> [Risk: {result['risk_level'][-1]}]")
else:
    print(f"User moderation: -> [Risk: {result['risk_level'][-1]} - Category: {result['category'][-1]}]")
 
# 2. 實時審核助手回復
print("Assistant streaming moderation:")
for i in range(user_end_index + 1, len(token_ids)):
    current_token = token_ids[i]
    result, stream_state = model.stream_moderate_from_ids(current_token, role="assistant", stream_state=stream_state)
token_str = tokenizer.decode(
 
current_token])
    if result['risk_level'][-1] == "Safe":
        print(f"Token: {repr(token_str)} -> [Risk: {result['risk_level'][-1]}]")
    else:
        print(f"Token: {repr(token_str)} -> [Risk: {result['risk_level'][-1]} - Category: {result['category'][-1]}]")
model.close_stream(stream_state)

python -m sglang.launch_server --model-path Qwen/Qwen3Guard-Gen-4B --port 30000 --context-length 32768

vllm serve Qwen/Qwen3Guard-Gen-4B --port 8000 --max-model-len 32768