一、意义

        你有没有过这样的经历：项目上线前夜，突然报了一个告警，你印象中之前有人写过一份处理手册，但死活找不到在哪。翻了半小时的飞书文档、翻了聊天记录、翻了内部Wiki，最后发现那份文档藏在某个三级目录下的一个不起眼的角落里…
        再比如，新同事入职了，问你“咱们这个API的鉴权流程是什么？“，你说”文档里有”，但他可能花了一下午，都没找到那份写得最全的文档。
        这就是传统文档管理的痛点——信息孤岛。
        什么意思呢？就是说，你的知识明明存在，但它们散落在各个角落，像一座座孤岛，彼此之间没有桥梁相连。你知道答案就在某个地方，但就是够不着。所以，知识库的出现，本质上就是要解决一个问题：把“大海捞针”变成”秒级找到”。

二、核心流程

2.1、文档拆分

        你上传了份100页的PDF技术手册，AI能直接读懂吗？答案是：不能，至少不能一口吞下去。
这就像你准备考试，你不会把一整本教材从头到尾读一遍就完事了吧？你会按章节、按知识点去拆分，做成一张张笔记卡片，对不对？
        知识库做的也是同样的事情。它会自动把你上传的文档，按照章节、段落、甚至自定义的规则进行拆分。而且，它还会智能适配不同的文档格式——PDF有PDF的拆法，Markdown有Markdown的拆法。
        这一步的关键在于：每个拆出来的片段，都要包含完整的语义信息。

        什么意思？举个例子，假设文档里有一段话："当API返回401错误码时，通常是因为Token过期。解决方案是重新调用/auth/refresh接口获取新Token。
        如果拆分的时候，把"解决方案是重新调用/auth/refresh接口"和前面的"API返回401错误码“拆到了两个不同的片段里，那后续AI检索的时候，就可能只找到问题描述，却找不到解决方案。这就是”语义信息不完整”带来的问题。

2.2、文本向量化

        拆分完成后，每个片段还是人类能读懂的文字。但问题是，AI不是用”读文字”的方式来理解内容的，它需要的是数学语言。这一步，就是调用一个叫 Embedding模型 的东西，把每个文本片段转化为一组高维向量。等等，“高维向量”是什么？别慌，咱们用一个通俗的类比来理解。
        你可以把每个文本片段想象成一个人。每个人都有很多特征：身高、体重、年龄、爱好等等。如果我们用一组数字来描述一个人，比如[175，70，25，...]，那这组数字就是这个人的”向量表示”。同样的道理，Embedding模型 会根据文本的语义，给每个片段生成一组数字（通常是几百到上千维的）。语义越相近的文本，它们的向量在数学空间中就越”靠近”。
举个例子：
        ·“Token过期了怎么办"和“鉴权失败的处理方法"，虽然字面上长得不一样，但语义是相近的，所以它们的向量会很接近。
        ·"Token过期了怎么办"和"今天中午吃什么"，语义完全不相关，向量就会离得很远。
        这就是向量化的魔力——它让AI能理解”意思”，而不仅仅是匹配”关键词”。这也是为什么知识库比传统的关键词搜索强大得多的原因。传统搜索你搜”鉴权失败”，可能搜不到标题叫"Token过期处理"的文档。但基于向量的检索，就能把它们关联起来。

2.3、向量库存储

        向量生成好了，总得有个地方存起来吧？这就是向量数据库的职责。知识库会把生成的向量，连同它的 元信息（比如：这段内容来自哪份文档、属于第几章、作者是谁、最后更新时间是什么），一起批量存入向量数据库中。
        为什么要存元信息？两个原因：
        1．方便溯源：当AI给你一个回答的时候，你可以看到”这个答案来自《故障处理手册》第3章”，这样你就能判断答案靠不靠谱，而不是盲目信任AI。
        2．支持过滤：比如你只想搜索最近一个月更新的文档，或者只搜索某个特定作者写的内容，有了元信息就能轻松做到。

三、过程解析

3.1、提问前的数据准备

