图中的输入的数据维度为8，卷积核的维度为5，卷积后输出的数据维度为8−5+1=4。
如果卷积核数量仍为1，输入数据的channel数量变为16，即输入数据维度为8×16。这里channel的概念相当于自然语言处理中的embedding，而该输入数据代表8个单词，其中每个单词的词向量维度大小为16。在这种情况下，过滤器的维度由5变为5×16，最终输出的数据维度仍为4。
如果过滤器数量为n，那么输出的数据维度就变为4×n。
一维卷积常用于序列模型，自然语言处理领域。

二维卷积运算

图中的输入的数据维度为14×14，过滤器大小为5×5，二者做卷积，输出的数据维度为10×10（14−5+1=10）。
上述内容没有引入channel的概念，也可以说channel的数量为1。如果将二维卷积中输入的channel的数量变为3，即输入的数据维度变为（14×14×3）。由于卷积操作中过滤器的channel数量必须与输入数据的channel数量相同，过滤器大小也变为5×5×3。在卷积的过程中，过滤器与数据在channel方向分别卷积，之后将卷积后的数值相加，即执行10×10次3个数值相加的操作，最终输出的数据维度为10×10。
以上都是在过滤器数量为1的情况下所进行的讨论。如果将过滤器的数量增加至16，即16个大小为10×10×3的过滤器，最终输出的数据维度就变为10×10×16。可以理解为分别执行每个过滤器的卷积操作，最后将每个卷积的输出在第三个维度（channel 维度）上进行拼接。
二维卷积常用于计算机视觉、图像处理领域。

三维卷积运算

假设输入数据的大小为a1×a2×a3，channel数为c，过滤器大小为f，即过滤器维度为f×f×f×c（一般不写channel的维度），过滤器数量为n。
基于上述情况，三维卷积最终的输出为(a1−f+1)×(a2−f+1)×(a3−f+1)×n。该公式对于一维卷积、二维卷积仍然有效，只有去掉不相干的输入数据维度就行。
三维卷积常用于医学领域（CT影响），视频处理领域（检测动作及人物行为）。

反卷积

原理

输入：2x2，卷积核：4x4，滑动步长：3，输出：7x7

即输入为2x2的图片经过4x4的卷积核进行步长为3的反卷积的过程：

输入图片每个像素进行一次full卷积，根据full卷积大小计算可以知道每个像素的卷积后大小为 1+4-1=4，即4x4大小的特征图，输入有4个像素所以4个4x4的特征图；
将4个特征图进行步长为3的fusion（即相加）；例如红色的特征图仍然是在原来输入位置（左上角），绿色还是在原来的位置（右上角），步长为3是指每隔3个像素进行fusion，重叠部分进行相加，即输出的第1行第4列是由红色特阵图的第一行第四列与绿色特征图的第一行第一列相加得到，其他如此类推。

可以看出翻卷积的大小是由卷积核大小与滑动步长决定， in是输入大小， k是卷积核大小， s是滑动步长， out是输出大小。

out = (in - 1) * s + k
上图过程就是， (2 - 1) * 3 + 4 = 7

tf.nn.conv2d_transpose

函数功能：计算反卷积（转置卷积）。

value：是输入的特征图，4维的tensor，float类型，需要进行反卷积的矩阵，格式是[n,h,w,c]，n是你这一个batch的图片数量，h和w是特征图的宽和高，c是特征图的数量；
filter：卷积核，4维，[h,w,out_channels, in_channels]，即[卷积核的高度，卷积核的宽度，卷积核个数，图像通道数]。
output_shape：反卷积操作输出的shape，细心的同学会发现卷积操作是没有这个参数的。
strides：步长，长度为4的1维向量，分别对应value四个维度的步长。
padding：string类型的量，只能是"SAME","VALID"其中之一，这个值决定了不同的卷积方式。
data_format：string类型的量，'NHWC'和'NCHW'其中之一，这是tensorflow新版本中新加的参数，它说明了value参数的数据格式。'NHWC'指tensorflow标准的数据格式[batch, height, width, in_channels]，'NCHW'指Theano的数据格式,[batch, in_channels，height, width]，默认值是'NHWC'。
name：操作的名称。

