广告ctr预估问题总结

2018-11-04

1. 为什么要做ctr(click througth rate)预估

互联网广告行业涉及媒体、广告主和用户三方。准确的预估ctr对广告主来说可以获得更高的用户转化率，对用户来说可以找到自己感兴趣的物品，对媒体方来说可以带来更多收益。

2. 广告ctr预估怎么转化为数学问题

$p(y=1|x)=\sigma(w, x)=\sigma(w, u_{x}, a_{x}, c_{x})$，$u_{x}$是用户特征、$a_{x}$是广告特征、$c_{x}$是上下文特征，本质上就是针对一个用户，在特定的情景下，预估其点击一个广告的概率问题。
这里的预估函数$\sigma(w, u_{x}, a_{x}, c_{x})$可以任意取，但问题在于怎么衡量其预估结果与真实情况之间的差距呢？
这里假设有n个样本$(x_{i}, y_{i})$，那么就希望预测结果$p(x_{i}, w)$和$y_{i}$尽可能接近。
整体上也就是希望似然函数$L(w, x_{i}, y_{i})=\prod\limits_{i=1}^{n}p(w, x_{i})^{y_{i}}(1-p(w,x_{i}))^{1-y_{i}}=\prod\limits_{i=1}^{n}\sigma(w, x_{i})^{y_{i}}(1-\sigma(w, x_{i}))^{1-y_{i}}$最大。
简单变化换后，就等价于最小化损失函数$Loss(w, x_{i}, y_{i}) = \frac{-\sum\limits_{i=1}^{n}(y_{i}log(\sigma(w, x_{i}))+(1-y_{i})log(1-\sigma(w,x_{i})))}{n}$
这样ctr预估问题就变成了一个机器学习的二分类问题。

3. ctr预估问题的难点在哪

广告数据$(u_{x},a_{x},c_{x})$的特点
- 数据多：亿级到千亿级
- 维度高: 百万维到上亿维
- 很稀疏：90%都是0
- 多域：年龄、性别、省份、时间、广告创意、广告文字、图片
- 离散值：每个特征的取值都是有限的
特征x与标签y之间的关系未知，选取的$\sigma$函数往往带有主观偏见；不同的模型对输入的要求不同，简单模型往往需要工程人员对输入做大量变换，复杂模型往往难以解释。
训练数据量大、维度高、模型复杂、响应时间严苛带来的线上工程化问题

4. 工业界怎么做ctr预估

主要是基于机器学习模型来对ctr进行计算，把ctr预估问题转化为二分类问题。
离线训练的效果基于auc和logloss来衡量
线上通过流量abtest来对比不同模型的效果
模型一般每天都要进行离线训练，定时和线上的模型进行同步；针对复杂模型，线上一般都要定制实现

5. 工业界ctr预估模型的演进

5.1 general linear model

Logistic Regression
- $p(w,x)=\frac{1}{1+e^{-w^{T}x}}$
Piece-Wise linear model
- 阿里巴巴提出
- $p(w,x)=\sum\limits_{i=1}^{m}\frac{e^{u_{i}^{T}x}}{\sum\limits_{j=1}^{m}e^{u_{j}^{T}x}}·\frac{1}{1+e^{-w_{i}^{T}x}}$
GBDT+LR
- FaceBook提出
- 使用gbdt对原始数据做特征交叉，然后再给到LR模型使用
优点：模型容易解释，容易存储
缺点：线性模型无法学习到不同特征之间的交叉效果，需要工程人员进行大量的手工特征交叉

5.2. non-linear model

5.2.1 Factorization Machine及其变种

分解机在模型层面上引入特征之间的组合，并通过对特征进行隐向量分解来更好的学习特征交互项的参数。比较适合于稀疏的广告数据。
FM
- $\phi_{FM}(w, x) = w_{0}+\sum\limits_{i=1}^{n}w_{i}x_{i}+\sum\limits_{i=1}^{n-1}\sum\limits_{j=i+1}^{n}w_{ij}x_{i}x_{j} = w_{0}+\sum\limits_{i=1}^{n}w_{i}x_{i}+\sum\limits_{i=1}^{n-1}\sum\limits_{j=i+1}^{n}(v_{i}^{T}v_{j})x_{i}x_{j}$
Field-aware FM
- $\phi_{FFM}(w, x) = w_{0}+\sum\limits_{i=1}^{n}w_{i}x_{i}+\sum\limits_{i=1}^{n-1}\sum\limits_{j=i+1}^{n}(v_{i,f_{j}} \cdot v_{j,f_{i}})x_{i}x_{j}$
- 考虑到不同特征域之间组合时重要性的差异
Field-weighted FM
- $\Phi_{FwFM}(x) = w_{0} + \sum\limits_{i=1}^{m}x_{i}w_{i} + \sum\limits_{i=1}^{m}\sum\limits_{j=i+1}^{m}x_{i}x_{j}r_{F(i),F(j)}$

5.2.2 Neural Network及和其他模型的融合

直觉是Neural Network Model有更大的capacity，可以学习更加复杂的特征的交叉关系
Wide & Deep
- Google提出
- 对LR和MLP进行融合，统一训练，综合dnn的记忆能力强和lr泛化能力强的特点
- $p(y=1|x)=\sigma(w_{wide}^{T}[x, \phi(x)]+w_{deep}^{T}a^{f}+b)$
DeepFM
- 华为提出
- 对FM和MLP进行融合，统一训练，避开lr模型需要进行的大量特征工程
Deep & Cross Network
- Google提出
- 相对于FM一般只能学习二阶域特征交叉，增加一个能显示学习更高阶域特征交叉的cross network，并与MLP进行融合
- $x_{l+1}=x_{0}x_{l}^{T}w_{l}+b_{l}+x_{l}$
Deep Interest Network
- 阿里巴巴提出
- 更好地利用电商网站用户历史行为数据中的多样性和局部激活的特点
- 利用注意力机制和用户的历史行为来对用户进行更好的表征
- $V_{u}=\sum\limits_{i=1}^{N}=g(V_{i}, V_{a})*V_{i}$
XDeepFM
- MSRA提出
- 一方面通过dnn来隐式地学习特征之间的交叉，一方面在模型层面显示地学习不同域的特征交叉，比DeepFM更加通用

6. 其他问题

转化率及点击率同时预估
- 阿里巴巴的Entire Space Multi-task Model，使用两个模型分别估计ctr和cvr，并且两个模型共享基础的Embedding结构且进行统一训练，避免单一使用转化数据带来的稀疏性问题

展开全文 >>

TensorFlow高级API-Dataset和Estimator的使用介绍

2018-05-02

1.TensorFlow简介

TensorFlow是Google Brain团队开发的第二代机器学习开源软件库。它基于有状态的数据流图(Data Flow Graph)来表达计算过程，名称中的Tensor称之为张量，Flow可理解为张量在数据流图的节点之间流动，从而完成前向计算过程。TF可以进行自动微分，极大简化了机器学习模型的构建过程。此外，支持分布式的训练在GPU上，核心引擎基于C++编写，提供了python、java、go等接口，目前已被很多公司用做深度学习模型的开发工具。

2. Dataset和Estimator简介

使用TensorFlow的底层API开发机器学习模型时，需要显式地定义模型中的变量和输入数据、以及对会话进行显式的声明和管理，这需要不小的编码量。而使用更高层的Dataset工具类可以很轻松、高效地处理大量的输入数据以及不同的数据格式，相比基于feed_dict的数据输入方式更加高效和规整。并且使用Estimator工具类可以简化机器学习模型的构建过程，Estimator可以自动管理图的构建、变量初始化、模型保存及恢复过程。对一般机器学习pipeline中的训练、评估、预测三个过程进行统一管理。

2.1 Dataset介绍

Dataset表示一系列元素的集合。其中的每个元素可以是单个或者多个Tensor对象。此外，Dataset包含了一系列作用于这些元素上的操作，包含map、flat_map、filter、repeat、shuffle、skip、take等，这类似于Spark中的RDD。

Dataset的创建
- 基于from_tensor_slices方法从tensor中创建，tensor通常来源于numpy Array或者pandas DataFrame
- 基于TextLineDataset从多个文本文件中创建
- 基于TFRecordDataset从多个TFRecord文件中创建，TFRecord是针对大批量数据以二进制进行存储的一种存储格式
Dataset的遍历
- 基于Iterator进行遍历，make_one_shot_iterator()方法可以生成iterator，调用Iterator的get_next()方法就可以得到下一个Dataset中的元素。Dataset主要用于后续Estimator的输入函数中，示例如下：

#自定义训练和评估输入函数
def train_input_fn(features, labels, batch_size):
    dataset = tf.data.Dataset.from_tensor_slices((dict(features), labels))
    dataset = dataset.shuffle(1000).repeat(1).batch(batch_size)
    iterator = dataset.make_one_shot_iterator()
    return iterator.get_next()

def eval_input_fn(features, labels, batch_size):
    dataset = tf.data.Dataset.from_tensor_slices((dict(features), labels))
    dataset = dataset.batch(batch_size)
    iterator = dataset.make_one_shot_iterator()
    return iterator.get_next()

