From ee5062ca9c018e6d16464b723600de7c931e4501 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 23 Mar 2026 22:41:55 +0800 Subject: [PATCH] =?UTF-8?q?test(14-04):=20=E6=B7=BB=E5=8A=A0=E9=95=BF?= =?UTF-8?q?=E6=96=87=E6=9C=AC=E5=88=86=E5=9D=97=E5=A4=84=E7=90=86=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加 TestSplitIntoChunks 测试类：测试文本分块功能 - 添加 TestLongTextChunking 测试类：测试长文本自动分块分段 - 添加 TestChunkIndexCorrection 测试类：测试分块索引修正 - 添加 TestChunkFallback 测试类：测试单个分块失败时降级到规则分段 测试覆盖： - 分块边界条件（空文本、短文本、长文本） - 句子边界分割（。！？\n） - 索引修正（start_index/end_index 相对于原文） - 分块失败降级 Co-Authored-By: Claude Opus 4.6 --- tests/test_semantic_robustness.py | 333 ++++++++++++++++++++++++++++++ 1 file changed, 333 insertions(+) diff --git a/tests/test_semantic_robustness.py b/tests/test_semantic_robustness.py index b417e17..687bd30 100644 --- a/tests/test_semantic_robustness.py +++ b/tests/test_semantic_robustness.py @@ -657,3 +657,336 @@ class TestExtractJSON: response = '这不包含任何 JSON 内容。' with pytest.raises(ValueError, match="未找到 JSON 内容"): extract_json_from_llm_response(response) + + +# ============================================================================ +# Test: 长文本分块处理 (Plan 14-04) +# ============================================================================ + +class TestSplitIntoChunks: + """测试文本分块功能（Plan 14-04 - Task 14-04-01）""" + + def test_split_into_chunks_basic(self): + """测试基本的文本分块功能""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + + strategy = UnifiedSegmentationStrategy() + + # 创建一个包含多个句子的文本（约 6000 字符） + text = "这是第一句。" * 1000 + "这是第二句。" * 1000 + "这是第三句。" * 1000 + + chunks = strategy._split_into_chunks(text, max_chunk_size=5000) + + # 验证分块数量 + assert len(chunks) > 1 + + # 验证每个分块不超过最大字符数 + for chunk_text, offset in chunks: + assert len(chunk_text) <= 5000 + + # 验证偏移量是递增的 + offsets = [offset for _, offset in chunks] + assert offsets == sorted(offsets) + + # 验证所有分块拼接后等于原文 + reconstructed = ''.join(chunk_text for chunk_text, _ in chunks) + assert reconstructed == text + + def test_split_into_chunks_short_text(self): + """测试短文本不分块""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + + strategy = UnifiedSegmentationStrategy() + + # 短文本（< 5000 字符） + text = "这是短文本。" * 100 + + chunks = strategy._split_into_chunks(text, max_chunk_size=5000) + + # 验证只有一个分块 + assert len(chunks) == 1 + + # 验证内容不变 + chunk_text, offset = chunks[0] + assert chunk_text == text + assert offset == 0 + + def test_split_into_chunks_sentence_boundaries(self): + """测试分块在句子边界分割""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + + strategy = UnifiedSegmentationStrategy() + + # 创建包含不同标点符号的文本 + text = "第一句。第二句！第三句？第四句\n" * 1000 + + chunks = strategy._split_into_chunks(text, max_chunk_size=5000) + + # 验证每个分块在句子边界结束（以标点符号或换行符结尾） + for chunk_text, offset in chunks: + # 分块应该以句子结束字符结尾 + assert chunk_text[-1] in '。！？\n' + + def test_split_into_chunks_empty_text(self): + """测试空文本分块""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + + strategy = UnifiedSegmentationStrategy() + + chunks = strategy._split_into_chunks("", max_chunk_size=5000) + + # 空文本应该返回空列表（没有句子可以分块） + assert len(chunks) == 0 + + +class TestLongTextChunking: + """测试长文本自动分块分段（Plan 14-04）""" + + def test_long_text_triggers_chunking(self, caplog): + """测试长文本（>5000字符）自动触发分块处理""" + import logging + import json + + # 创建模拟 AI 客户端 + mock_client = Mock() + mock_client.api_key = "test-key" + + # 创建长文本（>5000 字符） + long_text = "这是一个长句。" * 1000 + + # 计算会分成几个分块 + expected_chunks = (len(long_text) + 4999) // 5000 + + # 为每个分块设置响应 + responses = [] + for i in range(expected_chunks): + chunk_len = min(5000, len(long_text) - i * 5000) + response = json.dumps({ + "video_type": "知识类", + "segments": [ + { + "title": "测试标题", + "content": "分块内容" * (chunk_len // 4), + "start_index": 0, + "end_index": chunk_len + } + ], + "actual_segment_count": 1 + }, ensure_ascii=False) + responses.append(response) + + mock_client.generate.side_effect = responses + + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + from kb_workflow.srt.parser import Segment + + strategy = UnifiedSegmentationStrategy(ai_client=mock_client) + + # 创建长文本字幕 + segments = [Segment(1, 0.0, 1.0, long_text)] + + with caplog.at_level(logging.INFO): + result = strategy.segment(segments, {"title": "测试"}) + + # 验证调用了 LLM 多次（每个分块一次） + assert mock_client.generate.call_count == expected_chunks + + # 验证日志包含分块信息 + assert any("splitting into chunks" in record.message.lower() + for record in caplog.records) + + # 验证返回了分段结果 + assert len(result) > 0 + + def test_short_text_no_chunking(self, caplog): + """测试短文本（<=5000字符）不分块直接处理""" + import logging + import json + + mock_client = Mock() + mock_client.api_key = "test-key" + + # 短文本（< 5000 字符） + short_text = "这是短文本。" * 100 + + response = json.dumps({ + "video_type": "知识类", + "segments": [ + { + "title": "测试标题", + "content": short_text, + "start_index": 0, + "end_index": len(short_text) + } + ], + "actual_segment_count": 1 + }, ensure_ascii=False) + + mock_client.generate.return_value = response + + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + from kb_workflow.srt.parser import Segment + + strategy = UnifiedSegmentationStrategy(ai_client=mock_client) + + segments = [Segment(1, 0.0, 1.0, short_text)] + + with caplog.at_level(logging.INFO): + result = strategy.segment(segments, {"title": "测试"}) + + # 验证只调用了一次 LLM（不分块） + assert mock_client.generate.call_count == 1 + + # 验证日志不包含分块信息 + assert not any("splitting into chunks" in record.message.lower() + for record in caplog.records) + + # 验证返回了分段结果 + assert len(result) == 1 + + +class TestChunkIndexCorrection: + """测试分块索引修正（Plan 14-04）""" + + def test_chunk_index_correction(self): + """测试分块索引相对于原文位置""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + + strategy = UnifiedSegmentationStrategy() + + # 模拟分块结果 + chunk_text = "这是第一块。" + offset = 100 + + # 创建模拟的分段结果（索引相对于分块） + from kb_workflow.srt.semantic_engine import VideoSegment + + chunk_segments = [ + VideoSegment( + title="测试", + content=chunk_text, + start_index=0, + end_index=len(chunk_text) + ) + ] + + # 合并分块结果 + chunk_results = [(chunk_segments, offset)] + merged = strategy._merge_segment_results(chunk_results) + + # 验证索引被修正为相对于原文 + assert len(merged) == 1 + assert merged[0].start_index == offset + assert merged[0].end_index == offset + len(chunk_text) + + def test_multiple_chunks_index_correction(self): + """测试多个分块的索引修正""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy, VideoSegment + + strategy = UnifiedSegmentationStrategy() + + # 创建多个分块 + chunk1_text = "第一块内容。" * 100 + chunk2_text = "第二块内容。" * 100 + chunk3_text = "第三块内容。" * 100 + + # 模拟分块结果 + chunk1_segments = [VideoSegment(title="块1", content=chunk1_text, start_index=0, end_index=len(chunk1_text))] + chunk2_segments = [VideoSegment(title="块2", content=chunk2_text, start_index=0, end_index=len(chunk2_text))] + chunk3_segments = [VideoSegment(title="块3", content=chunk3_text, start_index=0, end_index=len(chunk3_text))] + + chunk_results = [ + (chunk1_segments, 0), # 第一个分块偏移量为 0 + (chunk2_segments, len(chunk1_text)), # 第二个分块偏移量为第一块的长度 + (chunk3_segments, len(chunk1_text) + len(chunk2_text)) # 第三个分块偏移量为前两块的长度之和 + ] + + # 合并分块结果 + merged = strategy._merge_segment_results(chunk_results) + + # 验证所有分块的索引都被正确修正 + assert len(merged) == 3 + + assert merged[0].start_index == 0 + assert merged[0].end_index == len(chunk1_text) + + assert merged[1].start_index == len(chunk1_text) + assert merged[1].end_index == len(chunk1_text) + len(chunk2_text) + + assert merged[2].start_index == len(chunk1_text) + len(chunk2_text) + assert merged[2].end_index == len(chunk1_text) + len(chunk2_text) + len(chunk3_text) + + +class TestChunkFallback: + """测试分块失败时的降级处理（Plan 14-04）""" + + def test_chunk_fallback_to_rule_based(self, caplog): + """测试单个分块失败时降级到规则分段""" + import logging + import json + + mock_client = Mock() + mock_client.api_key = "test-key" + + # 创建长文本（会分成多个分块） + long_text = "这是第一句。" * 1000 + "这是第二句。" * 1000 + + # 第一个分块成功，第二个分块失败 + success_response = json.dumps({ + "video_type": "知识类", + "segments": [ + { + "title": "成功分块", + "content": "内容", + "start_index": 0, + "end_index": 10 + } + ], + "actual_segment_count": 1 + }, ensure_ascii=False) + + # 第一个分块返回成功，第二个分块抛出异常 + mock_client.generate.side_effect = [success_response, Exception("LLM error")] + + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + from kb_workflow.srt.parser import Segment + + strategy = UnifiedSegmentationStrategy(ai_client=mock_client) + + segments = [Segment(1, 0.0, 1.0, long_text)] + + with caplog.at_level(logging.WARNING): + result = strategy.segment(segments, {"title": "测试"}) + + # 验证返回了分段结果（第一个分块的 LLM 结果 + 第二个分块的规则分段） + assert len(result) > 0 + + # 验证日志包含分块失败和降级信息 + assert any("failed" in record.message.lower() and "chunk" in record.message.lower() + for record in caplog.records) + + def test_fallback_chunk_returns_valid_segments(self): + """测试 _fallback_chunk 返回有效的 VideoSegment""" + from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy + + strategy = UnifiedSegmentationStrategy() + + chunk_text = "这是降级分块的内容。" * 100 + offset = 500 + + # 调用降级方法 + segments = strategy._fallback_chunk(chunk_text, offset) + + # 验证返回了分段列表 + assert len(segments) > 0 + + # 验证所有分段都是 VideoSegment 类型 + from kb_workflow.srt.semantic_engine import VideoSegment + assert all(isinstance(seg, VideoSegment) for seg in segments) + + # 验证第一个分段的索引正确 + assert segments[0].start_index == offset + + # 验证降级分段没有标题 + assert all(seg.title is None for seg in segments) +