2022 年 Python3 网络爬虫教程

大家好，我是崔庆才，由于爬虫技术不断迭代升级，一些旧的教程已经过时、案例已经过期，最前沿的爬虫技术比如异步、JavaScript 逆向、安卓逆向、智能解析、WebAssembly、大规模分布式、Kubernetes 等技术层出不穷，我最近新出了一套最新最全面的 Python3 网络爬虫系列教程。

博主自荐：截止 2022 年，可以将最前沿最全面的爬虫技术都涵盖的教程，如异步、JavaScript 逆向、安卓逆向、智能解析、WebAssembly、大规模分布式、Kubernetes 等，市面上目前就这一套了。

最新教程对旧的爬虫技术内容进行了全面更新，搭建了全新的案例平台进行全面讲解，保证案例稳定有效不过期。

教程请移步：

【2022 版】Python3 网络爬虫学习教程

2018 年 Python3 爬虫系列教程

以下为 2018 年版 Python3 网络爬虫系列教程

本内容来自于《Python3 网络爬虫开发实战》一书。 书籍购买地址： https://item.jd.com/12333540.html

本书通过多个实战案例详细介绍了 Python3 网络爬虫的知识，本书由图灵教育-人民邮电出版社出版发行，版权所有，禁止转载。

• 1-开发环境配置
• 1.1-Python3 的安装
• 1.2-请求库的安装
• 1.2.1-Requests 的安装
• 1.2.2-Selenium 的安装
• 1.2.3-ChromeDriver 的安装
• 1.2.4-GeckoDriver 的安装
• 1.2.5-PhantomJS 的安装
• 1.2.6-aiohttp 的安装
• 1.3-解析库的安装
• 1.3.1-lxml 的安装
• 1.3.2-Beautiful Soup 的安装
• 1.3.3-pyquery 的安装
• 1.3.4-tesserocr 的安装
• 1.4-数据库的安装
• 1.4.1-MySQL 的安装
• 1.4.2-MongoDB 安装
• 1.4.3-Redis 的安装
• 1.5-存储库的安装
• 1.5.1-PyMySQL 的安装
• 1.5.2-PyMongo 的安装
• 1.5.3-redis-py 的安装
• 1.5.4-RedisDump 的安装
• 1.6-Web 库的安装
• 1.6.1-Flask 的安装
• 1.6.2-Tornado 的安装
• 1.7-App 爬取相关库的安装
• 1.7.1-Charles 的安装
• 1.7.2-mitmproxy 的安装
• 1.7.3-Appium 的安装
• 1.8-爬虫框架的安装
• 1.8.1-pyspider 的安装
• 1.8.2-Scrapy 的安装
• 1.8.3-Scrapy-Splash 的安装
• 1.8.4-Scrapy-Redis 的安装
• 1.9-部署相关库的安装
• 1.9.1-Docker 的安装
• 1.9.2-Scrapyd 的安装
• 1.9.3-Scrapyd-Client 的安装
• 1.9.4-Scrapyd API 的安装
• 1.9.5-Scrapyrt 的安装
• 1.9.6-Gerapy 的安装
• 2-爬虫基础
• 2.1-HTTP 基本原理
• 2.2-网页基础
• 2.3-爬虫的基本原理
• 2.4-会话和 Cookies
• 2.5-代理的基本原理
• 3-基本库的使用
• 3.1-使用 urllib
• 3.1.1-发送请求
• 3.1.2-处理异常
• 3.1.3-解析链接
• 3.1.4-分析 Robots 协议
• 3.2-使用 requests
• 3.2.1-基本用法
• 3.2.2-高级用法
• 3.3-正则表达式
• 3.4-抓取猫眼电影排行
• 4-解析库的使用
• 4.1-使用 XPath
• 4.2-使用 Beautiful Soup
• 4.3-使用 pyquery
• 5-数据存储
• 5.1-文件存储
• 5.1.1-TXT 文本存储
• 5.1.2-JSON 文件存储
• 5.1.3-CSV 文件存储
• 5.2-关系型数据库存储
• 5.2.1-MySQL 存储
• 5.3-非关系型数据库存储
• 5.3.1-MongoDB 存储
• 5.3.2-Redis 存储
• 6-Ajax 数据爬取
• 6.1-什么是 Ajax
• 6.2-Ajax 分析方法
• 6.3-Ajax 结果提取
• 6.4-分析 Ajax 爬取今日头条街拍美图
• 7-动态渲染页面爬取
• 7.1-Selenium 的使用
• 7.2-Splash 的使用
• 7.3-Splash 负载均衡配置
• 7.4-使用 Selenium 爬取淘宝商品
• 8-验证码的识别
• 8.1-图形验证码的识别
• 8.2-极验滑动验证码的识别
• 8.3-点触验证码的识别
• 8.4-微博宫格验证码的识别
• 9-代理的使用
• 9.1-代理的设置
• 9.2-代理池的维护
• 9.3-付费代理的使用
• 9.4-ADSL 拨号代理
• 9.5-使用代理爬取微信公众号文章
• 10-模拟登录
• 10.1-模拟登录并爬取 GitHub
• 10.2-Cookies 池的搭建
• 11-App 的爬取
• 11.1-Charles 的使用
• 11.2-mitmproxy 的使用
• 11.3-mitmdump 爬取“得到”App 电子书信息
• 11.4-Appium 的基本使用
• 11.5-Appium 爬取微信朋友圈
• 11.6-Appium+mitmdump 爬取京东商品
• 12-pyspider 框架的使用
• 12.1-pyspider 框架介绍
• 12.2-pyspider 的基本使用
• 12.3-pyspider 用法详解
• 13-Scrapy 框架的使用
• 13.1-Scrapy 框架介绍
• 13.2-Scrapy 入门
• 13.3-Selector 的用法
• 13.4-Spider 的用法
• 13.5-Downloader Middleware 的用法
• 13.6-Spider Middleware 的用法
• 13.7-Item Pipeline 的用法
• 13.8-Scrapy 对接 Selenium
• 13.9-Scrapy 对接 Splash
• 13.10-Scrapy 通用爬虫
• 13.11-Scrapyrt 的使用
• 13.12-Scrapy 对接 Docker
• 13.13-Scrapy 爬取新浪微博
• 14-分布式爬虫
• 14.1-分布式爬虫原理
• 14.2-Scrapy-Redis 源码解析
• 14.3-Scrapy 分布式实现
• 14.4-Bloom Filter 的对接
• 15-分布式爬虫的部署
• 15.1-Scrapyd 分布式部署
• 15.2-Scrapyd-Client 的使用
• 15.3-Scrapyd 对接 Docker
• 15.4-Scrapyd 批量部署
• 15.5-Gerapy 分布式管理

前几天，大才发了一个自己写的框架，介绍地址在这里， GIT 地址在这里。

今天在阿里云上试用了一下，在这里做一个简单的说明。

1、配置环境

阿里云的版本是 2.7.5，所以用 pyenv 新安装了一个 3.6.4 的环境，安装后使用 pyenv global 3.6.4 即可使用 3.6.4 的环境，我个人比较喜欢这样，切换自如，互不影响。 如下图： 接下来按照大才的文章，pip install gerapy 即可，这一步没有遇到什么问题。有问题的同学可以向大才提 issue。

2. 开启服务

首先去阿里云的后台设置安全组 ，我的是这样： 然后到命令窗口对 8000 和 6800 端口放行即可。 接着执行

gerapy init cd gerapy gerapy migrate # 注意下一步 gerapy runserver 0.0.0.0:8000 【如果你是在本地，执行 gerapy runserver 即可，如果你是在阿里云上，你就要改成前面这样来执行】

现在在浏览器里访问：ip:8000 应该就可以看到主界面了 里面的各个的含义见大才的文章。

3.创建项目

在 gerapy 下的 projects 里面新建一个 scrapy 爬虫，在这里我搞的是最简单的：

scrapy startproject gerapy_test cd gerapy_test scrapy genspider baidu www.baidu.com

这样就是一个最简单的爬虫了，修改一个 settings.py 中的 ROBOTSTXT_OBEY=False, 然后修改一个 spiders 下面的 baidu.py, 这里随意，我这里设置的是输出返回的 response.url

4.安装 scrapyd

pip install scrapyd

安装好以后，命令行执行

scrapyd

然后浏览器中打开 ip:6800，如果你没有修改配置，应该这里会打不开，clients 那里配置的时候，也应该会显示为 error，就像这样： 后来找了一下原因发现 scrapyd 默认打开的也是 127.0.0.1 所以这个时候就要改一下配置，具体可以参考这里， 我是这么修改：

vim ~/.scrapyd.conf [scrapyd] bind_address = 0.0.0.0

在刷新一下，就会看到前面 error 变成了 normal

5. 打包，部署，调度

这几步大才的文章里都有详细说明，打包完，部署，在进入 clients 的调度界面，点击 run 按钮即可跑爬虫了 可以看到输出的结果了。

6.结语

建议大家可以试着用一下，很方便，我这里只是很简单的使用了一下。

本节我们来尝试使用 TensorFlow 搭建一个双向 LSTM (Bi-LSTM) 深度学习模型来处理序列标注问题，主要目的是学习 Bi-LSTM 的用法。

Bi-LSTM

我们知道 RNN 是可以学习到文本上下文之间的联系的，输入是上文，输出是下文，但这样的结果是模型可以根据上文推出下文，而如果输入下文，想要推出上文就没有那么简单了，为了弥补这个缺陷，我们可以让模型从两个方向来学习，这就构成了双向 RNN。在某些任务中，双向 RNN 的表现比单向 RNN 要好，本文要实现的文本分词就是其中之一。不过本文使用的模型不是简单的双向 RNN，而是 RNN 的变种 — LSTM。 如图所示为 Bi-LSTM 的基本原理，输入层的数据会经过向前和向后两个方向推算，最后输出的隐含状态再进行 concat，再作为下一层的输入，原理其实和 LSTM 是类似的，就是多了双向计算和 concat 过程。

数据处理

本文的训练和测试数据使用的是已经做好序列标注的中文文本数据。序列标注，就是给一个汉语句子作为输入，以“BEMS”组成的序列串作为输出，然后再进行切词，进而得到输入句子的划分。其中，B 代表该字是词语中的起始字，M 代表是词语中的中间字，E 代表是词语中的结束字，S 则代表是单字成词。 这里的原始数据样例如下：

1

人/b  们/e  常/s  说/s  生/b  活/e  是/s  一/s  部/s  教/b  科/m  书/e

这里一个字对应一个标注，我们首先需要对数据进行预处理，预处理的流程如下：

• 将句子切分
• 将句子的的标点符号去掉
• 将每个字及对应的标注切分
• 去掉长度为 0 的无效句子

首先我们将句子切分开来并去掉标点符号，代码实现如下：

1
2
3
4
5
6
7
8

# Read origin data
text = open('data/data.txt', encoding='utf-8').read()
# Get split sentences
sentences = re.split('[，。！？、‘’“”]/[bems]', text)
# Filter sentences whose length is 0
sentences = list(filter(lambda x: x.strip(), sentences))
# Strip sentences
sentences = list(map(lambda x: x.strip(), sentences))

这样我们就可以将句子切分开来并做好了清洗，接下来我们还需要把每个句子中的字及标注转为 Numpy 数组，便于下一步制作词表和数据集，代码实现如下：

1
2
3
4
5
6
7
8
9
10
11
12

import re
# To numpy array
words, labels = [], []
print('Start creating words and labels...')
for sentence in sentences:
    groups = re.findall('(.)/(.)', sentence)
    arrays = np.asarray(groups)
    words.append(arrays[:, 0])
    labels.append(arrays[:, 1])
print('Words Length', len(words), 'Labels Length', len(labels))
print('Words Example', words[0])
print('Labels Example', labels[0])

这里我们利用正则 re 库的 findall() 方法将字及标注分开，并分别添加到 words 和 labels 数组中，运行效果如下：

1
2
3

Words Length 321533 Labels Length 321533
Words Example ['人' '们' '常' '说' '生' '活' '是' '一' '部' '教' '科' '书']
Labels Example ['b' 'e' 's' 's' 'b' 'e' 's' 's' 's' 'b' 'm' 'e']

接下来我们有了这些数据就要开始制作词表了，词表制作起来无非就是输入词表和输出词表的不重复的正逆对应，制作词表的目的就是将输入的文字或标注转为 index，同时还能反向根据 index 获取对应的文字或标注，所以我们这里需要制作 word2id、id2word、tag2id、id2tag 四个字典。 为了解决 OOV 问题，我们还需要将无效字符也进行标注，这里我们统一取 0。制作时我们借助于 pandas 库的 Series 进行了去重和转换，另外还限制了每一句的最大长度，这里设置为 32，如果大于32，则截断，否则进行 padding，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

from itertools import chain
import pandas as pd
import numpy as np
# Merge all words
all_words = list(chain(*words))
# All words to Series
all_words_sr = pd.Series(all_words)
# Get value count, index changed to set
all_words_counts = all_words_sr.value_counts()
# Get words set
all_words_set = all_words_counts.index
# Get words ids
all_words_ids = range(1, len(all_words_set) + 1)

