语义搜索完全指南【2026】- 从原理到实战构建智能搜索系统

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

texts = [
    "语义搜索基于向量相似度",
    "Semantic search uses vector similarity",
    "今天天气真好"
]

embeddings = model.encode(texts)
print(f"向量维度: {embeddings.shape}")

import numpy as np
from numpy.linalg import norm

def cosine_similarity(vec1, vec2):
    return np.dot(vec1, vec2) / (norm(vec1) * norm(vec2))

similarity_01 = cosine_similarity(embeddings[0], embeddings[1])
similarity_02 = cosine_similarity(embeddings[0], embeddings[2])

print(f"语义搜索(中) vs 语义搜索(英): {similarity_01:.4f}")
print(f"语义搜索 vs 天气: {similarity_02:.4f}")

import numpy as np
from numpy.linalg import norm

def cosine_similarity(vec1, vec2):
    dot_product = np.dot(vec1, vec2)
    norm_product = norm(vec1) * norm(vec2)
    return dot_product / norm_product

def batch_cosine_similarity(query_vec, doc_vecs):
    query_norm = norm(query_vec)
    doc_norms = norm(doc_vecs, axis=1)
    dot_products = np.dot(doc_vecs, query_vec)
    return dot_products / (doc_norms * query_norm)

def euclidean_distance(vec1, vec2):
    return np.sqrt(np.sum((vec1 - vec2) ** 2))

def euclidean_to_similarity(distance, scale=1.0):
    return 1 / (1 + distance * scale)

def dot_product_similarity(vec1, vec2):
    return np.dot(vec1, vec2)

def normalize_vectors(vectors):
    norms = norm(vectors, axis=1, keepdims=True)
    return vectors / norms

from sentence_transformers import SentenceTransformer
from rank_bm25 import BM25Okapi
import numpy as np

class HybridSearch:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)
        self.documents = []
        self.embeddings = None
        self.bm25 = None
    
    def index(self, documents):
        self.documents = documents
        self.embeddings = self.model.encode(documents)
        
        tokenized = [doc.lower().split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized)
    
    def search(self, query, top_k=5, semantic_weight=0.5):
        query_embedding = self.model.encode(query)
        semantic_scores = np.dot(self.embeddings, query_embedding)
        semantic_scores = (semantic_scores - semantic_scores.min()) / (semantic_scores.max() - semantic_scores.min() + 1e-8)
        
        bm25_scores = np.array(self.bm25.get_scores(query.lower().split()))
        bm25_scores = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-8)
        
        hybrid_scores = semantic_weight * semantic_scores + (1 - semantic_weight) * bm25_scores
        
        top_indices = np.argsort(hybrid_scores)[::-1][:top_k]
        
        return [
            {"document": self.documents[i], "score": hybrid_scores[i]}
            for i in top_indices
        ]

search_engine = HybridSearch()
search_engine.index([
    "Python是一种流行的编程语言",
    "机器学习需要大量训练数据",
    "语义搜索理解查询意图",
    "向量数据库存储嵌入向量",
    "自然语言处理分析文本"
])

results = search_engine.search("如何处理文本数据", top_k=3)
for r in results:
    print(f"得分: {r['score']:.4f} | {r['document']}")

from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.utils import embedding_functions
import numpy as np
from typing import List, Dict

class SemanticSearchEngine:
    def __init__(self, model_name='BAAI/bge-base-zh-v1.5', persist_dir='./search_db'):
        self.model = SentenceTransformer(model_name)
        self.client = chromadb.PersistentClient(path=persist_dir)
        
        self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name=model_name
        )
        
        self.collection = self.client.get_or_create_collection(
            name="documents",
            embedding_function=self.embedding_fn,
            metadata={"hnsw:space": "cosine"}
        )
    
    def add_documents(self, documents: List[Dict], batch_size=100):
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i+batch_size]
            
            self.collection.add(
                documents=[doc['content'] for doc in batch],
                metadatas=[doc.get('metadata', {}) for doc in batch],
                ids=[doc['id'] for doc in batch]
            )
        
        print(f"已索引 {len(documents)} 个文档")
    
    def search(self, query: str, top_k: int = 5, filter_metadata: Dict = None) -> List[Dict]:
        where_filter = filter_metadata if filter_metadata else None
        
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            where=where_filter
        )
        
        search_results = []
        for i in range(len(results['documents'][0])):
            search_results.append({
                'id': results['ids'][0][i],
                'content': results['documents'][0][i],
                'metadata': results['metadatas'][0][i] if results['metadatas'] else {},
                'score': 1 - results['distances'][0][i]
            })
        
        return search_results
    
    def batch_search(self, queries: List[str], top_k: int = 5) -> List[List[Dict]]:
        results = self.collection.query(
            query_texts=queries,
            n_results=top_k
        )
        
        all_results = []
        for q_idx in range(len(queries)):
            query_results = []
            for i in range(len(results['documents'][q_idx])):
                query_results.append({
                    'id': results['ids'][q_idx][i],
                    'content': results['documents'][q_idx][i],
                    'score': 1 - results['distances'][q_idx][i]
                })
            all_results.append(query_results)
        
        return all_results

