#!/usr/bin/env python3
"""
LLM-free linting of a stored BM25/vector corpus (post-store check).
File: klea_utils/stores/postcheck.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import logging
from collections import Counter
from typing import Any
from langchain_core.documents import Document
logger = logging.getLogger(__name__)
#: A chunk whose ``page_content`` has fewer than this many characters is
#: flagged as suspicious -- typically a failed conversion or an OCR miss
#: on a scanned PDF.
MIN_CHUNK_CHARS = 20
#: Metadata keys that, when absent from a chunk, suggest the file's
#: DEFAULT entry did not apply or extraction failed. ``file_name`` is
#: machine-set so its presence is expected; the others come from the map.
REQUIRED_METADATA_KEYS = ("file_name", "title", "year", "doi")
#: Maximum characters of ``page_content`` shown per sample chunk.
SAMPLE_TEXT_WIDTH = 200
#: How many short chunks / missing-metadata files to list before truncating
#: the report (the rest are counted in the summary line).
MAX_LISTED = 20
#: Contiguous chunks shown at each sampling location. The user reviews a
#: small run of consecutive chunks per location rather than one isolated
#: chunk, to check that chunking stays coherent across boundaries.
PER_ANCHOR_SAMPLES = 3
[docs]
def lint_store(docs: list[Document]) -> dict[str, Any]:
"""Return a structured report of issues in a stored chunk corpus.
Runs only deterministic checks (no LLM): a corpus summary, suspicious
chunks (near-empty text, missing bibliographic metadata), and
structural problems (chunks without a ``file_name``, invalid types).
*docs* is the unpickled BM25 corpus (a list of
:class:`langchain_core.documents.Document`).
:param docs: List of stored chunk documents
:returns: A dict with ``total``, ``files``, ``chunks_per_file``,
``total_chars``, ``empty`` (list of ``(file_name, chars)``),
``missing_metadata`` (``{file_name: [missing keys]}``),
``no_file_name`` (count of chunks lacking ``file_name``),
``invalid_content`` (count of non-string ``page_content``), and
``invalid_year`` (count of non-int ``year``)
"""
total = len(docs)
total_chars = sum(
len(doc.page_content) if isinstance(doc.page_content, str) else 0
for doc in docs
)
files: Counter[str] = Counter()
for doc in docs:
file_name = doc.metadata.get("file_name")
if isinstance(file_name, str) and file_name:
files[file_name] += 1
chunks_per_file: dict[str, int | float] = {}
if files:
counts = list(files.values())
chunks_per_file = {
"min": min(counts),
"max": max(counts),
"avg": round(sum(counts) / len(counts), 2),
}
# Suspicious: near-empty content (conversion/OCR miss).
empty: list[tuple[str, int]] = []
for doc in docs:
text = doc.page_content
if not isinstance(text, str) or len(text.strip()) < MIN_CHUNK_CHARS:
file_name = doc.metadata.get("file_name", "?")
empty.append((file_name, len(text) if isinstance(text, str) else 0))
# Suspicious: missing bibliographic metadata.
missing_metadata: dict[str, list[str]] = {}
for doc in docs:
file_name = doc.metadata.get("file_name", "?")
missing = [k for k in REQUIRED_METADATA_KEYS if k not in doc.metadata]
if missing:
missing_metadata.setdefault(file_name, []).extend(missing)
# Structural: chunks that cannot be attributed to a source file.
no_file_name = sum(
1 for doc in docs if not isinstance(doc.metadata.get("file_name"), str)
)
# Structural: invalid page_content / year types.
invalid_content = sum(1 for doc in docs if not isinstance(doc.page_content, str))
invalid_year = sum(
1
for doc in docs
if "year" in doc.metadata and not isinstance(doc.metadata["year"], int)
)
return {
"total": total,
"files": len(files),
"chunks_per_file": chunks_per_file,
"total_chars": total_chars,
"empty": empty,
"missing_metadata": missing_metadata,
"no_file_name": no_file_name,
"invalid_content": invalid_content,
"invalid_year": invalid_year,
}
[docs]
def select_sample_windows(
docs: list[Document],
anchors: int,
per_anchor: int = PER_ANCHOR_SAMPLES,
) -> list[list[Document]]:
"""Pick ``anchors`` evenly-spaced windows of contiguous chunks to review.
Each window is ``per_anchor`` consecutive chunks starting at an anchor
position, so the user sees a small coherent run at several different
places in the corpus (early/middle/late) rather than the first few
chunks. Anchor starts are evenly spaced across the whole corpus and
avoid the very first/last chunk, giving roughly 25% / 50% / 75% for
three anchors.
For a corpus smaller than one window, a single window of the whole
corpus is returned. The selection is deterministic -- no RNG -- so
re-runs show the same samples for easy comparison.
:param docs: List of stored chunk documents
:param anchors: Number of sampling locations (``--samples``)
:param per_anchor: Contiguous chunks to show at each location
:returns: List of windows (each a list of :class:`Document`); empty
when ``anchors <= 0`` or the corpus is empty
"""
total = len(docs)
if anchors <= 0 or total == 0:
return []
if total <= per_anchor:
return [docs]
starts = [round(total * i / (anchors + 1)) for i in range(1, anchors + 1)]
return [docs[start : start + per_anchor] for start in starts]