60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
# ============================================================
|
|
# RAG 文档导入切块黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from server.knowledge.assets import KnowledgeStore
|
|
from server.knowledge.ingest import apply_knowledge_import, chunk_sections, preview_ingest
|
|
from server.knowledge.retrieval import hybrid_search
|
|
|
|
|
|
def test_chunk_markdown_and_ingest(tmp_path):
|
|
md = tmp_path / "换线补充SOP.md"
|
|
md.write_text(
|
|
"# 换线补充\n\n"
|
|
+ ("产线换型前必须完成尾检。" * 20) + "\n\n"
|
|
+ "## 夜班限制\n\n"
|
|
+ ("每日 16:00 后不安排跨族换线。" * 15),
|
|
encoding="utf-8",
|
|
)
|
|
preview = preview_ingest(md.name, path=str(md), kind="sop", title="换线补充SOP")
|
|
assert preview["chunkCount"] >= 1
|
|
assert preview["sectionCount"] >= 1
|
|
assert all("text" in c for c in preview["chunks"])
|
|
|
|
kb = KnowledgeStore(path=str(tmp_path / "kn.json"))
|
|
kb.assets = []
|
|
kb._write()
|
|
applied = apply_knowledge_import(preview, tags=["换线", "导入"], store=kb)
|
|
assert applied["chunkCount"] >= 1
|
|
asset = kb.get(applied["assetId"])
|
|
assert asset and asset.get("chunks")
|
|
units = kb.iter_search_units()
|
|
assert any(u.get("chunkId") for u in units)
|
|
hits = hybrid_search(units, "换线尾检", top_k=3)
|
|
assert hits
|
|
assert hits[0]["title"] == "换线补充SOP"
|
|
assert hits[0]["version"]
|
|
|
|
|
|
def test_chunk_overlap_bounds():
|
|
sections = [{"text": "甲" * 1200, "heading": "长段"}]
|
|
chunks = chunk_sections(sections)
|
|
assert len(chunks) >= 2
|
|
assert chunks[0]["text"]
|
|
assert chunks[1]["text"]
|
|
|
|
|
|
def test_ingest_version_bump(tmp_path):
|
|
f = tmp_path / "a.md"
|
|
f.write_text("# 标题\n\n内容甲乙丙丁戊己庚辛壬癸。\n", encoding="utf-8")
|
|
kb = KnowledgeStore(path=str(tmp_path / "kn.json"))
|
|
kb.assets = []
|
|
kb._write()
|
|
p1 = preview_ingest("a.md", path=str(f), title="同名文档")
|
|
a1 = apply_knowledge_import(p1, store=kb)
|
|
p2 = preview_ingest("a.md", path=str(f), title="同名文档")
|
|
a2 = apply_knowledge_import(p2, store=kb)
|
|
assert a1["version"] == "v1"
|
|
assert a2["version"] == "v2"
|