# Dict to transform
word2id = pd.Series(all_words_ids, index=all_words_set)
id2word = pd.Series(all_words_set, index=all_words_ids)

# Tag set and ids
tags_set = ['x', 's', 'b', 'm', 'e']
tags_ids = range(len(tags_set))

# Dict to transform
tag2id = pd.Series(tags_ids, index=tags_set)
id2tag = pd.Series(tags_set, index=tag2id)

max_length = 32

def x_transform(words):
    ids = list(word2id[words])
    if len(ids) >= max_length:
        ids = ids[:max_length]
    ids.extend([0] * (max_length - len(ids)))
    return ids

def y_transform(tags):
    ids = list(tag2id[tags])
    if len(ids) >= max_length:
        ids = ids[:max_length]
    ids.extend([0] * (max_length - len(ids)))
    return ids

print('Starting transform...')
data_x = list(map(lambda x: x_transform(x), words))
data_y = list(map(lambda y: y_transform(y), labels))
data_x = np.asarray(data_x)
data_y = np.asarray(data_y)

这样我们就完成了 word2id、id2word、tag2id、id2tag 四个字典的制作，并制作好了 Numpy 数组类型的 data_x 和 data_y，这里 data_x 和 data_y 单句示例如下：

1
2

Data X Example: [8, 43, 320, 88, 36, 198, 7, 2, 41, 163, 124, 245, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Data Y Example: [2, 4, 1, 1, 2, 4, 1, 1, 1, 2, 3, 4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]

可以看到数据的 x 部分，原始文字和标注结果都转化成了词表中的 index，同时不够 32 个字符就以 0 补全。 接下来我们将其保存成 pickle 文件，以备训练和测试使用：

1
2
3
4
5
6
7
8
9

print('Starting pickle to file...')
with open(join(path, 'data.pkl'), 'wb') as f:
    pickle.dump(data_x, f)
    pickle.dump(data_y, f)
    pickle.dump(word2id, f)
    pickle.dump(id2word, f)
    pickle.dump(tag2id, f)
    pickle.dump(id2tag, f)
print('Pickle finished')

好，现在数据预处理部分就完成了。

构造模型

接下来我们就需要利用 pickle 文件中的数据来构建模型了，首先进行 pickle 文件的读取，然后将数据分为训练集、开发集、测试集，详细流程不再赘述，赋值为如下变量：

1
2
3
4

# Load data
    data_x, data_y, word2id, id2word, tag2id, id2tag = load_data()
    # Split data
    train_x, train_y, dev_x, dev_y, test_x, test_y = get_data(data_x, data_y)

接下来我们使用 TensorFlow 自带的 Dataset 数据结构构造输入输出，利用 Dataset 我们可以构造一个 iterator 迭代器，每调用一次 get_next() 方法，我们就可以得到一个 batch，这里 Dataset 的初始化我们使用 from_tensor_slices() 方法，然后调用其 batch() 方法来初始化每个数据集的 batch_size，接着初始化同一个 iterator，并绑定到三个数据集上声明为三个 initializer，这样每调用 initializer，就会将 iterator 切换到对应的数据集上，代码实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# Train and dev dataset
train_dataset = tf.data.Dataset.from_tensor_slices((train_x, train_y))
train_dataset = train_dataset.batch(FLAGS.train_batch_size)

dev_dataset = tf.data.Dataset.from_tensor_slices((dev_x, dev_y))
dev_dataset = dev_dataset.batch(FLAGS.dev_batch_size)

test_dataset = tf.data.Dataset.from_tensor_slices((test_x, test_y))
test_dataset = test_dataset.batch(FLAGS.test_batch_size)

# A reinitializable iterator
iterator = tf.data.Iterator.from_structure(train_dataset.output_types, train_dataset.output_shapes)

train_initializer = iterator.make_initializer(train_dataset)
dev_initializer = iterator.make_initializer(dev_dataset)
test_initializer = iterator.make_initializer(test_dataset)

有了 Dataset 的 iterator，我们只需要调用一次 get_next() 方法即可得到 x 和 y_label 了，就不需要使用 placeholder 来声明了，代码如下：

1
2
3

# Input Layer
with tf.variable_scope('inputs'):
    x, y_label = iterator.get_next()

接下来我们需要实现 embedding 层，调用 TensorFlow 的 embedding_lookup 即可实现，这里没有使用 Pre Train 的 embedding，代码实现如下：

1
2
3
4

# Embedding Layer
with tf.variable_scope('embedding'):
    embedding = tf.Variable(tf.random_normal([vocab_size, FLAGS.embedding_size]), dtype=tf.float32)
inputs = tf.nn.embedding_lookup(embedding, x)

接下来我们就需要实现双向 LSTM 了，这里我们要构造一个 2 层的 Bi-LSTM 网络，实现的时候我们首先需要声明 LSTM Cell 的列表，然后调用 stack_bidirectional_rnn() 方法即可：

1
2
3
4

cell_fw = [lstm_cell(FLAGS.num_units, keep_prob) for _ in range(FLAGS.num_layer)]
cell_bw = [lstm_cell(FLAGS.num_units, keep_prob) for _ in range(FLAGS.num_layer)]
inputs = tf.unstack(inputs, FLAGS.time_step, axis=1)
output, _, _ = tf.contrib.rnn.stack_bidirectional_rnn(cell_fw, cell_bw, inputs=inputs, dtype=tf.float32)

这个方法内部是首先对每一层的 LSTM 进行正反向计算，然后对输出隐层进行 concat，然后输入下一层再进行计算，这里值得注意的地方是，我们不能把 LSTM Cell 提前组合成 MultiRNNCell 再调用 bidirectional_dynamic_rnn() 进行计算，这样相当于只有最后一层才进行 concat，是错误的。 现在我们得到的 output 就是 Bi-LSTM 的最后输出结果了。 接下来我们需要对输出结果进行一下 stack() 操作转化为一个 Tensor，然后将其 reshape() 一下，转化为 [-1, num_units * 2] 的 shape：

1
2

output = tf.stack(output, axis=1)
output = tf.reshape(output, [-1, FLAGS.num_units * 2])

这样我们再经过一层全连接网络将维度进行转换：

1
2
3
4
5
6
7

# Output Layer
with tf.variable_scope('outputs'):
    w = weight([FLAGS.num_units * 2, FLAGS.category_num])
    b = bias([FLAGS.category_num])
    y = tf.matmul(output, w) + b
    y_predict = tf.cast(tf.argmax(y, axis=1), tf.int32)
    print('Output Y', y_predict)

这样得到的最后的 y_predict 即为预测结果，shape 为 [batch_size]，即每一句都得到了一个最可能的结果标注。 接下来我们需要计算一下准确率和 Loss，准确率其实就是比较 y_predict 和 y_label 的相似度，Loss 即为二者交叉熵：

1
2
3
4
5
6
7
8
9

# Reshape y_label
y_label_reshape = tf.cast(tf.reshape(y_label, [-1]), tf.int32)
# Prediction
correct_prediction = tf.equal(y_predict, y_label_reshape)
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
# Loss
cross_entropy = tf.reduce_mean(tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y_label_reshape, logits=tf.cast(y, tf.float32)))
# Train
train = tf.train.AdamOptimizer(FLAGS.learning_rate).minimize(cross_entropy, global_step=global_step)

这里计算交叉熵使用的是 sparse_softmax_cross_entropy_with_logits() 方法，Optimizer 使用的是 Adam。 最后指定训练过程和测试过程即可，训练过程如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

for epoch in range(FLAGS.epoch_num):
    tf.train.global_step(sess, global_step_tensor=global_step)
    # Train
    sess.run(train_initializer)
    for step in range(int(train_steps)):
        smrs, loss, acc, gstep, _ = sess.run([summaries, cross_entropy, accuracy, global_step, train], feed_dict={keep_prob: FLAGS.keep_prob})
        # Print log
        if step % FLAGS.steps_per_print == 0:
            print('Global Step', gstep, 'Step', step, 'Train Loss', loss, 'Accuracy', acc)

    if epoch % FLAGS.epochs_per_dev == 0:
        # Dev
        sess.run(dev_initializer)
        for step in range(int(dev_steps)):
            if step % FLAGS.steps_per_print == 0:
                print('Dev Accuracy', sess.run(accuracy, feed_dict={keep_prob: 1}), 'Step', step)

这里训练时首先调用了 train_initializer，将 iterator 指向训练数据，这样每调用一次 get_next()，x 和 y_label 就会被赋值为训练数据的一个 batch，接下来打印输出了 Loss，Accuracy 等内容。另外对于开发集来说，每次进行验证的时候也需要重新调用 dev_initializer，这样 iterator 会再次指向开发集，这样每调用一次 get_next()，x 和 y_label 就会被赋值为开发集的一个 batch，然后进行验证。 对于测试来说，我们可以计算其准确率，然后将测试的结果输出出来，代码实现如下：

1
2
3
4
5
6
7
8
9
10

sess.run(test_initializer)
for step in range(int(test_steps)):
    x_results, y_predict_results, acc = sess.run([x, y_predict, accuracy], feed_dict={keep_prob: 1})
    print('Test step', step, 'Accuracy', acc)
    y_predict_results = np.reshape(y_predict_results, x_results.shape)
    for i in range(len(x_results)):
        x_result, y_predict_result = list(filter(lambda x: x, x_results[i])), list(
            filter(lambda x: x, y_predict_results[i]))
        x_text, y_predict_text = ''.join(id2word[x_result].values), ''.join(id2tag[y_predict_result].values)
        print(x_text, y_predict_text)

这里打印输出了当前测试的准确率，然后得到了测试结果，然后再结合词表将测试的真正结果打印出来即可。

运行结果

在训练过程中，我们需要构建模型图，然后调用训练部分的代码进行训练，输出结果类似如下：

1
2
3
4
5
6
7
8
9

Global Step 0 Step 0 Train Loss 1.67181 Accuracy 0.1475
Global Step 100 Step 100 Train Loss 0.210423 Accuracy 0.928125
Global Step 200 Step 200 Train Loss 0.208561 Accuracy 0.920625
Global Step 300 Step 300 Train Loss 0.185281 Accuracy 0.939375
Global Step 400 Step 400 Train Loss 0.186069 Accuracy 0.938125
Global Step 500 Step 500 Train Loss 0.165667 Accuracy 0.94375
Global Step 600 Step 600 Train Loss 0.201692 Accuracy 0.9275
Global Step 700 Step 700 Train Loss 0.13299 Accuracy 0.954375
...

随着训练的进行，准确率可以达到 96% 左右。 在测试阶段，输出了当前模型的准确率及真实测试输出结果，输出结果类似如下：

1
2
3
4

Test step 0 Accuracy 0.946125
据新华社北京７月９日电连日来 sbmebebmmesbes
董新辉为自己此生不能侍奉母亲而难过 bmesbebebebmmesbe
...

可见测试准确率在 95% 左右，对于测试数据，此处还输出了每句话的序列标注结果，如第一行结果中，“据”字对应的标注就是 s，代表单字成词，“新”字对应的标注是 b，代表词的起始，“华”字对应标注是 m，代表词的中间，“社”字对应的标注是 e，代表结束，这样 “据”、“新华社” 就可以被分成两个词了，可见还是有一定效果的。

结语

本节通过搭建一个 Bi-LSTM 网络实现了序列标注，并可实现分词，准确率可达到 95% 左右，但是最主要的还是学习 Bi-LSTM 的用法，本实例代码较多，部分代码已经省略，完整代码见：https://github.com/AIDeepLearning/BiLSTMWordBreaker。

参考来源

• TensorFlow入门 双端 LSTM 实现序列标注
• 基于双向LSTM的seq2seq字标注
• TensorFlow全新的数据读取方式：Dataset API入门教程
• TensorFlow Importing Data

Ansible简介

Ansible是由Python开发的一个运维工具，因为工作需要接触到Ansible，经常会集成一些东西到Ansible，所以对Ansible的了解越来越多。 那Ansible到底是什么呢？在我的理解中，原来需要登录到服务器上，然后执行一堆命令才能完成一些操作。而Ansible就是来代替我们去执行那些命令。并且可以通过Ansible控制多台机器，在机器上进行任务的编排和执行，在Ansible中称为playbook。 那Ansible是如何做到的呢？简单点说，就是Ansible将我们要执行的命令生成一个脚本，然后通过sftp将脚本上传到要执行命令的服务器上，然后在通过ssh协议，执行这个脚本并将执行结果返回。 那Ansible具体是怎么做到的呢？下面从模块和插件来看一下Ansible是如何完成一个模块的执行 PS：下面的分析都是在对Ansible有一些具体使用经验之后，通过阅读源代码进一步得出的执行结论，所以希望在看本文时，是建立在对Ansible有一定了解的基础上，最起码对于Ansible的一些概念有了解，例如inventory，module，playbooks等

Ansible模块

