☰
Python多模态风险识别:文本与图像融合实战及源码解析
2026/10/4 1:23:19 网站建设 项目流程

简介:本资源为Python实现基于文本与多模态数据的风险识别项目源码,源自字节跳动安全AI挑战赛色情导流用户识别赛题,面向计算机、人工智能相关专业学生及需要课程设计、期末大作业或竞赛练手的开发者,帮助理解多模态特征融合与风险用户识别的完整建模流程。压缩包共11个文件,约88KB,以7个Python脚本为核心,覆盖词向量训练、数据合并、K折训练、伪标签等环节,另含txt依赖清单、docx说明手册、md文档与sh运行脚本,便于快速复现与二次开发。目前已有360人学习。读者可获取从数据预处理、特征提取到模型训练评估的完整赛题方案,参考配置与运行脚本理清工程结构,并借助伪标签与K折思路提升识别效果,适合作为多模态风险识别入门与进阶的实践素材。

1. 文本加多模态做风险识别:为什么单看文字已经不够用了

电商评论里写「质量很好」,配图却是一张开裂的实物照;信贷申请材料文字工整,上传的营业执照却是 PS 拼接的。这类场景里,只跑一个文本分类模型,准确率会被图片里的信息直接拖垮。python实现基于文本和多模态数据的风险识别,要解决的就是这件事:把文本、图像甚至结构化字段放进同一个判断链路,输出一个可解释的风险分。它适合两类人——手上已有文本风控模型、想加一路图像信号的算法工程师,以及要快速搭出可演示原型的在校开发者。热搜里 python、多模态数据、风险识别、源码这几个词高频出现,说明大家真正卡住的不是概念,而是「文本和图像怎么对齐、特征怎么拼、源码怎么跑起来」。这篇就按我实际搭过的顺序,从数据组织讲到融合层,再讲部署时那些让人翻车的细节。

2. 多模态风险识别的数据组织与标签设计

2.1 文本、图像、结构化字段怎么对齐成一条样本

多模态项目第一个坑不在模型,在数据。文本和图像天然是两种存储,评论 ID、订单号、用户 ID 是它们之间唯一的桥。我一般会先定一张宽表,主键用业务唯一 ID,文本字段存原始字符串,图像字段存相对路径而不是二进制,结构化字段(金额、频次、设备数)单独列。这样做的原因是:训练时要能按 ID 快速取图,推理时要能只传文本降级运行。

import pandas as pd # 宽表结构:一行 = 一个风险事件样本 df = pd.DataFrame({ "sample_id": ["s001", "s002"], "text": ["质量很好下次还来", "货不对板要求退款"], "image_path": ["img/s001.jpg", "img/s002.jpg"], # 存相对路径,不存二进制 "amount": [199.0, 39.9], # 结构化字段 "device_cnt": [1, 5], "label": [0, 1] # 0 正常 1 风险 }) # 校验图文是否一一对应,缺失的直接剔除,别让模型学空图 df = df[df["image_path"].apply(lambda p: os.path.exists(p))]

逻辑说明:宽表把三种模态绑在同一主键下,image_path存路径是为了让数据集类按需加载,避免一次性把几万张图读进内存。参数上label用 0/1 二分类起步,多级风险(低/中/高)建议先做二分类再拆阈值,否则小样本类别会直接训崩。缺失图像样本要么剔除要么走纯文本分支,不要用全黑图填充,那会让模型学到「黑图=正常」的伪相关。

2.2 标签从哪来:弱监督与人工复核的配比

风险识别最贵的是标签。真实业务里正样本往往不到 5%,全靠人工标不现实。常见做法是先用规则打一批弱标签(命中敏感词、图像重复上传、金额异常),再抽 10% 到 20% 做人工复核,用复核结果去校准规则。我一般会把弱标签和人工标签分两列存,训练时用人工标签,弱标签只做预训练或样本筛选。

# 弱标签规则示例:命中任一条件先标为疑似风险 df["weak_label"] = ( (df["text"].str.contains("退款|投诉|假")) | (df["device_cnt"] > 3) | (df["amount"] > 500) ).astype(int) # 人工复核只覆盖弱标签为正的样本,控制成本 review_pool = df[df["weak_label"] == 1].sample(frac=0.2, random_state=42)

逻辑说明:weak_label是启发式规则产物,只用来缩小人工复核范围,不能直接当训练标签,否则模型只是在复现你的规则。frac=0.2是复核比例,样本量小时可以提到 0.3,但别超过 0.5,否则人工成本失控。这一步的产出是一份「干净标签子集」,后面所有指标都以它为准。

2.3 划分数据集时最容易忽略的泄漏问题

文本和图像如果来自同一用户或同一批次,随机划分会让训练集和验证集共享同一来源,指标虚高。血泪经验是:按用户 ID 或时间做分组划分,而不是按行随机。比如用GroupShuffleSplit按user_id分组,保证同一用户不出现在两边。

from sklearn.model_selection import GroupShuffleSplit gss = GroupShuffleSplit(n_splits=1, test_size=0.2, random_state=42) train_idx, val_idx = next(gss.split(df, groups=df["user_id"])) train_df, val_df = df.iloc[train_idx], df.iloc[val_idx]

