4.2 语义搜索与推荐系统实战

4.2 语义搜索与推荐系统实战

本节导读：将Embedding技术应用于语义搜索和推荐系统，从理论到工程实现，掌握完整的构建流程

学习目标

• 理解语义搜索与关键词搜索的本质区别
• 掌握基于向量相似度的搜索实现方案
• 学会构建个性化推荐系统的核心技术
• 了解混合检索策略和性能优化方法
• 掌握常见业务场景的最佳实践

核心概念

什么是语义搜索

传统的关键词搜索基于词汇匹配（如BM25），只能找到包含相同关键词的文档。而语义搜索利用Embedding将查询和文档映射到同一向量空间，通过计算向量相似度来匹配语义相关的内容，即使查询和文档没有共同词汇也能找到相关结果。

关键词搜索 vs 语义搜索对比：

向量检索的核心流程

完整的语义搜索系统包含以下步骤：

1. 文档入库：将文档通过Embedding模型转换为向量，存入向量数据库
2. 查询处理：将用户查询通过同一Embedding模型转换为向量
3. 相似度计算：计算查询向量与文档向量的相似度（通常使用余弦相似度）
4. 结果排序：按相似度排序返回Top-K结果

语义搜索系统实现

4.2.1 基础语义搜索引擎

技术选型：

• Embedding模型：选择sentence-transformers系列模型（如all-MiniLM-L6-v2）
• 向量存储：FAISS、Milvus或ChromaDB
• 后端框架：FastAPI

核心代码示例：

from sentence_transformers import SentenceTransformer
from typing import List, Tuple
import numpy as np

class SemanticSearchEngine:
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        """初始化语义搜索引擎"""
        self.model = SentenceTransformer(model_name)
        self.documents: List[str] = []
        self.embeddings: np.ndarray = None

    def index_documents(self, documents: List[str]):
        """索引文档集合"""
        self.documents = documents
        self.embeddings = self.model.encode(
            documents,
            convert_to_numpy=True,
            normalize_embeddings=True,
            show_progress_bar=True
        )

    def search(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
        """执行语义搜索"""
        query_embedding = self.model.encode(
            [query],
            convert_to_numpy=True,
            normalize_embeddings=True
        )[0]

        # 计算余弦相似度
        similarities = np.dot(self.embeddings, query_embedding)

        # 获取Top-K结果
        top_indices = np.argsort(similarities)[::-1][:top_k]
        results = [(self.documents[i], float(similarities[i])) for i in top_indices]
        return results

# 使用示例
engine = SemanticSearchEngine()
docs = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络处理数据",
    "自然语言处理研究计算机理解人类语言",
    "计算机视觉让机器能够理解图像"
]
engine.index_documents(docs)
results = engine.search("AI怎么理解人类说的话？", top_k=2)
for doc, score in results:
    print(f"[相似度: {score:.4f}] {doc}")
​

4.2.2 向量数据库集成

FAISS集成方案：

import faiss

class FAISSSearchEngine:
    def __init__(self, model_name: str = "all-MiniLM-L6-v2", dimension: int = 384):
        self.model = SentenceTransformer(model_name)
        self.dimension = dimension
        # 使用内积（已归一化的余弦相似度等价于内积）
        self.index = faiss.IndexFlatIP(dimension)
        self.documents: List[str] = []

    def index_documents(self, documents: List[str]):
        self.documents = documents
        embeddings = self.model.encode(
            documents,
            convert_to_numpy=True,
            normalize_embeddings=True
        )
        self.index.add(embeddings.astype('float32'))

    def search(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
        query_embedding = self.model.encode([query], normalize_embeddings=True)
        distances, indices = self.index.search(query_embedding.astype('float32'), top_k)
        results = [(self.documents[i], float(d)) for i, d in zip(indices[0], distances[0]) if i >= 0]
        return results
​

ChromaDB集成方案：

import chromadb
from chromadb.config import Settings

class ChromaSearchEngine:
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
        self.client = chromadb.Client(Settings(
            chroma_db_impl="duckdb+parquet",
            persist_directory="./chroma_db"
        ))
        self.collection = self.client.get_or_create_collection(
            name="semantic_search",
            metadata={"hnsw:space": "cosine"}
        )

    def index_documents(self, documents: List[str], ids: List[str] = None):
        embeddings = self.model.encode(documents).tolist()
        if ids is None:
            ids = [f"doc_{i}" for i in range(len(documents))]
        self.collection.add(
            embeddings=embeddings,
            documents=documents,
            ids=ids
        )

    def search(self, query: str, top_k: int = 5):
        query_embedding = self.model.encode([query]).tolist()
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=top_k
        )
        return list(zip(results['documents'][0], results['distances'][0]))