模块是Ansible执行的最小单位，可以是由Python编写，也可以是Shell编写，也可以是由其他语言编写。模块中定义了具体的操作步骤以及实际使用过程中所需要的参数 执行的脚本就是根据模块生成一个可执行的脚本。 那Ansible是怎么样将这个脚本上传到服务器上，然后执行获取结果的呢？

Ansible插件

connection插件

连接插件，根据指定的ssh参数连接指定的服务器，并切提供实际执行命令的接口

shell插件

命令插件，根据sh类型，来生成用于connection时要执行的命令

strategy插件

执行策略插件，默认情况下是线性插件，就是一个任务接着一个任务的向下执行，此插件将任务丢到执行器去执行。

action插件

动作插件，实质就是任务模块的所有动作，如果ansible的模块没有特别编写的action插件，默认情况下是normal或者async（这两个根据模块是否async来选择），normal和async中定义的就是模块的执行步骤。例如，本地创建临时文件，上传临时文件，执行脚本，删除脚本等等，如果想在所有的模块中增加一些特殊步骤，可以通过增加action插件的方式来扩展。

Ansible执行模块流程

1. ansible命令实质是通过ansible/cli/adhoc.py来运行，同时会收集参数信息
   1. 设置Play信息，然后通过TaskQueueManager进行run，
   2. TaskQueueManager需要Inventory(节点仓库)，variable_manager(收集变量),options(命令行中指定的参数),stdout_callback(回调函数)
2. 在task_queue_manager.py中找到run中
   1. 初始化时会设置队列
   2. 会根据options，，variable_manager，passwords等信息设置成一个PlayContext信息(playbooks/playcontext.py)
   3. 设置插件(plugins)信息callback_loader(回调), strategy_loader(执行策略), module_loader(任务模块)
   4. 通过strategy_loader（strategy插件）的run（默认的strategy类型是linear，线性执行），去按照顺序执行所有的任务（执行一个模块，可能会执行多个任务）
   5. 在strategy_loader插件run之后，会判断action类型。如果是meta类型的话会单独执行(不是具体的ansible模块时)，而其他模块时，会加载到队列_queue_task
   6. 在队列中会调用WorkerProcess去处理，在workerproces实际的run之后，会使用TaskExecutor进行执行
   7. 在TaskExecutor中会设置connection插件，并且根据task的类型（模块。或是include等）获取action插件，就是对应的模块，如果模块有自定义的执行，则会执行自定义的action，如果没有的会使用normal或者async，这个是根据是否是任务的async属性来决定
   8. 在Action插件中定义着执行的顺序，及具体操作，例如生成临时目录，生成临时脚本，所以要在统一的模式下，集成一些额外的处理时，可以重写Action的方法
   9. 通过Connection插件来执行Action的各个操作步骤

扩展Ansible实例

执行节点Python环境扩展

实际需求中，我们扩展的一些Ansible模块需要使用三方库，但每个节点中安装这些库有些不易于管理。ansible执行模块的实质就是在节点的python环境下执行生成的脚本，所以我们采取的方案是，指定节点上的Python环境，将局域网内一个python环境作为nfs共享。通过扩展Action插件，增加节点上挂载nfs，待执行结束后再将节点上的nfs卸载。具体实施步骤如下： 扩展代码：

重写ActionBase的execute_module方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194

# execute_module

from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

import json
import pipes

from ansible.compat.six import text_type, iteritems

from ansible import constants as C
from ansible.errors import AnsibleError
from ansible.release import __version__

try:
    from __main__ import display
except ImportError:
    from ansible.utils.display import Display
    display = Display()


class MagicStackBase(object):

    def _mount_nfs(self, ansible_nfs_src, ansible_nfs_dest):
        cmd = ['mount',ansible_nfs_src, ansible_nfs_dest]
        cmd = [pipes.quote(c) for c in cmd]
        cmd = ' '.join(cmd)
        result = self._low_level_execute_command(cmd=cmd, sudoable=True)
        return result

    def _umount_nfs(self, ansible_nfs_dest):
        cmd = ['umount', ansible_nfs_dest]
        cmd = [pipes.quote(c) for c in cmd]
        cmd = ' '.join(cmd)
        result = self._low_level_execute_command(cmd=cmd, sudoable=True)
        return result

    def _execute_module(self, module_name=None, module_args=None, tmp=None, task_vars=None, persist_files=False, delete_remote_tmp=True):
        '''
        Transfer and run a module along with its arguments.
        '''

        # display.v(task_vars)

        if task_vars is None:
            task_vars = dict()

        # if a module name was not specified for this execution, use
        # the action from the task
        if module_name is None:
            module_name = self._task.action
        if module_args is None:
            module_args = self._task.args

        # set check mode in the module arguments, if required
        if self._play_context.check_mode:
            if not self._supports_check_mode:
                raise AnsibleError("check mode is not supported for this operation")
            module_args['_ansible_check_mode'] = True
        else:
            module_args['_ansible_check_mode'] = False

        # Get the connection user for permission checks
        remote_user = task_vars.get('ansible_ssh_user') or self._play_context.remote_user

        # set no log in the module arguments, if required
        module_args['_ansible_no_log'] = self._play_context.no_log or C.DEFAULT_NO_TARGET_SYSLOG

        # set debug in the module arguments, if required
        module_args['_ansible_debug'] = C.DEFAULT_DEBUG

        # let module know we are in diff mode
        module_args['_ansible_diff'] = self._play_context.diff

        # let module know our verbosity
        module_args['_ansible_verbosity'] = display.verbosity

        # give the module information about the ansible version
        module_args['_ansible_version'] = __version__

        # set the syslog facility to be used in the module
        module_args['_ansible_syslog_facility'] = task_vars.get('ansible_syslog_facility', C.DEFAULT_SYSLOG_FACILITY)

        # let module know about filesystems that selinux treats specially
        module_args['_ansible_selinux_special_fs'] = C.DEFAULT_SELINUX_SPECIAL_FS

        (module_style, shebang, module_data) = self._configure_module(module_name=module_name, module_args=module_args, task_vars=task_vars)
        if not shebang:
            raise AnsibleError("module (%s) is missing interpreter line" % module_name)

        # get nfs info for mount python packages
        ansible_nfs_src = task_vars.get("ansible_nfs_src", None)
        ansible_nfs_dest = task_vars.get("ansible_nfs_dest", None)

        # a remote tmp path may be necessary and not already created
        remote_module_path = None
        args_file_path = None
        if not tmp and self._late_needs_tmp_path(tmp, module_style):
            tmp = self._make_tmp_path(remote_user)

        if tmp:
            remote_module_filename = self._connection._shell.get_remote_filename(module_name)
            remote_module_path = self._connection._shell.join_path(tmp, remote_module_filename)
            if module_style in ['old', 'non_native_want_json']:
                # we'll also need a temp file to hold our module arguments
                args_file_path = self._connection._shell.join_path(tmp, 'args')

        if remote_module_path or module_style != 'new':
            display.debug("transferring module to remote")
            self._transfer_data(remote_module_path, module_data)
            if module_style == 'old':
                # we need to dump the module args to a k=v string in a file on
                # the remote system, which can be read and parsed by the module
                args_data = ""
                for k,v in iteritems(module_args):
                    args_data += '%s=%s ' % (k, pipes.quote(text_type(v)))
                self._transfer_data(args_file_path, args_data)
            elif module_style == 'non_native_want_json':
                self._transfer_data(args_file_path, json.dumps(module_args))
            display.debug("done transferring module to remote")

        environment_string = self._compute_environment_string()

        remote_files = None

        if args_file_path:
            remote_files = tmp, remote_module_path, args_file_path
        elif remote_module_path:
            remote_files = tmp, remote_module_path

        # Fix permissions of the tmp path and tmp files.  This should be
        # called after all files have been transferred.
        if remote_files:
            self._fixup_perms2(remote_files, remote_user)


        # mount nfs
        if ansible_nfs_src and ansible_nfs_dest:
            result = self._mount_nfs(ansible_nfs_src, ansible_nfs_dest)
            if result['rc'] != 0:
                raise AnsibleError("mount nfs failed!!! {0}".format(result['stderr']))

        cmd = ""
        in_data = None

        if self._connection.has_pipelining and self._play_context.pipelining and not C.DEFAULT_KEEP_REMOTE_FILES and module_style == 'new':
            in_data = module_data
        else:
            if remote_module_path:
                cmd = remote_module_path

        rm_tmp = None
        if tmp and "tmp" in tmp and not C.DEFAULT_KEEP_REMOTE_FILES and not persist_files and delete_remote_tmp:
            if not self._play_context.become or self._play_context.become_user == 'root':
                # not sudoing or sudoing to root, so can cleanup files in the same step
                rm_tmp = tmp

        cmd = self._connection._shell.build_module_command(environment_string, shebang, cmd, arg_path=args_file_path, rm_tmp=rm_tmp)
        cmd = cmd.strip()
        sudoable = True
        if module_name == "accelerate":
            # always run the accelerate module as the user
            # specified in the play, not the sudo_user
            sudoable = False


        res = self._low_level_execute_command(cmd, sudoable=sudoable, in_data=in_data)

        # umount nfs
        if ansible_nfs_src and ansible_nfs_dest:
            result = self._umount_nfs(ansible_nfs_dest)
            if result['rc'] != 0:
                raise AnsibleError("umount nfs failed!!! {0}".format(result['stderr']))

        if tmp and "tmp" in tmp and not C.DEFAULT_KEEP_REMOTE_FILES and not persist_files and delete_remote_tmp:
            if self._play_context.become and self._play_context.become_user != 'root':
                # not sudoing to root, so maybe can't delete files as that other user
                # have to clean up temp files as original user in a second step
                tmp_rm_cmd = self._connection._shell.remove(tmp, recurse=True)
                tmp_rm_res = self._low_level_execute_command(tmp_rm_cmd, sudoable=False)
                tmp_rm_data = self._parse_returned_data(tmp_rm_res)
                if tmp_rm_data.get('rc', 0) != 0:
                    display.warning('Error deleting remote temporary files (rc: {0}, stderr: {1})'.format(tmp_rm_res.get('rc'), tmp_rm_res.get('stderr', 'No error string available.')))

        # parse the main result
        data = self._parse_returned_data(res)

        # pre-split stdout into lines, if stdout is in the data and there
        # isn't already a stdout_lines value there
        if 'stdout' in data and 'stdout_lines' not in data:
            data['stdout_lines'] = data.get('stdout', u'').splitlines()

        display.debug("done with _execute_module (%s, %s)" % (module_name, module_args))
        return data

集成到normal.py和async.py中，记住要将这两个插件在ansible.cfg中进行配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

from ansible.plugins.action import ActionBase
from ansible.utils.vars import merge_hash

from common.ansible_plugins import MagicStackBase


class ActionModule(MagicStackBase, ActionBase):

    def run(self, tmp=None, task_vars=None):
        if task_vars is None:
            task_vars = dict()

        results = super(ActionModule, self).run(tmp, task_vars)
        # remove as modules might hide due to nolog
        del results['invocation']['module_args']
        results = merge_hash(results, self._execute_module(tmp=tmp, task_vars=task_vars))
        # Remove special fields from the result, which can only be set
        # internally by the executor engine. We do this only here in
        # the 'normal' action, as other action plugins may set this.
        #
        # We don't want modules to determine that running the module fires
        # notify handlers.  That's for the playbook to decide.
        for field in ('_ansible_notify',):
            if field in results:
                results.pop(field)

        return results

• 配置ansible.cfg，将扩展的插件指定为ansible需要的action插件
• 重写插件方法，重点是execute_module
• 执行命令中需要指定Python环境，将需要的参数添加进去nfs挂载和卸载的参数

1

ansible 51 -m mysql_db -a "state=dump name=all target=/tmp/test.sql" -i hosts -u root -v -e "ansible_nfs_src=172.16.30.170:/web/proxy_env/lib64/python2.7/site-packages ansible_nfs_dest=/root/.pyenv/versions/2.7.10/lib/python2.7/site-packages ansible_python_interpreter=/root/.pyenv/versions/2.7.10/bin/python"

背景

用 Python 做过爬虫的小伙伴可能接触过 Scrapy，GitHub：https://github.com/scrapy/scrapy。Scrapy 的确是一个非常强大的爬虫框架，爬取效率高，扩展性好，基本上是使用 Python 开发爬虫的必备利器。如果使用 Scrapy 做爬虫，那么在爬取时，我们当然完全可以使用自己的主机来完成爬取，但当爬取量非常大的时候，我们肯定不能在自己的机器上来运行爬虫了，一个好的方法就是将 Scrapy 部署到远程服务器上来执行。 所以，这时候就出现了另一个库 Scrapyd，GitHub：https://github.com/scrapy/scrapyd，有了它我们只需要在远程服务器上安装一个 Scrapyd，启动这个服务，就可以将我们写的 Scrapy 项目部署到远程主机上了，Scrapyd 还提供了各种操作 API，可以自由地控制 Scrapy 项目的运行，API 文档：http://scrapyd.readthedocs.io/en/stable/api.html，例如我们将 Scrapyd 安装在 IP 为 88.88.88.88 的服务器上，然后将 Scrapy 项目部署上去，这时候我们通过请求 API 就可以来控制 Scrapy 项目的运行了，命令如下：

