意味的類似性と文の埋め込み: テキストの比較
2 つの文はどれくらい似ていますか?語彙的な意味（同じ単語）ではなく、 意味的な意味（同じ意味）。 「犬が猫を追いかける」「猫がやってくる」 「犬に追われた」は意味的にはほぼ同じですが、語彙的には大きく異なります。 この質問に答えるのが課題です 意味上の類似性.
セマンティック検索エンジン、レコメンデーション システム、 コンテンツの重複排除、質問応答、RAG (検索拡張生成)、 チャットボットとFAQのマッチング。この記事では、意味的類似性システムを構築します ゼロから: コサイン類似度から Sentence-BERT による文埋め込みまで、 FAISS による高速ベクトル検索まで。
これはシリーズの 9 番目の記事です 最新の NLP: BERT から LLM へ。 このトピックはシリーズに直接関係します AIエンジニアリング/RAG ここで、セマンティック埋め込みは高密度検索の中心です。
何を学ぶか
- コサイン類似度とドット積: 式とそれらをいつ使用するか
- 意味的類似性に対する BERT の制限と Sentence-BERT が必要な理由
- 文-BERT (SBERT): シャムアーキテクチャと三重項損失トレーニング
- HuggingFace の文変換モデル: どれを選択するか
- FAISS による大規模コーパスの意味検索
- イタリア語の文の埋め込み
- ベンチマーク: STS-B、SICK、および評価指標
- クロスエンコーダーとバイエンコーダー: 品質と速度のトレードオフ
- ドメイン上の文トランスフォーマーを微調整する
- FAQマッチングシステムの完全導入
- キャッシュと最適化を備えた本番環境に対応したパイプライン
1. 意味上の類似性の問題
これら 3 つの文グループとその課題について考えてみましょう。
意味上の類似性の例
- 高い類似性: 「銀行が金利を引き上げた」 / 「銀行機関が金利を引き上げた」
- 類似性が低い: 「銀行は金利を引き上げた」/「猫はソファで寝る」
- 欺瞞的 (同じ言葉、異なる意味): 「学校の机」「市場の魚売り場」
- 異言語学: 「犬は速く走る」（同じセマンティクス、異なる言語）
以下のような従来の指標 ジャカードの類似性 または BM25 それらは語彙の重複に依存しており、同義語や言い換えでは完全に失敗します。 単純な TF-IDF ですら意味がわかりません。解決策は次のとおりです。 セマンティック埋め込み: 近接した密なベクトル表現 幾何学的なものは意味上の近さを反映しています。
1.1 コサイン類似度: 基本的な指標
La 少し似ている 空間内の 2 つのベクトル間の角度を測定します 埋め込みの。範囲は -1 (反対) から 1 (同一) で、直交ベクトルの場合は 0 です。 数式は次のとおりです。
cos(A, B) = (A · B) / (||A|| · ||B||)
ベクトルが単位ノルムに正規化されると、コサイン類似度は一致します。 ドット積を使用すると、GPU ハードウェアでの計算がはるかに効率的になります。
import numpy as np
import torch
from torch.nn import functional as F
def cosine_similarity(vec1, vec2):
"""Cosine similarity tra due vettori numpy."""
dot_product = np.dot(vec1, vec2)
norm1 = np.linalg.norm(vec1)
norm2 = np.linalg.norm(vec2)
return dot_product / (norm1 * norm2)
# Versione PyTorch (batch-friendly)
def cosine_similarity_batch(emb1, emb2):
"""Cosine similarity tra batch di embedding (normalizzato)."""
# Normalizza a norma unitaria
emb1_norm = F.normalize(emb1, p=2, dim=1)
emb2_norm = F.normalize(emb2, p=2, dim=1)
return (emb1_norm * emb2_norm).sum(dim=1)
# Esempio con vettori semplici
vec_a = np.array([1.0, 0.5, 0.3, 0.8])
vec_b = np.array([0.9, 0.4, 0.4, 0.7]) # simile ad a
vec_c = np.array([-0.2, 0.8, -0.5, 0.1]) # diverso da a
print(f"sim(a, b) = {cosine_similarity(vec_a, vec_b):.4f}") # alta
print(f"sim(a, c) = {cosine_similarity(vec_a, vec_c):.4f}") # bassa
# Matrice di similarità per corpus di frasi
def similarity_matrix(embeddings):
"""Matrice di similarità N x N per un set di embedding."""
# Normalizza
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
normalized = embeddings / norms
# Prodotto matriciale per tutte le coppie
return normalized @ normalized.T
# Output: matrice (N, N) dove [i,j] = sim(frase_i, frase_j)1.2 その他の距離指標
類似性/距離メトリックの比較
| メトリック | Formula | 範囲 | 使用事例 |
|---|---|---|---|
| コサイン類似度 | cos(A,B) | [-1、1] | 標準的な意味の類似性 |
| ユークリッド距離 | ||A - B|| | [0, +inf) | クラスタリング、k-NN |
| 内積 | A・B | (-inf、+inf) | 正規化ベクトル = コサインの場合 |
| マンハッタンの距離 | 合計(|A-B|) | [0, +inf) | 異常値に対する堅牢性 |
| ピアソン相関 | cov(A,B)/シグマ | [-1、1] | STSベンチマークでの評価 |
2. 標準 BERT が類似性に対して機能しない理由
直感的には、BERT を使用して文の埋め込みを抽出し、比較することができます。 しかし、Reimers & Gurevych による研究 (2019) は、これが アプローチe 驚くほど効果がない.
主な問題は、BERT がマスク言語モデリング (MLM) で事前トレーニングされていることです。
次の文の予測 (NSP)。トークン [CLS] 情報をエンコードする
文ペア分類 (NSP) には便利ですが、最適化されていません。
比較したときに意味的な類似性を反映する埋め込みを生成する
類似性はほとんどありません。
さらに、すべてのトークンの平均プーリングにより埋め込みスペースが生成されます。 異方性: 方向は均一に分布していません。 意味的に異なる文のクラスターが重なり合います。
from transformers import BertModel, BertTokenizer
import torch
import numpy as np
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertModel.from_pretrained('bert-base-uncased')
def bert_mean_pooling(text):
"""Embedding di frase con mean pooling su BERT."""
inputs = tokenizer(text, return_tensors='pt',
truncation=True, max_length=128, padding=True)
with torch.no_grad():
outputs = model(**inputs)
# Mean pooling (esclude padding)
mask = inputs['attention_mask'].unsqueeze(-1)
embeddings = (outputs.last_hidden_state * mask).sum(1) / mask.sum(1)
return embeddings[0].numpy()
# Test: frasi semanticamente simili vs diverse
sent1 = "The weather is lovely today."
sent2 = "It's so beautiful today outside." # simile
sent3 = "My dog bit the mailman." # diversa
emb1 = bert_mean_pooling(sent1)
emb2 = bert_mean_pooling(sent2)
emb3 = bert_mean_pooling(sent3)
sim_1_2 = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
sim_1_3 = np.dot(emb1, emb3) / (np.linalg.norm(emb1) * np.linalg.norm(emb3))
print(f"sim(sent1, sent2) = {sim_1_2:.4f}") # ~0.93 - ok
print(f"sim(sent1, sent3) = {sim_1_3:.4f}") # ~0.87 - troppo alto!
# Problema: BERT tende a produrre embedding simili per tutte le frasi
# perchè il token [CLS] e trainato su NSP, non su similarità semantica
# La soluzione e Sentence-BERTSTS-B での BERT パフォーマンス (ベンチマーク)
STS-B (セマンティック テキスト類似性ベンチマーク) タスクでは、平均プーリングを使用した BERT それはちょうど届く ピアソン r = 0.54、アプローチのかなり下にあります SBERT (0.87) として監修されました。 [CLS]トークンだけでも0.20にしか達しません。 意味的な類似性を考慮すると、SBERT が正しい選択です。
3. 文-BERT (SBERT): 解決策
文-BERT (Reimers & Gurevych、EMNLP 2019) が問題を解決する 建築とともに シャム: BERT の 2 つのインスタンスが重みを共有します。 2 つの文を別々に処理し、損失関数によって表現が強制されます。 意味的にはベクトル空間で近いことに似ています。
3.1 シャムアーキテクチャ
重要な考え方は、2 つの「ネットワーク」がまったく同じ重みを共有するということです。 これらは 2 つの別個のモデルではなく、同じモデルが 2 回呼び出されます。 損失は出力ペアで計算されます。
- 回帰の目的: 予測されたコサイン類似度と人間のスコアの間の MSE (STS の場合)
- 分類の目的: [u, v, |u-v|] のクロスエントロピー (NLI の場合)
- 三重項損失: アンカー/ポジティブ/ネガティブでのマージン損失 (パラフレーズマイニングの場合)
from sentence_transformers import SentenceTransformer, util
import torch
# Carica un modello sentence-transformers
# Modello multilingua (include italiano!)
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# Modello ottimizzato per inglese (più accurato)
# model = SentenceTransformer('all-MiniLM-L6-v2')
# Encoding di frasi (batch-optimized)
sentences = [
"The weather is lovely today.",
"It's so beautiful today outside.",
"He drove to the stadium.",
"La giornata e bellissima oggi.", # italiano
"Il tempo e meraviglioso questa mattina.", # italiano simile
]
# Encode tutto in una volta (molto più efficiente del loop)
embeddings = model.encode(sentences, batch_size=32, show_progress_bar=False)
print(f"Embedding shape: {embeddings.shape}") # (5, 384)
# Calcola similarità
cos_scores = util.cos_sim(embeddings, embeddings)
print("\nMatrice di similarità:")
for i in range(len(sentences)):
for j in range(i+1, len(sentences)):
score = cos_scores[i][j].item()
if score > 0.6: # mostra solo coppie simili
print(f" {i+1} vs {j+1}: {score:.4f}")
print(f" '{sentences[i][:50]}'")
print(f" '{sentences[j][:50]}'")
# Pairwise similarity per coppie specifiche
sim = util.cos_sim(embeddings[0], embeddings[1]).item()
print(f"\nsim(EN1, EN2) = {sim:.4f}") # ~0.85 (frasi simili)
sim_cross = util.cos_sim(embeddings[0], embeddings[3]).item()
print(f"sim(EN1, IT1) = {sim_cross:.4f}") # ~0.75 (cross-lingual!)4. 文変換モデル: どちらを選択するか
主な文変換モデル (2024-2025)
| モデル | 言語 | 薄暗い。 | スピード | STS-B ピアソン |
|---|---|---|---|---|
| all-MiniLM-L6-v2 | EN | 384 | 非常に速い | 0.834 |
| all-mpnet-base-v2 | EN | 768 | 平均 | 0.869 |
| 言い換え-多言語-MiniLM-L12-v2 | 50以上の言語 | 384 | 速い | 0.821 |
| 言い換え-多言語-mpnet-base-v2 | 50以上の言語 | 768 | 平均 | 0.853 |
| intfloat/multilingual-e5-large | 100以上の言語 | 1024 | 遅い | 0.892 |
| テキスト埋め込み-3-small (OpenAI) | 多言語対応 | 1536年 | APIのみ | ~0.90 |
4.1 モデルの選択: 実践ガイド
選択は、言語、スピード、必要な品質という 3 つの主な要素によって決まります。
from sentence_transformers import SentenceTransformer
import time
import numpy as np
def benchmark_model(model_name, sentences, n_runs=3):
"""Benchmark velocità e qualità di un modello sentence-transformer."""
model = SentenceTransformer(model_name)
# Warmup
model.encode(sentences[:2])
# Misura velocità
times = []
for _ in range(n_runs):
start = time.time()
embs = model.encode(sentences)
times.append(time.time() - start)
avg_time = np.mean(times)
dim = embs.shape[1]
print(f"Model: {model_name}")
print(f" Embedding dim: {dim}")
print(f" Avg encoding time ({len(sentences)} sentences): {avg_time*1000:.1f}ms")
print(f" Throughput: {len(sentences)/avg_time:.0f} sentences/sec")
sentences_test = [
"Il sole splende oggi a Milano.",
"Oggi e una bella giornata soleggiata.",
"Roma e la capitale dell'Italia.",
"La Juventus ha vinto il campionato.",
"L'intelligenza artificiale sta cambiando il mondo.",
] * 20 # 100 frasi
# Benchmark modelli multilingua
for model_name in [
'paraphrase-multilingual-MiniLM-L12-v2',
'paraphrase-multilingual-mpnet-base-v2',
'intfloat/multilingual-e5-small',
]:
benchmark_model(model_name, sentences_test)
print()5. FAISS によるセマンティック検索
大規模なコーパス (数百万の文書) の場合、総当たり検索 （すべてのドキュメントとの類似性を計算します）そして遅すぎます。 フェイス （Facebook AI類似性検索）により近似最近傍検索が可能 さまざまなタイプのインデックスを使用したサブリニア時間で。
5.1 FAISS指数の種類
FAISS インデックス: 速度と精度のトレードオフ
| 索引 | タイプ | 使用事例 | リコール (%) | スピード |
|---|---|---|---|---|
| インデックスフラットL2 | ちょうど | < 100K ドキュメント | 100% | 遅い |
| インデックスフラットIP | まさに（些細なこと） | < 100K ドキュメント | 100% | 遅い |
| インデックスIVFFlat | 近似 | 100K～10M | ~95% | 速い |
| インデックスHNSW | 近似 | 100万以上 | ~99% | 非常に速い |
| インデックスIVFPQ | 圧縮された | 10M+、限られた RAM | ~85% | 非常に速い |
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
import time
model = SentenceTransformer('all-MiniLM-L6-v2')
# Corpus di esempio: articoli Wikipedia
corpus = [
"The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris.",
"Apple Inc. is an American multinational technology company founded by Steve Jobs.",
"Python is a high-level, general-purpose programming language.",
"The Mediterranean diet is based on traditional foods from countries bordering the sea.",
"Quantum computing uses quantum-mechanical phenomena such as superposition.",
"The Amazon River is the largest river by discharge volume in the world.",
"Artificial neural networks are computing systems inspired by biological neural networks.",
"The Sistine Chapel ceiling was painted by Michelangelo between 1508 and 1512.",
"Machine learning is a subset of artificial intelligence focused on algorithms.",
"The Colosseum is an oval amphitheatre in the centre of Rome, Italy.",
]
# Encode il corpus (offline, una volta sola)
print("Encoding corpus...")
start = time.time()
corpus_embeddings = model.encode(corpus, convert_to_numpy=True, show_progress_bar=False)
print(f"Encoded {len(corpus)} docs in {time.time()-start:.2f}s")
print(f"Embeddings shape: {corpus_embeddings.shape}") # (10, 384)
# Costruisci indice FAISS
dim = corpus_embeddings.shape[1] # 384
# IndexFlatIP: esatta, cosine similarity su vettori normalizzati
index_ip = faiss.IndexFlatIP(dim)
# Normalizza per usare cosine similarity (dot product su vettori unit-norm)
faiss.normalize_L2(corpus_embeddings)
index_ip.add(corpus_embeddings)
print(f"Index size: {index_ip.ntotal} vettori")
# IndexHNSW: approssimato ma molto veloce, buona per produzione
# M = numero di connessioni per nodo (16-64 in produzione)
index_hnsw = faiss.IndexHNSWFlat(dim, 32, faiss.METRIC_INNER_PRODUCT)
index_hnsw.hnsw.efConstruction = 200 # più alto = migliore recall in build
index_hnsw.hnsw.efSearch = 128 # più alto = migliore recall in search
# Ricerca semantica
def semantic_search(query, index, corpus, model, k=3):
"""Ricerca semantica: restituisce i k documenti più simili alla query."""
query_emb = model.encode([query], convert_to_numpy=True)
faiss.normalize_L2(query_emb)
start = time.time()
distances, indices = index.search(query_emb, k)
search_time = (time.time() - start) * 1000
print(f"\nQuery: '{query}'")
print(f"Search time: {search_time:.2f}ms")
for rank, (dist, idx) in enumerate(zip(distances[0], indices[0]), 1):
print(f" {rank}. [{dist:.4f}] {corpus[idx][:80]}")
return [(corpus[i], float(d)) for i, d in zip(indices[0], distances[0])]
# Test
semantic_search("ancient Roman architecture", index_ip, corpus, model)
semantic_search("programming language features", index_ip, corpus, model)
semantic_search("painting and art in Italy", index_ip, corpus, model)5.2 インデックスの永続性とロード
import faiss
import numpy as np
import json
import os
def build_and_save_index(corpus, model, index_path="faiss_index.bin",
corpus_path="corpus.json"):
"""Costruisce e salva un indice FAISS su disco."""
# Encode
embeddings = model.encode(corpus, convert_to_numpy=True, show_progress_bar=True)
faiss.normalize_L2(embeddings)
dim = embeddings.shape[1]
index = faiss.IndexFlatIP(dim)
index.add(embeddings)
# Salva indice FAISS
faiss.write_index(index, index_path)
# Salva corpus (per recuperare testi)
with open(corpus_path, 'w', encoding='utf-8') as f:
json.dump(corpus, f, ensure_ascii=False, indent=2)
print(f"Index saved: {index.ntotal} vettori -> {index_path}")
return index
def load_index(index_path="faiss_index.bin", corpus_path="corpus.json"):
"""Carica indice FAISS e corpus da disco."""
if not os.path.exists(index_path):
raise FileNotFoundError(f"Index not found: {index_path}")
index = faiss.read_index(index_path)
with open(corpus_path, 'r', encoding='utf-8') as f:
corpus = json.load(f)
print(f"Index loaded: {index.ntotal} vettori")
return index, corpus
# Utilizzo
# Prima volta: costruisce e salva
# index = build_and_save_index(my_corpus, model)
# Riavvii successivi: carica direttamente (molto più veloce)
# index, corpus = load_index()6. FAQ マッチング: 完全な使用例
意味的類似性の実際的な応用: 自動質問照合 既存の FAQ をユーザーに提供します。このパターンは多くのチャットボットやシステムの基礎となっています カスタマーサポート。
from sentence_transformers import SentenceTransformer, util
import torch
import json
class FAQMatcher:
"""Sistema di FAQ matching semantico con caching e persistenza."""
def __init__(self, model_name='paraphrase-multilingual-MiniLM-L12-v2',
threshold=0.7):
self.model = SentenceTransformer(model_name)
self.threshold = threshold
self.faqs = []
self.faq_embeddings = None
def load_faqs(self, faqs: list):
"""
faqs: lista di dizionari con 'question', 'answer', 'category'
"""
self.faqs = faqs
questions = [faq['question'] for faq in faqs]
print(f"Encoding {len(questions)} FAQ...")
self.faq_embeddings = self.model.encode(
questions,
convert_to_tensor=True,
show_progress_bar=False
)
print("FAQ pronte per la ricerca!")
def match(self, user_query: str, top_k: int = 3) -> list:
"""Trova le FAQ più simili alla domanda utente."""
if self.faq_embeddings is None:
raise ValueError("Carica prima le FAQ con load_faqs()")
query_emb = self.model.encode(user_query, convert_to_tensor=True)
scores = util.cos_sim(query_emb, self.faq_embeddings)[0]
top_k_indices = torch.topk(scores, k=min(top_k, len(self.faqs))).indices
results = []
for idx in top_k_indices:
score = scores[idx].item()
if score >= self.threshold:
results.append({
'question': self.faqs[idx]['question'],
'answer': self.faqs[idx]['answer'],
'category': self.faqs[idx].get('category', 'N/A'),
'score': round(score, 4)
})
return results
def respond(self, user_query: str) -> str:
"""Risposta automatica alla domanda utente."""
matches = self.match(user_query, top_k=1)
if not matches:
return f"Mi dispiace, non ho trovato una risposta per '{user_query}'. Contatta il supporto."
best = matches[0]
return f"[{best['category']}] {best['answer']} (Confidenza: {best['score']:.2f})"
# Esempio di utilizzo
faqs_ecommerce = [
{
"question": "Come posso restituire un prodotto?",
"answer": "Puoi restituire il prodotto entro 30 giorni dall'acquisto contattando il supporto.",
"category": "Resi"
},
{
"question": "Quanto tempo impiega la spedizione?",
"answer": "La consegna standard impiega 3-5 giorni lavorativi, l'express 24 ore.",
"category": "Spedizioni"
},
{
"question": "Come posso pagare?",
"answer": "Accettiamo carte di credito, PayPal, bonifico bancario e contrassegno.",
"category": "Pagamenti"
},
{
"question": "Il prodotto e in garanzia?",
"answer": "Tutti i prodotti hanno 2 anni di garanzia legale del consumatore.",
"category": "Garanzia"
},
{
"question": "Posso tracciare il mio ordine?",
"answer": "Si, riceverai un'email con il numero di tracking dopo la spedizione.",
"category": "Ordini"
},
]
matcher = FAQMatcher()
matcher.load_faqs(faqs_ecommerce)
test_queries = [
"Voglio rimandare indietro la merce",
"Quando arriva il pacco?",
"Accettate il bonifico?",
"Ho bisogno del codice di tracciamento",
"L'articolo si e rotto, cosa faccio?",
]
print("\n=== FAQ Matching ===")
for query in test_queries:
response = matcher.respond(query)
print(f"\nDomanda: {query}")
print(f"Risposta: {response}")7. クロスエンコーダーとバイエンコーダー
意味上の類似性には、異なるトレードオフをもたらす 2 つのアプローチがあります。 品質/スピード。適切なアーキテクチャを選択するには、それらを理解することが不可欠です。
バイエンコーダーとクロスエンコーダー
| 待ってます | バイエンコーダー (SBERT) | クロスエンコーダ |
|---|---|---|
| 建築 | 2 つの別々の BERT、エンベディングを生成 | ペアを処理する BERT |
| スピード | 非常に高速 (事前計算埋め込み) | 遅い (ペアごとに処理) |
| スケーラビリティ | 数百万の文書 | わずか数百組のカップル |
| 品質 | 良好 (STS-B では ~0.87 ピアソン) | 優れています (~0.92 ピアソン) |
| 使用事例 | 検索、意味検索 | 結果の再ランキング |
| 複雑さ O(n) | クエリの場合は O(1) (事前に計算された埋め込み) | クエリごとに O(n) |
from sentence_transformers import SentenceTransformer, CrossEncoder, util
# Bi-encoder per il retrieval iniziale (veloce)
bi_encoder = SentenceTransformer('all-MiniLM-L6-v2')
# Cross-encoder per il reranking (accurato)
cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
# Pipeline a due stadi (best of both worlds)
def retrieval_and_rerank(query, corpus, corpus_embeddings, top_k=100, final_k=5):
"""
Stage 1: Bi-encoder retrieval (veloce, ritorna top 100)
Stage 2: Cross-encoder reranking (accurato, sui top 100)
"""
# Stage 1: Bi-encoder retrieval
query_emb = bi_encoder.encode(query, convert_to_tensor=True)
hits = util.semantic_search(query_emb, corpus_embeddings, top_k=top_k)[0]
# Stage 2: Cross-encoder reranking
cross_inp = [[query, corpus[hit['corpus_id']]] for hit in hits]
cross_scores = cross_encoder.predict(cross_inp)
# Combina e riordina
for hit, score in zip(hits, cross_scores):
hit['cross_score'] = score
hits = sorted(hits, key=lambda x: x['cross_score'], reverse=True)[:final_k]
print(f"\nQuery: '{query}'")
for rank, hit in enumerate(hits, 1):
bi_score = hit['score']
cross_score = hit['cross_score']
doc = corpus[hit['corpus_id']][:80]
print(f" {rank}. [bi={bi_score:.3f}, cross={cross_score:.3f}] {doc}")
return hits
# Encode corpus una sola volta
corpus_embs = bi_encoder.encode(corpus, convert_to_tensor=True)
retrieval_and_rerank("ancient Roman buildings", corpus, corpus_embs)8. 評価: STS-B と指標
意味的類似性システムを正しく評価するには、次のことが必要です。 標準化されたベンチマーク データセット。 STS-B は英語の主なリファレンスです。 一方、STS-IT はイタリア語で利用できます。
from sentence_transformers import SentenceTransformer
from sentence_transformers.evaluation import EmbeddingSimilarityEvaluator
from datasets import load_dataset
import numpy as np
from scipy.stats import pearsonr, spearmanr
# Carica STS-B per l'evaluazione
stsb = load_dataset("mteb/stsbenchmark-sts")
val_data = stsb['validation']
sentences1 = val_data['sentence1']
sentences2 = val_data['sentence2']
scores = [s / 5.0 for s in val_data['score']] # normalizza 0-5 a 0-1
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# Evaluation automatica
evaluator = EmbeddingSimilarityEvaluator(
sentences1=sentences1,
sentences2=sentences2,
scores=scores,
name="sts-val"
)
pearson = model.evaluate(evaluator)
print(f"STS-B validation - Pearson: {pearson:.4f}")
# Valutazione manuale con correlazione di Pearson e Spearman
emb1 = model.encode(sentences1, show_progress_bar=False)
emb2 = model.encode(sentences2, show_progress_bar=False)
from numpy.linalg import norm
cos_sims = [
np.dot(e1, e2) / (norm(e1) * norm(e2))
for e1, e2 in zip(emb1, emb2)
]
pearson_r, _ = pearsonr(cos_sims, scores)
spearman_r, _ = spearmanr(cos_sims, scores)
print(f"Pearson: {pearson_r:.4f}")
print(f"Spearman: {spearman_r:.4f}")
# Analisi degli errori: trova le coppie più sbagliate
errors = [(abs(p - t), s1, s2, p, t)
for p, t, s1, s2 in zip(cos_sims, scores, sentences1, sentences2)]
errors.sort(reverse=True)
print("\n=== Top 3 Errori ===")
for err, s1, s2, pred, true in errors[:3]:
print(f" Errore: {err:.3f} | Pred: {pred:.3f} | True: {true:.3f}")
print(f" '{s1[:60]}'")
print(f" '{s2[:60]}'")9. ドメインの文トランスフォーマーを微調整する
事前トレーニングされたモデルは一般的なテキストには最適ですが、特定のドメインには最適です (医療、法律、技術) 注釈付きの文のペアを微調整する方が良いでしょう。
from sentence_transformers import (
SentenceTransformer,
InputExample,
losses,
evaluation
)
from torch.utils.data import DataLoader
# Dati di training: coppie (frase1, frase2, score)
# Score: 0.0 = totalmente diverse, 1.0 = identiche
train_examples = [
InputExample(texts=["Diagnosi di diabete tipo 2", "Paziente con iperglicemia cronica"], label=0.85),
InputExample(texts=["Prescrizione antibiotico", "Terapia con amoxicillina"], label=0.80),
InputExample(texts=["Intervento chirurgico al ginocchio", "Artroscopia del menisco"], label=0.75),
InputExample(texts=["Pressione arteriosa elevata", "Ipertensione arteriosa"], label=0.95),
InputExample(texts=["Dolore toracico", "Bruciore di stomaco"], label=0.30),
InputExample(texts=["Frattura del femore", "Infarto del miocardio"], label=0.05),
]
# Carica modello base
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# DataLoader
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
# Loss: CosineSimilarityLoss per regressione su score continuo
train_loss = losses.CosineSimilarityLoss(model)
# Valutazione su dati di test
test_examples = [
InputExample(texts=["Cefalea tensiva", "Mal di testa da stress"], label=0.88),
InputExample(texts=["Diabete gestazionale", "Diabete in gravidanza"], label=0.92),
]
evaluator_sentences1 = [e.texts[0] for e in test_examples]
evaluator_sentences2 = [e.texts[1] for e in test_examples]
evaluator_scores = [e.label for e in test_examples]
val_evaluator = evaluation.EmbeddingSimilarityEvaluator(
evaluator_sentences1, evaluator_sentences2, evaluator_scores
)
# Fine-tuning
model.fit(
train_objectives=[(train_dataloader, train_loss)],
evaluator=val_evaluator,
epochs=10,
evaluation_steps=50,
warmup_steps=100,
output_path='./medical-sentence-transformer',
save_best_model=True
)
print("Fine-tuning completato!")
print("Modello salvato in './medical-sentence-transformer'")10. パイプラインの本番環境対応
本番環境のセマンティック類似性システムは、埋め込みキャッシュを管理する必要があります。 コーパスの増分更新と品質モニタリング。
import faiss
import numpy as np
import json
import hashlib
from sentence_transformers import SentenceTransformer, util
from pathlib import Path
from typing import List, Dict, Optional
class SemanticSearchEngine:
"""
Motore di ricerca semantica production-ready con:
- Caching degli embedding su disco
- Aggiornamento incrementale
- Threshold configurabile
- Logging delle query
"""
def __init__(
self,
model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2',
cache_dir: str = './search_cache',
similarity_threshold: float = 0.5
):
self.model = SentenceTransformer(model_name)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.threshold = similarity_threshold
self.documents: List[Dict] = []
self.embeddings: Optional[np.ndarray] = None
self.index: Optional[faiss.Index] = None
def _doc_hash(self, doc: Dict) -> str:
"""Hash del documento per il caching."""
content = json.dumps(doc, sort_keys=True, ensure_ascii=False)
return hashlib.md5(content.encode()).hexdigest()
def add_documents(self, documents: List[Dict], text_field: str = 'text'):
"""Aggiunge documenti al corpus con caching."""
texts = [doc[text_field] for doc in documents]
new_embeddings = self.model.encode(texts, convert_to_numpy=True, show_progress_bar=True)
if self.embeddings is None:
self.embeddings = new_embeddings
else:
self.embeddings = np.vstack([self.embeddings, new_embeddings])
self.documents.extend(documents)
self._rebuild_index()
print(f"Corpus: {len(self.documents)} documenti")
def _rebuild_index(self):
"""Ricostruisce l'indice FAISS."""
dim = self.embeddings.shape[1]
self.index = faiss.IndexFlatIP(dim)
embs_normalized = self.embeddings.copy()
faiss.normalize_L2(embs_normalized)
self.index.add(embs_normalized)
def search(self, query: str, k: int = 5, text_field: str = 'text') -> List[Dict]:
"""Cerca i documenti più rilevanti per la query."""
if self.index is None or len(self.documents) == 0:
return []
query_emb = self.model.encode([query], convert_to_numpy=True)
faiss.normalize_L2(query_emb)
distances, indices = self.index.search(query_emb, min(k, len(self.documents)))
results = []
for dist, idx in zip(distances[0], indices[0]):
if dist >= self.threshold:
result = dict(self.documents[idx])
result['score'] = float(dist)
results.append(result)
return results
def save(self):
"""Persiste il motore di ricerca su disco."""
faiss.write_index(self.index, str(self.cache_dir / 'index.faiss'))
np.save(str(self.cache_dir / 'embeddings.npy'), self.embeddings)
with open(self.cache_dir / 'documents.json', 'w', encoding='utf-8') as f:
json.dump(self.documents, f, ensure_ascii=False, indent=2)
print(f"Engine saved to {self.cache_dir}")
# Utilizzo
engine = SemanticSearchEngine(similarity_threshold=0.6)
# Aggiungi documenti
docs = [
{"text": "Come configurare un ambiente Python con virtualenv.", "id": "py001", "category": "python"},
{"text": "Installazione e configurazione di Docker su Ubuntu.", "id": "docker001", "category": "devops"},
{"text": "Introduzione alle reti neurali con PyTorch.", "id": "ml001", "category": "ml"},
{"text": "Best practices per la sicurezza delle API REST.", "id": "api001", "category": "security"},
{"text": "Ottimizzazione delle query SQL con indici.", "id": "db001", "category": "database"},
]
engine.add_documents(docs)
# Ricerca
results = engine.search("come creare un ambiente virtuale Python")
for r in results:
print(f"[{r['score']:.3f}] {r['text']}")11. イタリア語の文埋め込み
イタリア語については、多言語モデルから微調整まで、さまざまなオプションがあります イタリア語コーパスに特有。ここに実践的なガイドがあります。
from sentence_transformers import SentenceTransformer, util
# Opzione 1: Modello multilingua (più pratico, supporta 50+ lingue)
model_multi = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
# Opzione 2: E5 multilingual (stato dell'arte per retrieval)
model_e5 = SentenceTransformer('intfloat/multilingual-e5-large')
# Frasi italiane di test
frasi_it = [
"Il governo italiano ha approvato la nuova legge sul lavoro.",
"Il parlamento ha votato la riforma del mercato del lavoro.", # simile
"La Serie A e il campionato di calcio italiano.", # diversa
"Juventus e Inter si sfideranno nel prossimo derby.", # correlata (calcio)
"Il PIL italiano e cresciuto del 2% nel 2024.", # diversa
]
# Test con modello multilingua
embeddings = model_multi.encode(frasi_it)
sim_matrix = util.cos_sim(embeddings, embeddings)
print("=== Similarità tra frasi italiane (multilingua mpnet) ===")
for i in range(len(frasi_it)):
for j in range(i+1, len(frasi_it)):
score = sim_matrix[i][j].item()
if score > 0.5:
print(f" {score:.3f} | '{frasi_it[i][:50]}'")
print(f" | '{frasi_it[j][:50]}'")
# Per E5: aggiungere prefisso "query: " o "passage: " per retrieval
query = "query: come va l'economia italiana?"
passages = [f"passage: {f}" for f in frasi_it]
q_emb = model_e5.encode(query)
p_embs = model_e5.encode(passages)
scores = util.cos_sim(q_emb, p_embs)
top3 = scores[0].topk(3)
print("\n=== Top 3 risultati con E5 ===")
for score, idx in zip(top3.values, top3.indices):
print(f" {score:.3f} | {frasi_it[idx]}")12. 一般的なエラーとアンチパターン
アンチパターン: BERT [CLS] を直接使用する
トークン [CLS] BERT は意味的な類似性に関しては最適化されていません。
(類似性タスクを微調整せずに) 直接使用すると結果が得られます
SBERTよりもはるかに悪い。常に専用の文変換テンプレートを使用してください。
アンチパターン: 異なるパターンの埋め込みを比較する
の埋め込み all-MiniLM-L6-v2 そしての
paraphrase-multilingual-mpnet-base-v2 彼らはベクトル空間に住んでいます
全く違います。異なるモデルで生成されたエンベディングを比較することはできません。
コーパス内のすべての文には常に同じパターンを使用してください。
アンチパターン: 正規化を無視する
FAISS を使用する場合 IndexFlatIP コサイン類似度の場合、
ベクトルを単位ノルムに正規化する必要があります faiss.normalize_L2()
インデックス作成時と検索時の両方で。このステップを忘れてください
明示的なエラーがなくても、誤った結果が生成されます。
ベストプラクティス: チェックリスト
- アメリカ合衆国 文変換者 意味上の類似性を得るために生の BERT の代わりに
- イタリア語または多言語コンテンツ用の多言語テンプレートを選択する
- IndexFlatIP を使用して FAISS インデックスを作成する前に、必ずベクトルを正規化します。
- 再起動のたびに再エンコードを避けるために、エンベディングをディスク上に保持します。
- スケーラブルで高品質な取得を実現するバイエンコーダー + クロスエンコーダー パイプライン
- 導入前に STS-B またはドメインのデータセットで評価する
- 本番環境での類似性スコアの分布を監視してドリフトを検出する
- 無関係な一致を除外するための最小信頼しきい値を設定します。
結論と次のステップ
文の埋め込みとの意味的な類似性は基本的な要素です 多くの最新の NLP アプリケーション: セマンティック検索、FAQ マッチング、重複排除、 推奨システムと RAG システム。 SBERT と文変換モデル FAISS は、わずか数行のコードでこれらの機能にアクセスできるようにしました。 ミリ秒単位の遅延を維持しながら、数百万のドキュメントに拡張できます。
イタリア語、多言語モデルの場合、 paraphrase-multilingual-mpnet-base-v2
e intfloat/multilingual-e5-large 彼らは優れたパフォーマンスを提供します
言語を超えた文脈でも。
重要なポイント
- アメリカ合衆国 スバート 意味的類似性については標準の BERT の代わりに (ピアソン 0.87 対 0.54)
- フェイス 大規模なコーパス研究には不可欠です
- バイエンコーダー + クロスエンコーダー パイプライン: 取得速度 + 再ランキングの品質
- イタリア語の多言語テンプレート:
paraphrase-multilingual-mpnet-base-v2omultilingual-e5-large - 常にレートアップ STS-B またはドメインのデータセット
- ドメイン固有の微調整
CosineSimilarityLoss最高の品質のために
シリーズは続く
- 第10条： 本番環境での NLP モニタリング — ドリフト検出と自動再トレーニング
- 第8条： ローカル LoRA 微調整 — コンシューマ GPU を使用して LLM をドメインに適応させる
- 関連シリーズ： AIエンジニアリング/RAG — 密な検索の核心としての意味的類似性
- 関連シリーズ： 高度なディープラーニング — 三重項損失、計量学習、および対照学習