上一篇我们已经基本实现了一个有记忆，有通过向量的RAG的智能体，但也只是实验室级别的，距离真正的生产标准还有一定的距离。而且我们V1版本主要以实验为主，很多代码耦合度较高，没有细化分类，所以我们本篇会根据开闭原则（对扩展开发，对修改关闭）重构项目。     

        下面我们重新创建一个项目DocMind，这次不添加版本号了，因为重构之后的代码，我们会按需放入项目中，而且是新增为主，通过新增代码来区分功能：

        1.创新新项目DocMind，使用uv环境，创建完成后将 pyproject.toml 修改如下：

[project]
name = "docmind"
version = "0.1.0"
description = "Add your description here"
requires-python = ">=3.14"
dependencies = [
    "chromadb>=1.5.9",
    "langchain>=1.3.13",
    "langchain-chroma>=1.1.0",
    "langchain-community>=0.4.2",
    "langchain-ollama>=1.1.0",
    "langchain-openai>=1.3.5",
    "langchain-text-splitters>=1.1.2",
    "openpyxl>=3.1.5",
    "pydantic>=2.13.4",
    "pypdf>=6.14.2",
    "python-docx>=1.2.0",
    "python-dotenv>=1.2.2",
]

[[tool.uv.index]]
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
default = true

        以上都是我们之前项目的依赖，修改完成后，终端执行，重新安装依赖:

uv sync

        还要添加一些本篇需要的依赖，终端执行：

uv add pymupdf4llm

        2.创建目录

New-Item -ItemType File -Path core/__init__.py, parsers/__init__.py, cleaners/__init__.py, chunkers/__init__.py, enhancers/__init__.py, stores/__init__.py, tools/__init__.py -Force

        如果没有这一堆 __init__.py，主流程代码就需要把所有文件堆在根目录，或者用非常丑陋的强行导入——那样项目会变成一团乱麻。有了它们，各个阶段才能像乐高积木一样，通过清晰的 import 互相组装。

        3. 根目录创建 .env 文件

# DeepSeek API
DEEPSEEK_API_KEY=sk-你的DeepSeek密钥
DEEPSEEK_BASE_URL=https://api.deepseek.com
LLM_MODEL=deepseek-v4-flash

# Ollama 远程服务器（嵌入向量模型）
OLLAMA_BASE_URL=http://ollama服务IP:11434
EMBEDDING_MODEL=qwen3-embedding:4b

# 分块参数
CHUNK_SIZE=500
CHUNK_OVERLAP=50

        4.根目录：embeddings.py（兼容层）

# @File  :  embeddings.py
"""
V1.5 嵌入模块（兼容层）
供 stores/vector_store.py 调用
"""

from langchain_ollama import OllamaEmbeddings
from core.config import Config


def get_ollama_embeddings():
    """连接远程 Ollama 服务器"""
    print(f"🔗 连接远程 Ollama: {Config.OLLAMA_BASE_URL}")
    return OllamaEmbeddings(
        model=Config.EMBEDDING_MODEL,
        base_url=Config.OLLAMA_BASE_URL,
    )

        5.core/__init__.py

# @file : core/__init__.py
"""
核心基础设施层
导出所有核心模块
"""

from core.config import Config
from core.models import DocumentChunk, ChunkWithEmbedding
from core.http_client import LoggingHttpClient

__all__ = [
    "Config",
    "DocumentChunk",
    "ChunkWithEmbedding",
    "LoggingHttpClient",
]

        6. core/config.py

# @file:core/config.py
import os
from dotenv import load_dotenv

load_dotenv()


class Config:
    """集中管理所有配置项"""

    DEEPSEEK_API_KEY: str = os.getenv("DEEPSEEK_API_KEY", "")
    DEEPSEEK_BASE_URL: str = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
    LLM_MODEL: str = os.getenv("LLM_MODEL", "deepseek-v4-flash")

    OLLAMA_BASE_URL: str = os.getenv("OLLAMA_BASE_URL", "http://192.168.0.2:11434")
    EMBEDDING_MODEL: str = os.getenv("EMBEDDING_MODEL", "qwen3-embedding")

    CHUNK_SIZE: int = int(os.getenv("CHUNK_SIZE", "500"))
    CHUNK_OVERLAP: int = int(os.getenv("CHUNK_OVERLAP", "50"))

    @classmethod
    def validate(cls):
        if not cls.DEEPSEEK_API_KEY:
            raise ValueError("请在 .env 文件中设置 DEEPSEEK_API_KEY")
        print(f"✅ 配置验证通过，Ollama 地址: {cls.OLLAMA_BASE_URL}")

        7.core/models.py

# @file : core/models.py
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field


class DocumentChunk(BaseModel):
    content: str = Field(description="切片文本内容")
    source_file: str = Field(description="来源文件路径")
    page_num: Optional[int] = Field(default=None, description="页码或行号")
    section_path: Optional[str] = Field(default=None, description="章节层级路径")
    chunk_type: str = Field(default="paragraph", description="内容类型")
    original_length: int = Field(default=0, description="清洗前字符数")
    clean_length: int = Field(default=0, description="清洗后字符数")
    parent_id: Optional[str] = Field(default=None, description="父级切片ID")
    child_ids: List[str] = Field(default_factory=list, description="子级切片ID列表")
    generated_questions: List[str] = Field(default_factory=list, description="Q2Q预生成问题")
    tags: List[str] = Field(default_factory=list, description="关键词标签")
    summary: Optional[str] = Field(default=None, description="一句话摘要")
    entities: List[str] = Field(default_factory=list, description="命名实体列表")


class ChunkWithEmbedding(BaseModel):
    chunk: DocumentChunk
    embedding: List[float]
    metadata: Dict[str, Any]

        8. core/models.py

# @file : core/models.py
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field


class DocumentChunk(BaseModel):
    content: str = Field(description="切片文本内容")
    source_file: str = Field(description="来源文件路径")
    page_num: Optional[int] = Field(default=None, description="页码或行号")
    section_path: Optional[str] = Field(default=None, description="章节层级路径")
    chunk_type: str = Field(default="paragraph", description="内容类型")
    original_length: int = Field(default=0, description="清洗前字符数")
    clean_length: int = Field(default=0, description="清洗后字符数")
    parent_id: Optional[str] = Field(default=None, description="父级切片ID")
    child_ids: List[str] = Field(default_factory=list, description="子级切片ID列表")
    generated_questions: List[str] = Field(default_factory=list, description="Q2Q预生成问题")
    tags: List[str] = Field(default_factory=list, description="关键词标签")
    summary: Optional[str] = Field(default=None, description="一句话摘要")
    entities: List[str] = Field(default_factory=list, description="命名实体列表")


class ChunkWithEmbedding(BaseModel):
    chunk: DocumentChunk
    embedding: List[float]
    metadata: Dict[str, Any]

        9.core/http_client.py

# @file : core/http_client.py
import httpx
import json


class LoggingHttpClient(httpx.Client):
    def send(self, request, *args, **kwargs):
        print(f"🔗 请求 URL: {request.url}")
        if request.content:
            try:
                body_json = json.loads(request.content)
                print("📤 请求体 (JSON):")
                print(json.dumps(body_json, indent=2, ensure_ascii=False))
            except:
                print("📤 请求体 (文本):", request.content.decode())
        return super().send(request, *args, **kwargs)

        10.parsers/__init__.py

