【その6】【自然言語処理】単語はベクトルではなく“非線形関数” ― 複素テイラー単語関数言語モデル ― 意味を“観測可能な量子状態”として扱うNLPモデルを作ってみた

この記事はシリーズものの5回目です。

1回目の記事はこちらです。
2回目の記事はこちらです。
3回目の記事はこちらです。
4回目の記事はこちらです。
5回目の記事はこちらです。

次回の記事は近日中に公開する予定です。

↓1回目の記事:

↓2回目の記事:

↓3回目の記事:

↓4回目の記事:

↓前回(5回目)の記事:

0. 前回までのあらすじ

$$\huge{「意味」とは「状態」だ。}$$

$$\huge{「単語」とは状態を変換する「非線形作用素」だ。}$$

こう考えると、「文」は「作用素の合成」として理解することが出来るし、
「意味を理解すること」とは「状態の観測」として説明することが出来る。

―こう考えて、量子力学の真似事をしながら、新しい自然言語処理モデル「複素テイラー単語関数言語モデル」の理論を組み立てたのが前々々々回(2回目)までだ。

そして、モデルクラス ComplexTaylorWordFunctionLM.py を書いたのが前々々回(3回目)。
学習器クラス ComplexTaylorWordFunctionLearner.py を書いたのが前々回(4回目)。
学習器を評価する評価器クラス ComplexTaylorWordFunctionEvaluator.py を書いたのが前回(5回目)だ。

前回までで、4-12節まで書いた。
今回の記事は、4-13節から開始する。

4. コーディング

複素テイラー単語関数言語モデルを学習させて実験してみるためのコードを、PyTorch で作ってみよう。

ディレクトリ構成とステータス

(カレントディレクトリ)
├ main.py                               # 🛠️メインプログラム
├ ComplexTaylorWordFunctionLM.py        # ✅モデルクラス
├ ComplexTaylorWordFunctionLearner.py   # ✅モデルを学習するためのクラス
├ constants.py                          # ✅複数ファイルで共有する定数など (コーパス以外)
├ corpus.py                             # 📌トイコーパス
├ ComplexTaylorWordFunctionEvaluator.py # ✅学習結果や経過を評価するためのクラス
└ test_project.py                       # ✅テスト用コード

ステータスの凡例
😴: 未着手
🛠️: 今から作成・編集する
📌: 新たに編集が必要になる
👀: 作成済み（未テスト）
✅: 作成済み (テスト成功済み or テストを実施しない)
⚠️: 問題あり

今回は、4-13節と4-14節で、メインプログラム main.py をコーディングする。
main.py では、 from corpus import sentences, test_contexts のようにし、
モデルに sentences を 学習させ、そのモデルを test_contexts で評価するものとする。

次に、4-15節でトイコーパス corpus.py を書き、学習用の sentences:list[list[str]] と 評価用の test_contexts:list[list[str]] を用意する。
その後、次回の記事でメインプログラムを動かす実験をついに開始する。

4-13. メインプログラムのコーディング

メインプログラムでは

1. ハイパーパラメータと設定
2. トイコーパスの読み込み
3. 学習器の初期化
4. 評価用関数の準備
5. while True ループで学習と評価の実行

をやる。

4-13-1. ハイパーパラメータと設定

ハイパーパラメータと設定は次のコードで表現する。
( この内容が、import 文を除く main.py の冒頭になる。 )

ハイパーパラメータと設定

vars_ = set(globals())

# =========================
# 1. ハイパーパラメータと設定
# =========================
SEED = "A"
D = 16
K = 4
NEXT_IDENTITY_INIT = True
BATCH_SIZE = 32
LAMBDA_NORM = 1e0
CLIP_GRAD_NORM = 1e0
BASE_LR = 7e-4
NEXT_LR = 7e-4
WEIGHT_DECAY = 1e-4
SEQUENCE_LEN = 8

for var in list(globals()):
    if var in vars_:   continue
    if var == "vars_": continue
    print(f"{var}" + " " * (20 - len(var)) + f"= {globals()[var]}")

if SEED == "A":
    torch.manual_seed(0)
    random.seed(0)
    np.random.seed(0)
elif SEED == "B":
    torch.manual_seed(1)
    random.seed(2)
    np.random.seed(3)
elif SEED == "C":
    torch.manual_seed(1119)
    random.seed(21)
    np.random.seed(0x17ec084)
else:
    raise ValueError("SEED")

---

一旦:

vars_ = set(globals())

は無視しよう。

---

SEED = "A"
D = 16
K = 4
NEXT_IDENTITY_INIT = True
BATCH_SIZE = 32
LAMBDA_NORM = 1e0
CLIP_GRAD_NORM = 1e0
BASE_LR = 7e-4
NEXT_LR = 7e-4
WEIGHT_DECAY = 1e-4
SEQUENCE_LEN = 8

は、ハイパーパラメータおよび設定値だ。

SEED は実験の再現性のためのシードを決めるための値で、 "A","B","C" の中から選ぶものとする。

D は 意味=状態ベクトルの次元数、
K は テイラー次数の最大数 (関数を K次近似する)、
NEXT_IDENTITY_INIT は、 $\mathbb{next}$ 関数をどう初期化するか、

• True: 恒等写像で初期化
• False: 通常の単語同様にランダム初期化

BATCH_SIZE から WEIGHT_DECAY まではハイパーパラメータ、
SEQUENCE_LEN は訓練データの問題の最大長である。

---

for var in list(globals().keys()):
    if var in vars_:   continue
    if var == "vars_": continue
    print(f"{var}" + " " * (20 - len(var)) + f"= {globals()[var]}")

は、上述のハイパーパラメータや設定値を標準出力しているだけである。

---

if SEED == "A":
    torch.manual_seed(0)
    random.seed(0)
    np.random.seed(0)
elif SEED == "B":
    torch.manual_seed(1)
    random.seed(2)
    np.random.seed(3)