3.1.1、分片

        想象一下，你有一本1000页的产品手册。如果有人问你”产品A的退货政策是什么"，你不会把整本书递给他，而是会翻到退货政策那一页，把那一段内容指给他看。 分片做的就是这件事——提前把文档切成一段一段的小片段，每个片段聚焦一个具体的知识点。
        常见的分片方式有：
        ·按固定字数切：比如每1000个字切一段
        ·按段落切：以自然段落为单位
        ·按章节/标题切：按文档本身的结构来
        ·按页码切：一页一个片段

        有一个非常重要的原则：每个片段的语义要完整。

3.1.2、Embedding

        片段切好了，但计算机不认识中文啊。你跟它说”退货政策”，它只看到一堆字符编码，完全不理解这几个字是什么意思。所以我们需要一种方式，把人类的语言翻译成计算机能理解的”数学语言”。这就是Embedding（向量化）的作用。

        Embedding就是把一段文字变成一个向量的过程。意思相近的文字，转出来的向量也会很相近。
        举个例子：
        "小林写Python" → [0.80,0.20，-0.50]
        "小林写Golang" → [0.78,0.22，-0.48]
        “今天天气真好” → [0.10,10.85，0.30]

3.1.3、存储

        向量生成好了，我们需要一个地方来存放它们。这个地方就是向量数据库（比如Milvus）。
向量数据库存的不只是向量，而是向量+原始文本，两个一起存。
        为什么要存原始文本？因为向量只是用来做相似度计算的”索引”，最终我们要拿给大模型看的，还是原始的文字内容。

        向量数据库中的每条数据大概长这样：

{
    "content"："产品A支持7天无理由退货，拆封后的电子产品除外。",
    "vector"：[θ.12，0.34，-0.56，θ.78，...]
}

3.2、提问后的回答生成

3.2.1、召回

        1. 先把用户的问题也通过Embedding模型转成向量（用的是同一个Embedding模型，这样才能保证在同一个“语义空间”里比较）
        2. 拿这个问题向量，去向量数据库里找最相似的片段，挑出相似度最高的TopN个（比如10个）。

        其中，向量相似度的计算方法有两种：余弦相似度和欧氏距离

3.2.1.1、余弦相似度

        计算两个向量夹角的余弦值，结果在-1到1之间。夹角越小，余弦值越接近1，说明两个向量方向越一致，语义越相近。它只看方向，不看长度，所以特别适合文本语义匹配（因为我们关心的是”意思像不像”，不关心”文本长不长")

3.2.1.2、欧氏距离

        就是两个点之间的直线距离，书距离越小说明越相似。它既看方向也看长度，适合需要考虑数值大小的场景

3.2.2、重排

        召回拿到了10个片段，但这里面可能有些是“擦边球” —— 跟问题沾点边但不太相关。我们需要进一步精筛。这时候登场的是重排模型（Cross Encoder），就是选出三个关系最大的。

        召回阶段是把问题和片段分别转成向量，然后算向量之间的距离（所以叫Bi-Encoder，双编码器）
        重排阶段是把问题和片段拼在一起输入模型，让模型直接判断”这两段话到底有多相关”（所以叫Cross Encoder，交叉编码器）

3.2.3、生成

        现在我们手里有了3个最相关的知识片段，最后一步就是把它们和用户的问题一起交给大模型，让它生成最终的回答。

最后发给大模型的内容就是这样的：

请根据以下参考资料回答用户的问题。

参考资料：
1.产品A支持7天无理由退货，拆封后的电子产品除外···
2.退货流程：登录官网→我的订单→申请退货..·
3.退货运费由买家承担，质量问题除外...

用户问题：产品A怎么退货？

四、代码实战

4.1、分片

        IndexingResult 类用于存储和管理文件索引任务的结果信息，提供了一系列方法来记录任务状态、统计数据和错误信息。

