分拣系统文档解析与向量库构建

关联：索引

术语小抄（初学者版，建议抄到笔记最前面）

• 文档解析：把 PDF/Word/Markdown 等“人读格式”转成“机器可处理的文本 + 结构信息（标题/章节/页码/条款编号）”。

• 文本分块（chunking）：把长文切成片段，作为检索与提示的基本单元；分块要保留上下文与可追溯信息。

• 向量嵌入（embedding）：把文本编码成向量；相似文本在向量空间距离更近。

• Sentence-BERT：以句子级语义表示为目标的嵌入模型家族，常用于语义检索、语义相似度计算。

• 元数据（metadata）：与每个分块绑定的来源信息（文件名、章节、规则号、页码、时间戳等），用于解释与追责。

• 如果后续要做“知识问答/故障咨询/规则解释”，关键在于：文档要能被检索命中、能定位来源、能复验。

• 你做的不是“把文档丢给大模型”，而是“把文档变成可检索的数据结构”。

• 至少 1 份文档解析成功，输出纯文本 + 基本结构信息（标题/章节/规则编号或页码）。

• 至少实现 1 种分块方法（按长度）并能打印 5 个样例块。

• 可选加分：实现 1 种“按语义”的分块（简单版本即可），并说明与按长度分块的差异。

1）设备手册（Manual）解析要点

• 特征：章节清晰、术语多、参数表多、故障码/告警码常见。
• 解析目标：尽量保留“章节层级 + 小节标题”，表格可先转成“键值行”或“列表行”文本。
• 风险：PDF 提取时容易丢换行、表格错位、页眉页脚污染文本，需要清洗。

2）分拣规则文档（Rules）解析要点

• 特征：条款编号明显（如 1.1/1.2/2.3 或 R-001）、条件句多（若/当/否则）、例外条款多。
• 解析目标：优先保留“规则编号 + 规则标题 + 规则正文”，不要把一条规则切碎。
• 风险：如果分块把“条件”和“结论”拆开，检索会命中但无法执行，工程上会导致误判。

适用场景：

• 快速上手；文档结构不稳定或解析不到标题；先保证“能跑通”。

核心思路：

• 设定 chunk_size 与 chunk_overlap，用重叠缓解“边界信息丢失”。
• 每个 chunk 必须带 metadata（来源、章节、rule_id/page 等至少 2 个字段）。

代码模板：纯 Python 的按长度滑窗分块（可直接运行）

from __future__ import annotations

from dataclasses import dataclass
from typing import Dict, List, Optional

@dataclass(frozen=True)
class Chunk:
    chunk_id: str
    text: str
    metadata: Dict[str, str]

def normalize_text(text: str) -> str:
    t = (text or "").replace("\r\n", "\n").replace("\r", "\n")
    lines = [ln.strip() for ln in t.split("\n")]
    lines = [ln for ln in lines if ln]
    return "\n".join(lines)

def chunk_by_length(
    text: str,
    *,
    chunk_size: int = 800,
    chunk_overlap: int = 120,
    metadata: Optional[Dict[str, str]] = None,
    chunk_id_prefix: str = "C",
) -> List[Chunk]:
    if chunk_size <= 0:
        raise ValueError("chunk_size must be > 0")
    if chunk_overlap < 0:
        raise ValueError("chunk_overlap must be >= 0")
    if chunk_overlap >= chunk_size:
        raise ValueError("chunk_overlap must be < chunk_size")

    t = normalize_text(text)
    if not t:
        return []

    meta = dict(metadata or {})
    out: List[Chunk] = []
    step = chunk_size - chunk_overlap

    start = 0
    idx = 0
    while start < len(t):
        end = min(start + chunk_size, len(t))
        chunk_text = t[start:end]
        out.append(
            Chunk(
                chunk_id=f"{chunk_id_prefix}-{idx:04d}",
                text=chunk_text,
                metadata=dict(meta),
            )
        )
        idx += 1
        start += step

    return out

