의미론적 유사성과 문장 임베딩: 텍스트 비교
두 문장은 얼마나 유사합니까? 어휘적인 의미(같은 단어)가 아니라, 의미적 의미(동일한 의미). “개가 고양이를 쫓는다” 그리고 “고양이가 온다” Chased by the canine"은 의미상으로는 거의 동일하지만 어휘적으로는 매우 다릅니다. 이 질문에 대답하는 것은 다음과 같은 과제입니다. 의미론적 유사성.
응용 프로그램은 의미 검색 엔진, 추천 시스템, 콘텐츠 중복 제거, 질문 응답, RAG(Retrieval-Augmented Generation), 챗봇과 FAQ 매칭. 이 글에서는 의미론적 유사성 시스템을 구축할 것입니다. 처음부터: 코사인 유사성부터 Sentence-BERT를 사용한 문장 임베딩까지, FAISS를 이용한 빠른 벡터 검색까지.
이 시리즈의 아홉 번째 기사입니다 최신 NLP: BERT에서 LLM까지. 이 주제는 시리즈와 직접 연결됩니다. AI 엔지니어링/RAG 여기서 의미론적 임베딩은 밀집 검색의 핵심입니다.
무엇을 배울 것인가
- 코사인 유사성과 내적: 공식 및 사용 시기
- 의미적 유사성에 대한 BERT의 한계와 Sentence-BERT가 필요한 이유
- Sentence-BERT(SBERT): Siamese 아키텍처 및 삼중항 손실 훈련
- HuggingFace의 문장 변환기 모델: 선택할 모델
- FAISS를 사용한 대규모 말뭉치에 대한 의미 검색
- 이탈리아어용 문장 임베딩
- 벤치마킹: STS-B, SICK 및 평가 지표
- 크로스 인코더와 바이 인코더: 품질/속도 절충
- 도메인에서 문장 변환기 미세 조정
- FAQ 매칭 시스템 완벽 구현
- 캐싱 및 최적화 기능을 갖춘 프로덕션 준비 파이프라인
1. 의미적 유사성의 문제
다음 세 가지 문장 그룹과 해당 문제를 고려해 보겠습니다.
의미론적 유사성의 예
- 높은 유사성: “은행이 금리를 올렸다” / “은행이 금리를 올렸다”
- 낮은 유사성: "은행이 금리를 올렸다" / "고양이는 소파에서 잔다"
- 기만적(같은 단어, 다른 의미): "학교 책상" / "시장의 생선 판매대"
- 교차 언어학: "개는 빨리 달린다"(동일한 의미, 다른 언어)
다음과 같은 전통적인 측정항목 자카드 유사성 또는 BM25 그들은 어휘 중복에 의존하고 동의어와 의역에서는 완전히 실패합니다. 단순한 TF-IDF조차도 의미를 포착하지 못합니다. 해결책은 다음과 같습니다. 의미론적 임베딩: 근접성이 있는 밀집된 벡터 표현 기하학은 의미론적 근접성을 반영합니다.
1.1 코사인 유사성: 기본 측정항목
La 약간의 유사성 공간에서 두 벡터 사이의 각도를 측정합니다. 임베딩의 범위는 -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 기타 거리 측정법
유사성/거리 측정항목 비교
| 미터법 | 공식 | 범위 | 사용 사례 |
|---|---|---|---|
| 코사인 유사성 | 왜냐하면(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)의 연구에 따르면 이는 전자에게 접근하다 놀랍게도 비효율적이다.
가장 큰 문제는 BERT가 MLM(Masked Language Modeling)으로 사전 훈련되어 있다는 것입니다.
다음 문장 예측(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의 두 인스턴스는 가중치를 공유합니다. 두 문장을 별도로 처리하고 손실 함수는 표현을 강제합니다. 벡터 공간에서 가깝다는 것과 의미상 유사합니다.
3.1 샴 아키텍처
핵심 아이디어는 두 "네트워크"가 정확히 동일한 가중치를 공유한다는 것입니다. 이들은 두 개의 별도 모델이 아니라 동일한 모델을 두 번 호출한 것입니다. 손실은 출력 쌍에서 계산됩니다.
- 회귀 목표: 예측된 코사인 유사성과 인간 점수 사이의 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 피어슨 |
|---|---|---|---|---|
| 모든-MiniLM-L6-v2 | EN | 384 | 매우 빠르다 | 0.834 |
| 모든 mpnet-base-v2 | EN | 768 | 평균 | 0.869 |
| 다른 표현-다국어-MiniLM-L12-v2 | 50개 이상의 언어 | 384 | 빠른 | 0.821 |
| 의역-다국어-mpnet-base-v2 | 50개 이상의 언어 | 768 | 평균 | 0.853 |
| intfloat/다국어-e5-large | 100개 이상의 언어 | 1024 | 느린 | 0.892 |
| 텍스트 임베딩-3-소형(OpenAI) | 다국어 | 1536년 | API 전용 | ~0.90 |
4.1 모델 선택: 실용 가이드
선택은 언어, 속도, 필요한 품질이라는 세 가지 주요 요소에 따라 달라집니다.
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를 이용한 의미 검색
대규모 자료(수백만 개의 문서)의 경우 무차별 검색 (모든 문서와의 유사성 계산) 너무 느립니다. FAISS (Facebook AI 유사성 검색)은 대략적인 최근접 이웃 검색을 허용합니다. 다양한 유형의 인덱스를 사용하여 하위 선형 시간에.
5.1 FAISS 지수의 종류
FAISS 지수: 속도/정확도 균형
| 색인 | 유형 | 사용 사례 | 리콜(%) | 속도 |
|---|---|---|---|---|
| 인덱스플랫L2 | 정확한 | 10만개 미만의 문서 | 100% | 느린 |
| 인덱스플랫IP | 바로 그거야 (사소한 것들) | 10만개 미만의 문서 | 100% | 느린 |
| 색인IVFFFlat | 근사치를 내다 | 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. 크로스 인코더 vs 바이 인코더
서로 다른 절충안을 제공하는 의미론적 유사성에 대한 두 가지 접근 방식이 있습니다. 품질/속도. 이를 이해하는 것은 올바른 아키텍처를 선택하는 데 필수적입니다.
바이 인코더와 크로스 인코더
| 나는 기다린다 | 바이 인코더(SBERT) | 크로스 엔코더 |
|---|---|---|
| 건축학 | 두 개의 별도 BERT가 임베딩을 생성합니다. | 쌍을 처리하는 BERT |
| 속도 | 매우 빠름(사전 계산 임베딩) | 느림(모든 쌍 처리) |
| 확장성 | 수백만 개의 문서 | 고작 몇백커플 |
| 품질 | 좋음(STS-B에서 ~0.87 Pearson) | 우수함(~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 그들은 뛰어난 성능을 제공합니다
언어 간 상황에서도 마찬가지입니다.
핵심 사항
- 미국 SERT 의미론적 유사성을 위해 표준 BERT 대신(Pearson 0.87 대 0.54)
- FAISS 대규모 코퍼스 연구에 꼭 필요한
- 이중 인코더 + 교차 인코더 파이프라인: 검색 속도 + 순위 재지정 품질
- 이탈리아어를 위한 다국어 템플릿:
paraphrase-multilingual-mpnet-base-v2omultilingual-e5-large - 항상 평가를 올려주세요 STS-B 또는 도메인의 데이터 세트
- 도메인별 미세 조정
CosineSimilarityLoss최고의 품질을 위해
시리즈는 계속됩니다
- 제10조: 프로덕션에서의 NLP 모니터링 — 드리프트 감지 및 자동 재교육
- 제8조: 로컬 LoRA 미세 조정 — 소비자 GPU를 사용하여 도메인에 LLM을 적용합니다.
- 관련 시리즈: AI 엔지니어링/RAG — 밀집 검색의 핵심인 의미론적 유사성
- 관련 시리즈: 고급 딥러닝 — 삼중 손실, 메트릭 학습 및 대조 학습