Python实现文本与多模态风险识别：CLIP+BERT融合实战-北京尧图网络

简介这份源码面向计算机、人工智能相关专业的课程设计与期末大作业场景也可作为安全算法竞赛的实战参考解决文本与多模态数据融合下的风险用户识别问题。项目源自字节跳动安全AI挑战赛色情导流用户识别赛道围绕词向量训练、数据合并、K折训练、伪标签等环节给出可运行的Python实现适合具备一定机器学习基础、希望理解多模态风险识别完整流程的学习者。压缩包共11个文件约88KB以7个py脚本为核心配套txt依赖清单、docx说明手册、md说明文档与sh运行脚本覆盖从环境配置到模型训练的主要环节。目前已有361人学习。读者可据此掌握赛题级方案的整体结构理解文本特征与多模态信息如何协同用于风险识别并借鉴K折交叉验证、伪标签等工程技巧快速搭建自己的实验基线。1. 文本加多模态做风险识别为什么单靠一个 BERT 会漏掉一半风险做过内容风控的同行大概都有过这种经历文本模型离线指标 AUC 跑到 0.93上线之后该拦的没拦住不该拦的误杀一片。翻日志才发现出问题的样本里大量是「图文不一致」——配文岁月静好配图却是另一回事或者反过来文字里全是黑话图片看着人畜无害。单模态模型在这种场景下就是个黑匣子它只看了一半信息却要给出一个全量判断。这个项目标题讲的是用 Python 实现一套基于文本和多模态数据的风险识别源码。拆开看核心是三件事文本侧做语义风险打分图像侧做视觉风险特征提取然后把两路信号融合成一个统一的风险概率。它解决的不是「文本分类怎么做」这种教科书问题而是真实业务里图文混合内容的风险判定——比如社区帖子、商品详情页、短视频封面加标题、用户举报工单。适合谁看有 Python 基础、做过至少一个分类任务、现在要处理图文混合数据的工程师。如果你只做过纯文本风控这套东西能帮你把漏召的那部分补回来。2. 多模态风险识别的技术选型为什么是 CLIP 加 BERT 而不是各训各的2.1 两条技术路线的取舍逻辑做图文风险识别绕不开一个架构决策文本和图像是分别建模再拼接还是从一开始就用一个联合编码器。分别建模的典型做法是 BERT 出文本向量、ResNet 或 ViT 出图像向量然后 concat 送进 MLP 分类头。这条路实现简单、训练快、显存友好但有个硬伤——两个模态之间没有交互模型学不到「这张图和这段文字放在一起才危险」这种跨模态信号。联合编码器路线里CLIP 是目前工程上最成熟的选择。它用对比学习把图像和文本映射到同一个语义空间天然具备跨模态对齐能力。你拿 CLIP 的 image encoder 和 text encoder 分别抽特征再做一个轻量的 cross-attention 融合层就能让模型在训练中自己学到图文之间的风险关联。代价是显存占用明显上升CLIP ViT-B/32 的视觉塔大概 1.5 亿参数文本塔 6300 万加上融合层单卡 8GB 显存跑 batch size 16 会比较紧张。我一般会这样选如果业务数据量在 10 万条以下、风险类型相对固定走 BERT ViT 分别编码再融合性价比最高如果数据量超过 30 万条、风险模式复杂多变直接上 CLIP 联合编码前期投入大但天花板高。还有一个折中方案是用 Chinese-CLIP它在中文学术和业务语料上做过对齐比原版 CLIP 对中文场景友好得多。2.2 数据管线的搭建从原始图文对到模型输入不管选哪条路线数据管线都是第一道坎。真实业务数据通常是这样的一条记录里有文本字段、一个或多个图片 URL、若干结构化字段发布时间、用户等级、举报次数。你需要把它整理成模型能吃的格式。import os import json import hashlib from pathlib import Path from PIL import Image import requests from io import BytesIO def build_multimodal_dataset(raw_records, image_cache_dir./image_cache): 将原始图文记录整理为多模态训练样本 raw_records: list of dict, 每条含 text, image_urls, label os.makedirs(image_cache_dir, exist_okTrue) samples [] for rec in raw_records: text rec.get(text, ).strip() if not text: continue # 图片下载与缓存用 URL 的 md5 做文件名避免重复下载 image_paths [] for url in rec.get(image_urls, [])[:3]: # 每条最多取3张图 fname hashlib.md5(url.encode()).hexdigest() .jpg fpath os.path.join(image_cache_dir, fname) if not os.path.exists(fpath): try: resp requests.get(url, timeout5) img Image.open(BytesIO(resp.content)).convert(RGB) img img.resize((224, 224)) # 统一尺寸CLIP ViT-B/32 输入224 img.save(fpath, JPEG, quality85) except Exception as e: print(fdownload failed: {url}, {e}) continue image_paths.append(fpath) if not image_paths: continue samples.append({ text: text, images: image_paths, label: rec[label], # 0 正常1 风险 record_id: rec.get(id, ) }) return samples # 使用示例 with open(raw_data.json, r, encodingutf-8) as f: raw json.load(f) dataset build_multimodal_dataset(raw) print(f有效样本数: {len(dataset)})这段代码做了三件事文本清洗、图片下载缓存、样本组装。几个关键参数需要根据实际情况调整。[:3]是每条记录最多取三张图因为大部分风险信号集中在前几张取太多会稀释注意力且增加显存压力。resize((224, 224))是 CLIP ViT-B/32 的标准输入尺寸如果你用 ViT-L/14 就要改成 224 但 patch size 不同实际输入是 224 或 336。timeout5是下载超时生产环境建议设 3 到 5 秒避免个别慢 URL 拖垮整个管线。注意图片缓存目录要定期清理否则跑几十万条数据后磁盘会被撑满。我一般会在管线里加一个 LRU 淘汰逻辑或者直接用对象存储加本地 SSD 缓存层。2.3 文本分支的实现BERT 微调与风险关键词注入文本分支是整个系统的基线。用bert-base-chinese或hfl/chinese-roberta-wwm-ext做微调输入就是原始文本输出是一个二分类概率。但纯 BERT 有个问题它对黑话、变体、谐音词的识别能力有限因为这些词在预训练语料里出现频率低。我的做法是在 BERT 之外挂一个关键词匹配模块把命中高风险词表的信号作为一个额外特征拼进分类头。词表不需要很大几百个核心词就够关键是要持续更新。import torch import torch.nn as nn from transformers import BertModel, BertTokenizer class TextRiskBranch(nn.Module): def __init__(self, bert_pathbert-base-chinese, num_keywords500): super().__init__() self.bert BertModel.from_pretrained(bert_path) self.keyword_embed nn.Embedding(num_keywords, 64) # BERT 输出 768 维关键词 64 维 self.classifier nn.Sequential( nn.Linear(768 64, 256), nn.ReLU(), nn.Dropout(0.3), nn.Linear(256, 2) ) def forward(self, input_ids, attention_mask, keyword_ids): # keyword_ids: [batch, num_hit_keywords]命中词索引未命中填0 bert_out self.bert(input_idsinput_ids, attention_maskattention_mask) cls_vec bert_out.last_hidden_state[:, 0, :] # [batch, 768] kw_vec self.keyword_embed(keyword_ids).mean(dim1) # [batch, 64] fused torch.cat([cls_vec, kw_vec], dim-1) logits self.classifier(fused) return logits # 关键词命中检测 def match_keywords(text, keyword2id): hits [0] * 10 # 最多取10个命中词 idx 0 for kw, kid in keyword2id.items(): if kw in text and idx 10: hits[idx] kid idx 1 return hitskeyword_embed的维度 64 是我试出来的经验值太小表达力不够太大容易过拟合。Dropout(0.3)在文本分支上比较关键因为风险识别任务的数据往往标注噪声大dropout 能缓解过拟合。match_keywords里限制最多 10 个命中词是因为一条文本里真正有信号的关键词通常不超过 5 个取太多反而引入噪声。3. 图像与跨模态融合CLIP 特征怎么接进风险分类头3.1 图像分支用 CLIP 视觉塔做特征提取图像分支不建议从零训 CNN直接用 CLIP 的 visual encoder 抽特征冻结底层、只训顶层适配层收敛快且泛化好。如果你用openai/clip-vit-base-patch32图像输入是 224x224输出是 512 维向量。from transformers import CLIPModel, CLIPProcessor class ImageRiskBranch(nn.Module): def __init__(self, clip_pathopenai/clip-vit-base-patch32, freeze_clipTrue): super().__init__() self.clip CLIPModel.from_pretrained(clip_path) if freeze_clip: for param in self.clip.parameters(): param.requires_grad False # CLIP vision output: 512 维 self.adapter nn.Sequential( nn.Linear(512, 256), nn.ReLU(), nn.Dropout(0.2), nn.Linear(256, 128) ) def forward(self, pixel_values): with torch.no_grad() if not self.training else torch.enable_grad(): vision_out self.clip.vision_model(pixel_valuespixel_values) pooled vision_out.pooler_output # [batch, 512] return self.adapter(pooled) # 图像预处理 processor CLIPProcessor.from_pretrained(openai/clip-vit-base-patch32) def preprocess_images(image_paths): images [Image.open(p).convert(RGB) for p in image_paths] inputs processor(imagesimages, return_tensorspt, paddingTrue) return inputs[pixel_values]freeze_clipTrue是默认选择因为 CLIP 预训练特征已经很强微调反而容易在小数据集上崩掉。adapter把 512 维压到 128 维是为了和文本分支的输出维度匹配方便后续融合。如果你显存够可以把freeze_clip设为 False 并只解冻最后两层 transformer block学习率设到 1e-5 量级。3.2 跨模态融合三种融合策略的对比与选择文本分支输出 256 维分类头之前图像分支输出 128 维融合方式直接决定模型上限。我试过三种融合方式实现复杂度显存开销适用场景拼接加 MLP低小数据量小、模态关联弱Cross-Attention中中图文关联强、数据量中等门控融合中小模态质量不稳定、需要动态加权拼接加 MLP 最简单把两个向量 concat 后过几层全连接。Cross-Attention 让文本向量去 attend 图像向量能学到细粒度关联但训练不稳定。门控融合是我目前最推荐的它用一个可学习的门控网络动态决定每个样本里文本和图像各占多少权重。class GatedFusion(nn.Module): def __init__(self, text_dim256, image_dim128, hidden128): super().__init__() self.gate nn.Sequential( nn.Linear(text_dim image_dim, hidden), nn.ReLU(), nn.Linear(hidden, 2), nn.Softmax(dim-1) ) self.final_classifier nn.Sequential( nn.Linear(text_dim image_dim, 128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 2) ) def forward(self, text_feat, image_feat): concat torch.cat([text_feat, image_feat], dim-1) gate_weights self.gate(concat) # [batch, 2] # 加权融合 weighted torch.cat([ text_feat * gate_weights[:, 0:1], image_feat * gate_weights[:, 1:2] ], dim-1) return self.final_classifier(weighted), gate_weights门控权重的输出可以拿来分析如果某个样本的文本权重接近 1、图像权重接近 0说明模型认为文本信号更可靠。上线后可以监控这个分布如果发现图像权重长期偏低说明图像分支没学好或者图像质量有问题。3.3 训练策略分阶段训练与损失函数设计多模态模型不建议一上来就端到端训。我的做法是分三阶段第一阶段单独训文本分支让它收敛到一个合理基线第二阶段冻结文本分支单独训图像分支的 adapter第三阶段解冻融合层和两个分支的顶层用较小学习率联合微调。损失函数用带类别权重的交叉熵因为风险识别任务里正负样本通常不均衡。如果正样本占比只有 5%给正类权重 5 到 10 倍。def train_epoch(model, dataloader, optimizer, criterion, device): model.train() total_loss 0 for batch in dataloader: input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) keyword_ids batch[keyword_ids].to(device) pixel_values batch[pixel_values].to(device) labels batch[label].to(device) optimizer.zero_grad() logits, gate_weights model(input_ids, attention_mask, keyword_ids, pixel_values) loss criterion(logits, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() total_loss loss.item() return total_loss / len(dataloader) # 类别权重计算 from sklearn.utils.class_weight import compute_class_weight import numpy as np labels [s[label] for s in dataset] class_weights compute_class_weight(balanced, classesnp.array([0, 1]), ylabels) criterion nn.CrossEntropyLoss(weighttorch.tensor(class_weights, dtypetorch.float))clip_grad_norm_的 max_norm 设 1.0 是防止梯度爆炸多模态模型里 cross-attention 层容易出这个问题。学习率方面文本分支用 2e-5图像 adapter 用 1e-4融合层用 5e-4分参数组设置。4. 避坑与排查多模态风险识别里最容易翻车的五个地方4.1 图片下载失败导致样本静默丢失现象训练集明明有 10 万条跑完管线只剩 6 万条模型指标虚高。原因build_multimodal_dataset里图片下载失败时直接continue把整条样本丢了。如果失败集中在某类图片比如某些域名的图床模型就学不到那部分分布。解决下载失败时不要丢样本而是用一张纯灰图占位同时记录一个image_missing标志位在融合层里把这个标志作为一个额外特征输入让模型自己决定是否忽略图像分支。4.2 文本长度截断把关键信号切掉现象某些风险文本明明有关键词模型却判正常。原因BERT 默认 max_length 是 512但中文长文本比如商品详情页经常超过这个长度关键风险词可能在后半段被截掉。解决不要简单截断用滑动窗口取前 256 和后 256 个 token分别过 BERT 后取平均。或者先用关键词匹配定位风险词位置优先保留包含风险词的片段。4.3 图像分支过拟合到背景颜色现象验证集准确率 0.95测试集掉到 0.7。原因训练集里风险样本的图片可能集中在某几种背景色或拍摄风格上模型学到了「背景暗就是风险」这种伪相关。解决训练时加颜色抖动、随机裁剪、随机翻转等增强。更关键的是做分层采样确保每个 batch 里正负样本的图像风格分布接近。我一般会用albumentations做增强比 torchvision 的 transform 更灵活。4.4 门控融合退化成单模态现象训练日志里 gate_weights 某一维长期接近 1另一维接近 0。原因两个模态的特征尺度不一致文本特征方差大图像特征方差小门控网络倾向于选信号强的那个。解决在融合前对两个模态特征做 LayerNorm把尺度拉到同一量级。另外可以在损失里加一个正则项惩罚门控权重过于极端的情况。4.5 推理时 batch 内 padding 导致结果不一致现象同一条样本单独推理和批量推理结果不一样。原因文本 padding 和图像 padding 在 batch 内对齐时attention mask 没处理好padding 位置参与了注意力计算。解决确保attention_mask正确传入 BERT图像 processor 的paddingTrue要配合return_tensorspt使用。推理时如果 batch size 变化最好固定 padding 到 max_length而不是动态 padding。5. 进阶技巧用风险类型解耦和置信度校准把误杀率压下去走到这一步模型能跑通、指标也还行但上线后你会发现一个更棘手的问题误杀。风险识别和普通分类不一样误杀的代价有时候比漏召还大。我踩过的坑是模型把所有「看起来像风险」的样本都判成风险召回上去了精确率崩了。第一个技巧是风险类型解耦。不要只做一个二分类而是把风险拆成若干子类型比如广告、辱骂、违禁品每个子类型单独出一个概率最后用一个规则层聚合。这样做的好处是每个子类型的决策阈值可以独立调广告类可以宽松一点违禁品类必须严格。实现上就是在融合层后面挂多个分类头训练时用多任务损失。class MultiHeadRiskModel(nn.Module): def __init__(self, fusion_dim384, num_risk_types5): super().__init__() self.shared nn.Sequential( nn.Linear(fusion_dim, 256), nn.ReLU(), nn.Dropout(0.3) ) # 每个风险类型一个独立分类头 self.heads nn.ModuleList([ nn.Linear(256, 2) for _ in range(num_risk_types) ]) # 全局风险头 self.global_head nn.Linear(256, 2) def forward(self, fused_feat): shared self.shared(fused_feat) type_logits [head(shared) for head in self.heads] global_logits self.global_head(shared) return type_logits, global_logits # 多任务损失 def multi_task_loss(type_logits, global_logits, type_labels, global_label, weights): loss 0 for i, logits in enumerate(type_logits): loss weights[i] * nn.functional.cross_entropy(logits, type_labels[:, i]) loss nn.functional.cross_entropy(global_logits, global_label) return loss第二个技巧是置信度校准。模型输出的 softmax 概率往往过于自信0.9 的概率实际准确率可能只有 0.7。用温度缩放temperature scaling在验证集上拟合一个温度参数把概率校准到接近真实置信度。这样你可以设一个「灰区」——概率在 0.4 到 0.6 之间的样本送人工审核而不是硬判。class TemperatureScaler(nn.Module): def __init__(self): super().__init__() self.temperature nn.Parameter(torch.ones(1) * 1.5) def forward(self, logits): return logits / self.temperature # 在验证集上拟合温度 def fit_temperature(model, val_loader, device): scaler TemperatureScaler().to(device) optimizer torch.optim.LBFGS([scaler.temperature], lr0.01, max_iter50) nll nn.CrossEntropyLoss() def closure(): optimizer.zero_grad() total_loss 0 for batch in val_loader: logits model(batch) # 获取未校准的 logits calibrated scaler(logits) total_loss nll(calibrated, batch[label].to(device)) total_loss.backward() return total_loss optimizer.step(closure) return scaler.temperature.item()温度值一般在 1.2 到 2.5 之间如果拟合出来超过 3说明模型过拟合严重得回头检查训练数据。第三个技巧是推理时的模态丢弃测试。上线前我会做一组实验只用文本推理、只用图像推理、两个都用对比三组结果的差异。如果只用文本的指标和两个都用的指标差距在 2 个点以内说明图像分支没贡献得回去检查图像特征是不是被门控网络忽略了。这个测试能帮你快速定位是融合层的问题还是某个分支本身没学好。最后说一个习惯每次改完模型结构或训练策略我都会固定一个 500 条的「金标集」跑一遍记录精确率、召回率和灰区比例。金标集不参与训练只用来做版本对比。这个习惯帮我避免了好几次「指标涨了但实际效果变差」的翻车。希望帮到你。本文还有配套的精品资源点击获取

Python实现文本与多模态风险识别：CLIP+BERT融合实战

相关推荐

智能体越界与对齐：从九起公开到上万起调查的工程实践

不依赖大模型：基于知识图谱的心理咨询智能问答系统构建

SpringBoot+Vue健康饮食推荐系统毕设实战

更多资讯

Aspose.Words for .NET 文档模板自动化实战指南

基于Wind Python接口实现Fama-French三因子与五因子模型构建

垃圾分类图像预处理：128×124降维如何保特征提精度

Linux x64 下 JDK 11 tar.gz 安装与 JAVA_HOME 配置避坑指南

GPU发起通信深度解剖：RDMA、NVSHMEM与NCCL机制与优化实践

OpcEnum 安装与排障：OPC DA 枚举器、DCOM 与注册表实战指南

今日更新

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本周精选

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？