2.2 Estimator介绍

Estimator封装了模型的构建、训练、评估、预估以及保存过程，将数据的输入从模型中分离出来。数据输入需要编写单独的函数。

预定义的Estimator
- tf.estimator中内置了GBDT、DNN、LinearClassifer等多种模型，在使用这些模型前，需要指定数据中的特征列，特征列告诉模型如何对数据进行特征工程，tf.feature_column提供了多种常用的用于创建特征列的方法，包含类别型、数值型、类别型特征onehot、数值型特征分桶等。也可以将keras模型转换成estimator。
自定义Estimator
- 预定义的Estimator不能满足需求时，就可以自定义Estimator。主要是要自定义model函数，函数签名为model_fn(features, labels, mode, param)，features和labels是输入函数返回的特征和标签，mode表示是要进行模型的训练、评估还是测试。param是传入的额外参数。针对Google提出的wide&deep模型的model_fn如下：

def wide_deep_model_fn(features, labels, mode, params):
    #分别产生deep和wide部分的输入
    deep_input = tf.feature_column.input_layer(features, params['deep_feature_columns'])
    wide_input = tf.feature_column.input_layer(features, params['wide_feature_columns'])
    
    hidden1 = tf.layers.dense(deep_input, units=params['hidden_units'][0], activation=tf.nn.relu)
    hidden2 = tf.layers.dense(hidden1, units=params['hidden_units'][1], activation=tf.nn.relu)
    deep_logits = tf.layers.dense(hidden2, 1, activation=None)
    wide_logits = tf.layers.dense(wide_input, 1, activation=None)
    logits = deep_logits + wide_logits
    loss = tf.losses.sigmoid_cross_entropy(multi_class_labels=labels, logits=tf.squeeze(logits))
    
    predicted_class = tf.cast(logits > 0, tf.float32)
    auc = tf.metrics.auc(labels, predicted_class)
    metrics = {'auc': auc}
    if mode == tf.estimator.ModeKeys.EVAL:
        return tf.estimator.EstimatorSpec(mode, loss=loss, eval_metric_ops=metrics)
    
    optimizer = tf.train.AdamOptimizer(learning_rate=0.01)
    train_op = optimizer.minimize(loss, global_step=tf.train.get_global_step())
    if mode == tf.estimator.ModeKeys.TRAIN:
        return tf.estimator.EstimatorSpec(mode, loss=loss, train_op=train_op)

3 Dataset和Estimator的完整使用流程

定义用于训练和评估的输入函数input_fn;
根据数据集的特点，定义好feature_column;
使用预定义的Estimator或者自定义模型函数;
调用Estimator的train、eval和predict方法产生结果。

展开全文 >>

NLP之记忆网络(Memory Network)学习笔记

2018-02-11

1. 原始记忆网络(Memory Network)

由FAIR的Jason Weston等在2015年提出，主要应用于文本问答任务中，相比之前的方法，在F1值上取得了显著的提升。核心思想是考虑到典型的循环神经网络(如Valinna-RNN、LSTM、GRU)依赖状态向量进行序列状态的记忆，记忆能力受限，很难记忆过去的一些事实；因而增加单独的、可以读写的记忆组件。

1.1 模型结构

输入组件$I$对输出的原始数据进行特征变换。
记忆组件$G$就是对输入的特征向量进行存储的槽；每次一个输入进来时，更新已有的记忆槽。
输出组件$O$根据输入选择匹配的记忆，并结合输入和记忆产生最终的输出。
回复组件根据输出$O$产生最终的回复结果。
整个模型的flow分为四步：
- 输入转换:$x \rightarrow I(x)$
- 根据新输入更新记忆：$m_{i}=G(m_{i}, I(x), m)$
- 输出计算：$o=O(I(x), m)$
- 根据输出产生最终的回复: $r=R(o)$

1.2 文本问答

针对文本问答任务，核心的推理在输出O和回复R模块。
- O模块选取k(k=2)个辅助推理的记忆内容：$o_{j}=\arg\max\limits_{i=1,2,…,N}s_{O}(x, m_{i}), j=1,2,…,k$
- R模块选取最终回复的词：$r=\arg\max_{w \in W}s_{R}([x,m_{o_{1}},m_{o_{2}}], w)$，$W$代表词典中所有词。
- 打分函数:$s(x,y)=\Phi_{x}(x)^{T}U^{T}U\Phi_{y}(y)$，$\Phi$用来对输入进行特征映射，$U$是需要学习的权重矩阵。
- 训练目标函数
  - 这里采用的是类似SVM中的最大边缘损失函数。

2.动态记忆网络(Dynamic Memory Network)

由Ankit Kumar等在2016年提出，并指出很多NLP任务可以看作QA任务。DMN在文本问答、文本分类、词性标注几个不同的任务中都取得了很好的效果。

2.1 模型架构

单独的情景记忆模块：情景记忆模块通常需要多轮迭代更新，实验结果显示多轮迭代对于需要推理的任务十分重要，下图中的两条线代表两次阅读输入事实，迭代次数也是个需要调整的超参数。每一轮的内容$e_{i}$根据输入事实的表征$c_{t}$、输入问题的表征$q$以及之前的记忆$m_{i-1}$来产生。
- $h_{t}^{i} = g_{t}^{i}GRU(c_{t}, h_{t-1}^{i})+(1-g_{t}^{i})h_{t-1}^{i}$
- $e^{i} = h_{T_{C}}^{i}$，$T_{C}$是输入事实的个数
记忆模块：根据之前的记忆内容、问题表征、情景记忆内容来产生。记忆表征初始化为$m^{0}=q$。
- $m^{i} = GRU(e^{i}, m_{i-1})$
输入表征、问题表征以及答案的产生均基于GRU。
采用了注意力机制来计算事实表征$c$、内部记忆$m$和问题表征$q$之间的匹配得分值。整体来说，attention的计算较为复杂。
- $z(c,m,q)=[c, m, q, c\circ q, |c-q|, |c-m|, c^{T}W^{(b)}q, c^{T}W^{(b)}m]$
- $G(c,m,q)=\sigma(W^{(2)}tanh(W^{(1)}z(c,m,q)+b^{(1)})+b^{(2)})$
- $g_{t}^{i} = G(c_{t}, m^{i-1}, q)$
相比记忆网络，动态记忆网络增加了attention机制，并且记忆内容端到端可微。

2.2 代码实现

整个DMN的代码如下，详细功能参考注释

