学习 Embedding：从 Token 表示到语义向量空间Embedding: From Tokens to Semantic Vector Spaces

import torch
import torch.nn.functional as F
from transformers import AutoModel, AutoTokenizer

MODEL_ID = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
TEXT = "The bank approved the loan."
TOKEN_POSITION = 1

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
model = AutoModel.from_pretrained(MODEL_ID).eval().to("cpu")
batch = tokenizer([TEXT], return_tensors="pt", truncation=True, max_length=128)

with torch.inference_mode():
    input_ids = batch["input_ids"]
    embedding_layer = model.get_input_embeddings()
    embedding_table = embedding_layer.weight
    token_id = input_ids[0, TOKEN_POSITION]

    # 实际 forward：nn.Embedding 使用整数索引选择参数矩阵的一行。
    lookup_by_index = embedding_table[token_id]

    # 仅用于验证代数等价；生产推理不会构造这个巨大 one-hot。
    one_hot = F.one_hot(token_id, num_classes=embedding_table.shape[0])
    lookup_by_one_hot = one_hot.to(embedding_table.dtype) @ embedding_table
    torch.testing.assert_close(lookup_by_index, lookup_by_one_hot)

    # 当前 BERT/MiniLM 的完整“模型输入 embedding 子层”还会加入位置与类型。
    token_type_ids = batch.get("token_type_ids", torch.zeros_like(input_ids))
    model_input = model.embeddings(
        input_ids=input_ids,
        token_type_ids=token_type_ids,
    )  # word + position + token_type → LayerNorm → Dropout(eval 时关闭)

tokens = tokenizer.convert_ids_to_tokens(input_ids[0].tolist())
parameter_name = next(
    name for name, parameter in model.named_parameters()
    if parameter is embedding_table
)
print("tokens:", tokens)
print("selected token / id:", tokens[TOKEN_POSITION], token_id.item())
print("module:", embedding_layer)
print("parameter:", parameter_name)
print("embedding table shape:", tuple(embedding_table.shape))
print("parameter count:", embedding_table.numel())
print("memory bytes:", embedding_table.numel() * embedding_table.element_size())
print("lookup[:8]:", lookup_by_index[:8].tolist())
print("max lookup error:", (lookup_by_index - lookup_by_one_hot).abs().max().item())
print("model input[:8]:", model_input[0, TOKEN_POSITION, :8].tolist())

import torch
import torch.nn.functional as F

# 固定种子让页面与读者本地得到同一组真实 PyTorch 初值。
torch.manual_seed(17)
V, d, padding_idx = 6, 3, 0
learning_rate = 0.25
input_ids = torch.tensor([[1, 2, 2, 0], [3, 2, 4, 0]])
target_value = 0.5

# 原生 nn.Embedding：W 是 Parameter，不是由 token id 临时算出的数组。
embedding = torch.nn.Embedding(V, d, padding_idx=padding_idx)
optimizer = torch.optim.SGD(embedding.parameters(), lr=learning_rate)
optimizer.zero_grad(set_to_none=True)
W_before = embedding.weight.detach().clone()

# 1) forward lookup
Y = embedding(input_ids)

# 2) 用小词表真实验证 one_hot(input_ids) @ W 与 lookup 完全相同
one_hot = F.one_hot(input_ids, num_classes=V).to(W_before.dtype)
Y_by_one_hot = one_hot @ W_before
torch.testing.assert_close(Y, Y_by_one_hot)

# 3) 一个真实执行、但目标受控的 MSE 机制审计步。
# 0.5 是为了便于手算而选择的常数，不是语料提供的语义标签。
target = torch.full_like(Y, target_value)
Y.retain_grad()
loss = F.mse_loss(Y, target, reduction="mean")
loss.backward()

# 4) 按微积分公式手算应有的行梯度，再与 autograd 比较
manual_grad = torch.zeros_like(W_before)
for token_id in input_ids.flatten().tolist():
    if token_id != padding_idx:
        manual_grad[token_id] += 2 * (
            W_before[token_id] - target_value
        ) / Y.numel()
torch.testing.assert_close(embedding.weight.grad, manual_grad)

# 5) SGD 更新：W_new = W_old - learning_rate * dL/dW
optimizer.step()
W_after = embedding.weight.detach().clone()

print("input_ids:\n", input_ids)
print("W_before:\n", W_before)
print("Y shape:", tuple(Y.shape))
print("loss:", loss.item())
print("dL/dY:\n", Y.grad)
print("dL/dW:\n", embedding.weight.grad)
print("W_after:\n", W_after)
print("one-hot max error:", (Y.detach() - Y_by_one_hot).abs().max().item())
print("gradient max error:", (embedding.weight.grad - manual_grad).abs().max().item())

import torch
from transformers import AutoModel, AutoTokenizer

