lllin000_PaperForge/docs/plans/memory-layer-implementation-plan.md

Memory Layer 改造 — 实施计划

基于：2026-07-06-memory-layer-overhaul-design.md 状态：待执行 策略：5 个 PR 顺序执行，PR n+1 依赖 PR n

---

总依赖图

PR 1: 结构树 + body_units 正确性
  │
  ▼
PR 2: Memory Builder 增量
  │
  ▼
PR 3: Content Discovery 行为（语义依赖 PR 2，PR 2 后做）
  │
  ▼
PR 4: Embedding 双 collection（依赖 PR 1 + PR 2）
  │
  ▼
PR 5: Retrieve 合并（依赖 PR 4）

---

PR 1：结构树和 body_units 正确性（最核心，不可拆分）

目标： 从 rendered markdown H1-H6 产出真实嵌套结构树，body_units 跟随树递归生成，不重复、不丢内容。

文件变更

操作	文件	改动
新增	paperforge/core/io.py	read_jsonl(path) — jsonl 逐行读取（待加）
修改	paperforge/worker/ocr_render.py	render_fulltext_markdown() 新增 return_events=False 参数；为 True 时返回 RenderOutput(markdown, heading_events, emitted_block_events)。不破坏旧返回类型。
修改	paperforge/worker/ocr_rebuild.py Phase 4/5	Phase 4 传 return_events=True → RenderOutput 传入 Phase 5
修改	paperforge/worker/ocr.py Phase 6	同上顺序调整
新增	render 阶段产出	render/render-map.json：{headings: [...], emitted_blocks: [...]}
重写	paperforge/retrieval/structure_tree.py	build_structure_tree(heading_events, emitted_block_events, structured_blocks) — stack 算法，interval 基于 emitted order
修改	paperforge/retrieval/units.py	build_body_units() / build_object_units() recursive walk，own_block_ids，role helper
修改	paperforge/memory/schema.py	body_units 表加字段；CURRENT_SCHEMA_VERSION → 4

实现要点

1.1 RenderOutput 与 render-map.json

不直接改 render_fulltext_markdown() 返回类型——当前返回 str，旧调用点很多。新增 return_events=False 参数：

@dataclass
class RenderOutput:
    markdown: str
    heading_events: list[dict]         # heading block 事件
    emitted_block_events: list[dict]   # 所有真正进入正文阅读流的 block

def render_fulltext_markdown(..., return_events=False) -> str | RenderOutput:
    ...
    if return_events:
        return RenderOutput(markdown=markdown,
                            heading_events=heading_events,
                            emitted_block_events=emitted_block_events)
    return markdown

heading_events 每条记录：

{"line_number": 42, "markdown_level": 3, "title": "Study population",
 "page": 2, "block_id": "b_10", "emitted_order": 7}

emitted_block_events 每条记录（每个真正进入正文阅读流的 block）：

{"emitted_order": 123, "line_start": 456, "line_end": 460,
 "page": 3, "block_id": "b_10", "role": "body_paragraph",
 "emitted_as": "body"}

写入 render/render-map.json：

{"headings": [...], "emitted_blocks": [...]}

build_structure_tree() 不再依赖 structured_blocks 的 _emitted_order（该字段不存在），而是接收 emitted_block_events。

1.2 Phase 4/5 顺序和数据流

× 当前:
  Phase 4: render_fulltext_markdown(...) → markdown (str)
  Phase 5: build_role_indexes
           build_structure_tree(structured)     ← 此时无 heading／emitted 信息
           write_render_outputs(...)

✓ 改为:
  Phase 4: rendered = render_fulltext_markdown(..., return_events=True)
           return rendered, health_overall

  Phase 5: write_render_outputs(..., markdown=rendered.markdown,
                                 heading_events=rendered.heading_events,
                                 emitted_block_events=rendered.emitted_block_events)
           build_role_indexes(...)
           build_structure_tree(
               heading_events=rendered.heading_events,
               emitted_block_events=rendered.emitted_block_events,
               structured_blocks=structured,
           )
           write_structure_tree(index/structure-tree.json)

两处都改：

• paperforge/worker/ocr_rebuild.py — rebuild 路径
• paperforge/worker/ocr.py — 初始 OCR 路径

write_render_outputs() 新增参数 heading_events / emitted_block_events，写入 render/render-map.json。

1.3 build_structure_tree() 使用 emitted_block_events

