给自建 AI 应用加文本水印与来源标注：EU 溯源合规落地

这篇能做出什么

照着做完，你会得到一个能跑的最小可用版本：

• 自建模型的输出文本里，嵌入一层统计可检出的水印，肉眼看不出来，但用密钥能算出一个置信分数；
• 每个生成结果旁边附一份签名的来源清单（C2PA 内容凭证），记录"由哪个模型、什么时候、以什么参数生成"；
• 对外 API 的返回值里带上机器可读的 provenance 字段，页面或客户端能直接渲染成"AI 生成"标签；
• 一个检测接口，可以拿一段文本回来问："这段内容是不是本系统生成的？置信度多少？"

对应欧盟 AI 法案第 50 条的透明度义务，落地路径可以拆成三根支柱：机器可读标记（水印 + 元数据）、用户可见披露（人看得见的标签）、可验证签名（第三方能校验，不能随便抵赖）。水印负责"内容本身带痕迹"，C2PA 负责"元数据可验证"。两者互补，缺一个都有短板。

> 合规时间表、具体条款措辞和适用范围，以欧盟官方公报与官方页面当前版本为准。本文只讲工程落地方式。

---

前置条件清单

环境

• Python 3.10 以上（以你所用框架的官方要求为准）
• PyTorch + Transformers，能加载并推理一个自建或本地部署的因果语言模型
• c2patool（C2PA 官方 Rust 参考实现的命令行工具），用于签名与校验
• 一对签名证书与私钥（开发阶段可自签，生产建议用受信任 CA 签发）
• 一个时间戳服务（TSA）地址，具体以证书签发方与 c2patool 文档为准

知识

• 知道什么是 logits、采样（temperature / top-p）
• 能看懂 JSON 和基本的命令行操作

需要注意的边界

• C2PA 规范目前对图像、视频、音频、PDF 等文件型资产支持成熟，对纯文本的支持仍在演进。可行做法是把文本包进 HTML 或 PDF 再签名，把水印直接打在文本 token 上。以 C2PA 官方规范当前版本为准。
• 水印是概率性的，不是密码学证明。短文本、被改写过的文本，检测结论要谨慎。

---

分步骤

第 1 步：理解两层标记的分工

顺序很重要：先生成文本并打水印 → 再把文本和溯源信息一起封进载体 → 最后对载体签名。签名之后任何字节改动都会让校验失败，所以签名必须是最后一步。

第 2 步：实现一个可自持的水印处理器

下面是一份自包含的 KGW 风格水印实现（绿色名单 + logits 偏置）。生成端和检测端共用同一份哈希逻辑，所以不需要依赖库的内部实现细节。

```python

watermark.py

import hashlib

import math

import torch

from transformers import LogitsProcessor

def greenlist_for(prev_token_id: int, vocab_size: int, gamma: float, key: bytes) -> torch.Tensor:

"""由 (密钥, 前一个 token) 确定性地导出一份绿色名单。

生成端和检测端调用同一函数，保证可复现。

"""

h = hashlib.sha256(key + prev_token_id.to_bytes(8, "big")).digest()

seed = int.from_bytes(h[:8], "big")

g = torch.Generator().manual_seed(seed)

perm = torch.randperm(vocab_size, generator=g)

k = max(1, int(vocab_size * gamma))

return perm[:k]

class KGWLogitsProcessor(LogitsProcessor):

"""给绿色名单里的 token 加一个正偏置，让模型更倾向选它们。"""

def __init__(self, vocab_size: int, key: bytes, gamma: float = 0.25, delta: float = 2.0):

self.vocab_size = vocab_size

self.key = key

self.gamma = gamma

self.delta = delta

def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor) -> torch.FloatTensor:

for b in range(input_ids.size(0)):

prev = int(input_ids[b, -1].item())

green = greenlist_for(prev, self.vocab_size, self.gamma, self.key)

mask = torch.zeros(scores.size(-1), dtype=torch.bool, device=scores.device)

mask[green.to(scores.device)] = True

scores[b, mask] += self.delta

return scores

```

参数直觉：gamma 是绿色名单占比，delta 是偏置强度。delta 调大检测更容易，但文本质量会下降；调小则反之。gamma=0.25、delta=2.0 是一个可以起步的组合，具体按你的模型实测调整。

第 3 步：接到生成流程里

