7.2 实现索引管线

7.2 实现索引管线

本节摘要：造第一件新零件 pipeline.py，把 3.2 切分、0.2 嵌入、4.1 向量库、5.1 词法索引串成一条可复现的管线。全量建库：扫描（共享忽略清单）→ 切分（max_lines 可调，7.4 要扫它）→ 嵌入（分批过 embed()）→ 双路持久化——向量进 npz、带正文的元数据进 metas.jsonl、每个文件的 mtime/哈希/chunk_ids 进 state.json（M1 的证据链）。增量更新：4.2 的两级判定落地——mtime 未变零成本跳过，mtime 变了再算 SHA-256 拦假变化；真变化的文件按 4.2 的"先删后插"处理，但取文件级粒度（变化文件的重切结果整体重嵌，同文件未变的方法不再单独省）——代码减半、语义不损，教学优先；向量侧用 numpy 掩码压实存活行、只嵌新增。三个设计决定（稳定 ID 含行号、正文必须持久化、model_ver 保险丝）逐一讲透。BM25 不持久化：查询侧按 metas.jsonl 重建（教学规模下毫秒级，生产再 pickle）。

学习目标

阅读完本节，你应当能够：

1. 运行 pipeline.py 完成全量建库，说出四个落盘产物各自的用途；
2. 解释增量更新的两级判定与"文件级先删后插"的取舍；
3. 说出三个设计决定的理由：ID 含行号、正文持久化、model_ver 保险丝；
4. 验证 M1：同仓库二次运行只重嵌变化文件。

一、全量建库

# pipeline.py — codeqa 索引管线：全量建库 + 增量更新（复用 3.2/4.1/4.2/5.1 部件）
import hashlib
import json
import sys
from pathlib import Path

import numpy as np

from chunker import chunk_file
from mini_store import MiniVectorStore

IGNORE = {".git", "__pycache__", "node_modules", ".venv", "dist"}
MODEL_VERSION = "my-code-embed-v1"      # 换嵌入模型必改：4.2 的版本保险丝

def iter_files(root: str, pattern: str = "**/*.py") -> list[str]:
    return [str(p).replace("\\", "/") for p in sorted(Path(root).rglob(pattern))
            if not (IGNORE & set(p.parts))]

def chunk_text(c) -> str:
    """身份前缀 + 正文（3.3 症状三修法；词法与语义两路共用同一文本）。"""
    prefix = f"# {c.kind} {c.symbol}"
    if c.parent:
        prefix += f" of class {c.parent}"
    return f"{prefix}\n{c.text}"

def make_meta(path: str, c) -> dict:
    return {"id": f"{path}:{c.start_line}-{c.end_line}:{c.symbol}",  # 4.2 稳定 ID
            "path": path, "symbol": c.symbol, "kind": c.kind, "parent": c.parent,
            "start": c.start_line, "end": c.end_line, "text": chunk_text(c)}

def build(root: str, out: str, embed, max_lines: int = 80) -> list[dict]:
    """全量建库：扫描 → 切分 → 嵌入 → 双路持久化 + state 落盘。"""
    metas, state = [], {}
    for path in iter_files(root):
        raw = Path(path).read_bytes()
        chunks = [make_meta(path, c) for c in chunk_file(path, max_lines=max_lines)]
        metas.extend(chunks)
        state[path] = {"mtime": Path(path).stat().st_mtime,
                       "sha256": hashlib.sha256(raw).hexdigest(),
                       "chunk_ids": [m["id"] for m in chunks],
                       "model_ver": MODEL_VERSION}
    if not metas:
        sys.exit(f"{root} 下没有可索引的 .py 文件")
    _save(out, metas, embed, vecs=None)
    _save_state(out, state)
    return metas
​