def build_structure_tree(heading_events, emitted_block_events, structured_blocks):
    # 1. heading_events 按 emitted_order 排序
    heading_events.sort(key=lambda h: h["emitted_order"])

    # 2. 建立 block_id → structured_block 的映射
    block_map = {b["block_id"]: b for b in structured_blocks}

    # 3. stack 建树
    stack, root_nodes = [], []
    for h in heading_events:
        node = {
            "node_id": f"sec:{h['block_id']}",
            "kind": "section",
            "title": h["title"],
            "level": h["markdown_level"],
            "page_span": [h["page"], h["page"]],
            "block_id": h["block_id"],
            "own_block_ids": [],
            "subtree_block_ids": [],
            "children": [],
            "objects": [],
        }
        while stack and stack[-1]["level"] >= h["markdown_level"]:
            stack.pop()
        if stack:
            stack[-1]["children"].append(node)
        else:
            root_nodes.append(node)
        stack.append(node)

    # 4. 计算 block intervals（基于 emitted_block_events）
    _assign_block_intervals(root_nodes, heading_events, emitted_block_events)
    return {"paper_id": ..., "nodes": root_nodes}

1.4 block interval 计算（基于 emitted order）

def _assign_block_intervals(nodes, heading_events, emitted_block_events):
    # 构建每个 heading 的 emitted_order 边界
    bounds_map = {}
    for i, h in enumerate(heading_events):
        next_sibling = None
        for h2 in heading_events[i+1:]:
            if h2["markdown_level"] <= h["markdown_level"]:
                next_sibling = h2["emitted_order"]
                break
        bounds_map[h["block_id"]] = {
            "start": h["emitted_order"],
            "end": next_sibling or float("inf"),
        }

    def compute(node):
        bounds = bounds_map.get(node["block_id"])
        if not bounds:
            return
        # subtree = bounds 范围内的所有 emitted blocks
        node["subtree_block_ids"] = [
            e["block_id"] for e in emitted_block_events
            if bounds["start"] <= e["emitted_order"] < bounds["end"]
        ]
        for child in node.get("children", []):
            compute(child)
        # own = subtree − children subtree − heading blocks
        child_ids = set()
        for child in node.get("children", []):
            child_ids.update(child.get("subtree_block_ids", []))
        node["own_block_ids"] = [
            bid for bid in node["subtree_block_ids"]
            if bid not in child_ids and bid != node["block_id"]
        ]

    for n in nodes:
        compute(n)

1.5 build_body_units() 递归

def _body_unit_role_kind(role):
    if role == "body_paragraph": return "body"
    if role in {"structured_insert", "non_body_insert", "backmatter_body"}:
        return "backmatter_body"
    return None  # reference_item, reference_heading, heading, caption, asset → excluded

def build_body_units(tree, structured_blocks):
    units = []
    block_map = {b["block_id"]: b for b in structured_blocks}

    def walk(node, path):
        this_path = path + [node["title"]]
        own = [block_map[bid] for bid in node.get("own_block_ids", [])
               if bid in block_map
               and _body_unit_role_kind(block_map[bid].get("role", ""))]
        if own:
            from itertools import groupby
            for kind, grp in groupby(own, key=lambda b: _body_unit_role_kind(b["role"])):
                if not kind: continue
                text = "\n\n".join(b.get("text","") for b in grp)
                parts = _split_if_oversized(text, 1000)
                n = len(parts)
                for pi, pt in enumerate(parts):
                    ord_ = pi + 1 if n > 1 else 0
                    suf = f":part_{ord_:03d}" if ord_ else ""
                    units.append({
                        "unit_id": f"{paper_id}:body:{node['node_id']}{suf}",
                        "paper_id": paper_id,
                        "section_path": "/".join(this_path),
                        "section_path_json": json.dumps(this_path),
                        "section_level": node["level"],
                        "section_title": node["title"],
                        "page_span": node["page_span"],
                        "unit_text": pt,
                        "token_estimate": len(pt)//4,
                        "unit_kind": kind,
                        "part_ordinal": ord_,
                        "indexable": True,
                        "veto_reason": "",
                        "quality_hints": [],
                    })
        for child in node.get("children", []):
            walk(child, path)

    for root in tree.get("nodes", []):
        walk(root, [])
    return units

1.6 build_object_units() own_block_ids

与 body_units 同理，只取 own_block_ids 范围内 role 为 figure/table/object 的 blocks。一个 object 只归属一个最具体的 section。

1.7 schema 升级

