模型窃取防护趋势：水印、限流与行为指纹的协同-尧图网络科技

模型窃取防护趋势：水印、限流与行为指纹的协同

一、单点限流为何挡不住模型窃取

模型窃取直接威胁 AI 资产。攻击者通过高频调用模型 API，收集输入-输出对，再蒸馏出一个功能等价的替代模型。一旦替代模型成型，原模型的训练成本、数据壁垒、商业护城河全部归零。

最早的防护手段是限流。按用户、按 IP、按 token 设置速率上限，让攻击者无法在合理时间内拿到足够样本。这套思路在 Web 时代有效，但在大模型时代很快遇到瓶颈。

合法用户也会触发限流。一个高并发的应用接入方，正常流量就可能超过单租户上限。攻击者则用代理池、多账号、分布式 IP 把请求摊薄，让限流器看到的每个来源都在阈值内。限流挡得住暴力提取，挡不住低速持续抽取。

更麻烦的是事后取证缺位。即便限流拦下了部分请求，被拿走的样本也无法追溯。模型一旦被蒸馏发布，原厂商几乎无法在法庭或技术上证明"这个模型是从我这里偷的"。归属问题不解决，防护就只剩一半。

防护范式转向协同。水印负责事后取证与归属，限流负责实时拦截暴力提取，行为指纹负责识别"看起来合法但模式异常"的低速窃取。三者覆盖不同时间尺度与威胁形态，协同形成完整链路。

二、三手段协同的防护架构

三种手段的能力边界各不相同。限流是实时门，作用在请求入口；水印是事后证据，嵌入在响应里；行为指纹是异步判别，作用在跨请求聚合层。三者互补。

水印分两类。训练水印在模型权重里嵌入触发样本，验证时用特定输入激发特征响应；推理水印在输出文本里嵌入统计扰动或词汇偏好，事后可从提取的替代模型上检测。前者对权重窃取有效，后者对 API 蒸馏有效。

行为指纹关注调用模式。同一身份下，正常用户的请求长度、主题分布、时间间隔有规律；窃取攻击者通常表现为高度均匀的请求模板、稳定的 token 长度、近乎机械的时间间隔。这些信号单看每条都合法，聚合后才能识别。

异步与同步的分工很重要。限流与水印嵌入必须在同步链路，否则会拖慢响应；行为指纹放在异步聚合层，不阻塞请求，只做事后判定与降级。三者通过统一的证据链关联，一旦行为指纹判定异常，可回溯该身份的水印响应做取证。

三、生产级协同防护网关

下面是一段协同防护网关的核心实现。它把限流、水印嵌入、行为指纹聚合串成一条链路，含并发安全、超时与降级：

