From 7a508e44f59a735cd6962ef0bcd043ceeb06c6b8 Mon Sep 17 00:00:00 2001 From: HeDaas-Code Date: Thu, 9 Jul 2026 17:17:39 +0000 Subject: [PATCH] =?UTF-8?q?test(intake):=20=E8=A1=A5=E5=85=85=E5=AE=9E?= =?UTF-8?q?=E4=BD=93=E5=BD=92=E5=B9=B6/chunk=E5=8E=BB=E9=87=8D/=E6=A3=80?= =?UTF-8?q?=E7=B4=A2=E9=80=80=E5=8C=96=E6=B5=8B=E8=AF=95=E7=BC=BA=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补齐 commit 972d563(修复 16 个 GitHub issue)中新增但无测试覆盖的逻辑: - chunker.dedup_chunks/_cosine_sim/_sha256_hex(Issue #18.a): 精确匹配(None/HashEmbeddings) / 真嵌入阈值 / embedder 失败降级 三条路径 - entity_resolver(Issue #17,新文件):levenshtein/jaro/jaro_winkler/ strings_similar 纯算法 + resolve_entities 三重信号融合(≥2合并/1待确认/0跳过) - index_store.search HashEmbeddings→SQLite LIKE 退化(Issue #12)+ merge_characters 边界与别名传播 全部确定性、离线(HashEmbeddings + SQLite 临时目录)、无 LLM 依赖。 验证:新增 28 测试 + 原有 50 = 78 passed,smoke 11/11,3 次重跑无不稳定。 --- tests/test_chunker_gaps.py | 137 +++++++++++++++++ tests/test_entity_resolver_gaps.py | 227 +++++++++++++++++++++++++++++ tests/test_index_store_gaps.py | 124 ++++++++++++++++ 3 files changed, 488 insertions(+) create mode 100644 tests/test_chunker_gaps.py create mode 100644 tests/test_entity_resolver_gaps.py create mode 100644 tests/test_index_store_gaps.py diff --git a/tests/test_chunker_gaps.py b/tests/test_chunker_gaps.py new file mode 100644 index 0000000..f17bb82 --- /dev/null +++ b/tests/test_chunker_gaps.py @@ -0,0 +1,137 @@ +"""chunker 去重逻辑的覆盖率缺口测试(Issue #18.a)。 + +`dedup_chunks` / `_cosine_sim` / `_sha256_hex` 在 commit 972d563(修复 16 个 +GitHub issue)中新增,用于在 NER 之前去除重复 chunk,避免同一文本被多次识别 +浪费 LLM 调用。此前无任何单元测试覆盖——本文件补齐: + +- SHA-256 摘要的确定性 +- 余弦相似度的边界条件(空向量 / 长度不匹配 / 零向量 / 正交 / 相同) +- dedup_chunks 三条路径: + * 精确匹配(None / HashEmbeddings)—— 退化为 SHA-256 完全匹配 + * 真嵌入路径 —— cosine ≥ threshold 视为重复 + * embedder 调用失败 —— 安全降级到精确匹配 + +所有用例确定性、离线、无 LLM 依赖。跑法:``python -m pytest tests/test_chunker_gaps.py``。 +""" +from __future__ import annotations + +from persona_distillation.chunker import ( + Chunk, + _cosine_sim, + _sha256_hex, + dedup_chunks, +) +from persona_distillation.intake.embedder import HashEmbeddings + + +# --------------------------------------------------------------------------- +# _sha256_hex +# --------------------------------------------------------------------------- +def test_sha256_hex_deterministic() -> None: + """相同文本产生相同摘要;不同文本产生不同摘要;与编码无关的稳定性。""" + assert _sha256_hex("荒川老师") == _sha256_hex("荒川老师") + assert _sha256_hex("荒川老师") != _sha256_hex("荒川老师 ") + # 空串也是合法输入 + assert isinstance(_sha256_hex(""), str) and len(_sha256_hex("")) == 64 + + +# --------------------------------------------------------------------------- +# _cosine_sim +# --------------------------------------------------------------------------- +def test_cosine_sim_edge_cases() -> None: + """余弦相似度边界:空/长度不匹配返回 0;相同向量返回 1;正交返回 0。""" + # 空向量 + assert _cosine_sim([], [1.0, 2.0]) == 0.0 + assert _cosine_sim([1.0], []) == 0.0 + # 长度不匹配 + assert _cosine_sim([1.0, 2.0], [1.0]) == 0.0 + # 相同向量 → 1.0 + assert _cosine_sim([1.0, 2.0, 3.0], [1.0, 2.0, 3.0]) == 1.0 + # 正交向量 → 0.0 + assert _cosine_sim([1.0, 0.0], [0.0, 1.0]) == 0.0 + # 零向量不应抛错(norm 兜底为 1.0,结果为 0.0) + assert _cosine_sim([0.0, 0.0], [1.0, 1.0]) == 0.0 + + +# --------------------------------------------------------------------------- +# dedup_chunks — 精确匹配路径(None / HashEmbeddings) +# --------------------------------------------------------------------------- +def _chunk(text: str, index: int = 0) -> Chunk: + return Chunk(text=text, index=index, char_start=0, char_end=len(text), token_count=len(text)) + + +def test_dedup_chunks_empty() -> None: + """空列表入参 → 空列表出参。""" + assert dedup_chunks([]) == [] + + +def test_dedup_chunks_exact_match_removes_duplicates() -> None: + """None embedder:SHA-256 精确匹配,去重保留首次出现,保持原顺序。""" + a = _chunk("荒川老师点点头。", 0) + b = _chunk("小明跑过来。", 1) + a2 = _chunk("荒川老师点点头。", 2) # 与 a 文本完全相同 + out = dedup_chunks([a, b, a2], embedder=None) + assert out == [a, b] + # 近似但非完全相同的不应被去重 + c = _chunk("荒川老师点点头!。", 3) + assert dedup_chunks([a, c], embedder=None) == [a, c] + + +def test_dedup_chunks_hash_embeddings_uses_exact_match() -> None: + """HashEmbeddings 是伪嵌入,cosine 无意义,必须走精确匹配路径。""" + a = _chunk("同一段文本。", 0) + b = _chunk("另一段文本。", 1) + a2 = _chunk("同一段文本。", 2) + out = dedup_chunks([a, b, a2], embedder=HashEmbeddings(dim=32)) + assert out == [a, b] + + +# --------------------------------------------------------------------------- +# dedup_chunks — 真嵌入路径 +# --------------------------------------------------------------------------- +class _FakeEmbedder: + """受控伪嵌入器:按文本返回预设向量,模拟真嵌入行为(非 HashEmbeddings)。""" + + def __init__(self, vectors: dict[str, list[float]]) -> None: + self._vectors = vectors + + def embed_documents(self, texts: list[str]) -> list[list[float]]: + return [self._vectors.get(t, [0.0, 0.0]) for t in texts] + + +def test_dedup_chunks_real_embedder_removes_near_duplicates() -> None: + """真嵌入路径:cosine ≥ threshold 视为重复被跳过。""" + # v1 与 v2 几乎平行(cosine≈1),v3 与它们正交 + v1 = [1.0, 0.01] + v2 = [1.0, 0.0] + v3 = [0.0, 1.0] + emb = _FakeEmbedder({"A": v1, "B": v2, "C": v3}) + a, b, c = _chunk("A", 0), _chunk("B", 1), _chunk("C", 2) + out = dedup_chunks([a, b, c], embedder=emb, threshold=0.95) + # B 与 A cosine≈1 ≥ 0.95 → 视为重复跳过;C 正交 → 保留 + assert out == [a, c] + + +def test_dedup_chunks_real_embedder_threshold_boundary() -> None: + """阈值边界:低于阈值的不被视为重复。""" + # cosine(v1,v2) = 1/(1*sqrt(2)) ≈ 0.707 + emb = _FakeEmbedder({"A": [1.0, 0.0], "B": [1.0, 1.0]}) + a, b = _chunk("A", 0), _chunk("B", 1) + # threshold=0.95:0.707 < 0.95 → 不去重 + assert dedup_chunks([a, b], embedder=emb, threshold=0.95) == [a, b] + # threshold=0.5:0.707 ≥ 0.5 → 去重 + assert dedup_chunks([a, b], embedder=emb, threshold=0.5) == [a] + + +class _ExplodingEmbedder: + def embed_documents(self, texts: list[str]) -> list[list[float]]: # noqa: ARG002 + raise RuntimeError("embedder unavailable") + + +def test_dedup_chunks_embedder_failure_falls_back_to_exact() -> None: + """真嵌入调用抛错时,安全降级到精确匹配,不崩溃且仍能去重完全相同的 chunk。""" + a = _chunk("同一段文本。", 0) + b = _chunk("另一段。", 1) + a2 = _chunk("同一段文本。", 2) + out = dedup_chunks([a, b, a2], embedder=_ExplodingEmbedder()) + assert out == [a, b] diff --git a/tests/test_entity_resolver_gaps.py b/tests/test_entity_resolver_gaps.py new file mode 100644 index 0000000..3350c3c --- /dev/null +++ b/tests/test_entity_resolver_gaps.py @@ -0,0 +1,227 @@ +"""跨 chunk 实体归并器的覆盖率缺口测试(Issue #17)。 + +`entity_resolver.py` 在 commit 972d563(修复 16 个 GitHub issue)中作为新文件 +引入,实现「别名 / 字符串相似 / 嵌入相似」三重信号融合的实体归并。此前无任何 +单元测试覆盖——本文件补齐: + +- 纯字符串相似度算法:levenshtein / jaro / jaro_winkler / strings_similar +- 余弦相似度边界、CharacterSignals / ResolveResult 聚合属性 +- resolve_entities 集成行为:≥2 重自动合并 / 1 重待确认 / 0 重跳过 / + 无嵌入降级 / auto_merge=False / 嵌入信号隔离 / target 选择 + +所有用例确定性、离线(HashEmbeddings + SQLite 临时目录)、无 LLM 依赖。 +跑法:``python -m pytest tests/test_entity_resolver_gaps.py``。 +""" +from __future__ import annotations + +import tempfile + +from persona_distillation.intake.embedder import HashEmbeddings +from persona_distillation.intake.entity_resolver import ( + CharacterSignals, + ResolveResult, + _cosine, + jaro, + jaro_winkler, + levenshtein, + resolve_entities, + strings_similar, +) +from persona_distillation.intake.index_store import IndexStore +from persona_distillation.intake.schemas import IndexCategory, NameIndexEntry + + +# --------------------------------------------------------------------------- +# levenshtein +# --------------------------------------------------------------------------- +def test_levenshtein_basic() -> None: + assert levenshtein("abc", "abc") == 0 + assert levenshtein("", "abc") == 3 + assert levenshtein("abc", "") == 3 + assert levenshtein("kitten", "sitting") == 3 # 经典用例 + # 单字符替换 + assert levenshtein("cat", "cot") == 1 + # 插入 + assert levenshtein("cat", "cats") == 1 + # 换位在 Levenshtein 中算 2(两次替换) + assert levenshtein("ab", "ba") == 2 + + +# --------------------------------------------------------------------------- +# jaro / jaro_winkler +# --------------------------------------------------------------------------- +def test_jaro_basic() -> None: + assert jaro("abc", "abc") == 1.0 + assert jaro("", "x") == 0.0 + assert jaro("abc", "xyz") == 0.0 # 无匹配字符 + # 经典用例:MARTHA / MARHTA = 0.9444 + assert abs(jaro("MARTHA", "MARHTA") - 0.9444) < 1e-3 + + +def test_jaro_winkler_basic() -> None: + assert jaro_winkler("abc", "abc") == 1.0 + # 公共前缀加权:MARTHA/MARHTA 前缀 3 → 0.961 + assert abs(jaro_winkler("MARTHA", "MARHTA") - 0.961) < 1e-3 + # 无公共前缀时不额外加权 + assert jaro_winkler("xyz", "abc") == 0.0 + + +# --------------------------------------------------------------------------- +# strings_similar +# --------------------------------------------------------------------------- +def test_strings_similar() -> None: + # 空值守卫 + assert strings_similar("", "x") is False + assert strings_similar("x", "") is False + # 完全相同 + assert strings_similar("荒川", "荒川") is True + # Levenshtein ≤ 2 路径(删 2 字) + assert strings_similar("荒川善次", "荒川") is True + # Jaro-Winkler ≥ 0.85 路径(轻微拼写差异) + assert strings_similar("MARTHA", "MARHTA") is True + # 完全不同 → False + assert strings_similar("荒川善次", "高桥太郎") is False + + +# --------------------------------------------------------------------------- +# _cosine(entity_resolver 内部实现,与 chunker._cosine_sim 独立) +# --------------------------------------------------------------------------- +def test_entity_resolver_cosine_edge_cases() -> None: + assert _cosine([], [1.0]) == 0.0 + assert _cosine([1.0, 2.0], [1.0]) == 0.0 # 长度不匹配 + assert _cosine([1.0, 0.0], [1.0, 0.0]) == 1.0 + assert _cosine([1.0, 0.0], [0.0, 1.0]) == 0.0 + + +# --------------------------------------------------------------------------- +# CharacterSignals / ResolveResult 聚合属性 +# --------------------------------------------------------------------------- +def test_character_signals_hit_count() -> None: + sig = CharacterSignals(name_a="a", name_b="b") + assert sig.hit_count == 0 + sig.alias_hit = True + assert sig.hit_count == 1 + sig.string_hit = True + sig.embedding_hit = True + assert sig.hit_count == 3 + + +def test_resolve_result_total_actions() -> None: + r = ResolveResult() + assert r.total_actions == 0 + r.auto_merged.append(("a", "b")) + assert r.total_actions == 1 + r.pending.append(CharacterSignals(name_a="c", name_b="d", string_hit=True)) + assert r.total_actions == 2 + + +# --------------------------------------------------------------------------- +# resolve_entities 集成(HashEmbeddings → 仅别名+字符串两重信号) +# --------------------------------------------------------------------------- +def _entry( + name: str, *, text: str, aliases: list[str] | None = None, chunk_index: int = 0 +) -> NameIndexEntry: + return NameIndexEntry( + character_name=name, + aliases=aliases or [], + category=IndexCategory.SPEECH, + text=text, + source="a.txt", + chunk_index=chunk_index, + ) + + +def _store() -> IndexStore: + return IndexStore(tempfile.mkdtemp(), embedding=HashEmbeddings(dim=32)) + + +def test_resolve_entities_single_character_noop() -> None: + """<2 个人物 → 空结果,不报错。""" + store = _store() + store.add(_entry("荒川", text="嘛。")) + result = resolve_entities(store, llm=None) + assert result.auto_merged == [] + assert result.pending == [] + store.close() + + +def test_resolve_entities_alias_plus_string_auto_merge() -> None: + """别名交叉 + 字符串相似(2 重,无嵌入)→ 自动合并到 mention 更多的 target。""" + store = _store() + # 荒川善次 带 alias「荒川」;荒川 是独立条目 → 经典跨 chunk 归并场景 + store.add(_entry("荒川善次", text="嘛,再看看吧。", aliases=["荒川"], chunk_index=0)) + store.add(_entry("荒川善次", text="书不还价。", chunk_index=1)) + store.add(_entry("荒川", text="五十二岁。", chunk_index=2)) + result = resolve_entities(store, llm=None, auto_merge=True) + # source=荒川 → target=荒川善次(mention 2 > 1) + assert ("荒川", "荒川善次") in result.auto_merged + # 合并后荒川 应消失 + names = {c["character_name"] for c in store.list_characters()} + assert "荒川" not in names + assert "荒川善次" in names + store.close() + + +def test_resolve_entities_single_hit_goes_pending() -> None: + """仅别名交叉 1 重(字符串不相似)→ 待确认,不自动合并。""" + store = _store() + store.add(_entry("荒川善次", text="嘛。", aliases=["Sensei"])) + store.add(_entry("Sensei", text="hello.")) + result = resolve_entities(store, llm=None, auto_merge=True) + assert result.auto_merged == [] + assert len(result.pending) == 1 + assert result.pending[0].alias_hit is True + assert result.pending[0].string_hit is False + # 未合并:两个名字都仍在 + names = {c["character_name"] for c in store.list_characters()} + assert names == {"荒川善次", "Sensei"} + store.close() + + +def test_resolve_entities_zero_hits_skipped() -> None: + """无别名交叉、字符串不相似(0 重)→ 不出现在结果里。""" + store = _store() + store.add(_entry("荒川善次", text="嘛。")) + store.add(_entry("高桥太郎", text="你好。")) + result = resolve_entities(store, llm=None) + assert result.auto_merged == [] + assert result.pending == [] + store.close() + + +def test_resolve_entities_auto_merge_false_goes_pending() -> None: + """auto_merge=False 时,即使 2 重命中也进 pending 而非合并。""" + store = _store() + store.add(_entry("荒川善次", text="嘛。", aliases=["荒川"])) + store.add(_entry("荒川", text="书。")) + result = resolve_entities(store, llm=None, auto_merge=False) + assert result.auto_merged == [] + assert len(result.pending) == 1 + assert result.pending[0].hit_count >= 2 + store.close() + + +# --------------------------------------------------------------------------- +# resolve_entities — 嵌入信号(受控伪嵌入,非 HashEmbeddings) +# --------------------------------------------------------------------------- +class _ConstantEmbedder: + """对所有文本返回同一向量 → 任意两人物 mean embedding cosine = 1.0。""" + + def embed_documents(self, texts: list[str]) -> list[list[float]]: # noqa: ARG002 + return [[1.0, 0.0] for _ in texts] + + +def test_resolve_entities_embedding_signal_isolated() -> None: + """无别名、字符串不相似,但嵌入一致 → embedding_hit=True,1 重 → 待确认。""" + store = IndexStore(tempfile.mkdtemp(), embedding=_ConstantEmbedder()) + store.add(_entry("荒川善次", text="嘛。")) + store.add(_entry("高桥太郎", text="你好。")) + result = resolve_entities(store, llm=None, auto_merge=True, embedding_cos_min=0.8) + assert result.auto_merged == [] # 仅 1 重不合并 + assert len(result.pending) == 1 + sig = result.pending[0] + assert sig.embedding_hit is True + assert sig.embedding_score >= 0.8 + assert sig.alias_hit is False + assert sig.string_hit is False + store.close() diff --git a/tests/test_index_store_gaps.py b/tests/test_index_store_gaps.py new file mode 100644 index 0000000..6938d06 --- /dev/null +++ b/tests/test_index_store_gaps.py @@ -0,0 +1,124 @@ +"""IndexStore 检索与归并的覆盖率缺口测试(Issue #12 / #17)。 + +commit 972d563(修复 16 个 GitHub issue)中两处关键改动此前无针对性单测: + +- **#12**:``search`` 在 ``HashEmbeddings``(伪嵌入)/ Chroma 不可用时,必须 + 退化到 SQLite ``LIKE`` 关键词匹配,并支持 ``character_name`` 过滤与 ``k`` 上限。 + 若退化路径失效,离线/单测场景会返回空结果或随机向量结果,掩盖真实检索问题。 +- **#17**:``merge_characters`` 是跨 chunk 实体归并的底层原子操作,负责把源人物 + 全部条目并入目标并传播别名。边界(source==target / 不存在)与别名合并正确性 + 直接影响实体归并的数据完整性。 + +所有用例确定性、离线(HashEmbeddings + SQLite 临时目录)、无 LLM 依赖。 +跑法:``python -m pytest tests/test_index_store_gaps.py``。 +""" +from __future__ import annotations + +import tempfile + +from persona_distillation.intake.embedder import HashEmbeddings +from persona_distillation.intake.index_store import IndexStore +from persona_distillation.intake.schemas import IndexCategory, NameIndexEntry + + +def _store() -> IndexStore: + return IndexStore(tempfile.mkdtemp(), embedding=HashEmbeddings(dim=32)) + + +def _entry( + name: str, *, text: str, category: IndexCategory = IndexCategory.SPEECH, + aliases: list[str] | None = None, chunk_index: int = 0, +) -> NameIndexEntry: + return NameIndexEntry( + character_name=name, + aliases=aliases or [], + category=category, + text=text, + source="a.txt", + chunk_index=chunk_index, + ) + + +# --------------------------------------------------------------------------- +# search — HashEmbeddings → SQLite LIKE 退化(Issue #12) +# --------------------------------------------------------------------------- +def test_search_like_fallback_finds_by_keyword() -> None: + """HashEmbeddings 下 search 走 LIKE:按文本关键词命中。""" + store = _store() + store.add(_entry("荒川", text="嘛,再看看吧。这版是岩波文库。")) + store.add(_entry("小明", text="来旧书店买书。")) + results = store.search("岩波") + assert len(results) == 1 + assert "岩波" in results[0].text + store.close() + + +def test_search_like_fallback_respects_character_filter() -> None: + """character_name 过滤:仅返回该人物的命中。""" + store = _store() + store.add(_entry("荒川", text="书不还价。")) + store.add(_entry("小明", text="书很便宜。")) # 同含「书」字 + results = store.search("书", character_name="荒川") + assert len(results) == 1 + assert results[0].character_name == "荒川" + store.close() + + +def test_search_like_fallback_respects_k_limit() -> None: + """k 上限:命中数超过 k 时只返回前 k 条。""" + store = _store() + for i in range(5): + store.add(_entry(f"人物{i}", text=f"第{i}条记录。", chunk_index=i)) + results = store.search("记录", k=2) + assert len(results) == 2 + store.close() + + +def test_search_like_no_match_returns_empty() -> None: + """关键词不存在 → 空列表(不抛错)。""" + store = _store() + store.add(_entry("荒川", text="嘛。")) + assert store.search("不存在的关键词XYZ") == [] + store.close() + + +# --------------------------------------------------------------------------- +# merge_characters(Issue #17 底层原子操作) +# --------------------------------------------------------------------------- +def test_merge_characters_source_equals_target_returns_zero() -> None: + store = _store() + store.add(_entry("荒川", text="嘛。")) + assert store.merge_characters("荒川", "荒川") == 0 + store.close() + + +def test_merge_characters_nonexistent_source_returns_zero() -> None: + store = _store() + store.add(_entry("荒川", text="嘛。")) + assert store.merge_characters("不存在", "荒川") == 0 + # target 不变 + assert store.count() == 1 + store.close() + + +def test_merge_characters_moves_entries_and_propagates_alias() -> None: + """合并:源条目迁入 target,返回迁移数,源名作为别名写入 target。""" + store = _store() + store.add(_entry("荒川", text="嘛。", aliases=["老师"])) + store.add(_entry("荒川", text="书。", chunk_index=1)) + store.add(_entry("荒川善次", text="五十二岁。", category=IndexCategory.APPEARANCE)) + + moved = store.merge_characters("荒川", "荒川善次") + assert moved == 2 # 荒川 的 2 条迁入 + # 源人物消失,target 累计 3 条 + names = {c["character_name"] for c in store.list_characters()} + assert names == {"荒川善次"} + assert store.count() == 3 + # 「荒川」作为别名传播到 target + target_aliases = set() + for e in store.get_character_entries("荒川善次"): + target_aliases |= set(e.aliases) + assert "荒川" in target_aliases + # 原别名「老师」也保留 + assert "老师" in target_aliases + store.close()