落盘产物四件，各司其职：code_index.npz（归一化向量矩阵）、code_index.jsonl（向量库的精简元数据，与 npz 行对齐）、metas.jsonl（带正文的完整元数据——重排要正文、BM25 重建要正文、评测对齐要 id，全靠它）、state.json（增量判定与版本保险丝的账本）。

二、统一落盘：为什么正文进 metas 而不进向量库

# pipeline.py 续 — 落盘与读取
def _save(out: str, metas: list[dict], embed, vecs=None):
    """vecs=None 全量嵌入；传入矩阵则直接使用（增量路径只嵌过新增）。"""
    Path(out).mkdir(parents=True, exist_ok=True)
    if vecs is None:
        store = MiniVectorStore(dim=len(embed([metas[0]["text"]])[0]))
        for i in range(0, len(metas), 64):              # 分批嵌入（4.1 的纪律）
            batch = metas[i:i + 64]
            store.add(embed([m["text"] for m in batch]),
                      [{k: v for k, v in m.items() if k != "text"} for m in batch])
    else:
        store = MiniVectorStore(dim=vecs.shape[1])
        store.vecs = np.asarray(vecs, dtype=np.float32)
        store.meta = [{k: v for k, v in m.items() if k != "text"} for m in metas]
    store.save(f"{out}/code_index")
    Path(f"{out}/metas.jsonl").write_text(
        "\n".join(json.dumps(m, ensure_ascii=False) for m in metas), encoding="utf-8")

def load_metas(out: str) -> list[dict]:
    return [json.loads(l) for l in
            Path(f"{out}/metas.jsonl").read_text(encoding="utf-8").splitlines()]

def _load_state(out: str) -> dict:
    p = Path(out) / "state.json"
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}

def _save_state(out: str, state: dict):
    Path(out).mkdir(parents=True, exist_ok=True)
    (Path(out) / "state.json").write_text(
        json.dumps(state, ensure_ascii=False, indent=1), encoding="utf-8")
​

mini_store 的 meta 不带正文是有意的（向量库只管存与查）；正文是检索之外多张嘴要吃的粮——重排要 (query, text) 对、BM25 重建要文本、6.3 的 snippet 要片段头、第 9 章评测要对 id。单一事实源放 metas.jsonl，其余视图按需投影。

三、增量更新：两级判定 + 文件级先删后插

# pipeline.py 续 — 增量更新（4.2 思想的教学版落地）
def update(root: str, out: str, embed, max_lines: int = 80) -> list[dict]:
    """mtime→哈希两级判定；真变化文件整体重切重嵌；向量只嵌新增。"""
    state = _load_state(out)
    old_metas = load_metas(out)
    if state and any(v.get("model_ver") != MODEL_VERSION for v in state.values()):
        sys.exit("嵌入模型版本与索引不一致：请删除索引目录后全量重建（4.2 保险丝）")

    files = set(iter_files(root))
    dead_ids: set[str] = set()
    fresh: list[dict] = []
    for path in sorted(files):
        st, old = Path(path).stat(), state.get(path)
        if old and old["mtime"] == st.st_mtime:
            continue                                   # 第 1 级：零成本跳过
        raw = Path(path).read_bytes()
        if old and old["sha256"] == hashlib.sha256(raw).hexdigest():
            state[path]["mtime"] = st.st_mtime         # 第 2 级：touch 类假变化
            continue
        if old:
            dead_ids.update(old["chunk_ids"])          # 真变化：旧 chunk 全部失效
        chunks = [make_meta(path, c) for c in chunk_file(path, max_lines=max_lines)]
        fresh.extend(chunks)                           # 文件级先删后插（见下注）
        state[path] = {"mtime": st.st_mtime,
                       "sha256": hashlib.sha256(raw).hexdigest(),
                       "chunk_ids": [m["id"] for m in chunks],
                       "model_ver": MODEL_VERSION}
    for path in [p for p in list(state) if p not in files]:   # 已删除的文件
        dead_ids.update(state[path]["chunk_ids"])
        del state[path]

    metas = [m for m in old_metas if m["id"] not in dead_ids] + fresh
    old_vecs = np.load(f"{out}/code_index.npz")
    alive = [i for i, m in enumerate(old_metas) if m["id"] not in dead_ids]
    vecs = old_vecs[alive]                             # 压实存活行（numpy 掩码）
    if fresh:                                          # 只嵌新增（增量省的就是这步）
        fv = np.asarray(embed([m["text"] for m in fresh]), dtype=np.float32)
        vecs = np.vstack([vecs, fv / np.linalg.norm(fv, axis=1, keepdims=True)])
    _save(out, metas, embed, vecs=vecs)
    _save_state(out, state)
    print(f"增量完成：{len(old_metas)} → {len(metas)} chunk（重嵌 {len(fresh)}）")
    return metas
