文档处理与检索接口

学习目标

这一课给 P14 的资料库补上「查」的能力：文档解析后切分成带来源的块，入 SQLite FTS5 全文索引，检索接口返回命中片段及其出处，并且把空资料、无结果这些边界场景的行为定死。

问题场景

资料库有了 20 份文档，AI 助手回答课程问题时需要引用原文。你不能把 20 份全文塞给模型——要按相关性挑几个片段，而且每个片段必须能回答「出自哪份文档、第几块」。这就是检索接口的存在意义：切分保来源，查询带出处。

核心解释

切分（chunking）的两条底线：块要有顺序号（拼回原文、定位引用都靠它），来源字段跟着块走（document_id + order_index 存进每一行，而不是靠行号隐式对应）。

# chunking.py
def chunk_text(text: str, size: int = 500) -> list[dict]:
    """按固定长度切块；真实项目可先按标题分级再切。"""
    paras = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, buf, order = [], "", 0
    for p in paras:
        if len(buf) + len(p) > size and buf:
            chunks.append({"order_index": order, "text": buf})
            order += 1
            buf = ""
        buf += p + "\n\n"
    if buf:
        chunks.append({"order_index": order, "text": buf})
    return chunks

检索用 SQLite FTS5——本站 fts_doc 表的同款技术：

# search.py
def build_index(conn, doc_id: str, collection: str, chunks: list[dict]):
    conn.execute("CREATE VIRTUAL TABLE IF NOT EXISTS doc_fts USING fts5(document_id, order_index, text)")
    with conn:
        conn.execute("DELETE FROM doc_fts WHERE document_id = ?", (doc_id,))  # 重建前先清
        for c in chunks:
            conn.execute("INSERT INTO doc_fts (document_id, order_index, text) VALUES (?, ?, ?)",
                         (doc_id, c["order_index"], c["text"]))

def search(conn, query: str, limit: int = 5) -> list[dict]:
    rows = conn.execute(
        "SELECT document_id, order_index, text, bm25(doc_fts) AS score "
        "FROM doc_fts WHERE doc_fts MATCH ? ORDER BY score LIMIT ?", (query, limit)
    ).fetchall()
    return [dict(r) for r in rows]

设计要点四个：

1. FTS5 的 MATCH 语法不是 LIKE。MATCH 'fastapi 校验' 走倒排索引，规模上去后比 LIKE 全表扫快几个数量级；本站 fts_doc 用 unicode61 分词器兼顾中文。
2. 重建走「先删后插」。文档重新解析时，同一 document_id 的旧块必须清掉再插，否则检索出过期内容——本站上传接口里 DELETE FROM doc_chunk WHERE document_id=? 就是这一步。
3. bm25() 是相关性分数。FTS5 内置排序函数，越小越相关；返回给前端时最好转成正数并附上文档标题。
4. 空态是正常态。资料集为空、查询无命中，都返回 {"total": 0, "items": []} 加 200，不是 500 也不是 404——前端「没有找到相关内容」的空态 UI 靠这个约定。

示例拆解

检索接口把来源信息拼完整：

@app.get("/search")
def search_docs(q: str = Query(min_length=1, max_length=100), limit: int = Query(5, ge=1, le=20)):
    hits = search(get_conn(), q, limit)
    items = []
    for h in hits:
        doc = get_doc(h["document_id"])
        items.append({                     # 来源三件套：哪份文档、第几块、原文
            "document_id": h["document_id"], "doc_title": doc["name"],
            "order_index": h["order_index"], "text": h["text"],
        })
    return {"total": len(items), "items": items}

验证三态：资料库空时查询返回 total=0；导入一份「FastAPI 校验」的 md 后查询 q=校验 命中并带出文档名和块序号；换一个不存在的词 q=量子力学 依然 total=0。AI 助手拿到这样的结果，就能在回答里标注「出自《FastAPI 校验》第 2 块」——这就是本站 AI 回复带课程引用的同款机制。

动手任务

1. 给 chunk_text 加标题感知：以 # 开头的行重置块边界，让块不跨章节。
2. 实现 reindex_document(doc_id)：重新解析并「先删后插」重建索引，重复执行两次验证结果一致。
3. 给 /search 加 collection 筛选参数，验证只在该资料集范围内命中。

常见错误

• 块不带 order_index：引用时无法定位「第几段」，也没法按序拼回原文；顺序信息必须落库。
• 重建索引不先删旧行：同一文档新旧块并存，检索返回过期内容；先 DELETE 再 INSERT。
• 对 FTS 用 LIKE 思维：MATCH 是语法敏感的，用户输入含特殊字符会报错；查询词要做转义或清洗。
• 无结果抛异常：把「查不到」当 500 处理，前端空态永远画不出来；空结果是 200 加空数组。

检查与复盘

自查清单：

• 每个检索结果都带 document_id、块序号和文档标题（来源三件套）
• 重复重建索引不会产生重复或过期内容
• 空资料集、无命中两个场景返回的是稳定空态而不是错误

参考资料

• SQLite 官方文档：FTS5 —— 建表、MATCH 与 bm25
• sqlite3 标准库文档 —— Python 侧的执行与参数绑定