danvics/haiku.rag
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add near-duplicate document detection to doctor

Browse source
This commit is contained in:
Yiorgis Gozadinos 2026-06-26 12:23:49 +03:00
parent ce27a7aae5
commit 43f2130b66
No known key found for this signature in database
6 changed files with 394 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
CHANGELOG.md
View file

@ -1,6 +1,10 @@
# Changelog
## [Unreleased]
### Added
- `doctor` reports groups of near-duplicate documents (revisions sharing most of their chunks), flagging the largest member as the likely one to keep. Tuned via the `doctor.duplicates` config block (`containment_threshold`, `candidate_threshold`, `twin_similarity`, `min_chunks`).
### Security
- Bumped dependencies in `uv.lock` to patched versions for known advisories: `aiohttp` 3.14.1, `cryptography` 49.0.0, `idna` 3.18, `langchain-core` 1.4.8, `langchain-text-splitters` 1.1.2, `langsmith` 0.9.1, `lxml` 6.1.1, `pillow` 12.2.0, `pydantic-settings` 2.14.2, `pyjwt` 2.13.0, `pytest` 9.1.1, `python-multipart` 0.0.32, `requests` 2.34.2, `starlette` 1.3.1, `urllib3` 2.7.0, `vcrpy` 8.2.1.

1
docs/cli.md
View file

@ -307,6 +307,7 @@ Checks include:
- the configured embedding identity matches the stored settings
- no database migrations are pending
- the vector index covers all chunks
- near-duplicate documents (revisions sharing most of their chunks) are grouped and reported, with the largest member flagged as the likely one to keep (advisory only, never deleted; tuned via `doctor.duplicates` in config)
- API keys are set for configured providers
It also probes the external endpoints the config uses and reports them under a Providers section:

7
docs/configuration/index.md
View file

@ -114,6 +114,13 @@ search:
vector_index_metric: cosine # cosine, l2, or dot
vector_refine_factor: 30
doctor:
duplicates: # Near-duplicate document detection (doctor command)
containment_threshold: 0.75 # flag a group when one doc shares >= this fraction of the smaller's chunks
candidate_threshold: 0.85 # centroid similarity gate for proposing candidate pairs (recall)
twin_similarity: 0.97 # cosine at which two chunks count as the same chunk
min_chunks: 3 # documents with fewer chunks are excluded
prompts:
domain_preamble: "" # Prepended to skill instructions

22
haiku_rag_slim/haiku/rag/config/models.py
View file

