语义搜索

核心结论

1. 语义搜索通过将文本映射为向量并计算相似度来召回结果，能有效处理同义词、查询意图和长尾自然语言问题，但可能漏掉精确词匹配，因此不能完全替代关键词检索。

2. 混合搜索将 BM25一种基于概率信息检索理论的文本检索模型，通过词频和逆文档频率对文档进行排名，考虑词项的相关性和稀有性。 词法分数与向量语义分数归一化后加权融合（如语义权重设为 0.5），能兼顾精确匹配和语义召回，但会带来更高的延迟、成本和调优复杂度。

3. 嵌入模型必须按语言、领域、隐私限制、延迟和预算选择，不能只看精度。例如 all-MiniLM-L6-v2 维度为 384，all-mpnet-base-v2 与 BGE-base-zh-v1.5 为 768，BGE-M3 为 1024，中文场景优先考虑 BGE 系列，多语言场景优先考虑 BGE-M3。

4. 语义搜索评测至少应覆盖 Recall@K、Precision@K、MRR、nDCG、无结果率、延迟分位数、索引体积和成本这八类指标，并且错误必须按查询类型、语言、租户、文档年龄和受保护内容切分分析，不能只看平均分。

5. 授权与生命周期过滤必须在检索和排序之前独立执行并单独验证，因为“相关”不等于“有权访问”，也不能用检索排序结果掩盖权限过滤的问题。

先说结论：这篇文章最打动我的地方，是它没有把语义搜索捧成万能药。它把边界划得很清楚：什么时候用词法，什么时候用向量，什么时候值得上混合方案。这种务实的态度在教程里很少见。

另外，它给出的评测思路值得认真读两遍。先定义怎么算相关，再去比方案，有标注数据的人会少走很多弯路。如果你的系统已经跑起来了，这篇文章也能提供一份现成的优化检查单。

核心摘要

• 语义搜索靠语义理解而不是关键词重合来召回，能抓住用户真正想找什么
• 核心流程是Embedding模型把文本变向量，再用向量相似度做匹配
• 嵌入模型得按语言、领域、隐私限制、延迟和预算来选，不能只看精度
• 搜索策略用标注好的查询去比纯关键词、纯语义和混合检索，谁赢用谁
• 可靠性优先：授权和时效性过滤先做，别让检索和排序质量混在一起评估

什么是语义搜索

语义搜索（Semantic Search基于语义理解的搜索技术，通过向量化表示和相似度计算理解查询意图，而非仅靠关键词匹配。）是建立在自然语言理解上的搜索技术。它不只看关键词，还能理解查询背后的意图和上下文。

语义搜索 vs 关键词搜索

我们拿“如何提升代码质量”这个查询来说。

关键词搜索会提取“代码”和“质量”两个词，去文档库里做精确匹配，结果只包含同时出现这两个词的文档。但一篇讲“代码审查最佳实践”的文章可能完全不含“代码质量”这个短语，就被漏掉了。

语义搜索把查询变成向量，在向量空间里找语义相近的文档。“代码审查最佳实践”和“代码质量”在语义上很近，所以能被召回。

语义搜索评测清单

评测不是最后补一步，应该从构建一开始就建立。以下是需要覆盖的几件事：

把代表性查询集和相关性标注做成版本控制的一部分。高频、长尾、精确词、多语言、过期文档、无答案、权限边界，这些类型都要有。

在同一份语料快照上跑纯词法、纯语义和混合基线，保证可对比。

指标至少包含 Recall@K、Precision@K、MRR、nDCG、无结果率、延迟分位数、索引体积和成本。

错误按查询类型、语言、租户、文档年龄和受保护内容切分来看。一个平均分盖掉的问题，往往才是系统最致命的。

授权和生命周期过滤单独验证。相关不等于有权访问。

Embedding模型、预处理、分块、语料、ANN参数、过滤、融合或重排，任何一项变化后都重新回归。

语义搜索能解决什么问题

问题1：同义词和近义词

搜“汽车”的时候，关键词搜不到含“轿车”和“机动车”的文档。语义搜索知道这几个词是同一个概念。

问题2：查询意图理解

用户搜“Python处理Excel”，想要的可能是pandas或openpyxl的使用教程。“处理”这个词太泛，关键词系统给不出好结果。

问题3：长尾查询

“为什么我的程序运行很慢”这种自然语言问法，关键词搜索几乎是废的。语义搜索能理解这是在问性能问题。

语义搜索的工作原理

语义搜索的核心就是把文本变成向量。语义相近的文本在向量空间里距离近，检索时用相似度衡量相关程度。

核心流程

索引阶段：文档集合先做文本分块，每个块经Embedding模型生成向量，存入向量数据库。

查询阶段：用户查询经过同一个Embedding模型生成查询向量，然后在向量数据库里做相似度搜索，返回排序结果。

关键步骤详解

1. 文本向量化（Embedding）

Embedding模型把一段文本映射到高维空间的一组数值。这组数值就是它的语义指纹。

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}")

2. 向量相似度计算

最常用的是余弦相似度，它只看两个向量的方向是否一致，不受向量长度影响。

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}")

3. 向量索引与检索

海量数据不能全量算一遍相似度，要用HNSWHierarchical Navigable Small World - a graph-based algorithm for approximate nearest neighbor search that provides efficient similarity search in high-dimensional spaces.、IVF这类近似最近邻索引来加速。它把搜索范围缩小到部分候选集，精度有轻微损失，但换来了速度。