池化运算的基本概念

介绍

在构建卷积神经网络时，经常会使用池化操作，而池化层往往在卷积层后面，通过池化操作来降低卷积层输出的特征维度，同时可以防止过拟合现象。

池化操作可以降低图像维度的原因，本质上是因为图像具有一种“静态性”的属性，这个意思是说在一个图像区域有用的特征极有可能在另一个区域同样有用。因此，为了描述一个大的图像，很直观的想法就是对不同位置的特征进行聚合统计。例如，可以计算图像在固定区域上特征的平均值 (或最大值)来代表这个区域的特征。

种类

平均池化（average pooling）：计算图像区域的平均值作为该区域池化后的值。
最大池化（max pooling）：选图像区域的最大值作为该区域池化后的值。

根据相关理论，特征提取的误差主要来自两个方面：

邻域大小受限造成的估计值方差增大；
卷积层参数误差造成估计均值的偏移。

一般来说，平均池化能减小第一种误差，更多的保留图像的背景信息；最大池化能减小第二种误差，更多的保留纹理信息。

Text-CNN原理

如图所示，TextCNN模型架构，句子中每个word使用k维向量来表示，于是句子（n个word）可表示为一个n*k的矩阵，作为CNN的输入。

CNN-rand：句子中的的word vector都是随机初始化的，同时当做CNN训练过程中需要优化的参数；

CNN-static：句子中的word vector是使用word2vec预先对Google News dataset (about 100 billion words)进行训练好的词向量表中的词向量。且在CNN训练过程中作为固定的输入，不作为优化的参数;

CNN-non-static：句子中的word vector是使用word2vec预先对Google News dataset (about 100 billion words)进行训练好的词向量表中的词向量。在CNN训练过程中作为固定的输入，做为CNN训练过程中需要优化的参数；

模型架构介绍

输入层

如上图中所示，对于模型的输入是由每个句子中的词的词向量组成的矩阵作为输入层的输入N*K,其中K为词向量的长度，N为句子的长度。词向量的表示方式有3种，CNN-rand、CNN-static、CNN-non-static。对于没有出现在训练好的词向量表中的词(未登录词)的词向量，论文实验中采取的是使用随机初始化为0或者偏小的正数表示。

卷积层

在输入层的基础上，使用Filter Window进行卷积操作得到Feature Map。实验中使用的3种类型大小的Filter Window,分别是3*K,4*K,5*K，K表示词向量的长度。其中每种类型大小的Filter Window 有100个含有不同值的Filter。每一个Filter能从输入的矩阵中抽取出一个Feature Map特征，在NLP中称为文本特征。

实验中对Feature Map的池化操作方式是Max-over-time Pooling的方式，即将每个Feature Map向量中最大的一个值抽取出来,组成一个一维向量。

全连接层

该层的输入为池化操作后形成的一维向量，经过激活函数输出，再加上Dropout层防止过拟合。并在全连接层上添加l2正则化参数。

输出层

该层的输入为全连接层的输出，经过SoftMax层作为输出层，进行分类。对于多分类问题可以使用SoftMax层,对于二分类问题可以使用一个含有sigmod激活函数的神经元作为输出层，实验中采用的是SoftMax层。

Text-CNN文本分类实战

数据集

百度网盘https://pan.baidu.com/s/1hugrfRu 密码：qfud

本数据集是清华NLP组提供的THUCNews新闻文本分类数据集的一个子集（原始的数据集大约74万篇文档，训练起来需要花较长的时间）。

本次训练使用了其中的10个分类（体育, 财经, 房产, 家居, 教育, 科技, 时尚, 时政, 游戏, 娱乐），每个分类6500条，总共65000条新闻数据。

数据集划分如下：

cnews.train.txt: 训练集(50000条)

cnews.val.txt: 验证集(5000条)

cnews.test.txt: 测试集(10000条)

预处理

预处理文件preprocess.py，代码中的函数说明如下：