​

文件级 vs chunk 级的取舍：4.2 的差分做到 chunk 级（同文件未变的方法不重嵌）；这里降为文件级（变化文件整体重嵌）——真实编辑常常牵动多行多个函数，chunk 级省的百分比有限，代码却能减半（教学规模下单文件重嵌也就几百毫秒）。生产上大文件多时再升级回 chunk 级：把 fresh.extend(chunks) 改成按 id 比对旧集合即可，骨架不变。

四、CLI 入口与 M1 验证

# pipeline.py 续 — 命令行入口
if __name__ == "__main__":
    import argparse
    ap = argparse.ArgumentParser(description="codeqa 索引管线")
    ap.add_argument("repo", help="要索引的仓库路径")
    ap.add_argument("--out", default="index")
    ap.add_argument("--max-lines", type=int, default=80, help="chunk 行数上限（3.1/3.3）")
    args = ap.parse_args()

    if (Path(args.out) / "metas.jsonl").exists() and (Path(args.out) / "state.json").exists():
        update(args.repo, args.out, embed, max_lines=args.max_lines)  # embed：0.2 入口
    else:
        chunks = build(args.repo, args.out, embed, max_lines=args.max_lines)
        print(f"全量建库完成：{len(chunks)} 个 chunk")
​

M1 验证三步（embed() 为第 0.2 节统一入口）：

1. python pipeline.py ./my_repo --out index——全量，记下 chunk 数与耗时；
2. 原样再跑一次——输出应为"重嵌 0"（两级判定全拦下）；
3. 改一个文件再加一行注释再跑——输出应为"重嵌 = 该文件的 chunk 数"，其余文件零触碰。

⚠️ 两个坑：①state.json 的 mtime 精度在部分文件系统上是秒级——连续两次快速保存可能被判"未变"，教学场景手动改文件再跑通常无感，遇到就对一次 sha256 兜底（或比较时加容差）；②删除文件靠"state 里有、扫描没有"反推——iter_files 的忽略清单若与建库时不同步，会误判整目录删除，IGNORE 动过就必须全量重建。

💡 BM25 为什么不持久化：词法索引从 metas.jsonl 重建在几万 chunk 内是亚秒级（经验值），教学管线少一个序列化格式就少一个版本坑；等仓库大到重建可感（十万 chunk 以上），把 code_tokenize 结果 pickle 进 index/ 即可，接口不变。

本节要点回顾

1. 四个落盘产物：npz（向量）、向量库 jsonl（精简 meta）、metas.jsonl（正文单一事实源）、state.json（增量账本+版本保险丝）；
2. 增量两级判定（mtime 零成本、哈希定谳）+ 文件级先删后插 + numpy 掩码压实、只嵌新增；
3. 文件级差分是 4.2 chunk 级的粗化版：代码减半、骨架不变，规模大了再升级；
4. M1 三步验证：全量 → 原样重跑（重嵌 0）→ 改一文件（只重嵌它）。

索引可复现了。下一节造查询侧：Retriever 把两路召回、融合、重排、上下文组装收进一个类，ask.py 把它接上命令行。