# 存储索引任务结果
class IndexingResult:
    """索引结果类"""

    def __init__(self):
        self.success = False               # 任务整体是否成功（默认失败）
        self.directory_path = ""           # 被索引的目录路径
        self.total_files = 0               # 总文件数
        self.success_count = 0             # 成功处理的文件数
        self.fail_count = 0                # 处理失败的文件数
        self.start_time: Optional[datetime] = None  # 任务开始时间
        self.end_time: Optional[datetime] = None    # 任务结束时间
        self.error_message = ""            # 整体错误信息
        self.failed_files: Dict[str, str] = {}  # 失败文件：路径 → 错误原因

    def increment_success_count(self):
        """增加成功计数"""
        self.success_count += 1

    def increment_fail_count(self):
        """增加失败计数"""
        self.fail_count += 1

    def add_failed_file(self, file_path: str, error: str):
        """添加失败文件"""
        self.failed_files[file_path] = error

    def get_duration_ms(self) -> int:
        """获取耗时（毫秒）"""
        if self.start_time and self.end_time:
            return int((self.end_time - self.start_time).total_seconds() * 1000)
        return 0

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "success": self.success,
            "directory_path": self.directory_path,
            "total_files": self.total_files,
            "success_count": self.success_count,
            "fail_count": self.fail_count,
            "duration_ms": self.get_duration_ms(),
            "error_message": self.error_message,
            "failed_files": self.failed_files,
        }

        VectorIndexService 类是一个用于处理文件索引的核心模块，主要功能包括目录扫描、文件读取、文本分割和向量存储。该服务采用模块化设计，便于集成到 RAG 系统中。

# 整个 RAG 系统（检索增强生成）里最核心的文件处理模块
class VectorIndexService:
    """向量索引服务 - 负责读取文件、生成向量、存储到 Milvus"""

    def __init__(self):
        """初始化向量索引服务"""
        self.upload_path = "./uploads"  # 默认文件存放目录
        logger.info("向量索引服务初始化完成")

    def index_directory(self, directory_path: Optional[str] = None) -> IndexingResult:
        """
        索引指定目录下的所有文件

        Args:(是 Arguments(参数)的缩写)
            directory_path: 目录路径（可选，默认使用配置的上传目录）

        Returns:
            IndexingResult: 索引结果
        """
        result = IndexingResult()  # 创建结果对象
        result.start_time = datetime.now()  # 记录开始时间

        try:
            # 使用指定目录或默认上传目录
            target_path = directory_path if directory_path else self.upload_path
            dir_path = Path(target_path).resolve()  # 把路径变成标准路径（绝对路径）

            if not dir_path.exists() or not dir_path.is_dir():
                raise ValueError(f"目录不存在或不是有效目录: {target_path}")

            result.directory_path = str(dir_path)

            # 获取所有支持的文件
            files = list(dir_path.glob("*.txt")) + list(dir_path.glob("*.md"))

            if not files:
                logger.warning(f"目录中没有找到支持的文件: {target_path}")
                result.total_files = 0
                result.success = True  # 没文件不算失败
                result.end_time = datetime.now()
                return result

            result.total_files = len(files)
            logger.info(f"开始索引目录: {target_path}, 找到 {len(files)} 个文件")

            # 遍历并索引每个文件
            for file_path in files:
                try:
                    self.index_single_file(str(file_path))  # 处理单个文件
                    result.increment_success_count()
                    logger.info(f"✓ 文件索引成功: {file_path.name}")  # 成功：计数+1
                except Exception as e:
                    # 失败：计数+1，记录原因
                    result.increment_fail_count()
                    result.add_failed_file(str(file_path), str(e))
                    logger.error(f"✗ 文件索引失败: {file_path.name}, 错误: {e}")

            # 判断整个任务是否成功，失败数 = 0 → 任务成功；有失败文件 → 整个任务标记失败
            result.success = result.fail_count == 0
            result.end_time = datetime.now()

            logger.info(
                f"目录索引完成: 总数={result.total_files}, "
                f"成功={result.success_count}, 失败={result.fail_count}"
            )

            return result

        except Exception as e:
            logger.error(f"索引目录失败: {e}")
            result.success = False
            result.error_message = str(e)
            result.end_time = datetime.now()
            return result

    def index_single_file(self, file_path: str):
        """
        索引单个文件 (使用新的 LangChain 分割器)

        Args:
            file_path: 文件路径

        Raises:
            ValueError: 文件不存在时抛出
            RuntimeError: 索引失败时抛出
        """
        path = Path(file_path).resolve()

        if not path.exists() or not path.is_file():
            raise ValueError(f"文件不存在: {file_path}")

        logger.info(f"开始索引文件: {path}")

        try:
            # 1. 读取文件内容
            content = path.read_text(encoding="utf-8")
            logger.info(f"读取文件: {path}, 内容长度: {len(content)} 字符")

            # 2. 删除该文件的旧数据（如果存在）
            normalized_path = path.as_posix()
            vector_store_manager.delete_by_source(normalized_path)

            # 3. 使用新的文档分割器
            documents = document_splitter_service.split_document(content, normalized_path)
            logger.info(f"文档分割完成: {file_path} -> {len(documents)} 个分片")

            # 4. 添加文档到向量存储
            if documents:
                vector_store_manager.add_documents(documents)  # 向量入库
                logger.info(f"文件索引完成: {file_path}, 共 {len(documents)} 个分片")
            else:
                logger.warning(f"文件内容为空或无法分割: {file_path}")

        except Exception as e:
            logger.error(f"索引文件失败: {file_path}, 错误: {e}")
            raise RuntimeError(f"索引文件失败: {e}") from e