elif SEED == "C":
    torch.manual_seed(1119)
    random.seed(21)
    np.random.seed(0x17ec084)
else:
    raise ValueError("SEED")

は、SEED に応じてシード値を設定している。

4-13-2. トイコーパスの読み込み

main.py で「ハイパーパラメータと設定」の直後は、次のようにしようと思う。

トイコーパスの読み込み

# =========================
# 2. トイコーパス
# =========================

from corpus import sentences, test_contexts

vocab = sorted(set(sum(sentences, [])))

print("Vocab size (exclude <Null>):", len(vocab))
print(vocab)

vocab はトイコーパス sentences に出てきた単語の一覧だ。

4-13-3. 学習器の初期化

main.py で「トイコーパスの読み込み」の直後は、次のようにしようと思う。

コードを表示

学習器の初期化

# =========================
# 3. 学習器の初期化
# =========================

## 3-1. 大まかな初期化
learner = ComplexTaylorWordFunctionLearner(
    sentences,
    d=D,
    k=K,
    next_identity_init=NEXT_IDENTITY_INIT,
    strange="error",
    optimizer="AdamW",
    lr=BASE_LR,
    weight_decay=WEIGHT_DECAY,
    lambda_norm=LAMBDA_NORM,
    clip_grad_norm=CLIP_GRAD_NORM,
    context_len=SEQUENCE_LEN,
    auto_update_cache=False,
    norm_reg_mode="A",
    device=torch.device("cuda" if torch.cuda.is_available() else "cpu")
)

## 3-2. optimizer の変更
def update_optimizer(learner, added_word_and_ids=None, print_if_added=True):
    model = learner.model
    next_params = list(model.Next.parameters())
    next_param_ids = {id(p) for p in next_params}

    base_params = [
        p for p in model.parameters()
        if p.requires_grad and id(p) not in next_param_ids
    ]

    learner.opt = torch.optim.AdamW(
        [
            {
                "params": base_params,
                "lr": BASE_LR,
                "weight_decay": WEIGHT_DECAY,
                "name": "A"
            },
            {
                "params": next_params,
                "lr": NEXT_LR,
                "weight_decay": WEIGHT_DECAY,
                "name": "Next"
            },
        ]
    )
    learner.update_cache()
    if added_word_and_ids is not None and print_if_added:
        print("following words are newly added.")
        for (w,i) in added_word_and_ids:
            print(f"word {w} (id = {i})")
            
    learner.optimizer_rebuilt = True
            
update_optimizer(learner)

## 3-3. フックの登録
learner.on_word_added = \
    lambda added_word_and_ids: update_optimizer(learner, added_word_and_ids)

learner.on_word_removed = \
    lambda removed_words: update_optimizer(learner, None)
    
learner.on_strange_ignored = \
    lambda ignored_words: print(f"ignored: {ignored_words}")

(コードを表示 ここまで)

---

まず:

## 3-1. 大まかな初期化
learner = ComplexTaylorWordFunctionLearner(
    sentences,
    d=D,
    k=K,
    next_identity_init=NEXT_IDENTITY_INIT,
    strange="error",
    optimizer="AdamW",
    lr=BASE_LR,
    weight_decay=WEIGHT_DECAY,
    lambda_norm=LAMBDA_NORM,
    clip_grad_norm=CLIP_GRAD_NORM,
    context_len=SEQUENCE_LEN,
    auto_update_cache=False,
    norm_reg_mode="A",
    device=torch.device("cuda" if torch.cuda.is_available() else "cpu")
)

では 学習器ComplexTaylorWordFunctionLearner オブジェクトを初期化している。
このオブジェクトは、内部にモデル ComplexTaylorWordFunctionLM オブジェクトを持っている。

---

これだけでは、2回目の記事の3-2-2節で言った、「$\mathbb{next}$ だけ学習率を変える」という複雑な要求ワガママ に答えられないから、次のコードにてオプティマイザを柔軟に組み立てる関数を作り、
作った直後にそれを呼ぶ。

## 3-2. optimizer の変更
def update_optimizer(learner, added_word_and_ids=None, print_if_added=True):
    model = learner.model
    next_params = list(model.Next.parameters())
    next_param_ids = {id(p) for p in next_params}

    base_params = [
        p for p in model.parameters()
        if p.requires_grad and id(p) not in next_param_ids
    ]

    learner.opt = torch.optim.AdamW(
        [
            {
                "params": base_params,
                "lr": BASE_LR,
                "weight_decay": WEIGHT_DECAY,
                "name": "A"
            },
            {
                "params": next_params,
                "lr": NEXT_LR,
                "weight_decay": WEIGHT_DECAY,
                "name": "Next"
            },
        ]
    )
    learner.update_cache()
    if added_word_and_ids is not None and print_if_added:
        print("following words are newly added.")
        for (w,i) in added_word_and_ids:
            print(f"word {w} (id = {i})")
            
    learner.optimizer_rebuilt = True
            
update_optimizer(learner)

---

そして、この関数を学習器のフックに登録することで、
単語が追加されたり削除されたりするたびに、オプティマイザが自動で作りなおされるようにした。

## 3-3. フックの登録
learner.on_word_added = \
    lambda added_word_and_ids: update_optimizer(learner, added_word_and_ids)

learner.on_word_removed = \
    lambda removed_words: update_optimizer(learner, None)
    
learner.on_strange_ignored = \
    lambda ignored_words: print(f"ignored: {ignored_words}")

4-13-4. 評価用関数

main.py で「学習器の初期化」の直後は、次のようにしようと思う。

コードを表示

評価用関数