​

4.2.3 混合检索策略

纯语义搜索在某些场景下可能不如关键词搜索精确（如产品编号、专有名词搜索）。混合检索结合了两者的优势：

from rank_bm25 import BM25Okapi
import jieba

class HybridSearchEngine:
    def __init__(self, model_name: str = "all-MiniLM-L6-v2", alpha: float = 0.7):
        """
        alpha: 语义搜索权重 (0-1), 关键词搜索权重为 1-alpha
        """
        self.semantic_engine = FAISSSearchEngine(model_name)
        self.alpha = alpha

    def index_documents(self, documents: List[str]):
        self.semantic_engine.index_documents(documents)
        # BM25需要分词后的文档
        tokenized = [list(jieba.cut(doc)) for doc in documents]
        self.bm25 = BM25Okapi(tokenized)

    def search(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
        # 语义搜索结果
        semantic_results = self.semantic_engine.search(query, top_k=top_k * 2)
        # BM25搜索结果
        tokenized_query = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # 归一化并融合分数
        all_docs = self.semantic_engine.documents
        combined_scores = {}
        for doc, sem_score in semantic_results:
            combined_scores[doc] = self.alpha * sem_score
        for i, doc in enumerate(all_docs):
            if doc in combined_scores:
                combined_scores[doc] += (1 - self.alpha) * bm25_scores[i]
            else:
                combined_scores[doc] = (1 - self.alpha) * bm25_scores[i]

        # 按融合分数排序
        sorted_results = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_results[:top_k]
​

推荐系统实现

4.2.4 基于Embedding的推荐原理

Embedding推荐系统的核心思想：将用户和物品映射到同一向量空间，通过向量相似度来衡量用户偏好与物品特征之间的匹配程度。

常见推荐范式：

• 内容推荐：基于物品内容的Embedding相似度
• 协同过滤推荐：基于用户行为矩阵分解得到的Embedding
• 混合推荐：融合内容和行为信息

4.2.5 内容推荐实现

class ContentBasedRecommender:
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
        self.items: List[dict] = []
        self.item_embeddings: np.ndarray = None

    def index_items(self, items: List[dict], text_field: str = "description"):
        """索引商品/内容"""
        self.items = items
        texts = [item[text_field] for item in items]
        self.item_embeddings = self.model.encode(
            texts,
            convert_to_numpy=True,
            normalize_embeddings=True
        )

    def recommend(self, user_preferences: List[str], top_k: int = 10) -> List[Tuple[dict, float]]:
        """
        基于用户偏好历史推荐
        user_preferences: 用户过去喜欢的物品文本描述
        """
        # 用用户偏好历史构建用户向量
        pref_embeddings = self.model.encode(user_preferences, normalize_embeddings=True)
        user_vector = np.mean(pref_embeddings, axis=0)
        user_vector = user_vector / np.linalg.norm(user_vector)

        # 计算用户向量与所有物品的相似度
        similarities = np.dot(self.item_embeddings, user_vector)
        top_indices = np.argsort(similarities)[::-1][:top_k]
        return [(self.items[i], float(similarities[i])) for i in top_indices]

# 使用示例
recommender = ContentBasedRecommender()
items = [
    {"id": 1, "name": "深度学习入门", "description": "神经网络和深度学习的基础教程"},
    {"id": 2, "name": "机器学习实战", "description": "使用Python进行机器学习项目开发"},
    {"id": 3, "name": "烹饪百科", "description": "世界各地的美食做法和烹饪技巧"},
    {"id": 4, "name": "NLP技术指南", "description": "自然语言处理的核心技术和应用实践"},
]
recommender.index_items(items)
user_history = ["深度学习入门"]
recs = recommender.recommend(user_history, top_k=3)
for item, score in recs:
    print(f"[匹配度: {score:.4f}] {item['name']} - {item['description']}")
​

4.2.6 双塔模型推荐

双塔模型是工业界常用的推荐架构，用户和物品各通过一个独立的编码塔得到Embedding：

import torch
import torch.nn as nn

class TwoTowerModel(nn.Module):
    def __init__(self, vocab_size: int, embed_dim: int = 128, hidden_dim: int = 256):
        super().__init__()
        # 用户塔
        self.user_tower = nn.Sequential(
            nn.Embedding(vocab_size, embed_dim),
            nn.Linear(embed_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, embed_dim),
        )
        # 物品塔
        self.item_tower = nn.Sequential(
            nn.Embedding(vocab_size, embed_dim),
            nn.Linear(embed_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, embed_dim),
        )

    def forward(self, user_ids, item_ids):
        user_embeds = self.user_tower(user_ids)
        item_embeds = self.item_tower(item_ids)
        # L2归一化
        user_embeds = nn.functional.normalize(user_embeds, p=2, dim=-1)
        item_embeds = nn.functional.normalize(item_embeds, p=2, dim=-1)
        return user_embeds, item_embeds
​

性能优化与工程实践

4.2.7 大规模搜索优化

优化策略：

1. 近似最近邻（ANN）：使用HNSW、IVF等算法替代暴力搜索
2. 批量编码：将文档批量通过模型，减少GPU调用次数
3. 缓存层：对热门查询结果进行缓存
4. 异步索引：新文档异步编码入库，不影响在线服务

FAISS IVF索引优化：

import faiss

class OptimizedFAISSSearch:
    def __init__(self, dimension: int, nlist: int = 100):
        """
        nlist: 聚类中心数量，越大越精确但越慢
        """
        quantizer = faiss.IndexFlatIP(dimension)
        self.index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT)

    def train_and_add(self, embeddings: np.ndarray):
        """训练聚类中心并添加向量"""
        self.index.train(embeddings)
        self.index.add(embeddings)

    def search(self, query: np.ndarray, top_k: int, nprobe: int = 10):
        """
        nprobe: 搜索时查询的聚类数量，越大越精确
        """
        return self.index.search(query, top_k, nprobe=nprobe)
