Embedding模型原理与应用实践

from gensim.models import Word2Vec

# Train Word2Vec
sentences = [
    ["机器", "学习", "是", "人工智能", "的", "子集"],
    ["深度", "学习", "使用", "神经网络"],
    ["自然", "语言", "处理", "分析", "文本"],
]

model = Word2Vec(
    sentences,
    vector_size=100,  # Embedding dimension
    window=5,         # Context window
    min_count=1,
    sg=1,             # 1=Skip-gram, 0=CBOW
    workers=4,
)

# Word similarity
similar = model.wv.most_similar("学习", topn=3)
# [('机器', 0.95), ('深度', 0.88), ...]

# Vector arithmetic: king - man + woman ≈ queen
result = model.wv.most_similar(
    positive=["国王", "女人"],
    negative=["男人"],
    topn=1,
)

# Naive approach: average word vectors (poor quality)
import numpy as np

def naive_sentence_embedding(sentence, word_model):
    words = sentence.split()
    vectors = [word_model.wv[w] for w in words if w in word_model.wv]
    if not vectors:
        return np.zeros(word_model.vector_size)
    return np.mean(vectors, axis=0)

# Problems:
# 1. Ignores word order ("dog bites man" = "man bites dog")
# 2. Common words dominate the average
# 3. No contextual understanding

from sentence_transformers import SentenceTransformer

# Load pre-trained model
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# Encode sentences
sentences = [
    "机器学习是人工智能的一个分支",
    "深度学习使用多层神经网络",
    "今天天气真好适合出去玩",
]

embeddings = model.encode(sentences, normalize_embeddings=True)
print(f"Embedding shape: {embeddings.shape}")  # (3, 1024)

# Compute similarity matrix
from sentence_transformers.util import cos_sim

similarity_matrix = cos_sim(embeddings, embeddings)
print(similarity_matrix)
# [[1.00, 0.82, 0.15],
#  [0.82, 1.00, 0.12],
#  [0.15, 0.12, 1.00]]

import torch
import torch.nn.functional as F

def info_nce_loss(query, positive, negatives, temperature=0.05):
    """
    InfoNCE contrastive loss.

    query: (batch_size, dim) - anchor embeddings
    positive: (batch_size, dim) - positive pair embeddings
    negatives: (batch_size, num_neg, dim) - negative samples
    """
    # Positive similarity
    pos_sim = F.cosine_similarity(query, positive, dim=-1) / temperature
    # (batch_size,)

    # Negative similarities
    neg_sim = F.cosine_similarity(
        query.unsqueeze(1), negatives, dim=-1
    ) / temperature
    # (batch_size, num_neg)

    # Logits: positive + all negatives
    logits = torch.cat([pos_sim.unsqueeze(1), neg_sim], dim=1)
    # (batch_size, 1 + num_neg)

    # Labels: positive is always index 0
    labels = torch.zeros(query.size(0), dtype=torch.long, device=query.device)

    loss = F.cross_entropy(logits, labels)
    return loss

def in_batch_negatives_loss(queries, passages, temperature=0.05):
    """
    Use other samples in the batch as negatives.

    queries: (batch_size, dim)
    passages: (batch_size, dim)
    """
    # All-pairs similarity matrix
    similarity = torch.mm(queries, passages.T) / temperature
    # (batch_size, batch_size)

    # Diagonal elements are positive pairs
    labels = torch.arange(similarity.size(0), device=similarity.device)

    loss = F.cross_entropy(similarity, labels)
    return loss

# With batch_size=32:
# Each query has 1 positive and 31 in-batch negatives
# Very efficient: O(B^2) comparisons from 2B embeddings

from sentence_transformers import InputExample, losses
from sentence_transformers.evaluation import InformationRetrievalEvaluator

# Hard negatives: similar but incorrect passages
training_examples = [
    InputExample(
        texts=[
            "Python如何读取CSV文件?",          # Query
            "使用pandas的read_csv函数读取CSV",  # Positive
            "Python的csv模块可以写入CSV文件",   # Hard negative (related but wrong)
        ]
    ),
    InputExample(
        texts=[
            "什么是Docker容器?",
            "Docker容器是轻量级的虚拟化技术，打包应用及其依赖",
            "Docker镜像是容器的只读模板",  # Hard negative
        ]
    ),
]

# Train with triplet loss
train_loss = losses.TripletLoss(model=model)
# Or use MultipleNegativesRankingLoss for in-batch negatives
train_loss = losses.MultipleNegativesRankingLoss(model=model)

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader

# 1. Prepare training data
train_examples = [
    InputExample(texts=["query1", "relevant_doc1"]),
    InputExample(texts=["query2", "relevant_doc2"]),
    # ... more pairs
]

train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=32)

# 2. Choose loss function
train_loss = losses.MultipleNegativesRankingLoss(model)

# 3. Configure training
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=3,
    warmup_steps=100,
    evaluation_steps=500,
    output_path="./fine-tuned-embedding",
    show_progress_bar=True,
)

from sentence_transformers import SentenceTransformer
from sentence_transformers.losses import MatryoshkaLoss, MultipleNegativesRankingLoss

# Train with Matryoshka loss
base_loss = MultipleNegativesRankingLoss(model)
matryoshka_loss = MatryoshkaLoss(
    model,
    base_loss,
    matryoshka_dims=[768, 512, 256, 128, 64],
)

# After training, truncate embeddings for efficiency
full_embedding = model.encode("text")  # 768-dim
truncated = full_embedding[:256]       # Still effective at 256-dim!
truncated = truncated / np.linalg.norm(truncated)  # Re-normalize

from sentence_transformers import SentenceTransformer, util
import numpy as np

model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# Index documents
documents = [
    "Python是一种解释型高级编程语言",
    "JavaScript主要用于Web前端开发",
    "Rust注重内存安全和并发性能",
    "Go语言由Google开发，适合后端服务",
    "TypeScript是JavaScript的超集，添加了静态类型",
]

doc_embeddings = model.encode(documents, normalize_embeddings=True)

# Search
query = "哪种编程语言适合写后端？"
query_embedding = model.encode(query, normalize_embeddings=True)

# Compute similarities
scores = util.cos_sim(query_embedding, doc_embeddings)[0]
top_results = torch.topk(scores, k=3)

for score, idx in zip(top_results.values, top_results.indices):
    print(f"Score: {score:.4f} | {documents[idx]}")

from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score

# Generate embeddings for clustering
texts = load_texts()  # List of documents
embeddings = model.encode(texts, normalize_embeddings=True, show_progress_bar=True)

# Find optimal k
silhouette_scores = []
for k in range(2, 11):
    kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
    labels = kmeans.fit_predict(embeddings)
    score = silhouette_score(embeddings, labels)
    silhouette_scores.append((k, score))
    print(f"k={k}, silhouette_score={score:.4f}")

# Cluster with best k
best_k = max(silhouette_scores, key=lambda x: x[1])[0]
kmeans = KMeans(n_clusters=best_k, random_state=42, n_init=10)
labels = kmeans.fit_predict(embeddings)

# Analyze clusters
for cluster_id in range(best_k):
    cluster_texts = [texts[i] for i in range(len(texts)) if labels[i] == cluster_id]
    print(f"\n--- Cluster {cluster_id} ({len(cluster_texts)} docs) ---")
    for text in cluster_texts[:3]:
        print(f"  {text[:80]}...")

from scipy.spatial.distance import cosine

def detect_anomalies(texts, model, threshold=0.3):
    """Detect texts that are semantically different from the majority."""
    embeddings = model.encode(texts, normalize_embeddings=True)

    # Compute centroid
    centroid = np.mean(embeddings, axis=0)
    centroid = centroid / np.linalg.norm(centroid)

    # Distance from centroid
    distances = [cosine(emb, centroid) for emb in embeddings]

    anomalies = []
    for i, (text, dist) in enumerate(zip(texts, distances)):
        if dist > threshold:
            anomalies.append({"text": text, "distance": dist, "index": i})

    return sorted(anomalies, key=lambda x: x["distance"], reverse=True)

# BGE-M3 supports 100+ languages
model = SentenceTransformer('BAAI/bge-m3')

# Encode texts in different languages
texts = {
    "zh": "机器学习是人工智能的一个分支",
    "en": "Machine learning is a branch of artificial intelligence",
    "ja": "機械学習は人工知能の一分野です",
    "ko": "머신러닝은 인공지능의 한 분야입니다",
}

embeddings = {lang: model.encode(text) for lang, text in texts.items()}

# Cross-lingual similarity
for lang_a, emb_a in embeddings.items():
    for lang_b, emb_b in embeddings.items():
        sim = float(util.cos_sim(emb_a, emb_b))
        if lang_a != lang_b:
            print(f"{lang_a}-{lang_b}: {sim:.4f}")
# All pairs show high similarity (>0.85) despite different languages