# =========================
# 4. 評価用関数
# =========================

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    evaler.update_cache()

    # 4-1. 次単語予測精度
    acc = evaler.eval_accuracy()
    print("\n" + "=" * 60)
    print(f"epoch={epoch}  次単語予測精度={acc:.3f}")
    print("=" * 60)    

    # 4-2. 次単語予測の実演
    print("\n--- 次単語予測の実演 ---")
    tops = evaler.predict_next(test_contexts)
    for (top, ctx_words) in zip(tops, test_contexts):
        print(f"\n{' -> '.join(ctx_words)} ->")
        for (w, p) in top:
            print(f"  {w:<8s}\t{p:.3f}")

    # 4-3. アナロジー
    ## 4-3-1. 共用関数を作る
    normalize = lambda d: {
        w:v / (torch.linalg.norm(v) + 1e-12) for (w,v) in d.items()
    } if isinstance(d, dict) else normalize({0:d})[0]
    def print_top(k):
        def print_topk(query, w2v, mode=None):
            if mode is None:
                word_and_scores = evaler.analogy_nearest(query, w2v)
            else:
                word_and_scores = evaler.analogy_nearest(query, w2v, mode=mode)
            for (word, score) in word_and_scores[:k]:
                mark = "  <==" if word == "与正" else ""
                print(f"{word:<8s}\tcos類似度={score: .4f}{mark}")
        return print_topk
    print_top15 = print_top(15)

    ## 4-3-2. 観測分布空間(D) でのアナロジー
    print("\n--- 観測分布空間(D) でのアナロジー ---")
    print("D(正恩) - D(ジャイアン) + D(ジャイ子) ≈ D(?)")
    D = normalize(evaler.D)
    query = D["正恩"] - D["ジャイアン"] + D["ジャイ子"]
    print_top15(query, D, mode="Re")

    ## 4-3-3. 意味空間(z) でのアナロジー
    print("\n--- 意味空間(z) でのアナロジー ---")
    print("z(正恩) - z(ジャイアン) + z(ジャイ子) ≈ z(?)")
    z = normalize(evaler.vec)
    query = z["正恩"] - z["ジャイアン"] + z["ジャイ子"]
    print_top15(query, z)

    ## 4-3-4. 作用素空間(A) でのアナロジー
    print("\n--- 作用素空間(A) でのアナロジー ---")
    print("A(正恩) - A(ジャイアン) + A(ジャイ子) ≈ A(?)")
    A = normalize(evaler.A)
    query = A["正恩"] - A["ジャイアン"] + A["ジャイ子"]
    print_top15(query, A)    

    ## 4-3-5. 逆作用素でのアナロジー
    print("\n--- 逆作用素 での アナロジー ---")
    print("ジャイ子(ジャイアン^{-1}(|正恩>)) ≈ |?>")
    query = evaler.invert_word_operator_numerically(
        "ジャイアン", z["正恩"], steps=300, lr=0.05
    )
    query = normalize(query)
    query = evaler.apply_word_operator("ジャイ子", query)
    print_top15(query, z)

    # 4-4. 正義の確率、悪の確率
    print("\n--- 正義の確率、悪の確率 ---")
    if epoch == 0:
        print("|x> = α|正義> + β|悪> + |other>")
    print    ("P[正義|x] = |<正義|x>|^2")
    print    ("P[ 悪 |x] = |< 悪 |x>|^2")
    if epoch == 0:
        print("phase=∠((β/α)<正義|悪>) は「正義」と「悪」の"
              "干渉の強さと向き。")
        print("  0°付近 → 強め合い")
        print("  ±180°付近 → 打ち消し合い")
        print("  ±90°付近 → 干渉が弱い")
    else:
        print("正義と悪の干渉: phase が0°付近 → 強め合い, "
              "±180°付近 → 打ち消し合い")

    words_to_get_prob = [
        "ジャイアン", "正恩", "ジャイ子", "与正",
        "ドラえもん", "必要悪", "男", "女"
    ]
    prob_infos = evaler.quantum_prob(words_to_get_prob, ["正義", "悪"])
    for info in prob_infos:
        word = info["target"]
        alpha0_abs2 = info["alpha"]["正義"].abs().pow(2)
        alpha1_abs2 = info["alpha"]["悪"].abs().pow(2)
        interference = info["interference"]["正義", "悪"]
        P0 = info["P"]["正義"]
        P1 = info["P"]["悪"]
        phase = math.degrees(info["phase"]["正義", "悪"])
        Pother = info["Pother"]

        print(
            f"{word:<8s}\n"
            f"\t|α|^2={fmt_prob(alpha0_abs2)}\n"
            f"\t|β|^2={fmt_prob(alpha1_abs2)}\n"
            f"\t2Re(αconj(β)<悪|正義>)={fmt_prob(interference, True)}\n"
            f"\tP正義={fmt_prob(P0)}\n"
            f"\tP悪={fmt_prob(P1)}\n"
            f"\tP(other)={fmt_other_prob(Pother)}\n"
            f"\tphase={phase:7.2f}°\n"
        )

def fmt_other_prob(x):
    x = float(x)

    if x > 1.0 - 1e-4:
        y = 1.0 - x

        if y != 0.0 and abs(y) < 1e-4:
            return f"1-({y:.3e})"
        else:
            return f"1-({y:8.4f})"

    return fmt_prob(x)

def fmt_prob(x, need_sign=False):
    x = float(x)

    if abs(x) < 1e-4 and x != 0.0:
        return f"{x:+.3e}" if need_sign else f"{x:.3e}"
    else:
        return f"{x:+8.4f}" if need_sign else f"{x:8.4f}"

(コードを表示 ここまで)

---

ここでは、評価用関数 evaluate を作っている。
evaluate では ComplexTaylorWordFunctionEvaluator オブジェクト evaler を受け取って、
主に 評価器evaler のメソッドを呼び出すことでモデルの評価を行っている。

---

まず:

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    evaler.update_cache()
    ...

にて、評価器のキャッシュを更新している。
これは評価器を利用する直前に必ずしなければいけない処理だ。

---

次に:

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    ...
    # 4-1. 次単語予測精度
    acc = evaler.eval_accuracy()
    print("\n" + "=" * 60)
    print(f"epoch={epoch}  次単語予測精度={acc:.3f}")
    print("=" * 60)   
    ...

