1.stop_words=[] with open("data\stop_words.txt","r",encoding="utf-8") as f_stopwords: for line in f_stopwords: line=line.replace("\r","").replace("\n","").strip() stop_words.append(line) print(len(stop_words)) stop_words=set(stop_words) print(len(stop_words)) print(len(stop_words)) 2.raw_word_list=[] rules=u"([\u4e00-\u9fa5]+)" pattern=re.compile(rules) f_writer=open("../data/Seg_The_Smiling_Proud_Wanderer.txt","w",encoding="utf-8") with open("data/The_Smiling_Proud_Wanderer.txt","r",encoding="utf-8") as f_reader: lines = f_reader.readlines() for line in lines: line = line.replace("\r", "").replace("\n", "").strip() if line == "" or line is None: continue line = " ".join(jieba.cut(line)) seg_list = pattern.findall(line) word_list = [] for word in seg_list: if word not in stop_words: word_list.append(word) if len(word_list) > 0: raw_word_list.extend(word_list) line = " ".join(word_list) f_writer.write(line + "\n") f_writer.flush() f_writer.close() print(len(raw_word_list)) print(len(set(raw_word_list))) vocabulary_size=len(set(raw_word_list)) 3.vocabulary_size = len(set(raw_word_list)) words = raw_word_list count = [['UNK', -1]] count.extend(collections.Counter(words).most_common(vocabulary_size - 1)) print("count", len(count)) dictionary = dict() for word, _ in count: dictionary[word] = len(dictionary) data = list() unk_count = 0 for word in words: if word in dictionary: index = dictionary[word] else: index = 0 unk_count = unk_count + 1 data.append(index) count[0][1] = unk_count reverse_dictionary = dict(zip(dictionary.values(), dictionary.keys())) del words print(reverse_dictionary[1000]) print(data[:200]) data_index = 0 def generate_batch(batch_size, num_skips, skip_window): global data_index batch = np.ndarray(shape=(batch_size), dtype=np.int32) labels = np.ndarray(shape=(batch_size, 1), dtype=np.int32) span = 2 * skip_window + 1 buffer = collections.deque(maxlen=span) for _ in range(span): buffer.append(data[data_index]) data_index = (data_index + 1) % len(data) for i in range(batch_size // num_skips): target = skip_window targets_to_avoid = [skip_window] for j in range(num_skips): while target in targets_to_avoid: target = random.randint(0, span - 1) targets_to_avoid.append(target) batch[i * num_skips + j] = buffer[skip_window] labels[i * num_skips + j, 0] = buffer[target] buffer.append(data[data_index]) data_index = (data_index + 1) % len(data) return batch, labels batch, labels = generate_batch(batch_size=128, num_skips=4, skip_window=2) for i in range(10): print(batch[i], reverse_dictionary[batch[i]], '->', labels[i, 0], reverse_dictionary[labels[i, 0]]) batch_size = 128 embedding_size = 300 skip_window = 2 num_skips = 4 valid_window = 100 num_sample = 64 learning_rate = 0.01 valid_word = ['令狐冲', '左冷禅', '林平之', '岳不群', '桃根仙'] valid_example = [dictionary[ii] for ii in valid_word] data_index = 0 def next_batch(batch_size, num_skips, skip_window): global data_index assert batch_size % num_skips == 0 assert num_skips <= 2 * skip_window batch = np.ndarray(shape=(batch_size), dtype=np.int32) labels = np.ndarray(shape=(batch_size, 1), dtype=np.int32) span = 2 * skip_window + 1 buffer = collections.deque(maxlen=span) if data_index + span > len(data): data_index = 0 buffer.extend(data[data_index:data_index + span]) data_index += span for i in range(batch_size // num_skips): context_words = [w for w in range(span) if w != skip_window] words_to_use = random.sample(context_words, num_skips) for j, context_word in enumerate(words_to_use): batch[i * num_skips + j] = buffer[skip_window] labels[i * num_skips + j, 0] = buffer[context_word] if data_index == len(data): buffer.extend(data[0:span]) data_index = span else: buffer.append(data[data_index]) data_index += 1 data_index = (data_index + len(data) - span) % len(data) return batch, labels with tf.device('/cpu:0'): embedding = tf.Variable(tf.random.normal([vocabulary_size, embedding_size])) nce_weights = tf.Variable(tf.random.normal([vocabulary_size, embedding_size])) nce_biases = tf.Variable(tf.zeros([vocabulary_size])) def get_embedding(x): with tf.device('/cpu:0'): x_embed = tf.nn.embedding_lookup(embedding, x) return x_embed def nce_loss(x_embed, y): with tf.device('/cpu:0'): y = tf.cast(y, tf.int64) loss = tf.reduce_mean( tf.nn.nce_loss(weights=nce_weights, biases=nce_biases, labels=y, inputs=x_embed, num_sampled=num_sample, num_classes=vocabulary_size)) return loss def evaluate(x_embed): with tf.device('/cpu:0'): x_embed = tf.cast(x_embed, tf.float32) x_embed_norm = x_embed / tf.sqrt(tf.reduce_sum(tf.square(x_embed))) embedding_norm = embedding / tf.sqrt(tf.reduce_sum(tf.square(embedding), 1, keepdims=True), tf.float32) cosine_sim_op = tf.matmul(x_embed_norm, embedding_norm, transpose_b=True) return cosine_sim_op optimizer = tf.optimizers.SGD(learning_rate) def run_optimization(x, y): with tf.device('/cpu:0'): with tf.GradientTape() as g: emb = get_embedding(x) loss = nce_loss(emb, y) gradients = g.gradient(loss, [embedding, nce_weights, nce_biases]) optimizer.apply_gradients(zip(gradients, [embedding, nce_weights, nce_biases])) 4.训练x_test = np.array(valid_example) num_steps = 200000 avg_loss = 0 for step in range(num_steps): batch_inputs, batch_labels = generate_batch(batch_size, num_skips, skip_window) run_optimization(batch_inputs, batch_labels) loss = nce_loss(get_embedding(batch_inputs), batch_labels) avg_loss = avg_loss + loss if step % 5000 == 0: if step > 0: avg_loss = avg_loss / 5000 loss = nce_loss(get_embedding(batch_inputs), batch_labels) print("step: %i, loss: %f" % (step, loss)) if step % 10000 == 0: sim = evaluate(get_embedding(x_test)).numpy() for i in range(len(valid_word)): val_word = reverse_dictionary[valid_example[i]] top_k = 10 nearest = (-sim[i, :]).argsort()[1:top_k+1] sim_str = "与" + val_word + "最近的前10词是" for k in range(top_k): close_word = reverse_dictionary[nearest[k]] sim_str = "%s %s," % (sim_str, close_word) print(sim_str) 5.import pandas as pd from gensim.models import Word2Vec from gensim.models.word2vec import LineSentence from sklearn.manifold import TSNE import matplotlib.pyplot as plt import logging from matplotlib.font_manager import FontProperties pd.options.mode.chained_assignment = None font = FontProperties(fname=r"c:/windows/fonts/simsun.ttc", size=14) logging.basicConfig(format='%(asctime)s : %(levelname)s : %(message)s', level=logging.INFO) new_vec = open("../data/Seg_The_Smiling_Proud_Wanderer.txt", 'r', encoding='utf-8') model = Word2Vec(LineSentence(new_vec), sg=0, size=200, window=10, min_count=40, workers=6) print('模型训练完成') def tsne_plot(model): labels = [] tokens = [] for word in model.wv.vocab: tokens.append(model[word]) labels.append(word) tsne_model = TSNE(perplexity=40, n_components=2, init='pca', n_iter=1000, random_state=20) new_values = tsne_model.fit_transform(tokens) x = [] y = [] for value in new_values: x.append(value[0]) y.append(value[1]) plt.figure(figsize=(16, 16)) for i in range(len(x)): plt.scatter(x[i], y[i]) plt.annotate(labels[i], fontproperties=font, xy=(x[i], y[i]), xytext=(5, 2), textcoords='offset points', ha='right', va='bottom') plt.show() tsne_plot(model)