if __name__ == "__main__":
    demo_text = """
    1.1 分拣规则概述
    若果径 >= 80mm 且表面瑕疵面积 < 1%，判定为 A 级；
    若果径 >= 75mm 且表面瑕疵面积 < 3%，判定为 B 级；
    否则判定为 C 级。

    1.2 例外条款
    当检测到霉斑时，直接判定为 D 级并进入人工复检通道。
    """
    chunks = chunk_by_length(
        demo_text,
        chunk_size=120,
        chunk_overlap=20,
        metadata={"source": "apple_rules_demo.md", "doc_type": "rules"},
        chunk_id_prefix="RULE",
    )
    for c in chunks[:5]:
        print(c.chunk_id, c.metadata)
        print(c.text)
        print("-" * 40)

解释与自检要点：

• normalize_text()：统一换行并清理空行，减少“页眉页脚/多余空行”对分块的干扰。
• chunk_size/chunk_overlap：这是字符级分块；重叠不是越大越好，过大会造成重复写入与检索噪声。
• 自检：打印前 5 个 chunk，确认没有出现“全是空白/全是标题/规则被切碎到看不懂”的情况。

适用场景：

• 规则条款或说明段落较长，需要尽量保证语义完整；希望检索命中后片段可直接用于解释/引用。

1. 先按“句子/分号/换行”做粗切分，得到 sentences。
2. 使用嵌入模型得到每句向量。
3. 计算相邻句的相似度，当相似度低于阈值（如 0.55）认为主题发生跳变，在此处分段（阈值仅为示例，需结合文档类型与检索效果做调参）。
4. 用 max_chars 控制单段最大长度，必要时再做一次按长度二次切分。

注意：

1）给你们的“苹果分拣规则文档”，写出你认为最合适的 chunk_size 与 chunk_overlap（给出理由：命中率/完整性/重复率）。
提示：规则条款一般更适合“块稍短 + 元数据更全”，设备手册一般更适合“块稍长 + 标题层级保留”。

2）指出你当前分块方案的 1 个风险点，并给出改进方向。
提示：风险点可从“边界切碎、表格错位、页眉页脚污染、同义表达导致命中偏差”等角度写。

• “能分块”只是第一步；要让系统“会找”，需要把 chunk 编成向量并建立索引。

• 成功安装并导入 faiss 与 sentence_transformers（至少在同一虚拟环境中）。

• 构建 1 个 FAISS 索引并写入 ≥ 30 条 chunk 向量（可更少，但建议 30+）。

• 至少 3 条查询返回正确的来源片段（输出要包含 source/section/rule_id/page 等证据字段）。

1）一句话理解嵌入

• 嵌入把文本映射到向量空间；相似文本的向量更接近。

• 检索时，我们把“问题”也嵌入成向量，找向量空间中最近的 chunk。

• 常见度量：

• L2 距离：越小越相似。

• 内积（Dot Product）：越大越相似。

• 余弦相似度：内积 + 向量归一化后等价（常用）。

对应数学公式（向量以 ($$\mathbf{x}$$, $$\mathbf{y}$$ $$\in$$ $$\mathbb{R}^d$$) 表示）：

L2 距离（欧氏距离）：

内积（点积）：

余弦相似度：

归一化（单位向量）：

等价关系（归一化后）：

3）Sentence-BERT 模型选择建议（示例）

• 英文/通用：all-MiniLM-L6-v2（快、常用）。
• 多语言：paraphrase-multilingual-MiniLM-L12-v2（中文可用，速度中等）。
• 中文更强可选：选择你们课程/机房可下载的中文句向量模型（确保来源可靠、许可证合规）。

三、FAISS 与“向量数据库”的关系（认知拓展）

先把概念说清楚（避免误解）：

• 向量索引库/检索库：偏“算法与数据结构”，提供相似度检索能力（既可以精确 KNN，也可以做近似 ANN；取决于索引类型）。FAISS 属于这一类。
• 向量数据库：偏“可服务化的数据系统”，通常包含：向量索引 + 元数据管理 + 过滤查询 + 持久化 + 并发访问 + 权限/隔离 + 监控运维等能力。