-- CURRENT_SCHEMA_VERSION: 3 → 4
ALTER TABLE body_units ADD COLUMN section_path_json TEXT NOT NULL DEFAULT '[]';
ALTER TABLE body_units ADD COLUMN section_level INTEGER NOT NULL DEFAULT 0;
ALTER TABLE body_units ADD COLUMN section_title TEXT NOT NULL DEFAULT '';
ALTER TABLE body_units ADD COLUMN part_ordinal INTEGER NOT NULL DEFAULT 0;

测试清单

1. duplicate heading title 通过 block_id 区分，不通过 title match
2. H2/H3/H3 sibling 产生正确 section_path（无跨级错误）
3. parent intro + child + parent summary:
   parent.own_block_ids = [intro_para, summary_para]（不含 child 内容）
4. child body 不重复出现在 parent unit
5. empty container 不产生 body unit
6. reference_item 不进入 body_units
7. structured_insert 进入 backmatter_body unit
8. mixed body + backmatter: 拆成两个 unit，不同 unit_kind
9. token cap: >1000 tokens 拆 part_001 / part_002
10. object 只归属最具体 section
11. rendered order ≠ structured order: tree interval 基于 emitted order
12. consumed object blocks 不进入 own_block_ids

---

PR 2：Memory Builder 增量

目标： memory build 不再被 canonical_index_hash 锁定，per-paper 检测 OCR 结果变化后增量重建 body_units。

文件变更

操作	文件	改动
修改	paperforge/memory/builder.py	结构化路径修正；不 early return；per-paper ocr_result_hash 增量；三级 fallback；schema v4 兼容
修改	paperforge/memory/schema.py	ensure_schema 支持 v4 upgrade path

实现要点

2.1 路径修正

# 改前
structured_path = ocr_dir / "structured-blocks.json"
structured_blocks = read_json(structured_path)

# 改后
structured_path = ocr_dir / "structure" / "blocks.structured.jsonl"
structured_blocks = read_jsonl(structured_path)

2.2 不 early return

# 改前
if canonical_hash matches and db_path.exists():
    return {"papers_indexed": N, "hash_match": True}

# 改后
index_changed = not _hash_matches(...)
if index_changed:
    _rebuild_metadata_tables(conn, items)    # papers / aliases / FTS
    _rebuild_all_units(conn, items, vault)   # body + object + manifest
else:
    _incremental_units_only(conn, items, vault)  # only body/object units

2.3 _incremental_units_only()

def _incremental_units_only(conn, items, vault):
    ocr_root = vault / "System" / "PaperForge" / "ocr"
    if not ocr_root.exists():
        return
    for entry in items:
        key = entry["zotero_key"]
        paper_dir = ocr_root / key
        tree_path = paper_dir / "index" / "structure-tree.json"
        blocks_path = paper_dir / "structure" / "blocks.structured.jsonl"
        if not tree_path.exists() or not blocks_path.exists():
            continue
        current_hash = _resolve_ocr_result_hash(paper_dir)
        row = conn.execute(
            "SELECT value FROM meta WHERE key=?", (f"manifest:{key}",)
        ).fetchone()
        if row:
            stored = json.loads(row[0])
            if stored.get("ocr_result_hash") == current_hash:
                continue
        _rebuild_paper_units(conn, key, paper_dir, tree_path, blocks_path)

2.4 _resolve_ocr_result_hash() 三级 fallback

def _resolve_ocr_result_hash(paper_dir):
    # 1. index/result-hash.txt
    rp = paper_dir / "index" / "result-hash.txt"
    if rp.exists():
        return rp.read_text().strip()
    # 2. hash of structured artifacts
    h = hashlib.sha256()
    for rel in ["structure/blocks.structured.jsonl", "index/structure-tree.json",
                 "index/role-index.json"]:
        p = paper_dir / rel
        if p.exists():
            h.update(p.read_bytes())
    if h.hexdigest() != hashlib.sha256(b"").hexdigest():
        return h.hexdigest()
    # 3. meta derived_version
    meta_p = paper_dir / "meta.json"
    if meta_p.exists():
        dv = json.loads(meta_p.read_text()).get("derived_version", {})
        return hashlib.sha256(json.dumps(dv, sort_keys=True).encode()).hexdigest()
    return ""

2.5 _rebuild_paper_units() 删除 + _upsert_body_units() 只插入

职责分离： _rebuild_paper_units() 负责 DELETE；_upsert_body_units() 只 INSERT + FTS。