@ -103,6 +103,27 @@ class AnalysisConfig(BaseModel):
max_executions: int = 15
class DuplicateDetectionConfig(BaseModel):
"""Thresholds for doctor's near-duplicate document detection.
Detection clusters documents that share most of their chunks (revisions of
one another). ``containment_threshold`` is the decision knob; the others
tune the cheap centroid pre-filter, what counts as a shared chunk, and which
tiny documents to skip.
"""
containment_threshold: float = 0.75
candidate_threshold: float = 0.85
twin_similarity: float = 0.97
min_chunks: int = 3
class DoctorConfig(BaseModel):
duplicates: DuplicateDetectionConfig = Field(
default_factory=DuplicateDetectionConfig
)
class PictureDescriptionConfig(BaseModel):
"""How the VLM runs over each picture when it runs at all.
@ -516,6 +537,7 @@ class AppConfig(BaseModel):
analysis: AnalysisConfig = Field(default_factory=AnalysisConfig)
processing: ProcessingConfig = Field(default_factory=ProcessingConfig)
search: SearchConfig = Field(default_factory=SearchConfig)
doctor: DoctorConfig = Field(default_factory=DoctorConfig)
providers: ProvidersConfig = Field(default_factory=ProvidersConfig)
prompts: PromptsConfig = Field(default_factory=PromptsConfig)
ingester: IngesterConfig = Field(default_factory=IngesterConfig)

171
haiku_rag_slim/haiku/rag/doctor.py
View file

@ -8,6 +8,7 @@ import numpy as np
from pydantic import BaseModel, Field
from haiku.rag.config import AppConfig
from haiku.rag.config.models import DuplicateDetectionConfig
from haiku.rag.store.engine import (
REQUIRED_TABLES,
Store,
@ -232,6 +233,153 @@ async def _column_values(table, column: str) -> list:
return [row[column] for row in rows]
# Backstop on how many centroid candidate pairs we verify, bounding memory and
# runtime on a pathologically self-similar corpus.
MAX_CANDIDATE_PAIRS = 200_000
class _DuplicateFamily(BaseModel):
members: list[str]
superset: str
pairs: list[tuple[str, str, float, float]]
def _unit(vector: np.ndarray) -> np.ndarray:
norm = np.linalg.norm(vector)
return vector / norm if norm else vector
def _containment(source: np.ndarray, target: np.ndarray, twin: float) -> float:
"""Fraction of ``source`` chunks with a near-identical chunk in ``target``."""
return float(((source @ target.T).max(axis=1) >= twin).mean())
def _duplicate_families(
doc_vectors: dict[str, np.ndarray], cfg: DuplicateDetectionConfig
) -> list[_DuplicateFamily]:
"""Cluster documents that share most of their chunks (revisions of one another).
Block-then-verify: cheap centroid similarity proposes candidate document
pairs, then directed chunk-overlap containment confirms them. Returns one
entry per connected component of confirmed pairs.
"""
# Normalize, drop unembedded (zero) vectors, apply the small-document floor.
normalized: dict[str, np.ndarray] = {}
for doc_id, matrix in doc_vectors.items():
m = np.asarray(matrix, dtype=float)
if m.ndim != 2 or m.shape[0] == 0:
continue
norms = np.linalg.norm(m, axis=1)
m = m[norms > 0]
if m.shape[0] < cfg.min_chunks:
continue
normalized[doc_id] = m / np.linalg.norm(m, axis=1)[:, None]
if len(normalized) < 2:
return []
order = sorted(normalized)
centroids = np.array([_unit(normalized[d].mean(axis=0)) for d in order])
# Stage 1: centroid candidate pairs, block-wise to avoid a full D×D matrix.
candidates: list[tuple[int, int]] = []
block = 512
for start in range(0, len(order), block):
sims = centroids[start : start + block] @ centroids.T
for row in range(sims.shape[0]):
gi = start + row
above = np.nonzero(sims[row, gi + 1 :] >= cfg.candidate_threshold)[0]
candidates.extend((gi, gi + 1 + int(j)) for j in above)
if len(candidates) >= MAX_CANDIDATE_PAIRS:
candidates = candidates[:MAX_CANDIDATE_PAIRS]
break
# Stage 2: confirm candidates with directed chunk-overlap containment.
adjacency: dict[int, set[int]] = {}
edges: dict[tuple[int, int], tuple[float, float]] = {}
for i, j in candidates:
a_to_b = _containment(
normalized[order[i]], normalized[order[j]], cfg.twin_similarity
)
b_to_a = _containment(
normalized[order[j]], normalized[order[i]], cfg.twin_similarity
)
if max(a_to_b, b_to_a) >= cfg.containment_threshold:
adjacency.setdefault(i, set()).add(j)
adjacency.setdefault(j, set()).add(i)
edges[(i, j)] = (a_to_b, b_to_a)
if not edges:
return []
# Cluster confirmed pairs into families (connected components).
families: list[_DuplicateFamily] = []
seen: set[int] = set()
for node in adjacency:
if node in seen:
continue
component: set[int] = set()
stack = [node]
while stack:
cur = stack.pop()
if cur in seen:
continue
seen.add(cur)
component.add(cur)
stack.extend(adjacency[cur] - seen)
members = sorted(order[i] for i in component)
# Largest document (most chunks) is the likely superset; smallest id on a tie.
superset = min(members, key=lambda d: (-normalized[d].shape[0], d))
pairs = sorted(
(order[i], order[j], round(ab, 3), round(ba, 3))
for (i, j), (ab, ba) in edges.items()
if i in component and j in component
)
families.append(
_DuplicateFamily(members=members, superset=superset, pairs=pairs)
)
return sorted(families, key=lambda f: f.members)
def _check_duplicate_documents(
doc_vectors: dict[str, np.ndarray],
uri_by_doc: dict[str, str | None],
title_by_doc: dict[str, str | None],
cfg: DuplicateDetectionConfig,
) -> CheckResult:
families = _duplicate_families(doc_vectors, cfg)
if not families:
return CheckResult(
name="duplicate_documents",
severity=Severity.OK,
message="No near-duplicate documents detected.",
)
def label(doc_id: str) -> str:
return uri_by_doc.get(doc_id) or title_by_doc.get(doc_id) or doc_id
lines: list[str] = []
for family in families:
members = ", ".join(label(m) for m in family.members)
overlaps = "; ".join(
f"{label(a)}→{label(b)}: {ab:.2f}, {label(b)}→{label(a)}: {ba:.2f}"
for a, b, ab, ba in family.pairs
)
lines.append(f"[{members}] keep≈{label(family.superset)} ({overlaps})")
total_docs = sum(len(f.members) for f in families)
return CheckResult(
name="duplicate_documents",
severity=Severity.WARN,
message=(
f"{len(families)} group(s) of documents with substantial chunk overlap "
f"(potential duplicates/revisions), {total_docs} documents."
),
remediation=(
"Review each group and remove redundant revisions; overlap may be intentional."
),
details=_sample(lines),
)
async def run_db_checks(
store: Store, config: AppConfig, stats: dict
) -> list[CheckResult]:
@ -244,7 +392,7 @@ async def run_db_checks(
doc_ids = set(await _column_values(store.documents_table, "id"))
meta_rows = (
await store.document_meta_table.query()
.select(["document_id", "metadata"])
.select(["document_id", "metadata", "uri", "title"])
.to_list()
)
meta_doc_ids = {row["document_id"] for row in meta_rows}
@ -254,6 +402,8 @@ async def run_db_checks(
)
for row in meta_rows
}
uri_by_doc = {row["document_id"]: row.get("uri") for row in meta_rows}
title_by_doc = {row["document_id"]: row.get("title") for row in meta_rows}
chunk_rows = (
await store.chunks_table.query()
@ -375,7 +525,11 @@ async def run_db_checks(
# Vector dimension consistency and unembedded (all-zero) vectors share one
# scan of the vector column — the heaviest check on large corpora.
arrow = await store.chunks_table.query().select(["id", "vector"]).to_arrow()
arrow = (
await store.chunks_table.query()
.select(["id", "vector", "document_id"])
.to_arrow()
)
stored = await SettingsRepository(store).get_current_settings()
stored_dim = stored.get("embeddings", {}).get("model", {}).get("vector_dim")
actual_dim = arrow.schema.field("vector").type.list_size
@ -419,6 +573,19 @@ async def run_db_checks(
)
)
# Near-duplicate documents (revisions sharing most chunks), grouped from the
# same vector scan rather than a second pass.
chunk_doc_ids_ordered = arrow.column("document_id").to_pylist()
indices_by_doc: dict[str, list[int]] = {}
for index, doc_id in enumerate(chunk_doc_ids_ordered):
indices_by_doc.setdefault(doc_id, []).append(index)
doc_vectors = {doc_id: vectors[idx] for doc_id, idx in indices_by_doc.items()}
results.append(
_check_duplicate_documents(
doc_vectors, uri_by_doc, title_by_doc, config.doctor.duplicates
)
)
# Pictures from image/PDF sources should carry raster bytes. Pictures that
# are external image references in a text document (markdown, HTML) have no
# embedded bytes by nature, so a missing raster there is expected.

191
tests/test_doctor.py
View file

@ -3,6 +3,7 @@ from importlib import metadata
from unittest.mock import AsyncMock, MagicMock
import lancedb
import numpy as np
import pytest
from typer.testing import CliRunner
@ -11,6 +12,8 @@ from haiku.rag.config.models import (
AppConfig,
ConversionOptions,
DoclingServeConfig,
DoctorConfig,
DuplicateDetectionConfig,
EmbeddingModelConfig,
EmbeddingsConfig,
ModelConfig,
@ -26,6 +29,7 @@ from haiku.rag.doctor import (
_check_api_keys,
_check_embedding_drift,
_check_vector_index,
_duplicate_families,
_model_present,
_probe_endpoint,
_provider_targets,
@ -933,3 +937,190 @@ async def test_probe_endpoint_connection_error():
reachable, error, _ = await _probe_with_handler(handler)
assert not reachable
assert error is not None and "refused" in error
# --- Duplicate-document detection ----------------------------------------
def _docs(spec: dict[str, list[int]], dim: int = 8) -> dict[str, np.ndarray]:
"""Build per-document chunk matrices from one-hot indices.
A shared index across documents is a shared (identical) chunk; distinct
indices are orthogonal, so they never count as twins.
"""
eye = np.eye(dim)
return {
doc: np.array([eye[i] for i in idxs], dtype=float) for doc, idxs in spec.items()
}
# Stage-2 (containment/clustering) unit tests disable the centroid gate
# (candidate_threshold=0.0) so every pair is verified; orthogonal one-hot chunks
# would otherwise drop centroids below the default gate. The gate itself is
# exercised by the end-to-end tests below.
def _stage2_cfg(**kw) -> DuplicateDetectionConfig:
return DuplicateDetectionConfig(candidate_threshold=0.0, **kw)
def test_duplicate_families_revision_pair():
families = _duplicate_families(
_docs({"a": [0, 1, 2, 3], "b": [0, 1, 2, 4]}), _stage2_cfg()
)
assert len(families) == 1
assert set(families[0].members) == {"a", "b"}
def test_duplicate_families_append_only_is_asymmetric():
families = _duplicate_families(
_docs({"a": [0, 1, 2], "b": [0, 1, 2, 3, 4, 5]}), _stage2_cfg()
)
assert len(families) == 1
fam = families[0]
assert fam.superset == "b" # the larger document
# directed containment: all of A is in B (1.0); only half of B is in A.
a_to_b = next(p for p in fam.pairs if p[:2] == ("a", "b"))
assert a_to_b[2] == pytest.approx(1.0)
assert a_to_b[3] == pytest.approx(0.5)
def test_duplicate_families_distinct_docs_none():
families = _duplicate_families(
_docs({"a": [0, 1, 2], "b": [3, 4, 5]}), _stage2_cfg()
)
assert families == []
def test_duplicate_families_three_way_chain_one_family():
families = _duplicate_families(
_docs(
{
"a": [0, 1, 2, 3],
"b": [0, 1, 2, 3, 4],
"c": [0, 1, 2, 3, 4, 5],
}
),
_stage2_cfg(),
)
assert len(families) == 1
assert set(families[0].members) == {"a", "b", "c"}
assert families[0].superset == "c"
def test_duplicate_families_tiny_docs_ignored():
# min_chunks = 3 excludes the one-chunk documents.
families = _duplicate_families(_docs({"a": [0], "b": [0]}), _stage2_cfg())
assert families == []
def test_duplicate_families_threshold_is_configurable():
# Share 3 of 5 chunks each -> containment 0.6 both ways.
spec = {"a": [0, 1, 2, 3, 4], "b": [0, 1, 2, 5, 6]}
assert _duplicate_families(_docs(spec), _stage2_cfg()) == []
flagged = _duplicate_families(_docs(spec), _stage2_cfg(containment_threshold=0.6))
assert len(flagged) == 1
assert set(flagged[0].members) == {"a", "b"}
async def _build_dup_db(path, docs: dict[str, list[int]], *, vector_dim: int = 8):
"""Build a multi-document database with one-hot chunk vectors."""
eye = np.eye(vector_dim)
db = await lancedb.connect_async(path)
settings_tbl = await db.create_table("settings", schema=SettingsRecord)
docs_tbl = await db.create_table("documents", schema=DocumentRecord)
meta_tbl = await db.create_table("document_meta", schema=DocumentMetaRecord)
chunk_model = create_chunk_model(vector_dim)
chunks_tbl = await db.create_table("chunks", schema=chunk_model)
items_tbl = await db.create_table("document_items", schema=DocumentItemRecord)
await settings_tbl.add(
[
SettingsRecord(
id="settings",
settings=json.dumps(
{
"version": CURRENT_VERSION,
"embeddings": {
"model": {
"provider": "ollama",
"name": "test",
"vector_dim": vector_dim,
}
},
}
),
)
]
)
for doc_id, idxs in docs.items():
await docs_tbl.add([DocumentRecord(id=doc_id, content="x")])
await meta_tbl.add(
[DocumentMetaRecord(document_id=doc_id, uri=f"test://{doc_id}")]
)
await items_tbl.add(
[
DocumentItemRecord(
document_id=doc_id, position=0, self_ref="#/texts/0", text="x"
)
]
)
await chunks_tbl.add(
[
chunk_model(
id=f"{doc_id}-c{n}",
document_id=doc_id,
content="x",
metadata=json.dumps({"doc_item_refs": ["#/texts/0"]}),
vector=eye[i].tolist(),
)
for n, i in enumerate(idxs)
]
)
return db
@pytest.mark.asyncio
async def test_duplicate_documents_check_warns_end_to_end(temp_db_path):
await _build_dup_db(temp_db_path, {"a": [0, 1, 2, 3], "b": [0, 1, 2, 3, 4]})
report = await run_doctor(_config(vector_dim=8), temp_db_path, {})
result = _result(report, "duplicate_documents")
assert result.severity is Severity.WARN
blob = " ".join(result.details)
assert "test://a" in blob and "test://b" in blob
@pytest.mark.asyncio
async def test_duplicate_documents_check_ok_when_distinct(temp_db_path):
await _build_dup_db(temp_db_path, {"a": [0, 1, 2], "b": [3, 4, 5]})
report = await run_doctor(_config(vector_dim=8), temp_db_path, {})
assert _result(report, "duplicate_documents").severity is Severity.OK
@pytest.mark.asyncio
async def test_duplicate_documents_check_reads_config(temp_db_path):
# Share 3 of 5 -> containment 0.6. Disable the centroid gate on both runs so
# only containment_threshold decides the outcome.
await _build_dup_db(temp_db_path, {"a": [0, 1, 2, 3, 4], "b": [0, 1, 2, 5, 6]})
base = _config(vector_dim=8)
base.doctor = DoctorConfig(
duplicates=DuplicateDetectionConfig(candidate_threshold=0.0)
)
assert (
_result(
await run_doctor(base, temp_db_path, {}), "duplicate_documents"
).severity
is Severity.OK
)
tuned = _config(vector_dim=8)
tuned.doctor = DoctorConfig(
duplicates=DuplicateDetectionConfig(
candidate_threshold=0.0, containment_threshold=0.6
)
)
assert (
_result(
await run_doctor(tuned, temp_db_path, {}), "duplicate_documents"
).severity
is Severity.WARN
)