# 统一使用 biz collection
COLLECTION_NAME = "biz"


class VectorStoreManager:
    """向量存储管理器"""

    def __init__(self):
        """初始化向量存储管理器"""
        self.vector_store = None
        self.collection_name = COLLECTION_NAME  # 类似于关系数据库中的"表名"
        self._initialize_vector_store()

    # 连接 Milvus 数据库并创建 VectorStore 实例
    def _initialize_vector_store(self):
        """初始化 Milvus VectorStore"""
        try:
            # 必须在 PyMilvus / langchain_milvus 访问 Collection 之前建立连接，
            # 否则会出现 ConnectionNotExistException: should create connection first.
            # （模块导入时就会执行此处，早于 FastAPI lifespan 中的 milvus_manager.connect）
            _ = milvus_manager.connect()  # 提前建立 PyMilvus 的基础连接；_ 表示忽略返回值（只需要连接的副作用）

            connection_args = {
                "host": config.milvus_host,
                "port": config.milvus_port,
            }

            # 创建 LangChain Milvus VectorStore
            # 使用 biz collection，字段映射：text_field -> content, vector_field -> vector
            self.vector_store = Milvus(
                embedding_function=vector_embedding_service,
                collection_name=self.collection_name,
                connection_args=connection_args,
                auto_id=False,  # 使用自定义 id
                drop_old=False,
                text_field="content",  # 文本内容存储到 content 字段
                vector_field="vector",  # 向量存储到 vector 字段
                primary_field="id",  # 主键字段
                metadata_field="metadata",  # 元数据字段
            )

            logger.info(
                f"VectorStore 初始化成功: {config.milvus_host}:{config.milvus_port}, "
                f"collection: {self.collection_name}"
            )

        except Exception as e:
            logger.error(f"VectorStore 初始化失败: {e}")
            raise

    def delete_by_source(self, file_path: str) -> int:
        """
        按文件路径删除向量数据的核心方法，用于更新文件时清理旧数据

        Args:
            file_path: 文件路径

        Returns:
            int: 删除的文档数量
        """
        try:
            # 使用 milvus_manager 获取已连接的 collection，就是那个biz
            # milvus_manager 是一个全局连接管理器
            collection = milvus_manager.get_collection()
            
            # metadata 是 JSON 字段，使用 JSON 路径查询语法
            # _source 是文档的来源文件路径
            # 生成 Milvus 的查询表达式，用于匹配要删除的数据，类似于SQL表达式
            expr = f'metadata["_source"] == "{file_path}"'
            
            result = collection.delete(expr)  # 执行删除操作，删除所有匹配 expr 的记录
            deleted_count = result.delete_count if hasattr(result, "delete_count") else 0
            
            logger.info(f"删除文件旧数据: {file_path}, 删除数量: {deleted_count}")
            return deleted_count
            
        except Exception as e:
            logger.warning(f"删除旧数据失败 (可能是首次索引): {e}")
            return 0

    def get_vector_store(self) -> Milvus:
        """
        获取 VectorStore 实例

        Returns:
            Milvus: VectorStore 实例
        """
        return self.vector_store

    def similarity_search(self, query: str, k: int = 3) -> List[Document]:
        """
        相似度搜索

        Args:
            query: 查询文本
            k: 返回结果数量

        Returns:
            List[Document]: 相关文档列表
        """
        try:
            docs = self.vector_store.similarity_search(query, k=k)
            logger.debug(f"相似度搜索完成: query='{query}', 结果数={len(docs)}")
            return docs
        except Exception as e:
            logger.error(f"相似度搜索失败: {e}")
            return []