read_file(): 读取文件数据;
build_vocab(): 构建词汇表，使用字符级的表示，这一函数会将词汇表存储下来，避免每一次重复处理;
read_vocab(): 读取上一步存储的词汇表，转换为{词：id}表示;
read_category(): 将分类目录固定，转换为{类别: id}表示;
to_words(): 将一条由id表示的数据重新转换为文字;
preocess_file(): 将数据集从文字转换为固定长度的id序列表示;
batch_iter(): 为神经网络的训练准备经过shuffle的批次的数据。


# coding: utf-8
 
import sys
from collections import Counter
 
import numpy as np
import tensorflow.contrib.keras as kr
 
if sys.version_info[0] > 2:
    is_py3 = True
else:
    reload(sys)
    sys.setdefaultencoding("utf-8")
    is_py3 = False
 
 
def native_word(word, encoding='utf-8'):
    """如果在python2下面使用python3训练的模型，可考虑调用此函数转化一下字符编码"""
    if not is_py3:
        return word.encode(encoding)
    else:
        return word
 
 
def native_content(content):
    if not is_py3:
        return content.decode('utf-8')
    else:
        return content
 
 
def open_file(filename, mode='r'):
    """
    常用文件操作，可在python2和python3间切换.
    mode: 'r' or 'w' for read or write
    """
    if is_py3:
        return open(filename, mode, encoding='utf-8', errors='ignore')
    else:
        return open(filename, mode)
 
 
def read_file(filename):
    """读取文件数据"""
    contents, labels = [], []
    with open_file(filename) as f:
        for line in f:
            try:
                label, content = line.strip().split('\t')
                if content:
                    contents.append(list(native_content(content)))
                    labels.append(native_content(label))
            except:
                pass
    return contents, labels
 
 
def build_vocab(train_dir, vocab_dir, vocab_size=5000):
    """根据训练集构建词汇表，存储"""
    data_train, _ = read_file(train_dir)
 
    all_data = []
    for content in data_train:
        all_data.extend(content)
 
    counter = Counter(all_data)
    count_pairs = counter.most_common(vocab_size - 1)
    words, _ = list(zip(*count_pairs))
    # 添加一个 <PAD> 来将所有文本pad为同一长度
    words = ['<PAD>'] + list(words)
    open_file(vocab_dir, mode='w').write('\n'.join(words) + '\n')
 
 
def read_vocab(vocab_dir):
    """读取词汇表"""
    # words = open_file(vocab_dir).read().strip().split('\n')
    with open_file(vocab_dir) as fp:
        # 如果是py2 则每个值都转化为unicode
        words = [native_content(_.strip()) for _ in fp.readlines()]
    word_to_id = dict(zip(words, range(len(words))))
    return words, word_to_id
 
 
def read_category():
    """读取分类目录，固定"""
    categories = ['体育', '财经', '房产', '家居', '教育', '科技', '时尚', '时政', '游戏', '娱乐']
 
    categories = [native_content(x) for x in categories]
 
    cat_to_id = dict(zip(categories, range(len(categories))))
 
    return categories, cat_to_id
 
 
def to_words(content, words):
    """将id表示的内容转换为文字"""
    return ''.join(words[x] for x in content)
 
 
def process_file(filename, word_to_id, cat_to_id, max_length=600):
    """将文件转换为id表示"""
    contents, labels = read_file(filename)
 
    data_id, label_id = [], []
    for i in range(len(contents)):
        data_id.append([word_to_id[x] for x in contents[i] if x in word_to_id])
        label_id.append(cat_to_id[labels[i]])
 
    # 使用keras提供的pad_sequences来将文本pad为固定长度
    x_pad = kr.preprocessing.sequence.pad_sequences(data_id, max_length)
    y_pad = kr.utils.to_categorical(label_id, num_classes=len(cat_to_id))  # 将标签转换为one-hot表示
 
    return x_pad, y_pad
 
 
def batch_iter(x, y, batch_size=64):
    """生成批次数据"""
    data_len = len(x)
    num_batch = int((data_len - 1) / batch_size) + 1
 
    indices = np.random.permutation(np.arange(data_len))
    x_shuffle = x[indices]
    y_shuffle = y[indices]
 
    for i in range(num_batch):
        start_id = i * batch_size
        end_id = min((i + 1) * batch_size, data_len)
        yield x_shuffle[start_id:end_id], y_shuffle[start_id:end_id]

