知识库搭建：文档处理、分块策略与索引优化

第6章：知识库搭建——文档处理、分块策略与索引优化

知识库的质量决定了 RAG 应用的上限。好的文档处理和分块策略，能让同一个 AI 模型的回答质量提升 30-50%。

本章导读

很多开发者把 Dify 知识库当作一个"文件上传" — 上传文档，等待处理完成，然后就开始用了。这种方式能让系统跑起来，但往往不能让系统表现好。

知识库的搭建是 RAG 应用质量的决定性环节。文档质量、分块策略、Embedding 模型的选择、索引参数的配置——每一个环节都会显著影响最终的检索质量，进而影响 AI 的回答准确性。

本章会从实战角度，系统讲解如何搭建高质量的知识库。我们会覆盖从文档准备到上线运营的全流程，包括一些 Dify 界面上没有明显提示但非常重要的细节。

读完本章，你将能够：

• 根据不同文档类型选择合适的预处理策略
• 理解分块策略背后的原理，做出合理的参数选择
• 为不同场景选择合适的 Embedding 模型
• 优化 Weaviate 的索引配置以获得最佳性能
• 建立知识库的版本管理和持续维护机制
• 诊断和解决常见的知识库质量问题

Level 1：基础认知（1-3 年经验）

知识库搭建的完整流程

构建一个高质量知识库需要经历以下阶段：

阶段 1：文档准备
  ↓ 收集、清洗、格式化原始文档

阶段 2：文档上传与预处理
  ↓ Dify 解析文档格式，提取纯文本

阶段 3：文本分块
  ↓ 将长文档切成适合检索的小块

阶段 4：向量化
  ↓ Embedding 模型将文本转化为向量

阶段 5：索引构建
  ↓ 向量存入向量数据库，建立检索索引

阶段 6：验证与调优
  ↓ 测试检索效果，调整参数

阶段 7：持续维护
  ↓ 文档更新、新增、删除

每个阶段的决策都会影响最终效果。让我们逐一深入。

文档准备：质量决定上限

上传到知识库的文档质量，是整个 RAG 系统的质量上限。垃圾进，垃圾出。

常见的文档质量问题和解决方法：

推荐的文档格式（按效果从好到差）：

1. Markdown (.md) — 最佳
   - 结构清晰（标题、列表、代码块）
   - 分块效果好（可按标题层级切分）
   - 文字密度高，没有格式噪音

2. 纯文本 (.txt)
   - 无格式噪音
   - 但缺乏结构信息，分块需要依赖内容

3. Word (.docx)
   - Dify 可以解析基本格式
   - 但表格、图片等复杂元素解析效果有限

4. PDF
   - 原生 PDF（数字创建）：解析效果较好
   - 扫描 PDF：需要 OCR，效果取决于 OCR 质量

5. HTML
   - Dify 会去除 HTML 标签，提取纯文本
   - 导航栏、广告等噪音内容需要预处理清除

文档预处理脚本示例：

import re
from pathlib import Path