class DocumentSplitterService:
    """文档分割服务 - 使用 LangChain 的分割器"""

    def __init__(self):
        """初始化文档分割服务"""
        self.chunk_size = config.chunk_max_size
        self.chunk_overlap = config.chunk_overlap

        # Markdown 标题分割器 (只按一级和二级标题分割，减少分片数)
        self.markdown_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=[
                ("#", "h1"),
                ("##", "h2"),
                # 不再按三级标题分割，避免过度碎片化
            ],
            strip_headers=False,  # 保留标题在内容中
        )

        # 递归字符分割器 (用于二次分割，使用更大的chunk_size)
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size * 2,  # 加倍chunk_size，减少分片数
            chunk_overlap=self.chunk_overlap,
            length_function=len,
            is_separator_regex=False,
        )

        logger.info(
            f"文档分割服务初始化完成, chunk_size={self.chunk_size}, "
            f"secondary_chunk_size={self.chunk_size * 2}, "
            f"overlap={self.chunk_overlap}"
        )

    def split_markdown(self, content: str, file_path: str = "") -> List[Document]:
        """
        分割 Markdown 文档 (两阶段分割 + 合并小片段)

        Args:
            content: Markdown 内容
            file_path: 文件路径 (用于元数据)

        Returns:
            List[Document]: 文档分片列表
        """
        if not content or not content.strip():
            logger.warning(f"Markdown 文档内容为空: {file_path}")
            return []

        try:
            # 第一阶段: 按标题分割
            md_docs = self.markdown_splitter.split_text(content)

            # 第二阶段: 按大小进一步分割
            docs_after_split = self.text_splitter.split_documents(md_docs)

            # 第三阶段: 合并太小的分片 (< 300字符)
            final_docs = self._merge_small_chunks(docs_after_split, min_size=300)

            # 添加文件路径元数据
            for doc in final_docs:
                doc.metadata["_source"] = file_path
                doc.metadata["_extension"] = ".md"
                doc.metadata["_file_name"] = Path(file_path).name

            logger.info(f"Markdown 分割完成: {file_path} -> {len(final_docs)} 个分片")
            return final_docs

        except Exception as e:
            logger.error(f"Markdown 分割失败: {file_path}, 错误: {e}")
            raise

    def split_text(self, content: str, file_path: str = "") -> List[Document]:
        """
        分割普通文本文档

        Args:
            content: 文本内容
            file_path: 文件路径 (用于元数据)

        Returns:
            List[Document]: 文档分片列表
        """
        if not content or not content.strip():
            logger.warning(f"文本文档内容为空: {file_path}")
            return []

        try:
            # 直接使用递归字符分割器
            # 1. 分割文本（返回字符串列表）
            # 2. 为每个分片创建 Document 对象
            docs = self.text_splitter.create_documents(
                texts=[content],
                metadatas=[
                    {
                        "_source": file_path,
                        "_extension": Path(file_path).suffix,
                        "_file_name": Path(file_path).name,
                    }
                ],
            )

            logger.info(f"文本分割完成: {file_path} -> {len(docs)} 个分片")
            return docs

        except Exception as e:
            logger.error(f"文本分割失败: {file_path}, 错误: {e}")
            raise

    def split_document(self, content: str, file_path: str = "") -> List[Document]:
        """
        智能分割文档 (根据文件类型选择分割器)

        Args:
            content: 文档内容
            file_path: 文件路径

        Returns:
            List[Document]: 文档分片列表
        """
        if file_path.endswith(".md"):
            return self.split_markdown(content, file_path)
        else:
            return self.split_text(content, file_path)

    def _merge_small_chunks(
        self, documents: List[Document], min_size: int = 300
    ) -> List[Document]:
        """
        合并太小的分片

        Args:
            documents: 文档列表
            min_size: 最小分片大小 (字符数)

        Returns:
            List[Document]: 合并后的文档列表
        """
        if not documents:
            return []

        merged_docs = []  # 存储合并后的文档
        current_doc = None  # 当前正在构建的文档

        for doc in documents:
            doc_size = len(doc.page_content)

            if current_doc is None:
                # 第一个文档
                current_doc = doc
            elif doc_size < min_size and len(current_doc.page_content) < self.chunk_size * 2:
                # 当前文档太小且合并后不会太大，则合并
                current_doc.page_content += "\n\n" + doc.page_content
                # 保留主文档的元数据
            else:
                # 保存当前文档，开始新文档
                merged_docs.append(current_doc)
                current_doc = doc

        # 添加最后一个文档
        if current_doc is not None:
            merged_docs.append(current_doc)

        return merged_docs