# @file : parsers/__init__.py
from parsers.base import load_documents
from parsers.pdf_parser import parse_pdf
from parsers.docx_parser import parse_docx
from parsers.excel_parser import parse_excel

__all__ = ["load_documents", "parse_pdf", "parse_docx", "parse_excel"]

        11.parsers/base.py

# @file : parsers/base.py
import os
from typing import List
from core.models import DocumentChunk
from parsers.pdf_parser import parse_pdf
from parsers.docx_parser import parse_docx
from parsers.excel_parser import parse_excel


def load_documents(file_path: str) -> List[DocumentChunk]:
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".pdf":
        return parse_pdf(file_path)
    elif ext == ".docx":
        return parse_docx(file_path)
    elif ext == ".xlsx":
        return parse_excel(file_path)
    else:
        raise ValueError(f"不支持的文件格式: {ext}")

        12. parsers/pdf_parser.py

# @file : pdf_parser
# 导入操作系统接口模块，用于文件和路径操作
import os
# 导入正则表达式模块，用于文本匹配与替换
import re
# 从 typing 模块导入 List 类型，用于类型提示
from typing import List
# 导入 PyMuPDF4LLM 库，用于将 PDF 转换为 Markdown 文本
import pymupdf4llm
# 从核心模型模块导入 DocumentChunk 数据类
from core.models import DocumentChunk
# 从文本清洗模块导入 clean_noise 清理函数
from cleaners.text_cleaner import clean_noise