class DynamicMemoryNetwork(object):
    '''
    定义动态记忆网络的结构
    '''
    def __init__(self, config):
        self.config = config #针对整个网络超参数的配置
        if self.config.train_mode:
            self.train, self.valid, self.word_embedding, self.max_q_len, self.max_sentences, self.max_sen_len, \
            self.num_supporting_facts, self.vocab_size = babi_input.load_babi(self.config, split_sentences=True)
        else:
            self.test, self.word_embedding, self.max_q_len, self.max_sentences, self.max_sen_len, \
            self.num_supporting_facts, self.vocab_size = babi_input.load_babi(self.config, split_sentences=True)
        self.question_holder = tf.placeholder(tf.int32, shape=(self.config.batch_size, self.max_q_len))
        self.input_placeholder = tf.placeholder(tf.int32, shape=(self.config.batch_size, self.max_sentences, self.max_sen_len))
        self.question_len_holder = tf.placeholder(tf.int32, shape=(self.config.batch_size, ))
        self.input_len_holder = tf.placeholder(tf.int32, shape=(self.config.batch_size, ))
        self.answer_holder = tf.placeholder(tf.int64, shape=(self.config.batch_size, ))
        self.rel_label_holder = tf.placeholder(tf.int32, shape=(self.config.batch_size, self.num_supporting_facts))
        self.dropout_placeholder = tf.placeholder(tf.float32)

        encoding = np.ones((self.config.embed_size, self.max_sen_len), dtype=np.float32)
        for i in range(1, self.config.embed_size+1):
            for j in range(1, self.max_sen_len+1):
                encoding[i-1, j-1] = (i - (self.config.embed_size-1)/2)*(j - (self.max_sen_len-1)/2)
        encoding = 1 + 4 * encoding / (self.config.embed_size*self.max_sen_len)
        self.encoding = np.transpose(encoding)
        self.embeddings = tf.Variable(self.word_embedding.astype(np.float32), name="embedding")

        self.output = self.inference()
        self.pred = self.get_predictions(self.output)
        self.calculate_loss = self.add_loss_op(self.output)
        self.train_step = self.add_training_op(self.calculate_loss)
        self.merged = tf.summary.merge_all()

    def get_input_represetation(self):
        '''
        计算输入事实的表征
        :return:
        '''
        inputs = tf.nn.embedding_lookup(self.embeddings, self.input_placeholder)
        inputs = tf.reduce_sum(inputs*self.encoding, 2)
        forward_gru_cell = tf.contrib.rnn.GRUCell(self.config.hidden_size)
        backward_gru_cell = tf.contrib.rnn.GRUCell(self.config.hidden_size)
        outputs, _ = tf.nn.bidirectional_dynamic_rnn(
            forward_gru_cell,
            backward_gru_cell,
            inputs,
            dtype=np.float32,
            sequence_length=self.input_len_holder
        )
        fact_vecs = tf.reduce_sum(tf.stack(outputs), axis=0)
        fact_vecs = tf.nn.dropout(fact_vecs, self.dropout_placeholder)

        return fact_vecs

    def get_question_representation(self):
        '''
        计算输入问题的表征
        :return:
        '''
        questions = tf.nn.embedding_lookup(self.embeddings, self.question_holder)
        gru_cell = tf.contrib.rnn.GRUCell(self.config.hidden_size)
        _, q_vec = tf.nn.dynamic_rnn(
            gru_cell,
            questions,
            dtype=np.float32,
            sequence_length=self.question_len_holder
        )

        return q_vec

    def get_attention(self, q_vec, fact_vec, prev_memory, reuse):
        '''
        计算注意力分值
        :param q_vec:
        :param fact_vec:
        :param prev_memory:
        :param reuse:
        :return:
        '''
        with tf.variable_scope('attention', reuse=reuse):
            features = [fact_vec*q_vec, fact_vec*prev_memory, tf.abs(fact_vec - q_vec), tf.abs(fact_vec - prev_memory)]
            feature_vec = tf.concat(features, 1)
            attention = tf.contrib.layers.fully_connected(feature_vec,
                                                          self.config.embed_size,
                                                          activation_fn=tf.nn.tanh,
                                                          reuse=reuse,
                                                          scope='fc1')
            attention = tf.contrib.layers.fully_connected(attention,
                                                          1,
                                                          activation_fn=None,
                                                          reuse=reuse,
                                                          scope='fc2')
            return attention

    def generate_episode(self, memory, q_vec, fact_vecs, hop_index):
        '''
        计算下一轮的情景记忆内容
        :param memory:
        :param q_vec:
        :param fact_vecs:
        :param hop_index:
        :return:
        '''
        attentions = [tf.squeeze(self.get_attention(q_vec, memory, fv, bool(hop_index) or bool(i)), axis=1)
                      for i, fv in enumerate(tf.unstack(fact_vecs, axis=1))]
        attentions = tf.transpose(tf.stack(attentions))
        self.attentions.append(attentions)
        attentions = tf.nn.softmax(attentions)
        attentions = tf.expand_dims(attentions, axis=-1)
        reuse = True if hop_index > 0 else False
        print('fact_vecs:', fact_vecs.shape)
        print('attentions:', attentions.shape)
        gru_inputs = tf.concat([fact_vecs, attentions], 2)
        print('gru_inputs:', gru_inputs.shape)
        with tf.variable_scope('attention_gru', reuse=reuse):
            _, episode = tf.nn.dynamic_rnn(
                AttentionGRUCell(self.config.hidden_size),
                gru_inputs,
                dtype=np.float32,
                sequence_length=self.input_len_holder
            )
        return episode

    def add_answer_module(self, rnn_output, q_vec):
        '''
        计算答案
        :param rnn_output:
        :param q_vec:
        :return:
        '''
        rnn_output = tf.nn.dropout(rnn_output, self.dropout_placeholder)
        output = tf.layers.dense(
            tf.concat([rnn_output, q_vec], 1),
            self.vocab_size,
            activation=None
        )

        return output

    def inference(self):
        '''
        根据记忆内容计算输出表征
        :return:
        '''
        with tf.variable_scope('question', initializer=tf.contrib.layers.xavier_initializer()):
            q_vec = self.get_question_representation()

        with tf.variable_scope('input', initializer=tf.contrib.layers.xavier_initializer()):
            fact_vecs = self.get_input_represetation()

        self.attentions = []

        with tf.variable_scope('memory', initializer=tf.contrib.layers.xavier_initializer()):
            prev_memory = q_vec

            for i in range(self.config.num_hops):
                episode = self.generate_episode(prev_memory, q_vec, fact_vecs, i)
                with tf.variable_scope('hop_%d' % i):
                    prev_memory = tf.layers.dense(
                        tf.concat([prev_memory, episode, q_vec], 1),
                        self.config.hidden_size,
                        activation=tf.nn.relu
                    )
            output = prev_memory

        with tf.variable_scope('answer', initializer=tf.contrib.layers.xavier_initializer()):
            output = self.add_answer_module(output, q_vec)

        return output

    def get_predictions(self, output):
        preds = tf.nn.softmax(output)
        return tf.argmax(preds, 1)

    def add_loss_op(self, output):
        gate_loss = 0
        if self.config.strong_supervision:
            for i, attention in enumerate(self.attentions):
                labels = tf.gather(tf.transpose(self.rel_label_holder), 0)
                gate_loss += tf.reduce_sum(tf.nn.sparse_softmax_cross_entropy_with_logits(logits=attention, labels=labels))

        loss = self.config.beta*tf.reduce_sum(tf.nn.sparse_softmax_cross_entropy_with_logits(logits=output, labels=self.answer_holder)) + gate_loss

        for v in tf.trainable_variables():
            if not 'bias' in v.name.lower():
                loss += self.config.l2 * tf.nn.l2_loss(v)

        tf.summary.scalar('loss', loss)

        return loss

    def add_training_op(self, loss):
        optimizer = tf.train.AdamOptimizer(learning_rate=self.config.lr)
        grad_and_vars = optimizer.compute_gradients(loss)

        if self.config.clip_grads:
            grad_and_vars = [(tf.clip_by_norm(grad, self.config.max_grad_value), var) for grad, var in grad_and_vars]

        train_op = optimizer.apply_gradients(grad_and_vars)
        return train_op

    def run_epoch(self, session, data, num_epoch=0, train_writer=None, train_op=None, verbose=2, train=False):
        '''
        计算训练集或测试集当前epoch的平均损失和准确率
        :param session:
        :param data:
        :param num_epoch:
        :param train_writer:
        :param train_op:
        :param verbose:
        :param train:
        :return:
        '''
        dropout = self.config.dropout
        total_steps = len(data[0])
        total_loss = []
        accuracy = 0
        p = np.random.permutation(len(data[0]))
        qp, ip, ql, il, im, a, r = data
        qp, ip, ql, il, im, a, r = qp[p], ip[p], ql[p], il[p], im[p], a[p], r[p]
        for step in range(total_steps/self.config.batch_size):
            index = range(step*self.config.batch_size, (step+1)*self.config.batch_size)
            feed = {
                self.question_holder: qp[index],
                self.input_placeholder: ip[index],
                self.question_len_holder: ql[index],
                self.input_len_holder: il[index],
                self.answer_holder: a[index],
                self.rel_label_holder: r[index],
                self.dropout_placeholder: dropout
            }
            if train_op is not None:
                loss, pred, summary, _ = session.run([self.calculate_loss, self.pred, self.merged, train_op],
                                                     feed_dict=feed)
            else:
                loss, pred, summary = session.run([self.calculate_loss, self.pred, self.merged], feed_dict=feed)
            if train_writer is not None:
                train_writer.add_summary(summary, num_epoch*total_steps+step)
            answers = a[step*self.config.batch_size:(step+1)*self.config.batch_size]
            accuracy += np.sum(answers==pred)/len(answers)
            total_loss.append(loss)

        return np.mean(total_loss), accuracy/float(total_steps)

基于注意力机制的AttentionGRUCell类的代码如下

class AttentionGRUCell(RNNCell):
    '''
    基于Attention机制的GRUCell实现
    '''

    def __init__(self, num_units, activation=tanh):
        RNNCell.__init__(self)
        self._num_units = num_units
        self._activation_fn = activation

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

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

    def __call__(self, inputs, state, scope=None):
        with tf.variable_scope(scope or 'attention_gru_cell'):
            with tf.variable_scope('gates'):
                inputs, z = tf.split(inputs, num_or_size_splits=[self._num_units, 1], axis=1)
                r = self._linear([inputs, state], self._num_units, True)
                r = tf.nn.sigmoid(r)
            with tf.variable_scope('candidate'):
                r = r * self._linear(state, self._num_units, False)
            with tf.variable_scope('input'):
                x = self._linear(inputs, self._num_units,True)
            h_hat = self._activation_fn(r + x)
            new_h = (1 - z) * state + z * h_hat
        return new_h, new_h

    def _linear(self, args, output_size, bias, bias_start=0.0):
        if not nest.is_sequence(args):
            args = [args]
        total_arg_size = 0
        shapes = [a.get_shape() for a in args]
        for shape in shapes:
            total_arg_size += shape[1].value

        dtype = [a.dtype for a in args][0]
        scope = tf.get_variable_scope()
        with tf.variable_scope(scope) as outer_scope:
            weights = tf.get_variable('weights', [total_arg_size, output_size], dtype=dtype)
            if len(args) == 1:
                res = tf.matmul(args[0], weights)
            else:
                res = tf.matmul(tf.concat(args, 1), weights)
            if not bias:
                return res
            with tf.variable_scope(outer_scope) as inner_scope:
                inner_scope.set_partitioner = None
                biases = tf.get_variable('biases', [output_size], dtype=dtype,
                                         initializer=tf.constant_initializer(bias_start, dtype=dtype))
            return tf.nn.bias_add(res, biases)