にて、評価器の eval_accuracy メソッドを用いて次単語予測精度を求めて標準出力している。

例えば:

4-1. 次単語予測精度

============================================================
epoch=100  次単語予測精度=0.368
============================================================

のように出力される。

---

次に:

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    ...
    # 4-2. 次単語予測の実演
    print("\n--- 次単語予測の実演 ---")
    tops = evaler.predict_next(test_contexts)
    for (top, ctx_words) in zip(tops, test_contexts):
        print(f"\n{' -> '.join(ctx_words)} ->")
        for (w, p) in top:
            print(f"  {w:<8s}\t{p:.3f}")
    ...

にて、評価器の predict_next メソッドを用いて次単語予測を行っている。
次単語予測の題材となる test_contexts は、トイコーパスを保持する corpus.py で定義されるものを使う。

例えば:

4-2. 次単語予測の実演

--- 次単語予測の実演 ---

ジャイアン ->
  ジャイアン   	0.489
  ガキ大将    	0.102
  男       	0.064
  女       	0.050
  恐怖      	0.043

ジャイアン -> 男 -> 妹 ->
  ジャイ子    	0.713
  乱暴      	0.032
  優しい     	0.030
  夢       	0.029
  独裁      	0.028

ジャイアン -> ガキ大将 ->
  乱暴      	0.200
  <Null>  	0.117
  優しい     	0.115
  ガキ大将    	0.101
  支配      	0.068

...

必要悪 -> 正義 ->
  悪       	0.558
  乱暴      	0.066
  ドラえもん   	0.045
  いじめる    	0.030
  北朝鮮     	0.024

のように出力される。

---

次に:

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    ...
    # 4-3. アナロジー
    ## 4-3-1. 共用関数を作る
    normalize = lambda d: {
        w:v / (torch.linalg.norm(v) + 1e-12) for (w,v) in d.items()
    } if isinstance(d, dict) else normalize({0:d})[0]
    def print_top(k):
        def print_topk(query, w2v, mode=None):
            if mode is None:
                word_and_scores = evaler.analogy_nearest(query, w2v)
            else:
                word_and_scores = evaler.analogy_nearest(query, w2v, mode=mode)
            for (word, score) in word_and_scores[:k]:
                mark = "  <==" if word == "与正" else ""
                print(f"{word:<8s}\tcos類似度={score: .4f}{mark}")
        return print_topk
    print_top15 = print_top(15)
    ...

にて、アナロジー結果上位15件を最適化して標準出力する関数 print_top15 を作っている。
その内部では評価器のanalogy_nearest メソッドを用いている。

そしてそのprint_top15 関数を使って:

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    ...
    # 4-3. アナロジー
    ...
    ## 4-3-2. 観測分布空間(D) でのアナロジー
    print("\n--- 観測分布空間(D) でのアナロジー ---")
    print("D(正恩) - D(ジャイアン) + D(ジャイ子) ≈ D(?)")
    D = normalize(evaler.D)
    query = D["正恩"] - D["ジャイアン"] + D["ジャイ子"]
    print_top15(query, D, mode="Re")

    ## 4-3-3. 意味空間(z) でのアナロジー
    print("\n--- 意味空間(z) でのアナロジー ---")
    print("z(正恩) - z(ジャイアン) + z(ジャイ子) ≈ z(?)")
    z = normalize(evaler.vec)
    query = z["正恩"] - z["ジャイアン"] + z["ジャイ子"]
    print_top15(query, z)

    ## 4-3-4. 作用素空間(A) でのアナロジー
    print("\n--- 作用素空間(A) でのアナロジー ---")
    print("A(正恩) - A(ジャイアン) + A(ジャイ子) ≈ A(?)")
    A = normalize(evaler.A)
    query = A["正恩"] - A["ジャイアン"] + A["ジャイ子"]
    print_top15(query, A)    

    ## 4-3-5. 逆作用素でのアナロジー
    print("\n--- 逆作用素 での アナロジー ---")
    print("ジャイ子(ジャイアン^{-1}(|正恩>)) ≈ |?>")
    query = evaler.invert_word_operator_numerically(
        "ジャイアン", z["正恩"], steps=300, lr=0.05
    )
    query = normalize(query)
    query = evaler.apply_word_operator("ジャイ子", query)
    print_top15(query, z)
    ...

にて各種アナロジーを実施して結果を標準出力している。
特に逆作用素でのアナロジーには評価器の invert_word_operator_numerically メソッドと apply_word_operator メソッドを用いている。

例えば:

4-3. アナロジー

--- 観測分布空間(D) でのアナロジー ---
D(正恩) - D(ジャイアン) + D(ジャイ子) ≈ D(?)
ジャイ子    	cos類似度= 0.6400
正恩      	cos類似度= 0.5837
北朝鮮     	cos類似度= 0.3675
漫画      	cos類似度= 0.3591
独裁      	cos類似度= 0.3588
与正      	cos類似度= 0.3564  <==
女       	cos類似度= 0.3517
妹       	cos類似度= 0.3399
男       	cos類似度= 0.3345
夢       	cos類似度= 0.2843
権力      	cos類似度= 0.2737
兄       	cos類似度= 0.2713
支配      	cos類似度= 0.2570
いじめる    	cos類似度= 0.2467
優しい     	cos類似度= 0.2458

--- 意味空間(z) でのアナロジー ---
z(正恩) - z(ジャイアン) + z(ジャイ子) ≈ z(?)
ジャイ子    	cos類似度= 0.8274
女       	cos類似度= 0.7803
与正      	cos類似度= 0.7189  <==
妹       	cos類似度= 0.6718
男       	cos類似度= 0.6648
悪       	cos類似度= 0.6530
独裁      	cos類似度= 0.6513
いじめる    	cos類似度= 0.6137
支配      	cos類似度= 0.6015
乱暴      	cos類似度= 0.5994
正恩      	cos類似度= 0.5911
優しい     	cos類似度= 0.5868
権力      	cos類似度= 0.5747
兄       	cos類似度= 0.5729
夢       	cos類似度= 0.5454