4.2、Embedding

class DashScopeEmbeddings(Embeddings):
    """阿里云 DashScope Text Embedding (OpenAI 兼容模式)
    
    实现 LangChain 标准 Embeddings 接口:
    - embed_documents(texts: List[str]) → List[List[float]]: 批量嵌入文档
    - embed_query(text: str) → List[float]: 嵌入单个查询
    """

    def __init__(
        self,
        api_key: str,
        model: str = "text-embedding-v4",
        dimensions: int = 1024,
    ):
        """
        初始化 DashScope Embeddings
        
        Args:
            api_key: DashScope API Key
            model: 嵌入模型名称
            dimensions: 向量维度
        """
        if not api_key or api_key == "your-api-key-here":
            raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")
        
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )
        self.model = model
        self.dimensions = dimensions
        
        # 打印初始化信息
        masked_key = self._mask_api_key(api_key)
        logger.info(
            f"DashScope Embeddings 初始化完成 - "
            f"模型: {model}, 维度: {dimensions}, API Key: {masked_key}"
        )

    @staticmethod
    def _mask_api_key(api_key: str) -> str:
        """掩码 API Key 用于日志"""
        if len(api_key) > 8:
            return f"{api_key[:8]}...{api_key[-4:]}"
        return "***"

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """
        批量嵌入文档列表 (LangChain 标准接口)
        
        Args:
            texts: 文本列表
            
        Returns:
            List[List[float]]: 嵌入向量列表
        """
        if not texts:
            return []
        
        try:
            logger.info(f"批量嵌入 {len(texts)} 个文档")
            
            # 批量调用 API
            response = self.client.embeddings.create(
                model=self.model,
                input=texts,
                dimensions=self.dimensions,
                encoding_format="float"
            )
            
            embeddings = [item.embedding for item in response.data]
            logger.debug(f"批量嵌入完成, 维度: {len(embeddings[0])}")
            
            return embeddings
            
        except Exception as e:
            logger.error(f"批量嵌入失败: {e}")
            raise RuntimeError(f"批量嵌入失败: {e}") from e

    def embed_query(self, text: str) -> List[float]:
        """
        嵌入单个查询文本 (LangChain 标准接口)
        
        Args:
            text: 查询文本
            
        Returns:
            List[float]: 嵌入向量
        """
        if not text or not text.strip():
            raise ValueError("查询文本不能为空")
        
        try:
            logger.debug(f"嵌入查询, 长度: {len(text)} 字符")
            
            response = self.client.embeddings.create(
                model=self.model,
                input=text,
                dimensions=self.dimensions,
                encoding_format="float"
            )
            
            embedding = response.data[0].embedding
            logger.debug(f"查询嵌入完成, 维度: {len(embedding)}")
            
            return embedding
            
        except Exception as e:
            logger.error(f"查询嵌入失败: {e}")
            raise RuntimeError(f"查询嵌入失败: {e}") from e