search_engine = SemanticSearchEngine()

documents = [
    {"id": "1", "content": "语义搜索通过理解查询意图来返回相关结果", "metadata": {"category": "search"}},
    {"id": "2", "content": "向量嵌入将文本转换为数值表示", "metadata": {"category": "embedding"}},
    {"id": "3", "content": "HNSW算法实现高效的近似最近邻搜索", "metadata": {"category": "algorithm"}},
    {"id": "4", "content": "RAG系统结合检索和生成提升回答质量", "metadata": {"category": "rag"}},
    {"id": "5", "content": "混合搜索结合关键词和语义搜索的优势", "metadata": {"category": "search"}}
]

search_engine.add_documents(documents)

results = search_engine.search("如何实现智能搜索", top_k=3)
print("\n搜索结果：")
for r in results:
    print(f"  [{r['score']:.4f}] {r['content']}")

from typing import List

class TextChunker:
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def chunk_by_size(self, text: str) -> List[str]:
        chunks = []
        start = 0
        
        while start < len(text):
            end = start + self.chunk_size
            
            if end < len(text):
                break_point = text.rfind('。', start, end)
                if break_point == -1:
                    break_point = text.rfind(' ', start, end)
                if break_point > start:
                    end = break_point + 1
            
            chunks.append(text[start:end].strip())
            start = end - self.chunk_overlap
        
        return chunks
    
    def chunk_by_paragraph(self, text: str) -> List[str]:
        paragraphs = text.split('\n\n')
        chunks = []
        current_chunk = ""
        
        for para in paragraphs:
            if len(current_chunk) + len(para) <= self.chunk_size:
                current_chunk += para + "\n\n"
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = para + "\n\n"
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks

chunker = TextChunker(chunk_size=300, chunk_overlap=30)

long_text = """
语义搜索是现代信息检索的核心技术。它通过理解查询的语义含义，而不仅仅是匹配关键词，来返回更相关的结果。

传统的关键词搜索依赖于精确的词汇匹配。如果用户搜索"汽车"，系统只会返回包含"汽车"这个词的文档，而不会返回包含"轿车"或"机动车"的文档。

语义搜索通过向量嵌入技术解决了这个问题。嵌入模型将文本转换为高维向量，语义相似的文本在向量空间中距离更近。这样，即使查询和文档使用不同的词汇，只要语义相近，就能被检索到。
"""

chunks = chunker.chunk_by_size(long_text)
for i, chunk in enumerate(chunks):
    print(f"块 {i+1}: {chunk[:50]}...")

class QueryOptimizer:
    def __init__(self, model):
        self.model = model
    
    def expand_query(self, query: str, expansions: List[str]) -> str:
        return f"{query} {' '.join(expansions)}"
    
    def rewrite_query(self, query: str) -> str:
        rewrites = {
            "怎么": "如何",
            "咋": "怎样",
            "啥": "什么"
        }
        for old, new in rewrites.items():
            query = query.replace(old, new)
        return query
    
    def multi_query_search(self, queries: List[str], search_fn, top_k=5):
        all_results = {}
        
        for query in queries:
            results = search_fn(query, top_k=top_k)
            for r in results:
                doc_id = r['id']
                if doc_id not in all_results:
                    all_results[doc_id] = r
                    all_results[doc_id]['query_count'] = 1
                else:
                    all_results[doc_id]['score'] = max(all_results[doc_id]['score'], r['score'])
                    all_results[doc_id]['query_count'] += 1
        
        sorted_results = sorted(
            all_results.values(),
            key=lambda x: (x['query_count'], x['score']),
            reverse=True
        )
        
        return sorted_results[:top_k]

class ResultReranker:
    def __init__(self, cross_encoder_model='cross-encoder/ms-marco-MiniLM-L-6-v2'):
        from sentence_transformers import CrossEncoder
        self.cross_encoder = CrossEncoder(cross_encoder_model)
    
    def rerank(self, query: str, results: List[Dict], top_k: int = 5) -> List[Dict]:
        pairs = [[query, r['content']] for r in results]
        
        scores = self.cross_encoder.predict(pairs)
        
        for i, score in enumerate(scores):
            results[i]['rerank_score'] = float(score)
        
        reranked = sorted(results, key=lambda x: x['rerank_score'], reverse=True)
        
        return reranked[:top_k]

from functools import lru_cache
import hashlib

class SearchCache:
    def __init__(self, max_size=1000):
        self.cache = {}
        self.max_size = max_size
    
    def _hash_query(self, query: str) -> str:
        return hashlib.md5(query.encode()).hexdigest()
    
    def get(self, query: str):
        key = self._hash_query(query)
        return self.cache.get(key)
    
    def set(self, query: str, results):
        if len(self.cache) >= self.max_size:
            oldest_key = next(iter(self.cache))
            del self.cache[oldest_key]
        
        key = self._hash_query(query)
        self.cache[key] = results
    
    def cached_search(self, query: str, search_fn):
        cached = self.get(query)
        if cached:
            return cached
        
        results = search_fn(query)
        self.set(query, results)
        return results