1

curl http://88.88.88.88:6800/schedule.json -d project=myproject -d spider=somespider

这样就相当于启动了 myproject 项目的 somespider 爬虫，而不用我们再用命令行方式去启动爬虫，同时 Scrapyd 还提供了查看爬虫状态、取消爬虫任务、添加爬虫版本、删除爬虫版本等等的一系列 API，所以说，有了 Scrapyd，我们可以通过 API 来控制爬虫的运行，摆脱了命令行的依赖。 另外爬虫部署还是个麻烦事，因为我们需要将爬虫代码上传到远程服务器上，这个过程涉及到打包和上传两个过程，在 Scrapyd 中其实提供了这个部署的 API，叫做 addversion，但是它接受的内容是 egg 包文件，所以说要用这个接口，我们必须要把我们的 Scrapy 项目打包成 egg 文件，然后再利用文件上传的方式请求这个 addversion 接口才可以完成上传，这个过程又比较繁琐了，所以又出现了一个工具叫做 Scrapyd-Client，GitHub：https://github.com/scrapy/scrapyd-client，利用它的 scrapyd-deploy 命令我们便可以完成打包和上传的两个功能，可谓是又方便了一步。 这样我们就已经解决了部署的问题，回过头来，如果我们要想实时查看服务器上 Scrapy 的运行状态，那该怎么办呢？像刚才说的，当然是请求 Scrapyd 的 API 了，如果我们想用 Python 程序来控制一下呢？我们还要用 requests 库一次次地请求这些 API ？这就太麻烦了吧，所以为了解决这个需求，Scrapyd-API 又出现了，GitHub：https://github.com/djm/python-scrapyd-api，有了它我们可以只用简单的 Python 代码就可以实现 Scrapy 项目的监控和运行：

1
2
3

from scrapyd_api import ScrapydAPI
scrapyd = ScrapydAPI('http://88.888.88.88:6800')
scrapyd.list_jobs('project_name')

这样它的返回结果就是各个 Scrapy 项目的运行情况。 例如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

{
    'pending': [
    ],
    'running': [
        {
            'id': u'14a65...b27ce',
            'spider': u'spider_name',
            'start_time': u'2018-01-17 22:45:31.975358'
        },
    ],
    'finished': [
        {
            'id': '34c23...b21ba',
            'spider': 'spider_name',
            'start_time': '2018-01-11 22:45:31.975358',
            'end_time': '2018-01-17 14:01:18.209680'
        }
    ]
}

这样我们就可以看到 Scrapy 爬虫的运行状态了。 所以，有了它们，我们可以完成的是：

• 通过 Scrapyd 完成 Scrapy 项目的部署
• 通过 Scrapyd 提供的 API 来控制 Scrapy 项目的启动及状态监控
• 通过 Scrapyd-Client 来简化 Scrapy 项目的部署
• 通过 Scrapyd-API 来通过 Python 控制 Scrapy 项目

是不是方便多了？ 可是？真的达到最方便了吗？肯定没有！如果这一切的一切，从 Scrapy 的部署、启动到监控、日志查看，我们只需要鼠标键盘点几下就可以完成，那岂不是美滋滋？更或者说，连 Scrapy 代码都可以帮你自动生成，那岂不是爽爆了？ 有需求就有动力，没错，Gerapy 就是为此而生的，GitHub：https://github.com/Gerapy/Gerapy。 本节我们就来简单了解一下 Gerapy 分布式爬虫管理框架的使用方法。

安装

Gerapy 是一款分布式爬虫管理框架，支持 Python 3，基于 Scrapy、Scrapyd、Scrapyd-Client、Scrapy-Redis、Scrapyd-API、Scrapy-Splash、Jinjia2、Django、Vue.js 开发，Gerapy 可以帮助我们：

• 更方便地控制爬虫运行
• 更直观地查看爬虫状态
• 更实时地查看爬取结果
• 更简单地实现项目部署
• 更统一地实现主机管理
• 更轻松地编写爬虫代码

安装非常简单，只需要运行 pip3 命令即可：

1

$ pip3 install gerapy

安装完成之后我们就可以使用 gerapy 命令了，输入 gerapy 便可以获取它的基本使用方法：

1
2
3
4
5
6
7

$ gerapy
Usage:
  gerapy init [--folder=<folder>]
  gerapy migrate
  gerapy createsuperuser
  gerapy runserver [<host:port>]
  gerapy makemigrations

如果出现上述结果，就证明 Gerapy 安装成功了。

初始化

接下来我们来开始使用 Gerapy，首先利用如下命令进行一下初始化，在任意路径下均可执行如下命令：

1

$ gerapy init

执行完毕之后，本地便会生成一个名字为 gerapy 的文件夹，接着进入该文件夹，可以看到有一个 projects 文件夹，我们后面会用到。 紧接着执行数据库初始化命令：

1
2

cd gerapy
gerapy migrate

这样它就会在 gerapy 目录下生成一个 SQLite 数据库，同时建立数据库表。 接着我们只需要再运行命令启动服务就好了：

1

gerapy runserver

这样我们就可以看到 Gerapy 已经在 8000 端口上运行了。 全部的操作流程截图如下： 接下来我们在浏览器中打开 http://localhost:8000/，就可以看到 Gerapy 的主界面了： 这里显示了主机、项目的状态，当然由于我们没有添加主机，所以所有的数目都是 0。 如果我们可以正常访问这个页面，那就证明 Gerapy 初始化都成功了。

主机管理

接下来我们可以点击左侧 Clients 选项卡，即主机管理页面，添加我们的 Scrapyd 远程服务，点击右上角的创建按钮即可添加我们需要管理的 Scrapyd 服务： 需要添加 IP、端口，以及名称，点击创建即可完成添加，点击返回即可看到当前添加的 Scrapyd 服务列表，样例如下所示： 这样我们可以在状态一栏看到各个 Scrapyd 服务是否可用，同时可以一目了然当前所有 Scrapyd 服务列表，另外我们还可以自由地进行编辑和删除。

项目管理

Gerapy 的核心功能当然是项目管理，在这里我们可以自由地配置、编辑、部署我们的 Scrapy 项目，点击左侧的 Projects ，即项目管理选项，我们可以看到如下空白的页面： 假设现在我们有一个 Scrapy 项目，如果我们想要进行管理和部署，还记得初始化过程中提到的 projects 文件夹吗？这时我们只需要将项目拖动到刚才 gerapy 运行目录的 projects 文件夹下，例如我这里写好了一个 Scrapy 项目，名字叫做 zhihusite，这时把它拖动到 projects 文件夹下： 这时刷新页面，我们便可以看到 Gerapy 检测到了这个项目，同时它是不可配置、没有打包的： 这时我们可以点击部署按钮进行打包和部署，在右下角我们可以输入打包时的描述信息，类似于 Git 的 commit 信息，然后点击打包按钮，即可发现 Gerapy 会提示打包成功，同时在左侧显示打包的结果和打包名称： 打包成功之后，我们便可以进行部署了，我们可以选择需要部署的主机，点击后方的部署按钮进行部署，同时也可以批量选择主机进行部署，示例如下： 可以发现此方法相比 Scrapyd-Client 的命令行式部署，简直不能方便更多。

监控任务

部署完毕之后就可以回到主机管理页面进行任务调度了，任选一台主机，点击调度按钮即可进入任务管理页面，此页面可以查看当前 Scrapyd 服务的所有项目、所有爬虫及运行状态： 我们可以通过点击新任务、停止等按钮来实现任务的启动和停止等操作，同时也可以通过展开任务条目查看日志详情： 另外我们还可以随时点击停止按钮来取消 Scrapy 任务的运行。 这样我们就可以在此页面方便地管理每个 Scrapyd 服务上的 每个 Scrapy 项目的运行了。

项目编辑

同时 Gerapy 还支持项目编辑功能，有了它我们不再需要 IDE 即可完成项目的编写，我们点击项目的编辑按钮即可进入到编辑页面，如图所示： 这样即使 Gerapy 部署在远程的服务器上，我们不方便用 IDE 打开，也不喜欢用 Vim 等编辑软件，我们可以借助于本功能方便地完成代码的编写。

代码生成

上述的项目主要针对的是我们已经写好的 Scrapy 项目，我们可以借助于 Gerapy 方便地完成编辑、部署、控制、监测等功能，而且这些项目的一些逻辑、配置都是已经写死在代码里面的，如果要修改的话，需要直接修改代码，即这些项目都是不可配置的。 在 Scrapy 中，其实提供了一个可配置化的爬虫 CrawlSpider，它可以利用一些规则来完成爬取规则和解析规则的配置，这样可配置化程度就非常高，这样我们只需要维护爬取规则、提取逻辑就可以了。如果要新增一个爬虫，我们只需要写好对应的规则即可，这类爬虫就叫做可配置化爬虫。 Gerapy 可以做到：我们写好爬虫规则，它帮我们自动生成 Scrapy 项目代码。 我们可以点击项目页面的右上角的创建按钮，增加一个可配置化爬虫，接着我们便可以在此处添加提取实体、爬取规则、抽取规则了，例如这里的解析器，我们可以配置解析成为哪个实体，每个字段使用怎样的解析方式，如 XPath 或 CSS 解析器、直接获取属性、直接添加值等多重方式，另外还可以指定处理器进行数据清洗，或直接指定正则表达式进行解析等等，通过这些流程我们可以做到任何字段的解析。 再比如爬取规则，我们可以指定从哪个链接开始爬取，允许爬取的域名是什么，该链接提取哪些跟进的链接，用什么解析方法来处理等等配置。通过这些配置，我们可以完成爬取规则的设置。 最后点击生成按钮即可完成代码的生成。 生成的代码示例结果如图所示，可见其结构和 Scrapy 代码是完全一致的。 生成代码之后，我们只需要像上述流程一样，把项目进行部署、启动就好了，不需要我们写任何一行代码，即可完成爬虫的编写、部署、控制、监测。

结语

以上便是 Gerapy 分布式爬虫管理框架的基本用法，如需了解更多，可以访问其 GitHub：https://github.com/Gerapy/Gerapy。 如果觉得此框架有不足的地方，欢迎提 Issue，也欢迎发 Pull Request 来贡献代码，如果觉得 Gerapy 有所帮助，还望赐予一个 Star！非常感谢！

本节来介绍一下使用 RNN 的 LSTM 来做 MNIST 分类的方法，RNN 相比 CNN 来说，速度可能会慢，但可以节省更多的内存空间。

初始化

首先我们可以先初始化一些变量，如学习率、节点单元数、RNN 层数等：

1
2
3
4
5
6
7
8
9
10
11

learning_rate = 1e-3
num_units = 256
num_layer = 3
input_size = 28
time_step = 28
total_steps = 2000
category_num = 10
steps_per_validate = 100
steps_per_test = 500
batch_size = tf.placeholder(tf.int32, [])
keep_prob = tf.placeholder(tf.float32, [])

然后还需要声明一下 MNIST 数据生成器：

1
2
3

import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
mnist = input_data.read_data_sets('MNIST_data/', one_hot=True)

接下来常规声明一下输入的数据，输入数据用 x 表示，标注数据用 y_label 表示：

1
2

x = tf.placeholder(tf.float32, [None, 784])
y_label = tf.placeholder(tf.float32, [None, 10])

这里输入的 x 维度是 [None, 784]，代表 batch_size 不确定，输入维度 784，y_label 同理。 接下来我们需要对输入的 x 进行 reshape 操作，因为我们需要将一张图分为多个 time_step 来输入，这样才能构建一个 RNN 序列，所以这里直接将 time_step 设成 28，这样一来 input_size 就变为了 28，batch_size 不变，所以reshape 的结果是一个三维的矩阵：

1

x_shape = tf.reshape(x, [-1, time_step, input_size])

RNN 层

接下来我们需要构建一个 RNN 模型了，这里我们使用的 RNN Cell 是 LSTMCell，而且要搭建一个三层的 RNN，所以这里还需要用到 MultiRNNCell，它的输入参数是 LSTMCell 的列表。 所以我们可以先声明一个方法用于创建 LSTMCell，方法如下：

1
2
3

def cell(num_units):
    cell = tf.nn.rnn_cell.BasicLSTMCell(num_units=num_units)
    return DropoutWrapper(cell, output_keep_prob=keep_prob)

这里还加入了 Dropout，来减少训练过程中的过拟合。 接下来我们再利用它来构建多层的 RNN：

1

cells = tf.nn.rnn_cell.MultiRNNCell([cell(num_units) for _ in range(num_layer)])

注意这里使用了 for 循环，每循环一次新生成一个 LSTMCell，而不是直接使用乘法来扩展列表，因为这样会导致 LSTMCell 是同一个对象，导致构建完 MultiRNNCell 之后出现维度不匹配的问题。 接下来我们需要声明一个初始状态：

1

h0 = cells.zero_state(batch_size, dtype=tf.float32)

然后接下来调用 dynamic_rnn() 方法即可完成模型的构建了：

1