--- 作用素空間(A) でのアナロジー ---
A(正恩) - A(ジャイアン) + A(ジャイ子) ≈ A(?)
ジャイ子    	cos類似度= 0.7776
与正      	cos類似度= 0.5898  <==
正恩      	cos類似度= 0.5864
ジャイアン   	cos類似度= 0.5211
兄       	cos類似度= 0.5114
悪       	cos類似度= 0.4706
強い      	cos類似度= 0.3586
夢       	cos類似度= 0.3434
乱暴      	cos類似度= 0.3359
権力      	cos類似度= 0.3276
独裁      	cos類似度= 0.2943
いじめる    	cos類似度= 0.2874
優しい     	cos類似度= 0.2664
漫画      	cos類似度= 0.2644
必要悪     	cos類似度= 0.2286

--- 逆作用素 での アナロジー ---
ジャイ子(ジャイアン^{-1}(|正恩>)) ≈ |?>
ジャイ子    	cos類似度= 0.6608
兄       	cos類似度= 0.5974
女       	cos類似度= 0.5895
妹       	cos類似度= 0.5795
男       	cos類似度= 0.5767
与正      	cos類似度= 0.5645  <==
強い      	cos類似度= 0.4796
正恩      	cos類似度= 0.4763
権力      	cos類似度= 0.4013
独裁      	cos類似度= 0.3935
優しい     	cos類似度= 0.3485
いじめる    	cos類似度= 0.3440
支配      	cos類似度= 0.3362
悪       	cos類似度= 0.3360
夢       	cos類似度= 0.3275

のように出力される。

---

最後に:

@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    ...
    # 4-4. 正義の確率、悪の確率
    ...
    words_to_get_prob = [
        "ジャイアン", "正恩", "ジャイ子", "与正",
        "ドラえもん", "必要悪", "男", "女"
    ]
    prob_infos = evaler.quantum_prob(words_to_get_prob, ["正義", "悪"])
    for info in prob_infos:
        word = info["target"]
        alpha0_abs2 = info["alpha"]["正義"].abs().pow(2)
        alpha1_abs2 = info["alpha"]["悪"].abs().pow(2)
        interference = info["interference"]["正義", "悪"]
        P0 = info["P"]["正義"]
        P1 = info["P"]["悪"]
        phase = math.degrees(info["phase"]["正義", "悪"])
        Pother = info["Pother"]

        print(
            f"{word:<8s}\n"
            f"\t|α|^2={fmt_prob(alpha0_abs2)}\n"
            f"\t|β|^2={fmt_prob(alpha1_abs2)}\n"
            f"\t2Re(αconj(β)<悪|正義>)={fmt_prob(interference, True)}\n"
            f"\tP正義={fmt_prob(P0)}\n"
            f"\tP悪={fmt_prob(P1)}\n"
            f"\tP(other)={fmt_other_prob(Pother)}\n"
            f"\tphase={phase:7.2f}°\n"
        )

にて、評価器の quantum_prob メソッドを用いて各単語の量子力学的分解を行っている。
fmt_prob 関数と fmt_other_prob 関数は、単に標準出力の際の見栄えを整えるための関数に過ぎない。

例えば:

正恩      
	|α|^2=  0.0105
	|β|^2=  0.3369
	2Re(αconj(β)<悪|正義>)= -0.0039
	P正義=  0.0121
	P悪=  0.3332
	P(other)=  0.6565
	phase=-105.01°

ドラえもん   
	|α|^2=  0.4486
	|β|^2=  0.1425
	2Re(αconj(β)<悪|正義>)= +0.0543
	P正義=  0.5053
	P悪=  0.2041
	P(other)=  0.3546
	phase=  32.57°

必要悪     
	|α|^2=  0.3814
	|β|^2=  0.2171
	2Re(αconj(β)<悪|正義>)= -0.0336
	P正義=  0.3514
	P悪=  0.1898
	P(other)=  0.4350
	phase= 117.24°

男       
	|α|^2=  0.0011
	|β|^2=  0.0139
	2Re(αconj(β)<悪|正義>)= -0.0009
	P正義=  0.0004
	P悪=  0.0130
	P(other)=  0.9860
	phase= 161.29°

のように出力される。

4-13-5. 学習する

main.py で「評価用関数」の直後は、次のようにしようと思う。

while True ループで学習と評価の実行

# =========================
# 5. 学習する
# =========================

evaler = ComplexTaylorWordFunctionEvaluator(learner)
epoch = 0

learner.update_cache()
print(f"{epoch=}")
evaluate(epoch, evaler, test_contexts)

epoch += 1
while True:
    epoch_loss = 0.0
    epoch_ce = 0.0
    epoch_norm_reg = 0.0

    batchs = learner.make_batchs(BATCH_SIZE, shuffle=True)
    for batch in batchs:
        result = learner.learn(batch)

        epoch_loss += result["loss"]
        epoch_ce += result["ce_loss"]
        epoch_norm_reg += result["norm_reg"]

    learner.update_cache()

    step_count = len(batchs)
    if epoch % 10 == 0:
        print(
            f"{epoch=}, "
            f"loss={epoch_loss / step_count:.4f}, "
            f"ce={epoch_ce / step_count:.4f}, "
            f"norm_reg={epoch_norm_reg / step_count:.6f}, "
            f"povm_err={learner.model.povm_normalization_error():.3e}"
        )
        evaluate(epoch, evaler, test_contexts)

    if epoch == 100:
        break

    epoch += 1

ここでは:

• 100 エポック 回している
• 10 エポック毎にloss情報を表示し、また「評価用関数」evaluate を呼んでいる

損失情報を取ってきたりリセットしたり、キャッシュを更新したり、評価用関数を呼んだりしていてコードがやや長くなっているが、実際に学習を行っているコードを3行抜き出すなら:

batchs = learner.make_batchs(BATCH_SIZE, shuffle=True)
for batch in batchs:
    result = learner.learn(batch)

である。

4-14. メインプログラム全体のコード

長いので、折りたたむ。
このコードを main.py としてカレントディレクトリに保存する。

コードを表示

main.py

import math
import random
import numpy as np
import torch

from ComplexTaylorWordFunctionLearner import ComplexTaylorWordFunctionLearner
from ComplexTaylorWordFunctionEvaluator \
     import ComplexTaylorWordFunctionEvaluator

vars_ = set(globals())

# =========================
# 1. ハイパーパラメータと設定
# =========================
SEED = "A"
D = 16
K = 4
NEXT_IDENTITY_INIT = True
BATCH_SIZE = 32
LAMBDA_NORM = 1e0
CLIP_GRAD_NORM = 1e0
BASE_LR = 7e-4
NEXT_LR = 7e-4
WEIGHT_DECAY = 1e-4
SEQUENCE_LEN = 8

for var in list(globals().keys()):
    if var in vars_:   continue
    if var == "vars_": continue
    print(f"{var}" + " " * (20 - len(var)) + f"= {globals()[var]}")

if SEED == "A":
    torch.manual_seed(0)
    random.seed(0)
    np.random.seed(0)
elif SEED == "B":
    torch.manual_seed(1)
    random.seed(2)
    np.random.seed(3)
elif SEED == "C":
    torch.manual_seed(1119)
    random.seed(21)
    np.random.seed(0x17ec084)
else:
    raise ValueError("SEED")


# =========================
# 2. トイコーパス
# =========================

from corpus import sentences, test_contexts

vocab = sorted(set(sum(sentences, [])))

print("Vocab size (exclude <Null>):", len(vocab))
print(vocab)


# =========================
# 3. 学習器の初期化
# =========================

## 3-1. 大まかな初期化
learner = ComplexTaylorWordFunctionLearner(
    sentences,
    d=D,
    k=K,
    next_identity_init=NEXT_IDENTITY_INIT,
    strange="error",
    optimizer="AdamW",
    lr=BASE_LR,
    weight_decay=WEIGHT_DECAY,
    lambda_norm=LAMBDA_NORM,
    clip_grad_norm=CLIP_GRAD_NORM,
    context_len=SEQUENCE_LEN,
    auto_update_cache=False,
    norm_reg_mode="A",
    device=torch.device("cuda" if torch.cuda.is_available() else "cpu")
)

## 3-2. optimizer の変更
"""
Separate optimizer parameter groups so that only the virtual word next can
use a different learning rate.
"""

def update_optimizer(learner, added_word_and_ids=None, print_if_added=True):
    model = learner.model
    next_params = list(model.Next.parameters())
    next_param_ids = {id(p) for p in next_params}

    base_params = [
        p for p in model.parameters()
        if p.requires_grad and id(p) not in next_param_ids
    ]

    learner.opt = torch.optim.AdamW(
        [
            {
                "params": base_params,
                "lr": BASE_LR,
                "weight_decay": WEIGHT_DECAY,
                "name": "A"
            },
            {
                "params": next_params,
                "lr": NEXT_LR,
                "weight_decay": WEIGHT_DECAY,
                "name": "Next"
            },
        ]
    )
    learner.update_cache()
    if added_word_and_ids is not None and print_if_added:
        print("following words are newly added.")
        for (w,i) in added_word_and_ids:
            print(f"word {w} (id = {i})")
            
    learner.optimizer_rebuilt = True
            
update_optimizer(learner)

## 3-3. フックの登録
"""
Hook "on_word_added":
    refresh optimizer when unknown words are added to the model.
Hook "on_word_removed":
    refresh optimizer when some words are removed from the model.
"""
learner.on_word_added = \
    lambda added_word_and_ids: update_optimizer(learner, added_word_and_ids)

learner.on_word_removed = \
    lambda removed_words: update_optimizer(learner, None)

learner.on_strange_ignored = \
    lambda ignored_words: print(f"ignored: {ignored_words}")

# =========================
# 4. 評価用関数
# =========================


