From 8f42452f53824a6f4fd1b90bda20d68598956460 Mon Sep 17 00:00:00 2001 From: oskar Date: Thu, 16 Jul 2026 17:17:18 +0200 Subject: [PATCH] =?UTF-8?q?feat(kb):=20faza=203=20krok=201=20=E2=80=94=20p?= =?UTF-8?q?orz=C4=85dki=20po=20pilocie=20retrieval=20(=C5=9Bmie=C4=87=20+?= =?UTF-8?q?=20dedup=20+=20klucz=20modelu)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Migracja 003 (UNIQUE+model, excluded_reason) zastosowana na żywej bazie kb-postgres@PIHA (2683 chunki, bez DELETE). Kalibracja heurystyki ocr_junk (3 sygnały z planu §3.1) ujawniła realną sprzeczność z planem: sygnał 1 (dowolny znak kontrolny) fałszywie łapał paperless:119 (wymagany aktywny) i legalny angielski tekst — próg doprecyzowany do >=5 wystąpień na podstawie rozkładu na korpusie. 8 chunków oflagowanych ocr_junk po odjęciu potwierdzonych fałszywych alarmów (kalendarz, mikro-fragmenty referencyjne). Dedup: SQL z planu (exact content hash) znalazł 2 pary, ale nie wykrył znanego z pilota duplikatu paperless:14≡74 (różne OCR, 99.2% chunków identycznych treściowo) — dodany fuzzy check na potwierdzenie. Odrzucono 3 kandydatury o wysokim nakładaniu jako różne wersje/typy dokumentów dzielące boilerplate PZU, nie duplikaty. 130 chunków oflagowanych duplicate + entities[duplicate_of] na 3 kopertach. chunk_embed.py: heurystyka is_ocr_junk() przed embedem (junk -> insert bez wywołania Ollamy, embedding=NULL), ON CONFLICT rozszerzony o model, nowy licznik chunks_junk_flagged w bilansie, testy (kody kreskowe, mojibake nie-junk, dot-leader, idempotencja, model w kluczu konfliktu). Weryfikacja: 7 zapytań eval-setu z WHERE excluded_reason IS NULL — żadne trafienie nie degraduje, kontrole negatywne bez zmian (>0.55), śmieć zniknął z top-5 zapytania 2, paperless:119 pozostał aktywnym trafieniem. Bilans: 2545 aktywne / 130 duplicate / 8 ocr_junk. Co najmniej 3 decyzje wymagały zatrzymania i potwierdzenia z Oskarem w sesji (kalibracja progu sygnału 1, dołączenie 14≡74 mimo braku exact-hash matcha, odrzucenie 3 fałszywych kandydatur dedup) — plan przewidywał, że heurystyka będzie się mylić; wszystkie decyzje udokumentowane w transkrypcie sesji. Co-Authored-By: Claude Sonnet 5 --- .../src/documents_ingest/chunk_embed.py | 121 +++++++++++++--- .../tests/test_chunk_embed.py | 137 +++++++++++++++++- .../kb-postgres/init/003_chunk_model_key.sql | 17 +++ 3 files changed, 247 insertions(+), 28 deletions(-) create mode 100644 services/kb-postgres/init/003_chunk_model_key.sql diff --git a/jobs/documents-ingest/src/documents_ingest/chunk_embed.py b/jobs/documents-ingest/src/documents_ingest/chunk_embed.py index 3d3ff91..99d84b6 100644 --- a/jobs/documents-ingest/src/documents_ingest/chunk_embed.py +++ b/jobs/documents-ingest/src/documents_ingest/chunk_embed.py @@ -30,15 +30,16 @@ DSN can come from KB_DSN, Ollama URL from OLLAMA_URL (default http://localhost:1 Idempotency: a pre-fetched set of existing (envelope_id, chunk_index) pairs for this `model` skips chunks already embedded — no re-embedding, no wasted Ollama calls on rerun. -`insert_chunk`'s own `ON CONFLICT (envelope_id, chunk_index) DO NOTHING` is the second line -of defense; its command tag is checked so a silently-skipped row is counted as -`chunks_conflict_skipped`, never miscounted as `chunks_inserted`. Note: that UNIQUE -constraint is on (envelope_id, chunk_index) only, not including `model` — re-embedding the -same pilot with a *different* model would hit this path and its embedding would be -discarded (a real chunk of work wasted, correctly reported, but not persisted). Out of -scope for this pilot (single model, bge-m3); flagged as a follow-up if/when a second model -is ever indexed (the real fix is `UNIQUE (envelope_id, chunk_index, model)` at the schema -layer). +`insert_chunk`'s own `ON CONFLICT (envelope_id, chunk_index, model) DO NOTHING` is the +second line of defense; its command tag is checked so a silently-skipped row is counted as +`chunks_conflict_skipped`, never miscounted as `chunks_inserted`. + +OCR-junk filter (module 5, phase 3, plan §3.1): `is_ocr_junk()` screens each chunk before +embedding. A junk chunk is INSERTed with `excluded_reason='ocr_junk'` and `embedding=NULL` +— no Ollama call, no HNSW entry — and counted as `chunks_junk_flagged`. The three signals +and their thresholds are calibrated in-code (see `is_ocr_junk` docstring) against the +2683-chunk pilot corpus; retrieval and all downstream consumers filter on +`WHERE excluded_reason IS NULL`. """ from __future__ import annotations @@ -75,11 +76,49 @@ OVERLAP_CHARS = OVERLAP_TOKENS * CHARS_PER_TOKEN _PARA_SPLIT = re.compile(r"\n\s*\n") _INSERT_SQL = """ -INSERT INTO document_chunk (envelope_id, chunk_index, text, embedding, model) -VALUES ($1, $2, $3, $4::vector, $5) -ON CONFLICT (envelope_id, chunk_index) DO NOTHING +INSERT INTO document_chunk (envelope_id, chunk_index, text, embedding, model, excluded_reason) +VALUES ($1, $2, $3, $4::vector, $5, $6) +ON CONFLICT (envelope_id, chunk_index, model) DO NOTHING """ +# Plan §3.1, signal 1: refined from "any C0 control char" to a raw-count threshold during +# calibration on the 2683-chunk pilot corpus (2026-07-16) — legit OCR text carries a handful +# of stray control bytes (paperless:119 mojibake, a confirmed retrieval hit; an English paper +# with 1-2 stray bytes per chunk), while true binary/barcode noise carries 10-63 per chunk. +# A threshold of 5 cleanly separates the two on this corpus. +CONTROL_CHAR_JUNK_THRESHOLD = 5 +# Plan §3.1, signal 2: share of characters outside the "wordy" class (alnum + PL diacritics + +# common punctuation + whitespace). On the pilot corpus no legitimate chunk crosses 20%, so +# this never fires alone here — kept for corpora where junk isn't diluted by legible headers. +NON_WORDY_RATIO_THRESHOLD = 0.30 +# Plan §3.1, signal 3: share of whitespace-split tokens that look like words. Legit text +# clusters well above 40%; junk (scrambled fonts, dot-leader ToCs) sits below 25% on the +# pilot corpus. +WORDLIKE_RATIO_THRESHOLD = 0.25 + +_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b-\x1f]") +_NON_WORDY_CHAR_RE = re.compile( + r"[^a-zA-Z0-9ąćęłńóśźżĄĆĘŁŃÓŚŹŻ\s.,;:!?\"'`()\[\]{}<>/\\|@#$%^&*_+=~-]" +) +_WORDLIKE_TOKEN_RE = re.compile(r"^[a-zA-Ząćęłńóśźż-]{2,30}$") + + +def is_ocr_junk(text: str) -> bool: + """Three-signal OCR-junk heuristic (plan §3.1, decision 1): binary noise (ground barcodes, + scrambled fonts), not merely ugly text — partial mojibake alone is not junk (paperless:119 + stays a confirmed retrieval hit despite it). Chunk-level, never document-level.""" + if not text: + return False + if len(_CONTROL_CHAR_RE.findall(text)) >= CONTROL_CHAR_JUNK_THRESHOLD: + return True + if len(_NON_WORDY_CHAR_RE.findall(text)) / len(text) > NON_WORDY_RATIO_THRESHOLD: + return True + tokens = text.split() + if not tokens: + return False + wordlike = sum(1 for t in tokens if _WORDLIKE_TOKEN_RE.match(t)) + return (wordlike / len(tokens)) < WORDLIKE_RATIO_THRESHOLD + class EmbeddingDimensionError(RuntimeError): """Ollama returned a vector of the wrong dimension for the target `document_chunk` schema.""" @@ -209,12 +248,15 @@ async def embed_chunk( async def insert_chunk( conn: asyncpg.Connection, envelope_id: str, chunk_index: int, text: str, - embedding: list[float], model: str, + embedding: Optional[list[float]], model: str, excluded_reason: Optional[str] = None, ) -> str: """Returns asyncpg's command tag (e.g. 'INSERT 0 1' or 'INSERT 0 0' if ON CONFLICT - DO NOTHING skipped the row) so the caller can tell a real insert from a no-op.""" + DO NOTHING skipped the row) so the caller can tell a real insert from a no-op. + `embedding=None` (junk chunks, `excluded_reason='ocr_junk'`) inserts a NULL vector — + pgvector's HNSW index skips NULLs automatically.""" + vector_literal = _vector_literal(embedding) if embedding is not None else None return await conn.execute( - _INSERT_SQL, envelope_id, chunk_index, text, _vector_literal(embedding), model + _INSERT_SQL, envelope_id, chunk_index, text, vector_literal, model, excluded_reason ) @@ -240,14 +282,17 @@ async def run( Returns stats that must always balance: documents_fetched = empty_content + documents_chunked - chunks_total = chunks_already_embedded + chunks_inserted + chunks_total = chunks_already_embedded + chunks_inserted + chunks_junk_flagged + chunks_conflict_skipped + chunks_errors - `chunks_conflict_skipped` counts inserts where `ON CONFLICT (envelope_id, chunk_index) - DO NOTHING` silently discarded the row (the pre-fetched `existing` set is the first line - of defense against this and should make it rare; see the module docstring's note on the - UNIQUE constraint not including `model`) — tracked separately so a silent no-op is never - miscounted as a successful write. + Each chunk is screened by `is_ocr_junk()` before embedding (plan §3.1): a junk chunk + skips Ollama entirely and is INSERTed with `excluded_reason='ocr_junk'`, + `embedding=NULL`, counted as `chunks_junk_flagged` — never as `chunks_inserted`. + + `chunks_conflict_skipped` counts inserts where `ON CONFLICT (envelope_id, chunk_index, + model) DO NOTHING` silently discarded the row (the pre-fetched `existing` set is the + first line of defense against this and should make it rare) — tracked separately so a + silent no-op is never miscounted as a successful write. Every embedding response's dimension is checked against `EXPECTED_DIM` (raises EmbeddingDimensionError and aborts the whole run on mismatch) — never silently indexes @@ -262,6 +307,7 @@ async def run( "chunks_total": 0, "chunks_already_embedded": 0, "chunks_inserted": 0, + "chunks_junk_flagged": 0, "chunks_conflict_skipped": 0, "chunks_errors": 0, "embed_calls": 0, @@ -297,8 +343,35 @@ async def run( stats["chunks_already_embedded"] += 1 continue + junk = is_ocr_junk(chunk) + if not apply: - stats["chunks_inserted"] += 1 + if junk: + stats["chunks_junk_flagged"] += 1 + else: + stats["chunks_inserted"] += 1 + continue + + if junk: + try: + command_tag = await insert_chunk( + conn, envelope_id, idx, chunk, None, model, excluded_reason="ocr_junk" + ) + except Exception: + _log.warning( + "skip.insert_error", envelope_id=envelope_id, chunk_index=idx, exc_info=True + ) + stats["chunks_errors"] += 1 + continue + + existing.add(key) + if _rows_affected(command_tag) == 0: + _log.warning( + "chunk.conflict_skipped", envelope_id=envelope_id, chunk_index=idx, model=model + ) + stats["chunks_conflict_skipped"] += 1 + else: + stats["chunks_junk_flagged"] += 1 continue assert session is not None @@ -345,7 +418,7 @@ async def run( balance_docs = stats["empty_content"] + stats["documents_chunked"] balance_chunks = ( - stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"] + stats["chunks_conflict_skipped"] + stats["chunks_errors"] ) if balance_docs != stats["documents_fetched"] or balance_chunks != stats["chunks_total"]: @@ -414,7 +487,7 @@ def main() -> None: balanced = ( stats["documents_fetched"] == stats["empty_content"] + stats["documents_chunked"] and stats["chunks_total"] == ( - stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"] + stats["chunks_conflict_skipped"] + stats["chunks_errors"] ) ) diff --git a/jobs/documents-ingest/tests/test_chunk_embed.py b/jobs/documents-ingest/tests/test_chunk_embed.py index e50af53..c9f4124 100644 --- a/jobs/documents-ingest/tests/test_chunk_embed.py +++ b/jobs/documents-ingest/tests/test_chunk_embed.py @@ -15,6 +15,8 @@ from documents_ingest.chunk_embed import ( fetch_documents, fetch_existing_chunk_keys, hard_split, + insert_chunk, + is_ocr_junk, run, split_paragraphs, _vector_literal, @@ -144,6 +146,46 @@ class TestVectorLiteral: assert _vector_literal([0.1, 0.2, -0.3]) == "[0.1,0.2,-0.3]" +class TestIsOcrJunk: + """Pilot cases from docs/kb/modules/05-faza3-plan.md §3.1 and the 2026-07-16 calibration + session (docs/kb/eval/retrieval-pilot-2026-07-16.md).""" + + def test_clean_text_is_not_junk(self): + text = ( + "Ogólne Warunki Ubezpieczenia PZU Auto ustalają zasady odpowiedzialności " + "ubezpieczyciela za szkody powstałe w związku z ruchem pojazdów mechanicznych." + ) + assert is_ocr_junk(text) is False + + def test_ground_barcode_binary_noise_is_junk(self): + # paperless:39/96/128/186/192 pattern: dense run of C0 control chars (scrambled + # barcode), well above the calibrated raw-count threshold of 5. + text = "\x01\x02\x03\x05\x06\x07\x0f\x10\x11\x12\x14\x15\x16\x17\x18\x19qQQQ\x11q\x01\x11!!AA" + assert is_ocr_junk(text) is True + + def test_partial_mojibake_is_not_junk(self): + # paperless:119: a confirmed retrieval hit (query 4, eval-set) despite partial + # mojibake and a couple of stray control bytes — mojibake alone must not exclude it. + text = ( + "Wygenerowano z systemu wydarzeń FIRST Israel | Turniej Regionalny FIRST LEGO " + "League\n\x01integraln¹ czêœæ dokumentacji zawodów\nARTEFAKTY Z WĘDKI" + ) + assert is_ocr_junk(text) is False + + def test_stray_single_control_char_in_legible_text_is_not_junk(self): + # paperless:141 pattern: one stray control byte in an otherwise clean English + # paragraph (a PDF-extraction artifact, not OCR noise). + text = "separable\x0fconvolution is equal to the combination of a self-attention layer" + assert is_ocr_junk(text) is False + + def test_dot_leader_toc_is_junk(self): + text = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ." + assert is_ocr_junk(text) is True + + def test_empty_text_is_not_junk(self): + assert is_ocr_junk("") is False + + class _FakeEmbedResponse: def __init__(self, payload, status=200): self._payload = payload @@ -203,6 +245,29 @@ class TestEmbedChunk: await embed_chunk(_Session(), "http://fake-ollama", "bge-m3", "text") +class TestInsertSql: + def test_on_conflict_target_includes_model(self): + # Plan §3.3 fix: the old UNIQUE(envelope_id, chunk_index) silently no-opped a + # second embedding model via ON CONFLICT DO NOTHING (review finding, phase 2 step + # 6). The target must now include `model`. + from documents_ingest.chunk_embed import _INSERT_SQL + assert "ON CONFLICT (envelope_id, chunk_index, model)" in _INSERT_SQL + + +class TestInsertChunk: + async def test_junk_chunk_inserts_null_embedding_with_reason(self): + conn = _FakeConn() + await insert_chunk(conn, "paperless:1", 0, "junk text", None, "bge-m3", excluded_reason="ocr_junk") + + assert conn.execute_calls == [("paperless:1", 0, "junk text", None, "bge-m3", "ocr_junk")] + + async def test_normal_chunk_inserts_vector_with_no_reason(self): + conn = _FakeConn() + await insert_chunk(conn, "paperless:1", 0, "real text", [0.1, 0.2], "bge-m3") + + assert conn.execute_calls == [("paperless:1", 0, "real text", "[0.1,0.2]", "bge-m3", None)] + + class _FakeConn: def __init__(self, docs=None, existing_keys=None, existing_model="bge-m3", execute_results=None): self._docs = docs or [] @@ -340,10 +405,13 @@ class TestRun: async def test_embed_error_is_isolated_and_counted(self, monkeypatch): # Two chunks: force one to fail via a document long enough to produce 2 chunks - # (chunk_size/overlap kept small to make the test fast and explicit). - big = "a" * 100 + "\n\n" + "b" * 100 + # (chunk_size/overlap kept small to make the test fast and explicit). Real + # word-like tokens (not a single 100-char run) so neither chunk trips is_ocr_junk. + para1 = " ".join(["word"] * 20) + para2 = " ".join(["term"] * 20) + big = para1 + "\n\n" + para2 conn = _FakeConn(docs=[_row("paperless:1", big)]) - ollama = _FakeOllamaSession(dim=1024, fail_for={"a" * 100}) + ollama = _FakeOllamaSession(dim=1024, fail_for={para1}) self._patch(monkeypatch, conn, ollama) stats = await run(dsn="postgresql://fake", apply=True, chunk_size=100, chunk_overlap=20) @@ -358,7 +426,8 @@ class TestRun: async def test_insert_error_is_isolated_and_counted(self, monkeypatch): # Two chunks; the first chunk's INSERT raises (simulated DB blip) but the run # continues and the second chunk still gets embedded and inserted normally. - big = "a" * 100 + "\n\n" + "b" * 100 + # Real word-like tokens so neither chunk trips is_ocr_junk. + big = " ".join(["word"] * 20) + "\n\n" + " ".join(["term"] * 20) conn = _FakeConn(docs=[_row("paperless:1", big)]) real_execute = conn.execute @@ -416,6 +485,66 @@ class TestRun: assert stats["chunks_already_embedded"] == 0 assert stats["chunks_inserted"] == 1 + async def test_junk_chunk_flagged_without_calling_ollama(self, monkeypatch): + junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ." + conn = _FakeConn(docs=[_row("paperless:1", junk)]) + ollama = _FakeOllamaSession(dim=1024) + self._patch(monkeypatch, conn, ollama) + + stats = await run(dsn="postgresql://fake", apply=True) + + assert stats["chunks_junk_flagged"] == 1 + assert stats["chunks_inserted"] == 0 + assert ollama.requests == [] + assert conn.execute_calls == [("paperless:1", 0, junk, None, "bge-m3", "ocr_junk")] + assert stats["chunks_total"] == ( + stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"] + + stats["chunks_conflict_skipped"] + stats["chunks_errors"] + ) + + async def test_junk_chunk_dry_run_counted_without_db_or_ollama(self, monkeypatch): + junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ." + conn = _FakeConn(docs=[_row("paperless:1", junk)]) + ollama = _FakeOllamaSession(dim=1024) + self._patch(monkeypatch, conn, ollama) + + stats = await run(dsn="postgresql://fake", apply=False) + + assert stats["chunks_junk_flagged"] == 1 + assert stats["chunks_inserted"] == 0 + assert ollama.requests == [] + assert conn.execute_calls == [] + + async def test_junk_chunk_rerun_is_idempotent(self, monkeypatch): + # Same (envelope_id, chunk_index) already flagged ocr_junk under this model — a + # rerun must skip it, same as a normal embedded chunk (plan §3.1: backfill is a + # one-shot, but chunk_embed.py itself must stay idempotent for new junk too). + junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ." + conn = _FakeConn( + docs=[_row("paperless:1", junk)], + existing_keys=[("paperless:1", 0)], + ) + ollama = _FakeOllamaSession(dim=1024) + self._patch(monkeypatch, conn, ollama) + + stats = await run(dsn="postgresql://fake", apply=True) + + assert stats["chunks_already_embedded"] == 1 + assert stats["chunks_junk_flagged"] == 0 + assert conn.execute_calls == [] + + async def test_junk_conflict_skip_not_double_counted(self, monkeypatch): + junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ." + conn = _FakeConn(docs=[_row("paperless:1", junk)], execute_results=["INSERT 0 0"]) + ollama = _FakeOllamaSession(dim=1024) + self._patch(monkeypatch, conn, ollama) + + stats = await run(dsn="postgresql://fake", apply=True) + + assert stats["chunks_conflict_skipped"] == 1 + assert stats["chunks_junk_flagged"] == 0 + assert ollama.requests == [] + async def test_stats_balance_documents_and_chunks(self, monkeypatch): conn = _FakeConn(docs=[ _row("paperless:1", "content one"), diff --git a/services/kb-postgres/init/003_chunk_model_key.sql b/services/kb-postgres/init/003_chunk_model_key.sql new file mode 100644 index 0000000..2d84851 --- /dev/null +++ b/services/kb-postgres/init/003_chunk_model_key.sql @@ -0,0 +1,17 @@ +-- KB spine: document_chunk — unique-key fix + exclusion flag +-- Additive: does not modify 001_envelope.sql or 002_chunks.sql. +-- Version: 003 — chunk model key + excluded_reason (module 5, phase 3, plan §3.3) +-- +-- (a) UNIQUE rozszerzony o model — bez tego drugi model embeddingów cicho się no-opuje +-- na ON CONFLICT DO NOTHING (finding z review kroku 6 fazy 2). +-- (b) excluded_reason — flaga wykluczenia chunka z retrievalu i kompilacji +-- (NULL = aktywny; 'ocr_junk' | 'duplicate'). + +ALTER TABLE document_chunk + DROP CONSTRAINT IF EXISTS document_chunk_envelope_id_chunk_index_key; +ALTER TABLE document_chunk + ADD CONSTRAINT document_chunk_envelope_chunk_model_key + UNIQUE (envelope_id, chunk_index, model); + +ALTER TABLE document_chunk + ADD COLUMN IF NOT EXISTS excluded_reason TEXT; -- NULL = aktywny