output, hs = tf.nn.dynamic_rnn(cells, inputs=x_shape, initial_state=h0)

这里 inputs 的输入就是 x 做了 reshape 之后的结果，初始状态通过 initial_state 传入，其返回结果有两个，一个 output 是所有 time_step 的输出结果，赋值为 output，它是三维的，第一维长度等于 batch_size，第二维长度等于 time_step，第三维长度等于 num_units。另一个 hs 是隐含状态，是元组形式，长度即 RNN 的层数 3，每一个元素都包含了 c 和 h，即 LSTM 的两个隐含状态。 这样的话 output 的最终结果可以取最后一个 time_step 的结果，所以可以使用：

1

output = output[:, -1, :]

或者直接取隐藏状态最后一层的 h 也是相同的：

1

h = hs[-1].h

在此模型中，二者是等价的。但注意如果用于文本处理，可能由于文本长度不一，而 padding，导致二者不同。

输出层

接下来我们再做一次线性变换和 Softmax 输出结果即可：

1
2
3
4
5
6

# Output Layer
w = tf.Variable(tf.truncated_normal([num_units, category_num], stddev=0.1), dtype=tf.float32)
b = tf.Variable(tf.constant(0.1, shape=[category_num]), dtype=tf.float32)
y = tf.matmul(output, w) + b
# Loss
cross_entropy = tf.nn.softmax_cross_entropy_with_logits(labels=y_label, logits=y)

这里的 Loss 直接调用了 softmax_cross_entropy_with_logits 先计算了 Softmax，然后计算了交叉熵。

训练和评估

最后再定义训练和评估的流程即可，在训练过程中每隔一定的 step 就输出 Train Accuracy 和 Test Accuracy：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# Train
train = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(cross_entropy)

# Prediction
correction_prediction = tf.equal(tf.argmax(y, axis=1), tf.argmax(y_label, axis=1))
accuracy = tf.reduce_mean(tf.cast(correction_prediction, tf.float32))

# Train
with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    for step in range(total_steps + 1):
        batch_x, batch_y = mnist.train.next_batch(100)
        sess.run(train, feed_dict={x: batch_x, y_label: batch_y, keep_prob: 0.5, batch_size: batch_x.shape[0]})
        # Train Accuracy
        if step % steps_per_validate == 0:
            print('Train', step, sess.run(accuracy, feed_dict={x: batch_x, y_label: batch_y, keep_prob: 0.5,
                                                               batch_size: batch_x.shape[0]}))
        # Test Accuracy
        if step % steps_per_test == 0:
            test_x, test_y = mnist.test.images, mnist.test.labels
            print('Test', step,
                  sess.run(accuracy, feed_dict={x: test_x, y_label: test_y, keep_prob: 1, batch_size: test_x.shape[0]}))

运行

直接运行之后，只训练了几轮就可以达到 98% 的准确率：

1
2
3
4
5
6
7
8
9
10
11

Train 0 0.27
Test 0 0.2223
Train 100 0.87
Train 200 0.91
Train 300 0.94
Train 400 0.94
Train 500 0.99
Test 500 0.9595
Train 600 0.95
Train 700 0.97
Train 800 0.98

可以看出来 LSTM 在做 MNIST 字符分类的任务上还是比较有效的。

本节代码

本节代码地址为：https://github.com/AIDeepLearning/LSTMClassification。

本文介绍下 RNN 及几种变种的结构和对应的 TensorFlow 源码实现，另外通过简单的实例来实现 TensorFlow RNN 相关类的调用。

RNN

RNN，循环神经网络，Recurrent Neural Networks。人们思考问题往往不是从零开始的，比如阅读时我们对每个词的理解都会依赖于前面看到的一些信息，而不是把前面看的内容全部抛弃再去理解某处的信息。应用到深度学习上面，如果我们想要学习去理解一些依赖上文的信息，RNN 便可以做到，它有一个循环的操作，可以使其可以保留之前学习到的内容。 RNN 的结构如下： 在上图网络结构中，对于矩形块 A 的那部分，通过输入xt（t时刻的特征向量），它会输出一个结果ht（t时刻的状态或者输出）。网络中的循环结构使得某个时刻的状态能够传到下一个时刻。 这些循环的结构让 RNNs 看起来有些难以理解，但我们可以把 RNNs 看成是一个普通的网络做了多次复制后叠加在一起组成的，每一网络会把它的输出传递到下一个网络中。我们可以把 RNNs 在时间步上进行展开，就得到下图这样： 所以最基本的 RNN Cell 输入就是 xt，它还会输出一个隐含内容传递到下一个 Cell，同时还会生成一个结果 ht，其最基本的结构如如下： 仅仅是输入的 xt 和隐藏状态进行 concat，然后经过线性变换后经过一个 tanh 激活函数便输出了，另外隐含内容和输出结果是相同的内容。 我们来分析一下 TensorFlow 里面 RNN Cell 的实现。 TensorFlow 实现 RNN Cell 的位置在 python/ops/rnncellimpl.py，首先其实现了一个 RNNCell 类，继承了 Layer 类，其内部有三个比较重要的方法，state_size()、output_size()、__call() 方法，其中 state_size() 和 output_size() 方法设置为类属性，可以当做属性来调用，实现如下：

1
2
3
4
5
6
7
8
9
10
11
12

@property
def state_size(self):
"""size(s) of state(s) used by this cell.
It can be represented by an Integer, a TensorShape or a tuple of Integers
or TensorShapes.
"""
    raise NotImplementedError("Abstract method")

@property
def output_size(self):
"""Integer or TensorShape: size of outputs produced by this cell."""
    raise NotImplementedError("Abstract method")

分别代表 Cell 的状态和输出维度，和 Cell 中的神经元数量有关，但这里两个方法都没有实现，意思是说我们必须要实现一个子类继承 RNNCell 类并实现这两个方法。 另外对于 call() 方法，实际上就是当初始化的对象直接被调用的时候触发的方法，实现如下：

1
2
3
4
5
6
7
8
9

def __call__(self, inputs, state, scope=None):
    if scope is not None:
        with vs.variable_scope(scope,
                               custom_getter=self._rnn_get_variable) as scope:
            return super(RNNCell, self).__call__(inputs, state, scope=scope)
    else:
        with vs.variable_scope(vs.get_variable_scope(),
                               custom_getter=self._rnn_get_variable):
            return super(RNNCell, self).__call__(inputs, state)

实际上是调用了父类 Layer 的 call() 方法，但父类中 call() 方法中又调用了 call() 方法，而 Layer 类的 call() 方法的实现如下：

1
2

def call(self, inputs, **kwargs):
    return inputs

父类的 call() 方法实现非常简单，所以要实现其真正的功能，只需要在继承 RNNCell 类的子类中实现 call() 方法即可。 接下来我们看下 RNN Cell 的最基本的实现，叫做 BasicRNNCell，其代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

class BasicRNNCell(RNNCell):
  """The most basic RNN cell.
  Args:
    num_units: int, The number of units in the RNN cell.
    activation: Nonlinearity to use.  Default: `tanh`.
    reuse: (optional) Python boolean describing whether to reuse variables
     in an existing scope.  If not `True`, and the existing scope already has
     the given variables, an error is raised.
  """

  def __init__(self, num_units, activation=None, reuse=None):
    super(BasicRNNCell, self).__init__(_reuse=reuse)
    self._num_units = num_units
    self._activation = activation or math_ops.tanh
    self._linear = None

  @property
  def state_size(self):
    return self._num_units

  @property
  def output_size(self):
    return self._num_units

  def call(self, inputs, state):
    """Most basic RNN: output = new_state = act(W * input + U * state + B)."""
    if self._linear is None:
      self._linear = _Linear([inputs, state], self._num_units, True)

    output = self._activation(self._linear([inputs, state]))
    return output, output

可以看到在初始化的时候，最终要的一个参数是 numunits，意思就是这个 Cell 中神经元的个数，另外还有一个参数 activation 即默认使用的激活函数，默认使用的 tanh，reuse 代表该 Cell 是否可以被重新使用。 在 statesize()、output_size() 方法里，其返回的内容都是 num_units，即神经元的个数，接下来 call() 方法中，传入的参数为 inputs 和 state，即输入的 x 和 上一次的隐含状态，首先实例化了一个 _Linear 类，这个类实际上就是做线性变换的类，将二者传递过来，然后直接调用，就实现了 w * [inputs, state] + b 的线性变换，其中 _Linear 类的 __call() 方法实现如下：

1
2
3
4
5
6
7
8
9
10

def __call__(self, args):
    if not self._is_sequence:
        args = [args]
    if len(args) == 1:
        res = math_ops.matmul(args[0], self._weights)
    else:
        res = math_ops.matmul(array_ops.concat(args, 1), self._weights)
    if self._build_bias:
        res = nn_ops.bias_add(res, self._biases)
    return res

很明显这里传递了 [inputs, state] 作为 call() 方法的 args，会执行 concat() 和 matmul() 方法，然后接着再执行 bias_add() 方法，这样就实现了线性变换。 最后回到 BasicRNNCell 的 call() 方法中，在 _linear() 方法外面又包括了一层 _activation() 方法，即对线性变换应用一次 tanh 激活函数处理，作为输出结果。 最后返回的结果是 output 和 output，第一个代表 output，第二个代表隐状态，其值也等于 output。 我们用一个实例来感受一下：

1
2
3
4
5
6
7
8
9

import tensorflow as tf

cell = tf.nn.rnn_cell.BasicRNNCell(num_units=128)
print(cell.state_size)
inputs = tf.placeholder(tf.float32, shape=[32, 100])
h0 = cell.zero_state(32, tf.float32)
output, h1 = cell(inputs=inputs, state=h0)
print(output, output.shape)
print(h1, h1.shape)

这里我们首先初始化了一个神经元个数为 128 的 BasicRNNCell 类，然后构造了一个 shape 为 [32, 100] 的变量作为 inputs，其代表 batch_size 为 32, 维度为 100，随后初始化了初始隐藏状态，调用了 zero_state() 方法，然后直接调用 cell，实际上是最终调用了其 call() 方法，最后得到 output 和 h1，打印输出结果：

1
2
3

128
Tensor("basic_rnn_cell/Tanh:0", shape=(32, 128), dtype=float32) (32, 128)
Tensor("basic_rnn_cell/Tanh:0", shape=(32, 128), dtype=float32) (32, 128)

可以看到，当输入变量维度为 100 的时候，经过一个 128 神经元 Cell 之后，输出维度变成了 128，其输出 shape 变成了 [32, 128]，且此时输出结果和隐藏状态是相同的。

LSTM

RNNs 的出现，主要是因为它们能够把以前的信息联系到现在，从而解决现在的问题。比如，利用前面的信息，能够帮助我们理解当前的内容。 有时候，我们在处理当前任务的时候，只需要看一下比较近的一些信息。比如在一个语言模型中，我们要通过上文来预测一下个词可能是什么，那么当我们看到 “the clouds are in the?”时，不需要更多的信息，我们就能够自然而然的想到下一个词应该是“sky”。在这样的情况下，我们所要预测的内容和相关信息之间的间隔很小，这种情况下 RNNs 就能够利用过去的信息， 很容易实现： 但是如果我们想依赖前文距离非常远的信息时，普通的 RNN 就非常难以做到了，随着间隔信息的增大，RNN 难以对其做关联： 但是 LSTM 可以用来解决这个问题。 LSTM，Long Short Term Memory Networks，是 RNN 的一个变种，经试验它可以用来解决更多问题，并取得了非常好的效果。 LSTM Cell 的结构如下： LSTMs 最关键的地方在于 Cell 的状态 和 结构图上面的那条横穿的水平线。 Cell 状态的传输就像一条传送带，向量从整个 Cell 中穿过，只是做了少量的线性操作。这种结构能够很轻松地实现信息从整个 Cell 中穿过而不做改变。 若只有上面的那条水平线是没办法实现添加或者删除信息的，信息的操作是是通过一种叫做门的结构来实现的。 这里我们可以把门分为三个：遗忘门（Forget Gate）、传入门（Input Gate）、输出门（Output Gate）。

遗忘门（Forget Gate）

首先是 LSTM 要决定让那些信息继续通过这个 Cell，这是通过 Forget Gate 的 sigmoid 神经层来实现的。它的输入是ht−1和xt，输出是一个数值都在 0，1 之间的向量，表示让 Ct−1 的各部分信息通过的比重。 0 表示“不让任何信息通过”， 1 表示“让所有信息通过”。

传入门（Input Gate）

下一步是决定让多少新的信息加入到 Cell 中来，一个叫做 Input Gate 的 sigmoid 层决定哪些信息需要更新，一个 New Input 通过 tanh 生成一个向量，也就是备选的用来更新的内容，Ct~ 。在下一步，我们把这两部分联合起来，对 Cell 的状态进行一个更新。 在经过 Forget Gate 和 Input Gate 处理后，我们就可以对输入的 Ct-1 做更新了，即把Ct−1 更新为 Ct，首先我们把旧的状态 Ct−1 和 ft 相乘， 把一些不想保留的信息忘掉。然后加上 it∗Ct~，这部分信息就是我们要添加的新内容，这样就可以完成对 Ct-1 的更新。