babi数据集的预处理、加载代码参考文末的参考资料链接[3]。

3. 参考资料

展开全文 >>

Contextual Multi-armed Bandit算法学习笔记

2018-01-02

1. multi-armed bandit问题定义

multi-arm-bandit

多臂老虎机问题的定义如下：
1. 一排K个老虎机$a_i\{i=1,2,…,K\}$；
2. 每个老虎机对应有回报$r_i\{i=1,2,…,K\}$，事先并不知道；
3. 在每一个时间t拉动一个老虎机会产生一个回报$r_{t}$；
4. 目标：根据特定的策略选择$a$，从而最小化累计遗憾值$L_{T}=E[\sum \limits_{\tau=1}^{T}r^{\star}-r_{\tau}]$。这里的$r^{\star}$是最优策略产生的回报。这里的最小化累积遗憾值相当于最大化累积回报值。
5. 这里就会产生一个强化学习中经典的exploration&exploitation问题，即每次贪婪地选择当前最优回报值的臂，还是探索潜在的新臂以期获得更高价值。
6. 常见场景：推荐系统中的冷启动问题、广告系统中的广告选择、金融衍生品设计
针对这里的策略选择方法，Richar Sutton在《Reinforcement Learning: An Introduction》一书中提到了如下contextual free的方案:
- $\epsilon-greedy$: 每一次以概率$\epsilon$选择当前最优策略，以概率$1-\epsilon$选择其他策略。这里的$\epsilon$可以灵活的控制模型的偏好程度。但这样做没有利用到探索各个臂时获得的信息，仍然不做区分的exploration效果不佳。
- UCB(Upper Confidence Bound): 采取一种乐观的态度，根据每个臂的预期回报的不确定性的上界来选择。对于每个臂尝试的次数越多，预期回报的置信区间越窄；某个臂的尝试次数越少，置信区间越宽。对于某个臂$a_{i}$，其期望回报上界为$\bar{x}_{i}=\sqrt{\frac{2lnn}{n_{i}}}$，$\bar{x}_{i}$为臂$i$的平均回报，$n_{i}$为臂$i$的选择次数，$n$为总的选择次数。UCB不需任何参数，需要遍历所有的臂，比较耗时；并且在刚开始各个臂选择次数比较少时，结果波动比较大。
- Thompson Smapling: 采用贝叶斯思想，每个臂有一个beta(win, lose)分布来估计产生回报的概率；不断试验，每次选一个臂，有回报则win增加1，否则lose增加1，不断调整beta分布的参数。每次选择臂时，每个分布随机产生一个数b，然后根据最大的b选择相应的臂。

2. contextual multi-armed bandit

每个arm包含特定的随时间而变化的上下文特征$x_{t}$，并且其回报$r_{t}$和上下文特征$x_{t}$直接相关，即存在一个函数$f$，使得$r_{t}=f(x_{t})$。这里的$x_{t}$即为实际应用中抽象出的辅助选择每个arm的特征，针对不同的互联网推荐系统场景，可产生不同的用户和物品的特征。针对不同的有关函数$f$的假设，会得到不同的模型，下面的LinUCB就是其中的典型一类。

2.1 LinUCB(Linear Upper Confidence Bound)

LinUCB由前雅虎的研究员Lihong Li(现就任微软)等在2010年提出，被应用到雅虎首页的个性化新闻推荐中。这里假设每个臂的预期回报$r_{t,a}$和上下文$x_{t}$之间存在线性关系, 即$E[r_{t,a}|x_{t,a}]=x_{t,a}^{t}\theta_{a}$.

考虑在新闻推荐的场景中，回报$r_{t,a}$可以简单的看成是否点击(0/1)，arm即为需要推荐的新闻。此时，可以给出一个对模型参数$\theta_{a}$的如下岭回归估计： $\hat{\theta_{a}}=(D_{a}^{T}D_{a}+I_{d})^{-1}D_{a}^{T}c_{a}$
这里的$D_{a}$是m个维度为d的训练样本构成的矩阵，$c_{a}$是标签向量。
此外，这里有$|x_{t,a}^{T}\hat{\theta_{a}}-E[r_{t,a}|x_{t,a}]| \leq \alpha \sqrt{x_{t,a}^{T}D_{a}^{T}D_{a}+I_{d})^{-1}x_{t,a}}$。
注意到这里arm的选择策略：$a_{t} = \arg\max\limits_{a \in A_{t}}(x_{t,a}^{T}+\alpha \sqrt{x_{t,a}^{T}A_{a}^{-1}x_{t,a}})$。
实际使用中，arm的选择通常存在一个Exploration&Exploitation的问题，这里采用的是UCB的方案，并且将置信上界的计算直接融入到模型优化中，实际算法如下图所示：

LinUCB

2.2 Hybrid LinUCB

上面的模型中各个arm的参数$\theta$互不影响，分开计算；但实际中可能有一些特征是所有arm共享的。因此，衍生出了如下模型：

$E[r_{t,a}|x_{t,a}] = z_{t,a}^{T}\beta^{*}+x_{t,a}^{T}\theta_{a}^{*}$

这里的参数$\beta^{*}$由所有的arm共享。实际算法如下图所示：

Hybrid-LinUCB

2.3 LinUCB算法的优点

考虑了上下文特征，可以处理动态的推荐资源池，在资源池增大时，效果更好。
收敛速度比UCB快，但需要合适的特征工程方案，这是实际使用中工程人员需要发挥的地方。

3. 参考资料

[1] A Contextual-Bandit Approach to Personalized News Article Recommendation
[2] 推荐系统的EE问题及Bandit算法
[3] Bandit算法与推荐系统

展开全文 >>

强化学习中的A3C算法

2017-12-17

1. A3C(Asynchronous Advantage Actor Critic)算法背景介绍

A3C算法[1]由DeepMind在2015年提出，基于Actor-Critic框架(参考下文)，并且通过异步并行训练多个agent的方式，在Atari 2600、TORCS、MuJoCo等视频游戏中都取得了优异的效果。此外，A3C既能适应于离散动作空间的控制，也能适应于连续动作空间的控制。

2. Actor-Critic(行动家-评论家)算法介绍

结合使用强化学习中经典的值函数方法和策略梯度方法，所谓的Actor即是策略输出函数，根据agent的状态state产生输出动作action，直观上可以认为它作为大脑来控制agent的行动；Critic即为每个状态的值函数，根据训练过程中获得的历史回报reward来自我调整，同时影响Actor的训练。
论文中的Actor和Critic都使用卷积神经网络来近似，两者共享相同的卷积层，但对应于不同的输出。

2.1 策略网络

$\pi(a|s, \theta)$即为策略函数，根据状态s直接输出动作a，$\theta$是需要优化的参数，优化目标是使得好的策略的出现概率高，坏的策略的出现概率低。
根据策略梯度定理，可以得到$\theta$的更新公式为：$\theta \leftarrow \theta + (\sum\limits_{t=1}^{T}\triangledown_{\theta}log\pi(a_{t}|s_{t}; \theta))(\sum\limits_{t=1}^{T}r(s_{t}, a_{t}))$，但直接这样做带来的问题是高方差、学习效率低，因为这里的梯度值对于所有action来说都是正的。
在Actor-Critic框架中，引入了基于状态值函数估计得到的Advantage函数$A^{\pi}(s,a)=r(s,a)+\gamma V^{\pi}(s’) - V^{\pi}(s)$来辅助更新策略函数的参数，这里使用值函数作为一种baseline，降低了实际更新时参数的方差，advantage函数可看作在状态s采取行动a时，相对于当前状态值函数的一种优势。因为值函数是对状态未来预期价值的一种估计。于是，这里$\theta$的更新公式变成$\theta \leftarrow \theta + (\sum\limits_{t=1}^{T}\triangledown_{\theta}log\pi(a_{t}|s_{t}; \theta))(\sum\limits_{t=1}^{T}(r(s_{t},a_{t})+\gamma V^{\pi}(s_{t+1}) - V^{\pi}(s_{t})))$。当然Advantage函数的形式不只这一种，具体可参考[2]。

2.2 值网络

在上述的Actor-Critic框架中，值网络$V(s, \theta’)$主要用来辅助策略网络的参数$\theta$的更新；同时，值网络同样需要进行训练，值网络的参数更新公式为：$\theta’ \rightarrow \theta’ + \partial(r_{t}-V(s_{t}, \theta’))^{2}/\partial \theta’$

3. A3C算法

actor-critic-architecture

在上述的算法流程图和示意图[3]中，值得注意的关键点有：

有一个中央控制大脑Global Network，存储着全局参数，负责异步接收各个agent(Worker)的参数梯度来更新全局参数。

各个agent同步中央大脑的参数到自身，再各自进行单独的actor训练。每个agent都先根据策略网络进行训练数据的收集，再累积多个时间步的批量样本来更新策略网络和值网络，这样保证了训练的稳定性。

4. 代码实现[4]

