Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion astrbot/core/knowledge_base/chunking/markdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,11 +129,14 @@ async def _sections_to_chunks(
# 扣除前缀长度,确保添加前缀后不超过 chunk_size
prefix_len = self._estimate_prefix_length(heading_path)
effective_chunk_size = max(chunk_size // 4, chunk_size - prefix_len)
effective_overlap = self._scale_overlap(
chunk_overlap, chunk_size, effective_chunk_size
)

sub_chunks = await self._fallback_chunker.chunk(
section_text,
chunk_size=effective_chunk_size,
chunk_overlap=chunk_overlap,
chunk_overlap=effective_overlap,
)
for i, sub_chunk in enumerate(sub_chunks):
chunk_text = self._apply_heading_context(
Expand All @@ -143,6 +146,22 @@ async def _sections_to_chunks(

return raw_chunks

@staticmethod
def _scale_overlap(
chunk_overlap: int, chunk_size: int, effective_chunk_size: int
) -> int:
"""按正文预算等比缩放重叠长度。

标题前缀会压缩子块可用的 chunk_size,而外部传入的 chunk_overlap 是针对
完整 chunk_size 配置的。原样传递会让合法配置在内部变成
overlap >= effective_chunk_size,被递归分块器拒绝。保持
overlap / chunk_size 的比例,并确保结果严格小于 effective_chunk_size。
"""
if chunk_size <= 0 or effective_chunk_size >= chunk_size:
return chunk_overlap
scaled = chunk_overlap * effective_chunk_size // chunk_size
return max(0, min(scaled, effective_chunk_size - 1))

def _build_context_prefix(self, heading_path: list[str]) -> str:
"""构建标题路径前缀"""
if self.include_heading_context and heading_path:
Expand Down
64 changes: 64 additions & 0 deletions tests/test_markdown_chunker.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
import pytest

from astrbot.core.knowledge_base.chunking.markdown import MarkdownChunker

# 600 characters of body made of unique markers (w000 ... w119), so a lost
# body segment is detected even though overlapping chunks duplicate text.
BODY_MARKERS = [f"w{i:03d}" for i in range(120)]
BODY = " ".join(BODY_MARKERS) + " "
LONG_HEADING_DOC = "# " + "A" * 200 + "\n\n## Child\n" + BODY


def _assert_body_preserved(chunks: list[str]) -> None:
joined = "\n".join(chunks)
missing = [marker for marker in BODY_MARKERS if marker not in joined]
assert not missing, f"body markers lost during chunking: {missing}"


@pytest.mark.asyncio
async def test_long_heading_prefix_keeps_valid_overlap_configuration():
"""A valid (chunk_size, chunk_overlap) pair must stay valid for the
recursive fallback even when the heading prefix shrinks the body budget."""
chunker = MarkdownChunker(chunk_size=256, chunk_overlap=100)

chunks = await chunker.chunk(LONG_HEADING_DOC)

assert chunks
_assert_body_preserved(chunks)


@pytest.mark.asyncio
async def test_overlap_override_is_scaled_too():
chunker = MarkdownChunker(chunk_size=1024, chunk_overlap=50)

chunks = await chunker.chunk(LONG_HEADING_DOC, chunk_size=256, chunk_overlap=100)

assert chunks
_assert_body_preserved(chunks)


@pytest.mark.asyncio
async def test_without_heading_context_behaviour_is_unchanged():
chunker = MarkdownChunker(
chunk_size=256, chunk_overlap=100, include_heading_context=False
)

chunks = await chunker.chunk(LONG_HEADING_DOC)

assert chunks
_assert_body_preserved(chunks)
assert all(len(chunk) <= 256 for chunk in chunks)


@pytest.mark.parametrize(
("overlap", "size", "effective", "expected"),
[
(100, 256, 64, 25), # proportional to the reduced budget
(100, 256, 256, 100), # no reduction, untouched
(100, 256, 300, 100), # larger budget, untouched
(255, 256, 64, 63), # never reaches the effective size
(0, 256, 64, 0),
],
)
def test_scale_overlap(overlap, size, effective, expected):
assert MarkdownChunker._scale_overlap(overlap, size, effective) == expected