kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(knowledge): 优化 laws 分块并加强超长保护

Browse Source 
- 归一化法规 markdown，避免标题/列表标记干扰层级识别

- 按章节条进行层级语义切分，优先保持条级完整

- 超长 chunk 优先句级切分，最终保留 token 上限兜底

- 补充分块回归测试，覆盖 markdown 条款和超长场景
This commit is contained in:
肖泽涛 2026-02-28 16:58:52 +08:00
parent 0695f6f4cd
commit 1f8ea859f8
2 changed files with 251 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
src/knowledge/chunking/ragflow_like/parsers/laws.py
View File

@ -5,6 +5,8 @@ from typing import Any
from src.knowledge.chunking.ragflow_like import nlp
_ARTICLE_PATTERN = re.compile(r"^(第[零一二三四五六七八九十百千万0-9]+条)[\s　:：]*(.*)$")
def _unescape_delimiter(delimiter: str) -> str:
return delimiter.replace("\\n", "\n").replace("\\r", "\r").replace("\\t", "\t").replace("\\\\", "\\")
@ -14,6 +16,39 @@ def _iter_lines(markdown_content: str) -> list[str]:
return [line.strip() for line in (markdown_content or "").splitlines() if line.strip()]
def _normalize_law_line(line: str) -> str:
# 法规 markdown 常见的 #、-、** 装饰会干扰层级识别，这里先做轻量归一化。
text = (line or "").strip()
text = re.sub(r"^#{1,6}\s+", "", text)
text = re.sub(r"^[-*+]\s+", "", text)
text = text.replace("**", "").replace("__", "").replace("`", "")
text = re.sub(r"[ \t]+", " ", text)
return text.strip()
def _expand_article_line(line: str) -> list[str]:
normalized = _normalize_law_line(line)
if not normalized:
return []
matched = _ARTICLE_PATTERN.match(normalized)
if not matched:
return [normalized]
article = matched.group(1).strip()
body = matched.group(2).strip()
if not body:
return [article]
return [article, body]
def _iter_law_sections(markdown_content: str) -> list[str]:
sections: list[str] = []
for line in _iter_lines(markdown_content):
sections.extend(_expand_article_line(line))
return [section for section in sections if section]
def _docx_heading_tree(markdown_content: str) -> list[str]:
lines: list[tuple[int, str]] = []
level_set: set[int] = set()
@ -49,7 +84,93 @@ def _docx_heading_tree(markdown_content: str) -> list[str]:
return [element for element in root.get_tree() if element]
def _hard_split_by_token_limit(text: str, chunk_token_num: int) -> list[str]:
token_iter = list(re.finditer(r"[A-Za-z0-9_]+|[\u4e00-\u9fff]", text or ""))
if not token_iter:
cleaned = (text or "").strip()
return [cleaned] if cleaned else []
chunks: list[str] = []
start = 0
index = 0
max_tokens = max(int(chunk_token_num or 0), 1)
while index < len(token_iter):
end_index = min(index + max_tokens, len(token_iter)) - 1
end = token_iter[end_index].end()
piece = text[start:end].strip()
if piece:
chunks.append(piece)
start = end
index = end_index + 1
tail = text[start:].strip()
if tail:
chunks.append(tail)
return chunks
def _ensure_chunk_token_limit(chunks: list[str], chunk_token_num: int, delimiter: str, overlapped_percent: int) -> list[str]:
"""
对输出 chunk 做 token 上限保护：
1) 先尝试按行用 naive_merge 再切一次；
2) 仍超长时才硬切，避免 embeddings 413。
"""
max_tokens = int(chunk_token_num or 0)
normalized = [chunk.strip() for chunk in chunks if chunk and chunk.strip()]
if max_tokens <= 0:
return normalized
protected: list[str] = []
for chunk in normalized:
if nlp.count_tokens(chunk) <= max_tokens:
protected.append(chunk)
continue
refined = nlp.naive_merge(
[(_line, "") for _line in _iter_lines(chunk)],
chunk_token_num=max_tokens,
delimiter=delimiter,
overlapped_percent=overlapped_percent,
)
if not refined:
refined = [chunk]
for item in refined:
cleaned = (item or "").strip()
if not cleaned:
continue
if nlp.count_tokens(cleaned) <= max_tokens:
protected.append(cleaned)
else:
sentence_refined = nlp.naive_merge(
[(_sentence, "") for _sentence in re.split(r"(?<=[。！？；;!?])", cleaned) if _sentence.strip()],
chunk_token_num=max_tokens,
delimiter=delimiter,
overlapped_percent=overlapped_percent,
)
if not sentence_refined:
sentence_refined = [cleaned]
for sentence_chunk in sentence_refined:
text = sentence_chunk.strip()
if not text:
continue
if nlp.count_tokens(text) <= max_tokens:
protected.append(text)
else:
protected.extend(_hard_split_by_token_limit(text, max_tokens))
return [chunk for chunk in protected if chunk.strip()]
def chunk_markdown(filename: str, markdown_content: str, parser_config: dict[str, Any] | None = None) -> list[str]:
"""
法规分块主流程（简化版）：
- docx 优先尝试标题树；
- 其余格式先做法规文本归一化，再按章/节/条树形切分（depth=3）；
- 最后统一执行超长保护。
"""
parser_config = parser_config or {}
delimiter = _unescape_delimiter(str(parser_config.get("delimiter", "\n") or "\n"))
@ -57,29 +178,30 @@ def chunk_markdown(filename: str, markdown_content: str, parser_config: dict[str
overlapped_percent = int(parser_config.get("overlapped_percent", 0) or 0)
if re.search(r"\.docx$", filename or "", re.IGNORECASE):
chunks = _docx_heading_tree(markdown_content)
if chunks:
return chunks
docx_chunks = _docx_heading_tree(markdown_content)
if len(docx_chunks) > 1:
return _ensure_chunk_token_limit(docx_chunks, chunk_token_num, delimiter, overlapped_percent)
sections = _iter_lines(markdown_content)
sections = _iter_law_sections(markdown_content)
if not sections:
return []
eng = nlp.is_english(sections)
nlp.remove_contents_table(sections, eng=eng)
nlp.remove_contents_table(sections, eng=nlp.is_english(sections))
typed_sections = [(s, "") for s in sections]
nlp.make_colon_as_title(typed_sections)
bull = nlp.bullets_category([s for s, _ in typed_sections])
merged = nlp.tree_merge(bull, typed_sections, depth=2)
if bull == nlp.MARKDOWN_BULLET_GROUP_INDEX:
# 归一化后仍命中 markdown 组时，回退到中文法规层级组，避免退化成“按章大块”。
bull = 0
if merged:
return merged
merged = nlp.tree_merge(bull, typed_sections, depth=3)
if not merged:
merged = nlp.naive_merge(
typed_sections,
chunk_token_num=chunk_token_num,
delimiter=delimiter,
overlapped_percent=overlapped_percent,
)
return nlp.naive_merge(
typed_sections,
chunk_token_num=chunk_token_num,
delimiter=delimiter,
overlapped_percent=overlapped_percent,
)
return _ensure_chunk_token_limit(merged, chunk_token_num, delimiter, overlapped_percent)

114
test/test_ragflow_like_chunking.py
View File

@ -6,7 +6,7 @@ import sys
sys.path.append(os.getcwd())
from src.knowledge.chunking.ragflow_like.dispatcher import chunk_markdown
from src.knowledge.chunking.ragflow_like.nlp import bullets_category
from src.knowledge.chunking.ragflow_like.nlp import bullets_category, count_tokens
from src.knowledge.chunking.ragflow_like.presets import (
CHUNK_ENGINE_VERSION,
get_chunk_preset_options,
@ -109,3 +109,115 @@ def test_chunk_preset_options_include_description() -> None:
options = get_chunk_preset_options()
assert len(options) == 4
assert all(isinstance(option.get("description"), str) and option["description"] for option in options)
def test_laws_chunking_should_apply_overlength_protection() -> None:
lines = ["#### 中华人民共和国企业所得税法实施条例", "##### 微信扫一扫：分享"]
lines.extend(
[
f"第{i}条 企业所得税法实施细则说明，适用于测试场景，确保条文长度足够用于验证分块策略。"
for i in range(1, 260)
]
)
content = "\n".join(lines)
max_chunk_tokens = 180
chunks = chunk_markdown(
markdown_content=content,
file_id="file_laws_long",
filename="laws.docx",
processing_params={
"chunk_preset_id": "laws",
"chunk_parser_config": {
"chunk_token_num": max_chunk_tokens,
"overlapped_percent": 20,
"delimiter": "\\n",
},
},
)
assert len(chunks) > 1
assert max(count_tokens(ck["content"]) for ck in chunks) <= max_chunk_tokens
def test_laws_chunking_should_prefer_sentence_boundary_split() -> None:
line = "第一条 企业所得税法实施细则用于测试分块语义边界。"
content = line * 120
chunks = chunk_markdown(
markdown_content=content,
file_id="file_laws_sentence",
filename="laws.docx",
processing_params={
"chunk_preset_id": "laws",
"chunk_parser_config": {
"chunk_token_num": 120,
"overlapped_percent": 0,
"delimiter": "\\n",
},
},
)
assert len(chunks) > 1
for ck in chunks:
text = ck["content"].strip()
assert text
assert count_tokens(text) <= 120
def test_laws_chunking_should_prefer_article_level_before_item_level() -> None:
content = """
第六章 特别纳税调整
第一百零六条 企业所得税法第三十八条规定的可以指定扣缴义务人的情形，包括：
(一)在资金、经营、购销等方面存在直接或者间接的控制关系;
(二)可以代表企业实施其他具有约束力的行为。
第一百零七条 税务机关可以依法核定应纳税所得额。
""".strip()
chunks = chunk_markdown(
markdown_content=content,
file_id="file_laws_article",
filename="laws.docx",
processing_params={
"chunk_preset_id": "laws",
"chunk_parser_config": {
"chunk_token_num": 1000,
"overlapped_percent": 0,
"delimiter": "\\n",
},
},
)
# 只要条下款项没有被拆成独立碎片，即可满足“条级优先”的目标。
target_chunks = [ck["content"] for ck in chunks if "第一百零六条" in ck["content"]]
assert target_chunks
assert any("(一)" in chunk and "(二)" in chunk for chunk in target_chunks)
def test_laws_markdown_articles_should_not_collapse_into_chapter_chunk() -> None:
content = """
## 第一章 总则
- **第一条** 为了规范担保活动，保障债权实现，制定本法。
- **第二条** 在借贷活动中，当事人可以依法设定担保。
- **第三条** 担保活动应当遵循平等、自愿、公平和诚实信用原则。
""".strip()
chunks = chunk_markdown(
markdown_content=content,
file_id="file_laws_markdown_article",
filename="laws.md",
processing_params={
"chunk_preset_id": "laws",
"chunk_parser_config": {
"chunk_token_num": 120,
"overlapped_percent": 0,
"delimiter": "\\n",
},
},
)
first_article_chunks = [ck["content"] for ck in chunks if "第一条" in ck["content"]]
assert first_article_chunks
# 条级切分时，第一条与第二条不应被合并到同一块。
assert all("第二条" not in chunk for chunk in first_article_chunks)
assert max(count_tokens(ck["content"]) for ck in chunks) <= 120