下面以经典Atari游戏中的Breakout为例，基于tensorflow给出整个A3C算法的代码，超参数定义部分省略。

4.1 定义网络结构


class A3CNet(object):
    '''
    策略网络和值网络结合的网络，二者共享网络的卷积层，策略网络负责根据状态选择动作，值网络负责指导策略网络的更新
    '''
    def __init__(self, state_shape, action_dim, scope):

        with tf.variable_scope('%s_placeholder' % scope):
            self.state = tf.placeholder(dtype=tf.float32, shape=[None] + state_shape, name='state')
            self.action = tf.placeholder(dtype=tf.float32, shape=[None, action_dim], name='action')
            self.target_q = tf.placeholder(dtype=tf.float32, shape=[None])

        #shared convnet part
        with tf.variable_scope('%s_shared' % scope):
            self.W1 = tf.get_variable(name='W1', shape=[8, 8, state_shape[-1], 16], initializer=tf.random_uniform_initializer())
            self.b1 = tf.get_variable(name='b1', shape=[16], initializer=tf.constant_initializer(0.0))
            conv1 = tf.nn.relu(tf.nn.bias_add(tf.nn.conv2d(self.state, self.W1, strides=[1, 4, 4, 1], padding='VALID', name='conv1'), self.b1))

            self.W2 = tf.get_variable(name='W2', shape=[4, 4, 16, 32], initializer=tf.random_uniform_initializer())
            self.b2 = tf.get_variable(name='b2', shape=[32], initializer=tf.constant_initializer(0.0))
            conv2 = tf.nn.relu(tf.nn.bias_add(tf.nn.conv2d(conv1, self.W2, strides=[1, 2, 2, 1], padding='VALID', name='conv1'), self.b2))

            flatten1 = tf.reshape(conv2, shape=[-1, 9*9*32], name='flatten1')

            self.W3 = tf.get_variable(name='W3', shape=[9*9*32, 256], initializer=tf.random_uniform_initializer())
            self.b3 = tf.get_variable(name='b3', shape=[256], initializer=tf.constant_initializer(0.0))

            fc1 = tf.nn.relu(tf.matmul(flatten1, self.W3) + self.b3)

        #policy net peculiar
        with tf.variable_scope('%s_policy' % scope):
            self.p_W1 = tf.get_variable(name='p_W1', shape=[256, 256], initializer=tf.random_uniform_initializer())
            self.p_b1 = tf.get_variable(name='p_b1', shape=[256], initializer=tf.constant_initializer(0.0))
            p_fc1 = tf.nn.relu(tf.matmul(fc1, self.p_W1) + self.p_b1)
            self.p_W2 = tf.get_variable(name='p_W2', shape=[256, action_dim], initializer=tf.random_uniform_initializer())
            self.p_b2 = tf.get_variable(name='p_b2', shape=[action_dim], initializer=tf.constant_initializer(0.0))
            p_fc2 = tf.nn.relu(tf.matmul(p_fc1, self.p_W2) + self.p_b2)
            self.policy_out = tf.nn.softmax(p_fc2, name='policy_out')

        #value net peculiar
        with tf.variable_scope('%s_value' % scope):
            self.v_W1 = tf.get_variable(name='v_W1', shape=[256, 256], initializer=tf.random_uniform_initializer())
            self.v_b1 = tf.get_variable(name='v_b1', shape=[256], initializer=tf.constant_initializer(0.0))
            v_fc1 = tf.nn.relu(tf.matmul(fc1, self.v_W1) + self.v_b1)
            self.v_W2 = tf.get_variable(name='v_W2', shape=[256, 1], initializer=tf.random_uniform_initializer())
            self.v_b2 = tf.get_variable(name='v_b2', shape=[1], initializer=tf.constant_initializer(0.0))
            v_fc1 = tf.nn.relu(tf.matmul(v_fc1, self.v_W2) + self.v_b2)
            self.value_out = tf.reshape(v_fc1, [-1], name='value_out')

        #loss value
        with tf.name_scope('%s_loss'%scope, values=[self.policy_out, self.value_out]):
            self.entropy = tf.reduce_sum(self.policy_out * tf.log(self.policy_out + flags.eps))
            self.td_error = self.target_q - self.value_out
            policy_prob = tf.log(tf.reduce_sum(tf.multiply(self.policy_out, self.action), reduction_indices=1))
            self.policy_loss = -tf.reduce_sum(policy_prob * tf.abs(self.td_error)) + self.entropy * flags.entropy_beta
            self.value_loss = tf.reduce_sum(tf.square(self.td_error))

            self.policy_grads = tf.gradients(self.policy_loss, [self.p_W1, self.p_b1, self.p_W2, self.p_b2])
            self.value_grads = tf.gradients(self.value_loss, [self.v_W1, self.v_b1, self.v_W2, self.v_b2])

            self.shared_loss = self.policy_loss + self.value_loss * 0.5
            self.share_grads = tf.gradients(self.shared_loss, [self.W1, self.b1, self.W2, self.b2, self.W3, self.b3])

    def get_policy(self, sess, state):
        return sess.run(self.policy_out, feed_dict={self.state: [state]})[0]

    def get_value(self, sess, state):
        return sess.run(self.value_out, feed_dict={self.state: [state]})[0]

    def get_vars(self):

        return [self.W1, self.b1, self.W2, self.b2, self.W3, self.b3,
                self.p_W1, self.p_b1, self.p_W2, self.p_b2,
                self.v_W1, self.v_b1, self.v_W2, self.v_b2]

上述代码中定义了策略网络和值网络的结构、损失值、梯度值。

4.2 定义各个Agent


