yunmei
1fb1d278d5
- core/rag-kb：领域 RAG 知识库按模板配置（PRD 5.4/7.3，EPIC #6 子任务） - templating.py：知识源三类分类 + 模板命名推导 + 零依赖轻量 YAML 配置加载 - documents.py：文档段落抽取分块，chunk 携带 文档/章节/段落 溯源信息 - store.py：from_template_config 按模板自动构建 + 中英混合词频检索 + 类别过滤 - config/kb.template.yaml：ti-cl4 模板 RAG 库资产示例（工艺规范/SOP/国标） - tests：29 用例全绿（模板化/分块溯源/检索排序/类别过滤/配置校验）
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""文档抽取（documents）单元测试：分块 / 章节识别 / 引用溯源。"""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import _bootstrap # noqa: F401
|
|
|
|
from rag_kb.documents import ( # noqa: E402
|
|
KbDocument,
|
|
build_document,
|
|
chunk_document,
|
|
)
|
|
|
|
SOP_TEXT = """# 沸腾氯化炉异常处置SOP
|
|
|
|
## 2.1 炉温骤升处置
|
|
当炉温骤升超过阈值时，立即降低氯气流量，并开启冷却循环水。
|
|
同时通知中控室确认炉压状态。
|
|
|
|
## 2.2 炉压异常
|
|
炉压异常升高时，检查排渣系统是否堵塞，必要时切换备用炉。
|
|
"""
|
|
|
|
|
|
class ChunkDocumentTest(unittest.TestCase):
|
|
def test_heading_and_sections(self):
|
|
doc = KbDocument(
|
|
doc_id="sop-001",
|
|
title="沸腾氯化炉异常处置SOP",
|
|
text=SOP_TEXT,
|
|
category="sop",
|
|
)
|
|
chunks = chunk_document(doc)
|
|
# 两个小节各至少一个段落
|
|
self.assertGreaterEqual(len(chunks), 2)
|
|
sections = {c.section for c in chunks}
|
|
self.assertIn("2.1 炉温骤升处置", sections)
|
|
self.assertIn("2.2 炉压异常", sections)
|
|
|
|
def test_source_traceability(self):
|
|
doc = KbDocument(
|
|
doc_id="sop-001",
|
|
title="沸腾氯化炉异常处置SOP",
|
|
text=SOP_TEXT,
|
|
category="sop",
|
|
)
|
|
chunks = chunk_document(doc)
|
|
first = chunks[0]
|
|
# 溯源串 = 标题 §章节 ¶序号（PRD 5.4 引用溯源）
|
|
self.assertTrue(first.source.startswith("沸腾氯化炉异常处置SOP §2.1 炉温骤升处置 ¶1"))
|
|
self.assertEqual(first.category, "sop")
|
|
# 段落序号全局递增
|
|
seqs = [c.seq for c in chunks]
|
|
self.assertEqual(seqs, sorted(seqs))
|
|
|
|
def test_long_paragraph_split_at_sentence_boundary(self):
|
|
text = "第一句。第二句！第三句？第四句；第五句。" * 10 # 200 字
|
|
doc = KbDocument(doc_id="d", title="长文档", text=text)
|
|
chunks = chunk_document(doc, max_chars=120)
|
|
# 每段不超过阈值（硬切兜底除外）
|
|
for c in chunks:
|
|
self.assertLessEqual(len(c.text), 120 + 20) # 句边界允许略超
|
|
# 拼接后应还原全文（顺序无损）
|
|
joined = "".join(c.text for c in chunks)
|
|
self.assertIn("第一句。第二句！", joined)
|
|
|
|
def test_pure_heading_skipped(self):
|
|
doc = KbDocument(doc_id="d", title="只有标题", text="# 第一章\n\n# 第二章\n")
|
|
chunks = chunk_document(doc)
|
|
self.assertEqual(chunks, []) # 纯标题行不产生段落
|
|
|
|
def test_empty_text_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
KbDocument(doc_id="d", title="空文档", text=" \n ")
|
|
|
|
|
|
class BuildDocumentTest(unittest.TestCase):
|
|
def test_doc_id_slug(self):
|
|
doc = build_document("沸腾氯化工艺规范", "正文内容")
|
|
self.assertEqual(doc.doc_id, "沸腾氯化工艺规范")
|
|
self.assertEqual(doc.category, "process")
|
|
|
|
def test_custom_doc_id(self):
|
|
doc = build_document("沸腾氯化工艺规范", "正文内容", doc_id="spec-001")
|
|
self.assertEqual(doc.doc_id, "spec-001")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|