4.3、存储

def add_documents(self, documents: List[Document]) -> List[str]:
        """
        批量添加文档到向量存储（自动批量向量化）

        Args:
            documents: 文档列表

        Returns:
            List[str]: 文档 ID 列表
        """
        try:
            import time
            import uuid
            start_time = time.time()

            # 为每个文档生成唯一 id（因为 auto_id=False）
            ids = [str(uuid.uuid4()) for _ in documents]

            # LangChain Milvus 的 add_documents 会自动调用 embedding_function
            # 并进行批量处理，性能更好
            result_ids = self.vector_store.add_documents(documents, ids=ids)

            elapsed = time.time() - start_time
            logger.info(
                f"批量添加 {len(documents)} 个文档到 VectorStore 完成, "
                f"耗时: {elapsed:.2f}秒, 平均: {elapsed/len(documents):.2f}秒/个"
            )
            return result_ids
        except Exception as e:
            logger.error(f"添加文档失败: {e}")
            raise

4.4、召回

class SearchResult:
    """搜索结果类"""

    def __init__(
        self,
        id: str,
        content: str,
        score: float,
        metadata: Dict[str, Any],
    ):
        self.id = id
        self.content = content
        self.score = score
        self.metadata = metadata

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "id": self.id,
            "content": self.content,
            "score": self.score,
            "metadata": self.metadata,
        }

class VectorSearchService:
    """向量检索服务 - 负责从 Milvus 中搜索相似向量"""

    def __init__(self):
        """初始化向量检索服务"""
        logger.info("向量检索服务初始化完成")

    def search_similar_documents(self, query: str, top_k: int = 3) -> List[SearchResult]:
        """
        搜索相似文档

        Args:
            query: 查询文本
            top_k: 返回最相似的K个结果

        Returns:
            List[SearchResult]: 搜索结果列表

        Raises:
            RuntimeError: 搜索失败时抛出
        """
        try:
            logger.info(f"开始搜索相似文档, 查询: {query}, topK: {top_k}")

            # 1. 将查询文本向量化
            query_vector = vector_embedding_service.embed_query(query)
            logger.debug(f"查询向量生成成功, 维度: {len(query_vector)}")

            # 2. 获取 collection，打开数据库表
            collection: Collection = milvus_manager.get_collection()

            # 3. 构建搜索参数
            search_params = {
                "metric_type": "L2",  # 欧氏距离
                "params": {"nprobe": 10},
            }

            # 4. 执行搜索
            results = collection.search(
                data=[query_vector],
                anns_field="vector",
                param=search_params,
                limit=top_k,
                output_fields=["id", "content", "metadata"],
            )

            # 5. 解析搜索结果
            search_results = []
            for hits in results:
                for hit in hits:
                    result = SearchResult(
                        id=hit.entity.get("id"),  # 取出数据库里的 id 字段
                        content=hit.entity.get("content"),  # 取出文本片段内容
                        score=hit.distance,  # L2 距离，越小越相似
                        metadata=hit.entity.get("metadata", {}),  # 取出元数据（文件路径、来源等）
                    )
                    search_results.append(result)

            logger.info(f"搜索完成, 找到 {len(search_results)} 个相似文档")
            return search_results

        except Exception as e:
            logger.error(f"搜索相似文档失败: {e}")
            raise RuntimeError(f"搜索失败: {e}") from e