class Worker(threading.Thread):
    '''
    负责对中央大脑的值网络和策略网络的参数进行更新和同步的Worker线程
    '''
    def __init__(self, thread_id, master):

        self.thread_id = thread_id
        threading.Thread.__init__(self, name='thread_%d'%thread_id)
        self.env = gym.make('Breakout-v0')
        self.master = master

        self.local_net = A3CNet(state_shape=[84, 84, 4], action_dim=self.env.action_space.n, scope='local_net_%s'%thread_id)
        self.sync = self.sync_network(master.global_net)
        self.accum_grads = self.create_accumulate_gradients()
        self.do_accum_grads_ops = self.do_accmulate_gradients()
        self.reset_accum_grads_ops = self.reset_accumulate_gradients()
        summaries = list()
        summaries.append(tf.summary.scalar('entropy/%d' % thread_id, self.local_net.entropy))
        summaries.append(tf.summary.scalar('policy_loss/%d' % thread_id, self.local_net.policy_loss))
        summaries.append(tf.summary.scalar('value_loss/%d' % thread_id, self.local_net.value_loss))
        summaries.append(tf.summary.scalar('shared_loss/%d' % thread_id, self.local_net.shared_loss))
        self.summary_op = tf.summary.merge(summaries)
        #裁剪各个local net的梯度值
        clip_accum_grads = [tf.clip_by_value(grad, -flags.grad_clip, flags.grad_clip) for grad in self.accum_grads]
        self.apply_gradients = master.optimizer.apply_gradients(
            zip(clip_accum_grads, master.global_net.get_vars()), global_step=master.global_step
        )

    #同步global net的参数到local net
    def sync_network(self, master):
        sync_ops = []
        with tf.name_scope(name='sync_ops_%d'%self.thread_id):
            for (worker_var, master_var) in zip(self.local_net.get_vars(), master.get_vars()):
                ops = tf.assign(worker_var, master_var)
                sync_ops.append(ops)
            return tf.group(*sync_ops, name='sync_group_%d'%self.thread_id)

    #创建各个local net的累积更新梯度值
    def create_accumulate_gradients(self):
        accum_grads = []

        with tf.name_scope(name='create_accum_%d' % self.thread_id, values=[self.local_net]):
            for var in self.local_net.get_vars():
                zero = tf.zeros(var.get_shape().as_list(), dtype=var.dtype)
                accum_grad = tf.Variable(zero, name=var.name.replace(':', '_')+'_accum_grad', trainable=False)
                accum_grads.append(accum_grad)
            return accum_grads

    #累积各个local net的梯度值
    def do_accmulate_gradients(self):
        accum_grad_ops = []

        with tf.name_scope(name='accum_ops_%d'%self.thread_id, values=[]):
            grads = self.local_net.share_grads + self.local_net.policy_grads + self.local_net.value_grads
            for (grad, var, accum_grad) in zip(grads, self.local_net.get_vars(), self.accum_grads):
                accum_ops = tf.assign_add(accum_grad, grad, name=var.name.replace(':', '_')+'accum_grad_ops')
                accum_grad_ops.append(accum_ops)
            return tf.group(*accum_grad_ops, name='accum_group_%d'%self.thread_id)

    #重置各个local net的累积梯度值
    def reset_accumulate_gradients(self):
        reset_grad_ops = []

        with tf.name_scope(name='reset_grad_ops_%d'%self.thread_id, values=[self.local_net]):
            for(var, accum_grad) in zip(self.local_net.get_vars(), self.accum_grads):
                zero = tf.zeros(shape=var.get_shape().as_list(), dtype=var.dtype)
                reset_ops = tf.assign(accum_grad, zero, name=var.name.replace(':', '_')+'reset_grad_ops')
                reset_grad_ops.append(reset_ops)
            return tf.group(*reset_grad_ops, name='accum_group_%d'%self.thread_id)

    #收集训练数据
    def forward_explore(self, train_step):
        terminal = False
        t_start = train_step
        rollout = {
            'state': [],
            'action': [],
            'reward': [],
            'done': []
        }
        state = np.zeros([84, 84, 4], dtype=np.float32)
        while not terminal and (train_step - t_start <= flags.t_max):
            pi_probs = self.local_net.get_policy(self.master.sess, state)
            #基于epsilon-greedy策略的action选择
            if random.random() < 0.8:
                action = np.argmax(pi_probs)
            else:
                action = random.randint(0, self.env.action_space.n - 1)
            observation, reward, terminal = None, None, None
            for _ in xrange(flags.frame_skip):
                observation, reward, terminal, _ = self.env.step(action)
                if terminal:
                    break
            observation = np.resize(observation, new_shape=[84, 84, 1])/256.0
            state = np.append(state[:, :, 1:], observation, axis=2)
            reward = np.clip(reward, -1, 1)
            train_step += 1
            rollout['state'].append(state)
            onehot_action = np.zeros(self.env.action_space.n)
            onehot_action[action] = 1
            rollout['action'].append(onehot_action)
            rollout['reward'].append(reward)
            rollout['done'].append(terminal)
        return train_step, rollout

    def run(self):
        sess = self.master.sess
        self.env.reset()
        loop = 0
        while flags.train_step <= flags.t_train:
            train_step = 0
            loop += 1
            sess.run(self.reset_accum_grads_ops)
            sess.run(self.sync)
            train_step, rollout = self.forward_explore(train_step)
            if rollout['done'][-1]:
                rollout['reward'][-1] = 0
                self.env.reset()
            else:
                rollout['reward'][-1] = self.local_net.get_value(sess, rollout['state'][-1])
            rollout['return'] = scipy.signal.lfilter([1], [1, flags.gamma], rollout['reward'][::-1], axis=0)[::-1]
            fetches = [self.do_accum_grads_ops, self.master.global_step]
            lc_net = self.local_net
            if loop % 10 == 0:
                fetches.append(self.summary_op)
            result = sess.run(fetches, feed_dict={
                lc_net.state: rollout['state'],
                lc_net.action: rollout['action'],
                lc_net.target_q: rollout['return']
            })
            if loop % 10 == 0:
                global_step, summary_str = result[1:3]
                self.master.summary_writer.add_summary(summary_str, global_step=global_step)
                self.master.global_step_val = int(global_step)
            sess.run(self.apply_gradients)
            flags.train_step += train_step
            if loop % 10 == 0 and self.thread_id == 0:
                print('thread_id=%d, loop=%d, train_step=%d' % (self.thread_id, loop, flags.train_step))
                self.test()
            if loop % 1000 == 0 and self.thread_id == 0:
                checkpoint_filename = self.master.saver.save(sess, flags.train_dir + '/' + 'a3c_model', global_step=self.master.global_step_eval)

    def test(self, episode=10, max_step=1e3):
        rewards = []
        start_time = time.time()
        while episode > 0:
            terminal = False
            self.env.reset()
            episode_reward = 0
            test_step = 0
            state = np.zeros([84, 84, 4], dtype=np.float32)
            while not terminal and test_step < max_step:
                pi_probs = self.local_net.get_policy(self.master.sess, state)
                action = np.argmax(pi_probs)
                observation, reward = None, None
                for _ in xrange(flags.frame_skip):
                    observation, reward, terminal, _ = self.env.step(action)
                    if terminal:
                        break
                observation = np.resize(observation, new_shape=[84, 84, 1])/256.0
                state = np.append(state[:, :, 1:], observation, axis=2)
                test_step += 1
                reward = np.clip(reward, -1, 1)
                episode_reward += reward
            rewards.append(episode_reward)
            episode -= 1
        elapsed_time = int(time.time() - start_time)
        print('train-step = %d, avg_reward = %.2f, time=%d(s)' % (flags.train_step, reduce(lambda x, y: x+y, rewards), elapsed_time))

上述代码定义了每个Agent同步来自全局大脑的参数、梯度累积、数据收集、独自训练、效果测试的过程。

4.3 执行实际的训练过程


class A3CAtari(object):
    '''
    定义中央大脑、执行整个网络的训练
    '''
    def __init__(self):
        self.env = gym.make('Breakout-v0')
        self.graph = tf.get_default_graph()
        self.global_net = A3CNet(state_shape=[84, 84, 4], action_dim=self.env.action_space.n, scope='global_net')
        self.optimizer = tf.train.AdamOptimizer(flags.learning_rate, name='global_optimizer')
        self.global_step = tf.get_variable('global_step', [], initializer= tf.constant_initializer(0), trainable=False)
        self.summary_writer = tf.summary.FileWriter(logdir=flags.train_dir, graph=self.graph)
        self.jobs = []
        for thread_id in xrange(flags.jobs):
            self.jobs.append(Worker(thread_id, self))
        self.sess = tf.Session(config=tf.ConfigProto(log_device_placement=False, allow_soft_placement=True))
        self.sess.run(tf.global_variables_initializer())
        self.global_step_eval = 0
        self.saver = tf.train.Saver(var_list=self.global_net.get_vars(), max_to_keep=3)
        #self.saver.restore(self.sess, flags.train_dir)

    def train(self):
        flags.train_step = 0
        signal.signal(signal.SIGINT, signal_handler)
        for job in self.jobs:
            job.start()
        for job in self.jobs:
            job.join()

def signal_handler():
    sys.exit(0)

def main(_):
    if not os.path.isdir(flags.train_dir):
        os.mkdir(flags.train_dir)
    for f in os.listdir(flags.train_dir):
        if re.search('.*tfevents.*', f):
            os.remove(os.path.join(flags.train_dir, f))
    model = A3CAtari()
    model.train()

if __name__ == '__main__':
    tf.app.run()

上述代码定义了中央大脑和各个训练Worker，启动整个网络的训练。

参考资料

[1] Asynchronous Methods for Deep Reinforcement Learning
[2] High-Dimensional Continuous Control Using Generalized Advantage Estimation
[3] Simple Reinforcement Learning with Tensorflow Part 8: Asynchronous Actor-Critic Agents (A3C)
[4] Github上A3C算法的实现

展开全文 >>

爬虫框架Scrapy学习笔记

2017-11-26

1. Scrapy框架简介

Scrapy是一个python编写的、快速、强大的爬虫框架。有如下特点：

1. 基于Twisted的异步、非阻塞的网络请求，不用等待上一个网络请求结束才开始下一个请求。
2. 基于Xpath和CSS从HTML和XML数据源中提取数据。
3. 支持将数据导出为Json/CSV/XML格式。
4. 交互式命令行工具创建爬虫、开始爬取、分析爬取数据、调试、部署爬虫。

2. scrapy爬取流程

scrapy_architecture

八步爬取流程
1. Engine根据初始url产生Request
2. Engine请求Scheduler生成Request对象
3. Scheduler返回Request给Engine
4. Engine发送请求给Downloader
5. Downloader产生Response，给到Engine
6. Engine转发Response给Spider
7. Spider处理Response产生item和新的Request给Engine
8. Engine发送item给Item Pipeline进一步处理，以及将处理过的Request给Scheduler来获取下一个Request
  3. 爬虫项目生成和结构
scrapy通过自带的shell命令生成爬虫项目，以及启动爬虫。操作步骤如下：
第一步: 生成爬虫项目
1
scrapy startproject tutorial

项目目录结构如下：

tutorial/
    scrapy.cfg            # deploy configuration file

    tutorial/             # project's Python module, you'll import your code from here
        __init__.py

        items.py          # project items definition file

        pipelines.py      # project pipelines file

        settings.py       # project settings file

        spiders/          # a directory where you'll later put your spiders
            __init__.py

第二步：生成爬虫类，进入tutorial目录，执行下面命令

1	scrapy genspider quotes quotes.toscrape.com

这里，quotes爬虫名称，quotes.toscrape.com是要爬取的网站域名。
第三步：编写爬虫类，编辑quotes.py文件如下：

import scrapy


class QuotesSpider(scrapy.Spider):
    name = "quotes"

    def start_requests(self):
        urls = [
            'http://quotes.toscrape.com/page/1/',
            'http://quotes.toscrape.com/page/2/',
        ]
        for url in urls:
            yield scrapy.Request(url=url, callback=self.parse)

    def parse(self, response):
        page = response.url.split("/")[-2]
        filename = 'quotes-%s.html' % page
        with open(filename, 'wb') as f:
            f.write(response.body)
        self.log('Saved file %s' % filename)

这里urls中的是起始爬取url，start_requests方法根据url产生请求，parse方法解析返回的response，并将body写入文件。
第四步：运行爬虫

1	scrapy crawl quotes

4. scrapy核心类介绍

