語言模型（language model）是什麼

				
					#coding:utf8

import torch
import torch.nn as nn
import numpy as np
import math
import random
import os
import re
import matplotlib.pyplot as plt

"""
基於 pytorch 的 rnn 語言模型
"""

class LanguageModel(nn.Module):
    def __init__(self, input_dim, vocab):
        super(LanguageModel, self).__init__()
        self.embedding = nn.Embedding(len(vocab) + 1, input_dim)
        self.layer = nn.RNN(input_dim, input_dim, num_layers=2, batch_first=True)
        self.classify = nn.Linear(input_dim, len(vocab) + 1)
        self.dropout = nn.Dropout(0.1)
        self.loss = nn.functional.cross_entropy

    #當輸入真實標籤，返回 loss 值；無真實標籤，返回預測值
    def forward(self, x, y=None):
        x = self.embedding(x)  #output shape:(batch_size, sen_len, input_dim)
        x, _ = self.layer(x)      #output shape:(batch_size, sen_len, input_dim)
        x = x[:, -1, :]        #output shape:(batch_size, input_dim)
        x = self.dropout(x)
        y_pred = self.classify(x)   #output shape:(batch_size, input_dim)
        if y is not None:
            return self.loss(y_pred, y)
        else:
            return torch.softmax(y_pred, dim=-1)

#讀取 corpus 獲得字符集
#輸出一份
def build_vocab_from_corpus(path):
    vocab = set()
    with open(path, encoding="utf8") as f:
        for index, char in enumerate(f.read()):
            vocab.add(char)
    vocab.add("<UNK>") #增加一個 unk token 來處理未登入詞
    writer = open("vocab.txt", "w", encoding="utf8")
    for char in sorted(vocab):
        writer.write(char + "\n")
    return vocab

#加載字表
def build_vocab(vocab_path):
    vocab = {}
    with open(vocab_path, encoding="utf8") as f:
        for index, line in enumerate(f):
            char = line[:-1]        #去掉結尾換行符
            vocab[char] = index + 1 #留出0位给pad token
        vocab["\n"] = 1
    return vocab

#加載 corpus
def load_corpus(path):
    return open(path, encoding="utf8").read()

#隨機產生一個樣本
#從文字中截取隨機窗口，前n個字作為輸入，最後一個字作為輸出
def build_sample(vocab, window_size, corpus):
    start = random.randint(0, len(corpus) - 1 - window_size)
    end = start + window_size
    window = corpus[start:end]
    target = corpus[end]
    # print(window, target)
    x = [vocab.get(word, vocab["<UNK>"]) for word in window]   #將字轉換成序號
    y = vocab[target]
    return x, y

#建立資料集
#sample_length 輸入所需的樣本數量。需要多少生成多少
#vocab 詞表
#window_size 樣本長度
#corpus 字符串
def build_dataset(sample_length, vocab, window_size, corpus):
    dataset_x = []
    dataset_y = []
    for i in range(sample_length):
        x, y = build_sample(vocab, window_size, corpus)
        dataset_x.append(x)
        dataset_y.append(y)
    return torch.LongTensor(dataset_x), torch.LongTensor(dataset_y)

#建立模型
def build_model(vocab, char_dim):
    model = LanguageModel(char_dim, vocab)
    return model


#計算文字ppl
def calc_perplexity(sentence, model, vocab, window_size):
    prob = 0
    model.eval()
    with torch.no_grad():
        for i in range(1, len(sentence)):
            start = max(0, i - window_size)
            window = sentence[start:i]
            x = [vocab.get(char, vocab["<UNK>"]) for char in window]
            x = torch.LongTensor([x])
            target = sentence[i]
            target_index = vocab.get(target, vocab["<UNK>"])
            if torch.cuda.is_available():
                x = x.cuda()
            pred_prob_distribute = model(x)[0]
            target_prob = pred_prob_distribute[target_index]
            prob += math.log(target_prob, 10)
    return 2 ** (prob * ( -1 / len(sentence)))


def train(corpus_path, save_weight=True):
    epoch_num = 10        #訓練輪數
    batch_size = 128       #每次訓練樣本個數
    train_sample = 10000   #每輪訓練總共訓練的樣本總數
    char_dim = 128        #每個字的維度
    window_size = 6       #樣本文字長度
    vocab = build_vocab("vocab.txt")       #建立字表
    corpus = load_corpus(corpus_path)     #載入corpus
    model = build_model(vocab, char_dim)    #建立模型
    if torch.cuda.is_available():
        model = model.cuda()
    optim = torch.optim.Adam(model.parameters(), lr=0.001)   #建立優化器
    for epoch in range(epoch_num):
        model.train()
        watch_loss = []
        for batch in range(int(train_sample / batch_size)):
            x, y = build_dataset(batch_size, vocab, window_size, corpus) #建立一組訓練樣本
            if torch.cuda.is_available():
                x, y = x.cuda(), y.cuda()
            optim.zero_grad()    #梯度歸零
            loss = model(x, y)   #計算loss
            watch_loss.append(loss.item())
            loss.backward()      #計算梯度
            optim.step()         #更新權重
        print("=========\n第%d輪平均loss:%f" % (epoch + 1, np.mean(watch_loss)))
    if not save_weight:
        return
    else:
        base_name = os.path.basename(corpus_path).replace("txt", "pth")
        model_path = os.path.join("model", base_name)
        torch.save(model.state_dict(), model_path)
        return

#訓練corpus資料夾下的所有corpus，依照檔案名稱將訓練後的模型放到莫得了資料夾
def train_all():
    for path in os.listdir("corpus"):
        corpus_path = os.path.join("corpus", path)
        train(corpus_path)


if __name__ == "__main__":
    # build_vocab_from_corpus("corpus/all.txt")
    # train("corpus.txt", True)
    train_all()