经过数据预处理，数据的格式如下：

模型结构及代码

模型结构

cnn_model.py


# coding: utf-8
 
import tensorflow as tf
 
 
class TCNNConfig(object):
    """CNN配置参数"""
 
    embedding_dim = 64  # 词向量维度
    seq_length = 600  # 序列长度
    num_classes = 10  # 类别数
    num_filters = 256  # 卷积核数目
    kernel_size = 5  # 卷积核尺寸
    vocab_size = 5000  # 词汇表达小
 
    hidden_dim = 128  # 全连接层神经元
 
    dropout_keep_prob = 0.5  # dropout保留比例
    learning_rate = 1e-3  # 学习率
 
    batch_size = 64  # 每批训练大小
    num_epochs = 10  # 总迭代轮次
 
    print_per_batch = 100  # 每多少轮输出一次结果
    save_per_batch = 10  # 每多少轮存入tensorboard
 
 
class TextCNN(object):
    """文本分类，CNN模型"""
 
    def __init__(self, config):
        self.config = config
 
        # 三个待输入的数据
        self.input_x = tf.placeholder(tf.int32, [None, self.config.seq_length], name='input_x')
        self.input_y = tf.placeholder(tf.float32, [None, self.config.num_classes], name='input_y')
        self.keep_prob = tf.placeholder(tf.float32, name='keep_prob')
 
        self.cnn()
 
    def cnn(self):
        """CNN模型"""
        # 词向量映射
        with tf.device('/cpu:0'):
            embedding = tf.get_variable('embedding', [self.config.vocab_size, self.config.embedding_dim])
            embedding_inputs = tf.nn.embedding_lookup(embedding, self.input_x)
 
        with tf.name_scope("cnn"):
            # CNN layer
            conv = tf.layers.conv1d(embedding_inputs, self.config.num_filters, self.config.kernel_size, name='conv')
            # global max pooling layer
            gmp = tf.reduce_max(conv, reduction_indices=[1], name='gmp')
 
        with tf.name_scope("score"):
            # 全连接层，后面接dropout以及relu激活
            fc = tf.layers.dense(gmp, self.config.hidden_dim, name='fc1')
            fc = tf.contrib.layers.dropout(fc, self.keep_prob)
            fc = tf.nn.relu(fc)
 
            # 分类器
            self.logits = tf.layers.dense(fc, self.config.num_classes, name='fc2')
            self.y_pred_cls = tf.argmax(tf.nn.softmax(self.logits), 1)  # 预测类别
 
        with tf.name_scope("optimize"):
            # 损失函数，交叉熵
            cross_entropy = tf.nn.softmax_cross_entropy_with_logits(logits=self.logits, labels=self.input_y)
            self.loss = tf.reduce_mean(cross_entropy)
            # 优化器
            self.optim = tf.train.AdamOptimizer(learning_rate=self.config.learning_rate).minimize(self.loss)
 
        with tf.name_scope("accuracy"):
            # 准确率
            correct_pred = tf.equal(tf.argmax(self.input_y, 1), self.y_pred_cls)
            self.acc = tf.reduce_mean(tf.cast(correct_pred, tf.float32))

训练

若之前进行过训练，请把tensorboard/textcnn删除，避免TensorBoard多次训练结果重叠。

cmd进入cnn_train.py所在路径，输入python cnn_train.py train，开始训练。

cnn_train.py代码如下：


# -*- coding: utf-8 -*-
 
from __future__ import print_function
 
import os
import sys
import time
from datetime import timedelta
 
import numpy as np
import tensorflow as tf
from sklearn import metrics
 
from cnn_model import TCNNConfig, TextCNN
from preprocess import read_vocab, read_category, batch_iter, process_file, build_vocab
 
base_dir = 'data/cnews'
train_dir = os.path.join(base_dir, 'cnews.train.txt')
test_dir = os.path.join(base_dir, 'cnews.test.txt')
val_dir = os.path.join(base_dir, 'cnews.val.txt')
vocab_dir = os.path.join(base_dir, 'cnews.vocab.txt')
 