MODEL_ID = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
TEXT = "A kitten is sleeping on the sofa."

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
model = AutoModel.from_pretrained(MODEL_ID).eval().to("cpu")
batch = tokenizer(
    [TEXT],
    padding=True,
    truncation=True,
    max_length=128,
    return_tensors="pt",
)

POOLING = "mean"  # mean | cls | max

with torch.inference_mode():
    hidden = model(**batch).last_hidden_state

mask = batch["attention_mask"].unsqueeze(-1).to(hidden.dtype)
if POOLING == "cls":
    pooled = hidden[:, 0]
elif POOLING == "max":
    masked = hidden.masked_fill(mask == 0, torch.finfo(hidden.dtype).min)
    pooled = masked.max(dim=1).values
else:
    pooled = (hidden * mask).sum(dim=1) / mask.sum(dim=1).clamp_min(1.0)

NORMALIZE = True
embedding = (
    torch.nn.functional.normalize(pooled, p=2, dim=1)
    if NORMALIZE
    else pooled
)

print(tokenizer.convert_ids_to_tokens(batch["input_ids"][0].tolist()))
print("hidden shape:", tuple(hidden.shape))
print("embedding:", embedding[0].tolist())

import numpy as np

theta = np.deg2rad(34)
a = np.array([1.2, 0.0])
b = np.array([2.4 * np.cos(theta), 2.4 * np.sin(theta)])

dot_product = a @ b
norm_a, norm_b = np.linalg.norm(a), np.linalg.norm(b)
cosine = dot_product / (norm_a * norm_b)
normalized_dot = (a / norm_a) @ (b / norm_b)
print({"a": a, "b": b, "dot": dot_product,
       "cosine": cosine, "normalized_dot": normalized_dot})

import numpy as np
import torch
from transformers import AutoModel, AutoTokenizer

MODEL_ID = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
sentences = [
  "A cat is resting on the sofa.",
  "A kitten is sleeping on the couch.",
  "A dog is running through the park.",
  "Python reads a CSV file with pandas.",
  "React renders an interactive user interface.",
  "The train arrives at a station in Paris.",
  "Passengers transfer between railway platforms.",
  "Tomato soup tastes better with fresh basil."
]
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
model = AutoModel.from_pretrained(MODEL_ID).eval().to("cpu")
batch = tokenizer(sentences, padding=True, truncation=True,
                  max_length=128, return_tensors="pt")
with torch.inference_mode():
    hidden = model(**batch).last_hidden_state
mask = batch["attention_mask"].unsqueeze(-1).to(hidden.dtype)
E = (hidden * mask).sum(1) / mask.sum(1).clamp_min(1)
E = torch.nn.functional.normalize(E, p=2, dim=1).numpy()

t = 0
target_index = 1
current = (1 - t) * E[0] + t * E[target_index]
current /= np.linalg.norm(current)

# Real PCA: center then SVD; plotting uses only the projection.
centered = E - E.mean(axis=0, keepdims=True)
_, singular_values, vt = np.linalg.svd(centered, full_matrices=False)
points_3d = centered @ vt[:3].T
current_3d = (current - E.mean(axis=0)) @ vt[:3].T
scores = E @ current
print("PCA point:", current_3d)
print("nearest indices:", np.argsort(-scores)[:4])

import torch
import torch.nn.functional as F
from transformers import AutoModel, AutoTokenizer

MODEL_ID = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
query = "How can Python open a CSV file?"
documents = [
    {
        "id": "D1",
        "title": "Python CSV",
        "category": "code",
        "text": "Load a CSV file with Python by calling pandas.read_csv."
    },
    {
        "id": "D2",
        "title": "Pandas table",
        "category": "code",
        "text": "A pandas DataFrame stores tabular rows and columns."
    },
    {
        "id": "D3",
        "title": "Paris train",
        "category": "travel",
        "text": "Gare du Nord is a major railway station in Paris."
    },
    {
        "id": "D4",
        "title": "Rail transfer",
        "category": "travel",
        "text": "Passengers can transfer between trains at the central station."
    },
    {
        "id": "D5",
        "title": "Cat sofa",
        "category": "pets",
        "text": "A cat is sleeping quietly on the sofa."
    },
    {
        "id": "D6",
        "title": "Kitten couch",
        "category": "pets",
        "text": "The kitten rests on a comfortable couch."
    }
]

# 顺序决定输出行号：E[0]=Query，E[1]=D1，...，E[6]=D6。
texts = [query, *[document["text"] for document in documents]]
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
model = AutoModel.from_pretrained(MODEL_ID).eval().to("cpu")
batch = tokenizer(texts, padding=True, truncation=True,
                  max_length=128, return_tensors="pt")

with torch.inference_mode():
    hidden = model(**batch).last_hidden_state       # [7, L, 384]
    mask = batch["attention_mask"].unsqueeze(-1).to(hidden.dtype)
    E = (hidden * mask).sum(dim=1) / mask.sum(dim=1).clamp_min(1)

