语义搜索入门：用 Hermes 搭配嵌入模型实现智能检索

关键词搜索的局限

传统的关键词搜索（比如 Elasticsearch 的 BM25）有一个根本性的问题：它只匹配字面文字，不理解语义。

用户搜”怎么解决内存溢出”，关键词搜索只能匹配包含”内存溢出”这几个字的文档。但”OOM 错误处理方法””Java 堆空间不足的解决方案”这些语义相关但用词不同的文档，关键词搜索就捞不出来。

语义搜索通过嵌入模型（Embedding Model）把文本转成高维向量，让意思相近的文本在向量空间中距离相近。这样搜索”内存溢出”就能匹配到”OOM 处理”——因为它们的向量距离足够近。

嵌入模型是怎么工作的

嵌入模型把一段文本映射成一个固定长度的浮点数组（向量）。关键性质：语义相似的文本，其向量的余弦相似度更高。

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("BAAI/bge-m3")

texts = [
    "怎么解决 Python 内存溢出",
    "Python OOM 错误处理方法",
    "今天天气真好",
]

embeddings = model.encode(texts, normalize_embeddings=True)

sim_01 = np.dot(embeddings[0], embeddings[1])  # 高相似度
sim_02 = np.dot(embeddings[0], embeddings[2])  # 低相似度

print(f"'内存溢出' vs 'OOM错误': {sim_01:.4f}")
print(f"'内存溢出' vs '天气真好': {sim_02:.4f}")

嵌入模型选型

BGE 系列（中文场景首选）

# BGE-M3: 中英双语最佳选择
model = SentenceTransformer("BAAI/bge-m3")

# BGE 系列在做检索时，建议对查询加前缀
query_embedding = model.encode(
    "为这个句子生成表示以用于检索相关文章：怎么解决Python内存溢出",
    normalize_embeddings=True
)

选型对比

模型	维度	大小	中文效果	最大长度	推荐场景
bge-m3	1024	2.2GB	优秀	8192	中英混合，长文本
bge-large-zh-v1.5	1024	1.3GB	优秀	512	纯中文
bge-small-zh-v1.5	512	96MB	良好	512	资源受限
multilingual-e5-large	1024	1.1GB	良好	512	多语言

如果不知道选什么，直接选 bge-m3。

向量数据库选型和使用

ChromaDB（入门首选）

import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path="./chroma_data")

embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-m3", device="cuda"
)

collection = client.get_or_create_collection(
    name="knowledge_base",
    embedding_function=embedding_fn,
    metadata={"hnsw:space": "cosine"}
)

# 插入文档
collection.add(
    documents=["Python 的垃圾回收机制基于引用计数", "Go 语言使用三色标记法"],
    ids=["doc_0", "doc_1"],
    metadatas=[{"topic": "memory"}, {"topic": "memory"}]
)

# 搜索
results = collection.query(query_texts=["怎么解决内存泄漏"], n_results=3)

Milvus（生产级）

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

connections.connect("default", host="localhost", port="19530")

fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2000),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024)
]

schema = CollectionSchema(fields, description="知识库")
collection = Collection("knowledge_base", schema)

index_params = {"metric_type": "COSINE", "index_type": "HNSW", "params": {"M": 16, "efConstruction": 256}}
collection.create_index("embedding", index_params)

维度	ChromaDB	Milvus
部署	嵌入式，无需额外服务	需要单独部署
数据规模	万到十万级	百万到亿级
适合场景	原型开发、小项目	生产环境

完整搜索 Pipeline：检索 + 重排 + 生成

from sentence_transformers import SentenceTransformer, CrossEncoder
from llama_cpp import Llama


