feat(vector): page-aware PDF chunking for predictable per-page retrieval
Add PageAwareChunker, which splits paginated documents (PDFs) on page boundaries first and only character-splits pages larger than chunk_size. No chunk spans a page boundary, so page_number is always exact and stored excerpts never lead with a neighbouring page's text. When chunk_size is at least the largest page, this yields exactly one chunk per page: a predictable vector count (== page count), a flat per-page embedding cost, and zero cross-page overlap duplication. Gated by DOCUMENT_CHUNK_PAGE_AWARE (default true). When false, the legacy char-based DocumentChunker + post-hoc assign_page_numbers path runs unchanged. Only doc_type="file" with page_boundaries (PDFs) takes the page-aware path; notes/deck/news are unaffected. Measured on a 15-page record (query "leadership award louis", target = top-half of page 15): char-based degraded the target to dense-rank 10 at cs=2048 (OCR) and mislabeled its page; page-aware restored rank 1 across every fusion/modality and chunk size, with correct page labels and clean snippets. BREAKING CHANGE: PDFs are re-chunked page-aware by default. Existing deployments will re-index PDF content on the next vector sync (different chunk counts and page_number labels). Set DOCUMENT_CHUNK_PAGE_AWARE=false to retain the previous char-based behaviour. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
40e56aeaea
commit
2f2a7f9659
@@ -2,6 +2,7 @@
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import anyio
|
||||
from langchain_text_splitters import RecursiveCharacterTextSplitter
|
||||
@@ -97,3 +98,150 @@ class DocumentChunker:
|
||||
self.overlap,
|
||||
)
|
||||
return chunks
|
||||
|
||||
|
||||
class PageAwareChunker:
|
||||
"""Page-first chunker for paginated documents (PDFs).
|
||||
|
||||
Unlike :class:`DocumentChunker`, which splits the concatenated document text
|
||||
on character boundaries and is therefore page-agnostic, this chunker splits
|
||||
on the page boundaries FIRST and only falls back to character splitting for
|
||||
pages larger than ``chunk_size``. As a result:
|
||||
|
||||
* No chunk ever spans a page boundary, so ``page_number`` is always exact
|
||||
and the stored excerpt never leads with a neighbouring page's text (the
|
||||
char-based path can bury a short page's content in the tail of a chunk
|
||||
whose majority — and thus :func:`assign_page_numbers` label — is the
|
||||
previous page).
|
||||
* Chunks-per-page is ``ceil(page_chars / chunk_size)``. When ``chunk_size``
|
||||
is at least the largest page, that is exactly one chunk per page, giving a
|
||||
predictable vector count (== page count), a flat per-page embedding cost,
|
||||
and zero cross-page overlap duplication.
|
||||
|
||||
Page numbers are assigned inline, so callers must NOT additionally run
|
||||
``assign_page_numbers`` on the result.
|
||||
"""
|
||||
|
||||
def __init__(self, chunk_size: int = 2048, overlap: int = 200):
|
||||
"""
|
||||
Initialize page-aware chunker.
|
||||
|
||||
Args:
|
||||
chunk_size: Number of characters per chunk (default: 2048). Pages at
|
||||
or below this size become a single chunk; larger pages are
|
||||
character-split (with overlap) within the page only.
|
||||
overlap: Overlapping characters between sub-chunks of an oversized
|
||||
page (default: 200). Pages that fit in one chunk carry no
|
||||
overlap.
|
||||
"""
|
||||
self.chunk_size = chunk_size
|
||||
self.overlap = overlap
|
||||
|
||||
# Only used for pages that exceed chunk_size. Same hierarchical splitter
|
||||
# as DocumentChunker so oversized pages keep semantic-boundary splitting.
|
||||
self.splitter = RecursiveCharacterTextSplitter(
|
||||
chunk_size=chunk_size,
|
||||
chunk_overlap=overlap,
|
||||
add_start_index=True,
|
||||
strip_whitespace=True,
|
||||
)
|
||||
|
||||
async def chunk_text(
|
||||
self, content: str, page_boundaries: list[dict[str, Any]]
|
||||
) -> list[ChunkWithPosition]:
|
||||
"""
|
||||
Split ``content`` into per-page chunks using ``page_boundaries``.
|
||||
|
||||
Args:
|
||||
content: Full document text. Offsets in ``page_boundaries`` must
|
||||
index into this string (the extractor contract — see
|
||||
``document_processors``).
|
||||
page_boundaries: Ordered list of ``{"page", "start_offset",
|
||||
"end_offset"}`` dicts. When empty, falls back to plain
|
||||
character chunking (no page numbers), matching
|
||||
:class:`DocumentChunker` for non-paginated input.
|
||||
|
||||
Returns:
|
||||
List of chunks with character positions and ``page_number`` set.
|
||||
"""
|
||||
if not content:
|
||||
return [ChunkWithPosition(text="", start_offset=0, end_offset=0)]
|
||||
|
||||
# No page info (e.g. a non-PDF that reached this path) — degrade to the
|
||||
# char-based behaviour so the caller still gets sensible chunks.
|
||||
if not page_boundaries:
|
||||
docs = await anyio.to_thread.run_sync( # type: ignore[attr-defined]
|
||||
self.splitter.create_documents,
|
||||
[content],
|
||||
)
|
||||
return [
|
||||
ChunkWithPosition(
|
||||
text=doc.page_content,
|
||||
start_offset=doc.metadata.get("start_index", 0),
|
||||
end_offset=doc.metadata.get("start_index", 0)
|
||||
+ len(doc.page_content),
|
||||
)
|
||||
for doc in docs
|
||||
]
|
||||
|
||||
chunks = await anyio.to_thread.run_sync( # type: ignore[attr-defined]
|
||||
self._chunk_by_page,
|
||||
content,
|
||||
page_boundaries,
|
||||
)
|
||||
|
||||
logger.debug(
|
||||
"Page-aware chunked document into %s chunks across %s pages "
|
||||
"(chunk_size=%s, overlap=%s)",
|
||||
len(chunks),
|
||||
len(page_boundaries),
|
||||
self.chunk_size,
|
||||
self.overlap,
|
||||
)
|
||||
return chunks
|
||||
|
||||
def _chunk_by_page(
|
||||
self, content: str, page_boundaries: list[dict[str, Any]]
|
||||
) -> list[ChunkWithPosition]:
|
||||
"""CPU-bound per-page splitting (runs in a worker thread)."""
|
||||
chunks: list[ChunkWithPosition] = []
|
||||
for boundary in page_boundaries:
|
||||
page = boundary["page"]
|
||||
start = boundary["start_offset"]
|
||||
end = boundary["end_offset"]
|
||||
page_text = content[start:end]
|
||||
|
||||
# Skip blank pages: embedding an empty/whitespace-only string wastes
|
||||
# a provider call and a vector slot.
|
||||
if not page_text.strip():
|
||||
continue
|
||||
|
||||
if len(page_text) <= self.chunk_size:
|
||||
stripped = page_text.strip()
|
||||
# Tighten offsets to the stripped text so they stay meaningful
|
||||
# even though the whole page is one chunk.
|
||||
lead = len(page_text) - len(page_text.lstrip())
|
||||
chunk_start = start + lead
|
||||
chunks.append(
|
||||
ChunkWithPosition(
|
||||
text=stripped,
|
||||
start_offset=chunk_start,
|
||||
end_offset=chunk_start + len(stripped),
|
||||
page_number=page,
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
# Oversized page: split within the page only, keeping offsets
|
||||
# absolute and the page number fixed.
|
||||
for doc in self.splitter.create_documents([page_text]):
|
||||
sub_start = start + doc.metadata.get("start_index", 0)
|
||||
chunks.append(
|
||||
ChunkWithPosition(
|
||||
text=doc.page_content,
|
||||
start_offset=sub_start,
|
||||
end_offset=sub_start + len(doc.page_content),
|
||||
page_number=page,
|
||||
)
|
||||
)
|
||||
return chunks
|
||||
|
||||
@@ -30,7 +30,10 @@ from nextcloud_mcp_server.observability.metrics import (
|
||||
from nextcloud_mcp_server.observability.tracing import trace_operation
|
||||
from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter
|
||||
from nextcloud_mcp_server.vector import payload_keys
|
||||
from nextcloud_mcp_server.vector.document_chunker import DocumentChunker
|
||||
from nextcloud_mcp_server.vector.document_chunker import (
|
||||
DocumentChunker,
|
||||
PageAwareChunker,
|
||||
)
|
||||
from nextcloud_mcp_server.vector.html_processor import html_to_markdown
|
||||
from nextcloud_mcp_server.vector.placeholder import (
|
||||
delete_placeholder_point,
|
||||
@@ -682,27 +685,46 @@ async def _index_document(
|
||||
logger.error("Failed to process file %s: %s", file_path, e)
|
||||
raise
|
||||
|
||||
# Tokenize and chunk (using configured chunk size and overlap)
|
||||
# Tokenize and chunk (using configured chunk size and overlap). Paginated
|
||||
# files (PDFs with page_boundaries) use the page-aware chunker when enabled,
|
||||
# which assigns page numbers inline; everything else uses the char-based
|
||||
# chunker followed by post-hoc page assignment.
|
||||
page_boundaries = file_metadata.get("page_boundaries")
|
||||
use_page_aware = (
|
||||
settings.document_chunk_page_aware
|
||||
and doc_task.doc_type == "file"
|
||||
and page_boundaries is not None
|
||||
)
|
||||
with trace_operation(
|
||||
"vector_sync.chunk_text",
|
||||
attributes={
|
||||
"vector_sync.input_chars": len(content),
|
||||
"vector_sync.chunk_size": settings.document_chunk_size,
|
||||
"vector_sync.overlap": settings.document_chunk_overlap,
|
||||
"vector_sync.page_aware": use_page_aware,
|
||||
},
|
||||
) as chunk_span:
|
||||
chunker = DocumentChunker(
|
||||
chunk_size=settings.document_chunk_size,
|
||||
overlap=settings.document_chunk_overlap,
|
||||
)
|
||||
chunks = await chunker.chunk_text(content)
|
||||
if use_page_aware:
|
||||
page_boundaries_list = cast(list[dict[str, Any]], page_boundaries)
|
||||
chunks = await PageAwareChunker(
|
||||
chunk_size=settings.document_chunk_size,
|
||||
overlap=settings.document_chunk_overlap,
|
||||
).chunk_text(content, page_boundaries_list)
|
||||
else:
|
||||
chunks = await DocumentChunker(
|
||||
chunk_size=settings.document_chunk_size,
|
||||
overlap=settings.document_chunk_overlap,
|
||||
).chunk_text(content)
|
||||
record_document_chunks(doc_task.doc_type, len(chunks))
|
||||
if chunk_span is not None:
|
||||
chunk_span.set_attribute(_ATTR_CHUNK_COUNT, len(chunks))
|
||||
|
||||
# Assign page numbers to chunks if page boundaries are available (PDFs)
|
||||
page_boundaries = file_metadata.get("page_boundaries")
|
||||
if doc_task.doc_type == "file" and page_boundaries is not None:
|
||||
# Assign page numbers for the char-based path (page-aware already sets them).
|
||||
if (
|
||||
not use_page_aware
|
||||
and doc_task.doc_type == "file"
|
||||
and page_boundaries is not None
|
||||
):
|
||||
# Type narrowing: page_boundaries is guaranteed to be list[dict] here
|
||||
page_boundaries_list = cast(list[dict[str, Any]], page_boundaries)
|
||||
with trace_operation(
|
||||
|
||||
Reference in New Issue
Block a user