逻辑说明:groups传用户维度,test_size=0.2是验证集比例。如果业务是时序场景,直接按时间切,用前 80% 训练、后 20% 验证,别用随机。这一步做错,后面融合层调得再漂亮,上线也会打回原形。

3. 文本分支:从分词到风险语义向量

3.1 中文风险文本的预处理与截断策略

风险文本往往短、口语化、带错别字,直接上大模型分词器未必最优。我一般先做一轮清洗:去 URL、去连续重复字符、统一全半角,再用预训练分词器编码。截断长度设 128 还是 256,取决于你的文本分布——评论类 128 够用,工单类建议 256。超过长度的部分直接截尾,不要截头,因为风险信号常在句尾(「要求退款」「已投诉」)。

import re from transformers import AutoTokenizer def clean_text(t): t = re.sub(r"http\S+", "", t) # 去链接 t = re.sub(r"(.)\1{3,}", r"\1\1", t) # 连续重复字压缩 return t.strip() tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese") enc = tokenizer( [clean_text(x) for x in train_df["text"]], max_length=128, truncation=True, padding="max_length", return_tensors="pt" )

逻辑说明:max_length=128是文本分支的硬参数,改大显存翻倍,改小可能丢信号,建议先用验证集跑 128/256 两档对比。padding="max_length"保证 batch 内等长,方便后续和图像特征拼接。清洗函数里的重复字压缩能减少噪声,但别过度清洗,把「好好好」压成「好好」可能改变语义强度。

3.2 用预训练模型抽风险语义向量

文本分支的产出不是分类结果,而是一个定长向量,供后面融合。常见做法是取[CLS]位置的输出,或者对最后一层做平均池化。我一般用[CLS],因为它在预训练阶段就被训练成句子级表示。

import torch from transformers import AutoModel text_encoder = AutoModel.from_pretrained("bert-base-chinese") with torch.no_grad(): out = text_encoder(**enc) text_vec = out.last_hidden_state[:, 0, :] # 取 [CLS],形状 [B, 768]

逻辑说明:last_hidden_state[:, 0, :]取的是每个样本的[CLS]向量,维度 768。如果显存紧张,可以冻结前 8 层只微调后 4 层,或者直接用句向量模型。注意torch.no_grad()只在纯推理时用,训练时要放开梯度。这一步的输出text_vec就是文本模态的「身份证」。

3.3 文本分支单独评估:先跑通再融合

很多人一上来就搭融合网络,结果文本分支本身就没调好,融合后根本不知道是谁的锅。我的习惯是先把文本分支单独训一个分类头,看它在验证集上的 AUC 和召回,达标了再进融合。

from torch import nn class TextOnly(nn.Module): def __init__(self, encoder, hidden=768): super().__init__() self.encoder = encoder self.head = nn.Linear(hidden, 2) # 二分类 def forward(self, input_ids, attention_mask): out = self.encoder(input_ids=input_ids, attention_mask=attention_mask) cls = out.last_hidden_state[:, 0, :] return self.head(cls)

逻辑说明:nn.Linear(768, 2)是分类头,输出两类 logits。训练时用交叉熵,正样本少就加pos_weight。这一步的指标是基线,融合后如果没超过它,说明图像分支在帮倒忙,要回去查图像质量或融合方式。

4. 图像分支与多模态融合层的实现

4.1 图像预处理与轻量骨干选型

风险场景的图像多是截图、证件照、商品图,分辨率参差。预处理统一到 224×224,归一化用 ImageNet 均值方差即可。骨干网络我一般选 ResNet18 或 EfficientNet-B0,理由是推理快、显存友好,风险识别不需要 ImageNet 冠军级别的容量。