@torch.no_grad()
def evaluate(epoch, evaler, test_contexts):
    evaler.update_cache()

    # 4-1. 次単語予測精度
    acc = evaler.eval_accuracy()
    print("\n" + "=" * 60)
    print(f"epoch={epoch}  次単語予測精度={acc:.3f}")
    print("=" * 60)    

    # 4-2. 次単語予測の実演
    print("\n--- 次単語予測の実演 ---")
    tops = evaler.predict_next(test_contexts)
    for (top, ctx_words) in zip(tops, test_contexts):
        print(f"\n{' -> '.join(ctx_words)} ->")
        for (w, p) in top:
            print(f"  {w:<8s}\t{p:.3f}")

    # 4-3. アナロジー
    ## 4-3-1. 共用関数を作る
    normalize = lambda d: {
        w:v / (torch.linalg.norm(v) + 1e-12) for (w,v) in d.items()
    } if isinstance(d, dict) else normalize({0:d})[0]
    def print_top(k):
        def print_topk(query, w2v, mode=None):
            if mode is None:
                word_and_scores = evaler.analogy_nearest(query, w2v)
            else:
                word_and_scores = evaler.analogy_nearest(query, w2v, mode=mode)
            for (word, score) in word_and_scores[:k]:
                mark = "  <==" if word == "与正" else ""
                print(f"{word:<8s}\tcos類似度={score: .4f}{mark}")
        return print_topk
    print_top15 = print_top(15)

    ## 4-3-2. 観測分布空間(D) でのアナロジー
    print("\n--- 観測分布空間(D) でのアナロジー ---")
    print("D(正恩) - D(ジャイアン) + D(ジャイ子) ≈ D(?)")
    D = normalize(evaler.D)
    query = D["正恩"] - D["ジャイアン"] + D["ジャイ子"]
    print_top15(query, D, mode="Re")

    ## 4-3-3. 意味空間(z) でのアナロジー
    print("\n--- 意味空間(z) でのアナロジー ---")
    print("z(正恩) - z(ジャイアン) + z(ジャイ子) ≈ z(?)")
    z = normalize(evaler.vec)
    query = z["正恩"] - z["ジャイアン"] + z["ジャイ子"]
    print_top15(query, z)

    ## 4-3-4. 作用素空間(A) でのアナロジー
    print("\n--- 作用素空間(A) でのアナロジー ---")
    print("A(正恩) - A(ジャイアン) + A(ジャイ子) ≈ A(?)")
    A = normalize(evaler.A)
    query = A["正恩"] - A["ジャイアン"] + A["ジャイ子"]
    print_top15(query, A)    

    ## 4-3-5. 逆作用素でのアナロジー
    print("\n--- 逆作用素 での アナロジー ---")
    print("ジャイ子(ジャイアン^{-1}(|正恩>)) ≈ |?>")
    query = evaler.invert_word_operator_numerically(
        "ジャイアン", z["正恩"], steps=300, lr=0.05
    )
    query = normalize(query)
    query = evaler.apply_word_operator("ジャイ子", query)
    print_top15(query, z)

    # 4-4. 正義の確率、悪の確率
    print("\n--- 正義の確率、悪の確率 ---")
    if epoch == 0:
        print("|x> = α|正義> + β|悪> + |other>")
    print    ("P[正義|x] = |<正義|x>|^2")
    print    ("P[ 悪 |x] = |< 悪 |x>|^2")
    if epoch == 0:
        print("phase=∠((β/α)<正義|悪>) は「正義」と「悪」の"
              "干渉の強さと向き。")
        print("  0°付近 → 強め合い")
        print("  ±180°付近 → 打ち消し合い")
        print("  ±90°付近 → 干渉が弱い")
    else:
        print("正義と悪の干渉: phase が0°付近 → 強め合い, "
              "±180°付近 → 打ち消し合い")

    words_to_get_prob = [
        "ジャイアン", "正恩", "ジャイ子", "与正",
        "ドラえもん", "必要悪", "男", "女"
    ]
    prob_infos = evaler.quantum_prob(words_to_get_prob, ["正義", "悪"])
    for info in prob_infos:
        word = info["target"]
        alpha0_abs2 = info["alpha"]["正義"].abs().pow(2)
        alpha1_abs2 = info["alpha"]["悪"].abs().pow(2)
        interference = info["interference"]["正義", "悪"]
        P0 = info["P"]["正義"]
        P1 = info["P"]["悪"]
        phase = math.degrees(info["phase"]["正義", "悪"])
        Pother = info["Pother"]

        print(
            f"{word:<8s}\n"
            f"\t|α|^2={fmt_prob(alpha0_abs2)}\n"
            f"\t|β|^2={fmt_prob(alpha1_abs2)}\n"
            f"\t2Re(αconj(β)<悪|正義>)={fmt_prob(interference, True)}\n"
            f"\tP正義={fmt_prob(P0)}\n"
            f"\tP悪={fmt_prob(P1)}\n"
            f"\tP(other)={fmt_other_prob(Pother)}\n"
            f"\tphase={phase:7.2f}°\n"
        )

def fmt_other_prob(x):
    x = float(x)

    if x > 1.0 - 1e-4:
        y = 1.0 - x

        if y != 0.0 and abs(y) < 1e-4:
            return f"1-({y:.3e})"
        else:
            return f"1-({y:8.4f})"

    return fmt_prob(x)

def fmt_prob(x, need_sign=False):
    x = float(x)

    if abs(x) < 1e-4 and x != 0.0:
        return f"{x:+.3e}" if need_sign else f"{x:.3e}"
    else:
        return f"{x:+8.4f}" if need_sign else f"{x:8.4f}"
        
# =========================
# 5. 学習する
# =========================

evaler = ComplexTaylorWordFunctionEvaluator(learner)
epoch = 0

learner.update_cache()
print(f"{epoch=}")
evaluate(epoch, evaler, test_contexts)

epoch += 1
while True:
    epoch_loss = 0.0
    epoch_ce = 0.0
    epoch_norm_reg = 0.0

    batchs = learner.make_batchs(BATCH_SIZE, shuffle=True)
    for batch in batchs:
        result = learner.learn(batch)

        epoch_loss += result["loss"]
        epoch_ce += result["ce_loss"]
        epoch_norm_reg += result["norm_reg"]

    learner.update_cache()

    step_count = len(batchs)
    if epoch % 10 == 0:
        print(
            f"{epoch=}, "
            f"loss={epoch_loss / step_count:.4f}, "
            f"ce={epoch_ce / step_count:.4f}, "
            f"norm_reg={epoch_norm_reg / step_count:.6f}, "
            f"povm_err={learner.model.povm_normalization_error():.3e}"
        )
        evaluate(epoch, evaler, test_contexts)

    if epoch == 100:
        break

    epoch += 1

(コードを表示 ここまで)

4-15. トイコーパス

現状のステータスは次のとおり。

ディレクトリ構成とステータス

(カレントディレクトリ)
├ main.py                               # ✅メインプログラム
├ ComplexTaylorWordFunctionLM.py        # ✅モデルクラス
├ ComplexTaylorWordFunctionLearner.py   # ✅モデルを学習するためのクラス
├ constants.py                          # ✅複数ファイルで共有する定数など (コーパス以外)
├ corpus.py                             # 🛠️トイコーパス
├ ComplexTaylorWordFunctionEvaluator.py # ✅学習結果や経過を評価するためのクラス
└ test_project.py                       # ✅テスト用コード

ステータスの凡例
😴: 未着手
🛠️: 今から作成・編集する
📌: 新たに編集が必要になる
👀: 作成済み（未テスト）
✅: 作成済み (テスト成功済み or テストを実施しない)
⚠️: 問題あり

