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

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

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

この記事は前回書いた記事の続きです。

www.pytry3g.com

 

前回はJapaneseTextEncoderに指定した品詞を辞書に加えないように改良しました。

今回は以下の4つの点をJapaneseTextEncoderに加えてみます。

  1. データセットの用意
  2. パディング(padding)
  3. ミニバッチの用意
  4. PyTorchへの対応

これら4つの改良点について順番に書いていきます。

環境

関連リンク

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

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

1. データセットの用意

テキストコーパスをJapaneseTextEncoderに渡したら自動的に単語ID化したデータセットを作るようにしました。

JapaneseTextEncoderのコンストラクタ__init__の最後に以下を追加。

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

この変更により例えば下のような3つの文章からなるテキストコーパスをJapaneseTextEncoderに渡すと単語ID化されたdatasetが作られる。

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

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

2. パディング(padding)

次にパディングについて。これは今回の改良の目玉となるもの。

パディングについては以前書いた記事で少し触れました。

可変長のデータを扱うときはパディングを使うことにより入力データのサイズを揃えることができミニバッチ学習が可能になります。

今回実装するパディングの最大入力数はテキストコーパスの文章を形態素解析してできた形態素のリストの中で最も長いものとします。

下に例を示します。

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

>>> encoder = JapaneseTextEncoder(
                corpus,
                append_eos=True,
                padding=True
            )

>>> 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]]

JapaneseTextEncoderに渡したテキストコーパスの中で最も長いものはセネガルつええ、ボルト三体くらいいるわ笑笑になっています。作られたdatasetの中身を見てみると全てのデータのサイズが最も長い文章のサイズに合わせて作られています。

3. ミニバッチ学習

パディング機能を追加したのでミニバッチ学習ができるようになりました。

以下、ミニバッチ学習をするためのデータを返す関数です。デフォルトでバッチサイズは50としています。

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

PyTorchへの対応

とりあえずPyTorchに対応?できるようにしました。

データを渡せばtorch.tensorに変換してくれます。デフォルトではfloat型に変換します。

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

ソースコード

以下JapaneseTextEncoderのソースコード

import random
import torch
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 = [self.encode(sentence) for sentence in self.corpus]

    @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 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, data, size=50, shuffle=False):
        batch_dataset = []
        if shuffle:
            random.shuffle(data)
        for i in range(0, len(data), size):
            start = i
            end = start + size
            batch_dataset.append(data[start:end])
        return batch_dataset

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

    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

おわりに

今回の改良によりパディングを使ったミニバッチ学習が可能になったので、Seq2Seqを使った対話エージェントの学習に使ってみるつもりです。