どん底から這い上がるまでの記録

どん底から這い上がりたいけど這い上がれない人がいろいろ書くブログ(主にプログラミング)

日本語のテキストコーパスから辞書を作るライブラリを作りたい④

 

この記事は前回書いた記事の続きになります。

www.pytry3g.com

 

前回JapaneseTextEncoderにいくつかの機能を追加しましたが、JapaneseTextEncoderを使って言語モデルやSeq2Seqを実装をするにあたり改善すべき点がありそうなので、それらについてコードの変更をしていきます。

変更点

今回変更するところはほとんど前回の部分です。

  1. データセットの用意ー>変更
  2. ミニバッチ学習ー>変更
  3. PyTorchへの対応ー>廃止
  4. データのシャッフルー>追加

1. データセットの用意

前回JapaneseTextEncoderのコンストラクタ__init__の最後にテキストから単語IDに変換していましたが、これだとSeq2Seqのようなデータセットにはこの方法は使えないので言語モデルのデータセットの生成とSeq2Seqのデータセットの生成の部分を以下のように分けてみました。

言語モデル

言語モデル用のデータセット生成のために新しくbuild()を追加。

    def build(self):
        self.dataset = [self.encode(sentence) for sentence in self.corpus]

 

例えば以下のようなリストがself.corpusに入っていたとすると、

corpus = [
  "セネガルつええ、ボルト三体くらいいるわ笑笑", 
  "しょーみコロンビアより強い", 
  "それなまちがいないわ"
]

build()でテキストから単語IDのリストに変換するとこんな感じになる。

[[4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 14],
 [15, 16, 17, 18, 19, 20],
 [21, 22, 23, 24, 13]]

Seq2Seq

Seq2Seq用のデータセット生成のために新しくbuild_seq2seq()を追加。

    def build_seq2seq(self, corpus):
        """ Corpus must be list of lists.
            Each contained lists have two sentences message and reply.
            [
                ['message_1', 'reply_1'],
                ['message_2', 'reply_2']
            ]
        """
        self.dataset = [[self.encode(message), self.encode(reply)] for message, reply in corpus]

build()とは異なり、メソッドにデータセットを渡す必要がある。なお、データセットの形式は以下のようにする。

[["このメンツなら原口だな。", "高齢化の弊害"],
 ["コロンビアってサッカーつよいんですか", "つよいよ"]]

リストの中にメッセージと応答文を含んだリストが入った状態のものを用意する必要がある。

2. ミニバッチ学習

大きな変更はないが、デフォルトでデータセットをシャッフルするように変更し、データセットインスタンス変数に変更。

ただし、これは言語モデルでは使えなさそう、Seq2Seqには使えるはず。

    def get_batch_dataset(self, size=50, shuffle=True):
        batch_dataset = []
        if shuffle:
            self.shuffle()
        for i in range(0, len(self.dataset), size):
            start = i
            end = start + size
            batch_dataset.append(self.dataset[start:end])
        return batch_dataset

3. PyTorchへの対応

たぶん使わないから廃止。

    def to_tensor(self, data, dtype=torch.float):
        return torch.tensor(data, dtype=dtype)

4. データのシャッフル

データをシャッフルする。get_batch_dataset()でも使用。

    def shuffle(self):
        random.shuffle(self.dataset)

使用例

はやくこれを使って言語モデルとSeq2Seqをやりたい、、、

ソースコード

以下JapaneseTextEncoderのコード

import random
from collections import Counter
from reserved_tokens import SOS_INDEX
from reserved_tokens import EOS_INDEX
from reserved_tokens import UNKNOWN_INDEX
from reserved_tokens import RESERVED_ITOS
from reserved_tokens import PADDING_INDEX