输出门 （Output Gate）

最后我们需要来决定输出什么值，输出主要是依赖于 Cell 的状态 Ct，但是又不仅仅依赖于 Ct，而是需要经过一个过滤的处理。首先，我们还是使用一个 sigmoid 层来决定 Ct 中的哪部分信息会被输出。然后我们把 Ct 通过一个 tanh 激活函数处理，然后把其输出和 sigmoid 计算出来的权重相乘，这样就得到了最后输出的结果。 到了最后，其输出结果有三个内容，其中输出结果就是最上面的箭头代指的内容，即最终计算的结果，隐层包括两部分内容，一个是 Ct，一个是最下方的 ht，我们可以将其合并为一个变量来表示。 接下来我们来看下 LSTMCell 的 TensorFlow 代码实现。 首先它的类是 BasicLSTMCell 类，继承了 RNNCell 类，其初始化方法 init() 实现如下：

1
2
3
4
5
6
7
8
9
10
11

def __init__(self, num_units, forget_bias=1.0,
               state_is_tuple=True, activation=None, reuse=None):
    super(BasicLSTMCell, self).__init__(_reuse=reuse)
    if not state_is_tuple:
      logging.warn("%s: Using a concatenated state is slower and will soon be "
                   "deprecated.  Use state_is_tuple=True.", self)
    self._num_units = num_units
    self._forget_bias = forget_bias
    self._state_is_tuple = state_is_tuple
    self._activation = activation or math_ops.tanh
    self._linear = None

这里必须传入的参数仍然是 num_units，即神经元的个数，然后 forget_bias 是初始化 Forget Gate 的偏置大小，state_is_tuple 指的是输出状态类型是元组类型，activation 代表默认激活函数，reuse 代表是否可以被重复使用。 接下来看下 state_size() 方法和 output_size() 方法，实现如下：

1
2
3
4
5
6
7
8

@property
def state_size(self):
    return (LSTMStateTuple(self._num_units, self._num_units)
        if self._state_is_tuple else 2 * self._num_units)

@property
def output_size(self):
    return self._num_units

这里 state_size() 方法变了，因为输出的 state 需要将 Ct 和隐含状态合并，所以它需要包含两部分的内容，如果传入的参数 state_is_tuple 为 True 的话，状态会被表示成一个元组，否则会是 num_units 乘以 2 的数字，默认是元组形式。output_size() 方法则保持不变。 对于 call() 方法，其实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

def call(self, inputs, state):
    """Long short-term memory cell (LSTM).

    Args:
      inputs: `2-D` tensor with shape `[batch_size x input_size]`.
      state: An `LSTMStateTuple` of state tensors, each shaped
        `[batch_size x self.state_size]`, if `state_is_tuple` has been set to
        `True`.  Otherwise, a `Tensor` shaped
        `[batch_size x 2 * self.state_size]`.

    Returns:
      A pair containing the new hidden state, and the new state (either a
        `LSTMStateTuple` or a concatenated state, depending on
        `state_is_tuple`).
    """
    sigmoid = math_ops.sigmoid
    # Parameters of gates are concatenated into one multiply for efficiency.
    if self._state_is_tuple:
        c, h = state
    else:
        c, h = array_ops.split(value=state, num_or_size_splits=2, axis=1)

    if self._linear is None:
        self._linear = _Linear([inputs, h], 4 * self._num_units, True)
    # i = input_gate, j = new_input, f = forget_gate, o = output_gate
    i, j, f, o = array_ops.split(
        value=self._linear([inputs, h]), num_or_size_splits=4, axis=1)

    new_c = (
        c * sigmoid(f + self._forget_bias) + sigmoid(i) * self._activation(j))
    new_h = self._activation(new_c) * sigmoid(o)

    if self._state_is_tuple:
        new_state = LSTMStateTuple(new_c, new_h)
    else:
        new_state = array_ops.concat([new_c, new_h], 1)
    return new_h, new_state

首先为了获取 c, h，需要将其从 state 中分离开来，如果传入的 state 是元组的话可以直接分解，否则需要调用 split() 方法来分解：

1
2
3
4

if self._state_is_tuple:
    c, h = state
else:
    c, h = array_ops.split(value=state, num_or_size_splits=2, axis=1)

接下来定义了几个门的实现：

1

i, j, f, o = array_ops.split(value=self._linear([inputs, h]), num_or_size_splits=4, axis=1)

放到一起来用 Linear 计算然后分成了 4 份，分别代表 Input Gate、New Input、Forget Gate、Output Gate，用 i、j、f、o 来表示，这时候四个变量都经过了线性变换，乘以权重并做了偏置操作。 接下来就是更新 Ct-1 为 Ct 和得到隐含状态输出了，都是遵循 LSTM 内部的公式实现：

1
2

new_c = (c * sigmoid(f + self._forget_bias) + sigmoid(i) * self._activation(j))
new_h = self._activation(new_c) * sigmoid(o)

这里值得注意的是还多加了一个 _forget_bias 变量，即设置了初始化偏置，以免初始输出为 0 的问题。 最后将 new_c 和 new_h 进行合并，如果要输出元组，那么就合并为元组，否则二者进行 concat 操作，返回的结果是 new_h、new_state，前者即 Cell 的输出结果，后者代表隐含状态：

1
2
3
4
5

if self._state_is_tuple:
    new_state = LSTMStateTuple(new_c, new_h)
else:
    new_state = array_ops.concat([new_c, new_h], 1)
return new_h, new_state

我们再用一个实例来感受一下 BasicLSTMCell 的用法：

1
2
3
4
5
6
7
8
9
10
11

import tensorflow as tf

cell = tf.nn.rnn_cell.BasicLSTMCell(num_units=128)
print(cell.state_size)
inputs = tf.placeholder(tf.float32, shape=(32, 100))
h0 = cell.zero_state(32, tf.float32)
output, h1 = cell(inputs=inputs, state=h0)
print(h1)
print(h1.h, h1.h.shape)
print(h1.c, h1.c.shape)
print(output, output.shape)

这里我们首先初始化了一个神经元个数为 128 的 BasicRNNCell 类，然后构造了一个 shape 为 [32, 100] 的变量作为 inputs，其代表 batch_size 为 32, 维度为 100，随后初始化了初始隐藏状态，调用了 zero_state() 方法，然后直接调用 cell，实际上是最终调用了其 call() 方法，最后得到 output 和 h1，此时 h1 是一个元组，它还可以分离成 h 和 c，分别打印其对象和维度，结果如下：

1
2
3
4
5

LSTMStateTuple(c=128, h=128)
LSTMStateTuple(c=<tf.Tensor 'add_1:0' shape=(32, 128) dtype=float32>, h=<tf.Tensor 'mul_2:0' shape=(32, 128) dtype=float32>)
Tensor("mul_2:0", shape=(32, 128), dtype=float32) (32, 128)
Tensor("add_1:0", shape=(32, 128), dtype=float32) (32, 128)
Tensor("mul_2:0", shape=(32, 128), dtype=float32) (32, 128)

可以看到其维度都是 [32, 128]，而且 h1.h 和 output 是相同的。 另外 LSTM 有许多变种，其中一个比较有名的就是 Gers & Schmidhuber (2000) 提出的，它在原来的基础上行添加了 Peephole Connections，使得遗忘门可以受 Ct-1 的影响。 另外还有一个变种就是将 Forget Gate 和 Input Gate 二者联合起来，做到要么遗忘老的输入新的，要么保留老的不输入新的。 但接下来还有一个更常用的变种，俺就是 GRU，它是由 Cho, et al. (2014) 提出的，在提出的同时他还提出了 Seq2Seq 模型，为 Generation Model 做好了铺垫。

GRU

GRU，Gated Recurrent Unit，在 GRU 中，只有两个门：重置门（Reset Gate）和更新门（Update Gate）。同时在这个结构中，把 Ct 和隐藏状态进行了合并，整体结构比标准的 LSTM 结构要简单，而且这个结构后来也非常流行。 接下来我们看下 TensorFlow 中 GRUCell 的实现，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

class GRUCell(RNNCell):
  """Gated Recurrent Unit cell (cf. http://arxiv.org/abs/1406.1078).

  Args:
    num_units: int, The number of units in the GRU cell.
    activation: Nonlinearity to use.  Default: `tanh`.
    reuse: (optional) Python boolean describing whether to reuse variables
     in an existing scope.  If not `True`, and the existing scope already has
     the given variables, an error is raised.
    kernel_initializer: (optional) The initializer to use for the weight and
    projection matrices.
    bias_initializer: (optional) The initializer to use for the bias.
  """

  def __init__(self,
               num_units,
               activation=None,
               reuse=None,
               kernel_initializer=None,
               bias_initializer=None):
    super(GRUCell, self).__init__(_reuse=reuse)
    self._num_units = num_units
    self._activation = activation or math_ops.tanh
    self._kernel_initializer = kernel_initializer
    self._bias_initializer = bias_initializer
    self._gate_linear = None
    self._candidate_linear = None

  @property
  def state_size(self):
    return self._num_units

  @property
  def output_size(self):
    return self._num_units

  def call(self, inputs, state):
    """Gated recurrent unit (GRU) with nunits cells."""
    if self._gate_linear is None:
      bias_ones = self._bias_initializer
      if self._bias_initializer is None:
        bias_ones = init_ops.constant_initializer(1.0, dtype=inputs.dtype)
      with vs.variable_scope("gates"):  # Reset gate and update gate.
        self._gate_linear = _Linear(
            [inputs, state],
            2 * self._num_units,
            True,
            bias_initializer=bias_ones,
            kernel_initializer=self._kernel_initializer)

    value = math_ops.sigmoid(self._gate_linear([inputs, state]))
    r, u = array_ops.split(value=value, num_or_size_splits=2, axis=1)

    r_state = r * state
    if self._candidate_linear is None:
      with vs.variable_scope("candidate"):
        self._candidate_linear = _Linear(
            [inputs, r_state],
            self._num_units,
            True,
            bias_initializer=self._bias_initializer,
            kernel_initializer=self._kernel_initializer)
    c = self._activation(self._candidate_linear([inputs, r_state]))
    new_h = u * state + (1 - u) * c
    return new_h, new_h

在 state_size()、output_size() 方法里，其返回的内容都是 num_units，即神经元的个数。 接下来 call() 方法中，因为 Reset Gate rt 和 Update Gate zt 分别用变量 r、u 表示，它们需要先对 ht-1 即 state 和 xt 做合并，然后再实现线性变换，再调用 sigmod 函数得到：

1
2

value = math_ops.sigmoid(self._gate_linear([inputs, state]))
r, u = array_ops.split(value=value, num_or_size_splits=2, axis=1)

然后需要求解 ht~，首先用 rt 和 ht-1 即 state 相乘：

1

r_state = r * state

然后将其放到线性函数里面，在调用 tanh 激活函数即可：

1

c = self._activation(self._candidate_linear([inputs, r_state]))

最后计算隐含状态和输出结果，二者一致：

1
2

new_h = u * state + (1 - u) * c
return new_h, new_h

这样即可返回得到输出结果和隐藏状态。 我们用一个实例感受一下：

1
2
3
4
5
6
7
8
9

import tensorflow as tf

cell = tf.nn.rnn_cell.GRUCell(num_units=128)
print(cell.state_size)
inputs = tf.placeholder(tf.float32, shape=[32, 100])
h0 = cell.zero_state(32, tf.float32)
output, h1 = cell(inputs=inputs, state=h0)
print(output, output.shape)
print(h1, h1.shape)

运行结果：

1
2
3

128
Tensor("gru_cell/add:0", shape=(32, 128), dtype=float32) (32, 128)
Tensor("gru_cell/add:0", shape=(32, 128), dtype=float32) (32, 128)

这个结果和 BasicRNNCell 并无二致，但 GRUCell 内部的结构使模型的效果更加优化，一般我们也会选取 GRUCell 来代替原生的 BasicRNNCell。

结语

以上便是对 RNN 及一些变种的说明及代码原理分析和实例用法，此部分掌握之后对 Dynamic RNN、多层 RNN 及 RNN Cell 的改写会有很大帮助，需要好好掌握。

上一节使用了最简单的网络来处理了 MNIST 数据集，但只有 92% 的正确率，接下来我们使用卷积神经网络来实现更高的正确率。

权重初始化

在上一节初始化 w 和 b 的时候，我们使用了置零初始化。但在卷积神经网络中，我们需要在初始化的时候权重加入少量噪声来打破对称性和避免零梯度，偏置项直接使用一个较小的正数来避免节点输出恒为零的问题。 所以权重我们可以使用截尾正态分布函数 truncated_normal() 来生成初始化张量，我们可以给它指定均值或标准差，均值默认是 0， 标准差默认是 1，例如我们生成一个 [10] 的张量，代码如下：

1
2
3
4

import tensorflow as tf
initial = tf.truncated_normal([10], stddev=0.1)
with tf.Session() as sess:
    print(sess.run(initial))