scrapy.spiders.Spider：爬虫基类，所有自定义爬虫都要继承它。关键属性和方法如下：
- name：爬虫名字
- start_urls：起始爬取url列表
- start_requests()：产生初始爬取的、可迭代的Request对象
- parse(response): 处理Response的回调函数
- log(message): 产生日志
- from_crawler(crawler, args, *kwargs)：创建爬虫的类方法
scrapy.selector.Selector: 基于lxml库的数据提取器类
- xpath(query): 根据xpath字符串查询节点
- css(query)
- extract()：返回字符串列表。
- re(regex)
scrapy.item.Item: 通用的数据容器类
- Field：指定元数据的类
- fields：包含所有field的字典
  - scrapy.loader.ItemLoader：提取item的类
  - load_item()
  - Item Pipeline: 对爬取数据构造的Item对象进行后续清洗的类，通常需要实现以下方法。
  - process_item(self, item, spider)
  - open_spider(self, spider)
  - close_spider(self, spider)
  - from_crawler(cls, crawler)
  - scrapy.exporters.BaseItemExporter：导出item到外部存储，支持Json、CSV、XML，通常和Item Pipeline结合使用。
  - start_exporting()
  - export_item(item)
  - finish_exporting()

Item Pipeline和Item Exporter结合的示例如下：

from scrapy import signals
from scrapy.exporters import XmlItemExporter

class XmlExportPipeline(object):

    def __init__(self):
        self.files = {}

     @classmethod
     def from_crawler(cls, crawler):
         pipeline = cls()
         crawler.signals.connect(pipeline.spider_opened, signals.spider_opened)
         crawler.signals.connect(pipeline.spider_closed, signals.spider_closed)
         return pipeline

    def spider_opened(self, spider):
        file = open('%s_products.xml' % spider.name, 'w+b')
        self.files[spider] = file
        self.exporter = XmlItemExporter(file)
        self.exporter.start_exporting()

    def spider_closed(self, spider):
        self.exporter.finish_exporting()
        file = self.files.pop(spider)
        file.close()

    def process_item(self, item, spider):
        self.exporter.export_item(item)
        return item

scrapy.http.Request：Http请求类，由Downloader执行，返回Response
scrapy.http.FormRequest(url, formdata): Post请求类，formdata是填充表单数据的字典。
scrapy.http.Response：Http响应类
- urljoin(url): 根据相对url和response的url构造绝对url
- follow(url, callback): 根据url产生Request对象
scrapy.linkextractors.LinkExtractor: 提取response中的链接
- extrack_links()
scrapy.crawler.CrawlerProcess: 同一个进程内运行多个爬虫类
scrapy.crawler.CrawlerRunner: 跟踪、管理、运行多个爬虫的工具类
scrapy.downloadermiddlewares.DownloaderMiddleware: 位于Engine类和Downloader类之间，处理request和response，需要在settings.py中配置DOWNLOADER_MIDDLEWARES项，包含如下内置的子类：
- CookiesMiddleware
- DownloadTimeoutMiddleware
- HttpCompressionMiddleware
- RobotsTxtMiddleware
scrapy.spidermiddlewares.SpiderMiddleware: 位于Engine类和Spider类之间，需要在settings.py中配置SPIDER_MIDDLEWARES项，包含如下内置的子类：
- DepthMiddleware
- HttpErrorMiddleware

参考资料

展开全文 >>

广告ctr预估中的FM和FFM模型

2017-11-26

1. 在线广告广告ctr预估

点击率(click-through rate)和转化率(conversion rate)是衡量广告流量的两个重要指标。点击率与点击价值的乘积决定了广告的排列顺序。在大规模广告系统中，常用的ctr预估模型是LR(Logistic Regression)，LR易于并行化，配合大量的人工特征工程，可以产生很好的效果。但LR本质上是一种广义线性模型，分类能力有限。因此，近年来出现了以GBDT(Gradient Boosting Decision Tree)、FM(actorization Machine)及其改进FFM(Field-aware Factorization Machine)为代表的非线性模型，在一些CTR预估竞赛中表现出优异的效果。本文将对FM和FFM进行简短概要的介绍。

2. FM模型

FM(Factorization Machine)由Steffen Randle于2010年提出，起初是用于推荐系统中稀疏数据场景下的评分预测问题。它属于一种在多项式模型上衍生出来的模型。