def _rebuild_paper_units(conn, key, paper_dir, tree_path, blocks_path):
    # ── 删除（一个地方做）──
    conn.execute("DELETE FROM body_units WHERE paper_id = ?", (key,))
    conn.execute("DELETE FROM body_units_fts WHERE paper_id = ?", (key,))
    conn.execute("DELETE FROM object_units WHERE paper_id = ?", (key,))

    # ── 构建 ──
    tree = json.loads(tree_path.read_text())
    blocks = read_jsonl(blocks_path)
    from paperforge.retrieval.units import build_body_units, build_object_units
    body_units = build_body_units(tree=tree, structured_blocks=blocks)
    object_units = build_object_units(tree=tree, structured_blocks=blocks, ...)

    # ── 插入 + FTS（不做 DELETE）──
    _upsert_body_units(conn, body_units)
    _upsert_object_units(conn, object_units)

    # ── manifest ──
    from paperforge.retrieval.manifest import build_paper_manifest
    manifest = build_paper_manifest(paper_id=key, ...)
    _write_manifest_row(conn, manifest)

_upsert_body_units() 只 INSERT + FTS，DELETE 已在 _rebuild_paper_units() 完成：

def _upsert_body_units(conn, body_units):
    for unit in body_units:
        conn.execute("""INSERT INTO body_units (...) VALUES (...)""", ...)
    paper_ids = list({u["paper_id"] for u in body_units})
    for pid in paper_ids:
        conn.execute("""INSERT INTO body_units_fts(...)
                        SELECT rowid, unit_id, paper_id, section_path, unit_text
                        FROM body_units
                        WHERE paper_id = ? AND indexable = 1""", (pid,))

---

PR 3：Content Discovery 行为

目标： body_units_fts 做 primary 检索，无结果时不静默 fallback。

依赖： PR 2 后做（依赖 body_units 表稳定 + indexable 语义稳定 + schema v4）。

文件变更

操作	文件	改动
修改	paperforge/retrieval/gateway.py	_run_body_unit_discovery() 无结果时返回空 + coverage 提示

实现要点

def _run_body_unit_discovery(vault, query, limit=5):
    body_rows = _search_body_units_fts(vault, query, limit)
    coverage = _get_body_coverage(conn)
    if not body_rows:
        return PFResult(ok=True, data={
            "results": [],
            "coverage": coverage,
            "next_action": {
                "command": f"paperforge search {query}",
                "reason": (f"正文检索无匹配。正文索引覆盖 "
                           f"{coverage['body_papers']}/{coverage['ocr_papers']} 篇 OCR 完成论文。"
                           f"尝试 paperforge search 进行元数据全文搜索。")
            }
        })
    return PFResult(ok=True, data={"results": body_rows, "coverage": coverage})

def _get_body_coverage(conn):
    body = conn.execute("SELECT COUNT(DISTINCT paper_id) FROM body_units WHERE indexable=1").fetchone()[0]
    ocr = conn.execute("SELECT COUNT(*) FROM papers WHERE ocr_status='done'").fetchone()[0]
    lib = conn.execute("SELECT COUNT(*) FROM papers").fetchone()[0]
    return {"body_papers": body, "ocr_papers": ocr, "library_papers": lib}

---

PR 4：Embedding 双 collection

目标： 新建 paperforge_body collection，embed builder 从 DB 读 body_units。

文件变更

操作	文件	改动
修改	paperforge/embedding/_chroma.py	get_collection() 支持 name 参数；delete_paper_vectors() 双删
修改	paperforge/embedding/builder.py	新增 embed_body_units()；get_body_units_for_embedding() 读 DB
修改	paperforge/embedding/search.py	新增 merge_retrieve()
修改	paperforge/embedding/status.py	统计两个 collection
修改	paperforge/commands/embed.py	build 流程以 DB 分流、resume 检测 body_units_hash

实现要点

4.1 get_collection(name=)

def get_collection(vault, name="paperforge_fulltext"):
    db_path = get_vector_db_path(vault)
    client = chromadb.PersistentClient(str(db_path))
    return client.get_or_create_collection(
        name=name, metadata={"hnsw:space": "cosine"},
    )

4.2 embed build 分流（以 DB 为准）

是否走 body_units 路径以 DB body_units 表为准（Memory Builder 是唯一生产者），不重复检测文件。

def _has_body_units_in_db(vault, key) -> bool:
    db_path = get_memory_db_path(vault)
    if not db_path.exists():
        return False
    conn = get_connection(db_path, read_only=True)
    cnt = conn.execute(
        "SELECT COUNT(*) FROM body_units WHERE paper_id=? AND indexable=1",
        (key,)
    ).fetchone()[0]
    conn.close()
    return cnt > 0