save_dir = 'checkpoints/textcnn'
save_path = os.path.join(save_dir, 'best_validation')  # 最佳验证结果保存路径
 
 
def get_time_dif(start_time):
    """获取已使用时间"""
    end_time = time.time()
    time_dif = end_time - start_time
    return timedelta(seconds=int(round(time_dif)))
 
 
def feed_data(x_batch, y_batch, keep_prob):
    feed_dict = {
        model.input_x: x_batch,
        model.input_y: y_batch,
        model.keep_prob: keep_prob
    }
    return feed_dict
 
 
def evaluate(sess, x_, y_):
    """评估在某一数据上的准确率和损失"""
    data_len = len(x_)
    batch_eval = batch_iter(x_, y_, 128)
    total_loss = 0.0
    total_acc = 0.0
    for x_batch, y_batch in batch_eval:
        batch_len = len(x_batch)
        feed_dict = feed_data(x_batch, y_batch, 1.0)
        loss, acc = sess.run([model.loss, model.acc], feed_dict=feed_dict)
        total_loss += loss * batch_len
        total_acc += acc * batch_len
 
    return total_loss / data_len, total_acc / data_len
 
 
def train():
    print("Configuring TensorBoard and Saver...")
    # 配置 Tensorboard，重新训练时，请将tensorboard文件夹删除，不然图会覆盖
    tensorboard_dir = 'tensorboard/textcnn'
    if not os.path.exists(tensorboard_dir):
        os.makedirs(tensorboard_dir)
 
    tf.summary.scalar("loss", model.loss)
    tf.summary.scalar("accuracy", model.acc)
    merged_summary = tf.summary.merge_all()
    writer = tf.summary.FileWriter(tensorboard_dir)
 
    # 配置 Saver
    saver = tf.train.Saver()
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
 
    print("Loading training and validation data...")
    # 载入训练集与验证集
    start_time = time.time()
    x_train, y_train = process_file(train_dir, word_to_id, cat_to_id, config.seq_length)
    x_val, y_val = process_file(val_dir, word_to_id, cat_to_id, config.seq_length)
    time_dif = get_time_dif(start_time)
    print("Time usage:", time_dif)
 
    # 创建session
    session = tf.Session()
    session.run(tf.global_variables_initializer())
    writer.add_graph(session.graph)
 
    print('Training and evaluating...')
    start_time = time.time()
    total_batch = 0  # 总批次
    best_acc_val = 0.0  # 最佳验证集准确率
    last_improved = 0  # 记录上一次提升批次
    require_improvement = 1000  # 如果超过1000轮未提升，提前结束训练
 
    flag = False
    for epoch in range(config.num_epochs):
        print('Epoch:', epoch + 1)
        batch_train = batch_iter(x_train, y_train, config.batch_size)
        for x_batch, y_batch in batch_train:
            feed_dict = feed_data(x_batch, y_batch, config.dropout_keep_prob)
 
            if total_batch % config.save_per_batch == 0:
                # 每多少轮次将训练结果写入tensorboard scalar
                s = session.run(merged_summary, feed_dict=feed_dict)
                writer.add_summary(s, total_batch)
 
            if total_batch % config.print_per_batch == 0:
                # 每多少轮次输出在训练集和验证集上的性能
                feed_dict[model.keep_prob] = 1.0
                loss_train, acc_train = session.run([model.loss, model.acc], feed_dict=feed_dict)
                loss_val, acc_val = evaluate(session, x_val, y_val)  # todo
 
                if acc_val > best_acc_val:
                    # 保存最好结果
                    best_acc_val = acc_val
                    last_improved = total_batch
                    saver.save(sess=session, save_path=save_path)
                    improved_str = '*'
                else:
                    improved_str = ''
 
                time_dif = get_time_dif(start_time)
                msg = 'Iter: {0:>6}, Train Loss: {1:>6.2}, Train Acc: {2:>7.2%},' \
                      + ' Val Loss: {3:>6.2}, Val Acc: {4:>7.2%}, Time: {5} {6}'
                print(msg.format(total_batch, loss_train, acc_train, loss_val, acc_val, time_dif, improved_str))
 
            session.run(model.optim, feed_dict=feed_dict)  # 运行优化
            total_batch += 1
 
            if total_batch - last_improved > require_improvement:
                # 验证集正确率长期不提升，提前结束训练
                print("No optimization for a long time, auto-stopping...")
                flag = True
                break  # 跳出循环
        if flag:  # 同上
            break
 
 