from torchvision import transforms from torchvision.models import resnet18 img_tf = transforms.Compose([ transforms.Resize((224, 224)), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) img_encoder = resnet18(weights="IMAGENET1K_V1") img_encoder.fc = nn.Identity() # 去掉原分类头,输出 512 维特征

逻辑说明:Resize((224,224))是硬参数,和骨干输入绑定。fc = nn.Identity()把 ResNet 最后的全连接换成恒等映射,输出 512 维图像向量。weights="IMAGENET1K_V1"用预训练权重,小样本场景下比从头训稳得多。如果图像里有大量文字(截图类),可以额外接一个 OCR 分支,但那是另一个工程量级,先别贪。

4.2 早期融合、晚期融合怎么选

融合方式决定模型上限。早期融合是文本向量和图像向量直接拼接后过分类头;晚期融合是两路各自出分再加权。风险识别里我倾向早期融合,因为文本和图像的交互信号(图文矛盾)只有在特征层才能被学到。

class FusionModel(nn.Module): def __init__(self, text_encoder, img_encoder, text_dim=768, img_dim=512): super().__init__() self.text_encoder = text_encoder self.img_encoder = img_encoder self.fc = nn.Sequential( nn.Linear(text_dim + img_dim, 256), nn.ReLU(), nn.Dropout(0.3), nn.Linear(256, 2) ) def forward(self, input_ids, attention_mask, images): t = self.text_encoder(input_ids=input_ids, attention_mask=attention_mask).last_hidden_state[:, 0, :] v = self.img_encoder(images) x = torch.cat([t, v], dim=1) # 拼接:768 + 512 = 1280 return self.fc(x)

逻辑说明:torch.cat([t, v], dim=1)是早期融合的核心,拼接后维度 1280。Dropout(0.3)防过拟合,样本少时可提到 0.5。如果两路量纲差异大,拼接前各加一层LayerNorm会更稳。晚期融合实现简单但学不到交互,图文矛盾类风险会漏。

4.3 训练循环与类别不平衡处理

风险样本少,损失函数要动。常见做法是交叉熵加pos_weight,或者换 Focal Loss。我一般先用带权交叉熵,简单可控。

from torch.optim import AdamW pos_weight = torch.tensor([neg_count / pos_count]) # 负正比 criterion = nn.CrossEntropyLoss(weight=torch.tensor([1.0, pos_weight.item()])) optimizer = AdamW(model.parameters(), lr=2e-5, weight_decay=0.01) for epoch in range(5): model.train() for batch in train_loader: optimizer.zero_grad() logits = model(batch["input_ids"], batch["attention_mask"], batch["image"]) loss = criterion(logits, batch["label"]) loss.backward() optimizer.step()

逻辑说明:pos_weight用负正样本比,正样本越少权重越大,但别超过 20,否则模型全预测为正。lr=2e-5是预训练模型微调的常用学习率,融合层可以单独设大一点(如 1e-3)。训练轮数 5 起步,看验证集 AUC 早停。

5. 避坑与排查:多模态风险识别最常见的 5 个翻车点

5.1 指标虚高:验证集和训练集共享了同一来源

现象:验证集 AUC 0.98,上线后掉到 0.6。原因:按行随机划分,同一用户的文本和图像同时进了训练和验证。解决:按用户 ID 或时间做分组划分,重跑一遍指标,通常 AUC 会回落到真实水平。

5.2 图像分支拖后腿:模型学会了「有图=正常」

现象:融合后召回反而低于纯文本基线。原因:正常样本几乎都有图,风险样本图像缺失多,模型把「有无图」当成了强特征。解决:对缺失图像样本做掩码,或在训练时随机丢弃部分图像输入,逼模型不依赖单一模态。

5.3 显存爆炸:batch 里文本和图像同时加载

现象:batch_size 设 32 直接 OOM。原因:文本编码器和图像编码器同时前向,激活值叠加。解决:文本和图像分开编码再拼接,或者用梯度累积把等效 batch 做大,实际 batch 设 8 到 16。

5.4 分词截断把风险信号截没了

现象:长工单文本的风险召回低。原因:max_length=128截尾,而风险描述常在末尾。解决:改截头保留尾部,或把长度提到 256,用验证集确认召回变化。

5.5 融合层过拟合:训练集 loss 降验证集不降

现象:训练 3 轮后验证 loss 反弹。原因:融合层参数多、样本少。解决:加 Dropout、加 weight_decay、冻结文本编码器前若干层,或者先训文本分支再解冻融合。

6. 把风险分做成可解释输出与上线前的验证习惯

模型出分只是第一步,业务方要的是「为什么判风险」。我一般会在融合层后加一个简单归因:文本侧看注意力权重最高的几个词,图像侧看 Grad-CAM 高亮区域,两者拼成一句人话解释。这不是为了炫技,是为了让运营能复核、能申诉。

# 文本侧:取注意力权重最高的 token 作为关键词 attn = out.attentions[-1].mean(dim=1) # 平均多头 weights = attn[0, 0, :] # [CLS] 对各 token 的关注 topk = torch.topk(weights, k=5).indices keywords = tokenizer.convert_ids_to_tokens(enc["input_ids"][0][topk])

逻辑说明:attentions[-1]取最后一层注意力,mean(dim=1)对多头平均,[0, 0, :]是[CLS]对所有位置的权重。k=5取前五关键词,够业务看即可。图像侧用pytorch-grad-cam生成热力图,叠加原图输出。注意注意力权重不等于因果,只作参考。

上线前我固定做三件事:一是用分组划分的验证集跑一遍,确认 AUC 和召回;二是构造一批图文矛盾的对抗样本(文字正面、图像负面),看模型是否真学到了跨模态信号;三是把纯文本、纯图像、融合三组指标并排打表,确认融合确实有增益。

验证项纯文本纯图像融合判断标准
验证集 AUC0.820.710.89融合需高于两者
风险召回0.650.520.78召回优先于精确
对抗样本准确率0.550.600.80融合需明显领先

这张表是我每次迭代必看的,任何一列融合没赢,就先别急着上线。最后说个习惯:多模态项目的后悔药是数据版本管理,文本、图像、标签三者任一变动都要打版本号,否则两周后你根本复现不出当时那个 0.89。我吃过这个亏,现在每个实验目录里必存一份数据快照的哈希。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询