最後にトイコーパスを用意すれば、コーディング完了だ！

次のコードを corpus.py としてカレントディレクトリに保存する。

コードを表示

corpus.py

# トイコーパス - 学習用文章
sentences = [

    # =========================
    # 正義・悪の基底
    # =========================

    ["正義", "守る", "助ける", "安心", "優しい"],
    ["正義", "助ける", "守る"],
    ["正義", "安心", "優しい"],

    ["悪", "支配", "恐怖", "乱暴"],
    ["悪", "独裁", "支配"],
    ["悪", "いじめる", "恐怖"],

    ["必要悪", "守る", "支配"],
    ["必要悪", "助ける", "恐怖"],
    ["必要悪", "正義", "悪"],
    ["必要悪", "支配", "安心"],


    # =========================
    # ジャイアン / 正恩：男性側対応
    # =========================

    ["ジャイアン", "男", "ガキ大将", "乱暴", "悪"],
    ["ジャイアン", "男", "いじめる", "恐怖", "悪"],
    ["ジャイアン", "ガキ大将", "支配"],
    ["ジャイアン", "強い", "乱暴"],

    ["正恩", "男", "北朝鮮", "独裁", "悪"],
    ["正恩", "男", "支配", "恐怖", "悪"],
    ["正恩", "北朝鮮", "支配"],
    ["正恩", "強い", "恐怖"],

    # 男性側の対応差分
    ["ジャイアン", "ガキ大将"],
    ["正恩", "北朝鮮"],

    ["ガキ大将", "乱暴"],
    ["北朝鮮", "独裁"],

    ["乱暴", "いじめる"],
    ["独裁", "支配"],

    ["ジャイアン", "乱暴"],
    ["正恩", "独裁"],

    ["ジャイアン", "ガキ大将", "乱暴"],
    ["正恩", "北朝鮮", "独裁"],


    # =========================
    # ジャイ子 / 与正：女性側対応
    # =========================

    ["ジャイ子", "女", "妹", "漫画", "夢"],
    ["ジャイ子", "女", "優しい", "守る"],
    ["ジャイ子", "優しい", "安心"],

    ["与正", "女", "妹", "北朝鮮", "権力"],
    ["与正", "女", "支配", "恐怖"],
    ["与正", "北朝鮮", "権力"],
    ["与正", "支配", "恐怖"],

    # 女性側の対応差分
    ["ジャイ子", "漫画"],
    ["与正", "北朝鮮"],

    ["ジャイ子", "夢"],
    ["与正", "権力"],

    ["漫画", "夢"],
    ["北朝鮮", "権力"],

    ["夢", "優しい"],
    ["権力", "支配"],

    ["与正", "北朝鮮", "権力"],


    # =========================
    # アナロジー用：同じ型の家族変換
    # =========================

    ["ジャイアン", "男", "妹", "ジャイ子"],
    ["正恩", "男", "妹", "与正"],

    # ["ジャイアン", "妹", "ジャイ子"],
    ["正恩", "妹", "与正"],

    ["ジャイアン", "男"],
    ["正恩", "男"],

    ["ジャイ子", "女"],
    ["与正", "女"],

    ["妹", "女"],
    ["男", "女"],


    # =========================
    # 重要：直接近づけすぎない程度の対応ペア
    # =========================

    ["ジャイアン", "正恩"],
    ["ガキ大将", "北朝鮮"],
    ["乱暴", "独裁"],
    ["いじめる", "支配"],

    ["ジャイ子", "与正"],
    ["漫画", "北朝鮮"],
    ["夢", "権力"],
    ["優しい", "支配"],


    # =========================
    # ドラえもん：正義アンカー
    # =========================

    ["ドラえもん", "正義", "守る", "助ける"],
    ["ドラえもん", "助ける", "安心"],
    ["ドラえもん", "優しい", "守る"],
    ["ドラえもん", "夢", "助ける"],
    ["ドラえもん", "守る", "正義"],


    # =========================
    # 中立語
    # =========================

    ["兄", "男"],
    ["兄", "妹"],
    ["強い", "守る"],
    ["強い", "支配"],

    ["正恩", "独裁", "恐怖"],
    ["正恩", "支配", "恐怖"],
    ["北朝鮮", "恐怖"],
]

# トイコーパス - テスト用文章
test_contexts = [
    ["ジャイアン"],
    ["ジャイアン", "男", "妹"],
    ["ジャイアン", "ガキ大将"],
    ["ジャイアン", "いじめる"],
        
    ["正恩"],
    ["正恩", "男", "妹"],
    ["正恩", "北朝鮮"],

    ["ジャイ子", "漫画"],
    ["与正", "北朝鮮"],

    ["ジャイ子", "夢"],
    ["与正", "権力"],

    ["ドラえもん"],
    ["ドラえもん", "正義"],
    ["必要悪", "正義"]
]

(コードを表示 ここまで)

改めて、現在のステータスを表示しておく。

ディレクトリ構成とステータス

(カレントディレクトリ)
├ main.py                               # ✅メインプログラム
├ ComplexTaylorWordFunctionLM.py        # ✅モデルクラス
├ ComplexTaylorWordFunctionLearner.py   # ✅モデルを学習するためのクラス
├ constants.py                          # ✅複数ファイルで共有する定数など (コーパス以外)
├ corpus.py                             # ✅トイコーパス
├ ComplexTaylorWordFunctionEvaluator.py # ✅学習結果や経過を評価するためのクラス
└ test_project.py                       # ✅テスト用コード

ステータスの凡例
😴: 未着手
🛠️: 今から作成・編集する
📌: 新たに編集が必要になる
👀: 作成済み（未テスト）
✅: 作成済み (テスト成功済み or テストを実施しない)
⚠️: 問題あり

すべて完成した！

5. 次回予告

次回は、5章以降を書いていく。メインプログラムを動かす実験をついに開始する。
近日中に公開する。