def test():
    print("Loading test data...")
    start_time = time.time()
    x_test, y_test = process_file(test_dir, word_to_id, cat_to_id, config.seq_length)
 
    session = tf.Session()
    session.run(tf.global_variables_initializer())
    saver = tf.train.Saver()
    saver.restore(sess=session, save_path=save_path)  # 读取保存的模型
 
    print('Testing...')
    loss_test, acc_test = evaluate(session, x_test, y_test)
    msg = 'Test Loss: {0:>6.2}, Test Acc: {1:>7.2%}'
    print(msg.format(loss_test, acc_test))
 
    batch_size = 128
    data_len = len(x_test)
    num_batch = int((data_len - 1) / batch_size) + 1
 
    y_test_cls = np.argmax(y_test, 1)
    y_pred_cls = np.zeros(shape=len(x_test), dtype=np.int32)  # 保存预测结果
    for i in range(num_batch):  # 逐批次处理
        start_id = i * batch_size
        end_id = min((i + 1) * batch_size, data_len)
        feed_dict = {
            model.input_x: x_test[start_id:end_id],
            model.keep_prob: 1.0
        }
        y_pred_cls[start_id:end_id] = session.run(model.y_pred_cls, feed_dict=feed_dict)
 
    # 评估
    print("Precision, Recall and F1-Score...")
    print(metrics.classification_report(y_test_cls, y_pred_cls, target_names=categories))
 
    # 混淆矩阵
    print("Confusion Matrix...")
    cm = metrics.confusion_matrix(y_test_cls, y_pred_cls)
    print(cm)
 
    time_dif = get_time_dif(start_time)
    print("Time usage:", time_dif)
 
 
if __name__ == '__main__':
    if len(sys.argv) != 2 or sys.argv[1] not in ['train', 'test']:
        raise ValueError("""usage: python run_cnn.py [train / test]""")
 
    print('Configuring CNN model...')
    config = TCNNConfig()
    if not os.path.exists(vocab_dir):  # 如果不存在词汇表，重建
        build_vocab(train_dir, vocab_dir, config.vocab_size)
    categories, cat_to_id = read_category()
    words, word_to_id = read_vocab(vocab_dir)
    config.vocab_size = len(words)
    model = TextCNN(config)
    # train()
    
    if sys.argv[1] == 'train':
        train()
    else:
        test()

参考

Deep Learning读书笔记（6）–【卷积神经网络】 https://www.colabug.com/2430279.html

卷积神经网络（CNN）之一维卷积、二维卷积、三维卷积详解 https://www.cnblogs.com/szxspark/p/8445406.html

卷积神经网络CNN（1）——图像卷积与反卷积（后卷积，转置卷积） https://blog.csdn.net/Fate_fjh/article/details/52882134

CNN中的卷积、反卷积与反池化 https://blog.csdn.net/qinghuaci666/article/details/80848642

Tensorflow中tf.conv2d_transpose()的操作 https://blog.csdn.net/mieleizhi0522/article/details/80441571

池化种类以及区别 https://blog.csdn.net/yangqingse/article/details/79841889

Tensorflow实现CNN文本分类(详细解释及TextCNN代码解释) https://blog.csdn.net/chivalrousli/article/details/77168397

卷积神经网络(TextCNN)在句子分类上的实现 https://blog.csdn.net/u014732537/article/details/79573174

CNN字符级中文文本分类（tensorflow） https://blog.csdn.net/u011439796/article/details/77692621

CNN字符级中文文本分类（tensorflow） https://github.com/gaussic/text-classification-cnn-rnn

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/凡人多烦事01/article/detail/239110