嵌入模型选择指南

嵌入模型是语义搜索的地基。选错模型，后面的一切调优都是白费。

主流嵌入模型对比

模型选择决策树

主要业务语言是什么？如果只用英文，看性能要求。追求轻量就选all-MiniLM-L6-v2，要均衡选all-mpnet-base-v2，要精度就考虑API模型。预算够就上text-embedding-3-large，成本敏感用small版本。中文业务直接看BGE系列，base-zh-v1.5对标中文场景已经够用。多语言场景BGE-M3是稳的选择。

本地模型 vs API模型

向量相似度计算详解

余弦相似度

余弦相似度衡量两个向量的方向差异，数值范围在 -1 到 1 之间，越大表示越相似。它是最适合文本语义场景的度量方式。

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

选择哪种度量方式

语义搜索 vs 全文搜索 vs 混合搜索

三种搜索方式对比

全文搜索靠BM25这类算法做词法匹配，速度快、精确，但理解不了语义。语义搜索懂同义词，能处理改写，却可能把精确匹配的文档漏掉。混合搜索两者结合，复杂度也上去了。

混合搜索实现

把BM25的分数和语义相似度的分数做归一化，再加权融合。实现不复杂，但权重怎么调才是学问。

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']}")

构建语义搜索系统实战

完整的语义搜索系统

这一节用一个完整的Python类把前面讲的东西串起来，代码可直接用于中小规模的个人项目或Demo。

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']}")

文本分块策略

长文档不能直接喂给Embedding模型，先分块再索引。分块做得好不好，直接决定检索效果。

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]}...")

优化语义搜索的技巧

1. 查询优化

查询改写和多查询扩展能显著提升召回。把口语化的词替换成更书面的说法，或者扩展出多个同义查询，再合并结果。

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]

2. 结果重排序

用CrossEncoder对召回的结果做二次排序。双塔模型负责召回，CrossEncoder负责精排，两者搭配效果最好。

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]

3. 缓存策略

高频查询的缓存能省下大量重复计算。注意缓存容量别无限膨胀，超出上限要淘汰旧条目。

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

常见问题

语义搜索和向量搜索是一回事吗？

不完全是一回事。向量搜索是在向量空间里做近邻检索，向量可以表示文本、图像、用户、商品或任何特征。语义搜索是按含义检索的产品目标，通常还需要文档处理、过滤、词法信号和排序。只建一个向量索引，离完整的语义搜索系统还差得远。

如何评估语义搜索的效果？

常用指标有四个：召回率（Recall@K）表示相关文档被检出来的比例；精确率（Precision@K）表示返回结果里相关文档占多少；MRR看第一个相关结果排在哪；NDCG则综合考虑排序位置。关键是要有一份标注好的评测集，没有标注就没有客观评估。

语义搜索适合所有场景吗？

看情况。精确ID、引号短语、代码符号、新名称或严格延迟预算，词法检索更稳。改写和概念查询才值得上语义。混合搜索只是候选方案，用标注评测去验证它是否真的同时优于两种单路方案，否则白搭。

如何处理语义搜索中的冷启动问题？

新领域没有标注数据很正常。先用预训练的通用嵌入模型跑起来，再用小批量领域数据微调，同时保留关键词搜索作为兜底。线上跑起来之后，慢慢用用户反馈来迭代。

语义搜索的延迟如何优化？

轻量模型优先，比如all-MiniLM-L6-v2。向量索引要调HNSW参数，通常能换来几个数量级的提速。高频查询做缓存，批量请求合并处理。有GPU的话把Embedding计算放上去，预计算热门查询的结果也管用。

总结

语义搜索能帮你做出理解用户意图的检索系统，但它有清晰的边界。词法检索和向量检索各有各的位置，混合方案要经过评测验证，这些都要在选型阶段想清楚。

关键要点回顾

• 语义搜索靠向量相似度，能处理同义词和查询意图
• 嵌入模型的选择要在语言、性能和成本之间权衡
• 混合搜索只有在标注工作负载上胜过单路方案时才值得引入
• 文本分块、查询优化、结果重排序是提升质量的关键动作
• 专用向量数据库不是必需品，现有搜索引擎或数据库的扩展能力可能已经够用

延伸阅读

• 向量嵌入完全指南 - 深入理解Embedding技术
• 向量数据库完全指南 - 选择合适的向量存储方案
• RAG检索增强生成指南 - 构建基于检索的AI应用

常见问题（FAQ）

语义搜索和关键词搜索有什么区别？

关键词搜索依赖字面匹配，会漏掉同义词或意图改写；语义搜索通过理解查询意图和上下文，将文本转为向量做相似度匹配，能召回语义相近但字面不同的内容，两者可互补，混合检索效果更稳。

怎么选合适的嵌入模型？

按业务语言、领域、隐私、延迟和预算选。英文轻量用MiniLM，均衡用mpnet；中文用BGE；多语言用BGE-M3或API。本地模型保隐私，API模型省心但数据外发。别只看精度，要结合场景权衡。

语义搜索效果怎么评测？

用带标注的查询集，覆盖高频、长尾、精确词等类型。在同一语料上对比纯词法、纯语义和混合基线，看Recall@K、Precision@K、MRR等指标。错误按查询类型、语言等切分，避免平均分掩盖问题。