结果如下：

1
2

[-0.13058113  0.03201858 -0.19349943 -0.06061752 -0.10267895 -0.11079147
  0.1881365  -0.01057311 -0.02797078  0.01180232]

另外 constant() 方法是用于生成常量的方法，例如生成一个 [10] 的常量张量，代码如下：

1
2
3
4

import tensorflow as tf
initial = tf.constant(0.2, shape=[10])
with tf.Session() as sess:
    print(sess.run(initial))

结果如下：

1

[ 0.2  0.2  0.2  0.2  0.2  0.2  0.2  0.2  0.2  0.2]

所以这里我们可以将这两个方法封装成一个函数来尝试：

1
2
3
4
5
6
7

def weight(shape, stddev=0.1, mean=0):
    initial = tf.truncated_normal(shape=shape, mean=mean, stddev=stddev)
    return tf.Variable(initial)

def bias(shape, value):
    initial = tf.constant(value=value, shape=shape)
    return tf.Variable(initial)

卷积

这次我们需要使用卷积神经网络来处理图片，所以这里的核心部分就是卷积和池化了，首先我们来了解一下卷积和池化。 卷积常用的方法为 conv2d() ，它的 API 如下：

1

tf.nn.conv2d(input, filter, strides, padding, use_cudnn_on_gpu=None, name=None)

这个方法是 TensorFlow 实现卷积常用的方法，也是搭建卷积神经网络的核心方法，参数介绍如下：

• input，指需要做卷积的输入图像，它要求是一个 Tensor，具有 [batch_size, in_height, in_width, in_channels] 这样的 shape，具体含义是 [batch_size 的图片数量, 图片高度, 图片宽度, 输入图像通道数]，注意这是一个 4 维的 Tensor，要求类型为 float32 和 float64 其中之一。
• filter，相当于 CNN 中的卷积核，它要求是一个 Tensor，具有 [filter_height, filter_width, in_channels, out_channels] 这样的shape，具体含义是 [卷积核的高度，卷积核的宽度，输入图像通道数，输出通道数（即卷积核个数）]，要求类型与参数 input 相同，有一个地方需要注意，第三维 in_channels，就是参数 input 的第四维。
• strides，卷积时在图像每一维的步长，这是一个一维的向量，长度 4，具有 [stride_batch_size, stride_in_height, stride_in_width, stride_in_channels] 这样的 shape，第一个元素代表在一个样本的特征图上移动，第二三个元素代表在特征图上的高、宽上移动，第四个元素代表在通道上移动。
• padding，string 类型的量，只能是 SAME、VALID 其中之一，这个值决定了不同的卷积方式。
• use_cudnn_on_gpu，布尔类型，是否使用 cudnn 加速，默认为true。

返回的结果是 [batch_size, out_height, out_width, out_channels] 维度的结果。 我们这里拿一张 3x3 的图片，单通道（通道为1）的图片，拿一个 1x1 的卷积核进行卷积：

1
2
3
4

input = tf.Variable(tf.random_normal([1, 3, 3, 1]))
filter = tf.Variable(tf.random_normal([1, 1, 1, 1]))
op = tf.nn.conv2d(input, filter, strides=[1, 1, 1, 1], padding='VALID')
print(op.shape)

结果如下：

1

(1, 3, 3, 1)

很清晰，一张图片，拿一个 1x1 的核去做卷积，得到的结果输出是 3x3 的，输出通道为 1，batch_size 照旧。 再将卷积核扩大，用一个 3x3 的卷积核：

1
2
3
4

input = tf.Variable(tf.random_normal([1, 3, 3, 1]))
filter = tf.Variable(tf.random_normal([3, 3, 1, 1]))
op = tf.nn.conv2d(input, filter, strides=[1, 1, 1, 1], padding='VALID')
print(op.shape)

结果如下：

1

(1, 1, 1, 1)

最后输出的是一个 1x1 的值。 将图片扩大为 7x7，卷积核仍然使用 3x3：

1
2
3
4

input = tf.Variable(tf.random_normal([1, 7, 7, 1]))
filter = tf.Variable(tf.random_normal([3, 3, 1, 1]))
op = tf.nn.conv2d(input, filter, strides=[1, 1, 1, 1], padding='VALID')
print(op.shape)

结果如下：

1

(1, 5, 5, 1)

最后输出的是一个 5x5 的值。 这时如果我们把 padding 模式改为 SAME，表示卷积核可以停留在图像边缘：

1
2
3
4

input = tf.Variable(tf.random_normal([1, 7, 7, 1]))
filter = tf.Variable(tf.random_normal([3, 3, 1, 1]))
op = tf.nn.conv2d(input, filter, strides=[1, 1, 1, 1], padding='SAME')
print(op.shape)

结果如下：

1

(1, 7, 7, 1)

则输出的内容和原图像是相同的。 这时如果更改 batch_size 和 out_channels，比如 batch_size 修改为 3，out_channels 修改为 6：

1
2
3
4

input = tf.Variable(tf.random_normal([3, 7, 7, 1]))
filter = tf.Variable(tf.random_normal([3, 3, 1, 6]))
op = tf.nn.conv2d(input, filter, strides=[1, 1, 1, 1], padding='SAME')
print(op.shape)

结果如下：

1

(3, 7, 7, 6)

输出结果的 batch_size 和 out_channels 会随之变化。 当 strides 的步长不为 1 的时候，我们将 stride_in_height 和 stride_in_width 修改为 2，相当于每次移动两步：

1
2
3
4

input = tf.Variable(tf.random_normal([3, 7, 7, 1]))
filter = tf.Variable(tf.random_normal([3, 3, 1, 6]))
op = tf.nn.conv2d(input, filter, strides=[1, 2, 2, 1], padding='VALID')
print(op.shape)

结果如下：

1

(3, 3, 3, 6)

最后我们用一个例子来感受一下：

1
2
3
4
5
6
7
8
9

import tensorflow as tf

input = tf.Variable(tf.random_normal([2, 4, 4, 5]))
filter = tf.Variable(tf.random_normal([2, 2, 5, 2]))
op = tf.nn.conv2d(input, filter, strides=[1, 1, 1, 1], padding='VALID')
sess = tf.InteractiveSession()
tf.global_variables_initializer().run()
print(op.shape)
print(sess.run(op))

这里 input、filter 通过指定 shape 的方式调用 random_normal() 方法进行随机初始化，input 的维度为 [2, 4, 4, 5]，即 batch_size 为 2，图片是 4x4，输入通道数为 5，卷积核大小为 2x2，输入通道 5，输出通道 2，步长为 1，padding 方式选用 VALID，最后输出得到输出的 shape 和结果。 结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

(2, 3, 3, 2)
[[[[  2.05039382  -8.82934952]
   [ -9.77668381   3.63882256]
   [ -4.46390772  -5.91670704]]

  [[  8.41201782  -6.72245312]
   [ -1.47592044  13.03628349]
   [  5.44015312   2.46059227]]

  [[ -3.18967772   1.24733043]
   [-10.1108532   -6.44734669]
   [  1.99426246   2.91549349]]]


 [[[ -1.66685319   0.32011557]
   [ -5.66163826  -0.37670898]
   [ -0.74658942   1.31723833]]

  [[ -5.85412216  -0.29930949]
   [ -0.75974303  -1.84006214]
   [ -2.05475235   4.9572196 ]]

  [[ -4.09344864   1.39405775]
   [ -1.28887582  -2.82365012]
   [  4.87360907  10.8071022 ]]]]

可以看到 input 维度为 [2, 4, 4, 5]，filter 维度为 [2, 2, 5, 2] 时，生成的结果维度为 [2, 3, 3, 2]。

池化

池化层往往在卷积层后面，通过池化来降低卷积层输出的特征向量，同时改善结果。 在这里介绍一个常用的最大值池化 max_pool() 方法，其 API 如下：

1

tf.nn.max_pool(value, ksize, strides, padding, name=None)

是CNN当中的最大值池化操作，其实用法和卷积很类似。 参数介绍如下：

• value，需要池化的输入，一般池化层接在卷积层后面，所以输入通常是 feature map，依然是 [batch_size, height, width, channels] 这样的shape。
• ksize，池化窗口的大小，取一个四维向量，一般是 [batch_size, height, width, channels]，因为我们不想在 batch 和 channels 上做池化，所以这两个维度设为了1。
• strides，和卷积类似，窗口在每一个维度上滑动的步长，一般也是 [stride_batch_size, stride_height, stride_width, stride_channels]。
• padding，和卷积类似，可以取 VALID、SAME，返回一个 Tensor，类型不变，shape 仍然是 [batch_size, height, width, channels] 这种形式。

在这里输入为 [3, 7, 7, 2]，池化窗口设置为 [1, 2, 2, 1]，步长为 [1, 1, 1, 1]，padding 模式设置为 VALID。

1
2
3

input = tf.Variable(tf.random_normal([3, 7, 7, 2]))
op = tf.nn.max_pool(input, ksize=[1, 2, 2, 1], strides=[1, 1, 1, 1], padding='VALID')
print(op.shape)

结果如下：

1

(3, 6, 6, 2)

类似的原理，我们可以得到这样的的结果。

卷积和池化

所以了解了以上卷积和池化方法的用法，我们可以定义如下两个工具方法：

1
2
3
4
5

def conv2d(input, filter, strides=[1, 1, 1, 1], padding='SAME'):
    return tf.nn.conv2d(input, filter, strides=strides, padding=padding)

def max_pool(input, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], padding='SAME'):
    return tf.nn.max_pool(input, ksize=ksize, strides=strides, padding=padding)

这两个方法分别实现了卷积和池化，并设置了默认步长和核大小。 接下来就让我们开始神经网络的构建吧。

初始化

首先我们需要初始化一些数据，包括输入的 x 和对一个的标注 y_label：

1
2

x = tf.placeholder(tf.float32, shape=[None, 784])
y_label = tf.placeholder(tf.float32, shape=[None, 10])

第一层卷积

现在我们可以开始实现第一层了。它由一个卷积接一个 max pooling 完成。卷积在每个 5x5 的 patch 中算出 32 个特征。卷积的权重张量形状是 [5, 5, 1, 32]，前两个维度是 patch 的大小，接着是输入的通道数目，最后是输出的通道数目，而对于每一个输出通道都有一个对应的偏置量，我们首先初始化 w 和 b

1
2

w_conv1 = weight([5, 5, 1, 32])
b_conv1 = bias([32])

为了用这一层，我们把 x 变成一个四维向量，其第 2、3 维对应图片的宽、高，最后一维代表图片的颜色通道数，因为是灰度图所以这里的通道数为 1，如果是彩色图，则为 3。 随后我们需要对图片做 reshape 操作，将其

1

x_reshape = tf.reshape(x, [-1, 28, 28, 1])

我们把 x_reshape 和权值向量进行卷积，加上偏置项，然后应用 ReLU 激活函数，最后进行 max pooling：

1
2

h_conv1 = tf.nn.relu(conv2d(x_reshape, w_conv1) + b_conv1)
h_pool1 = max_pool(h_conv1)

第二层卷积

现在我们已经实现了一层卷积，为了构建一个更深的网络，我们再继续增加一层卷积，将通道数变成 64，所以这里的初始化权重和偏置为：

1
2

w_conv2 = weight([5, 5, 32, 64])
b_conv2 = bias([64])

随后我们把上一层池化结果 h_pool1 和权值向量进行卷积，加上偏置项，然后应用 ReLU 激活函数，最后进行 max pooling：

1
2

h_conv2 = tf.nn.relu(conv2d(h_pool1, w_conv2) + b_conv2)
h_pool2 = max_pool(h_conv2)

密集连接层

现在，图片尺寸减小到7x7，我们再加入一个有 1024 个神经元的全连接层，用于处理整个图片。我们把池化层输出的张量 reshape 成一些向量，乘上权重矩阵，加上偏置，然后对其使用 ReLU。

1
2
3
4

w_fc1 = weight([7 * 7 * 64, 1024])
b_fc1 = bias([1024])
h_pool2_flat = tf.reshape(h_pool2, [-1, 7 * 7 * 64])
h_fc1 = tf.nn.relu(tf.matmul(h_pool2_flat, w_fc1) + b_fc1)

Dropout

为了减少过拟合，我们在输出层之前加入 dropout。我们用一个 placeholder 来代表一个神经元的输出在 dropout 中保持不变的概率。这样我们可以在训练过程中启用 dropout，在测试过程中关闭 dropout。 TensorFlow 的 tf.nn.dropout 操作除了可以屏蔽神经元的输出外，还会自动处理神经元输出值的 scale，所以用 dropout 的时候可以不用考虑 scale。

1
2

keep_prob = tf.placeholder(tf.float32)
h_fc1_dropout = tf.nn.dropout(h_fc1, keep_prob=keep_prob)

输出层

最后，我们添加一个 Softmax 输出层，这里我们需要将 1024 维转为 10 维，所以需要声明一个 [1024, 10] 的权重和 [10] 的偏置：

1
2
3