​

4.2.8 常见问题与解答

Q：语义搜索什么时候不如关键词搜索？
A：当搜索目标包含精确的专业术语、产品编号、人名等低频词时，语义搜索可能因为这些词在训练语料中出现频率低而导致Embedding不够精确。此时混合检索是最佳选择。

Q：如何选择合适的Embedding模型？
A：选择模型时考虑以下因素：语言（中文场景选择支持中文的模型）、模型大小（影响推理速度）、训练数据领域（通用场景或特定领域）、性能基准（在MTEB等榜单上的表现）。

Q：向量数据库如何选型？
A：小规模（<100万）用FAISS或ChromaDB；中等规模（100万-1亿）用Milvus或Weaviate；超大规模（>1亿）需要分布式方案如Milvus集群或Elasticsearch的稠密向量插件。

Q：如何处理长文档的搜索？
A：将长文档切分为段落或句子，对每个片段独立编码。搜索时返回最相关的片段，同时返回所属文档信息。

本节小结

本节完整介绍了基于Embedding的语义搜索和推荐系统的实现方案。从基础的语义搜索引擎构建，到向量数据库集成和混合检索策略，再到推荐系统的双塔模型，覆盖了从原型到生产的全流程。通过具体的代码示例，开发者可以直接复用这些模式来构建自己的搜索和推荐系统。

延伸阅读

• 论文：DPR（Dense Passage Retrieval）—— 密集段落检索经典方法
• 工具：sentence-transformers库官方文档和模型库
• 实践：FAISS官方教程中的索引优化策略

关键词：语义搜索, 向量检索, 推荐系统, 混合检索, FAISS, 双塔模型, BM25, ChromaDB
难度：中级
预计阅读：45分钟