```python

generate.py

import os

import torch

from transformers import AutoModelForCausalLM, AutoAI 词典：Token">Tokenizer, LogitsProcessorList

from watermark import KGWLogitsProcessor

MODEL_ID = os.environ["MODEL_ID"] # 你的本地模型路径或名称

WATERMARK_KEY = os.environ["WATERMARK_KEY"].encode() # 至少 32 字节随机串

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)

model = AutoModelForCausalLM.from_pretrained(MODEL_ID, device_map="auto")

VOCAB = model.config.vocab_size

processor = LogitsProcessorList([KGWLogitsProcessor(VOCAB, WATERMARK_KEY)])

def run_model(prompt: str, max_new_tokens: int = 256):

inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

prompt_len = inputs["input_ids"].shape[-1]

with torch.no_grad():

out = model.generate(

**inputs,

max_new_tokens=max_new_tokens,

do_sample=True,

temperature=0.8,

top_p=0.95,

logits_processor=processor,

pad_token_id=tokenizer.eos_token_id,

)

gen_ids = out[0][prompt_len:].tolist()

text = tokenizer.decode(gen_ids, skip_special_tokens=True)

return text, gen_ids

```

注意 gen_ids 要只保留生成部分，检测时用不到 prompt 的 token。

第 4 步：写检测函数

```python

detect.py

import math

from watermark import greenlist_for

def detect(token_ids, vocab_size: int, key: bytes,

gamma: float = 0.25, z_threshold: float = 4.0):

"""返回 (绿色命中率, z 分数, 是否判定为水印文本)。"""

if len(token_ids) < 2:

return 0.0, 0.0, False

hits = 0

total = 0

for t in range(1, len(token_ids)):

green = set(greenlist_for(token_ids[t - 1], vocab_size, gamma, key).tolist())

total += 1

if token_ids[t] in green:

hits += 1

if total == 0:

return 0.0, 0.0, False

ratio = hits / total

expected = gamma * total

std = math.sqrt(total * gamma * (1 - gamma))

z = (hits - expected) / std if std > 0 else 0.0

return ratio, z, z >= z_threshold

```

判定逻辑：没有水印的文本，绿色命中率应当接近 gamma；有水印时显著高于它。z 衡量偏离程度，z >= 4 是一个常用的保守阈值。文本越短，z 的波动越大，所以 少于大约 50 个 token 的片段不要下结论。

第 5 步：把溯源信息写成 C2PA 清单

先准备清单 JSON。c2pa.actions 是规范定义的断言，用来描述"发生了什么操作"；digitalSourceType 用 IPTC 的数字来源类型词表，trainedAlgorithmicMedia 表示由训练过的算法生成。

```json

{

"claim_generator": "my-ai-app/1.0",

"claim_generator_info": [

{ "name": "my-ai-app", "version": "1.0" }

],

"title": "AI 生成文本",

"assertions": [

{

"label": "c2pa.actions",

"data": {

"actions": [

{

"action": "c2pa.created",

"digitalSourceType": "http://cv.iptc.org/newscodes/digitalsourcetype/trainedAlgorithmicMedia",

"softwareAgent": { "name": "my-ai-app", "version": "1.0" }

}

]

}

},

{

"label": "com.example.ai.provenance",

"data": {

"model_id": "your-model-id",

"generated_at": "2025-01-01T00:00:00Z",

"prompt_sha256": "<对 prompt 取哈希，不落原文>",

"watermark": {

"scheme": "kgw",

"key_id": "kgw-2026-01",

"gamma": 0.25,

"delta": 2.0

},

"disclosure": "本内容由 AI 生成"

}

}

]

}

```

自定义断言标签建议用反向域名（com.example.ai.provenance），避免和规范保留标签冲突。清单里只放 key_id，不放密钥本身。

第 6 步：把文本封进载体并签名

把生成文本渲染成 HTML 或 PDF，再对文件签名。HTML 做一层最简单的外壳：

```html

<!-- article.template.html -->

<!doctype html>

<html lang="zh-CN">

<head><meta charset="utf-8"><title>AI 生成内容</title></head>

<body>

<p class="ai-disclosure">本内容由 AI 生成</p>

<article>{{TEXT}}</article>

</body>

</html>

```

然后用 c2patool 签名：

```bash

c2patool ./build/article.html \

--manifest ./build/manifest.json \

--output ./build/article.signed.html \

--cert ./certs/signer.pem \

--private-key ./certs/signer.key \

--ta_url "https://你的时间戳服务地址"

```

参数名以 c2patool 官方文档当前版本为准。时间戳不是可选项——没有可信时间戳，证书过期后就无法证明"当时确实签过"。

校验：

```bash

c2patool ./build/article.signed.html --detailed

```

输出里会包含签名校验状态和完整清单。把它接进发布流水线，任何一步签名失败就阻断发布。

第 7 步：在应用层暴露机器可读字段与可见标签

