adminandClaude Fable 5
60f7360cec
缺陷 #52b(验收 §6 第 2 条 + 回归 G3):同一 PDF 重复上传 4 次 → 4 个 doc_id × 200 chunk = 800 条,FAISS 检索 top_k 被同内容不同 doc_id 的块占满。 修复(vstore/vault.py): - add() 先算内容哈希 SHA256(原始文件字节),查 index.json 的 content_sha 字段判重;同内容 → 移除旧 doc 的检索行/密文/chunk 行,再落一条新 doc (新 doc_id,meta 以本次上传为准,created_at 刷新) - 去重后该内容在库中恒仅 1 个 doc_id、chunk 总数不翻倍;同名不同内容 按内容哈希不误判(G3-2);不同内容互不覆盖(G3-1) - content_sha 落 index.json(doc_meta 新字段);rebuild_index 对历史密文 缺 content_sha 时以密文内容补算(重建后去重依据不丢) - 与现有删除机制一致:index.remove_item 过滤重建 faiss、chunks.remove_document 过滤重建 chunk faiss(IndexFlatIP 不支持逐行删,现有机制即重建) 测试:vstore/tests/test_vault_dedup.py(5 用例):同内容两次 → 1 个 doc_id/ chunk 不翻倍/旧密文清理;meta 以本次为准;不同内容 → 2 个 doc_id; 同名不同内容不误判;重启(新实例 load)后仍去重。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
124 lines
5.2 KiB
Python
124 lines
5.2 KiB
Python
"""test_vault_dedup.py —— 内容 SHA256 去重（缺陷 #52b，验收 §6 第 2 条 + 回归 G3）。
|
||
|
||
运行：cd vstore && /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest tests/test_vault_dedup.py -v
|
||
（或 vstore/pyproject.toml testpaths 下全量 pytest；venv = plantai-rag-new-venv）
|
||
|
||
G3 断言：
|
||
- 同内容 add 两次 → 库中该内容仅 1 个 doc_id、chunk 总数不翻倍；
|
||
- 不同内容 add → 2 个 doc_id 互不覆盖；
|
||
- 同名不同内容 → 按内容哈希不误判（保留 2 个 doc_id）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
|
||
import pytest
|
||
|
||
from vstore.vault import Vault
|
||
|
||
|
||
@pytest.fixture
|
||
def vault(tmp_path, embed_fn, monkeypatch):
|
||
from vstore.settings import VstorePaths, set_paths
|
||
|
||
set_paths(VstorePaths(root=str(tmp_path), vault_path="", index_path="", chunk_path=""))
|
||
# 桩掉 chunk 批量嵌入（vault._chunked_embed 动态查找），不触网真 Ollama
|
||
dim = 1024
|
||
monkeypatch.setattr(
|
||
"vstore.embed.embed_batch_chunked",
|
||
lambda texts, *a, **k: [[1.0 / dim] * dim for _ in texts],
|
||
)
|
||
return Vault(tmp_path / "vault", tmp_path / "index")
|
||
|
||
|
||
def _doc_ids_for_content(vault, content: bytes) -> list[str]:
|
||
"""库中 content_sha == sha256(content) 的 doc 行 id 列表（磁盘事实）。"""
|
||
sha = hashlib.sha256(content).hexdigest()
|
||
import json
|
||
|
||
rows = []
|
||
try:
|
||
data = json.loads(vault.index.json_path.read_text(encoding="utf-8"))
|
||
items = data.get("items", []) if isinstance(data, dict) else []
|
||
except (ValueError, OSError):
|
||
items = []
|
||
for item in items:
|
||
if isinstance(item, dict) and str(item.get("content_sha") or "") == sha:
|
||
rows.append(item)
|
||
return [r.get("id") for r in rows]
|
||
|
||
|
||
def _make_pdf(text: str) -> bytes:
|
||
"""真实最小 PDF 字节（pymupdf 可解析、含目标文本）。"""
|
||
import pymupdf
|
||
|
||
doc = pymupdf.open()
|
||
page = doc.new_page()
|
||
page.insert_text((72, 72), text)
|
||
return doc.tobytes()
|
||
|
||
|
||
def test_same_content_added_twice_keeps_one_doc_id(vault):
|
||
"""#52b：同内容 add 两次 → 1 个 doc_id、chunk 数不翻倍、旧密文被清。"""
|
||
content = _make_pdf("同一份论文内容。")
|
||
first = vault.add(content, "paper.pdf", {"name": "论文", "category": "资料"})
|
||
assert first["content_sha"] == hashlib.sha256(content).hexdigest()
|
||
chunks_after_first = vault.chunks.count()
|
||
assert chunks_after_first >= 1
|
||
|
||
second = vault.add(content, "paper.pdf", {"name": "论文", "category": "资料"})
|
||
|
||
ids = _doc_ids_for_content(vault, content)
|
||
assert ids == [second["document_id"]], f"库中该内容 doc_id 应为仅新 doc：{ids}"
|
||
assert second["document_id"] != first["document_id"]
|
||
# 旧密文被清理（覆盖语义不留两份）
|
||
assert not (vault.vault_dir / f"{first['document_id']}.enc").exists()
|
||
assert (vault.vault_dir / f"{second['document_id']}.enc").exists()
|
||
# chunk 总数不翻倍（旧 doc_id 的 chunk 已清理）
|
||
assert vault.chunks.count() == chunks_after_first
|
||
# chunk 行全部属于新 doc_id（旧 doc_id 不再出现在检索结果）
|
||
assert {r["doc_id"] for r in vault.chunks.rows()} == {second["document_id"]}
|
||
|
||
|
||
def test_duplicate_reuses_latest_meta(vault):
|
||
"""#52b：覆盖语义——重复上传 meta 以本次为准（created_at 刷新、name 覆盖）。"""
|
||
content = "内容甲。".encode("utf-8")
|
||
first = vault.add(content, "a.txt", {"name": "旧名"})
|
||
second = vault.add(content, "a.txt", {"name": "新名"})
|
||
rows = vault.list_rows()
|
||
assert len(rows) == 1
|
||
assert rows[0]["name"] == "新名"
|
||
assert rows[0]["id"] == second["document_id"]
|
||
assert first["document_id"] not in [r["id"] for r in rows]
|
||
|
||
|
||
def test_different_content_keeps_two_doc_ids(vault):
|
||
"""G3-1：不同内容上传 → 2 个 doc_id 互不覆盖。"""
|
||
a = vault.add("内容甲。".encode("utf-8"), "a.txt", {"name": "甲"})
|
||
b = vault.add("内容乙。".encode("utf-8"), "b.txt", {"name": "乙"})
|
||
assert a["document_id"] != b["document_id"]
|
||
assert len(vault.list_rows()) == 2
|
||
assert set(_doc_ids_for_content(vault, "内容甲。".encode("utf-8"))) == {a["document_id"]}
|
||
assert set(_doc_ids_for_content(vault, "内容乙。".encode("utf-8"))) == {b["document_id"]}
|
||
|
||
|
||
def test_same_name_different_content_not_deduped(vault):
|
||
"""G3-2：同名不同内容 → 按 SHA256 不误判（保留 2 个 doc_id）。"""
|
||
a = vault.add(_make_pdf("内容甲。"), "same.pdf", {"name": "同名"})
|
||
b = vault.add(_make_pdf("内容乙。"), "same.pdf", {"name": "同名"})
|
||
assert a["document_id"] != b["document_id"]
|
||
assert len(vault.list_rows()) == 2
|
||
assert vault.chunks.count() >= 2
|
||
|
||
|
||
def test_duplicate_dedup_persists_across_reload(vault, tmp_path, monkeypatch):
|
||
"""#52b：去重依据（content_sha）落盘持久——重启后（新实例 load）仍不判重。"""
|
||
content = "持久化内容。".encode("utf-8")
|
||
vault.add(content, "a.txt", {"name": "甲"})
|
||
vault2 = Vault(tmp_path / "vault", tmp_path / "index")
|
||
vault2.index.load()
|
||
vault2.chunks.load()
|
||
item = vault2.add(content, "a.txt", {"name": "甲"})
|
||
assert _doc_ids_for_content(vault2, content) == [item["document_id"]]
|
||
assert vault2.chunks.count() >= 1
|