class JapaneseTextEncoder:
    """ Encodes the text using a tokenizer.

    Args:
        corpus (list of strings): Text strings to build dictionary on.
        min_occurrences (int, optional): Minimum number of occurences for a token to be
            added to dictionary.
        append_sos (bool, optional): If 'True' append SOS token onto the begin to the encoded vector.
        append_eos (bool, optional): If 'True' append EOS token onto the end to the encoded vector.
        padding (bool, optional): If 'True' pad a sequence.
        filters (list of strings): Part of Speech strings to remove.
        reserved_tokens (list of str, optional): Tokens added to dictionary; reserving the first
            'len(reserved_tokens') indices.

    Example:
        >>> corpus = ["セネガルつええ、ボルト三体くらいいるわ笑笑", \
                      "しょーみコロンビアより強い", \
                      "それなまちがいないわ"]

        >>> encoder = JapaneseTextEncoder(
                corpus,
                append_eos=True,
                padding=True
            )
        >>> encoder.encode("コロンビア強い")
        [18, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]
        >>> encoder.vocab
        ['<pad>', '<unk>', '</s>', '<s>', 'セネガル', 'つえ', 'え', '、', 'ボルト', '三', '体', 'くらい', ' いる', 'わ', '笑', 'しょ', 'ー', 'み', 'コロンビア', 'より', '強い', 'それ', 'な', 'まちがい', 'ない']
        >>> encoder.decode(encoder.encode("コロンビア強い"))
        コロンビア強い</s>

        >>> encoder.dataset
        [[4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 14, 2],
         [15, 16, 17, 18, 19, 20, 0, 0, 0, 0, 0, 0, 2],
         [21, 22, 23, 24, 13, 0, 0, 0, 0, 0, 0, 0, 2]]

    """
    def __init__(self,
                 corpus,
                 min_occurrences=1,
                 append_sos=False,
                 append_eos=False,
                 padding=False,
                 filters=None,
                 reserved_tokens=RESERVED_ITOS):
        try:
            import MeCab
        except ImportError:
            print("Please install MeCab.")
            raise

        if not isinstance(corpus, list):
            raise TypeError("Corpus must be a list of strings.")

        self.corpus = corpus # sentence of list
        self.tagger = MeCab.Tagger("-Ochasen")
        self.append_sos = append_sos
        self.append_eos = append_eos
        self.padding = padding
        self.tokens = Counter()
        self.filters = ["BOS/EOS"]
        if filters is not None:
            if not isinstance(filters, list):
                raise TypeError("Filters must be a list of POS.")
            self.filters += filters

        self.maxlen = 0 # length of a sequence
        for sentence in self.corpus:
            tokens = self.tokenize(sentence)
            if tokens:
                self.tokens.update(tokens)
                self.maxlen = max(self.maxlen, len(tokens))

        self.itos = reserved_tokens.copy()
        self.stoi = {token: index for index, token in enumerate(reserved_tokens)}
        for token, cnt in self.tokens.items():
            if cnt >= min_occurrences:
                self.itos.append(token)
                self.stoi[token] = len(self.itos) - 1

        self.dataset = None # e.x. [[0, 1, 2], [3, 4, 2]]

    @property
    def vocab(self):
        return self.itos

    @property
    def word2id(self):
        return self.stoi

    @property
    def id2word(self):
        return {index: token for token, index in self.stoi.items()}

    def build(self):
        self.dataset = [self.encode(sentence) for sentence in self.corpus]

    def build_seq2seq(self, corpus):
        """ Corpus must be list of lists.
            Each contained lists have two sentences message and reply.
            [
                ['message_1', 'reply_1'],
                ['message_2', 'reply_2']
            ]
        """
        self.dataset = [[self.encode(message), self.encode(reply)] for message, reply in corpus]

    def encode(self, sentence, sos_index=SOS_INDEX, eos_index=EOS_INDEX, unknown_index=UNKNOWN_INDEX, padding_index=PADDING_INDEX):
        tokens = self.tokenize(sentence)
        if tokens is None:
            raise TypeError("Invalid type None...")
        indices = [self.stoi.get(token, unknown_index) for token in tokens]
        if self.padding:
            indices += [padding_index] * (self.maxlen-len(indices))
        if self.append_sos:
            indices.insert(0, sos_index)
        if self.append_eos:
            indices.append(eos_index)
        return indices

    def decode(self, indices):
        tokens = [self.itos[index] for index in indices]
        tokens = list(filter(lambda x: x != "<pad>", tokens))
        return "".join(tokens)

    def get_batch_dataset(self, size=50, shuffle=True):
        batch_dataset = []
        if shuffle:
            self.shuffle()
        for i in range(0, len(self.dataset), size):
            start = i
            end = start + size
            batch_dataset.append(self.dataset[start:end])
        return batch_dataset

    def shuffle(self):
        random.shuffle(self.dataset)

    def tokenize(self, sentence):
        tag = self.tagger.parseToNode(sentence)
        tokens = []
        while tag:
            features = tag.feature.split(",")
            pos = features[0]
            token = tag.surface
            if pos in self.filters:
                tag = tag.next
                continue
            tokens.append(token)
            tag = tag.next
        return tokens if tokens else None