class SemanticSearchPipeline:
    def __init__(self, embedding_model="BAAI/bge-m3", reranker_model="BAAI/bge-reranker-v2-m3",
                 llm_path="./Hermes-3-Llama-3.1-8B-Q5_K_M.gguf", db_path="./chroma_data"):
        self.embedder = SentenceTransformer(embedding_model)
        self.reranker = CrossEncoder(reranker_model, max_length=512)
        self.llm = Llama(model_path=llm_path, n_ctx=8192, n_gpu_layers=-1, chat_format="chatml")

        client = chromadb.PersistentClient(path=db_path)
        self.collection = client.get_or_create_collection(name="search_index", metadata={"hnsw:space": "cosine"})

    def search(self, query, top_k=3, initial_k=20, generate_answer=True):
        # 阶段1: 粗筛
        query_embedding = self.embedder.encode(query, normalize_embeddings=True).tolist()
        candidates = self.collection.query(query_embeddings=[query_embedding], n_results=initial_k)

        if not candidates["documents"][0]:
            return {"answer": "没有找到相关内容。", "sources": []}

        # 阶段2: 精排
        pairs = [(query, text) for text in candidates["documents"][0]]
        rerank_scores = self.reranker.predict(pairs)

        ranked = sorted(
            zip(candidates["documents"][0], candidates["ids"][0], rerank_scores),
            key=lambda x: x[2], reverse=True
        )[:top_k]

        sources = [{"id": item[1], "text": item[0][:200], "score": float(item[2])} for item in ranked]

        if not generate_answer:
            return {"sources": sources}

        # 阶段3: 生成
        context = "\n\n---\n\n".join([item[0] for item in ranked])
        messages = [
            {"role": "system", "content": "基于参考资料回答问题。如果资料中没有相关信息，请明确说明。"},
            {"role": "user", "content": f"参考资料：\n{context}\n\n问题：{query}"}
        ]
        response = self.llm.create_chat_completion(messages=messages, max_tokens=1024, temperature=0.3)

        return {"answer": response["choices"][0]["message"]["content"], "sources": sources}

重排序为什么重要

向量检索用的是 Bi-Encoder（查询和文档分别独立编码），效率高但精度有限。重排用的是 Cross-Encoder（查询和文档拼在一起做注意力交互），精度更高但速度慢。

最佳实践：

Bi-Encoder 粗筛：从百万文档中取 top-20（毫秒级）
Cross-Encoder 精排：对 20 个候选精确排序，取 top-3（几十到几百毫秒）

加上重排后检索准确率通常能提升 10-20 个百分点。

混合检索：向量 + 关键词

纯向量检索对精确匹配不够好。混合检索把向量检索和 BM25 结合：

from rank_bm25 import BM25Okapi
import jieba

class HybridSearcher:
    def __init__(self, embedder, documents):
        self.embedder = embedder
        self.documents = documents
        tokenized = [list(jieba.cut(doc)) for doc in documents]
        self.bm25 = BM25Okapi(tokenized)
        self.doc_embeddings = embedder.encode(documents, normalize_embeddings=True)

    def search(self, query, top_k=5, vector_weight=0.6):
        keyword_weight = 1 - vector_weight

        query_tokens = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(query_tokens)
        if bm25_scores.max() > 0:
            bm25_scores = bm25_scores / bm25_scores.max()

        query_emb = self.embedder.encode(query, normalize_embeddings=True)
        vector_scores = np.dot(self.doc_embeddings, query_emb)
        vector_scores = (vector_scores + 1) / 2

        hybrid_scores = vector_weight * vector_scores + keyword_weight * bm25_scores
        top_indices = np.argsort(hybrid_scores)[::-1][:top_k]

        return [(idx, float(hybrid_scores[idx])) for idx in top_indices]

混合检索在 RAG 知识库系统中效果很明显。

性能优化

1. 批量编码

# 差：逐条编码
embeddings = [model.encode(text) for text in texts]

# 好：批量编码
embeddings = model.encode(texts, batch_size=64, normalize_embeddings=True)

2. GPU 加速 — 嵌入模型在 GPU 上编码速度比 CPU 快 10-20 倍。

3. 维度裁剪 — BGE-M3 完整 1024 维，可以只用前 512 维，检索质量只有很小下降但存储和计算量大幅减少。

4. 量化存储 — 向量从 float32 量化到 int8，存储缩小为 1/4。

把所有东西整合成一个可部署的服务，可以用 FastAPI 包装成 API。

语义搜索不是一个组件而是一个系统——嵌入模型选型、向量数据库部署、检索策略优化、重排模型配置、生成模型集成，每个环节都影响最终效果。建议从简单方案开始（ChromaDB + BGE），跑通后再逐步优化。cocoloop 社区里有不少实际项目经验可以参考。