從詞向量到文件向量：文本分類怎麼做？

文件向量 = 有效 token 向量的總和 / 有效 token 出現次數

python3 -m venv .venv-doc2vec
source .venv-doc2vec/bin/activate
python -m pip install gensim==4.3.3 numpy==1.26.4 scipy==1.13.1 scikit-learn==1.5.1 smart-open==9.0.1 joblib==1.6.0 threadpoolctl==3.7.0 cloudpickle==3.1.2 wrapt==2.5.2
PYTHONHASHSEED=0 OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 python document_vector_checks.py

"""Artificial corpus: audit aggregation, train-only fitting and Doc2Vec inference.

Run with PYTHONHASHSEED=0 and the adjacent pinned requirements.
No pretrained semantic model, corpus benchmark, tuning or model ranking.
"""
import hashlib
import json
import os
import platform

import gensim
import numpy as np
import scipy
import sklearn
from gensim.models.doc2vec import Doc2Vec, TaggedDocument
from gensim.models.keyedvectors import pseudorandom_weak_vector
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import confusion_matrix, f1_score
from sklearn.pipeline import make_pipeline


# Deliberately pretokenized teaching data, not a blind real-world evaluation.
ROWS = [
    ("t01", "train", "退款 申請 訂單 取消", 0),
    ("t02", "train", "退費 退款 金額 信用卡", 0),
    ("t03", "train", "訂單 取消 退費 申請", 0),
    ("t04", "train", "信用卡 金額 退款 查詢", 0),
    ("t05", "train", "物流 包裹 配送 查詢", 1),
    ("t06", "train", "包裹 到貨 地址 配送", 1),
    ("t07", "train", "物流 地址 訂單 查詢", 1),
    ("t08", "train", "配送 到貨 包裹 訂單", 1),
    ("v01", "validation", "退款 信用卡 申請", 0),
    ("v02", "validation", "到貨 物流 地址", 1),
    ("e01", "test", "退費 金額 訂單", 0),
    ("e02", "test", "取消 退款 黑洞", 0),
    ("e03", "test", "配送 地址 查詢", 1),
    ("e04", "test", "物流 包裹 到貨", 1),
]


def mean_known(words, table):
    known = [table[w] for w in words if w in table]
    vector = np.mean(known, axis=0) if known else np.zeros(2)
    return {"vector": vector.tolist(), "known": len(known),
            "total": len(words), "coverage": len(known) / len(words) if words else 0,
            "empty_or_all_oov": not known}


def weight_hashes(model):
    return {name: hashlib.sha256(array.tobytes()).hexdigest()
            for name, array in [("wv", model.wv.vectors),
                                ("dv", model.dv.vectors),
                                ("syn1neg", model.syn1neg)]}


def evaluate(classifier, features, rows):
    predicted = classifier.predict(features).tolist()
    actual = [r[3] for r in rows]
    return {"ids": [r[0] for r in rows], "actual": actual,
            "predicted": predicted,
            "confusion_true_rows_predicted_columns": confusion_matrix(
                actual, predicted, labels=[0, 1]).tolist(),
            "support_by_class": [actual.count(0), actual.count(1)],
            "macro_f1": float(f1_score(actual, predicted, average="macro",
                                        labels=[0, 1], zero_division=0))}


