Milvus + RAG 实战：构建电子书语义检索助手的架构设计与实现

1次阅读

没有评论

共计 2881 个字符，预计需要花费 8 分钟才能阅读完成。

在数字化阅读时代，电子书检索功能往往局限于简单的关键词匹配，导致用户体验不佳。本文将通过 Milvus 向量数据库和 RAG 技术，构建一个能够理解自然语言的电子书语义检索系统，显著提升检索准确率和使用体验。

传统的关键词检索方式存在几个主要问题：

同义词问题：无法识别不同词汇表达的相同含义。例如搜索 ” 汽车 ” 时，不会返回包含 ” 轿车 ” 或 ” 车辆 ” 的结果。
语义泛化不足：无法理解查询的深层含义。比如搜索 ” 如何治疗感冒 ”，可能不会返回 ” 感冒的药物治疗方案 ” 相关内容。
上下文缺失：无法根据上下文调整搜索结果，导致相关性差。

在选择向量数据库时，我们对比了几种主流方案：

FAISS：Facebook 开源的向量检索库，适合小规模部署，但缺乏分布式支持和生产级功能。
Pinecone：全托管服务，使用简单但成本较高，且定制化能力有限。
Milvus：专为向量搜索优化的开源数据库，支持分布式部署、多种索引类型和丰富的查询功能，是电子书场景的理想选择。

最终选择 Milvus 的原因在于其优秀的可扩展性、丰富的 API 支持以及活跃的社区生态。

我们使用 Sentence-Transformer 模型将文本转换为向量表示。这个预训练模型能够很好地捕捉句子级别的语义信息。

from sentence_transformers import SentenceTransformer

# 加载预训练模型
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

# 文本向量化
text = "如何学习机器学习"
embedding = model.encode(text)

在 Milvus 中，我们创建一个专门用于存储电子书内容的集合：

定义集合结构，包含 ID、文本内容和向量字段
选择 IVF_FLAT 索引类型，平衡查询速度和内存使用
根据数据规模设置适当的分片数

from pymilvus import CollectionSchema, FieldSchema, DataType

# 定义字段
book_id = FieldSchema(name="book_id", dtype=DataType.INT64, is_primary=True)
text_content = FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=1000)
embedding = FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384)

# 创建集合模式
schema = CollectionSchema(fields=[book_id, text_content, embedding], description="Ebook search collection")

检索增强生成 (RAG) 将 Milvus 的语义检索结果与大语言模型结合：

用户输入查询
系统检索最相关的电子书片段
将检索结果作为上下文输入 LLM 生成最终回答

# 伪代码示例
def rag_query(user_query):
    # 向量化查询
    query_embedding = model.encode(user_query)

    # Milvus 语义检索
    search_results = collection.search(query_embedding, top_k=3)

    # 构建 LLM 提示
    context = "\n".join([result.content for result in search_results])
    prompt = f"根据以下上下文回答用户问题:{context}\n 问题:{user_query}"

    # 调用 LLM 生成回答
    response = llm.generate(prompt)
    return response

以下是关键功能的完整实现：

import logging
from pymilvus import connections, Collection, utility
from sentence_transformers import SentenceTransformer

# 初始化
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class EbookSearchEngine:
    def __init__(self):
        # 连接 Milvus
        connections.connect("default", host="localhost", port="19530")

        # 加载模型
        self.model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

        # 初始化集合
        self.collection = Collection("ebook_search")

    def search(self, query: str, top_k: int = 5):
        try:
            # 向量化查询
            query_embedding = self.model.encode(query)

            # 搜索参数
            search_params = {
                "metric_type": "L2",
                "params": {"nprobe": 10}
            }

            # 执行搜索
            results = self.collection.search(data=[query_embedding],
                anns_field="embedding",
                param=search_params,
                limit=top_k,
                output_fields=["content"]
            )

            return [hit.entity.get("content") for hit in results[0]]
        except Exception as e:
            logger.error(f"Search failed: {str(e)}")
            return []

    # 其他方法如插入数据、创建集合等...

我们在 10 万条电子书章节数据上进行了测试：