把 RAG 知识库拆成 5 个模块，你会更容易理解“为什么 FAISS 能用、但不等同于数据库”：

1. 文档处理：解析、清洗、分块（输出 chunk + metadata）
2. 向量生成：嵌入模型把 chunk 编成向量（输出 vectors）
3. 向量索引：存向量并支持相似度检索（FAISS 就在这里）
4. 元数据与过滤：按来源、章节、规则编号、版本、设备型号等过滤与追溯（需要 metadata 存储与查询）
5. 服务化与运维：API、并发、备份、权限、监控、扩容（生产系统常需要）

常见向量数据库/向量检索方案（让学生知道生态，不要求都掌握）：

• Milvus：典型向量数据库（服务化/可扩展），适合更大规模与工程化部署。

• Qdrant / Weaviate：服务化向量数据库（强调 API 友好、过滤与工程能力），常用于应用落地。

• Elasticsearch / OpenSearch：在搜索引擎体系中提供向量检索（更擅长关键词 + 向量混合检索与复杂过滤）。

• PostgreSQL + pgvector：把向量作为数据库字段存储，适合“结构化业务数据 + 向量检索”一体化的场景。

• 文档规则/设备手册类知识通常较稳定，数据量可控：FAISS 足够支撑 RAG 闭环（重点在分块与元数据）。

• 当出现“高并发 + 持续增量入库 + 强过滤/权限/多租户 + 运维要求”时，再评估迁移到 Milvus 等向量数据库更合理。

说明：

方案 A：pip（优先尝试）

py -m pip install --upgrade pip
py -m pip install numpy
py -m pip install sentence-transformers
py -m pip install faiss-cpu

解释与自检要点：

• sentence-transformers：提供 Sentence-BERT 推理接口，首次运行会下载模型文件（需要网络与磁盘空间）。
• 自检（同一环境中执行）：

py -c "import faiss; import numpy as np; print('faiss_ok', getattr(faiss, '__version__', 'unknown')); print('numpy_ok', np.__version__)"
py -c "from sentence_transformers import SentenceTransformer; m=SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2'); print('model_dim', m.get_sentence_embedding_dimension())"

如果你使用本讲配套项目 11_faiss_kb_project/（推荐），也可以在项目目录用 requirements 一键安装：

cd .\11_faiss_kb_project
py -m pip install -r .\requirements.txt

方案 B：conda（当 pip 安装失败时）

conda install -c conda-forge faiss-cpu
conda install -c conda-forge sentence-transformers

解释与自检要点：

• conda-forge 往往更容易解决二进制依赖问题。
• 安装后仍按“方案 A 的自检命令”验证导入与模型维度。

目标：

配套项目实操：11_faiss_kb_project（与本实现对齐，推荐）

1）在课件目录运行建库脚本（会默认读取同目录下两份示例文档：设备手册 + 分拣规则）：

cd .\11_faiss_kb_project
py .\build_kb.py

2）自检：运行成功后，默认会生成：

• .\11_faiss_kb_project\faiss_store\index.faiss
• .\11_faiss_kb_project\faiss_store\chunks.jsonl

3）运行检索脚本（top-k 命中 + 输出证据字段）：

py .\query_kb.py --query "检测到霉斑应该怎么处理？" --top-k 5
py .\query_kb.py --query "告警码 E-101 表示什么？" --top-k 5

代码模板：构建索引、写入、持久化、再查询（单文件演示版）

from __future__ import annotations

import json
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Dict, List

import numpy as np
import faiss
from sentence_transformers import SentenceTransformer

@dataclass(frozen=True)
class Chunk:
    chunk_id: str
    text: str
    metadata: Dict[str, str]

def normalize_text(text: str) -> str:
    t = (text or "").replace("\r\n", "\n").replace("\r", "\n")
    lines = [ln.strip() for ln in t.split("\n")]
    lines = [ln for ln in lines if ln]
    return "\n".join(lines)