def preprocess_document(input_path: str, output_path: str):
    """
    通用文档预处理：去除常见噪音内容
    """
    with open(input_path, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 1. 去除连续的空白行（超过 2 个空行替换为 1 个）
    content = re.sub(r'\n{3,}', '\n\n', content)
    
    # 2. 去除常见的页眉页脚模式（根据实际文档调整）
    patterns_to_remove = [
        r'第\s*\d+\s*页\s*/\s*共\s*\d+\s*页',  # "第X页/共Y页"
        r'版权所有.*?保留所有权利',                # 版权声明
        r'www\.[a-zA-Z0-9-]+\.[a-zA-Z]{2,}',    # URL
        r'confidential|内部资料|仅供内部使用',      # 保密标识
    ]
    
    for pattern in patterns_to_remove:
        content = re.sub(pattern, '', content, flags=re.IGNORECASE)
    
    # 3. 统一标点符号（全角转半角）
    content = content.replace('，', ',').replace('。', '.').replace('：', ':')
    # 注意：这一步对中文文档可能影响可读性，谨慎使用
    
    # 4. 去除首尾空白
    content = content.strip()
    
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(content)
    
    print(f"处理完成：{Path(input_path).name} → {Path(output_path).name}")
    print(f"原始大小：{Path(input_path).stat().st_size} bytes")
    print(f"处理后大小：{Path(output_path).stat().st_size} bytes")

分块策略的选择

分块是将长文档切成小片段的过程。分块的目标：每个片段独立完整，足以回答某类特定问题。

Dify 内置的分块方式：

在创建知识库时，选择「自动」或「自定义」分块：

自动分块（推荐新手使用）：
  - Dify 自动识别文档结构
  - 按段落和语义边界切分
  - 大多数情况下效果不错

自定义分块（推荐进阶用户）：
  参数：
  ├── 分块大小：每块的最大字符数（默认 500）
  ├── 重叠：相邻块之间重复的字符数（默认 50）
  └── 分隔符：按什么标志分块（如 \n\n 按段落）

分块大小的选择指南：

一个重要的实践原则：分块大小应该和你预期的问题复杂度匹配。用户问一个简单问题（"价格是多少？"），需要的片段可以很小；用户问一个复杂问题（"这个产品的技术架构是什么？"），需要的片段要更大一些。

在 Dify 中创建知识库的完整操作

步骤 1：进入知识库模块

在 Dify 左侧导航中点击「知识库」→「创建知识库」

步骤 2：选择数据源

数据源选项：
├── 上传文件（本章重点）
├── 通过 URL 同步（网页内容）
└── 连接 Notion（企业用户）

步骤 3：选择分块和清洗策略

在「数据处理方式」中选择：

• 自动分段与清洗（简单场景）
• 自定义分段规则（专业场景）

自定义分段配置示例：

分段标识符：\n\n  （按空行分段）
最大分段长度：600 字符
分段重叠：60 字符
文字预处理规则：
  ☑ 删除所有 URL 和电子邮件地址
  ☑ 删除所有 HTML 标签
  ☑ 连续空格替换为单个空格

步骤 4：选择索引方式

高质量：
  - 使用 LLM 对分段内容进行摘要（QA 对）
  - 同时使用向量索引和关键词索引
  - 效果最好，消耗 token

经济：
  - 直接向量化文档内容
  - 只使用倒排索引（关键词）
  - 低成本，效果略差

推荐：绝大多数场景选择「高质量」模式，token 消耗可以接受（处理 100 页文档约消耗 2-5 万 token，成本不到 $1）。

步骤 5：选择 Embedding 模型

推荐选择：
├── text-embedding-3-small（OpenAI）— 成本低，效果好，多语言
├── bge-m3（本地部署）— 中文效果最好，无 API 费用
└── text-embedding-3-large（OpenAI）— 效果最好，成本较高

Level 2：机制深解（3-5 年经验）

理解 Dify 的文档处理管道

当你上传文档后，Dify 在后台执行以下处理流程：

# Dify 文档处理管道（伪代码）
class DocumentProcessingPipeline:
    def process(self, file: UploadedFile, dataset: Dataset, config: IndexingConfig):
        
        # 步骤 1：文档解析（将二进制文件转为纯文本）
        parser = self.get_parser(file.extension)  # PDF解析器、Word解析器等
        raw_text = parser.parse(file.content)
        
        # 步骤 2：文本清洗
        cleaner = TextCleaner(config.cleaning_rules)
        cleaned_text = cleaner.clean(raw_text)
        
        # 步骤 3：文本分块
        splitter = TextSplitter(
            chunk_size=config.segment_max_tokens,
            chunk_overlap=config.segment_overlap,
            separator=config.separator
        )
        chunks = splitter.split(cleaned_text)
        
        # 步骤 4：向量化（批量处理，提高效率）
        embedding_model = self.get_embedding_model(dataset.embedding_model_id)
        embeddings = []
        
        for batch in self.batch_chunks(chunks, batch_size=100):
            batch_texts = [chunk.text for chunk in batch]
            batch_embeddings = embedding_model.encode(batch_texts)
            embeddings.extend(batch_embeddings)
        
        # 步骤 5：存储
        for chunk, embedding in zip(chunks, embeddings):
            # 存入向量数据库
            self.vector_db.upsert(chunk, embedding, dataset.id)
            # 存入关系数据库（用于管理和关键词检索）
            self.relational_db.insert(chunk, dataset.id)
        
        # 步骤 6：更新文档状态
        self.update_document_status(file.id, status=DocumentStatus.COMPLETED)

各种文档格式的解析质量：

PDF 解析质量取决于 PDF 类型：

数字 PDF（程序生成）：
  - 文字提取准确率：> 99%
  - 表格：基本能提取，但格式可能丢失
  - 数学公式：通常无法识别

扫描 PDF（图片扫描）：
  - 依赖 OCR，准确率：70-95%（取决于扫描质量和 OCR 引擎）
  - Dify 默认不做 OCR，需要在配置中开启
  - 开启 OCR 需要配置 OCR 服务（如 Azure Form Recognizer）

改善 PDF 解析质量的方法：
1. 使用 PyMuPDF (fitz) 提取文字
2. 使用 pdfplumber 提取表格
3. 使用 Adobe PDF Services API 做高质量 OCR

父子分块策略（Parent-Child Chunking）

Dify v0.10+ 引入了父子分块功能，这是一种显著提升检索质量的高级策略。

基本思路：

• 子块（Child）：小片段（200-300 字符），用于精细检索（向量相似度更准确）
• 父块（Parent）：大片段（1000-2000 字符），传给 LLM 作为上下文（包含更完整信息）

检索流程：

1. 用用户问题检索子块（精确定位）
2. 找到子块后，取出其对应的父块
3. 将父块内容传给 LLM（完整上下文）

效果对比（以技术文档为例）：

传统分块（500字块）：
  检索到：500字的片段，可能在句子中间截断
  问题：上下文不完整，LLM 无法给出完整答案

父子分块（200字子块 + 1500字父块）：
  子块检索精准定位到相关段落
  父块提供完整的节/章上下文
  LLM 基于完整信息给出准确回答

实测结果：
  传统分块 Recall@5：78%
  父子分块 Recall@5：87%（提升约 12%）

在 Dify 中配置父子分块：

在知识库设置 → 分段方式 → 选择「父子分段」：

父块设置：
  分块大小：1500 字符
  分隔符：\n\n（按段落）

子块设置：
  分块大小：200 字符
  重叠：20 字符

元数据的重要性

为知识库中的文档片段附加元数据，可以大幅提升检索的精确性和可用性。

Dify 支持的元数据类型：

# 在 Dify 中通过 API 上传文档时可以附加元数据
metadata = {
    "document_type": "product_manual",  # 文档类型
    "version": "2.0",                   # 版本号
    "effective_date": "2024-01-01",     # 生效日期
    "department": "engineering",         # 部门
    "confidentiality": "public",        # 保密级别
    "language": "zh",                   # 语言
}

# 上传文档
response = requests.post(
    f"{DIFY_API_URL}/datasets/{dataset_id}/documents/create-by-file",
    headers={"Authorization": f"Bearer {API_KEY}"},
    files={"file": open("product_manual_v2.pdf", "rb")},
    data={
        "indexing_technique": "high_quality",
        "doc_metadata": json.dumps(metadata)
    }
)

基于元数据的检索过滤（在工作流中）：

知识检索节点配置：
  检索查询：{{user_question}}
  过滤条件：
    - document_type == "product_manual"  （只检索产品手册）
    - version == "2.0"                    （只检索最新版本）
    - language == "zh"                    （只检索中文文档）

这种方式可以实现精确的知识库隔离，特别适合多产品、多版本的企业知识库管理场景。

文档更新的最佳实践

企业知识库面临的一个普遍挑战：文档需要持续更新，如何在不影响服务的情况下更新知识库？

方案 1：直接替换（简单场景）

# 通过 Dify API 更新文档
# 步骤 1：删除旧文档
curl -X DELETE \
  -H "Authorization: Bearer $API_KEY" \
  "https://api.dify.ai/v1/datasets/$DATASET_ID/documents/$OLD_DOC_ID"

# 步骤 2：上传新文档
curl -X POST \
  -H "Authorization: Bearer $API_KEY" \
  -F "file=@new_document.pdf" \
  -F "indexing_technique=high_quality" \
  "https://api.dify.ai/v1/datasets/$DATASET_ID/documents/create-by-file"

注意：删除和重新上传之间有一个时间窗口，这段时间内知识库中没有该文档的内容。对于关键文档，这可能导致查询失败。

方案 2：蓝绿知识库（零停机更新）

class KnowledgeBaseUpdater:
    """
    蓝绿知识库更新策略：
    同时维护两个知识库（蓝：当前生产，绿：待更新）
    更新时在绿库上操作，更新完成后切换流量到绿库
    """
    
    def __init__(self, dify_client):
        self.client = dify_client
        self.blue_dataset_id = "dataset-blue-xxx"   # 当前生产知识库
        self.green_dataset_id = "dataset-green-xxx"  # 待更新知识库
        self.active = "blue"  # 当前活跃知识库
    
    def update_knowledge_base(self, new_documents: list[str]):
        # 1. 在非活跃知识库（绿）中更新文档
        inactive = self.green_dataset_id if self.active == "blue" else self.blue_dataset_id
        
        # 清空旧内容
        self.client.clear_dataset(inactive)
        
        # 上传新文档
        for doc_path in new_documents:
            self.client.upload_document(inactive, doc_path)
        
        # 等待索引完成
        self.wait_for_indexing_complete(inactive)
        
        # 验证新知识库（运行测试查询）
        if self.validate_knowledge_base(inactive):
            # 切换流量：更新应用的知识库 ID 配置
            self.switch_traffic(inactive)
            self.active = "green" if self.active == "blue" else "blue"
            print(f"成功切换到知识库：{inactive}")
        else:
            print("验证失败，保留原知识库")
    
    def validate_knowledge_base(self, dataset_id: str) -> bool:
        """运行验证查询，确保新知识库可以正常使用"""
        test_queries = [
            "产品导出格式有哪些",
            "如何注册账号",
            "密码重置步骤",
        ]
        
        for query in test_queries:
            results = self.client.retrieve(dataset_id, query, top_k=3)
            if not results or results[0].score < 0.5:
                print(f"验证失败：查询 '{query}' 没有得到高质量结果")
                return False
        
        return True

Level 3：源码与原理（5 年以上）

Dify 文档分块的实现细节

Dify 使用 LangChain 的 RecursiveCharacterTextSplitter 作为基础分块实现（虽然已经逐步迁移到自研实现，但核心算法相同）：

# Dify 的文本分块实现（api/core/indexing_runner.py 相关逻辑）
class FixedRecursiveCharacterTextSplitter:
    """
    递归字符分块器：
    按照分隔符列表依次尝试分块，先用优先级高的分隔符
    如果分块仍然太大，用下一级分隔符继续切分
    """
    
    DEFAULT_SEPARATORS = [
        "\n\n",   # 优先按段落分
        "\n",     # 其次按换行分
        "。",     # 按句号分（中文）
        ".",      # 按句号分（英文）
        "；",     # 按分号分
        " ",      # 最后按空格分
        "",       # 实在没有分隔符，按字符数强制切
    ]
    
    def __init__(self, chunk_size: int, chunk_overlap: int, separators: list[str] = None):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separators = separators or self.DEFAULT_SEPARATORS
    
    def split_text(self, text: str) -> list[str]:
        return self._split_text(text, self.separators)
    
    def _split_text(self, text: str, separators: list[str]) -> list[str]:
        # 尝试当前优先级的分隔符
        separator = separators[0]
        splits = text.split(separator)
        
        chunks = []
        current_chunk = ""
        
        for split in splits:
            if len(current_chunk) + len(split) + len(separator) <= self.chunk_size:
                current_chunk += split + separator
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                
                # 如果单个片段仍然太大，用下一级分隔符继续切
                if len(split) > self.chunk_size and len(separators) > 1:
                    sub_chunks = self._split_text(split, separators[1:])
                    chunks.extend(sub_chunks[:-1])
                    current_chunk = sub_chunks[-1] + separator
                else:
                    current_chunk = split + separator
        
        if current_chunk.strip():
            chunks.append(current_chunk.strip())
        
        # 处理重叠：在相邻块之间加入重叠内容
        if self.chunk_overlap > 0:
            chunks = self._add_overlap(chunks)
        
        return chunks
    
    def _add_overlap(self, chunks: list[str]) -> list[str]:
        """在相邻块之间添加重叠内容"""
        if len(chunks) <= 1:
            return chunks
        
        overlapped_chunks = [chunks[0]]
        
        for i in range(1, len(chunks)):
            prev_chunk = chunks[i - 1]
            current_chunk = chunks[i]
            
            # 取上一个块的末尾 overlap 字符作为当前块的前缀
            overlap_text = prev_chunk[-self.chunk_overlap:]
            overlapped_chunks.append(overlap_text + current_chunk)
        
        return overlapped_chunks

理解为什么需要递归：

考虑这种情况：一个段落有 2000 字符，超过了 500 字符的分块限制。如果按 \n\n 分段，这个段落不会被切开。然后算法会退回到 \n 尝试，如果还不够，继续退回到句号，以此类推，直到能切出合适大小的块。这种递归策略保证了在任何情况下都能切出不超过 chunk_size 的块，同时尽量保留语义完整性。

Embedding 批量处理的优化

Dify 在向量化文档时使用批量请求来减少 API 调用次数和延迟：

class EmbeddingBatchProcessor:
    """
    批量处理文档向量化，优化 API 调用效率
    """
    
    def __init__(self, model, batch_size: int = 100):
        self.model = model
        self.batch_size = batch_size
    
    def process_chunks(self, chunks: list[DocumentChunk]) -> list[DocumentChunk]:
        """
        批量向量化文档片段
        
        OpenAI text-embedding-3-small 的限制：
        - 单次请求最多 2048 个文本
        - 每个文本最多 8191 tokens
        - Rate limit: 1,000,000 TPM（每分钟 token 数）
        """
        
        # 按 batch_size 分批处理
        processed_chunks = []
        
        for i in range(0, len(chunks), self.batch_size):
            batch = chunks[i:i + self.batch_size]
            batch_texts = [chunk.content for chunk in batch]
            
            # 批量调用 Embedding API
            try:
                embeddings = self.model.embed_documents(batch_texts)
            except RateLimitError:
                # 触发限速时，等待后重试
                time.sleep(60)
                embeddings = self.model.embed_documents(batch_texts)
            
            for chunk, embedding in zip(batch, embeddings):
                chunk.embedding = embedding
                processed_chunks.append(chunk)
            
            # 记录进度
            progress = (i + len(batch)) / len(chunks) * 100
            print(f"向量化进度：{progress:.1f}% ({i + len(batch)}/{len(chunks)})")
        
        return processed_chunks

实际速度参考（text-embedding-3-small，1000 个文档片段，每片段 100 tokens）：

小批量（batch_size=10）：
  API 调用次数：100 次
  总时间：约 30 秒（受网络延迟影响大）

大批量（batch_size=100）：
  API 调用次数：10 次
  总时间：约 8 秒（网络延迟被摊薄）

最大批量（batch_size=2048）：
  API 调用次数：1 次（理论上）
  总时间：约 3-5 秒（限于 API 处理时间）
  
  注意：batch_size 不是越大越好，过大的批次可能触发 token 限制

Weaviate 的数据模型与查询优化

了解 Weaviate 的数据模型，有助于在需要时进行性能调优：

# Weaviate collection 的完整配置（Dify 创建知识库时使用的配置）
collection_config = {
    "class": "DatasetXxx",
    "description": "Dify dataset collection",
    
    # 向量索引配置
    "vectorIndexConfig": {
        "distance": "cosine",   # 使用余弦相似度
        "ef": 64,               # 动态候选列表大小
        "efConstruction": 128,  # 建索引时的候选列表大小
        "maxConnections": 64,   # 每个节点最大连接数
        "vectorCacheMaxObjects": 1000000,  # 向量缓存对象数
        "dynamicEfFactor": 8,   # 动态 ef 的倍数因子
        "dynamicEfMin": 25,
        "dynamicEfMax": 500,
    },
    
    # 数据属性定义
    "properties": [
        {
            "name": "text",
            "dataType": ["text"],
            "indexInverted": True,   # 开启关键词索引（用于混合检索）
        },
        {
            "name": "doc_id",
            "dataType": ["string"],
            "indexInverted": False,  # 只用于过滤，不需要关键词索引
        },
        {
            "name": "dataset_id",
            "dataType": ["string"],
            "indexInverted": False,
        },
        # ... 其他元数据字段
    ],
    
    # 倒排索引配置（用于 BM25 全文检索）
    "invertedIndexConfig": {
        "bm25": {
            "b": 0.75,    # BM25 的 b 参数（文档长度归一化因子）
            "k1": 1.2,    # BM25 的 k1 参数（词频饱和度）
        },
        "stopwords": {
            "preset": "en",  # 英文停用词
        },
        "cleanupIntervalSeconds": 60,
    }
}

关键性能调优：

# 高并发场景下的 Weaviate 查询优化
def optimized_vector_search(
    client: weaviate.Client,
    collection_name: str,
    query_vector: list[float],
    top_k: int = 5,
    score_threshold: float = 0.4
) -> list[dict]:
    
    results = (
        client.query
        .get(collection_name, ["text", "doc_id", "dataset_id"])
        .with_near_vector({
            "vector": query_vector,
            "certainty": score_threshold  # 在 Weaviate 层直接过滤低分结果
                                          # 减少传回 Dify 的数据量
        })
        .with_limit(top_k * 2)  # 多取一些，给 Reranker 更多候选
        .with_additional(["certainty", "distance", "id"])
        .do()
    )
    
    documents = results.get("data", {}).get("Get", {}).get(collection_name, [])
    
    # 过滤并转换格式
    return [
        {
            "text": doc["text"],
            "doc_id": doc["doc_id"],
            "score": doc["_additional"]["certainty"],
        }
        for doc in documents
        if doc["_additional"]["certainty"] >= score_threshold
    ]

Level 4：生产陷阱与决策（专家视角）

陷阱 1：知识库"超载"问题

一个很常见的错误：把所有内容都放进一个知识库，期望 AI 能从中找到任何问题的答案。

问题：当知识库包含 10000 个以上的文档片段时，检索精度开始下降。原因是：

• 向量空间中的片段数量越多，"噪音"就越多
• 相同的词可能在不同上下文下有不同含义，干扰检索
• BM25 检索时，低 IDF 词汇的权重被拉低

解决方案：知识库分层

知识库架构建议：

层级 1：通用知识库（全公司共用）
  - 公司概述、产品总览、政策法规
  - 不频繁更新的基础内容
  - 大小：< 1000 个文档片段

层级 2：产品知识库（按产品线分）
  - 产品A手册知识库
  - 产品B手册知识库
  - 各自独立，减少干扰

层级 3：部门专用知识库
  - 技术文档库（工程师用）
  - 销售知识库（销售用）
  - 客服话术库（客服用）

在 Dify 中实现多知识库路由（使用工作流）：

工作流节点：
[开始] → 接收问题
  ↓
[LLM：问题分类] → 判断属于哪个类型
  {product_a, product_b, general, technical}
  ↓
[条件分支]
  ├── product_a → [知识检索：产品A知识库]
  ├── product_b → [知识检索：产品B知识库]
  ├── technical → [知识检索：技术文档库]
  └── general → [知识检索：通用知识库]
  ↓
[LLM：生成回答]

陷阱 2：文档解析失败的静默处理

Dify 在处理文档时，如果某个文档解析失败，会静默标记为"错误"状态，但不会阻止其他文档处理，也不会主动通知你。

如何发现解析失败：

# 通过 API 检查知识库中的文档状态
curl -H "Authorization: Bearer $API_KEY" \
  "https://api.dify.ai/v1/datasets/$DATASET_ID/documents?page=1&limit=50" \
  | python3 -c "
import sys, json
data = json.load(sys.stdin)
for doc in data.get('data', []):
    if doc.get('indexing_status') != 'completed':
        print(f\"FAILED: {doc.get('name')} - Status: {doc.get('indexing_status')}\")
"

常见解析失败原因和解决方法：

建立解析监控：

import requests
import time

def monitor_indexing_status(dataset_id: str, api_key: str, expected_count: int):
    """
    监控知识库文档的索引状态，直到所有文档处理完成
    """
    headers = {"Authorization": f"Bearer {api_key}"}
    
    while True:
        response = requests.get(
            f"https://api.dify.ai/v1/datasets/{dataset_id}/documents",
            headers=headers,
            params={"page": 1, "limit": 100}
        )
        
        documents = response.json().get("data", [])
        
        completed = [d for d in documents if d["indexing_status"] == "completed"]
        failed = [d for d in documents if d["indexing_status"] == "error"]
        processing = [d for d in documents if d["indexing_status"] in ["waiting", "parsing", "cleaning", "splitting", "indexing"]]
        
        print(f"进度：{len(completed)}/{expected_count} 完成, {len(failed)} 失败, {len(processing)} 处理中")
        
        if failed:
            for doc in failed:
                print(f"  [错误] {doc['name']}: {doc.get('error', '未知错误')}")
        
        if len(completed) + len(failed) >= expected_count:
            print("所有文档处理完成！")
            break
        
        time.sleep(30)  # 每 30 秒检查一次

陷阱 3：知识库的"记忆漂移"问题

在生产环境中运行一段时间后，知识库可能会出现"记忆漂移"现象：

• 同一个问题，今天和三个月前得到不同的答案
• 明明更新了文档，但 AI 还是给出旧版本的信息

根因排查清单：

□ 检查文档更新是否成功
  → 通过 API 查看文档的 updated_at 时间
  
□ 检查旧文档是否被删除
  → 如果只上传新文档而没有删除旧文档，旧内容仍然存在

□ 检查向量数据库中是否有孤立数据
  → 有时 Dify 元数据删除成功，但向量数据未同步删除

□ 检查 Embedding 模型是否更换
  → 更换 Embedding 模型后，旧向量无效，必须重建索引

□ 检查知识库缓存
  → Dify 对检索结果有短时缓存（通常几分钟），刚更新的内容可能还在缓存中

建立知识库健康检查定时任务：

import schedule
import time

def knowledge_base_health_check():
    """每天凌晨检查知识库健康状态"""
    
    # 1. 检查文档数量是否符合预期
    doc_count = get_document_count(DATASET_ID)
    if doc_count < EXPECTED_MIN_DOCS:
        alert(f"知识库文档数量低于预期：{doc_count} < {EXPECTED_MIN_DOCS}")
    
    # 2. 运行标准测试查询
    test_cases = [
        ("导出格式", "CSV"),      # 查询词，预期包含的关键词
        ("注册账号", "邮箱"),
        ("密码重置", "24小时"),
    ]
    
    for query, expected_keyword in test_cases:
        results = retrieve(DATASET_ID, query, top_k=3)
        if not any(expected_keyword in r.text for r in results):
            alert(f"检索质量下降：查询'{query}'未找到包含'{expected_keyword}'的片段")
    
    # 3. 检查平均检索分数
    scores = [r.score for r in retrieve(DATASET_ID, "测试查询", top_k=5)]
    avg_score = sum(scores) / len(scores) if scores else 0
    if avg_score < 0.5:
        alert(f"检索平均分数过低：{avg_score:.3f}")

# 设置定时任务
schedule.every().day.at("02:00").do(knowledge_base_health_check)

while True:
    schedule.run_pending()
    time.sleep(60)

知识库性能基准与预算规划

处理时间基准（不同规模知识库，Dify 默认配置）：

规模：1000 个文档片段（~100 页文档）
  向量化时间：约 1-3 分钟
  总处理时间（含解析、存储）：约 3-8 分钟
  Embedding 费用（text-embedding-3-small）：约 $0.002

规模：10,000 个文档片段（~1000 页文档）
  向量化时间：约 10-30 分钟
  总处理时间：约 20-60 分钟
  Embedding 费用：约 $0.02

规模：100,000 个文档片段（~10,000 页文档）
  向量化时间：约 1-3 小时
  总处理时间：约 2-6 小时
  Embedding 费用：约 $0.2

规模：1,000,000 个文档片段（~100,000 页文档）
  需要评估 Weaviate 容量（默认配置可能需要扩容）
  向量化时间：约 10-30 小时
  Embedding 费用：约 $2

向量存储空间估算：

每个文档片段的存储空间：
  向量（1536维 × 4字节）：6,144 bytes ≈ 6KB
  文本（平均500字符）：约 1KB
  元数据：约 0.5KB
  总计：约 7.5KB / 片段

10万个片段的存储需求：
  向量数据库（Weaviate）：约 750MB
  PostgreSQL（元数据）：约 100MB
  总计：约 1GB（含索引开销约 1.5-2GB）

本章小结

高质量的知识库不是"上传文档就完事了"，而是需要在文档准备、分块策略、Embedding 选择、索引配置、持续维护等多个环节精心设计。

核心要点：

1. 文档质量是上限：垃圾文档进，垃圾回答出；花时间做好文档预处理，效果提升立竿见影
2. 分块策略要匹配业务：FAQ 用小块，技术文档用大块，复杂场景用父子分块
3. 元数据是精确路由的关键：为文档附加版本、类型等元数据，在检索时过滤，提升精确性
4. 监控与维护是持续工作：知识库不是一次性建设，需要随业务变化持续更新和优化
5. 知识库要分层设计：不要把所有内容塞进一个知识库，按业务域、产品、部门分层管理

本章是 Dify 知识库部分的终章。至此，你已经掌握了从零构建一个高质量知识问答系统的完整知识体系：从 Dify 的基础概念（第1-2章）、到快速上手（第3章）、模型选择（第4章）、RAG 原理（第5章）、知识库搭建（第6章）。

接下来的章节将深入 Dify 的高级功能：工作流的复杂编排、Agent 的生产部署、与业务系统的深度集成等。