diff --git a/backend/app/knowledge/parser.py b/backend/app/knowledge/parser.py index 3ab4bf3..144c161 100644 --- a/backend/app/knowledge/parser.py +++ b/backend/app/knowledge/parser.py @@ -55,7 +55,7 @@ def parse_note( fallback_title = Path(file_path).stem title = frontmatter.get("title") or _first_heading(markdown) or fallback_title - resolved_tags = list(tags) if tags else _parse_tags(frontmatter.get("tags")) + resolved_tags = list(tags) if tags is not None else _parse_tags(frontmatter.get("tags")) blocks = parse_blocks(markdown, note_id) return ParsedNote( diff --git a/backend/app/repository.py b/backend/app/repository.py index ba3918c..7be9169 100644 --- a/backend/app/repository.py +++ b/backend/app/repository.py @@ -8,6 +8,7 @@ app/retrieval/vectorstore.py）。本层只负责 notes / blocks / blocks_fts from __future__ import annotations import json +import sqlite3 from dataclasses import dataclass, field from datetime import datetime @@ -63,6 +64,7 @@ class FtsHit: def replace_note_metadata( *, + conn: sqlite3.Connection, note_id: str, title: str, file_path: str, @@ -72,53 +74,49 @@ def replace_note_metadata( updated_at: datetime, blocks: list[NoteBlock], ) -> list[str]: - """整体替换一条笔记的元数据、Block 与 FTS5 索引（单事务）。 + """整体替换一条笔记的元数据、Block 与 FTS5 索引。 - 返回替换前的旧 block_id 列表，供调用方清理 vec_blocks 中已失效的向量。 + 不在此处开启/提交事务：由调用方（index_note）在同一连接上把「元数据 + 向量」包进 + 单个事务，保证原子性。返回替换前的旧 block_id 列表，供调用方清理失效向量。 """ - conn = connect() - try: - with transaction(conn): - old_block_ids = [ - row["block_id"] - for row in conn.execute("SELECT block_id FROM blocks WHERE note_id = ?", (note_id,)) - ] - conn.execute( - """ - INSERT INTO notes (note_id, title, file_path, folder, tags, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(note_id) DO UPDATE SET - title = excluded.title, - file_path = excluded.file_path, - folder = excluded.folder, - tags = excluded.tags, - updated_at = excluded.updated_at - """, - (note_id, title, file_path, folder, json.dumps(tags, ensure_ascii=False), - _iso(created_at), _iso(updated_at)), - ) - conn.execute("DELETE FROM blocks WHERE note_id = ?", (note_id,)) - conn.execute("DELETE FROM blocks_fts WHERE note_id = ?", (note_id,)) - for position, block in enumerate(blocks): - conn.execute( - """ - INSERT INTO blocks - (block_id, note_id, heading_path, start_offset, end_offset, - content, content_hash, token_count, position) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - (block.block_id, note_id, json.dumps(block.heading_path, ensure_ascii=False), - block.start_offset, block.end_offset, block.content, - block.content_hash, block.token_count, position), - ) - # FTS5 存分词后的可检索文本；原文仍由 blocks.content 保留用于展示 - conn.execute( - "INSERT INTO blocks_fts (block_id, note_id, heading_path, content) VALUES (?, ?, ?, ?)", - (block.block_id, note_id, segment(" ".join(block.heading_path)), segment(block.content)), - ) - return old_block_ids - finally: - conn.close() + old_block_ids = [ + row["block_id"] + for row in conn.execute("SELECT block_id FROM blocks WHERE note_id = ?", (note_id,)) + ] + conn.execute( + """ + INSERT INTO notes (note_id, title, file_path, folder, tags, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(note_id) DO UPDATE SET + title = excluded.title, + file_path = excluded.file_path, + folder = excluded.folder, + tags = excluded.tags, + updated_at = excluded.updated_at + """, + (note_id, title, file_path, folder, json.dumps(tags, ensure_ascii=False), + _iso(created_at), _iso(updated_at)), + ) + conn.execute("DELETE FROM blocks WHERE note_id = ?", (note_id,)) + conn.execute("DELETE FROM blocks_fts WHERE note_id = ?", (note_id,)) + for position, block in enumerate(blocks): + conn.execute( + """ + INSERT INTO blocks + (block_id, note_id, heading_path, start_offset, end_offset, + content, content_hash, token_count, position) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + (block.block_id, note_id, json.dumps(block.heading_path, ensure_ascii=False), + block.start_offset, block.end_offset, block.content, + block.content_hash, block.token_count, position), + ) + # FTS5 存分词后的可检索文本；原文仍由 blocks.content 保留用于展示 + conn.execute( + "INSERT INTO blocks_fts (block_id, note_id, heading_path, content) VALUES (?, ?, ?, ?)", + (block.block_id, note_id, segment(" ".join(block.heading_path)), segment(block.content)), + ) + return old_block_ids def delete_note(note_id: str) -> list[str]: @@ -216,17 +214,6 @@ def fts_search(match: str, limit: int = 100) -> list[FtsHit]: conn.close() -def fts_count(match: str) -> int: - """返回 FTS5 命中总数，用于分页 total（不受候选池截断影响）。""" - conn = connect() - try: - return conn.execute( - "SELECT COUNT(*) FROM blocks_fts WHERE blocks_fts MATCH ?", (match,) - ).fetchone()[0] - finally: - conn.close() - - def get_block_hits(block_ids: list[str]) -> list[BlockHit]: if not block_ids: return [] diff --git a/backend/app/retrieval/engine.py b/backend/app/retrieval/engine.py index 0009b50..09bffb2 100644 --- a/backend/app/retrieval/engine.py +++ b/backend/app/retrieval/engine.py @@ -29,6 +29,10 @@ from app.textutils import make_snippet, match_query CANDIDATE_POOL = 50 # 分页窗口上限：候选池至少覆盖 offset+limit，但设上限防止超大 offset 撑爆内存 MAX_CANDIDATE_POOL = 200 +# 带 metadata 过滤时放大召回倍数，缓解「先截断候选池再过滤」造成的漏召回 +OVERSCAN_FACTOR = 4 +# FTS 一次性取全量命中上限：保证 fts 模式 total 准确、过滤不漏召回；超出则截断 +FTS_FETCH_LIMIT = 1000 class RetrievalEngine: @@ -43,30 +47,35 @@ class RetrievalEngine: self.vector_store = vector_store async def search(self, request: SearchRequest) -> SearchResponse: + has_filters = bool( + request.folders or request.note_ids or request.tags + or request.created_from or request.created_to + or request.updated_from or request.updated_to + ) # 候选池至少覆盖本次请求的 offset+limit，保证分页能取到目标页；设上限防内存失控 window = min(request.offset + request.limit, MAX_CANDIDATE_POOL) pool_size = max(CANDIDATE_POOL, window) + # 带过滤时放大召回；FTS 则一次性取全量命中（≤FTS_FETCH_LIMIT）避免截断漏召回 + recall = min(pool_size * OVERSCAN_FACTOR, MAX_CANDIDATE_POOL) if has_filters else pool_size # 1. 按模式收集候选（FTS 与 Vector 各产出「按相关性降序」的 block_id 列表） fts_ranked: list[str] = [] vec_ranked: list[str] = [] fts_scores: dict[str, float] = {} vec_scores: dict[str, float] = {} - fts_total = 0 if request.mode in (SearchMode.fts, SearchMode.hybrid): match = match_query(request.query) if match: - fts_hits = repository.fts_search(match, pool_size) + fts_limit = FTS_FETCH_LIMIT if request.mode == SearchMode.fts else recall + fts_hits = repository.fts_search(match, fts_limit) fts_ranked = [h.block_id for h in fts_hits] # bm25 越小越相关，取反后统一为「越大越相关」 fts_scores = {h.block_id: -h.bm25 for h in fts_hits} - if request.mode == SearchMode.fts: - fts_total = repository.fts_count(match) if request.mode in (SearchMode.vector, SearchMode.hybrid): query_vec = await self.embedding.embed_query(request.query) - vec_hits = await self.vector_store.search(query_vec, top_k=pool_size) + vec_hits = await self.vector_store.search(query_vec, top_k=recall) vec_ranked = [v.id for v in vec_hits] vec_scores = {v.id: v.score for v in vec_hits} @@ -104,11 +113,9 @@ class RetrievalEngine: ordered = normalize_scores(ordered) - # 5. 分页：fts 用真实命中总数；vector/hybrid 为 KNN 候选集，无全局 total - if request.mode == SearchMode.fts: - total = fts_total - else: - total = len(ordered) + # 5. 分页：total = 过滤后候选集大小。fts 已取全量（≤FTS_FETCH_LIMIT）故为真实命中数； + # vector/hybrid 为 KNN 候选集，无全局 total。 + total = len(ordered) page = ordered[request.offset : request.offset + request.limit] items = [self._build_result(hits[block_id], request, score) for block_id, score in page] return SearchResponse( diff --git a/backend/app/retrieval/vectorstore.py b/backend/app/retrieval/vectorstore.py index 2dd50e4..99e5c1e 100644 --- a/backend/app/retrieval/vectorstore.py +++ b/backend/app/retrieval/vectorstore.py @@ -6,6 +6,8 @@ vec0 虚拟表返回的 distance 是欧氏距离（非平方）。入库前向 from __future__ import annotations +import sqlite3 +from contextlib import nullcontext from dataclasses import dataclass from typing import Protocol, runtime_checkable @@ -38,30 +40,36 @@ class VectorStore(Protocol): class SqliteVecStore: """sqlite-vec 默认实现。""" - async def upsert(self, records: list[VectorRecord]) -> None: + async def upsert(self, records: list[VectorRecord], *, conn: sqlite3.Connection | None = None) -> None: if not records: return - conn = connect() + owns = conn is None + conn = conn or connect() try: - with transaction(conn): + with transaction(conn) if owns else nullcontext(): for record in records: + # vec0 不支持 UPDATE，采用 delete-then-insert 实现幂等 upsert，避免主键冲突 + conn.execute("DELETE FROM vec_blocks WHERE block_id = ?", (record.id,)) conn.execute( "INSERT INTO vec_blocks (block_id, embedding) VALUES (?, ?)", (record.id, sqlite_vec.serialize_float32(record.vector)), ) finally: - conn.close() + if owns: + conn.close() - async def delete(self, ids: list[str]) -> None: + async def delete(self, ids: list[str], *, conn: sqlite3.Connection | None = None) -> None: if not ids: return - conn = connect() + owns = conn is None + conn = conn or connect() try: - with transaction(conn): + with transaction(conn) if owns else nullcontext(): for bid in ids: conn.execute("DELETE FROM vec_blocks WHERE block_id = ?", (bid,)) finally: - conn.close() + if owns: + conn.close() async def search(self, vector: list[float], *, top_k: int) -> list[VectorHit]: conn = connect() diff --git a/backend/app/services/index_service.py b/backend/app/services/index_service.py index d85eb6a..d8f276f 100644 --- a/backend/app/services/index_service.py +++ b/backend/app/services/index_service.py @@ -6,6 +6,7 @@ MVP 阶段重建是同步的（数据量小），完成后直接返回 completed from __future__ import annotations +import shutil from datetime import datetime, timezone from pathlib import Path from uuid import uuid4 @@ -13,6 +14,7 @@ from uuid import uuid4 from app import repository from app.config import get_settings from app.contracts import IndexJob, IndexRebuildRequest, IndexStatus +from app.errors import ApiError from app.knowledge.parser import parse_note from app.services.note_service import index_note from app.retrieval.vectorstore import SqliteVecStore @@ -22,10 +24,13 @@ vector_store = SqliteVecStore() _jobs: dict[str, IndexJob] = {} -def _scan_vault() -> list[tuple[str, str, str]]: - """扫描 Vault 下所有 Markdown，返回 (rel_path, folder, markdown)。""" +def _scan_vault() -> list[tuple[str, str, str, datetime, datetime]]: + """扫描 Vault 下所有 Markdown，返回 (rel_path, folder, markdown, created, updated)。 + + 先读入内存：若文件读取失败，rebuild 尚未清空旧索引，不会造成数据损失。 + """ vault = get_settings().vault_path - result: list[tuple[str, str, str]] = [] + result: list[tuple[str, str, str, datetime, datetime]] = [] if not vault.exists(): return result for path in sorted(vault.rglob("*.md")): @@ -33,27 +38,52 @@ def _scan_vault() -> list[tuple[str, str, str]]: folder = path.relative_to(vault).parent.as_posix() if folder == ".": folder = "" - result.append((rel, folder, path.read_text(encoding="utf-8"))) + stat = path.stat() + created = datetime.fromtimestamp(stat.st_ctime, tz=timezone.utc) + updated = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc) + result.append((rel, folder, path.read_text(encoding="utf-8"), created, updated)) return result async def rebuild(request: IndexRebuildRequest) -> IndexJob: job_id = "job_" + uuid4().hex[:12] - # MVP：scope（all/notes/vectors）与 note_ids 增量暂不区分，统一全量重建 - repository.clear_all() - await vector_store.clear() - - vault = get_settings().vault_path - for rel, folder, markdown in _scan_vault(): - path = vault / rel - stat = path.stat() - created = datetime.fromtimestamp(stat.st_ctime, tz=timezone.utc) - updated = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc) - parsed = parse_note( - markdown=markdown, file_path=rel, folder=folder, tags=None, - created_at=created, updated_at=updated, + # 增量重建（scope != all 或指定 note_ids）尚未实现，明确拒绝而非静默全量重建 + if request.scope != "all" or request.note_ids: + raise ApiError( + 400, + "UNSUPPORTED_SCOPE", + "only full rebuild (scope='all' with empty note_ids) is supported", + {"scope": request.scope, "note_ids": request.note_ids}, ) - await index_note(parsed) + + # 先扫描到内存（失败不会清旧索引），再快照旧库用于失败回滚 + docs = _scan_vault() + settings = get_settings() + backup_path = settings.db_path.with_suffix(".db.bak") if settings.db_path.exists() else None + if backup_path is not None: + shutil.copy2(settings.db_path, backup_path) + + try: + repository.clear_all() + await vector_store.clear() + for rel, folder, markdown, created, updated in docs: + parsed = parse_note( + markdown=markdown, file_path=rel, folder=folder, tags=None, + created_at=created, updated_at=updated, + ) + await index_note(parsed) + except BaseException: + # 重建失败：恢复旧索引，避免留下半成品；记录 failed 任务后向上抛 + if backup_path is not None and backup_path.exists(): + shutil.copy2(backup_path, settings.db_path) + _jobs[job_id] = IndexJob( + job_id=job_id, status="failed", scope=request.scope, + created_at=datetime.now(timezone.utc), + ) + raise + finally: + if backup_path is not None: + backup_path.unlink(missing_ok=True) job = IndexJob(job_id=job_id, status="completed", scope=request.scope, created_at=datetime.now(timezone.utc)) _jobs[job_id] = job diff --git a/backend/app/services/note_service.py b/backend/app/services/note_service.py index e0feb19..7e6060e 100644 --- a/backend/app/services/note_service.py +++ b/backend/app/services/note_service.py @@ -13,6 +13,7 @@ from pathlib import Path from app import repository from app.config import get_settings from app.contracts import Note, NoteBlock, NoteSummary +from app.database.db import connect, transaction from app.errors import ApiError from app.knowledge.parser import ParsedNote, parse_note from app.retrieval.embedding import HashEmbeddingProvider @@ -89,34 +90,41 @@ def _delete_markdown(rel_path: str) -> None: async def index_note(parsed: ParsedNote) -> None: - """把解析结果写入元数据 + FTS5 + 向量（三层可重建索引）。 + """把解析结果写入元数据 + FTS5 + 向量（三层可重建索引），单事务保证原子性。 - 替换元数据时拿到旧 block_id：清理已删除/内容变化的旧向量，只为新增 block 写向量， - 避免失效向量残留（内容未变的 block 其向量仍有效，无需重复写入）。 + 元数据与向量在同一连接、同一事务内提交，避免「新元数据已提交、向量写入失败」的 + 半提交状态。替换元数据时拿到旧 block_id：清理已删除/内容变化的旧向量，只为新增 + block 写向量（内容未变的 block 其向量仍有效，无需重复写入）。 """ vectors = await embedding.embed_documents([block.content for block in parsed.blocks]) - old_block_ids = repository.replace_note_metadata( - note_id=parsed.note_id, - title=parsed.title, - file_path=parsed.file_path, - folder=parsed.folder, - tags=parsed.tags, - created_at=parsed.created_at, - updated_at=parsed.updated_at, - blocks=parsed.blocks, - ) - old_ids = set(old_block_ids) - new_ids = {block.block_id for block in parsed.blocks} - stale_ids = [bid for bid in old_ids if bid not in new_ids] - if stale_ids: - await vector_store.delete(stale_ids) - missing_ids = [bid for bid in new_ids if bid not in old_ids] - records = [ - VectorRecord(id=block.block_id, vector=vector) - for block, vector in zip(parsed.blocks, vectors) - if block.block_id in missing_ids - ] - await vector_store.upsert(records) + conn = connect() + try: + with transaction(conn): + old_block_ids = repository.replace_note_metadata( + conn=conn, + note_id=parsed.note_id, + title=parsed.title, + file_path=parsed.file_path, + folder=parsed.folder, + tags=parsed.tags, + created_at=parsed.created_at, + updated_at=parsed.updated_at, + blocks=parsed.blocks, + ) + old_ids = set(old_block_ids) + new_ids = {block.block_id for block in parsed.blocks} + stale_ids = [bid for bid in old_ids if bid not in new_ids] + if stale_ids: + await vector_store.delete(stale_ids, conn=conn) + missing_ids = [bid for bid in new_ids if bid not in old_ids] + records = [ + VectorRecord(id=block.block_id, vector=vector) + for block, vector in zip(parsed.blocks, vectors) + if block.block_id in missing_ids + ] + await vector_store.upsert(records, conn=conn) + finally: + conn.close() repository.set_index_meta({"embedding_model": embedding.model_id, "embedding_dim": str(embedding.dim)}) @@ -124,7 +132,8 @@ async def create_note(*, title: str, markdown: str, folder: str | None, tags: li rel_path, clean_folder = _rel_path(folder, title) now = datetime.now(timezone.utc) parsed = parse_note( - markdown=markdown, file_path=rel_path, folder=clean_folder, tags=tags, + # 创建时空标签视为「未显式指定」，由 frontmatter 推导（创建无「清空」语义） + markdown=markdown, file_path=rel_path, folder=clean_folder, tags=tags or None, created_at=now, updated_at=now, ) parsed.title = title # 显式传入的 title 优先于正文推导（与 update_note 保持一致） @@ -156,12 +165,14 @@ async def update_note( old_md = _read_markdown(record.file_path) new_md = old_md if markdown is None else markdown + # PATCH 语义：tags=None 保持原标签；[] 清空；非空列表替换（区别于 create 的 frontmatter 推导） + effective_tags = record.tags if tags is None else tags _write_markdown(record.file_path, new_md) now = datetime.now(timezone.utc) try: parsed = parse_note( - markdown=new_md, file_path=record.file_path, folder=record.folder, tags=tags, + markdown=new_md, file_path=record.file_path, folder=record.folder, tags=effective_tags, created_at=record.created_at, updated_at=now, ) if title is not None: diff --git a/backend/tests/test_retrieval.py b/backend/tests/test_retrieval.py index d6cb7ec..5f137dd 100644 --- a/backend/tests/test_retrieval.py +++ b/backend/tests/test_retrieval.py @@ -321,3 +321,115 @@ def test_search_pagination_total_reflects_all_matches(vault) -> None: engine.search(SearchRequest(query="段", mode=SearchMode.fts, limit=10, offset=55)) ) assert page2.items # 跨过旧候选池边界仍能取到结果 + + +# --------------------------------------------------------------------------- # +# 审阅回归：PATCH tags 语义 / 向量-块一致性 / 过滤漏召回 / rebuild 语义与回滚 +# --------------------------------------------------------------------------- # +def test_patch_tags_semantics(vault) -> None: + """PATCH 省略 tags 保留、tags=[] 清空、非空替换（审阅 #5）。""" + from app.services import note_service + + note = asyncio.run( + note_service.create_note(title="标签语义", markdown="# 标题\n\n正文。", folder="", tags=["a"]) + ) + assert note.tags == ["a"] + + updated = asyncio.run(note_service.update_note(note.note_id, title="改名")) # tags=None + assert updated.tags == ["a"] # 省略 tags 保留原标签 + + updated = asyncio.run(note_service.update_note(note.note_id, tags=["b"])) + assert updated.tags == ["b"] # 非空列表替换 + + updated = asyncio.run(note_service.update_note(note.note_id, tags=[])) + assert updated.tags == [] # 空列表清空 + + +def test_patch_partial_content_no_orphan_vectors(vault) -> None: + """修改正文只删部分 block 后，vec_blocks 与 blocks 的 ID 集合一致（审阅 #2/#3）。""" + from app.database.db import connect + from app.services import note_service + + def ids(table: str) -> set[str]: + conn = connect() + try: + return {row[0] for row in conn.execute(f"SELECT block_id FROM {table}")} + finally: + conn.close() + + note = asyncio.run( + note_service.create_note( + title="部分修改", markdown="# 标题\n\n段落一。\n\n段落二。", folder="", tags=[] + ) + ) + assert ids("vec_blocks") == ids("blocks") + + asyncio.run( + note_service.update_note(note.note_id, markdown="# 标题\n\n段落一改了。\n\n新增段落。") + ) + # 更新后不变量：向量集合与块集合一一对应，无残留、无缺失 + assert ids("vec_blocks") == ids("blocks") + + +def test_fts_metadata_filter_recalls_beyond_candidate_pool(vault) -> None: + """metadata 过滤不能受候选池截断影响：目标块排在 50 名之外也应被召回（审阅 #4）。""" + from app.retrieval.engine import engine + from app.services import index_service + + files: dict[str, str] = {} + # 60 篇短填充笔记：bm25 高，占据 FTS 前 60 位 + for i in range(60): + files[f"批量/填充{i}.md"] = f"---\ntitle: 填充{i}\ntags: 填充\n---\n\n检索\n" + # 目标笔记：长正文使 bm25 变低，排在候选池（50）之外 + long_body = "检索 " + "甲乙丙丁戊己庚辛壬癸子丑寅卯辰巳午未申酉戌亥天地玄黄宇宙洪荒日月盈昃" + files["批量/目标.md"] = f"---\ntitle: 目标\ntags: 目标\n---\n\n{long_body}\n" + + _write_vault(vault, files) + asyncio.run(index_service.rebuild(IndexRebuildRequest(scope="all"))) + + resp = asyncio.run( + engine.search(SearchRequest(query="检索", mode=SearchMode.fts, tags=["目标"])) + ) + assert resp.page.total == 1 + assert resp.items[0].title == "目标" + + +def test_rebuild_rejects_unsupported_scope_and_note_ids(vault) -> None: + """增量 scope / note_ids 未实现时明确拒绝，而非静默全量重建（审阅 #6）。""" + from app.errors import ApiError + from app.services import index_service + + with pytest.raises(ApiError) as exc: + asyncio.run(index_service.rebuild(IndexRebuildRequest(scope="notes"))) + assert exc.value.status_code == 400 + assert exc.value.code == "UNSUPPORTED_SCOPE" + + with pytest.raises(ApiError) as exc: + asyncio.run(index_service.rebuild(IndexRebuildRequest(scope="all", note_ids=["note_x"]))) + assert exc.value.code == "UNSUPPORTED_SCOPE" + + +def test_rebuild_failure_restores_old_index(vault, monkeypatch) -> None: + """重建中途失败应恢复旧索引，不留下半成品（审阅 #6）。""" + from app import repository + from app.services import index_service + from app.services import note_service as ns + + _write_vault(vault, SAMPLE_NOTES) + asyncio.run(index_service.rebuild(IndexRebuildRequest(scope="all"))) + before = repository.stats() + + real_embed = ns.embedding.embed_documents + call = {"n": 0} + + async def _flaky(contents): + call["n"] += 1 + if call["n"] > 1: + raise RuntimeError("embed down") + return await real_embed(contents) + + monkeypatch.setattr(ns.embedding, "embed_documents", _flaky) + with pytest.raises(RuntimeError): + asyncio.run(index_service.rebuild(IndexRebuildRequest(scope="all"))) + + assert repository.stats() == before # 旧索引已恢复，无半成品