二阶FM的预测公式： $\phi_{FM}(w, x) = w_{0}+\sum\limits_{i=1}^{n}w_{i}x_{i}+\sum\limits_{i=1}^{n-1}\sum\limits_{j=i+1}^{n}w_{ij}x_{i}x_{j} = w_{0}+\sum\limits_{i=1}^{n}w_{i}x_{i}+\sum\limits_{i=1}^{n-1}\sum\limits_{j=i+1}^{n}(v_{i}^{T}v_{j})x_{i}x_{j}$ 注意这里直接计算$\phi_{FM}(w, x)$的时间复杂度是$O(kn^{2})$。经过变换二次项，时间复杂度可降为$O(kn)$。即： $\sum\limits_{i=1}^{n}\sum\limits_{j=i+1}^{n}<v_{i}, v_{j}>x_{i}x_{j}=\frac{1}{2}\sum\limits_{f=1}^{k}((\sum\limits_{i=1}^{n}v_{i,f}x_{i})^{2}-\sum\limits_{i=1}^{n}v_{i,f}^{2}x_{i}^{2})$ FM通过将特征变量交叉项的系数分解到隐变量空间中，可以更好的在稀疏数据中估计二次项的参数。FM模型以其通用的表达形式可以直接适用于回归、分类和排序任务。
模型学习
针对FM模型，可以直接使用SGD进行模型参数更新，公式如下：
\begin{equation}
\frac{\partial \phi_{FM}(w, x)}{\partial w}=
\left\{
\begin{aligned}
1, \theta = w_{0} \\
x_{i}, \theta = w_{i} \\
x_{i}\sum\limits_{j=1}^{n}v_{j,f}x_{j}-v_{i,f}x_{i}^{2}, \theta = v_{i,f}
\end{aligned}
\right.
\end{equation}

3. FM模型的改进: FFM模型

FFM(Field-aware Factorization Machine)由Yuchin Juan等在2014年参加kaggle criteo展示广告点击率预测比赛时提出。在FM模型基础之上引入了特征域的概念。

预测公式： $\phi_{FFM}(w, x) = w_{0}+\sum\limits_{i=1}^{n}w_{i}x_{i}+\sum\limits_{i=1}^{n-1}\sum\limits_{j=i+1}^{n}(v_{i,f_{1}} \cdot v_{j,f_{2}})x_{i}x_{j}$ 这里，$f_{1}, f_{2}$是特征$x_{i}$和$x_{j}$所属的域，每一个特征所属的域是唯一的，$x_{i}$和$x_{j}$被分解到了对应域的隐空间中。二次项的总共有$\frac{n(n-1)}{2}$个。
模型学习：
1. 计算子梯度
  $g_{i, f_{1}} = \triangledown_{w_{i,f_{1}}}f(w)=\lambda w_{i,f_{1}} + \kappa w_{j, f_{2}}$
  $g_{j, f_{2}} = \triangledown_{w_{j, f_{2}}}f(w) = \lambda w_{j,f_{2}} + \kappa w_{i, f_{1}}$
  这里，$\kappa = \frac{-y}{1+exp(y\phi_{FFM}(w,x))}$
2. 计算总梯度
  $(G_{i, f_{1}})_{d} = (G_{i, f_{1}})_{d} + (g_{i, f_{1}})_{d}^{2}$
  $(G_{j, f_{2}})_{d} = (G_{j, f_{2}})_{d} + (g_{j, f_{2}})_{d}^{2}$
3. 参数更新
  $(w_{i, f_{1}})_{d} = (w_{i, f_{1}})_{d} - \frac{\eta}{\sqrt{(G_{i, f_{1}})_{d}}}(g_{i,f_{1}})_{d}$
  $(w_{j, f_{2}})_{d} = (w_{j, f_{2}})_{d} - \frac{\eta}{\sqrt{(G_{j, f_{2}})_{d}}}(g_{j,f_{2}})_{d}$
  
  这里，参数更新时，参考了Adagrad算法的自适应学习率方法。FFM模型整体的训练时间复杂度是$O(kn^{2})$，相比FM更高。FFM在实际使用时容易过拟合，需要使用Early Stopping策略。并且需要将原始数据转换成”field_id:feature_id:value”的格式。实际使用时，数值型特征分配单独的field编号，类别特征ohe后的所有特征属于同一个域。
FFM的适用场景
- 数据集包含大量类别特征，并进行了ohe编码
- 转换后的数据集比较稀疏

4. 其他杂项

针对展示广告中的点击率或转化率预估模型，通常需要求解如下无约束优化问题：

$min_{w} \frac{\lambda}{2}||w||_{2}^{2}+\sum\limits_{i=1}^{m}log(1+exp(-y_{i}\phi(w, x_{i})))$

这里的特征一般含用户侧(性别、年龄、教育背景、收入、偏好等)、广告侧(创意、素材、标题等)、上下文(时间、位置等)等特征，实际使用时需要针对模型的特性做不同的特征工程。
优化算法一般有一阶优化方法SGD(Stochastic Gradient Descent)、二阶拟牛顿优化方法L-BFGS(Limited memory Broyden–Fletcher–Goldfarb–Shanno)、Trust-Region。这里就暂时不展开讲述，留在后续文章中介绍。

5. 参考资料

[1] 深入FFM原理与实践-美团点评技术博客
[2] Factorization Machines-steffen randle
[3] Field-aware Factorization Machines for CTR Prediction
[4] kaggle: Display Advertising Challenge

展开全文 >>

Reinforcement Learning-学习笔记

2017-11-26

1. 定义和特点

强化学习(Reinforcement Learning)解决的是代理(Agent)在未知环境(Environment)中通过交互(Interaction)进行学习的方法。不同于监督学习和无监督学习，属于第三种机器学习范式。(By Richard Sutton)

没有样本标签，只有行动的奖励信号。

最终反馈结果的获得是延迟的。

主要针对有前后关系的序列数据，无独立同分布假设。

当前行为会影响后续收到的回报值大小。

2. 四个基本组件

策略(Policy): 当前Agent所处状态到下一步采取行动的映射，是强化学习的学习目标。
回报(Reward): 衡量Agent采取的行动的好坏的值。
状态值函数(Value Function): 从当前状态开始获得的预期长期累计回报，代表对当前状态价值高低的估计。
环境模型(Model): 对当前环境中各个变量变化情况的建模表示。典型的是离散变量下的概率分布模型。

3. 四个重要概念

Agent: 学习在未知环境中做决策的对象。例如棋类游戏中的棋子。
Environment: Agent的交互对象，针对Agent的行动，给出回报及下一步的状态。
Policy: $\pi_{t}(a_{t}|s_{t})$

Agent-Environment

长期累积回报: $G_{t}=R_{t+1}+\gamma R_{t+2}+\gamma^{2}R_{t+3}+\cdots=\sum\limits_{k=0}^{n}\gamma^{k}R_{t+k+1}$, $\gamma$ 是折扣因子。Agent的目标就是通过与环境的不断交互，学习最优的策略，从而最大化长期累计回报。

4. MDP(Markov Decision Process)

MDP描述了一种用概率分布建模的环境，环境中的状态转移具有马尔科夫性，即下一时刻的状态只与当前状态s和采取的动作a有关，而与之前的状态和动作无关。针对强化学习要解决的环境交互式学习任务，往往简化其环境是MDP，方便求解。

$P(S_{t+1}=s', R_{t+1}=r|S_{0},A_{0},R_{1},...,S_{t},A_{t})=P(S_{t+1}=s', R_{t+1}|S_{t}=s,A_{t}=a)$

5. 状态值函数，动作价值函数与Bellamn方程

$v_{\pi}(s) = E_{\pi}[G_{t}|S_{t}=s]=E_{\pi}[\sum\limits_{k=0}^{n}\gamma^{k}R_{t+k+1}|S_{t}=s]$

$q_{\pi}(s, a) = E_{\pi}[G_{t}|S_{t}=s, A_{t}=a]=E_{\pi}[\sum\limits_{k=0}^{n}\gamma^{k}R_{t+k+1}|S_{t}=s,A_{t}=a]$

Bellman方程: $v_{\pi}(s) = \sum\limits_{a}\pi(a|s)\sum\limits_{s’, r}p(s’,r|s,a)[r+\gamma v_{\pi}(s’)]$ <—> 状态值函数递归求解方程

最优值函数: $v_{\star}(s) = \max\limits_{\pi}v_{\pi}(s) \Rightarrow v_{\star}(s)=\max\limits_{a \in A(s)}=\sum\limits_{s’, r}p(s’,r|s,a)[r+\gamma v_{\star}(s’)]$

最优动作价值函数: $q_{\star}(s, a) = \max\limits_{\pi}q_{\pi}(s, a) \Rightarrow q_{\star}(s, a)=\sum\limits_{s’, r}p(s’, r|s,a)[r + \gamma \max\limits_{a’}q_{\star}(s’, a’)]$

动作价值函数在求解最优策略时更加直接，所以更加常用。

6. 依赖模型的最优策略函数计算：策略迭代与值迭代

策略迭代: 包含策略评估和策略提升两个部分。

Policy-Iteraion

值迭代: 直接迭代计算状态值函数。

Value-Iteraion

策略迭代和值迭代都是一种动态规划的思路，都需要事先知道环境的具体MDP模型，但实际情况中很难预先估计。

7. 模型无关的值函数计算: Monte-Carlo和Temporal-Difference

模型无关的值函数估计方法不需要事先知道环境的具体模型，主要通过与环境交互过程中产生的实际经验${s_{t},a_{t},r_{t+1},s_{t+1},a_{t+1},r_{t+2},…,s_{T},a_{T},r_{T+1}}$来直接估计值函数。

Monte-Carlo方法: 使用完整经验片段(episode)中最终回报的平均值来估计值函数。

Monte-Carlo

Temporal-Difference方法: 从不完整的经验片段中估计值函数。

Temporal-Difference

Q-Learning: TD方法的一种，直接估计动作价值Q函数。

Q-Learning

8. 近似求解方案

前面的强化学习算法一般是将(s,a)存在一张表中，在状态空间或动作空间很大时，耗费内存空间并且泛化能力差，使用函数来近似值函数或动作价值函数可以避免这些问题。即$v_{\pi}(s) \approx \hat{v}(s,w)或q_{\pi}(s)=\hat{q}(s,a,w)$，这里w是函数的参数。

常见的近似函数优线性函数和神经网络。线性函数往往需要精心构造表示状态的特征，而神经网络可以直接接受原始输入。

Deepmind开发的玩Atari游戏的DQN和围棋程序Alphago的策略网络都使用了卷积神经网络来近似值函数或动作价值函数，展现了优异的结果。

9. 策略梯度方法

直接近似策略函数: $\pi\{a|s,\theta\}=p(A_{t}=a|S_{t}=s,\theta_{t}=\theta)$。直接根据状态计算出最终的动作值.

REINFORCE方法: 根据策略函数梯度更新参数值。

REINFORCE

Actor-Critic方法: 结合使用值函数近似和策略函数近似。值函数作为critic来知道策略函数的训练。

Actor-Critic

10. 强化学习的应用

游戏AI: 视频游戏(Atari、星际争霸)，棋类游戏(围棋)
Web服务: 新闻(Yahoo!)、广告、商品个性化推荐(Alibaba)
机器人控制
聊天机器人

11. 参考资料

[1]Reinforcement Learning Lecture by David Silver
[2]CS 294: Deep Reinforcement Learning by UC Berkeley
[3]强化学习by莫凡
[4]Reinforcement Learning: An Introduction (2nd Edition) by Richar Sutton
[5] https://github.com/ShangtongZhang/reinforcement-learning-an-introduction
[6]OpenAI Gym(比较不同强化学习算法的软件环境平台，包含很多游戏环境)
[7]Deepmind Rearch Blog
[8]Deep Reinforcement Learning: An Overview by Yuxi Li

展开全文 >>

1. 为什么要做ctr(click througth rate)预估

2. 广告ctr预估怎么转化为数学问题

3. ctr预估问题的难点在哪

4. 工业界怎么做ctr预估

5. 工业界ctr预估模型的演进

5.1 general linear model

5.2. non-linear model

5.2.1 Factorization Machine及其变种

5.2.2 Neural Network及和其他模型的融合

6. 其他问题

1.TensorFlow简介

2. Dataset和Estimator简介

2.1 Dataset介绍

2.2 Estimator介绍

3 Dataset和Estimator的完整使用流程

1. 原始记忆网络(Memory Network)

1.1 模型结构

1.2 文本问答

2.动态记忆网络(Dynamic Memory Network)

2.1 模型架构

2.2 代码实现

3. 参考资料

1. multi-armed bandit问题定义

2. contextual multi-armed bandit

2.1 LinUCB(Linear Upper Confidence Bound)

2.2 Hybrid LinUCB

2.3 LinUCB算法的优点

3. 参考资料

1. A3C(Asynchronous Advantage Actor Critic)算法背景介绍

2. Actor-Critic(行动家-评论家)算法介绍

2.1 策略网络

2.2 值网络

3. A3C算法

4. 代码实现[4]

4.1 定义网络结构

4.2 定义各个Agent

4.3 执行实际的训练过程

参考资料

1. Scrapy框架简介

2. scrapy爬取流程

3. 爬虫项目生成和结构

4. scrapy核心类介绍

参考资料

1. 在线广告广告ctr预估

2. FM模型

3. FM模型的改进: FFM模型

4. 其他杂项

5. 参考资料

1. 定义和特点

2. 四个基本组件

3. 四个重要概念

4. MDP(Markov Decision Process)

5. 状态值函数，动作价值函数与Bellamn方程

6. 依赖模型的最优策略函数计算：策略迭代与值迭代

7. 模型无关的值函数计算: Monte-Carlo和Temporal-Difference

8. 近似求解方案

9. 策略梯度方法

10. 强化学习的应用

11. 参考资料