# 定义解析 PDF 文件的函数，接收文件路径，返回 DocumentChunk 列表
def parse_pdf(file_path: str) -> List[DocumentChunk]:
    # 初始化一个空列表，用于存储所有的文档块
    chunks = []
    # 调用 pymupdf4llm 将 PDF 内容转换为 Markdown 格式的字符串 (只能还是文本转换，不能ocr识别)
    md_text = pymupdf4llm.to_markdown(file_path)
    # 按两个换行符分割 Markdown 文本，得到内容块列表（段落、表格、标题等）
    blocks = md_text.split("\n\n")
    # 初始化页码计数器，后续用于估算每个块所在的页码
    page_num = 1
    # 初始化当前章节标题变量，用于记录最近遇到的标题
    current_section = ""

    # 遍历所有内容块，i 为索引，block 为块内容
    for i, block in enumerate(blocks):
        # 如果块为空或只包含空白字符，则跳过该块
        if not block or not block.strip():
            continue
        # 调用 clean_noise 清洗当前块，去除噪声（多余空格、特殊符号等）
        clean_block = clean_noise(block)
        # 如果清洗后的内容为空，则跳过该块
        if not clean_block:
            continue

        # 默认设置块类型为普通段落
        chunk_type = "paragraph"
        # 如果清洗内容以竖线开头且包含 "---"，则判定为表格
        if clean_block.startswith("|") and "---" in clean_block:
            # 将块类型设为表格
            chunk_type = "table"
        # 否则如果清洗内容以井号开头，则判定为标题
        elif clean_block.startswith("#"):
            # 将块类型设为标题
            chunk_type = "heading"
            # 使用正则去除开头的井号和空格，提取纯标题文本，并更新当前章节
            current_section = re.sub(r'^#+\s*', '', clean_block).strip()

        # 创建 DocumentChunk 对象，封装当前块的所有信息
        chunk = DocumentChunk(
            # 清洗后的文本内容
            content=clean_block,
            # 源文件名（从路径中提取）
            source_file=os.path.basename(file_path),
            # 估算页码（每 5 个块大致递增 1 页）
            page_num=page_num + (i // 5),
            # 如果是标题块则不记录章节路径，否则记录当前章节
            section_path=current_section if chunk_type != "heading" else None,
            # 块类型（paragraph / table / heading）
            chunk_type=chunk_type,
            # 原始块的字符长度（清洗前）
            original_length=len(block),
            # 清洗后块的字符长度
            clean_length=len(clean_block)
        )
        # 将构造好的块对象添加到结果列表中
        chunks.append(chunk)

    # 返回包含所有文档块的列表
    return chunks

        13.parsers/docx_parser.py

# @file : parsers/docx_parser.py

# 导入操作系统接口模块，用于文件路径操作（如提取文件名）
import os
# 从 typing 模块导入 List 类型，用于函数返回值的类型提示
from typing import List
# 从 python-docx 库导入 Document 类，并重命名为 DocxDocument，用于操作 .docx 文件
from docx import Document as DocxDocument
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk
# 从文本清洗模块导入 clean_noise 函数，用于清除文本中的噪声
from cleaners.text_cleaner import clean_noise


# 定义解析 DOCX 文件的函数，接收文件路径，返回 DocumentChunk 列表
def parse_docx(file_path: str) -> List[DocumentChunk]:
    # 初始化一个空列表，用于存储所有的文档块
    chunks = []
    # 使用 python-docx 打开并加载指定的 DOCX 文件
    doc = DocxDocument(file_path)
    # 初始化当前章节标题变量，用于跟踪最近的标题
    current_section = ""

    # 遍历文档中的所有段落（paragraphs 是段落对象列表）
    for para in doc.paragraphs:
        # 获取段落文本并去除首尾空白字符
        text = para.text.strip()
        # 如果段落文本为空（或仅空白），则跳过该段落
        if not text:
            continue
        # 调用 clean_noise 函数清洗当前段落文本，去除噪声
        clean_text = clean_noise(text)
        # 如果清洗后的文本为空，则跳过该段落
        if not clean_text:
            continue

        # 判断块类型：如果清洗后文本长度小于20且不含句号，则视为标题，否则为段落
        chunk_type = "heading" if (len(clean_text) < 20 and "。" not in clean_text) else "paragraph"
        # 如果当前块是标题，则更新当前章节标题为该清洗后文本
        if chunk_type == "heading":
            current_section = clean_text

        # 创建 DocumentChunk 对象，封装当前段落的所有信息
        chunk = DocumentChunk(
            # 清洗后的文本内容
            content=clean_text,
            # 源文件名（从文件路径中提取）
            source_file=os.path.basename(file_path),
            # 如果是标题块则不记录章节路径，否则记录当前章节标题
            section_path=current_section if chunk_type != "heading" else None,
            # 块类型（heading 或 paragraph）
            chunk_type=chunk_type,
            # 原始段落的字符长度（清洗前）
            original_length=len(text),
            # 清洗后段落的字符长度
            clean_length=len(clean_text)
        )
        # 将构造好的块对象添加到结果列表中
        chunks.append(chunk)

    # 返回包含所有文档块的列表
    return chunks

        14. excel_parser

# @file : parsers/excel_parser.py
# 导入操作系统接口模块，用于文件和路径操作（如提取文件名）
import os
# 从 typing 模块导入 List 类型，用于函数返回值的类型提示
from typing import List
# 从 openpyxl 库导入 load_workbook 函数，用于加载 Excel 工作簿
from openpyxl import load_workbook
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk
# 从文本清洗模块导入 clean_noise 函数，用于清除文本中的噪声
from cleaners.text_cleaner import clean_noise


# 定义解析 Excel 文件的函数，接收文件路径，返回 DocumentChunk 列表
def parse_excel(file_path: str) -> List[DocumentChunk]:
    # 初始化一个空列表，用于存储所有的文档块
    chunks = []
    # 使用 openpyxl 以只读模式加载指定的 Excel 工作簿（read_only=True 可优化内存占用）
    wb = load_workbook(file_path, read_only=True)

    # 遍历工作簿中的每一个工作表（worksheets 属性返回所有工作表对象）
    for sheet in wb.worksheets:
        # 获取当前工作表的名称
        sheet_name = sheet.title
        # 初始化行号计数器（从0开始，后续在循环开始时递增）
        row_num = 0
        # 遍历当前工作表的每一行，values_only=True 表示只获取单元格的值（不包含样式等）
        for row in sheet.iter_rows(values_only=True):
            # 行号递增（从1开始计数）
            row_num += 1
            # 将当前行的所有非空单元格转换为字符串，并组成列表
            cells = [str(cell) for cell in row if cell is not None]
            # 如果该行所有单元格均为空，则跳过该行
            if not cells:
                continue
            # 将单元格列表格式化为 Markdown 表格的一行，如 "| 值1 | 值2 | 值3 |"
            table_row = "| " + " | ".join(cells) + " |"
            # 调用 clean_noise 清洗当前表格行文本，去除噪声
            clean_text = clean_noise(table_row)
            # 如果清洗后的文本为空，则跳过该行
            if not clean_text:
                continue
            # 创建 DocumentChunk 对象，封装当前 Excel 行的所有信息
            chunk = DocumentChunk(
                # 清洗后的文本内容（表格行格式）
                content=clean_text,
                # 源文件名（从文件路径中提取）
                source_file=os.path.basename(file_path),
                # 页码/行号：此处以行号作为类似页码的标识（便于追踪位置）
                page_num=row_num,
                # 章节路径：记录当前工作表名称，便于溯源
                section_path=f"Sheet: {sheet_name}",
                # 块类型固定为表格（因为 Excel 数据以表格行呈现）
                chunk_type="table",
                # 原始表格行的字符长度（清洗前）
                original_length=len(table_row),
                # 清洗后文本的字符长度
                clean_length=len(clean_text)
            )
            # 将构造好的块对象添加到结果列表中
            chunks.append(chunk)

    # 返回包含所有文档块（每行对应一个块）的列表
    return chunks

        15.cleaners/__init__.py

# @file : cleaners/__init__.py
from cleaners.text_cleaner import clean_noise
__all__ = ["clean_noise"]

        16.cleaners/text_cleaner.py

# @file : cleaners/text_cleaner.py
# 导入正则表达式模块，用于文本模式匹配与替换
import re

# 定义清洗噪声的函数，接收原始文本字符串，返回清洗后的字符串
def clean_noise(text: str) -> str:
    # 使用正则替换掉页码标记，如 "第 1 页 / 共 10 页" 或 "第1页/共10页"（允许空格和斜杠变体）
    text = re.sub(r'第\s*\d+\s*页\s*[/|/]\s*共\s*\d+\s*页', '', text)
    # 使用正则替换掉英文页码标记，如 "Page 1 of 10"（不区分大小写）
    text = re.sub(r'Page\s*\d+\s*of\s*\d+', '', text, flags=re.IGNORECASE)
    # 将多个连续换行符（可能中间夹有空白）替换为单个换行符，以规范化段落间距
    text = re.sub(r'\n\s*\n', '\n', text)
    # 返回去除首尾空白字符后的文本
    return text.strip()

        17.chunkers/__init__.py

# @file : chunkers/__init__.py
from chunkers.chunker_factory import split_chunk
from chunkers.semantic_chunker import SemanticChunker
from chunkers.hierarchical_chunker import HierarchicalChunker

__all__ = ["split_chunk", "SemanticChunker", "HierarchicalChunker"]

        18. chunkers/semantic_chunker.py

# @file : chunkers/semantic_chunker.py
# 从 typing 模块导入 List 类型，用于类型提示
from typing import List
# 从 langchain_text_splitters 导入递归字符文本分割器，用于按语义切分长文本
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 从核心配置模块导入 Config，获取分块大小和重叠等配置参数
from core.config import Config
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk


# 定义语义分块器类，负责将过长的文档块拆分为更小的子块
class SemanticChunker:
    # 定义类的初始化方法，在创建实例时调用
    def __init__(self):
        # 创建一个 RecursiveCharacterTextSplitter 实例并赋值给实例变量 splitter
        self.splitter = RecursiveCharacterTextSplitter(
            # 从配置中获取每个块的最大字符数
            chunk_size=Config.CHUNK_SIZE,
            # 从配置中获取块之间重叠的字符数
            chunk_overlap=Config.CHUNK_OVERLAP,
            # 设置分割优先级：优先按双换行、单换行、句号、感叹号、问号、分号、逗号、空格、字符依次切分
            separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
        )

    # 定义拆分方法，接收一个 DocumentChunk 对象，返回拆分后的 DocumentChunk 列表
    def split(self, chunk: DocumentChunk) -> List[DocumentChunk]:
        # 如果当前块的内容长度不超过配置的最大块大小，则无需拆分，直接返回包含原块的列表
        if len(chunk.content) <= Config.CHUNK_SIZE:
            return [chunk]
        # 调用分割器的 split_text 方法，将内容按分隔符切分为多个子文本片段
        sub_texts = self.splitter.split_text(chunk.content)
        # 初始化空列表，用于存储生成的子块对象
        children = []
        # 遍历每个子文本片段
        for sub_text in sub_texts:
            # 为每个子片段创建一个新的 DocumentChunk 对象，并添加到 children 列表中
            children.append(DocumentChunk(
                # 子块的内容为切分后的子文本
                content=sub_text,
                # 继承原块的源文件名
                source_file=chunk.source_file,
                # 继承原块的页码
                page_num=chunk.page_num,
                # 继承原块的章节路径
                section_path=chunk.section_path,
                # 继承原块的块类型
                chunk_type=chunk.chunk_type,
                # 记录父块 ID：如果原块有 parent_id 则使用，否则使用原块的内存地址作为父 ID（用于追踪层级关系）
                parent_id=chunk.parent_id or str(id(chunk)),
                # 记录子块的原始长度（切分后的长度）
                original_length=len(sub_text),
                # 记录子块清洗后的长度（去除首尾空白后的长度）
                clean_length=len(sub_text.strip())
            ))
        # 返回拆分后的所有子块列表
        return children

        19.chunkers/hierarchical_chunker.py

# @fiel : chunkers/hierarchical_chunker.py
# 从 typing 模块导入 List 类型，用于函数返回值的类型提示
from typing import List
# 从 langchain_text_splitters 导入 Markdown 标题分割器，用于按标题层级切分 Markdown 文档
from langchain_text_splitters import MarkdownHeaderTextSplitter
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk


# 定义层级分块器类，负责按 Markdown 标题层级将文档块拆分为带有层级结构的子块
class HierarchicalChunker:
    # 定义类的初始化方法，在创建实例时调用
    def __init__(self):
        # 设置要分割的 Markdown 标题层级，元组列表包含（标题标记，对应层级名称）
        self.headers_to_split_on = [("#", "h1"), ("##", "h2"), ("###", "h3"), ("####", "h4")]
        # 创建 MarkdownHeaderTextSplitter 实例并赋值给实例变量 splitter
        self.splitter = MarkdownHeaderTextSplitter(
            # 传入要分割的标题层级配置
            headers_to_split_on=self.headers_to_split_on,
            # 设置保留标题文本在内容中（不剥离），便于后续使用
            strip_headers=False
        )

    # 定义拆分方法，接收一个 DocumentChunk 对象，返回拆分后的 DocumentChunk 列表
    def split(self, chunk: DocumentChunk) -> List[DocumentChunk]:
        # 调用分割器的 split_text 方法，按标题将内容切分为多个带元数据的文档片段
        md_splits = self.splitter.split_text(chunk.content)
        # 初始化空列表，用于存储生成的子块对象
        children = []
        # 遍历每个分割后的文档片段
        for md_doc in md_splits:
            # 获取当前片段的元数据（包含标题层级信息）
            metadata = md_doc.metadata
            # 从元数据中提取所有非空的标题值，用 " > " 连接组成层级路径
            section_path = " > ".join([v for k, v in metadata.items() if v])
            # 为每个片段创建一个新的 DocumentChunk 对象，并添加到 children 列表中
            children.append(DocumentChunk(
                # 子块的内容为片段的主体文本
                content=md_doc.page_content,
                # 继承原块的源文件名
                source_file=chunk.source_file,
                # 继承原块的页码
                page_num=chunk.page_num,
                # 如果有元数据则使用生成的层级路径，否则继承原块的章节路径
                section_path=section_path or chunk.section_path,
                # 如果元数据存在（即该片段包含标题）则类型为 heading，否则继承原块类型
                chunk_type="heading" if metadata else chunk.chunk_type,
                # 记录父块 ID：如果原块有 parent_id 则使用，否则使用原块的内存地址
                parent_id=chunk.parent_id or str(id(chunk)),
                # 记录子块的原始长度（注意这里用了原块内容长度，可能需修正，但按原代码保留）
                original_length=len(chunk.content),
                # 记录子块清洗后的长度（即当前片段的长度）
                clean_length=len(md_doc.page_content)
            ))
        # 返回拆分后的所有子块列表
        return children

        20.chunkers/chunker_factory.py

# @file : chunkers/chunker_factory.py
# 从 typing 模块导入 List 类型，用于函数返回值的类型提示
from typing import List
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk
# 从语义分块器模块导入 SemanticChunker 类，用于按语义切分长文本
from chunkers.semantic_chunker import SemanticChunker
# 从层级分块器模块导入 HierarchicalChunker 类，用于按 Markdown 标题层级切分
from chunkers.hierarchical_chunker import HierarchicalChunker

# 初始化全局变量 _semantic 为 None，用于缓存语义分块器单例
_semantic = None
# 初始化全局变量 _hierarchical 为 None，用于缓存层级分块器单例
_hierarchical = None

# 定义获取语义分块器实例的函数，实现单例模式
def get_semantic():
    # 声明使用全局变量 _semantic
    global _semantic
    # 如果 _semantic 尚未初始化，则创建一个新的 SemanticChunker 实例
    if _semantic is None:
        _semantic = SemanticChunker()
    # 返回语义分块器实例
    return _semantic

# 定义获取层级分块器实例的函数，实现单例模式
def get_hierarchical():
    # 声明使用全局变量 _hierarchical
    global _hierarchical
    # 如果 _hierarchical 尚未初始化，则创建一个新的 HierarchicalChunker 实例
    if _hierarchical is None:
        _hierarchical = HierarchicalChunker()
    # 返回层级分块器实例
    return _hierarchical

# 定义智能拆分函数，根据内容特征选择合适的拆分策略
def split_chunk(chunk: DocumentChunk) -> List[DocumentChunk]:
    # 判断内容是否以 Markdown 标题开头（支持 #, ##, ###）
    if any(chunk.content.startswith(header) for header in ["#", "##", "###"]):
        # 若包含标题，则获取层级分块器实例
        hierarchical = get_hierarchical()
        # 调用层级分块器进行拆分，得到初步子块列表
        result = hierarchical.split(chunk)
        # 对过长的子块再走语义分块，初始化最终结果列表
        final = []
        # 遍历每个初步子块
        for sub in result:
            # 如果子块内容长度超过 500 字符，则需进一步切分
            if len(sub.content) > 500:
                # 获取语义分块器实例
                semantic = get_semantic()
                # 对当前子块进行语义切分，并将结果扩展到 final 列表
                final.extend(semantic.split(sub))
            else:
                # 如果子块长度适中，直接保留
                final.append(sub)
        # 返回经过两级处理后的最终块列表
        return final
    else:
        # 若不包含标题，直接使用语义分块器进行切分
        semantic = get_semantic()
        # 返回语义分块的结果
        return semantic.split(chunk)

        21.enhancers/__init__.py

# @file : enhancers/__init__.py
from enhancers.q2q_augmenter import augment_chunks_with_questions
from enhancers.metadata_generator import augment_chunks_with_metadata

__all__ = ["augment_chunks_with_questions", "augment_chunks_with_metadata"]

        22.enhancers/q2q_augmenter.py

# @file : enhancers/q2q_augmenter.py

# 导入正则表达式模块，用于文本清洗和模式匹配
import re
# 从 typing 模块导入 List 类型，用于类型提示
from typing import List
# 从 langchain_openai 导入 ChatOpenAI 类，用于调用 OpenAI 风格的大语言模型
from langchain_openai import ChatOpenAI
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk


# 定义为单个文档块生成相关问题列表的函数，接收文档块和 LLM 实例，返回问题字符串列表
def generate_questions_for_chunk(chunk: DocumentChunk, llm: ChatOpenAI) -> List[str]:
    # 截取文档块内容的前 800 个字符，作为提示词的上下文（避免超出 token 限制）
    content_preview = chunk.content[:800]
    # 构建发送给 LLM 的提示词，要求生成 3 个简短问题，每行一个且不带编号
    prompt = f"""
请根据以下文档片段，生成 3 个用户可能会问的简短问题（每行一个，不要编号）：
{content_preview}
"""
    # 尝试执行 LLM 调用，捕获可能出现的异常
    try:
        # 调用 LLM 的 invoke 方法，传入提示词，获取响应对象
        response = llm.invoke(prompt)
        # 提取响应内容，去除首尾空白，并按换行符拆分为行
        lines = response.content.strip().split("\n")
        # 初始化空列表，用于存储清洗后的问题
        questions = []
        # 遍历每一行
        for line in lines:
            # 去除当前行的首尾空白
            line = line.strip()
            # 如果行为空，则跳过
            if not line:
                continue
            # 使用正则移除行首的数字序号（如 "1." "2、" 等）
            cleaned = re.sub(r'^\d+[.、]\s*', '', line)
            # 如果清洗后非空，则添加到问题列表
            if cleaned:
                questions.append(cleaned)
        # 最多返回前 5 个问题（防止意外输出过多）
        return questions[:5]
    # 捕获所有异常，打印错误信息并返回空列表
    except Exception as e:
        print(f"⚠️ 生成问题失败: {e}")
        return []


# 定义批量增强文档块的函数，为每个块生成问题并存入 generated_questions 属性
def augment_chunks_with_questions(chunks: List[DocumentChunk], llm: ChatOpenAI) -> List[DocumentChunk]:
    # 获取文档块总数，用于进度显示
    total = len(chunks)
    # 遍历所有文档块，i 为索引，chunk 为当前块对象
    for i, chunk in enumerate(chunks):
        # 打印进度信息，显示当前处理的文件及页码
        if chunk.page_num is not None:
            print(f"  Q2Q {i + 1}/{total}: {chunk.source_file} (第{chunk.page_num}页)")
        else:
            print(f"  Q2Q {i + 1}/{total}: {chunk.source_file}")
        # 如果文档块内容长度小于 20 个字符，则跳过（太短的内容不适合生成问题）
        if len(chunk.content) < 20:
            continue
        # 调用生成问题函数，并将结果赋值给当前块的 generated_questions 属性
        chunk.generated_questions = generate_questions_for_chunk(chunk, llm)
    # 返回增强后的文档块列表（原列表被修改，但为了语义清晰仍返回）
    return chunks

        23.enhancers/metadata_generator.py

# @file : enhancers/metadata_generator.py
# 从 typing 模块导入 List 和 Tuple 类型，用于类型提示
from typing import List, Tuple
# 从 langchain_openai 导入 ChatOpenAI 类，用于调用大语言模型
from langchain_openai import ChatOpenAI
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk


# 定义为单个文档块生成元数据（标签、摘要、实体）的函数，返回三元组
def generate_metadata_for_chunk(chunk: DocumentChunk, llm: ChatOpenAI) -> Tuple[List[str], str, List[str]]:
    # 截取文档块内容的前 600 个字符作为预览，避免超出模型上下文限制
    content_preview = chunk.content[:600]
    # 构建发送给 LLM 的提示词，要求提取标签、摘要和实体三类信息
    prompt = f"""
请分析以下文档片段，提取三类信息（每行一个标签）：
标签：关键词1, 关键词2, 关键词3
摘要：一句话摘要（20字内）
实体：实体1, 实体2, 实体3

片段：{content_preview}
"""
    # 尝试执行 LLM 调用，捕获可能发生的异常
    try:
        # 调用 LLM 的 invoke 方法，传入提示词获取响应对象
        response = llm.invoke(prompt)
        # 提取响应内容，去除首尾空白，并按换行符拆分成行列表
        lines = response.content.strip().split("\n")
        # 初始化标签列表、摘要字符串、实体列表
        tags, summary, entities = [], "", []
        # 遍历每一行响应文本
        for line in lines:
            # 去除当前行的首尾空白
            line = line.strip()
            # 判断行是否以“标签”或“标签:”开头
            if line.startswith("标签") or line.startswith("标签:"):
                # 提取冒号后的部分，兼容中文冒号和英文冒号
                parts = line.split("：")[-1] if "：" in line else line.split(":")[-1]
                # 按逗号分割并去除每个标签的空格，生成标签列表
                tags = [t.strip() for t in parts.split(",") if t.strip()]
            # 否则判断行是否以“摘要”或“摘要:”开头
            elif line.startswith("摘要") or line.startswith("摘要:"):
                # 提取冒号后的部分，并去除首尾空白
                summary = line.split("：")[-1] if "：" in line else line.split(":")[-1]
                summary = summary.strip()
            # 否则判断行是否以“实体”或“实体:”开头
            elif line.startswith("实体") or line.startswith("实体:"):
                # 提取冒号后的部分
                parts = line.split("：")[-1] if "：" in line else line.split(":")[-1]
                # 按逗号分割并去除每个实体的空格，生成实体列表
                entities = [e.strip() for e in parts.split(",") if e.strip()]
        # 返回提取到的标签、摘要和实体
        return tags, summary, entities
    # 捕获所有异常，打印错误信息并返回默认空值
    except Exception as e:
        print(f"⚠️ 生成元数据失败: {e}")
        return [], "", []


# 定义批量增强文档块的函数，为每个块生成元数据并存入对象属性
def augment_chunks_with_metadata(chunks: List[DocumentChunk], llm: ChatOpenAI) -> List[DocumentChunk]:
    # 获取文档块总数，用于进度显示
    total = len(chunks)
    # 遍历所有文档块，i 为索引，chunk 为当前块对象
    for i, chunk in enumerate(chunks):
        # 打印进度信息，显示当前处理的文件及页码
        print(f"  元数据 {i+1}/{total}: {chunk.source_file} (第{chunk.page_num}页)")
        # 如果文档块内容长度小于 30 个字符，则跳过（内容太少无法提取有效元数据）
        if len(chunk.content) < 30:
            continue
        # 调用生成元数据函数，获取标签、摘要和实体
        tags, summary, entities = generate_metadata_for_chunk(chunk, llm)
        # 将标签列表赋值给当前块的 tags 属性
        chunk.tags = tags
        # 将摘要字符串赋值给当前块的 summary 属性
        chunk.summary = summary
        # 将实体列表赋值给当前块的 entities 属性
        chunk.entities = entities
    # 返回增强后的文档块列表（原列表被修改，此处返回便于链式调用）
    return chunks

        24.stores/__init__.py

from stores.vector_store import get_vector_store, search_knowledge_base, get_parent_context
from stores.index_pipeline import index_pipeline

__all__ = ["get_vector_store", "search_knowledge_base", "get_parent_context", "index_pipeline"]

        25.stores/vector_store.py

# @file : stores/vector_store.py
# 导入 hashlib 模块，用于生成哈希值作为文档 ID
import hashlib
# 从 typing 模块导入多种类型，用于类型提示（列表、可选、字典、任意类型、元组）
from typing import List, Optional, Dict, Any, Tuple
# 从 langchain_chroma 导入 Chroma 向量数据库类
from langchain_chroma import Chroma
# 从 langchain_core.documents 导入 Document 类，用于 LangChain 文档对象
from langchain_core.documents import Document
# 从核心配置模块导入 Config，获取配置参数
from core.config import Config
# 从核心模型模块导入 DocumentChunk 数据类，用于封装文档块
from core.models import DocumentChunk
# 从 embeddings 模块导入获取 Ollama 嵌入函数的函数
from embeddings import get_ollama_embeddings

# 初始化全局变量 _vector_store 为 None，用于缓存向量存储单例
_vector_store = None

# 定义获取向量存储实例的函数，接收集合名称参数，返回 Chroma 对象
def get_vector_store(collection_name: str = "docmind_kb_v15") -> Chroma:
    # 声明使用全局变量 _vector_store
    global _vector_store
    # 如果 _vector_store 尚未初始化，则创建新的 Chroma 实例
    if _vector_store is None:
        _vector_store = Chroma(
            # 设置向量数据库的集合名称
            collection_name=collection_name,
            # 设置嵌入函数，从 Ollama 获取
            embedding_function=get_ollama_embeddings(),
            # 设置持久化目录，用于存储向量数据
            persist_directory="./chroma_db_v15"
        )
    # 返回向量存储实例
    return _vector_store

# 定义内部函数，构建增强文本（包含问题、摘要、原文）
def _build_augmented_text(chunk: DocumentChunk) -> str:
    # 初始化列表，存储各部分文本
    parts = []
    # 如果文档块包含生成的问题列表，则组装问题部分
    if chunk.generated_questions:
        # 添加“用户可能这样问”前缀，并用分号连接所有问题
        parts.append("【用户可能这样问】" + "；".join(chunk.generated_questions))
    # 如果文档块包含摘要，则添加摘要部分
    if chunk.summary:
        parts.append("【摘要】" + chunk.summary)
    # 添加原文部分
    parts.append("【原文】" + chunk.content)
    # 将所有部分用两个换行符连接，返回增强后的文本
    return "\n\n".join(parts)

# 定义内部函数，将 DocumentChunk 转换为元数据字典
def _chunk_to_metadata(chunk: DocumentChunk) -> Dict[str, Any]:
    # 返回包含各种元数据字段的字典
    return {
        # 源文件名
        "source_file": chunk.source_file,
        # 页码，若为 None 则设为 0
        "page_num": chunk.page_num or 0,
        # 章节路径，若为 None 则设为空字符串
        "section_path": chunk.section_path or "",
        # 块类型
        "chunk_type": chunk.chunk_type,
        # 原始长度
        "original_length": chunk.original_length,
        # 清洗后长度
        "clean_length": chunk.clean_length,
        # 标签列表转为逗号分隔的字符串
        "tags": ",".join(chunk.tags) if chunk.tags else "",
        # 摘要，若为 None 则设为空字符串
        "summary": chunk.summary or "",
        # 实体列表转为逗号分隔的字符串
        "entities": ",".join(chunk.entities) if chunk.entities else "",
        # 父块 ID，若为 None 则设为空字符串
        "parent_id": chunk.parent_id or "",
        # 生成问题的数量
        "question_count": len(chunk.generated_questions),
    }

# 定义内部函数，将 DocumentChunk 转换为 LangChain Document 对象
def _chunk_to_langchain_doc(chunk: DocumentChunk) -> Document:
    # 调用 _build_augmented_text 构建增强文本作为页面内容
    augmented_text = _build_augmented_text(chunk)
    # 调用 _chunk_to_metadata 获取元数据字典
    metadata = _chunk_to_metadata(chunk)
    # 计算内容哈希值，用于生成唯一文档 ID
    content_hash = hashlib.md5(chunk.content.encode("utf-8")).hexdigest()
    # 组合文档 ID：源文件名 + 页码 + 哈希前8位
    doc_id = f"{chunk.source_file}_{chunk.page_num}_{content_hash[:8]}"
    # 创建并返回 LangChain Document 对象
    return Document(page_content=augmented_text, metadata=metadata, id=doc_id)

# 定义知识库搜索函数，接收查询和返回数量，返回文档块和相似度分数的列表
def search_knowledge_base(query: str, k: int = 3) -> List[Tuple[DocumentChunk, float]]:
    # 获取向量存储实例
    store = get_vector_store()
    # 执行相似度搜索，返回文档和分数的列表
    docs_with_scores = store.similarity_search_with_score(query, k=k)
    # 初始化结果列表
    results = []
    # 遍历搜索出的文档及其分数
    for doc, score in docs_with_scores:
        # 提取元数据
        meta = doc.metadata
        # 从元数据和页面内容构造 DocumentChunk 对象
        chunk = DocumentChunk(
            # 页面内容（即增强文本）作为 content
            content=doc.page_content,
            # 源文件名
            source_file=meta.get("source_file", ""),
            # 页码
            page_num=meta.get("page_num"),
            # 章节路径
            section_path=meta.get("section_path"),
            # 块类型
            chunk_type=meta.get("chunk_type", "paragraph"),
            # 标签列表（从逗号分隔字符串解析）
            tags=meta.get("tags", "").split(",") if meta.get("tags") else [],
            # 摘要
            summary=meta.get("summary", ""),
            # 实体列表（从逗号分隔字符串解析）
            entities=meta.get("entities", "").split(",") if meta.get("entities") else [],
            # 父块 ID
            parent_id=meta.get("parent_id"),
        )
        # 将块和分数添加到结果列表
        results.append((chunk, score))
    # 返回搜索结果
    return results

# 定义获取父级上下文的函数，根据子块的 parent_id 查找父块
def get_parent_context(chunk: DocumentChunk) -> Optional[DocumentChunk]:
    # 如果当前块没有 parent_id，则直接返回 None
    if not chunk.parent_id:
        return None
    # 尝试执行查询
    try:
        # 获取向量存储实例
        store = get_vector_store()
        # 使用相似度搜索并添加过滤条件，只返回 parent_id 匹配的文档（取第一个）
        docs = store.similarity_search(chunk.content, k=1, filter={"parent_id": chunk.parent_id})
        # 如果找到文档
        if docs:
            # 提取第一个文档的元数据
            meta = docs[0].metadata
            # 构造并返回父级 DocumentChunk 对象
            return DocumentChunk(
                # 页面内容
                content=docs[0].page_content,
                # 源文件名
                source_file=meta.get("source_file", ""),
                # 页码
                page_num=meta.get("page_num"),
                # 章节路径
                section_path=meta.get("section_path"),
                # 块类型
                chunk_type=meta.get("chunk_type", "paragraph"),
            )
    # 捕获异常并打印错误信息
    except Exception as e:
        print(f"⚠️ 获取父级上下文失败: {e}")
    # 未找到或发生异常则返回 None
    return None

        26.stores/index_pipeline.py

# @file : stores/index_pipeline.py
# 导入操作系统接口模块，用于文件和目录操作
import os
# 从 typing 模块导入 List 类型，用于类型提示
from typing import List
# 从 langchain_openai 导入 ChatOpenAI 类，用于调用大语言模型
from langchain_openai import ChatOpenAI
# 从核心配置模块导入 Config，获取分块大小等配置
from core.config import Config
# 从核心模型模块导入 DocumentChunk 数据类
from core.models import DocumentChunk
# 从解析器基础模块导入 load_documents 函数，用于根据文件类型解析文档
from parsers.base import load_documents
# 从分块器工厂模块导入 split_chunk 函数，进行智能分块
from chunkers.chunker_factory import split_chunk
# 从增强器模块导入 Q2Q 问题生成函数
from enhancers.q2q_augmenter import augment_chunks_with_questions
# 从增强器模块导入元数据生成函数
from enhancers.metadata_generator import augment_chunks_with_metadata
# 从向量存储模块导入获取向量存储实例和转换函数
from stores.vector_store import get_vector_store, _chunk_to_langchain_doc


# 定义索引管道函数，接收文档目录和可选的 LLM 实例，返回成功索引的文档数量
def index_pipeline(docs_dir: str = "./docs", llm: ChatOpenAI = None) -> int:
    # 检查文档目录是否存在
    if not os.path.exists(docs_dir):
        # 如果不存在，则创建目录（包括必要的父目录）
        os.makedirs(docs_dir, exist_ok=True)
        # 打印提示信息，要求用户放入文档后重启
        print("📁 已创建 docs 文件夹，请放入文档后重启")
        # 返回 0 表示未索引任何文档
        return 0

    # 获取文档目录下所有支持的文件（.pdf, .docx, .xlsx）
    files = [f for f in os.listdir(docs_dir) if f.endswith((".pdf", ".docx", ".xlsx"))]
    # 如果没有找到任何支持的文件
    if not files:
        # 打印提示信息
        print("📂 docs 文件夹为空")
        # 返回 0
        return 0

    # 打印开始解析文档的步骤信息
    print("📄 步骤1: 解析文档...")
    # 初始化空列表，用于存储所有解析出的原始块
    all_chunks = []
    # 遍历每个文件名
    for filename in files:
        # 拼接完整的文件路径
        file_path = os.path.join(docs_dir, filename)
        # 打印正在处理的文件名
        print(f"  - {filename}")
        # 调用 load_documents 解析该文件，返回文档块列表
        chunks = load_documents(file_path)
        # 将解析出的块扩展到总列表中
        all_chunks.extend(chunks)

    # 如果没有解析到任何有效内容
    if not all_chunks:
        # 打印警告信息
        print("⚠️ 未解析到有效内容")
        # 返回 0
        return 0
    # 打印解析出的原始切片数量
    print(f"  共解析 {len(all_chunks)} 个原始切片")

    # 打印智能分块步骤信息
    print("✂️ 步骤2: 智能分块（层级 + 语义）...")
    # 初始化最终块列表
    final_chunks = []
    # 遍历每个原始块
    for chunk in all_chunks:
        # 如果当前块内容长度超过配置的最大块大小
        if len(chunk.content) > Config.CHUNK_SIZE:
            # 调用 split_chunk 进行分块（先层级后语义）
            sub_chunks = split_chunk(chunk)
            # 将分块后的子块扩展到最终列表
            final_chunks.extend(sub_chunks)
        else:
            # 否则直接保留该块
            final_chunks.append(chunk)
    # 打印最终生成的切片数量
    print(f"  共生成 {len(final_chunks)} 个最终切片")

    # 如果传入了 LLM 实例
    if llm:
        # 打印 Q2Q 生成问题步骤信息
        print("🤔 步骤3: Q2Q 预生成问题...")
        # 调用问题生成增强函数，更新 final_chunks
        final_chunks = augment_chunks_with_questions(final_chunks, llm)

    # 如果传入了 LLM 实例
    if llm:
        # 打印元数据生成步骤信息
        print("🏷️ 步骤4: LLM 生成元数据...")
        # 调用元数据生成增强函数，更新 final_chunks
        final_chunks = augment_chunks_with_metadata(final_chunks, llm)

    # 打印存入向量库步骤信息
    print("💾 步骤5: 存入向量库...")
    # 获取向量存储实例（单例）
    store = get_vector_store()
    # 将所有最终块转换为 LangChain Document 对象列表
    docs_to_add = [_chunk_to_langchain_doc(chunk) for chunk in final_chunks]
    # 如果有文档需要添加
    if docs_to_add:
        # 调用向量存储的 add_documents 方法批量添加
        store.add_documents(docs_to_add)
        # 打印成功索引的切片数量
        print(f"✅ 成功索引 {len(docs_to_add)} 个切片")

    # 打印空行和索引统计信息
    print("\n📊 索引统计:")
    # 打印总原始切片数
    print(f"  总原始切片数: {len(all_chunks)}")
    # 打印最终切片数
    print(f"  最终切片数: {len(final_chunks)}")
    # 打印压缩比（原始/最终）
    print(f"  压缩比: {len(all_chunks) / len(final_chunks):.2f}x")
    # 打印向量库集合名称
    print(f"  向量库集合: docmind_kb_v15")
    # 返回成功添加的文档数量
    return len(docs_to_add)

        27.tools/__init__.py

from tools.rag_tool import query_knowledge_base
__all__ = ["query_knowledge_base"]

        28.tools/rag_tool.py

# @file : tools/rag_tool.py
# 从 langchain.tools 导入 tool 装饰器，用于定义工具函数
from langchain.tools import tool
# 从 pydantic 导入 BaseModel 和 Field，用于定义输入参数模型
from pydantic import BaseModel, Field
# 从向量存储模块导入搜索知识库和获取父级上下文的函数
from stores.vector_store import search_knowledge_base, get_parent_context


# 定义搜索输入参数模型，继承自 BaseModel
class SearchInput(BaseModel):
    # 定义 query 字段，描述为"用户问题中的核心检索关键词，提取最相关的名词或短语"
    query: str = Field(description="用户问题中的核心检索关键词，提取最相关的名词或短语")


# 使用 @tool 装饰器将函数定义为工具，并指定参数模式为 SearchInput
# 🆕 显式提供 description，完全绕过 docstring 解析
@tool(args_schema=SearchInput, description="从企业文档知识库中检索相关信息。")
# 定义查询知识库的工具函数，接收 query 字符串，返回上下文文本
def query_knowledge_base(query: str) -> str:
    # 调用搜索函数，获取最多 3 个相关块及其相似度分数
    results = search_knowledge_base(query, k=3)
    # 如果没有搜索结果，返回提示信息
    if not results:
        return "知识库中未找到相关信息。请先上传文档。"
    # 初始化上下文部分列表
    context_parts = []
    # 遍历每个搜索结果（块和分数）
    for chunk, score in results:
        # 构建基本来源信息：文件名和页码
        part = f"【来源】{chunk.source_file} 第{chunk.page_num}页\n"
        # 如果块有章节路径，添加章节信息
        if chunk.section_path:
            part += f"【章节】{chunk.section_path}\n"
        # 如果块有标签，添加前5个标签
        if chunk.tags:
            part += f"【标签】{', '.join(chunk.tags[:5])}\n"
        # 添加块的内容
        part += f"【内容】{chunk.content}"
        # 尝试获取父级上下文
        parent = get_parent_context(chunk)
        # 如果存在父级块，添加父级内容（截取前300字符）
        if parent:
            part += f"\n【父级上下文】{parent.content[:300]}..."
        # 将组装好的部分添加到列表中
        context_parts.append(part)
    # 用分隔符连接所有部分，返回完整的上下文
    return "\n\n---\n\n".join(context_parts)

        29.main.py

# @file : main.py
# 从 langchain_core.messages 导入 AIMessage 类，用于处理 AI 回复消息
from langchain_core.messages import AIMessage
# 从 langchain_openai 导入 ChatOpenAI 类，用于初始化大语言模型客户端
from langchain_openai import ChatOpenAI
# 从 langchain.agents 导入 create_agent 函数，用于创建智能体
from langchain.agents import create_agent
# 从 langgraph.checkpoint.memory 导入 MemorySaver，用于对话状态记忆
from langgraph.checkpoint.memory import MemorySaver

# 从核心配置模块导入 Config，获取应用配置
from core.config import Config
# 从核心 HTTP 客户端模块导入 LoggingHttpClient，用于带日志的 HTTP 请求
from core.http_client import LoggingHttpClient
# 从工具模块导入知识库查询工具函数
from tools.rag_tool import query_knowledge_base
# 从向量存储索引管道模块导入索引函数
from stores.index_pipeline import index_pipeline

# ---------- 初始化 ----------
# 验证配置项是否完整有效（如 API 密钥等）
Config.validate()
# 创建带有日志功能的 HTTP 客户端，设置超时时间为 60 秒
http_client = LoggingHttpClient(timeout=60.0)

# 初始化 ChatOpenAI 大语言模型实例
llm = ChatOpenAI(
    # 模型名称（从配置中读取）
    model=Config.LLM_MODEL,
    # API 密钥（从配置中读取 DeepSeek API 密钥）
    api_key=Config.DEEPSEEK_API_KEY,
    # API 基础 URL（从配置中读取 DeepSeek 端点）
    base_url=Config.DEEPSEEK_BASE_URL,
    # 温度参数设为 0.7，控制回复的随机性
    temperature=0.7,
    # 传入自定义 HTTP 客户端
    http_client=http_client
)

# ---------- 启动时索引 ----------
# 打印分隔线和启动信息
print("=" * 60)
print("DocMind V1.5 启动中...")
print("=" * 60)
# 调用索引管道，指定文档目录和 LLM 实例
index_pipeline(docs_dir="./docs", llm=llm)

# ---------- 创建 Agent ----------
# 创建内存检查点保存器，用于存储对话历史
memory = MemorySaver()
# 使用 create_agent 创建智能体
agent = create_agent(
    # 指定使用的模型
    model=llm,
    # 提供工具列表（知识库查询工具）
    tools=[query_knowledge_base],
    # 设置系统提示词，指导智能体行为
    system_prompt=(
        "你是企业文档智能助手 DocMind V1.5。\n"
        "1. 回答问题时，优先使用 query_knowledge_base 工具检索知识库。\n"
        "2. 检索结果已包含来源页码、章节标签和父级上下文，请综合使用。\n"
        "3. 回答时尽量注明信息来源（如'根据第3页...'）。"
    ),
    # 传入检查点保存器，实现对话记忆
    checkpointer=memory
)

# ---------- 对话循环 ----------
# 打印就绪信息和提示
print("\n" + "=" * 60)
print("DocMind V1.5 已就绪")
print("📌 新增特性: OCR扫描件 | 语义分块 | Q2Q问题 | 元数据指纹 | 父子索引")
print("输入 'exit' 退出")
print("=" * 60)

# 设置线程 ID，用于区分不同用户的对话
thread_id = "user_001"
# 进入无限循环，持续接收用户输入
while True:
    # 获取用户输入（去除首尾空白）
    user_input = input("\n你: ")
    # 如果用户输入 'exit'（不区分大小写），则退出循环
    if user_input.lower() == "exit":
        break
    # 如果输入为空或仅有空白，则跳过本次循环
    if not user_input.strip():
        continue

    # 调用智能体的 invoke 方法，传入用户消息和配置（包含线程 ID）
    result = agent.invoke(
        {"messages": [{"role": "user", "content": user_input}]},
        config={"configurable": {"thread_id": thread_id}}
    )

    # 初始化变量，用于存储最后一条 AI 消息
    last_ai_msg = None
    # 反向遍历智能体返回的消息列表（从后向前）
    for msg in reversed(result['messages']):
        # 如果消息是 AIMessage 类型且包含内容
        if isinstance(msg, AIMessage) and msg.content:
            # 赋值并跳出循环
            last_ai_msg = msg
            break

    # 打印分隔线
    print("-" * 60)
    # 如果找到了 AI 消息，则打印其内容
    if last_ai_msg:
        print(f"🤖 DocMind V1.5: {last_ai_msg.content}")
    else:
        # 否则打印未获取到有效回复的提示
        print("🤖 DocMind V1.5: (未获取到有效回复)")
    # 打印分隔线
    print("-" * 60)

V1.0 vs V1.5 完整对比表

一、项目结构与代码组织

对比维度	V1.0（扁平结构）	V1.5（分层架构）
项目名称	docmind-v1	docmind-v1.5
目录结构	所有代码平铺在根目录（8个文件）	7个分层包，职责清晰（28个文件）
核心分层	无	core/ parsers/ cleaners/ chunkers/ enhancers/ stores/ tools/
导入方式	直接从文件名导入	从包名导入，内部隐藏实现细节
代码可读性	中等（需要翻阅多个函数才能理解流程）	高（每个包的名称即代表其职责）
新增功能影响	必须修改现有文件（如 document_loader.py）	在对应包中新增文件，原有代码零改动

---

二、文档解析能力

对比维度	V1.0	V1.5
PDF 解析	pypdf 纯文本提取	pymupdf4llm（OCR + 表格 + 多栏排版）
扫描件支持	❌ 不支持（乱码/空白）	✅ 支持（集成 Tesseract OCR）
表格识别	❌ 丢失行列结构	✅ 转为 Markdown Table 保留结构
多栏排版	❌ 提取顺序错乱	✅ 智能重组阅读顺序
Word 解析	python-docx 按段落提取	python-docx + 噪声清洗
Excel 解析	openpyxl 拼接为纯文本	openpyxl 转为 Markdown 表格行
返回格式	List[str]（纯文本）	List[DocumentChunk]（结构化对象）

---

三、分块策略

对比维度	V1.0	V1.5
分块方式	RecursiveCharacterTextSplitter 按 500 字符硬切	智能分块：层级分块 + 语义分块双策略
句子完整性	❌ 可能切断句子	✅ 按句号、问号、段落边界切分
结构保留	❌ 不保留章节信息	✅ 按 Markdown 标题（# ## ###）切分，保留层级路径
分块策略	单一策略	组合策略：层级分块 → 语义精修

---

四、存储内容与增强

对比维度	V1.0	V1.5
存储内容	仅存储原文向量	存储“Q2Q问题 + 摘要 + 原文”拼接向量
Q2Q 预生成问题	❌ 不支持	✅ LLM 为每个切片生成 3~5 个口语化问题
元数据	极少（仅文件名）	全量元数据指纹：页码、章节路径、内容类型、标签、摘要、实体
LLM 元数据生成	❌ 不支持	✅ 自动生成标签/摘要/命名实体
父子索引	❌ 扁平索引	✅ 建立父子关系，检索时自动带回父级上下文
数据模型	List[str] 裸列表	DocumentChunk（Pydantic 强类型实体）

---

五、代码质量与工程化

对比维度	V1.0	V1.5
配置管理	os.getenv + 手动 validate()	同上（保留，与 Pydantic 分离）
数据校验	无	Pydantic BaseModel 定义所有实体
调试能力	LoggingHttpClient 拦截 DeepSeek 请求	同左 + 各模块独立日志输出
类型安全	类型注解部分缺失	全量类型注解，IDE 友好
模块解耦	各模块直接依赖，耦合紧密	通过 __init__.py 控制导出，内部可独立重构

---

六、扩展性（对后续版本的影响）

场景	V1.0（扁平结构）	V1.5（分层架构）
新增 PDF 解析增强	改 document_loader.py（牵一发动全身）	改 parsers/pdf_parser.py（不影响其他格式）
新增分块策略	改 embeddings.py，加 if 分支	在 chunkers/ 新增文件，工厂自动注册
新增检索增强	改 vector_store.py，加参数	在 enhancers/ 新增文件，管线自动调用
新增 Agent 工具	在 main.py 里添加 @tool	在 tools/ 新增文件，main.py 只负责导入
升级到 Milvus	全部重写 vector_store.py	在 stores/ 新增 milvus_store.py，通过接口切换
V2.0 新增编排层	改 main.py，越来越臃肿	新增 orchestrator/ 包，main.py 保持清爽
V3.0 新增检索策略	改 vector_store.py，加多路召回	新增 retrievers/ 包，专注检索优化
V4.0 新增可观测性	到处加日志埋点，侵入性强	新增 observability/ 包，通过钩子注入，零侵入

---

七、技术依赖对比

依赖	V1.0	V1.5	说明
pypdf	✅	✅（保留，仅作备用）	早期 PDF 解析
pymupdf4llm	❌	✅ 新增	OCR + 表格 + 多栏
pymupdf	❌	✅ 新增	PDF 底层引擎
python-docx	✅	✅	Word 解析
openpyxl	✅	✅	Excel 解析
langchain-ollama	✅	✅	Ollama 嵌入（已从 community 迁移）
pydantic	✅	✅	数据校验（V1.5 全面使用）
pydantic-settings	❌	❌	未使用（保持 os.getenv）
reportlab	❌	❌（仅测试脚本用）	PDF 生成（用户测试时使用）