diff --git a/admin/app/services/rag_service.ts b/admin/app/services/rag_service.ts index 7bf77f52..74505480 100644 --- a/admin/app/services/rag_service.ts +++ b/admin/app/services/rag_service.ts @@ -45,6 +45,7 @@ const EXCLUDE_EVAL_FILTER = { } import type { KbIngestStateValue } from '../../types/kb_ingest_state.js' import { ZIMExtractionService } from './zim_extraction_service.js' +import { chunkTextHash, dedupeZimChunks } from '../utils/zim_chunk_dedupe.js' import { ZIM_BATCH_SIZE } from '../../constants/zim_extraction.js' import { hasMoreArticleBatches } from '../utils/zim_batch_decision.js' import { EMBEDDING_MODEL_NAME } from '../../constants/ollama.js' @@ -186,6 +187,12 @@ export class RagService { field_name: 'active', field_schema: 'bool', }) + // Lets processZIMFile ask "which of these texts are already stored for this + // file?" in one filtered scroll instead of a table scan (#1438). + await this.qdrant!.createPayloadIndex(collectionName, { + field_name: 'text_hash', + field_schema: 'keyword', + }) // Backfill: stamp `active: true` on any point that predates this field. // `is_empty` (not a match-all filter) means this only ever touches points @@ -632,13 +639,29 @@ export class RagService { `[RAG] Extracted ${zimChunks.length} chunks from ZIM file with enhanced metadata (file totalArticles=${totalArticles})` ) + // Store each distinct text once per file. Scraped-site ZIMs repeat their + // banners, disclaimers and sidebars in every article, and those identical + // vectors degrade HNSW recall for the whole collection (#1438). Earlier + // batches win: a text already in Qdrant for this file is skipped here. + const storedHashes = await this._findStoredTextHashes( + filepath, + zimChunks.map((c) => c.text) + ) + const { kept, droppedInBatch, droppedStored } = dedupeZimChunks(zimChunks, storedHashes) + if (droppedInBatch + droppedStored > 0) { + logger.info( + `[RAG] Skipping ${droppedInBatch + droppedStored} duplicate chunks (${droppedInBatch} repeated in this batch, ${droppedStored} already stored)` + ) + } + // Process each chunk individually with its metadata let totalChunks = 0 - for (let i = 0; i < zimChunks.length; i++) { - const zimChunk = zimChunks[i] + for (let i = 0; i < kept.length; i++) { + const { chunk: zimChunk, hash: textHash } = kept[i] const result = await this.embedAndStoreText(zimChunk.text, { source: filepath, content_type: 'zim_article', + text_hash: textHash, // Without this the ZIM path writes points with no `collection` at all, so // getKnowledgeCollections() (which facets on it) never sees them. ...(collection ? { collection } : {}), @@ -673,7 +696,7 @@ export class RagService { } if (onProgress) { - await onProgress(((i + 1) / zimChunks.length) * 100) + await onProgress(((i + 1) / kept.length) * 100) } } @@ -2319,6 +2342,47 @@ export class RagService { } } + /** + * Which of these chunk texts does Qdrant already hold for `source`? Returns + * their hashes (see `chunkTextHash`). Points written before #1438 carry no + * `text_hash`, so they are invisible here; only newly ingested files get the + * cross-batch guarantee, and a re-index of an older file starts clean anyway + * because its points are deleted first. + */ + private async _findStoredTextHashes(source: string, texts: string[]): Promise> { + const found = new Set() + if (texts.length === 0) return found + + // First call in a batch job: this both connects the client and creates the + // `text_hash` payload index the filter below relies on. + await this._ensureCollection(RagService.CONTENT_COLLECTION_NAME) + + const hashes = [...new Set(texts.map((t) => chunkTextHash(t)))] + let offset: string | number | undefined + + do { + const page = await this.qdrant!.scroll(RagService.CONTENT_COLLECTION_NAME, { + filter: { + must: [ + { key: 'source', match: { value: this.sanitizeText(source) } }, + { key: 'text_hash', match: { any: hashes } }, + ], + }, + with_payload: ['text_hash'], + with_vector: false, + limit: hashes.length, + ...(offset !== undefined ? { offset } : {}), + }) + for (const point of page.points) { + const hash = point.payload?.text_hash + if (typeof hash === 'string') found.add(hash) + } + offset = (page.next_page_offset as string | number | null | undefined) ?? undefined + } while (offset !== undefined) + + return found + } + /** * Delete all Qdrant points whose `source` payload matches the given path. * Unlike deleteFileBySource(), this does NOT touch the file on disk — used diff --git a/admin/app/utils/zim_chunk_dedupe.ts b/admin/app/utils/zim_chunk_dedupe.ts new file mode 100644 index 00000000..a53dc7ea --- /dev/null +++ b/admin/app/utils/zim_chunk_dedupe.ts @@ -0,0 +1,64 @@ +import { createHash } from 'node:crypto' + +/** + * Exact-duplicate suppression for ZIM chunks (#1438). + * + * Section-level chunking turns every article's shared chrome (site banners, + * disclaimers, sidebar link lists, survey prompts) into a chunk of its own, so a + * scraped-site ZIM such as MedlinePlus ends up with a third of its points being + * byte-identical copies: the same text embedded once per article. Thousands of + * identical vectors form cliques in Qdrant's HNSW graph that the search walk + * cannot leave, and approximate search then stops returning points an exact + * search still finds. The fix is simply to never store the same text twice for + * the same file. + * + * This is the pure, I/O-free core: hashing and the keep/drop decision. The + * caller (`RagService.processZIMFile`) supplies the hashes already stored for + * the file, because a ZIM is ingested in independent batches and the first + * batch that saw a text is the one that keeps it. + */ + +/** Stable fingerprint of a chunk's text, independent of surrounding whitespace. */ +export function chunkTextHash(text: string): string { + const normalized = text.replace(/\s+/g, ' ').trim() + return createHash('sha1').update(normalized).digest('hex') +} + +export interface DedupeResult { + /** Chunks to embed, in their original order, each tagged with its hash. */ + kept: Array<{ chunk: T; hash: string }> + /** Chunks dropped because an earlier chunk in this batch has the same text. */ + droppedInBatch: number + /** Chunks dropped because an earlier batch already stored the same text. */ + droppedStored: number +} + +/** + * Keep the first occurrence of each text; drop later copies, whether they come + * from this batch or (via `storedHashes`) from a batch that already ran. + */ +export function dedupeZimChunks( + chunks: T[], + storedHashes: ReadonlySet +): DedupeResult { + const seen = new Set() + const kept: Array<{ chunk: T; hash: string }> = [] + let droppedInBatch = 0 + let droppedStored = 0 + + for (const chunk of chunks) { + const hash = chunkTextHash(chunk.text) + if (seen.has(hash)) { + droppedInBatch++ + continue + } + seen.add(hash) + if (storedHashes.has(hash)) { + droppedStored++ + continue + } + kept.push({ chunk, hash }) + } + + return { kept, droppedInBatch, droppedStored } +} diff --git a/admin/tests/unit/zim_chunk_dedupe.spec.ts b/admin/tests/unit/zim_chunk_dedupe.spec.ts new file mode 100644 index 00000000..5dd9006b --- /dev/null +++ b/admin/tests/unit/zim_chunk_dedupe.spec.ts @@ -0,0 +1,65 @@ +import * as assert from 'node:assert/strict' +import { test } from 'node:test' + +import { chunkTextHash, dedupeZimChunks } from '../../app/utils/zim_chunk_dedupe.js' + +test('chunkTextHash ignores whitespace differences only', () => { + assert.equal( + chunkTextHash('Official websites use .gov'), + chunkTextHash(' Official websites\nuse .gov ') + ) + assert.notEqual( + chunkTextHash('Official websites use .gov'), + chunkTextHash('Official websites use .GOV') + ) +}) + +test('dedupeZimChunks keeps the first copy of a text within a batch (#1438)', () => { + const banner = + 'Official websites use .gov. A .gov website belongs to an official government organization.' + const chunks = [ + { text: 'Aspirin is used to treat pain.', articleTitle: 'Aspirin' }, + { text: banner, articleTitle: 'Aspirin' }, + { text: 'Ibuprofen is an NSAID.', articleTitle: 'Ibuprofen' }, + { text: banner, articleTitle: 'Ibuprofen' }, + { text: banner, articleTitle: 'Warfarin' }, + ] + + const result = dedupeZimChunks(chunks, new Set()) + + assert.deepEqual( + result.kept.map((k) => k.chunk.articleTitle + ': ' + k.chunk.text.slice(0, 8)), + ['Aspirin: Aspirin ', 'Aspirin: Official', 'Ibuprofen: Ibuprofe'] + ) + assert.equal(result.droppedInBatch, 2) + assert.equal(result.droppedStored, 0) + assert.equal(result.kept[1].hash, chunkTextHash(banner)) +}) + +test('dedupeZimChunks drops texts an earlier batch already stored', () => { + const disclaimer = + 'The information on this site should not be used as a substitute for professional medical care.' + const chunks = [{ text: disclaimer }, { text: disclaimer }, { text: 'Take with food.' }] + + const result = dedupeZimChunks(chunks, new Set([chunkTextHash(disclaimer)])) + + assert.deepEqual( + result.kept.map((k) => k.chunk.text), + ['Take with food.'] + ) + // The second in-batch copy counts against the batch, not the store: it would + // have been dropped either way, and the counts should add up to the input. + assert.equal(result.droppedStored, 1) + assert.equal(result.droppedInBatch, 1) + assert.equal(result.kept.length + result.droppedStored + result.droppedInBatch, chunks.length) +}) + +test('dedupeZimChunks passes unique chunks through untouched', () => { + const chunks = [{ text: 'a' }, { text: 'b' }, { text: 'c' }] + const result = dedupeZimChunks(chunks, new Set()) + assert.deepEqual( + result.kept.map((k) => k.chunk), + chunks + ) + assert.equal(result.droppedInBatch + result.droppedStored, 0) +})