1. 贝叶斯公式

import math
import jieba
import re
import os
import json
from collections import defaultdictjieba.initialize()"""
贝叶斯分类实践P(A|B) = (P(A) * P(B|A)) / P(B)
事件A：文本属于类别x1。文本属于类别x的概率，记做P(x1)
事件B：文本为s (s=w1w2w3..wn)
P(x1|s) = 文本为s，属于x1类的概率.   #求解目标#
P(x1|s) = P(x1|w1, w2, w3...wn) = P(w1, w2..wn|x1) * P(x1) / P(w1, w2, w3...wn)P(x1) 任意样本属于x1的概率。x1样本数/总样本数
P(w1, w2..wn|x1) = P(w1|x1) * P(w2|x1)...P(wn|x1)  词的独立性假设
P(w1|x1) x1类样本中，w1出现的频率公共分母的计算，使用全概率公式：
P(w1, w2, w3...wn) = P(w1,w2..Wn|x1)*P(x1) + P(w1,w2..Wn|x2)*P(x2) ... P(w1,w2..Wn|xn)*P(xn)
"""class BayesApproach:def __init__(self, data_path):self.p_class = defaultdict(int)self.word_class_prob = defaultdict(dict)self.load(data_path)def load(self, path):self.class_name_to_word_freq = defaultdict(dict)self.all_words = set()  #汇总一个词表with open(path, encoding="utf8") as f:for line in f:line = json.loads(line)class_name = line["tag"]title = line["title"]words = jieba.lcut(title)self.all_words = self.all_words.union(set(words))self.p_class[class_name] += 1  #记录每个类别样本数量word_freq = self.class_name_to_word_freq[class_name]#记录每个类别下的词频for word in words:if word not in word_freq:word_freq[word] = 1else:word_freq[word] += 1self.freq_to_prob()return#将记录的词频和样本频率都转化为概率def freq_to_prob(self):#样本概率计算total_sample_count = sum(self.p_class.values())self.p_class = dict([c, self.p_class[c] / total_sample_count] for c in self.p_class)#词概率计算self.word_class_prob = defaultdict(dict)for class_name, word_freq in self.class_name_to_word_freq.items():total_word_count = sum(count for count in word_freq.values()) #每个类别总词数for word in word_freq:#加1平滑，避免出现概率为0，计算P(wn|x1)prob = (word_freq[word] + 1) / (total_word_count + len(self.all_words))self.word_class_prob[class_name][word] = probself.word_class_prob[class_name]["<unk>"] = 1/(total_word_count + len(self.all_words))return#P(w1|x1) * P(w2|x1)...P(wn|x1)def get_words_class_prob(self, words, class_name):result = 1for word in words:unk_prob = self.word_class_prob[class_name]["<unk>"]result *= self.word_class_prob[class_name].get(word, unk_prob)return result#计算P(w1, w2..wn|x1) * P(x1)def get_class_prob(self, words, class_name):#P(x1)p_x = self.p_class[class_name]# P(w1, w2..wn|x1) = P(w1|x1) * P(w2|x1)...P(wn|x1)p_w_x = self.get_words_class_prob(words, class_name)return p_x * p_w_x#做文本分类def classify(self, sentence):words = jieba.lcut(sentence) #切词results = []for class_name in self.p_class:prob = self.get_class_prob(words, class_name)  #计算class_name类概率results.append([class_name, prob])results = sorted(results, key=lambda x:x[1], reverse=True) #排序#计算公共分母：P(w1, w2, w3...wn) = P(w1,w2..Wn|x1)*P(x1) + P(w1,w2..Wn|x2)*P(x2) ... P(w1,w2..Wn|xn)*P(xn)#不做这一步也可以，对顺序没影响，只不过得到的不是0-1之间的概率值pw = sum([x[1] for x in results]) #P(w1, w2, w3...wn)results = [[c, prob/pw] for c, prob in results]#打印结果for class_name, prob in results:print("属于类别[%s]的概率为%f" % (class_name, prob))return resultsif __name__ == "__main__":path = "../data/train_tag_news.json"ba = BayesApproach(path)query = "中国三款导弹可发射多弹头 美无法防御很急躁"ba.classify(query)

2. 支持向量机（SVM）