for entry in papers:
    key = entry["zotero_key"]
    has_body = _has_body_units_in_db(vault, key)

    if has_body:
        body_units = get_body_units_for_embedding(vault, key)
        if not body_units:
            continue
        if resume:
            col = get_collection(vault, name="paperforge_body")
            # Chroma: 不依赖多字段 where 隐式 AND
            existing = col.get(where={"paper_id": key}, limit=1)
            metas = existing.get("metadatas", [])
            if metas and metas[0].get("body_units_hash") == current_hash:
                papers_skipped += 1; continue
        delete_paper_vectors(vault, key)
        embed_body_units(vault, key, body_units)
    else:
        # 文件存在但 DB 无 body_units → 提示先 memory build
        ocr_root = vault / "System" / "PaperForge" / "ocr" / key
        has_files = ((ocr_root / "structure" / "blocks.structured.jsonl").exists()
                     and (ocr_root / "index" / "structure-tree.json").exists())
        if has_files:
            print(f"Skip {key}: has structured blocks but no body_units in DB. "
                  f"Run `paperforge memory build` first.")
            continue
        # 旧路径
        fulltext_path = vault / (entry.get("fulltext_path") or "")
        if resume:
            col = get_collection(vault, name="paperforge_fulltext")
            existing = col.get(where={"paper_id": key}, limit=1)
            if existing.get("ids"):
                papers_skipped += 1; continue
        delete_paper_vectors(vault, key)
        chunks = chunk_fulltext(fulltext_path)
        if chunks:
            embed_paper(vault, key, chunks)

---

PR 5：Retrieve 合并

目标： 同时查两个 collection，unit-level 去重，per-paper cap。

文件变更

操作	文件	改动
新增	paperforge/embedding/search.py	merge_retrieve()
修改	paperforge/commands/retrieve.py	调 merge_retrieve() 替代 retrieve_chunks()

实现要点

RETRIEVAL_COLLECTIONS = ["paperforge_fulltext", "paperforge_body"]

def merge_retrieve(vault, query, limit=5, expand=True):
    provider = OpenAICompatibleProvider(vault)
    q_emb = provider.encode_single(query)
    n = limit * 2

    all_results = []
    for name in RETRIEVAL_COLLECTIONS:
        try:
            col = get_collection(vault, name=name)
            res = col.query(
                query_embeddings=[q_emb],
                n_results=n,
                include=["documents", "metadatas", "distances"],
            )
            for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0]):
                all_results.append({
                    "paper_id": meta.get("paper_id", ""),
                    "section_path": meta.get("section_path", ""),
                    "chunk_text": doc,
                    "score": round(1.0 - dist, 4),
                    "source": "legacy_chunk" if name == "paperforge_fulltext" else "body_unit",
                    "unit_id": meta.get("chunk_index", ""),
                })
        except Exception:
            continue

    # 排序 + unit-level 去重 + per-paper cap
    all_results.sort(key=lambda r: r["score"], reverse=True)
    seen = set()
    per_paper = {}
    merged = []
    for r in all_results:
        # dedupe key: (source, unit_id) 或 (source, paper_id+text_hash) 防空 unit_id
        dedupe_key = (r["source"], r["unit_id"]) if r.get("unit_id") else (
            r["source"], r["paper_id"], hash(r["chunk_text"])
        )
        if dedupe_key in seen:
            continue
        seen.add(dedupe_key)
        pid = r["paper_id"]
        if per_paper.get(pid, 0) >= 2:
            continue
        per_paper[pid] = per_paper.get(pid, 0) + 1
        merged.append(r)
        if len(merged) >= limit:
            break
    return merged

---

执行顺序

Step 1: PR 1（最核心，不可拆）
  → 验证: body_units 表有行，section_path 正确
  → 验证: rendered order ≠ structured order 时边界正确

Step 2: PR 2
  → 验证: rebuild 后 memory build 增量更新 body_units
  → 验证: content-discovery 返回正文结果

Step 3: PR 3（依赖 PR 2）
  → 验证: content-discovery 无结果时返回空 + coverage 提示

Step 4: PR 4
  → 验证: embed build 写入 paperforge_body
  → 验证: resume 正确跳过已嵌入的论文（通过 body_units_hash）

Step 5: PR 5
  → 验证: retrieve 返回合并结果，unit-level 去重 + per-paper cap