在本教程中，我们使用 AugLy 为图像、文本和音频构建一个全面的多模态增强与鲁棒性工作流。我们首先解决现代依赖兼容性问题，并生成确定性的合成数据集，使实验保持自包含且可复现。然后我们探索 AugLy 的函数式和基于类的 API、元数据与强度跟踪、概率组合、边界框感知变换以及自定义变换。我们将工作流扩展到实际的鲁棒性实验，通过在图像失真下对感知哈希复制检测进行基准测试，并针对对抗扰动、Unicode 混淆、净化和对抗训练来评估文本分类器。我们还集成了音频增强，构建可查询的元数据仓库，并将 AugLy 变换直接连接到 PyTorch 数据集和 DataLoader，从而让我们从端到端视角看到增强既是一种数据生成机制，也是一种可度量的鲁棒性工具。
复制代码已复制使用其他浏览器
import subprocess, sys, importlib
def _sh(cmd):
print(f"$ {cmd}")
subprocess.run(cmd, shell=True, check=False,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
def _need(mod):
try:
importlib.import_module(mod)
return False
except ImportError:
return True
if _need("augly"):
_sh("apt-get -qq install -y libmagic1 > /dev/null 2>&1")
_sh(f'"{sys.executable}" -m pip install -q --no-deps augly')
_sh(f'"{sys.executable}" -m pip install -q "iopath>=0.1.8" "python-magic>=0.4.22" '
f'"regex>=2021.4.4" "nlpaug==1.1.3"')
import numpy as np
from PIL import Image, ImageDraw, ImageFont, ImageFilter
for _name, _builtin in (("float", float), ("int", int), ("bool", bool)):
if not hasattr(np, _name):
setattr(np, _name, _builtin)
def _size(font, text):
left, top, right, bottom = font.getbbox(text)
return (right, bottom)
if not hasattr(ImageFont.FreeTypeFont, "getsize"):
ImageFont.FreeTypeFont.getsize = lambda self, t, *a, **k: _size(self, t)
if not hasattr(ImageFont.FreeTypeFont, "getsize_multiline"):
def _getsize_multiline(self, text, direction=None, spacing=4, features=None,
language=None, stroke_width=0):
lines = text.split("\n")
w = max((_size(self, ln)[0] for ln in lines), default=0)
h = sum(_size(self, ln)[1] for ln in lines) + spacing * (len(lines) - 1)
return (w, h)
ImageFont.FreeTypeFont.getsize_multiline = _getsize_multiline
import os, io, json, math, random, string, textwrap, unicodedata, warnings
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
import matplotlib.pyplot as plt
import pandas as pd
import augly.image as imaugs
import augly.text as textaugs
import augly.utils as augutils
from augly.image.transforms import BaseTransform as ImageBaseTransform
warnings.filterwarnings("ignore")
pd.set_option("display.width", 160)
SEED = 1234
random.seed(SEED)
np.random.seed(SEED)
print("\n" + "=" * 78)
print("AugLy ready. assets at:", augutils.ASSETS_BASE_DIR)
print("image augs :", len([f for f in dir(imaugs) if f[0].islower()]))
print("text augs :", len([f for f in dir(textaugs) if f[0].islower()]))
print("=" * 78 + "\n")
def make_image(idx: int, w: int = 320, h: int = 240) -> Tuple[Image.Image, Tuple[int, int, int, int]]:
"""程序化生成的“照片” + 一个 pascal_voc 格式的真实边界框。"""
rng = random.Random(SEED + idx)
img = Image.new("RGB", (w, h), tuple(rng.randint(20, 90) for _ in range(3)))
d = ImageDraw.Draw(img)
for _ in range(70):
x0, y0 = rng.randint(0, w), rng.randint(0, h)
d.line([x0, y0, x0 + rng.randint(-60, 60), y0 + rng.randint(-60, 60)],
fill=tuple(rng.randint(60, 160) for _ in range(3)), width=rng.randint(1, 3))
ow, oh = rng.randint(70, 130), rng.randint(60, 110)
ox, oy = rng.randint(10, w - ow - 10), rng.randint(10, h - oh - 10)
box = (ox, oy, ox + ow, oy + oh)
colour = tuple(rng.randint(150, 255) for _ in range(3))
if idx % 3 == 0:
d.ellipse(box, fill=colour, outline=(255, 255, 255), width=3)
elif idx % 3 == 1:
d.rectangle(box, fill=colour, outline=(255, 255, 255), width=3)
else:
d.polygon([(ox + ow // 2, oy), (ox + ow, oy + oh), (ox, oy + oh)],
fill=colour, outline=(255, 255, 255))
return img, box
N_IMAGES = 24
IMAGES, BOXES = zip(*[make_image(i) for i in range(N_IMAGES)])
IMAGES, BOXES = list(IMAGES), list(BOXES)
DEMO_IMG, DEMO_BOX = IMAGES[0], BOXES[0]
def make_text_dataset(n_per_class: int = 260):
"""由模板构建的小型情感语料库 -> 可学习但并非平凡。"""
rng = random.Random(SEED)
pos_adj = ["excellent", "delightful", "superb", "charming", "brilliant",
"flawless", "wonderful", "outstanding", "impressive", "lovely"]
neg_adj = ["terrible", "awful", "dreadful", "disappointing", "clumsy",
"broken", "miserable", "useless", "painful", "sloppy"]
subj = ["the movie", "this restaurant", "the hotel room", "their support team",
"the new phone", "the sequel", "this laptop", "the delivery service"]
tail_p = ["and I would recommend it to anyone", "worth every rupee",
"I left completely satisfied", "easily the best of the year",
"it exceeded all my expectations"]
tail_n = ["and I want a refund", "a total waste of money",
"I left extremely frustrated", "easily the worst of the year",
"it failed every expectation"]
rows = []
for _ in range(n_per_class):
rows.append((f"{rng.choice(subj)} was {rng.choice(pos_adj)} {rng.choice(tail_p)}", 1))
rows.append((f"{rng.choice(subj)} was {rng.choice(neg_adj)} {rng.choice(tail_n)}", 0))
rng.shuffle(rows)
return [r[0] for r in rows], [r[1] for r in rows]
TEXTS, LABELS = make_text_dataset()
DEMO_TEXT = "The quick brown fox jumps over the lazy dog near the river bank"
def make_audio(seconds: float = 2.0, sr: int = 16000) -> Tuple[np.ndarray, int]:
"""一个啁啾声 + 谐波 + 一点噪声 = 某种你确实能听出变化的东西。"""
t = np.linspace(0, seconds, int(sr * seconds), endpoint=False)
f = np.linspace(220, 880, t.size)
sig = 0.5 * np.sin(2 * np.pi * f * t) + 0.2 * np.sin(2 * np.pi * 2 * f * t)
sig += 0.02 * np.random.RandomState(SEED).randn(t.size)
env = np.minimum(1.0, np.minimum(t * 8, (seconds - t) * 8))
return (sig * env).astype(np.float32), sr
AUDIO, SR = make_audio()
def show_grid(pairs, cols=4, title="", figsize_scale=2.9):
"""pairs: (caption, PIL.Image) 的列表。"""
rows = math.ceil(len(pairs) / cols)
fig, axes = plt.subplots(rows, cols, figsize=(cols * figsize_scale, rows * figsize_scale))
axes = np.atleast_1d(axes).ravel()
for ax, (cap, im) in zip(axes, pairs):
ax.imshow(im)
ax.set_title(cap, fontsize=8)
ax.axis("off")
for ax in axes[len(pairs):]:
ax.axis("off")
if title:
fig.suptitle(title, fontsize=13, y=1.0)
plt.tight_layout()
plt.show()
def as_str(out) -> str:
"""AugLy text augs 对某些变换中的 str 输入返回 str，对其他变换返回 list。"""
return out[0] if isinstance(out, list) else out
print("\n### §2 IMAGE AUGMENTATION + METADATA " + "#" * 38)
functional_result = imaugs.pixelization(DEMO_IMG, ratio=0.25)
class_result = imaugs.Pixelization(ratio=0.25, p=1.0)(DEMO_IMG)
print("functional == class:", np.array_equal(np.array(functional_result), np.array(class_result)))
IMAGE_ZOO = {
"blur": lambda im, m: imaugs.blur(im, radius=3.0, metadata=m),
"brightness": lambda im, m: imaugs.brightness(im, factor=1.7, metadata=m),
"color_jitter": lambda im, m: imaugs.color_jitter(im, brightness_factor=1.3,
contrast_factor=1.4,
saturation_factor=1.6, metadata=m),
"crop": lambda im, m: imaugs.crop(im, x1=.15, y1=.15, x2=.85, y2=.85, metadata=m),
"encoding_quality": lambda im, m: imaugs.encoding_quality(im, quality=8, metadata=m),
"grayscale": lambda im, m: imaugs.grayscale(im, metadata=m),
"hflip": lambda im, m: imaugs.hflip(im, metadata=m),
"meme_format": lambda im, m: imaugs.meme_format(im, text="TOP TEXT",
caption_height=90, metadata=m),
"opacity": lambda im, m: imaugs.opacity(im, level=0.45, metadata=m),
"overlay_emoji": lambda im, m: imaugs.overlay_emoji(im, opacity=0.9,
emoji_size=0.35, metadata=m),
"overlay_screenshot": lambda im, m: imaugs.overlay_onto_screenshot(im, metadata=m),
"overlay_stripes": lambda im, m: imaugs.overlay_stripes(im, line_width=0.4,
line_opacity=0.7, metadata=m),
"overlay_text": lambda im, m: imaugs.overlay_text(im, opacity=0.9, metadata=m),
"pad_square": lambda im, m: imaugs.pad_square(im, metadata=m),
"perspective": lambda im, m: imaugs.perspective_transform(im, sigma=40.0, metadata=m),
"pixelization": lambda im, m: imaugs.pixelization(im, ratio=0.15, metadata=m),
"random_noise": lambda im, m: imaugs.random_noise(im, var=0.03, metadata=m),
"rotate": lambda im, m: imaugs.rotate(im, degrees=17, metadata=m),
"saturation": lambda im, m: imaugs.saturation(im, factor=3.0, metadata=m),
"scale": lambda im, m: imaugs.scale(im, factor=0.35, metadata=m),
"sharpen": lambda im, m: imaugs.sharpen(im, factor=8.0, metadata=m),
"shuffle_pixels": lambda im, m: imaugs.shuffle_pixels(im, factor=0.15, metadata=m),
"skew": lambda im, m: imaugs.skew(im, skew_factor=0.35, metadata=m),
"vflip": lambda im, m: imaugs.vflip(im, metadata=m),
}
gallery, image_meta = [("ORIGINAL", DEMO_IMG)], []
for name, fn in IMAGE_ZOO.items():
m = []
try:
out = fn(DEMO_IMG, m)
gallery.append((f"{name}\nintensity={m[0]['intensity']:.1f}", out))
image_meta.append(m[0])
except Exception as e:
print(f" [skip] {name}: {type(e).__name__}: {e}")
show_grid(gallery, cols=5, title="§2 AugLy 图像增强（含 AugLy 自带的强度评分）")
meta_df = pd.DataFrame(image_meta)[["name", "intensity", "src_width", "src_height",
"dst_width", "dst_height"]]
print(meta_df.sort_values("intensity", ascending=False).head(10).to_string(index=False))
我们在现代 Colab 环境中搭建了 AugLy，同时为 NumPy 和 Pillow 添加了兼容性补丁。我们生成了确定性的合成图像、文本和音频数据集，无需外部下载。在探索图像增强和元数据之前，我们还初始化了可复用的可视化和工具函数。
Copy CodeCopiedUse a different Browser
print("\n### §3 COMPOSITION & REPRODUCIBILITY " + "#" * 39)
REUPLOAD_PIPELINE = imaugs.Compose([
imaugs.OneOf([
imaugs.OverlayOntoScreenshot(),
imaugs.MemeFormat(text="LOL", caption_height=80),
imaugs.OverlayStripes(line_width=0.3, line_opacity=0.5),
], p=0.9),
imaugs.RandomAspectRatio(min_ratio=0.7, max_ratio=1.4, p=0.5),
imaugs.RandomEmojiOverlay(p=0.7),
imaugs.RandomBrightness(min_factor=0.7, max_factor=1.4, p=0.6),
imaugs.EncodingQuality(quality=12, p=1.0),
])
def run_pipeline(img, seed=None):
"""AugLy 图像变换使用全局 `random` 模块 -> 为其设置种子以保证确定性。"""
if seed is not None:
random.seed(seed)
np.random.seed(seed)
meta = []
return REUPLOAD_PIPELINE(img, metadata=meta), meta
a, meta_a = run_pipeline(DEMO_IMG, seed=7)
b, meta_b = run_pipeline(DEMO_IMG, seed=7)
c, _ = run_pipeline(DEMO_IMG, seed=99)
print("same seed -> identical output:", np.array_equal(np.array(a), np.array(b)))
print("applied chain (seed=7) :", " -> ".join(m["name"] for m in meta_a))
show_grid([("original", DEMO_IMG), ("seed=7", a), ("seed=7 again", b), ("seed=99", c)],
cols=4, title="§3 带种子的、可复现的增强流水线")
print("\n### §4 边界框感知增强 " + "#" * 45)
BBOX_OPS = [
("crop", lambda im, m, bb: imaugs.crop(im, x1=.1, y1=.1, x2=.9, y2=.9,
metadata=m, bboxes=bb, bbox_format="pascal_voc")),
("hflip", lambda im, m, bb: imaugs.hflip(im, metadata=m, bboxes=bb,
bbox_format="pascal_voc")),
("rotate 20", lambda im, m, bb: imaugs.rotate(im, degrees=20, metadata=m, bboxes=bb,
bbox_format="pascal_voc")),
("pad", lambda im, m, bb: imaugs.pad(im, w_factor=0.25, h_factor=0.25,
metadata=m, bboxes=bb, bbox_format="pascal_voc")),
("meme_format", lambda im, m, bb: imaugs.meme_format(im, text="BOXED", caption_height=80,
metadata=m, bboxes=bb,
bbox_format="pascal_voc")),
]
def draw_box(img, box, colour=(0, 255, 0)):
out = img.copy().convert("RGB")
ImageDraw.Draw(out).rectangle([float(v) for v in box], outline=colour, width=4)
return out
bbox_panels = [("original", draw_box(DEMO_IMG, DEMO_BOX))]
for label, op in BBOX_OPS:
m = []
try:
out = op(DEMO_IMG, m, [DEMO_BOX])
dst = m[0]["dst_bboxes"][0]
bbox_panels.append((f"{label}\n{tuple(round(v) for v in dst)}", draw_box(out, dst)))
print(f" {label:12s} {DEMO_BOX} -> {tuple(round(v, 1) for v in dst)}")
except Exception as e:
print(f" [skip] {label}: {type(e).__name__}: {e}")
show_grid(bbox_panels, cols=3, title="§4 边界框自动跟随像素")
print("\n### §5 自定义变换 " + "#" * 51)
class RecompressionChain(ImageBaseTransform):
"""模拟一张图像在经历 N 轮平台重新编码后仍然存活。
继承 BaseTransform（而不是使用 ApplyLambda）能为你带来：`p`
概率门控、`force=True`，以及完整参与 Compose/OneOf。
"""
def __init__(self, n_rounds: int = 3, min_q: int = 12, max_q: int = 45,
downscale: float = 0.85, p: float = 1.0):
super().__init__(p)
self.n_rounds, self.min_q, self.max_q, self.downscale = n_rounds, min_q, max_q, downscale
def apply_transform(self, image, metadata=None, bboxes=None, bbox_format=None):
src_w, src_h = image.size
out, qualities = image, []
for _ in range(self.n_rounds):
q = random.randint(self.min_q, self.max_q)
qualities.append(q)
out = imaugs.encoding_quality(out, quality=q)
out = imaugs.scale(out, factor=self.downscale)
out = out.resize((src_w, src_h), Image.BILINEAR)
if metadata is not None:
metadata.append({
"name": "recompression_chain",
"src_width": src_w, "src_height": src_h,
"dst_width": out.size[0], "dst_height": out.size[1],
"n_rounds": self.n_rounds, "qualities": qualities,
"intensity": float(100 * (1 - np.mean(qualities) / 100)),
})
return out
vignette = imaugs.ApplyLambda(aug_function=lambda im: Image.composite(
im, Image.new("RGB", im.size, (0, 0, 0)),
Image.radial_gradient("L").resize(im.size).point(lambda v: 255 - v)))
random.seed(SEED)
custom_meta = []
show_grid([
("original", DEMO_IMG),
("RecompressionChain(n=3)", RecompressionChain(n_rounds=3)(DEMO_IMG, metadata=custom_meta)),
("RecompressionChain(n=6)", RecompressionChain(n_rounds=6, min_q=5, max_q=20)(DEMO_IMG)),
("ApplyLambda vignette", vignette(DEMO_IMG)),
], cols=4, title="§5 Custom transforms drop straight into the AugLy API")
print(" custom metadata:", custom_meta[0])
CUSTOM_PIPELINE = imaugs.Compose([RecompressionChain(n_rounds=2, p=1.0),
imaugs.RandomEmojiOverlay(p=1.0)])
_ = CUSTOM_PIPELINE(DEMO_IMG)
print(" composed with built-ins: OK")
我们使用 Compose 和 OneOf 构建概率增强流水线，同时通过显式随机种子控制可复现性。我们演示 AugLy 如何在空间变换中自动传播边界框坐标。随后我们实现一个自定义的 BaseTransform，并将其与内置的 AugLy 变换组合使用。
Copy CodeCopiedUse a different Browser
print("\n### §6 COPY-DETECTION ROBUSTNESS BENCHMARK " + "#" * 33)
from scipy.fftpack import dct
def phash(img: Image.Image, hash_size: int = 8, highfreq: int = 4) -> np.ndarray:
"""Classic DCT perceptual hash -> 64-bit signature as a bool array."""
size = hash_size * highfreq
px = np.asarray(img.convert("L").resize((size, size), Image.LANCZOS), dtype=np.float64)
d = dct(dct(px, axis=0, norm="ortho"), axis=1, norm="ortho")[:hash_size, :hash_size]
return (d > np.median(d[1:, 1:])).ravel()
def hamming(a, b) -> int:
return int(np.count_nonzero(a != b))
INDEX = np.stack([phash(im) for im in IMAGES])
ATTACKS = {
"brightness x1.6": lambda im: imaugs.brightness(im, factor=1.6),
"blur r=3": lambda im: imaugs.blur(im, radius=3.0),
"jpeg q=8": lambda im: imaugs.encoding_quality(im, quality=8),
"crop 80%": lambda im: imaugs.crop(im, x1=.1, y1=.1, x2=.9, y2=.9),
"rotate 12": lambda im: imaugs.rotate(im, degrees=12),
"hflip": lambda im: imaugs.hflip(im),
"grayscale": lambda im: imaugs.grayscale(im),
"pixelize 0.2": lambda im: imaugs.pixelization(im, ratio=0.2),
"noise var=.03": lambda im: imaugs.random_noise(im, var=0.03),
"emoji overlay": lambda im: imaugs.overlay_emoji(im, emoji_size=0.35, opacity=0.9),
"meme format": lambda im: imaugs.meme_format(im, text="LOL", caption_height=70),
"screenshot": lambda im: imaugs.overlay_onto_screenshot(im),
"perspective s=40": lambda im: imaugs.perspective_transform(im, sigma=40.0),
"scale 0.35": lambda im: imaugs.scale(im, factor=0.35),
"stripes": lambda im: imaugs.overlay_stripes(im, line_width=0.4, line_opacity=0.7),
"re-encode chain": lambda im: RecompressionChain(n_rounds=3)(im),
"REUPLOAD pipeline": lambda im: REUPLOAD_PIPELINE(im),
}
rows = []
for attack, fn in ATTACKS.items():
random.seed(SEED)
np.random.seed(SEED)
hits, dists, failures = 0, [], 0
for i, im in enumerate(IMAGES):
try:
q = phash(fn(im))
except Exception:
failures += 1
continue
d = np.array([hamming(q, h) for h in INDEX])
hits += int(d.argmin() == i)
dists.append(int(d[i]))
n = len(IMAGES) - failures
rows.append({"attack": attack,
"top1_recall": hits / max(n, 1),
"mean_hamming": float(np.mean(dists)) if dists else np.nan,
"errors": failures})
bench = pd.DataFrame(rows).sort_values("top1_recall")
print(bench.to_string(index=False, float_format=lambda v: f"{v:.3f}"))
fig, ax = plt.subplots(1, 2, figsize=(14, 6))
colours = ["#c0392b" if r < .5 else "#e67e22" if r < .9 else "#27ae60"
for r in bench.top1_recall]
ax[0].barh(bench.attack, bench.top1_recall, color=colours)
ax[0].set_xlabel("top-1 retrieval recall"); ax[0].set_xlim(0, 1.05)
ax[0].axvline(0.9, ls="--", c="k", lw=1)
ax[0].set_title("pHash survival per AugLy attack")
ax[1].scatter(bench.mean_hamming, bench.top1_recall, s=70, c=colours)
for _, r in bench.iterrows():
ax[1].annotate(r.attack, (r.mean_hamming, r.top1_recall), fontsize=7,
xytext=(3, 3), textcoords="offset points")
ax[1].set_xlabel("与真实匹配的平均 Hamming 距离 (0-64)")
ax[1].set_ylabel("top-1 召回率")
ax[1].set_title("失真 vs. 检索失败")
plt.tight_layout(); plt.show()
worst = bench.head(3).attack.tolist()
print(f"\n pHash 在以下情况下失效: {worst}")
print(" -> 正是你会加入训练的那些增强，或者用")
print(" 几何不变的嵌入来处理，而不是哈希。")
我们在合成图像语料库上构建了一个感知哈希索引，并评估其针对大量 AugLy 失真的鲁棒性。我们测量每次攻击的 top-1 检索召回率和 Hamming 距离，以量化不同变换对副本检测的影响。我们将结果可视化，以识别对感知匹配造成最严重退化的增强。
Copy CodeCopiedUse a different Browser
print("\n### §7 TEXT ATTACK / DEFEND / HARDEN " + "#" * 39)
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
X_tr, X_te, y_tr, y_te = train_test_split(TEXTS, LABELS, test_size=0.3,
random_state=SEED, stratify=LABELS)
def new_model():
return make_pipeline(
TfidfVectorizer(analyzer="word", ngram_range=(1, 2), sublinear_tf=True),
LogisticRegression(max_iter=1000, C=4.0),
)
baseline = new_model().fit(X_tr, y_tr)
clean_acc = accuracy_score(y_te, baseline.predict(X_te))
print(f"干净测试准确率: {clean_acc:.3f}\n")
ATTACK_SUITE = {
"拼写错误 (键盘+拼写错误)": textaugs.SimulateTypos(aug_word_p=0.45, typo_type="all"),
"Unicode 同形字": textaugs.ReplaceSimilarUnicodeChars(aug_word_p=0.8, aug_char_p=0.4),
"leet 语相似字符": textaugs.ReplaceSimilarChars(aug_word_p=0.8, aug_char_p=0.4),
"零宽字符注入": textaugs.InsertZeroWidthChars(granularity="word", cadence=2.0),
"标点注入": textaugs.InsertPunctuationChars(granularity="word", cadence=2.0,
vary_chars=True),
"空白注入": textaugs.InsertWhitespaceChars(granularity="word", cadence=3.0),
"趣味字体": textaugs.ReplaceFunFonts(aug_p=0.8, granularity="word",
vary_fonts=True),
"上下颠倒": textaugs.ReplaceUpsideDown(aug_p=0.6, granularity="word"),
"双向文本": textaugs.ReplaceBidirectional(granularity="word"),
"拆分单词": textaugs.SplitWords(aug_word_p=0.5),
"合并单词": textaugs.MergeWords(aug_word_p=0.5),
"大小写混排": textaugs.ChangeCase(granularity="word", cadence=2.0, case="upper"),
}
print("这些攻击作用在一个句子上的效果：")
for name, aug in ATTACK_SUITE.items():
random.seed(SEED)
print(f" {name:26s} {as_str(aug(DEMO_TEXT))[:72]}")
ZERO_WIDTH = dict.fromkeys(
[0x200B, 0x200C, 0x200D, 0x2060, 0x2061, 0x2062, 0x2063, 0x2064, 0xFEFF,
0x202A, 0x202B, 0x202C, 0x202D, 0x202E, 0x2066, 0x2067, 0x2068, 0x2069, 0x200E, 0x200F]
)
PUNCT_TABLE = str.maketrans("", "", "".join(c for c in string.punctuation if c not in "'"))
def sanitize(text: str) -> str:
"""针对 Unicode 混淆攻击的低成本、高收益预处理。"""
t = text.translate(ZERO_WIDTH)
t = unicodedata.normalize("NFKD", t)
t = "".join(ch for ch in t if not unicodedata.combining(ch))
t = t.translate(PUNCT_TABLE)
return " ".join(t.lower().split())
random.seed(SEED)
TRAIN_AUGS = [ATTACK_SUITE[k] for k in
["typos (keyboard+misspell)", "unicode homoglyphs", "leetspeak lookalikes",
"zero-width injection", "punctuation injection", "fun fonts", "CaSe ChAoS"]]
aug_X, aug_y = list(X_tr), list(y_tr)
for aug in TRAIN_AUGS:
for x, y in zip(X_tr, y_tr):
aug_X.append(as_str(aug(x)))
aug_y.append(y)
print(f"\ntrain set: {len(X_tr)} -> {len(aug_X)} examples after AugLy adversarial training")
hardened = make_pipeline(
TfidfVectorizer(analyzer="char_wb", ngram_range=(3, 5), sublinear_tf=True,
preprocessor=sanitize),
LogisticRegression(max_iter=2000, C=4.0),
).fit(aug_X, aug_y)
results = []
for name, aug in ATTACK_SUITE.items():
random.seed(SEED)
X_atk = [as_str(aug(x)) for x in X_te]
results.append({
"attack": name,
"baseline": accuracy_score(y_te, baseline.predict(X_atk)),
"+ sanitize": accuracy_score(y_te, baseline.predict([sanitize(x) for x in X_atk])),
"+ sanitize + adv-train": accuracy_score(y_te, hardened.predict(X_atk)),
})
res = pd.DataFrame(results).set_index("attack").sort_values("baseline")
res.loc["-- CLEAN --"] = [clean_acc,
accuracy_score(y_te, baseline.predict([sanitize(x) for x in X_te])),
accuracy_score(y_te, hardened.predict(X_te))]
print("\n" + res.to_string(float_format=lambda v: f"{v:.3f}"))
print(f"\nmean accuracy under attack: baseline {res.iloc[:-1,0].mean():.3f}"
f" -> sanitized {res.iloc[:-1,1].mean():.3f}"
f" -> hardened {res.iloc[:-1,2].mean():.3f}")
plot_df = res.drop(index="-- CLEAN --")
ax = plot_df.plot.barh(figsize=(11, 7), width=0.78,
color=["#c0392b", "#e6a817", "#27ae60"])
ax.axvline(clean_acc, ls="--", c="k", lw=1.2, label=f"clean acc = {clean_acc:.2f}")
ax.set_xlabel("accuracy under attack"); ax.set_xlim(0, 1.05)
ax.set_title("§7 AugLy as a red-team suite — and as the fix")
ax.legend(loc="lower right", fontsize=8)
plt.tight_layout(); plt.show()
我们创建一个文本分类基线，并系统性地将其暴露于拼写错误、Unicode 同形字、不可见字符、标点变化以及其他对抗性变换之下。我们实现 Unicode 规范化和清洗，以移除若干类混淆。随后我们在训练期间使用 AugLy 生成的对抗样本，并将由此得到的加固模型与基线进行比较。
Copy CodeCopiedUse a different Browser
print("\n### §8 AUDIO AUGMENTATION " + "#" * 50)
try:
import augly.audio as audaugs
AUDIO_ZOO = {
"pitch_shift +4": lambda a, sr, m: audaugs.pitch_shift(a, sr, n_steps=4.0, metadata=m),
"time_stretch 1.5x": lambda a, sr, m: audaugs.time_stretch(a, sr, rate=1.5, metadata=m),
"change_volume -12dB":lambda a, sr, m: audaugs.change_volume(a, sr, volume_db=-12.0, metadata=m),
"clicks": lambda a, sr, m: audaugs.clicks(a, sr, seconds_between_clicks=0.25,
metadata=m),
"background_noise": lambda a, sr, m: audaugs.add_background_noise(a, sr, snr_level_db=5.0,
metadata=m),
"low_pass 1kHz": lambda a, sr, m: audaugs.low_pass_filter(a, sr, cutoff_hz=1000.0,
metadata=m),
"high_pass 2kHz": lambda a, sr, m: audaugs.high_pass_filter(a, sr, cutoff_hz=2000.0,
metadata=m),
"peaking_eq": lambda a, sr, m: audaugs.peaking_equalizer(a, sr, center_hz=800.0,
gain_db=-12.0, metadata=m),
"harmonic": lambda a, sr, m: audaugs.harmonic(a, sr, metadata=m),
"percussive": lambda a, sr, m: audaugs.percussive(a, sr, metadata=m),
"clip (50%)": lambda a, sr, m: audaugs.clip(a, sr, duration_factor=0.5, metadata=m),
"loop x2": lambda a, sr, m: audaugs.loop(a, sr, n=1, metadata=m),
"normalize": lambda a, sr, m: audaugs.normalize(a, sr, metadata=m),
"speed 1.4x": lambda a, sr, m: audaugs.speed(a, sr, factor=1.4, metadata=m),
"tempo 0.8x": lambda a, sr, m: audaugs.tempo(a, sr, factor=0.8, metadata=m),
"reverb": lambda a, sr, m: audaugs.reverb(a, sr, reverberance=80.0, metadata=m),
}
ok, audio_meta = [], []
for name, fn in AUDIO_ZOO.items():
m = []
try:
out, out_sr = fn(AUDIO.copy(), SR, m)
ok.append((name, np.asarray(out).squeeze(), out_sr))
audio_meta.append({"name": m[0]["name"], "intensity": m[0].get("intensity"),
"dst_duration": np.asarray(out).squeeze().shape[-1] / out_sr})
except Exception as e:
print(f" [skip] {name}: {type(e).__name__}: {str(e)[:90]}")
print("\n" + pd.DataFrame(audio_meta).to_string(index=False,
float_format=lambda v: f"{v:.3f}"))
n = min(len(ok), 8)
fig, axes = plt.subplots(n + 1, 1, figsize=(11, 1.5 * (n + 1)), sharex=False)
axes[0].plot(AUDIO, lw=.5, color="k"); axes[0].set_ylabel("orig", fontsize=7)
for ax, (name, sig, _) in zip(axes[1:], ok[:n]):
ax.plot(sig, lw=.5); ax.set_ylabel(name, fontsize=6, rotation=0, ha="right", va="center")
for ax in axes:
ax.set_xticks([]); ax.set_yticks([])
fig.suptitle("§8 Audio waveforms after AugLy augmentation")
plt.tight_layout(); plt.show()
try:
from IPython.display import Audio, display
print("original:"); display(Audio(AUDIO, rate=SR))
for name, sig, sr_out in ok[:3]:
print(name); display(Audio(sig, rate=sr_out))
except Exception:
pass
except ImportError as e:
print(f" audio module unavailable ({e}).")
print(" On Colab librosa/torch/torchaudio are preinstalled; elsewhere run:")
print(' pip install "librosa>=0.8.1" soundfile audioread torch torchaudio')
print("\n### §9 METADATA WAREHOUSE " + "#" * 50)
warehouse = []
random.seed(SEED)
for idx, im in enumerate(IMAGES[:8]):
for name, fn in IMAGE_ZOO.items():
meta = []
try:
out = fn(im, meta)
except Exception:
continue
m = meta[0]
warehouse.append({
"source_id": idx,
"augmentation": m["name"],
"intensity": m.get("intensity"),
"src_w": m.get("src_width"), "src_h": m.get("src_height"),
"dst_w": m.get("dst_width"), "dst_h": m.get("dst_height"),
"area_ratio": (m.get("dst_width", 0) * m.get("dst_height", 0)) /
max(m.get("src_width", 1) * m.get("src_height", 1), 1),
})
wh = pd.DataFrame(warehouse)
print(wh.head(8).to_string(index=False, float_format=lambda v: f"{v:.2f}"))
print(f"\nlogged {len(wh)} augmented samples from {wh.source_id.nunique()} sources")
print("\nhardest augmentations by mean intensity:")
print(wh.groupby("augmentation").intensity.mean().sort_values(ascending=False)
.head(8).to_string(float_format=lambda v: f"{v:.1f}"))
wh.to_csv("augly_metadata.csv", index=False)
print("\nwrote augly_metadata.csv (join this to your training manifest)")
我们将增强工作流扩展到音频，应用音高偏移、时间拉伸、滤波、噪声注入和混响等变换，同时优雅地跳过不可用的依赖项。我们检查生成的波形，并在支持的情况下直接在 Colab 中播放增强后的样本。我们还构建了一个元数据仓库，记录增强类型、强度、尺寸和面积变化，以供下游分析使用。
Copy CodeCopiedUse a different Browser
print("\n### §10 PYTORCH DATASET / DATALOADER " + "#" * 39)
try:
import torch
from torch.utils.data import Dataset, DataLoader
import torchvision.transforms as T
class AugLyDataset(Dataset):
"""AugLy transforms are drop-in for torchvision because they are PIL->PIL."""
def __init__(self, images, labels, train=True):
self.images, self.labels = images, labels
aug = imaugs.Compose([
imaugs.OneOf([imaugs.RandomBlur(min_radius=0.5, max_radius=2.5),
imaugs.RandomPixelization(min_ratio=0.3, max_ratio=1.0),
imaugs.EncodingQuality(quality=20)], p=0.8),
imaugs.RandomBrightness(min_factor=0.7, max_factor=1.4, p=0.6),
imaugs.RandomEmojiOverlay(p=0.3),
RecompressionChain(n_rounds=1, p=0.3),
]）
steps = ([aug] if train else []) + [
T.Resize((128, 128)),
T.ToTensor(),
T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
]
self.tf = T.Compose(steps)
def __len__(self):
return len(self.images)
def __getitem__(self, i):
return self.tf(self.images[i]), self.labels[i]
ds = AugLyDataset(IMAGES, [i % 3 for i in range(len(IMAGES))], train=True)
dl = DataLoader(ds, batch_size=8, shuffle=True, num_workers=0)
xb, yb = next(iter(dl))
print(f" batch tensor {tuple(xb.shape)} dtype={xb.dtype} labels={yb.tolist()}")
denorm = (xb[:8] * torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1)
+ torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1)).clamp(0, 1)
show_grid([(f"batch[{i}] y={yb[i].item()}", denorm[i].permute(1, 2, 0).numpy())
for i in range(min(8, len(denorm)))],
cols=4, title="§10 直接从 DataLoader 取出的一批增强数据")
np_img = np.asarray(DEMO_IMG)
np_out = imaugs.aug_np_wrapper(np_img, imaugs.overlay_emoji,
**{"opacity": 0.8, "y_pos": 0.4})
print(f" aug_np_wrapper: {np_img.shape} -> {np_out.shape} ({np_out.dtype})")
except ImportError:
print(" 未安装 torch/torchvision — 跳过（Colab 默认已安装）。")
print("\n" + "=" * 78)
print("完成。从这里可以延伸的想法：")
print(" * 把 §6 中的 pHash 换成真正的嵌入（CLIP / DINOv2）并重新运行表格")
print(" * 把 §9 的 CSV 输入课程学习：先在低强度增强上训练")
print(" * 加入 augly.video（pip install 'augly' + apt install ffmpeg）并")
print(" 用同样的方式对帧级鲁棒性做基准测试")
print("=" * 78)
我们将 AugLy 直接集成到 PyTorch 的 Dataset 和 DataLoader 中，使增强可以作为训练时预处理流程的一部分运行。我们在增强之后应用归一化和张量转换，并可视化生成的一个训练批次，以验证完整的数据路径。我们还演示了 AugLy 的 NumPy 原生封装，并总结了基于嵌入和视频鲁棒性基准测试的实用扩展方向。
总之，我们展示了如何将 AugLy 用作一个系统化的鲁棒性工程框架，而不仅仅是独立增强函数的集合。我们测量了不同图像变换对复制检测检索的影响，展示了对抗性文本变换如何暴露传统分类器的弱点，并评估了净化和对抗训练作为互补防御手段的效果。我们还保留了增强元数据和强度信息，使每个生成的样本都保持可追溯和可分析，同时自定义变换让我们能够对特定应用的失真进行建模。通过将图像、文本、音频和 PyTorch 工作流整合到一个可复现的流水线中，我们为构建增强感知的训练系统、鲁棒性基准测试和生产数据流水线奠定了基础。
在此查看完整代码。所有功劳归于该项目的研究人员。另外，欢迎在 Twitter 上关注我们，别忘了加入我们拥有 15 万+ 成员的 ML SubReddit 并订阅我们的 Newsletter。等等！你在 Telegram 上吗？现在你也可以在 Telegram 上加入我们了。
需要与我们合作推广你的 GitHub 仓库或 Hugging Face 页面或产品发布或网络研讨会等？联系我们
文章 End-to-End Multimodal Data Augmentation and Adversarial Robustness Benchmark with AugLy for Images, Text, Audio, and PyTorch 首次发布于 MarkTechPost。