Qwen3Guard-Stream 流式实时安全检测 — 生成过程中逐 token 拦截

Stream 版的核心创新

传统安全护栏的做法是等模型生成完毕后再审核——如果输出有害内容，用户已经看到了。Qwen3Guard-Stream 的创新是逐 token 实时检测：

主模型开始生成 token
每个 token 都传给 Stream 版判断
如果检测到有害内容，立即截断生成
用户看到的永远是安全的内容

架构示意

用户输入 → 主模型生成 ──→ [Stream Guard 逐 token 检测] ──→ 用户
                                    │
                                    ├─ 安全 → 继续生成
                                    └─ 有害 → 截断 + 替换为安全回复

优势

即时响应：有害内容生成 1-2 个 token 就被截断，用户看不到
适配性强：可以和任何主模型配合（不限于 Qwen3）
效率高：Guard 模型可以和主模型并行推理

集成方式

代理层（Proxy）

在主模型前面加一个安全代理：

class SafeModelProxy:
    def __init__(self, main_model, guard_model):
        self.main = main_model
        self.guard = guard_model
    
    def safe_generate(self, prompt):
        safe_output = []
        for token in self.main.generate_tokens(prompt):
            # Stream Guard 实时检测
            if self.guard.is_safe(token, safe_output):
                safe_output.append(token)
            else:
                # 有害内容，截断并返回安全回复
                safe_output.append("（内容已被安全拦截）")
                break
        return "".join(safe_output)

SGLang / vLLM 集成

部分推理框架支持在生成管线中直接集成安全护栏，减少中间环节的开销。

精度与延迟的权衡

逐 token 检测会带来额外开销。实际部署时需要权衡：

Guard 模型规格：0.6B 开销最小但精度略低，4B 平衡，8B 精度最高
检测策略：可以每 N 个 token 检测一次，或者只对高危类别做细粒度检测
并行推理：让 Guard 和主模型并行跑，减少串行延迟

适用场景

场景	推荐
在线聊天机器人	Stream（必须）
AI 客服	Stream（必须）
离线批量处理	Gen（Stream 没必要）
RLHF 训练	Gen（奖励信号）

限制

提前截断：Guard 只能检测已生成的内容，无法预见还没生成的。极端情况下有害内容可能生成 1-2 个 token 才被检测到。
精度天花板：Guard 模型本身也不是完美的，可能有漏报（false negative）和误报（false positive）。
延迟：Stream 版会增加每个 token 的生成延迟。

相关阅读