w_fc2 = weight([1024, 10])
b_fc1 = bias([10])
y = tf.nn.softmax(tf.matmul(h_fc1_dropout, w_fc2) + b_fc1)

训练和评估模型

为了进行训练和评估，我们使用与之前简单的单层 Softmax 神经网络模型几乎相同的一套代码，只是我们会用更加复杂的 Adam 优化器来做梯度最速下降，在 feed_dict 中加入额外的参数 keep_prob 来控制 dropout 比例，然后每 100次 迭代输出一次日志：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# Loss
cross_entropy = -tf.reduce_sum(y_label * tf.log(y))
train = tf.train.AdamOptimizer(1e-4).minimize(cross_entropy)

# Prediction
correct_prediction = tf.equal(tf.argmax(y_label, axis=1), tf.argmax(y, axis=1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))

# Train
with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    for step in range(total_steps + 1):
        batch = mnist.train.next_batch(batch_size)
        sess.run(train, feed_dict={x: batch[0], y_label: batch[1], keep_prob: dropout_keep_prob})
        # Train accuracy
        if step % steps_per_test == 0:
            print('Training Accuracy', step,
                  sess.run(accuracy, feed_dict={x: batch[0], y_label: batch[1], keep_prob: 1}))

# Final Test
print('Test Accuracy', sess.run(accuracy, feed_dict={x: mnist.test.images, y_label: mnist.test.labels, keep_prob: 1}))

运行

以上代码，在最终测试集上的准确率大概是99.2%。 运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

Training Accuracy 0 0.05
Training Accuracy 100 0.7
Training Accuracy 200 0.85
Training Accuracy 300 0.9
Training Accuracy 400 0.93
Training Accuracy 500 0.91
Training Accuracy 600 0.94
Training Accuracy 700 0.95
Training Accuracy 800 0.95
Training Accuracy 900 0.95
Training Accuracy 1000 0.97
Training Accuracy 1100 0.95
Training Accuracy 1200 0.96
Training Accuracy 1300 0.99
Training Accuracy 1400 0.98
Training Accuracy 1500 0.95
Training Accuracy 1600 0.97
Training Accuracy 1700 1.0
Training Accuracy 1800 0.95
Training Accuracy 1900 0.95
Training Accuracy 2000 0.95
Training Accuracy 2100 0.96
Training Accuracy 2200 0.96
Training Accuracy 2300 0.98
Training Accuracy 2400 0.97
Training Accuracy 2500 0.96
Training Accuracy 2600 0.99
Training Accuracy 2700 0.96
Training Accuracy 2800 0.98
Training Accuracy 2900 0.95
Training Accuracy 3000 0.99

结语

本节我们实现了卷积神经网络来处理图像相关问题，将准确率大大提高，可见神经网络是非常强大的。

本节代码

本节代码地址为：https://github.com/AIDeepLearning/MNIST。

我们本节要用 MNIST 数据集训练一个可以识别数据的深度学习模型来帮助识别手写数字。

MNIST

MNIST 是一个入门级计算机视觉数据集，包含了很多手写数字图片，如图所示： 数据集中包含了图片和对应的标注，在 TensorFlow 中提供了这个数据集，我们可以用如下方法进行导入：

1
2
3

from tensorflow.examples.tutorials.mnist import input_data
mnist = input_data.read_data_sets('MNIST_data/', one_hot=True)
print(mnist)

输出结果如下：

1
2
3
4
5

Extracting MNIST_data/train-images-idx3-ubyte.gz
Extracting MNIST_data/train-labels-idx1-ubyte.gz
Extracting MNIST_data/t10k-images-idx3-ubyte.gz
Extracting MNIST_data/t10k-labels-idx1-ubyte.gz
Datasets(train=<tensorflow.contrib.learn.python.learn.datasets.mnist.DataSet object at 0x101707ef0>, validation=<tensorflow.contrib.learn.python.learn.datasets.mnist.DataSet object at 0x1016ae4a8>, test=<tensorflow.contrib.learn.python.learn.datasets.mnist.DataSet object at 0x1016f9358>)

在这里程序会首先下载 MNIST 数据集，然后解压并保存到刚刚制定好的 MNIST_data 文件夹中，然后输出数据集对象。 数据集中包含了 55000 行的训练数据集（mnist.train）、5000 行验证集（mnist.validation）和 10000 行的测试数据集（mnist.test），文件如下所示： 正如前面提到的一样，每一个 MNIST 数据单元有两部分组成：一张包含手写数字的图片和一个对应的标签。我们把这些图片设为 xs，把这些标签设为 ys。训练数据集和测试数据集都包含 xs 和 ys，比如训练数据集的图片是 mnist.train.images ，训练数据集的标签是 mnist.train.labels，每张图片是 28 x 28 像素，即 784 个像素点，我们可以把它展开形成一个向量，即长度为 784 的向量。 所以训练集我们可以转化为 [55000, 784] 的向量，第一维就是训练集中包含的图片个数，第二维是图片的像素点表示的向量。

Softmax

Softmax 可以看成是一个激励（activation）函数或者链接（link）函数，把我们定义的线性函数的输出转换成我们想要的格式，也就是关于 10 个数字类的概率分布。因此，给定一张图片，它对于每一个数字的吻合度可以被 Softmax 函数转换成为一个概率值。Softmax 函数可以定义为： 展开等式右边的子式，可以得到： 比如判断一张图片中的动物是什么，可能的结果有三种，猫、狗、鸡，假如我们可以经过计算得出它们分别的得分为 3.2、5.1、-1.7，Softmax 的过程首先会对各个值进行次幂计算，分别为 24.5、164.0、0.18，然后计算各个次幂结果占总次幂结果的比重，这样就可以得到 0.13、0.87、0.00 这三个数值，所以这样我们就可以实现差别的放缩，即好的更好、差的更差。 如果要进一步求损失值可以进一步求对数然后取负值，这样 Softmax 后的值如果值越接近 1，那么得到的值越小，即损失越小，如果越远离 1，那么得到的值越大。

实现回归模型

首先导入 TensorFlow，命令如下：

1

import tensorflow as tf

接下来我们指定一个输入，在这里输入即为样本数据，如果是训练集那么则是 55000 x 784 的矩阵，如果是验证集则为 5000 x 784 的矩阵，如果是测试集则是 10000 x 784 的矩阵，所以它的行数是不确定的，但是列数是确定的。 所以可以先声明一个 placeholder 对象：

1

x = tf.placeholder(tf.float32, [None, 784])

这里第一个参数指定了矩阵中每个数据的类型，第二个参数指定了数据的维度。 接下来我们需要构建第一层网络，表达式如下： 这里实际上是对输入的 x 乘以 w 权重，然后加上一个偏置项作为输出，而这两个变量实际是在训练的过程中动态调优的，所以我们需要指定它们的类型为 Variable，代码如下：

1
2

w = tf.Variable(tf.zeros([784, 10]))
b = tf.Variable(tf.zeros([10]))

接下来需要实现的就是上图所述的公式了，我们再进一步调用 Softmax 进行计算，得到 y：

1

y = tf.nn.softmax(tf.matmul(x, w) + b)

通过上面几行代码我们就已经把模型构建完毕了，结构非常简单。

损失函数

为了训练我们的模型，我们首先需要定义一个指标来评估这个模型是好的。其实，在机器学习，我们通常定义指标来表示一个模型是坏的，这个指标称为成本（cost）或损失（loss），然后尽量最小化这个指标。但是这两种方式是相同的。 一个非常常见的，非常漂亮的成本函数是“交叉熵”（cross-entropy）。交叉熵产生于信息论里面的信息压缩编码技术，但是它后来演变成为从博弈论到机器学习等其他领域里的重要技术手段。它的定义如下： y 是我们预测的概率分布, y_label 是实际的分布，比较粗糙的理解是，交叉熵是用来衡量我们的预测用于描述真相的低效性。 我们可以首先定义 y_label，它的表达式是：

1

y_label = tf.placeholder(tf.float32, [None, 10])

接下来我们需要计算它们的交叉熵，代码如下：

1

cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_label * tf.log(y), reduction_indices=[1]))

首先用 reduce_sum() 方法针对每一个维度进行求和，reduction_indices 是指定沿哪些维度进行求和。 然后调用 reduce_mean() 则求平均值，将一个向量中的所有元素求算平均值。 这样我们最后只需要优化这个交叉熵就好了。 所以这样我们再定义一个优化方法：

1

train = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)

这里使用了 GradientDescentOptimizer，在这里，我们要求 TensorFlow 用梯度下降算法（gradient descent algorithm）以 0.5 的学习速率最小化交叉熵。梯度下降算法（gradient descent algorithm）是一个简单的学习过程，TensorFlow 只需将每个变量一点点地往使成本不断降低的方向移动即可。

运行模型

定义好了以上内容之后，相当于我们已经构建好了一个计算图，即设置好了模型，我们把它放到 Session 里面运行即可：

1
2
3
4
5

with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    for step in range(total_steps + 1):
        batch_x, batch_y = mnist.train.next_batch(batch_size)
        sess.run(train, feed_dict={x: batch_x, y_label: batch_y})

该循环的每个步骤中，我们都会随机抓取训练数据中的 batch_size 个批处理数据点，然后我们用这些数据点作为参数替换之前的占位符来运行 train。 这里需要一些变量的定义：

1
2

batch_size = 100
total_steps = 5000

测试模型

那么我们的模型性能如何呢？ 首先让我们找出那些预测正确的标签。tf.argmax() 是一个非常有用的函数，它能给出某个 Tensor 对象在某一维上的其数据最大值所在的索引值。由于标签向量是由 0,1 组成，因此最大值 1 所在的索引位置就是类别标签，比如 tf.argmax(y, 1) 返回的是模型对于任一输入 x 预测到的标签值，而 tf.argmax(y_label, 1) 代表正确的标签，我们可以用 tf.equal() 方法来检测我们的预测是否真实标签匹配（索引位置一样表示匹配）。

1

correct_prediction = tf.equal(tf.argmax(y, axis=1), tf.argmax(y_label, axis=1))

这行代码会给我们一组布尔值。为了确定正确预测项的比例，我们可以把布尔值转换成浮点数，然后取平均值。例如，[True, False, True, True] 会变成 [1, 0, 1, 1] ，取平均值后得到 0.75。

1

accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))

最后，我们计算所学习到的模型在测试数据集上面的正确率，定义如下：

1
2
3

steps_per_test = 100
if step % steps_per_test == 0:
    print(step, sess.run(accuracy, feed_dict={x: mnist.test.images, y_label: mnist.test.labels}))

这个最终结果值应该大约是92%。 这样我们就通过完成了训练和测试阶段，实现了一个基本的训练模型，后面我们会继续优化模型来达到更好的效果。 运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

0 0.453
100 0.8915
200 0.9026
300 0.9081
400 0.9109
500 0.9108
600 0.9175
700 0.9137
800 0.9158
900 0.9176
1000 0.9167
1100 0.9186
1200 0.9206
1300 0.9161
1400 0.9218
1500 0.9179
1600 0.916
1700 0.9196
1800 0.9222
1900 0.921
2000 0.9223
2100 0.9214
2200 0.9191
2300 0.9228
2400 0.9228
2500 0.9218
2600 0.9197
2700 0.9225
2800 0.9238
2900 0.9219
3000 0.9224
3100 0.9184
3200 0.9253
3300 0.9216
3400 0.9218
3500 0.9212
3600 0.9225
3700 0.9224
3800 0.9225
3900 0.9226
4000 0.9201
4100 0.9138
4200 0.9184
4300 0.9222
4400 0.92
4500 0.924
4600 0.9234
4700 0.9219
4800 0.923
4900 0.9254
5000 0.9218

结语

本节通过一个 MNIST 数据集来简单体验了一下真实数据的训练和预测过程，但是准确率还不够高，后面我们会学习用卷积的方式来进行模型训练，准确率会更高。

本节代码

本节代码地址为：https://github.com/AIDeepLearning/MNIST。

本篇内容基于 Python3 TensorFlow 1.4 版本。

本节内容

本节通过最简单的示例 —— 平面拟合来说明 TensorFlow 的基本用法。

构造数据

TensorFlow 的引入方式是：

1

import tensorflow as tf

接下来我们构造一些随机的三维数据，然后用 TensorFlow 找到平面去拟合它，首先我们用 Numpy 生成随机三维点，其中变量 x 代表三维点的 (x, y) 坐标，是一个 2x100 的矩阵，即 100 个 (x, y)，然后变量 y 代表三位点的 z 坐标，我们用 Numpy 来生成这些随机的点：

1
2
3
4
5
6

import numpy as np
x_data = np.float32(np.random.rand(2, 100))
y_data = np.dot([0.300, 0.200], x_data) + 0.400

print(x_data)
print(y_data)

这里利用 Numpy 的 random 模块的 rand() 方法生成了 2x100 的随机矩阵，这样就生成了 100 个 (x, y) 坐标，然后用了一个 dot() 方法算了矩阵乘法，用了一个长度为 2 的向量跟此矩阵相乘，得到一个长度为 100 的向量，然后再加上一个常量，得到 z 坐标，输出结果样例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51