#!/usr/bin/env python3  
#coding: utf-8#使用基于词向量的分类器
#对比几种模型的效果import json
import jieba
import numpy as np
from gensim.models import Word2Vec
from sklearn.metrics import classification_report
from sklearn.svm import SVC
from collections import defaultdictLABELS = {'健康': 0, '军事': 1, '房产': 2, '社会': 3, '国际': 4, '旅游': 5, '彩票': 6, '时尚': 7, '文化': 8, '汽车': 9, '体育': 10, '家居': 11, '教育': 12, '娱乐': 13, '科技': 14, '股票': 15, '游戏': 16, '财经': 17}#输入模型文件路径
#加载训练好的模型
def load_word2vec_model(path):model = Word2Vec.load(path)return model#加载数据集
def load_sentence(path, model):sentences = []labels = []with open(path, encoding="utf8") as f:for line in f:line = json.loads(line)title, content = line["title"], line["content"]sentences.append(" ".join(jieba.lcut(title)))labels.append(line["tag"])train_x = sentences_to_vectors(sentences, model)train_y = label_to_label_index(labels)return train_x, train_y#tag标签转化为类别标号
def label_to_label_index(labels):return [LABELS[y] for y in labels]#文本向量化，使用了基于这些文本训练的词向量
def sentences_to_vectors(sentences, model):vectors = []for sentence in sentences:words = sentence.split()vector = np.zeros(model.vector_size)for word in words:try:vector += model.wv[word]# vector = np.max([vector, model.wv[word]], axis=0)except KeyError:vector += np.zeros(model.vector_size)vectors.append(vector / len(words))return np.array(vectors)def main():model = load_word2vec_model("model.w2v")train_x, train_y = load_sentence("../data/train_tag_news.json", model)test_x, test_y = load_sentence("../data/valid_tag_news.json", model)classifier = SVC()classifier.fit(train_x, train_y)y_pred = classifier.predict(test_x)print(classification_report(test_y, y_pred))if __name__ == "__main__":main()

核函数：

假设存在一个特征映射函数 ϕ，使得 K(x,y)=ϕ(x)⋅ϕ(y)。核技巧通过直接使用 K(x,y) 计算内积，而无需明确地知道或计算 ϕ(x)。核函数的作用是可以低维映射到高维，从而进行分类。

3. CNN神经网络

这个地方有点小坑，bias偏移量的个数等于卷积核的个数，如下面表示的是8个卷积核，每个卷积核是6*2的矩阵，和我们输入的6*8的矩阵去做对位相乘，得到的是6*（8 - 2 + 1）也就是6*7的矩阵然后sum求和得到一个数值，求完和以后加上一个常数值bias，第几个卷积核，加的就是第几个bias值。这样计算完第一个卷积，得到的是一维的数组，形状变成1 * 7。因为有8个卷积核，就需要计算八次，最后得到8*7的矩阵。这里的卷积核8和输入的矩阵列维度8没有关系。

import torch
import torch.nn as nn
import numpy as np#使用pytorch的1维卷积层input_dim = 6
hidden_size = 8
kernel_size = 2
torch_cnn1d = nn.Conv1d(input_dim, hidden_size, kernel_size)
for key, weight in torch_cnn1d.state_dict().items():print(key, weight.shape)x = torch.rand((6, 8))  #embedding_size * max_lengthdef numpy_cnn1d(x, state_dict):weight = state_dict["weight"].numpy()bias = state_dict["bias"].numpy()sequence_output = []for i in range(0, x.shape[1] - kernel_size + 1):window = x[:, i:i+kernel_size]kernel_outputs = []for kernel in weight:kernel_outputs.append(np.sum(kernel * window))sequence_output.append(np.array(kernel_outputs) + bias)return np.array(sequence_output).Tprint(x.shape)
print(torch_cnn1d(x.unsqueeze(0)))
print(torch_cnn1d(x.unsqueeze(0)).shape)
print(numpy_cnn1d(x.numpy(), torch_cnn1d.state_dict()))

4. LSTM 