```python

app.py

import uuid

from fastapi import FastAPI

from pydantic import BaseModel

from generate import run_model, VOCAB, WATERMARK_KEY

from detect import detect

app = FastAPI()

class GenRequest(BaseModel):

prompt: str

max_new_tokens: int = 256

class DetectRequest(BaseModel):

token_ids: list[int]

@app.post("/v1/generate")

def generate(req: GenRequest):

text, token_ids = run_model(req.prompt, req.max_new_tokens)

ratio, z, _ = detect(token_ids, VOCAB, WATERMARK_KEY)

asset_id = uuid.uuid4().hex

return {

"text": text,

"provenance": {

"ai_generated": True,

"disclosure": "本内容由 AI 生成",

"watermark": {

"scheme": "kgw",

"key_id": "kgw-2026-01",

"z_score": round(z, 2),

},

"content_credentials": f"/provenance/{asset_id}.html",

},

}

@app.post("/v1/detect")

def detect_endpoint(req: DetectRequest):

ratio, z, is_watermarked = detect(req.token_ids, VOCAB, WATERMARK_KEY)

return {"green_ratio": round(ratio, 4), "z_score": round(z, 2),

"watermarked": is_watermarked}

```

披露文案要出现在用户第一次看到内容的位置，不能只塞在页面底部或者只留在 API 字段里。终端的做法是：正文上方一行明显的标签，旁边挂一个"内容凭证"入口，点开能看到 C2PA 清单摘要。

第 8 步：密钥与日志

```bash

.env（不要提交进仓库）

WATERMARK_KEY=<32 字节以上的随机串>

MODEL_ID=<你的模型标识>

C2PA_KEY_ID=kgw-2026-01

```

密钥放入 KMS 或密钥管理服务，按季度轮换。轮换时保留旧 key_id 的解密能力，否则历史内容无法复检。同时在数据库里记录：asset_id、key_id、z_score、prompt_sha256、模型版本、生成参数、清单哈希。这份日志是应对审计的材料。

---

常见坑与排错

检测分数一直是 0 附近

先确认检测端和生成端用的是同一个 key、同一个 gamma、同一个 tokenizer。换了模型版本或分词器，绿色名单就完全对不上了。另外确认检测输入只有生成部分，把 prompt 混进去会稀释统计量。

文本读起来变别扭

delta 太大了。把 delta 降到 1.0～1.5 试试，或者把 gamma 调小。也可以只对前 N 个 token 施加偏置，减少对长文的影响。

短文本检测不准

这是原理决定的。少于约 50 个 token 时统计量没意义，接口应当返回"样本不足"，而不是返回"未检测到水印"。

C2PA 校验显示签发者未知

自签名证书的正常现象。开发阶段可以接受，对外发布建议换成受信任 CA 签发的证书，并配置好时间戳。

签名之后内容一改就校验失败

这是设计如此。C2PA 对文件字节做硬绑定，任何重排版、复制粘贴、格式转换都会让绑定失效。所以：纯文本传播场景靠水印兜底，C2PA 放在你可控的分发载体（HTML / PDF / 下载包）上。

有人故意改写来绕过水印

改写、翻译、摘要都会削弱乃至抹掉水印。这是水印的固有局限。不要对外宣称"无法去除"。定位应该是：提高大规模自动伪造的成本，并为正常传播留下可核验痕迹。

性能

每生成一个 token 都要跑一次 randperm，词表大时开销明显。生产环境把绿色名单按 (key_id, prev_token_id) 缓存，或者换成位图/布隆过滤器实现。

---

下一步建议

• 换更强的水印方案：KGW 是入门基线。Transformer 生态里已经内置了水印 logits processor，Google 也开源了 SynthID-Text 的采样式水印实现，抗改写能力更好，值得对比评估。以各项目官方文档当前版本为准。
• 把检测做成独立服务：和生成服务解耦，便于灰度、也可用于审核外部送检内容。
• 打通多模态：图像/音频用对应的水印方案，再用 C2PA 统一承载溯源清单，保持一套凭证体系。
• 提供"内容凭证"查看页：用户点开能看到模型、生成时间、水印状态。这比一行小字更能满足透明度要求。
• 关注"持久凭证"方向：行业在推动把水印作为软绑定与 C2PA 清单关联，让元数据被剥离后仍能靠水印找回溯源信息。相关规范仍在演进，以官方页面为准。
• 补一份内部合规清单：生成时是否打标、发布时是否带可见披露、签名是否成功、日志是否完整、密钥是否轮换。把这五项做成发布前的自动检查。

最后提醒一句：技术手段能覆盖大部分场景，但欧盟的透明度要求里，"让人看得懂" 和 "机器读得到" 是两件事。水印和元数据解决后者，披露文案和界面设计解决前者，两件事都要做。