import asyncio import hashlib import time from collections import defaultdict, deque from dataclasses import dataclass, field @dataclass class RequestContext: user_id: str prompt: str timestamp: int = field(default_factory=lambda: time.time_ns()) class TokenBucket: # 令牌桶限流：协程安全，按 user_id 隔离 def __init__(self, rate: float, capacity: int): self._rate = rate self._capacity = capacity self._buckets: dict[str, tuple[float, float]] = {} self._lock = asyncio.Lock() async def allow(self, user_id: str) -> bool: async with self._lock: now = time.monotonic() tokens, last = self._buckets.get(user_id, (self._capacity, now)) elapsed = now - last tokens = min(self._capacity, tokens + elapsed * self._rate) if tokens < 1.0: self._buckets[user_id] = (tokens, now) return False tokens -= 1.0 self._buckets[user_id] = (tokens, now) return True class WatermarkEmbedder: # 推理水印：在输出末尾以稳定概率插入零宽字符或词汇偏好 # 这里用占位逻辑演示嵌入位置；真实实现需语言学无害化处理 def __init__(self, marker_rate: float = 0.3): self._marker_rate = marker_rate self._counter = 0 def embed(self, text: str, user_id: str) -> str: # 用 user_id 派生标记位，便于事后归属 self._counter += 1 tag = hashlib.sha1(f"{user_id}|{self._counter}".encode()).hexdigest()[:4] # 占位：实际嵌入需保证语义不变，例如词汇替换或零宽字符 return f"{text}\n<!--wm:{tag}-->" class BehaviorFingerprint: # 行为指纹：维护每个用户的请求间隔与长度分布，异常即标记 def __init__(self, window: int = 100): self._window = window self._history: dict[str, deque] = defaultdict(lambda: deque(maxlen=window)) self._lock = asyncio.Lock() async def observe(self, ctx: RequestContext) -> bool: async with self._lock: hist = self._history[ctx.user_id] hist.append({"ts": ctx.timestamp, "len": len(ctx.prompt)}) if len(hist) < 20: return False # 简化判定：请求长度方差过低、间隔方差过低，判定为机械模式 lengths = [h["len"] for h in hist] intervals = [ hist[i]["ts"] - hist[i - 1]["ts"] for i in range(1, len(hist)) ] len_var = self._variance(lengths) int_var = self._variance(intervals) return len_var < 5.0 and int_var < 1e10 @staticmethod def _variance(seq: list[float]) -> float: if not seq: return 0.0 mean = sum(seq) / len(seq) return sum((x - mean) ** 2 for x in seq) / len(seq) class ProtectionGateway: def __init__(self): self._limiter = TokenBucket(rate=2.0, capacity=10) self._watermark = WatermarkEmbedder() self._fingerprint = BehaviorFingerprint() self._suspicious: set[str] = set() async def handle(self, ctx: RequestContext) -> dict: # 限流为硬门，超阈值直接拒绝 if not await self._limiter.allow(ctx.user_id): return {"status": "rate_limited", "user_id": ctx.user_id} # 异步触发行为指纹聚合，不阻塞主链路 asyncio.create_task(self._observe(ctx)) # 模型推理（占位）+ 水印嵌入 try: raw = await asyncio.wait_for(self._infer(ctx.prompt), timeout=5.0) except asyncio.TimeoutError: return {"status": "timeout", "user_id": ctx.user_id} marked = self._watermark.embed(raw, ctx.user_id) return { "status": "ok", "user_id": ctx.user_id, "suspicious": ctx.user_id in self._suspicious, "output": marked, } async def _observe(self, ctx: RequestContext): try: if await self._fingerprint.observe(ctx): self._suspicious.add(ctx.user_id) except Exception: # 指纹聚合失败不能影响主链路 pass async def _infer(self, prompt: str) -> str: # 占位：实际调用模型推理 await asyncio.sleep(0.05) return f"response_for: {prompt[:16]}" # 使用示例 async def demo(): gw = ProtectionGateway() ctx = RequestContext(user_id="u_001", prompt="请总结这段文本") print(await gw.handle(ctx))

令牌桶用 asyncio.Lock 保证并发安全；水印嵌入在同步链路里完成，保证每条响应都可归属；行为指纹放异步任务，单次失败不污染主链路；异常身份只标记不立即封禁，避免误伤合法高并发用户。

四、协同防护的权衡：鲁棒性、误报与延迟

三种手段协同后，会带来新的权衡，要正面应对。

水印的鲁棒性是最头疼的问题。推理水印要在被蒸馏、被重写、被剪枝后仍可检测。这意味着嵌入强度要够，但又不能让输出质量明显下降。低强度水印容易被攻击者用文本归一化洗掉；高强度水印会影响生成质量与可读性。落地时要按"水印可检测率 vs 用户感知度"做量化权衡，而不是拍一个固定强度。这个没有完美方案，只能根据业务场景选折中点。

行为指纹的误报是体验杀手。合法的高频调用方（如客服机器人、批量处理任务）天然呈现机械模式，会被误判为窃取。解决办法是把判定阈值按业务画像分层，对白名单接入方放宽规则，对新身份从严。同时，异常判定只触发降级（如降低温度、限制敏感能力），不直接封禁，给人审介入留窗口。

延迟叠加是工程上的硬约束。限流与水印在同步链路，每多一道就多几毫秒；行为指纹虽异步，但聚合层本身要扛高写入。若三者各自为政，P99 延迟会快速劣化。统一网关、共享上下文、异步落库是必须的工程动作。

还有一条常被忽略：水印本身是双刃。一旦嵌入策略泄露，攻击者可反向清洗或伪造水印，制造"这是从你这里偷的"的反诬。水印密钥与嵌入逻辑必须按敏感资产管理，定期轮换，否则协同防护会从内部被瓦解。

五、总结

模型窃取防护从单一限流走向水印、限流与行为指纹的协同，是因为单点手段都有盲区。限流挡暴力提取，水印做事后归属，行为指纹识别低速窃取——三者覆盖不同时间尺度。工程上，同步门与异步判别要分层，延迟与误报要量化权衡，水印密钥要按敏感资产管理。协同防护说到底，就是一条统一的证据链，三种手段在这条链上各自守一段。