import torch
import torch.nn as nn
import numpy as np'''
用矩阵运算的方式复现一些基础的模型结构
清楚模型的计算细节，有助于加深对于模型的理解，以及模型转换等工作
'''#构造一个输入
length = 6
input_dim = 12
hidden_size = 7
x = np.random.random((length, input_dim))
# print(x)#使用pytorch的lstm层
torch_lstm = nn.LSTM(input_dim, hidden_size, batch_first=True)
for key, weight in torch_lstm.state_dict().items():print(key, weight.shape)def sigmoid(x):return 1/(1 + np.exp(-x))#将pytorch的lstm网络权重拿出来，用numpy通过矩阵运算实现lstm的计算
def numpy_lstm(x, state_dict):weight_ih = state_dict["weight_ih_l0"].numpy()weight_hh = state_dict["weight_hh_l0"].numpy()bias_ih = state_dict["bias_ih_l0"].numpy()bias_hh = state_dict["bias_hh_l0"].numpy()#pytorch将四个门的权重拼接存储，我们将它拆开w_i_x, w_f_x, w_c_x, w_o_x = weight_ih[0:hidden_size, :], \weight_ih[hidden_size:hidden_size*2, :],\weight_ih[hidden_size*2:hidden_size*3, :],\weight_ih[hidden_size*3:hidden_size*4, :]w_i_h, w_f_h, w_c_h, w_o_h = weight_hh[0:hidden_size, :], \weight_hh[hidden_size:hidden_size * 2, :], \weight_hh[hidden_size * 2:hidden_size * 3, :], \weight_hh[hidden_size * 3:hidden_size * 4, :]b_i_x, b_f_x, b_c_x, b_o_x = bias_ih[0:hidden_size], \bias_ih[hidden_size:hidden_size * 2], \bias_ih[hidden_size * 2:hidden_size * 3], \bias_ih[hidden_size * 3:hidden_size * 4]b_i_h, b_f_h, b_c_h, b_o_h = bias_hh[0:hidden_size], \bias_hh[hidden_size:hidden_size * 2], \bias_hh[hidden_size * 2:hidden_size * 3], \bias_hh[hidden_size * 3:hidden_size * 4]w_i = np.concatenate([w_i_h, w_i_x], axis=1)w_f = np.concatenate([w_f_h, w_f_x], axis=1)w_c = np.concatenate([w_c_h, w_c_x], axis=1)w_o = np.concatenate([w_o_h, w_o_x], axis=1)b_f = b_f_h + b_f_xb_i = b_i_h + b_i_xb_c = b_c_h + b_c_xb_o = b_o_h + b_o_xc_t = np.zeros((1, hidden_size))h_t = np.zeros((1, hidden_size))sequence_output = []for x_t in x:x_t = x_t[np.newaxis, :]hx = np.concatenate([h_t, x_t], axis=1)# f_t = sigmoid(np.dot(x_t, w_f_x.T) + b_f_x + np.dot(h_t, w_f_h.T) + b_f_h)f_t = sigmoid(np.dot(hx, w_f.T) + b_f)# i_t = sigmoid(np.dot(x_t, w_i_x.T) + b_i_x + np.dot(h_t, w_i_h.T) + b_i_h)i_t = sigmoid(np.dot(hx, w_i.T) + b_i)# g = np.tanh(np.dot(x_t, w_c_x.T) + b_c_x + np.dot(h_t, w_c_h.T) + b_c_h)g = np.tanh(np.dot(hx, w_c.T) + b_c)c_t = f_t * c_t + i_t * g# o_t = sigmoid(np.dot(x_t, w_o_x.T) + b_o_x + np.dot(h_t, w_o_h.T) + b_o_h)o_t = sigmoid(np.dot(hx, w_o.T) + b_o)h_t = o_t * np.tanh(c_t)sequence_output.append(h_t)return np.array(sequence_output), (h_t, c_t)torch_sequence_output, (torch_h, torch_c) = torch_lstm(torch.Tensor([x]))
numpy_sequence_output, (numpy_h, numpy_c) = numpy_lstm(x, torch_lstm.state_dict())print(torch_sequence_output)
print(numpy_sequence_output)
print("--------")
print(torch_h)
print(numpy_h)
print("--------")
print(torch_c)
print(numpy_c)##############################################################使用pytorch的GRU层
torch_gru = nn.GRU(input_dim, hidden_size, batch_first=True)
# for key, weight in torch_gru.state_dict().items():
#     print(key, weight.shape)#将pytorch的GRU网络权重拿出来，用numpy通过矩阵运算实现GRU的计算
def numpy_gru(x, state_dict):weight_ih = state_dict["weight_ih_l0"].numpy()weight_hh = state_dict["weight_hh_l0"].numpy()bias_ih = state_dict["bias_ih_l0"].numpy()bias_hh = state_dict["bias_hh_l0"].numpy()#pytorch将3个门的权重拼接存储，我们将它拆开w_r_x, w_z_x, w_x = weight_ih[0:hidden_size, :], \weight_ih[hidden_size:hidden_size * 2, :],\weight_ih[hidden_size * 2:hidden_size * 3, :]w_r_h, w_z_h, w_h = weight_hh[0:hidden_size, :], \weight_hh[hidden_size:hidden_size * 2, :], \weight_hh[hidden_size * 2:hidden_size * 3, :]b_r_x, b_z_x, b_x = bias_ih[0:hidden_size], \bias_ih[hidden_size:hidden_size * 2], \bias_ih[hidden_size * 2:hidden_size * 3]b_r_h, b_z_h, b_h = bias_hh[0:hidden_size], \bias_hh[hidden_size:hidden_size * 2], \bias_hh[hidden_size * 2:hidden_size * 3]w_z = np.concatenate([w_z_h, w_z_x], axis=1)w_r = np.concatenate([w_r_h, w_r_x], axis=1)b_z = b_z_h + b_z_xb_r = b_r_h + b_r_xh_t = np.zeros((1, hidden_size))sequence_output = []for x_t in x:x_t = x_t[np.newaxis, :]hx = np.concatenate([h_t, x_t], axis=1)z_t = sigmoid(np.dot(hx, w_z.T) + b_z)r_t = sigmoid(np.dot(hx, w_r.T) + b_r)h = np.tanh(r_t * (np.dot(h_t, w_h.T) + b_h) + np.dot(x_t, w_x.T) + b_x)h_t = (1 - z_t) * h + z_t * h_tsequence_output.append(h_t)return np.array(sequence_output), h_t# torch_sequence_output, torch_h = torch_gru(torch.Tensor([x]))
# numpy_sequence_output, numpy_h = numpy_gru(x, torch_gru.state_dict())
#
# print(torch_sequence_output)
# print(numpy_sequence_output)
# print("--------")
# print(torch_h)
# print(numpy_h)