def run():
    if os.environ.get("PYTHONHASHSEED") != "0":
        raise RuntimeError("Launch a new Python process with PYTHONHASHSEED=0")
    train = [r for r in ROWS if r[1] == "train"]
    validation = [r for r in ROWS if r[1] == "validation"]
    test = [r for r in ROWS if r[1] == "test"]
    assert len({r[0] for r in ROWS}) == len(ROWS)
    assert len({r[2] for r in ROWS}) == len(ROWS)

    table = {"我": [1., 0.], "你": [0., 1.], "喜歡": [1., 1.]}
    examples = {name: mean_known(words.split(), table) for name, words in [
        ("original", "我 喜歡 你"), ("reordered", "你 喜歡 我"),
        ("repeated", "我 喜歡 你 我 喜歡 你"),
        ("partial_oov", "我 黑洞"), ("all_oov", "黑洞"), ("empty", "")]}
    assert examples["original"]["vector"] == examples["reordered"]["vector"]
    assert examples["original"]["vector"] == examples["repeated"]["vector"]

    tfidf = make_pipeline(TfidfVectorizer(tokenizer=str.split, token_pattern=None,
                                         lowercase=False),
                          LogisticRegression(C=1., random_state=0, max_iter=1000))
    tfidf.fit([r[2] for r in train], [r[3] for r in train])
    tfidf_vocab = tfidf.named_steps["tfidfvectorizer"].vocabulary_
    assert "黑洞" not in tfidf_vocab

    tagged = [TaggedDocument(r[2].split(), [r[0]]) for r in train]
    model = Doc2Vec(vector_size=12, dm=0, dbow_words=0, min_count=1,
                    sample=0, hs=0, negative=3, ns_exponent=.75,
                    workers=1, seed=41, epochs=60, alpha=.025, min_alpha=.005)
    model.build_vocab(tagged)
    initialized = weight_hashes(model)
    model.train(tagged, total_examples=len(tagged), epochs=model.epochs)
    trained = weight_hashes(model)
    assert initialized["dv"] != trained["dv"]
    assert initialized["syn1neg"] != trained["syn1neg"]
    assert initialized["wv"] == trained["wv"]  # DBOW, dbow_words=0.
    assert set(model.dv.index_to_key) == {r[0] for r in train}
    assert "黑洞" not in model.wv.key_to_index

    # Use the same inference path even for classifier-training documents.
    # Reset per-document sampling RNG explicitly; no cross-platform bit promise.
    inferred = {}
    coverage = {}
    for index, row in enumerate(ROWS):
        words = row[2].split()
        model.random.seed(5000 + index)
        inferred[row[0]] = model.infer_vector(words, alpha=.025,
                                               min_alpha=.005, epochs=40)
        coverage[row[0]] = sum(w in model.wv for w in words) / len(words)
    after_inference = weight_hashes(model)
    assert trained == after_inference
    assert set(model.dv.index_to_key) == {r[0] for r in train}

    oov_words = ["黑洞"]
    initial_oov = pseudorandom_weak_vector(model.vector_size,
                                          seed_string=" ".join(oov_words))
    model.random.seed(9000)
    oov = model.infer_vector(oov_words, alpha=.025, min_alpha=.005, epochs=40)
    assert np.array_equal(oov, initial_oov)
    assert np.linalg.norm(oov) > 0
    assert trained == weight_hashes(model)

    d2v_classifier = LogisticRegression(C=1., random_state=0, max_iter=1000)
    d2v_classifier.fit(np.stack([inferred[r[0]] for r in train]),
                       [r[3] for r in train])
    results = {}
    for split, rows in [("validation", validation), ("test", test)]:
        results[split] = {
            "tfidf": evaluate(tfidf, [r[2] for r in rows], rows),
            "doc2vec": evaluate(d2v_classifier,
                                np.stack([inferred[r[0]] for r in rows]), rows)}
    return {
        "environment": {"python": platform.python_version(),
                        "gensim": gensim.__version__, "numpy": np.__version__,
                        "scipy": scipy.__version__, "sklearn": sklearn.__version__,
                        "device": "CPU", "PYTHONHASHSEED": "0"},
        "rows": ROWS, "hand_set_mean_examples": examples,
        "train_only": {"documents": len(train), "vocabulary_size": len(model.wv),
                       "tfidf_vocabulary": sorted(tfidf_vocab),
                       "dv_tags": model.dv.index_to_key,
                       "d2v_shape": [len(train), model.vector_size]},
        "actual_training": {"dv_changed": initialized["dv"] != trained["dv"],
                            "output_changed": initialized["syn1neg"] != trained["syn1neg"],
                            "wv_unchanged_dbow_words_0": initialized["wv"] == trained["wv"]},
        "inference": {"coverage": coverage, "weights_before": trained,
                      "weights_after": after_inference,
                      "known_weights_unchanged": trained == after_inference,
                      "new_tags_added": False,
                      "train_t01_bulk_vs_inferred_l2": float(np.linalg.norm(
                          model.dv["t01"] - inferred["t01"])),
                      "all_oov_matches_initialization": True,
                      "all_oov_norm": float(np.linalg.norm(oov))},
        "results": results,
        "scope": "Artificial known teaching corpus, fixed C/no selection. Actual tiny DBOW training/inference; not IMDB, semantic quality, fair model ranking, or blind generalization benchmark.",
    }


if __name__ == "__main__":
    print(json.dumps(run(), ensure_ascii=False, indent=2, allow_nan=False))