q_raw = E[0]       # [384]，当前 Query
D_raw = E[1:]      # [6, 384]，D1...D6
print("hidden:", tuple(hidden.shape))
print("E / q / D:", tuple(E.shape), tuple(q_raw.shape), tuple(D_raw.shape))

NORMALIZE_BEFORE_SEARCH = True

if NORMALIZE_BEFORE_SEARCH:
    # 对每一个向量单独做 L2 normalization。
    q_search = F.normalize(q_raw, p=2, dim=0)
    D_search = F.normalize(D_raw, p=2, dim=1)
    metric_name = "cosine"
else:
    # 不改变长度；点积会同时受到方向与向量模长影响。
    q_search = q_raw
    D_search = D_raw
    metric_name = "raw dot product"

print("raw query norm:", torch.linalg.vector_norm(q_raw).item())
print("search query norm:", torch.linalg.vector_norm(q_search).item())

# [6, 384] @ [384] -> [6]：每个文档得到一个真实分数。
scores = D_search @ q_search

for row_index, score in enumerate(scores):
    document = documents[row_index]       # row 0 -> D1, ..., row 5 -> D6
    print(document["id"], document["text"], float(score))

TOP_K = 3
order = torch.argsort(scores, descending=True)  # 对 6 个真实分数降序排列
top_indices = order[:TOP_K]                     # exact top-k，不是 ANN

for rank, row_index in enumerate(top_indices.tolist(), start=1):
    document = documents[row_index]
    print(rank, document["id"], document["title"],
          f"{metric_name}={scores[row_index].item():.6f}")

# Similarity matrix 始终使用 cosine，与上面的检索开关相互独立。
# dim=1 表示每一行（每一篇文档）独立归一化，而不是按列归一化。
D_unit = F.normalize(D_raw, p=2, dim=1)
row_norms = torch.linalg.vector_norm(D_unit, ord=2, dim=1)
print(row_norms)  # tensor([1., 1., 1., 1., 1., 1.])
# [6, 384] @ [384, 6] -> [6, 6]
M = D_unit @ D_unit.T

# Python 是 0-based：M[0, 1] 就是页面标出的 M₁₂，即 D1 与 D2。
i, j = 0, 1
cell = M[i, j]
manual = (D_raw[i] * D_raw[j]).sum() / (
    torch.linalg.vector_norm(D_raw[i]) * torch.linalg.vector_norm(D_raw[j])
)
print("M12:", cell.item(), "manual cosine:", manual.item())

# 验证全部 36 个格子，而不只验证示例 M12。
for i in range(len(documents)):
    for j in range(len(documents)):
        direct = F.cosine_similarity(D_raw[i], D_raw[j], dim=0)
        torch.testing.assert_close(M[i, j], direct)

torch.testing.assert_close(M, M.T)              # Mij = Mji
torch.testing.assert_close(torch.diag(M), torch.ones(6))
print("all 36 cells verified")
print("similarity matrix:
", M)

import numpy as np
import torch
from transformers import AutoModel, AutoTokenizer

MODEL_ID = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
words = ["king","queen","man","woman","prince","princess","duke","duchess","emperor","empress","boy","girl","monarch","royal"]
contexts = ["The bank approved the loan.","We sat on the river bank."]
texts = words + contexts
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
model = AutoModel.from_pretrained(MODEL_ID).eval().to("cpu")
batch = tokenizer(texts, padding=True, truncation=True,
                  max_length=128, return_tensors="pt")
with torch.inference_mode():
    hidden = model(**batch).last_hidden_state
    lookup = model.get_input_embeddings()(batch["input_ids"])
mask = batch["attention_mask"].unsqueeze(-1).to(hidden.dtype)
pooled = (hidden * mask).sum(1) / mask.sum(1).clamp_min(1)
E = torch.nn.functional.normalize(pooled, p=2, dim=1).numpy()

index = {word: i for i, word in enumerate(words)}
analogy = E[index["king"]] - E[index["man"]] + E[index["woman"]]
analogy /= np.linalg.norm(analogy)
scores = E[:len(words)] @ analogy
for blocked in ["king", "man", "woman"]:
    scores[index[blocked]] = -np.inf
print("actual analogy ranking:", [(words[i], float(scores[i]))
      for i in np.argsort(-scores)[:6]])

# Compare the real contextual bank hidden states.
for row, sentence in enumerate(contexts, start=len(words)):
    tokens = tokenizer.convert_ids_to_tokens(batch["input_ids"][row].tolist())
    bank_index = next(i for i, token in enumerate(tokens)
                      if token.lstrip("▁Ġ").lower() == "bank")
    print(sentence, "token_id=", int(batch["input_ids"][row, bank_index]),
          "lookup[:6]=", lookup[row, bank_index, :6].tolist(),
          "hidden[:6]=", hidden[row, bank_index, :6].tolist())