简介这份源码面向Python开发者、安全方向学生及参加数据挖掘竞赛的选手聚焦文本与多模态数据融合的风险识别任务可用于课程设计、期末大作业或安全AI赛题复现。资源包共11个文件以7个Python脚本为核心覆盖词向量训练、数据合并、K折训练、伪标签等完整流程另附requirements.txt依赖清单、run.sh运行脚本、readme.md说明及docx手册压缩包约88KB结构紧凑便于快速上手。该方案源自字节跳动安全AI挑战赛色情导流用户识别项目已有360人学习。读者可借此掌握文本特征与多模态信息联合建模的工程实现理解K折交叉验证、伪标签半监督策略及评估脚本的编写思路并参考配置与运行脚本完成本地复现适合作为风险识别方向的实战参考与二次开发基础。1. 文本加多模态做风险识别为什么单看文本已经不够用了金融风控、内容审核、电商反欺诈这些场景里纯文本模型早就不是新鲜事。但真正在一线跑过业务的人会发现一个尴尬的现实用户提交的申请材料里文本字段可以编得滴水不漏可上传的截图、证件照、商品图却经常露馅。我见过一个案例某贷款申请人的职业信息、收入描述、单位名称全部自洽文本模型给出 0.12 的违约概率但他上传的工作证明截图里公章边缘有明显的 PS 痕迹字体间距也不对。单模态模型看不到这些。这就是「python实现基于文本和多模态数据的风险识别源码」这个方向要解决的问题把文本特征和图像特征在同一个模型里做融合让风险识别的准确率不再被单一模态的天花板卡住。适合谁看有 Python 基础、做过 sklearn 或 PyTorch 小项目、想把这个方案落到实际风控或审核业务里的工程师。下面从数据管线怎么搭、模型怎么设计、参数怎么调、坑在哪一步步拆开讲。2. 多模态风险识别的数据管线怎么搭从原始文件到模型输入2.1 文本和图像各自怎么清洗、怎么对齐多模态项目翻车最多的环节不是模型是数据对齐。文本和图像必须能通过一个唯一 ID 关联起来否则后面的融合层根本没法训练。常见做法是维护一张样本表每行包含sample_id、text_content、image_path、label四个核心字段。文本侧的处理链路去 HTML 标签、统一编码、截断到固定长度。中文场景下用 jieba 分词还是直接按字切取决于你的预训练模型。如果用 BERT 系列直接按字切更稳如果用 TextCNN 从头训分词后做 embedding 效果更好。图像侧的处理链路统一尺寸、归一化、必要时做数据增强。风险识别场景里图像增强要克制翻转和裁剪可能把关键的风险信号比如公章位置、水印方向破坏掉。我一般只做 resize 和归一化最多加一点亮度扰动。import os import pandas as pd from PIL import Image from torch.utils.data import Dataset, DataLoader from torchvision import transforms import jieba class RiskDataset(Dataset): def __init__(self, csv_path, img_root, tokenizer, max_len128, img_size224): self.df pd.read_csv(csv_path) self.img_root img_root self.tokenizer tokenizer self.max_len max_len # 图像变换只做 resize 和归一化不做翻转 self.img_transform transforms.Compose([ transforms.Resize((img_size, img_size)), transforms.ToTensor(), transforms.Normalize(mean[0.485, 0.456, 0.406], std[0.229, 0.224, 0.225]) ]) def __len__(self): return len(self.df) def __getitem__(self, idx): row self.df.iloc[idx] # 文本处理按字切分后转 id text str(row[text_content]).strip() tokens self.tokenizer( text, max_lengthself.max_len, paddingmax_length, truncationTrue, return_tensorspt ) # 图像处理读取并变换 img_path os.path.join(self.img_root, row[image_path]) image Image.open(img_path).convert(RGB) image self.img_transform(image) label int(row[label]) return { input_ids: tokens[input_ids].squeeze(0), attention_mask: tokens[attention_mask].squeeze(0), image: image, label: label }这段代码的关键点在于tokenizer负责把文本转成模型能吃的 id 序列img_transform负责把图像统一到 224x224。max_len128对大多数风控文本够用如果你的文本里有长段描述可以加到 256但要注意显存。img_size224是 ResNet 系列的标准输入换成 ViT 的话通常是 224 或 384。注意图像增强里的随机翻转在风险识别场景要慎用。公章、签名、水印这些关键信号对方向敏感翻转后模型学到的特征可能和真实分布对不上。2.2 用 DataLoader 把两个模态拼成一个 batch单模态的 DataLoader 只返回一个张量多模态要返回一个字典。PyTorch 的默认 collate_fn 对字典的处理没问题但要注意每个字段的维度必须一致。from transformers import BertTokenizer from torch.utils.data import DataLoader tokenizer BertTokenizer.from_pretrained(bert-base-chinese) dataset RiskDataset( csv_pathdata/samples.csv, img_rootdata/images, tokenizertokenizer, max_len128, img_size224 ) loader DataLoader( dataset, batch_size32, shuffleTrue, num_workers4, pin_memoryTrue ) # 验证一个 batch 的维度 batch next(iter(loader)) print(batch[input_ids].shape) # [32, 128] print(batch[image].shape) # [32, 3, 224, 224] print(batch[label].shape) # [32]batch_size32是个保守起点。如果你的显存够比如 24G 以上可以试 64。num_workers4在 Linux 上通常没问题Windows 下如果报错就改成 0。pin_memoryTrue在 GPU 训练时能加快数据传输CPU 训练时反而多余。这里有个容易忽略的点文本和图像的采样必须同步。如果你的数据集里正负样本比例是 1:10而图像模态的缺失率又比较高直接 shuffle 可能导致某个 batch 里全是负样本。稳妥的做法是先做分层采样保证每个 batch 的正负比例大致均衡。3. 模型结构怎么设计文本分支、图像分支和融合层3.1 文本分支用 BERT 还是 TextCNN选型取决于你的数据量和算力。数据量在 1 万条以下TextCNN 从头训可能比 BERT 微调更稳因为 BERT 在小数据上容易过拟合。数据量超过 5 万条BERT 微调的优势就出来了。我一般会先用 TextCNN 跑一个 baseline确认数据管线没问题再换 BERT 看提升幅度。TextCNN 的结构很简单embedding 层 多尺度卷积 最大池化 全连接。import torch import torch.nn as nn class TextBranch(nn.Module): def __init__(self, vocab_size, embed_dim128, num_filters128, filter_sizes(2, 3, 4), dropout0.3): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim, padding_idx0) self.convs nn.ModuleList([ nn.Conv1d(embed_dim, num_filters, fs) for fs in filter_sizes ]) self.dropout nn.Dropout(dropout) self.fc nn.Linear(num_filters * len(filter_sizes), 256) def forward(self, input_ids, attention_maskNone): # input_ids: [batch, seq_len] x self.embedding(input_ids) # [batch, seq_len, embed_dim] x x.permute(0, 2, 1) # [batch, embed_dim, seq_len] conv_outs [] for conv in self.convs: c torch.relu(conv(x)) # [batch, num_filters, seq_len - fs 1] c torch.max(c, dim2)[0] # [batch, num_filters] conv_outs.append(c) x torch.cat(conv_outs, dim1) # [batch, num_filters * 3] x self.dropout(x) x self.fc(x) # [batch, 256] return xembed_dim128对中文小数据集够用num_filters128配合三个卷积核尺寸输出维度是 384再压到 256。dropout0.3是防止过拟合的常规操作如果你的训练集很小可以加到 0.5。3.2 图像分支用 ResNet 做特征提取图像分支直接用 torchvision 里的预训练 ResNet把最后的全连接层换掉就行。预训练权重能帮你省掉大量标注数据。from torchvision import models class ImageBranch(nn.Module): def __init__(self, output_dim256, freeze_backboneFalse): super().__init__() self.backbone models.resnet18(pretrainedTrue) if freeze_backbone: for param in self.backbone.parameters(): param.requires_grad False in_features self.backbone.fc.in_features self.backbone.fc nn.Sequential( nn.Linear(in_features, 512), nn.ReLU(), nn.Dropout(0.3), nn.Linear(512, output_dim) ) def forward(self, image): return self.backbone(image)freeze_backboneFalse表示微调整个 ResNet。如果你的图像数据少于 5000 张建议设成 True只训最后的全连接层。resnet18是最轻量的选择换成resnet50会慢不少但特征质量更高。3.3 融合层拼接、注意力还是门控最简单的融合是直接拼接两个分支的输出然后过几层全连接。这种方式参数少、训练快但两个模态的贡献是固定的模型没法根据样本动态调整。进阶做法是加一个注意力机制让模型自己学每个模态的权重。我一般用门控融合class GatedFusion(nn.Module): def __init__(self, text_dim256, image_dim256, hidden_dim256): super().__init__() self.gate nn.Sequential( nn.Linear(text_dim image_dim, hidden_dim), nn.Sigmoid() ) self.fc nn.Sequential( nn.Linear(text_dim image_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.3), nn.Linear(hidden_dim, 2) # 二分类输出 ) def forward(self, text_feat, image_feat): concat torch.cat([text_feat, image_feat], dim1) g self.gate(concat) # [batch, hidden_dim] weighted concat * g # 门控加权 out self.fc(weighted) # [batch, 2] return out门控机制的核心是Sigmoid输出的权重向量它决定了每个特征维度在最终判断里占多大比重。训练初期门控值接近 0.5随着训练推进会逐渐分化。如果你的某个模态数据质量很差门控值会自然降低相当于模型自己学会了「不信这个模态」。4. 训练参数怎么调学习率、损失函数和早停策略4.1 分层学习率是必须的文本分支和图像分支的预训练程度不一样用同一个学习率会出问题。BERT 微调通常用 2e-5 到 5e-5ResNet 微调用 1e-4 到 1e-3融合层从头训可以用 1e-3。如果统一用 1e-3BERT 那部分会震荡统一用 2e-5ResNet 又学得太慢。def build_optimizer(model, text_lr2e-5, image_lr1e-4, fusion_lr1e-3): text_params list(model.text_branch.parameters()) image_params list(model.image_branch.parameters()) fusion_params list(model.fusion.parameters()) optimizer torch.optim.AdamW([ {params: text_params, lr: text_lr}, {params: image_params, lr: image_lr}, {params: fusion_params, lr: fusion_lr} ], weight_decay0.01) return optimizerAdamW比Adam多了正确的权重衰减实现在多模态训练里更稳。weight_decay0.01是常规值数据量小可以加到 0.05。4.2 损失函数选交叉熵还是 Focal Loss风险识别场景通常有类别不平衡问题负样本远多于正样本。标准交叉熵在这种情况下会让模型偏向预测多数类。Focal Loss 通过降低易分类样本的权重让模型更关注难样本。import torch.nn.functional as F class FocalLoss(nn.Module): def __init__(self, alpha0.25, gamma2.0): super().__init__() self.alpha alpha self.gamma gamma def forward(self, logits, targets): ce_loss F.cross_entropy(logits, targets, reductionnone) pt torch.exp(-ce_loss) focal_loss self.alpha * (1 - pt) ** self.gamma * ce_loss return focal_loss.mean()alpha0.25和gamma2.0是 Focal Loss 的经典参数。alpha控制正负样本的权重正样本少就设小一点gamma控制难易样本的聚焦程度越大越关注难样本。如果你的数据不平衡比例超过 1:20可以把alpha降到 0.1。4.3 早停看验证集的 F1 还是 AUC风险识别场景里AUC 反映的是排序能力F1 反映的是阈值确定后的分类效果。我一般两个都看但早停用验证集的 F1因为最终业务落地时你需要一个明确的阈值。from sklearn.metrics import f1_score, roc_auc_score def evaluate(model, loader, device): model.eval() all_preds, all_labels, all_probs [], [], [] with torch.no_grad(): for batch in loader: input_ids batch[input_ids].to(device) image batch[image].to(device) labels batch[label].to(device) logits model(input_ids, image) probs F.softmax(logits, dim1)[:, 1] preds (probs 0.5).long() all_preds.extend(preds.cpu().numpy()) all_labels.extend(labels.cpu().numpy()) all_probs.extend(probs.cpu().numpy()) f1 f1_score(all_labels, all_preds) auc roc_auc_score(all_labels, all_probs) return f1, auc早停的 patience 一般设 5 到 10 个 epoch。如果验证集 F1 连续 5 轮不提升就停掉。注意保存最佳模型时用torch.save(model.state_dict(), best_model.pt)不要只存整个模型对象后面加载会方便很多。5. 避坑与排查多模态风险识别里最容易翻车的 5 个地方5.1 文本和图像 ID 对不上训练 loss 不降现象训练几个 epoch 后 loss 一直在 0.69 附近震荡准确率接近随机猜。原因CSV 里的image_path和实际文件名不匹配或者图像读取失败返回了全黑图。DataLoader 不会报错模型学到的是噪声。解决在 Dataset 的__getitem__里加断言图像读取失败时直接抛异常而不是静默返回。另外用df[image_path].apply(lambda x: os.path.exists(os.path.join(img_root, x)))做一次全量检查。5.2 图像增强过度模型学不到风险信号现象训练集准确率很高验证集 F1 只有 0.5 左右。原因用了 RandomHorizontalFlip、RandomRotation 等增强把公章、签名这些关键信号的方向和位置破坏了。模型在训练集上记住了增强后的模式但验证集是原始图。解决风险识别场景只保留 Resize 和 Normalize最多加一点 ColorJitter 的亮度扰动。如果一定要做增强先可视化几张增强后的图确认关键信号还在。5.3 两个模态的学习率没分开BERT 部分梯度爆炸现象训练初期 loss 突然变成 NaN。原因文本分支用了和融合层一样的学习率比如 1e-3BERT 的预训练权重被大幅更新梯度爆炸。解决用分层学习率文本分支 2e-5图像分支 1e-4融合层 1e-3。另外加梯度裁剪torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)。5.4 验证集和测试集的样本分布不一致现象验证集 F1 0.85上线后实际效果只有 0.6。原因划分数据集时用了随机划分但实际业务里时间是最重要的维度。用未来数据预测过去模型学到了时间泄露的特征。解决按时间划分用前 80% 的数据训练后 20% 验证。如果数据量够做滚动窗口验证。5.5 门控融合的权重全部趋近于 0.5现象训练完后检查门控输出发现所有样本的权重都在 0.5 附近等于没融合。原因门控层的初始化用了默认的 Xavier输出接近 0.5。如果两个模态的特征尺度差异大门控学不到有区分度的权重。解决在门控层前加 BatchNorm把两个模态的特征归一化到相近尺度。另外可以给门控加一个稀疏正则鼓励它输出极端值。6. 进阶技巧用模态 dropout 提升模型鲁棒性训练多模态模型时模型很容易过度依赖某一个模态。比如图像质量普遍较好时文本分支可能被完全忽略。上线后如果图像接口挂了整个模型就废了。模态 dropout 的思路很简单训练时以一定概率把某个模态的输入置零强迫模型学会在只有单模态的情况下也能做出合理判断。class MultimodalRiskModel(nn.Module): def __init__(self, text_branch, image_branch, fusion, modal_dropout0.15): super().__init__() self.text_branch text_branch self.image_branch image_branch self.fusion fusion self.modal_dropout modal_dropout def forward(self, input_ids, image): text_feat self.text_branch(input_ids) image_feat self.image_branch(image) # 训练时随机丢弃一个模态 if self.training: if torch.rand(1).item() self.modal_dropout: text_feat torch.zeros_like(text_feat) elif torch.rand(1).item() self.modal_dropout: image_feat torch.zeros_like(image_feat) return self.fusion(text_feat, image_feat)modal_dropout0.15表示每个模态有 15% 的概率被置零。这个值不能太大否则模型欠拟合也不能太小否则起不到正则作用。我一般从 0.1 开始试观察验证集 F1 的变化。验证模态 dropout 是否生效可以做一个简单的测试加载训练好的模型把图像输入全部置零看文本分支单独输出的 AUC。如果 AUC 掉到 0.5 附近说明模型完全依赖图像如果还能保持在 0.7 以上说明文本分支学到了有效特征。还有一个实用技巧是「模态互补采样」。在构造 batch 时故意让一部分样本只有文本、一部分只有图像、一部分两个都有。这样模型在每个 batch 里都能见到单模态和双模态的情况学到的融合策略更稳健。def collate_with_modality_mask(batch, text_only_ratio0.2, image_only_ratio0.2): input_ids torch.stack([b[input_ids] for b in batch]) images torch.stack([b[image] for b in batch]) labels torch.tensor([b[label] for b in batch]) n len(batch) for i in range(n): r torch.rand(1).item() if r text_only_ratio: images[i] torch.zeros_like(images[i]) elif r text_only_ratio image_only_ratio: input_ids[i] torch.zeros_like(input_ids[i]) return {input_ids: input_ids, image: images, label: labels}这个 collate 函数在 DataLoader 里替换默认的即可。text_only_ratio0.2和image_only_ratio0.2意味着 40% 的样本是单模态60% 是双模态。实际用的时候根据你的业务场景调整如果图像接口不稳定就加大text_only_ratio。最后说一个我自己的习惯每次训完模型我都会把验证集里预测错误的样本单独拉出来看尤其是那些文本和图像信号矛盾的样本。这些样本往往暴露了融合层的缺陷比盯着 loss 曲线有用得多。希望帮到你。本文还有配套的精品资源点击获取