def chunk_by_length(text: str, *, chunk_size: int = 900, chunk_overlap: int = 140, metadata: Dict[str, str]) -> List[Chunk]:
    if chunk_size <= 0:
        raise ValueError("chunk_size must be > 0")
    if chunk_overlap < 0:
        raise ValueError("chunk_overlap must be >= 0")
    if chunk_overlap >= chunk_size:
        raise ValueError("chunk_overlap must be < chunk_size")

    t = normalize_text(text)
    if not t:
        return []
    step = chunk_size - chunk_overlap
    out: List[Chunk] = []
    start = 0
    idx = 0
    while start < len(t):
        end = min(start + chunk_size, len(t))
        out.append(Chunk(chunk_id=f"C-{idx:04d}", text=t[start:end], metadata=dict(metadata)))
        idx += 1
        start += step
    return out

def embed_texts(model: SentenceTransformer, texts: List[str]) -> np.ndarray:
    vec = model.encode(texts, batch_size=32, show_progress_bar=False, convert_to_numpy=True, normalize_embeddings=True)
    vec = np.ascontiguousarray(vec.astype("float32"))
    return vec

def build_faiss_ip_index(vectors: np.ndarray) -> faiss.Index:
    if vectors.ndim != 2:
        raise ValueError("vectors must be 2D array: (n, dim)")
    dim = int(vectors.shape[1])
    index = faiss.IndexFlatIP(dim)
    index.add(vectors)
    return index

def save_store(dir_path: Path, *, index: faiss.Index, chunks: List[Chunk]) -> None:
    dir_path.mkdir(parents=True, exist_ok=True)
    faiss.write_index(index, str(dir_path / "index.faiss"))
    (dir_path / "chunks.jsonl").write_text(
        "\n".join(json.dumps(asdict(c), ensure_ascii=False) for c in chunks),
        encoding="utf-8",
    )

def load_store(dir_path: Path) -> tuple[faiss.Index, List[Chunk]]:
    index = faiss.read_index(str(dir_path / "index.faiss"))
    lines = (dir_path / "chunks.jsonl").read_text(encoding="utf-8").splitlines()
    chunks = [Chunk(**json.loads(ln)) for ln in lines if ln.strip()]
    return index, chunks

def search(index: faiss.Index, chunks: List[Chunk], query_vec: np.ndarray, *, top_k: int = 5) -> List[dict]:
    if query_vec.ndim == 1:
        query_vec = query_vec.reshape(1, -1)
    scores, ids = index.search(query_vec.astype("float32"), top_k)
    out: List[dict] = []
    for score, i in zip(scores[0].tolist(), ids[0].tolist()):
        if i < 0:
            continue
        c = chunks[i]
        out.append(
            {
                "score": float(score),
                "chunk_id": c.chunk_id,
                "source": c.metadata.get("source", ""),
                "doc_type": c.metadata.get("doc_type", ""),
                "section": c.metadata.get("section", ""),
                "rule_id": c.metadata.get("rule_id", ""),
                "alarm_code": c.metadata.get("alarm_code", ""),
                "text_preview": c.text[:160].replace("\n", " "),
            }
        )
    return out

if __name__ == "__main__":
    doc_text = """
    1.1 分拣规则概述
    若果径 >= 80mm 且表面瑕疵面积 < 1%，判定为 A 级；
    若果径 >= 75mm 且表面瑕疵面积 < 3%，判定为 B 级；
    否则判定为 C 级。

    1.2 例外条款
    当检测到霉斑时，直接判定为 D 级并进入人工复检通道。
    """

    chunks = chunk_by_length(
        doc_text,
        chunk_size=220,
        chunk_overlap=40,
        metadata={"source": "apple_rules_demo.md", "doc_type": "rules", "section": "1", "rule_id": "1.1-1.2"},
    )

    model_name = "paraphrase-multilingual-MiniLM-L12-v2"
    model = SentenceTransformer(model_name)

    vectors = embed_texts(model, [c.text for c in chunks])
    index = build_faiss_ip_index(vectors)

    store_dir = Path("./faiss_store")
    save_store(store_dir, index=index, chunks=chunks)

    index2, chunks2 = load_store(store_dir)
    q = "霉斑应该分到哪个等级？"
    qv = embed_texts(model, [q])[0]
    hits = search(index2, chunks2, qv, top_k=5)
    print(json.dumps({"query": q, "hits": hits}, ensure_ascii=False, indent=2))

解释与自检要点：

• normalize_embeddings=True：把向量做归一化，后续用 IndexFlatIP 的内积检索就等价余弦相似度；这能减少“向量长度差异”带来的干扰。
• chunks.jsonl：把 chunk 与 metadata 持久化，避免“只存向量不存来源”的不可追溯问题。
• 自检 1：运行脚本后，目录下应生成 faiss_store/index.faiss 与 chunks.jsonl。
• 自检 2：查询 霉斑 应命中“例外条款”相关 chunk，并输出包含 source/rule_id 的结果。

你是 RAG（检索增强生成）工程师。请审计并优化我的“分块 + 嵌入 + FAISS 检索”逻辑，要求给出可以直接落地的改进点。

现状：
1）分块：按长度滑窗，chunk_size={...} overlap={...}；metadata 目前只有 source/doc_type/section/rule_id/alarm_code。
2）嵌入：SentenceTransformer({模型名})，normalize_embeddings=True。
3）索引：FAISS IndexFlatIP；持久化 index.faiss + chunks.jsonl。

请输出：
A）至少 6 条改进建议（按优先级排序），每条建议都要包含“原因 + 具体做法 + 如何验收”。
B）针对分拣规则文档，给出 3 条你认为最关键的 metadata 字段设计，并解释它们如何提升可追溯性与检索质量。
C）列出 5 个最容易踩坑的点（例如归一化/维度不一致/重复写入/噪声文本等），并给出快速排查方法。

我的分拣文档片段：{粘贴 2~3 条规则 + 1 段手册}

校验点（你们要人工审计）：

• AI 是否提出“去噪/去重/脱敏/保留规则编号”的工程建议，而不是只给“换大模型”。

课程思政融入点（质量与责任：工业智能系统开发底线）

• 模块化设计意识：文档解析、分块、嵌入、索引、检索必须解耦；任何一环可替换、可回归测试，避免“改一处坏一片”。

• 质量意识：检索结果必须可追溯、可复验；不能用“看起来像”的回答替代证据。

• 责任意识：工业系统的规则解释与设备告警涉及生产安全与质量追责；对数据来源、脱敏、日志留存与结果审计负责。

• 收集并解析分拣系统相关文档（如苹果分拣规则文档/设备手册），输出可读文本与结构信息。

• 设计并实现文本分块逻辑（至少 1 种：按长度），并说明分块参数与理由。

• 安装并配置 FAISS 向量数据库（本地索引），完成向量写入与持久化。

• 实现文本向量嵌入并写入数据库，完成至少 3 条查询的 top-k 检索验证（必须输出来源证据字段）。

• 使用 AI 设计 1 种分块方案，对比自研方案并做 1 次优化迭代（记录差异与理由）。

• 记录文本处理与数据库搭建的核心步骤（命令、脚本运行输出、截图/日志）。

大模型任务（教师可发放，学生可复用）

• 生成分拣系统文档解析脚本（按文档类型区分策略，输出结构化条目）。
• 生成 FAISS 向量数据库配置与使用脚本（含持久化与查询示例）。
• 优化向量嵌入逻辑（模型选择、归一化、去噪、去重、metadata 设计），提升检索准确性。
• 解答文本分块与向量嵌入中的技术疑问（要求给出“原因 + 复现 + 修复 + 回归验证”）。

作业（按要求布置）

1）提交文本分块代码及分块后的文档片段，附分块方案说明（含 AI 优化建议）。

2）提交 FAISS 数据库配置截图及向量嵌入测试日志（含嵌入成功截图）。

3）记录文本处理过程中遇到的 1 个问题，详细描述问题现象、排查过程与解决方案。