From 6541fee4323a94ddf8a5061fdaa36d4ec146d64f Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:00:37 +0300 Subject: [PATCH 01/10] fix(ingestion): derive deterministic chunk UIDs so re-ingest stops duplicating `TextChunk.uid` defaults to `uuid4()`, so every ingest of the same document minted brand-new chunk IDs. The lexical graph MERGEs chunks on that ID, so nothing ever matched and re-ingesting silently duplicated the whole chunk layer. Measured on the benchmark corpus, ingesting one unchanged document three times in a row: round 1: 17 Chunks, 171 MENTIONED_IN round 2: 34 Chunks, 343 MENTIONED_IN round 3: 51 Chunks, 513 MENTIONED_IN Entities were unaffected (94 -> 95) because those MERGE on a normalised name, which is stable. Only the lexical layer duplicated. The UID is now derived from the owning document's UID, the chunk index and a hash of the chunk text. Including the document UID keeps two documents that share a paragraph as distinct chunks; including the text means an edited document produces new chunks rather than silently overwriting old text under a recycled `doc:index` key. This runs after chunking and overwrites unconditionally, so chunk identity is decided in exactly one place. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/graphrag_sdk/ingestion/pipeline.py | 40 ++++++++++++++++ graphrag_sdk/tests/test_pipeline.py | 48 +++++++++++++++++++ 2 files changed, 88 insertions(+) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index 9428757f..94250491 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -38,6 +38,44 @@ _PATTERN_MISMATCH_SAMPLE_SIZE = 3 +def _assign_deterministic_chunk_uids( + doc_info: DocumentInfo, chunks: TextChunks +) -> None: + """Replace random chunk UIDs with a content-derived, stable identity. + + ``TextChunk.uid`` defaults to ``uuid4()``, so every ingest of the *same* + document minted brand-new chunk IDs. The lexical graph MERGEs chunks on + that ID, so nothing ever matched and re-ingesting silently duplicated the + whole chunk layer. + + Measured on the benchmark corpus before this fix — ingesting one unchanged + document three times in a row:: + + round 1: 17 Chunks, 171 MENTIONED_IN + round 2: 34 Chunks, 343 MENTIONED_IN + round 3: 51 Chunks, 513 MENTIONED_IN + + Entities were unaffected (94 -> 95) because those MERGE on a normalised + name, which *is* stable. Only the lexical layer duplicated. + + The UID is derived from the owning document's UID, the chunk index and a + hash of the chunk text. Including the document UID keeps two documents that + happen to share a paragraph as distinct chunks; including the text means an + edited document produces new chunks rather than silently overwriting the + old text under a recycled ``doc:index`` key. + + Chunkers that deliberately assign their own meaningful UIDs are not + special-cased: this runs after chunking and overwrites unconditionally, so + identity is decided in exactly one place. + """ + doc_key = doc_info.uid or doc_info.path or "" + for chunk in chunks.chunks: + digest = hashlib.sha256( + f"{doc_key}\x00{chunk.index}\x00{chunk.text}".encode() + ).hexdigest() + chunk.uid = f"chunk-{digest[:32]}" + + class IngestionPipeline: """Sequential orchestrator for knowledge graph construction. @@ -181,6 +219,8 @@ async def run( ctx.log("No chunks produced, pipeline complete") return IngestionResult(document_info=document.document_info) + _assign_deterministic_chunk_uids(document.document_info, chunks) + # Step 3: Build lexical graph (MANDATORY — not a strategy). # Hash the loaded text so ``GraphRAG.update()`` can short-circuit # when content is unchanged. SHA-256 hex; cost is negligible diff --git a/graphrag_sdk/tests/test_pipeline.py b/graphrag_sdk/tests/test_pipeline.py index 9421d5c6..5d97dae9 100644 --- a/graphrag_sdk/tests/test_pipeline.py +++ b/graphrag_sdk/tests/test_pipeline.py @@ -696,3 +696,51 @@ def test_prune_pattern_mismatch_sample_is_bounded(self, caplog): # Total count is reported, but the sampled list does not contain 50 entries. assert "Pruned 50" in msg assert msg.count("('Company', 'Person')") <= 3 + + +class TestDeterministicChunkUids: + """Bug #12 — re-ingesting an unchanged document duplicated its chunks. + + ``TextChunk.uid`` defaulted to ``uuid4()``, so the lexical graph's + ``MERGE (c:Chunk {id: ...})`` never matched an existing chunk. Measured on + the benchmark corpus, ingesting one unchanged file three times: + 17 -> 34 -> 51 Chunks and 171 -> 343 -> 513 MENTIONED_IN. After the fix the + same three rounds hold at 17 Chunks / 170 MENTIONED_IN. + """ + + @staticmethod + def _chunks(texts, doc_uid="doc-1"): + from graphrag_sdk.core.models import DocumentInfo, TextChunk, TextChunks + from graphrag_sdk.ingestion.pipeline import _assign_deterministic_chunk_uids + + info = DocumentInfo(uid=doc_uid, path="/tmp/a.txt") + chunks = TextChunks( + chunks=[TextChunk(text=t, index=i) for i, t in enumerate(texts)] + ) + _assign_deterministic_chunk_uids(info, chunks) + return [c.uid for c in chunks.chunks] + + def test_same_document_yields_same_uids(self): + a = self._chunks(["alpha", "beta"]) + b = self._chunks(["alpha", "beta"]) + assert a == b + + def test_uids_replace_the_random_default(self): + from graphrag_sdk.core.models import TextChunk + + assert self._chunks(["alpha"])[0] != TextChunk(text="alpha", index=0).uid + + def test_different_text_yields_different_uid(self): + assert self._chunks(["alpha"]) != self._chunks(["alpha edited"]) + + def test_different_index_yields_different_uid(self): + """Two chunks with identical text must stay distinct nodes.""" + uids = self._chunks(["same", "same"]) + assert uids[0] != uids[1] + + def test_different_document_yields_different_uid(self): + assert self._chunks(["alpha"], "doc-1") != self._chunks(["alpha"], "doc-2") + + def test_uids_are_unique_within_a_document(self): + uids = self._chunks([f"chunk {i}" for i in range(50)]) + assert len(set(uids)) == 50 From 123c3fd3c0ae0a5a81d29ab1ac3a8f8b92c7f7f0 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:00:49 +0300 Subject: [PATCH 02/10] perf(chunking): lower SentenceTokenCapChunking default from 512 to 384 tokens 384 wins on the effect that dominates graph quality: the relationship extractor returns a roughly constant number of relationships per call regardless of how much text it is given. Doubling the input grew the reply by 1.3%, so smaller chunks mean more calls and more extracted facts. Measured on an 11-document benchmark with the current default model, chunk 384 vs 768: entity F1 0.574 vs 0.563, relation F1 0.237 vs 0.223. With the previous default model, quality fell monotonically as chunks grew (0.233 / 0.221 / 0.204 at 384 / 768 / 1536). This cuts against most RAG guidance, which suggests 1024+. Those defaults come from pipelines that use an LLM for entity extraction and so do not share this per-call ceiling. Raising the cap is a way to trade recall for lower cost, not a quality improvement. The docstring also records that 384 is no longer justified by GLiNER's `config.max_len`: the extractor now windows long text and the default model's limit is 2048, so the original NER argument for the number no longer applies. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../chunking_strategies/sentence_token_cap.py | 30 +++++++++++++++++-- 1 file changed, 27 insertions(+), 3 deletions(-) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/sentence_token_cap.py b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/sentence_token_cap.py index c3721d2c..63a76be6 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/sentence_token_cap.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/sentence_token_cap.py @@ -27,19 +27,43 @@ class SentenceTokenCapChunking(ChunkingStrategy): sentences for context continuity. Args: - max_tokens: Token cap per chunk. Default 512. + max_tokens: Token cap per chunk. Default 384. + + 384 was originally chosen to match GLiNER's ``config.max_len``, + because the entity extractor called ``predict_entities()`` on whole + chunks and silently dropped anything past that limit. **Both of + those facts have since changed**: the extractor now windows long + text, and the default model's limit is 2048. The NER argument for + 384 no longer applies. + + 384 is still the default because it independently won on a second, + larger effect. The relationship extractor returns a roughly constant + number of relationships per call regardless of how much text it is + given — doubling the input grew the reply by 1.3% — so smaller + chunks mean more calls and more extracted facts. Measured on an + 11-document benchmark with the current default model, chunk 384 vs + 768: entity F1 0.574 vs 0.563, relation F1 0.237 vs 0.223. With the + previous model, quality fell monotonically as chunks grew + (0.233 / 0.221 / 0.204 at 384 / 768 / 1536). + + Note this cuts against most RAG guidance, which suggests 1024+. + Those defaults come from pipelines that use an LLM for entity + extraction and do not share this per-call ceiling. + + Raising the cap is therefore a way to trade recall for lower cost, + not a quality improvement. overlap_sentences: Sentences shared between consecutive chunks. Default 2. encoding_name: tiktoken encoding to use. Default ``cl100k_base`` (GPT-4/3.5). Example:: - chunker = SentenceTokenCapChunking(max_tokens=512, overlap_sentences=2) + chunker = SentenceTokenCapChunking(max_tokens=384, overlap_sentences=2) result = await chunker.chunk(text, ctx) """ def __init__( self, - max_tokens: int = 512, + max_tokens: int = 384, overlap_sentences: int = 2, encoding_name: str = "cl100k_base", ) -> None: From d2adfd01b22ecc8fd9b4411db74af073dd3f4009 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:00:56 +0300 Subject: [PATCH 03/10] perf(chunking): carry the 384 default through the remaining chunkers All four chunking strategies defaulted to 512 independently, so changing one left the others mis-tuned and a user swapping chunkers silently changed their extraction quality. StructuralChunking and ContextualChunking now default to 384 and point at SentenceTokenCapChunking for the measurement. The CallableChunking docstring example and the markdown example are updated to match so copy-pasted code does not reintroduce 512. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- graphrag_sdk/examples/06_markdown_document_aware.py | 2 +- .../ingestion/chunking_strategies/callable_chunking.py | 2 +- .../ingestion/chunking_strategies/contextual_chunking.py | 6 +++--- .../ingestion/chunking_strategies/structural_chunking.py | 8 +++++--- 4 files changed, 10 insertions(+), 8 deletions(-) diff --git a/graphrag_sdk/examples/06_markdown_document_aware.py b/graphrag_sdk/examples/06_markdown_document_aware.py index 92508532..33834567 100644 --- a/graphrag_sdk/examples/06_markdown_document_aware.py +++ b/graphrag_sdk/examples/06_markdown_document_aware.py @@ -194,7 +194,7 @@ async def main(): result = await rag.ingest( md_path, loader=MarkdownLoader(), - chunker=StructuralChunking(max_tokens=512), + chunker=StructuralChunking(max_tokens=384), ) print(f"Done: {result.nodes_created} nodes, {result.relationships_created} edges, " f"{result.chunks_indexed} chunks indexed") diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/callable_chunking.py b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/callable_chunking.py index 93873895..15205a38 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/callable_chunking.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/callable_chunking.py @@ -34,7 +34,7 @@ class CallableChunking(ChunkingStrategy): from llama_index.core.node_parser import SentenceSplitter from llama_index.core import Document - splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50) + splitter = SentenceSplitter(chunk_size=384, chunk_overlap=50) chunker = CallableChunking( lambda text: [n.text for n in splitter.get_nodes_from_documents([Document(text=text)])] ) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/contextual_chunking.py b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/contextual_chunking.py index c02b8268..f629d1cf 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/contextual_chunking.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/contextual_chunking.py @@ -54,7 +54,7 @@ class ContextualChunking(ChunkingStrategy): **not** be passed — configure those directly on the base chunker. If omitted, defaults to ``SentenceTokenCapChunking(max_tokens, overlap_sentences, encoding_name)``. - max_tokens: Token cap per chunk for the default base chunker. Default 512. + max_tokens: Token cap per chunk for the default base chunker. Default 384. Ignored (and forbidden) when *base_chunker* is provided. overlap_sentences: Sentence overlap for the default base chunker. Default 2. Ignored (and forbidden) when *base_chunker* is provided. @@ -70,7 +70,7 @@ class ContextualChunking(ChunkingStrategy): chunker = ContextualChunking(llm=my_llm, max_tokens=256, overlap_sentences=1) # Custom base chunker — configure it directly, pass no shorthand kwargs - chunker = ContextualChunking(llm=my_llm, base_chunker=StructuralChunking(max_tokens=512)) + chunker = ContextualChunking(llm=my_llm, base_chunker=StructuralChunking(max_tokens=384)) result = await chunker.chunk_document(doc, ctx) """ @@ -109,7 +109,7 @@ def __init__( # Use encoding_name solely for document-truncation token counting _encoding_name = "cl100k_base" else: - _max_tokens = max_tokens if max_tokens is not ContextualChunking._UNSET else 512 + _max_tokens = max_tokens if max_tokens is not ContextualChunking._UNSET else 384 _overlap = ( overlap_sentences if overlap_sentences is not ContextualChunking._UNSET else 2 ) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py index 9c1852c2..9440deaf 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py @@ -23,7 +23,9 @@ class StructuralChunking(ChunkingStrategy): Args: fallback_chunker: Chunker to handle oversized elements. Default: SentenceTokenCapChunking. - max_tokens: Maximum tokens per structural chunk. Default: 512. + max_tokens: Maximum tokens per structural chunk. Default: 384 + (measured best for relationship extraction; see + SentenceTokenCapChunking for why). encoding_name: tiktoken encoding to use. Default ``cl100k_base``. """ @@ -55,10 +57,10 @@ def __init__( "fallback chunker directly." ) self.fallback_chunker = fallback_chunker - self.max_tokens = 512 + self.max_tokens = 384 self.encoding_name = "cl100k_base" else: - self.max_tokens = max_tokens if max_tokens is not StructuralChunking._UNSET else 512 + self.max_tokens = max_tokens if max_tokens is not StructuralChunking._UNSET else 384 self.encoding_name = ( encoding_name if encoding_name is not StructuralChunking._UNSET else "cl100k_base" ) From ee036dac2ff47abffd8c3bf90af6ff840c91ddc0 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:01:15 +0300 Subject: [PATCH 04/10] docs(api): document the 384 default chunker in GraphRAG The convenience API's docstring still advertised max_tokens=512, which is no longer what it builds. Also states why 384 is the default so callers know that raising it trades recall for cost rather than improving quality. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- graphrag_sdk/src/graphrag_sdk/api/main.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/graphrag_sdk/src/graphrag_sdk/api/main.py b/graphrag_sdk/src/graphrag_sdk/api/main.py index bae09970..5ce1d192 100644 --- a/graphrag_sdk/src/graphrag_sdk/api/main.py +++ b/graphrag_sdk/src/graphrag_sdk/api/main.py @@ -1369,10 +1369,13 @@ async def ingest( Uses sensible defaults for any unspecified strategy: - Loader: auto-detected from file extension (PDF or text) - - Chunker: SentenceTokenCapChunking(max_tokens=512, overlap_sentences=2) + - Chunker: SentenceTokenCapChunking(max_tokens=384, overlap_sentences=2) — sentence-aware, never splits entity names at chunk boundaries. - Override with ``chunker=FixedSizeChunking(...)`` if you need - character-window chunking. + 384 is the measured best size for relationship extraction, which + returns a roughly fixed number of facts per call — smaller chunks + mean more calls and more facts. Override with + ``chunker=FixedSizeChunking(...)`` + if you need character-window chunking. - Extractor: GraphExtraction with configured LLM - Resolver: ExactMatchResolution From 7e3f02f5a73d34e1171e40317c16b3e5bff8da02 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:25:13 +0300 Subject: [PATCH 05/10] style: apply ruff format to pipeline.py CI runs 'ruff format --check src/' with line-length 100. The hand-wrapped signature and hashlib call in _assign_deterministic_chunk_uids both fit on one line at that width, so the formatter joins them. No behaviour change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index 94250491..3d51da63 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -38,9 +38,7 @@ _PATTERN_MISMATCH_SAMPLE_SIZE = 3 -def _assign_deterministic_chunk_uids( - doc_info: DocumentInfo, chunks: TextChunks -) -> None: +def _assign_deterministic_chunk_uids(doc_info: DocumentInfo, chunks: TextChunks) -> None: """Replace random chunk UIDs with a content-derived, stable identity. ``TextChunk.uid`` defaults to ``uuid4()``, so every ingest of the *same* @@ -70,9 +68,7 @@ def _assign_deterministic_chunk_uids( """ doc_key = doc_info.uid or doc_info.path or "" for chunk in chunks.chunks: - digest = hashlib.sha256( - f"{doc_key}\x00{chunk.index}\x00{chunk.text}".encode() - ).hexdigest() + digest = hashlib.sha256(f"{doc_key}\x00{chunk.index}\x00{chunk.text}".encode()).hexdigest() chunk.uid = f"chunk-{digest[:32]}" From e44a60746dd49c79c6c91f4fcae78acbc4367e96 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Mon, 7 Sep 2026 10:17:37 +0300 Subject: [PATCH 06/10] fix(ingestion): skip re-ingest of a document whose content is unchanged IngestionPipeline.run now short-circuits when the Document node already exists with an identical content hash: no chunking, extraction, resolution or index calls, and IngestionResult.metadata reports skipped_unchanged. Measured end to end (11-doc corpus, re-ingest of one document): 33 provider calls / 11 duplicate Chunk nodes / +101 MENTIONED_IN edges -> 0 / 0 / 0. --- .../src/graphrag_sdk/ingestion/pipeline.py | 24 +++++++ graphrag_sdk/tests/test_pipeline.py | 71 +++++++++++++++++++ 2 files changed, 95 insertions(+) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index 3d51da63..35632864 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -229,6 +229,30 @@ async def run( # short-circuit in update() will never fire — correct, just # not optimal. content_hash = hashlib.sha256(document.text.encode("utf-8")).hexdigest() + + # Unchanged re-ingest short-circuit. Stable chunk UIDs made the + # lexical layer idempotent, but extraction still re-ran and, being + # LLM work, named a few entities differently each time: measured + # +3 to +16 entity nodes, +18 to +50 RELATES and +30 MENTIONED_IN + # per re-ingest of one unchanged document. Same rule update() + # uses for its no-op: a stored Document with this exact content + # hash means every chunk, entity and edge below is already there. + # Changed text (new hash) still takes the full path; a changed + # ontology or strategy with the same text is update()'s job. + doc_uid = document.document_info.uid + if doc_uid: + existing = await self.graph_store.get_document_record(doc_uid) + if existing is not None and existing.content_hash == content_hash: + ctx.log( + f"Document '{doc_uid}' already ingested with identical content; " + f"skipping extraction. Call update() to re-extract." + ) + return IngestionResult( + document_info=document.document_info, + chunks_indexed=len(chunks.chunks), + metadata={"skipped_unchanged": True, "content_hash": content_hash}, + ) + ctx.log("Step 3/9: Building lexical graph (provenance chain)") await self._build_lexical_graph( document.document_info, chunks, ctx, content_hash=content_hash diff --git a/graphrag_sdk/tests/test_pipeline.py b/graphrag_sdk/tests/test_pipeline.py index 5d97dae9..4657c325 100644 --- a/graphrag_sdk/tests/test_pipeline.py +++ b/graphrag_sdk/tests/test_pipeline.py @@ -406,6 +406,77 @@ async def extract(self, chunks, ontology, ctx): assert mention_rels[0].end_node_id == "chunk-0" +class TestUnchangedReingestShortCircuit: + """Re-ingesting a document whose content hash is already stored is a no-op. + + Stable chunk UIDs (Bug #12) made the chunk layer idempotent, but + extraction still re-ran and, being LLM work, named a few entities + differently each time: measured +3 to +16 entity nodes, +18 to +50 + RELATES and +30 MENTIONED_IN per re-ingest of one unchanged document. + """ + + def _pipeline(self, mock_graph_store, mock_vector_store, extractor): + return IngestionPipeline( + loader=StubLoader("Alice works at Acme Corp."), + chunker=StubChunker(), + extractor=extractor, + resolver=StubResolver(), + graph_store=mock_graph_store, + vector_store=mock_vector_store, + ontology=Ontology(), + ) + + @staticmethod + def _stored(content_hash): + from graphrag_sdk.core.models import DocumentRecord + + return AsyncMock(return_value=DocumentRecord(path="test.txt", content_hash=content_hash)) + + async def test_identical_content_skips_extraction_and_writes( + self, ctx, mock_graph_store, mock_vector_store + ): + import hashlib + + extractor = StubExtractor() + extractor.extract = AsyncMock(wraps=extractor.extract) + digest = hashlib.sha256(b"Alice works at Acme Corp.").hexdigest() + mock_graph_store.get_document_record = self._stored(digest) + + pipeline = self._pipeline(mock_graph_store, mock_vector_store, extractor) + result = await pipeline.run("test.txt", ctx, document_info=DocumentInfo(uid="doc-1")) + + extractor.extract.assert_not_called() + mock_graph_store.upsert_nodes.assert_not_called() + mock_graph_store.upsert_relationships.assert_not_called() + assert result.metadata["skipped_unchanged"] is True + assert result.nodes_created == 0 + + async def test_changed_content_takes_the_full_path( + self, ctx, mock_graph_store, mock_vector_store + ): + extractor = StubExtractor() + extractor.extract = AsyncMock(wraps=extractor.extract) + mock_graph_store.get_document_record = self._stored("0" * 64) + + pipeline = self._pipeline(mock_graph_store, mock_vector_store, extractor) + result = await pipeline.run("test.txt", ctx, document_info=DocumentInfo(uid="doc-1")) + + extractor.extract.assert_called_once() + assert "skipped_unchanged" not in result.metadata + + async def test_new_document_takes_the_full_path( + self, ctx, mock_graph_store, mock_vector_store + ): + extractor = StubExtractor() + extractor.extract = AsyncMock(wraps=extractor.extract) + mock_graph_store.get_document_record = AsyncMock(return_value=None) + + pipeline = self._pipeline(mock_graph_store, mock_vector_store, extractor) + await pipeline.run("test.txt", ctx, document_info=DocumentInfo(uid="doc-1")) + + extractor.extract.assert_called_once() + + class TestRemapMentionsUnit: """Direct unit tests for ``IngestionPipeline._remap_mentions`` covering the chain-following contract independently of the full pipeline.""" From d38cf36c4de3799b247ad982d23d488ea0487566 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:01:58 +0300 Subject: [PATCH 07/10] =?UTF-8?q?fix(ingestion):=20address=20#309=20review?= =?UTF-8?q?=20=E2=80=94=20pipeline-level=20stable=20ids,=20skip=20before?= =?UTF-8?q?=20chunking,=20hash=20after=20success?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review points from galshubeli, Copilot and CodeRabbit on #309, each verified against the code before changing it: * Deterministic chunk ids and the unchanged-document skip only worked through GraphRAG.ingest(): loaders leave DocumentInfo.uid at its uuid4() default, so a direct IngestionPipeline.run(source) minted a new Document (and new chunk ids) every time. The pipeline now derives the id itself when the caller passes no document_info — os.path.normpath(source) in file mode, a text hash in text mode — the same rule GraphRAG._resolve_document_id applies. update() always passes an explicit pending id and is unaffected. * The unchanged-document check ran after chunk_document(); a 'no-op' still chunked (a provider call per chunk for contextual/semantic chunkers) and reported chunks_indexed=len(chunks). It now runs right after load, before step 2, and reports chunks_indexed=0. * content_hash was written with the Document node in step 3, before extraction; a failure in steps 4-9 left a Document that the next ingest skipped as complete. The hash is now written by a final upsert only after steps 8/9 finish, so a partial run is retried in full. * ContextualChunking prepends an LLM summary; the chunk id hashed that enriched text, so a reworded summary re-minted the id. The id now hashes metadata['original_chunk'] when present. * structural_chunking.py passed uid=sc.uid for fallback sub-chunks; the pipeline overwrites it unconditionally. Removed. * docs/strategies.mdx and docs/benchmark.mdx still said 512. Updated. * CHANGELOG: both fixes, the 384 default with its measured +35 % LLM-call / +19 % input-token cost, the one-time extra chunk copy on first re-ingest after upgrade, and the edited-file re-id behaviour. Tests: end-to-end re-ingest through pipeline.run() with no document_info (same chunk ids twice; different files with the same text do not collide; text mode stable), chunker not called and chunks_indexed == 0 on skip, skip fires with a path-derived id, hash absent after a failed extraction, contextual original_chunk hashing. Two existing tests updated for the Document-then-hash write order. Full suite: 1112 passed, 41 skipped. --- CHANGELOG.md | 33 ++++ docs/benchmark.mdx | 2 +- docs/strategies.mdx | 6 +- .../structural_chunking.py | 3 +- .../src/graphrag_sdk/ingestion/pipeline.py | 132 +++++++++---- graphrag_sdk/tests/test_ingestion_e2e.py | 6 +- graphrag_sdk/tests/test_pipeline.py | 182 +++++++++++++++++- 7 files changed, 310 insertions(+), 54 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a5b14920..841d5f45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Re-ingesting the same file no longer duplicates its chunks. Chunk ids are + now derived from document id + position + text instead of a fresh + `uuid4()`, so `MERGE` finds the existing node (17 → 17 → 17 chunks across + three ingests, previously 17 → 34 → 51). `ContextualChunking` hashes the + original chunk text, not the LLM-enriched one. Applies to + `IngestionPipeline.run()` directly as well as through `GraphRAG.ingest()`: + when no `document_info` is supplied the pipeline now derives a stable + Document id from the normalised source path (or, in text mode, from the + text). **Upgrade note:** graphs built before this change hold random chunk + ids; the first re-ingest of an existing document adds one more copy of its + chunk layer (matching nothing), and is stable from the second re-ingest on. + Because position is part of the id, inserting a paragraph into an edited + document re-ids every later chunk; only byte-identical files are a no-op. +- Re-ingesting an unchanged document is now a true no-op. The pipeline hashes + the loaded text and, if the stored Document carries the same + `content_hash`, returns before chunking with + `IngestionResult.metadata["skipped_unchanged"] = True` and + `chunks_indexed = 0` — no chunker, NER, LLM or graph calls (measured: 33 + provider calls and ~30 s → 0 and 0.01 s). The hash is written only after a + run completes, so a partially failed ingest is retried in full rather than + skipped. + +### Changed + +- Default chunk size lowered from 512 to 384 tokens in + `SentenceTokenCapChunking`, `StructuralChunking`, `ContextualChunking` and + the documented `CallableChunking` example. Measured on the benchmark corpus: + entity F1 0.574 vs 0.563 and relation F1 0.237 vs 0.223 against 768; with + the current extraction prompt, exact relation F1 2.2× and answer accuracy + 27 → 32 % for the full stack. **Cost:** ~35 % more extraction LLM calls and + ~19 % more input tokens per ingest (103 → 157 calls on the 11-document + corpus), since the per-call instructions are re-sent once per chunk. Pass + `max_tokens=512` to keep the old size. - Fixed vector-search ordering so chunk, entity, and relationship searches use similarity scores, with higher values indicating closer matches. diff --git a/docs/benchmark.mdx b/docs/benchmark.mdx index eec843cb..40b8cbb7 100644 --- a/docs/benchmark.mdx +++ b/docs/benchmark.mdx @@ -74,7 +74,7 @@ model and the judge. | Generation temperature | 0.7 | Appendix H.2 | | Framework | GraphRAG-SDK 1.3.0 (PyPI) on FalkorDB | — | | Graph layout | one graph per corpus document | — | -| Chunking | `SentenceTokenCapChunking`, max_tokens 512, overlap 2 sentences | SDK default | +| Chunking | `SentenceTokenCapChunking`, max_tokens 384 (was 512 before 1.2), overlap 2 sentences | SDK default | | Retrieval | `MultiPathRetrieval` — chunk_top_k 15, rel_top_k 15, max_entities 30, max_relationships 20, keyword_limit 10 | SDK default | | Embeddings | `text-embedding-3-large` @ 1024 dimensions | Declared below | | Text-to-Cypher | enabled | Declared below | diff --git a/docs/strategies.mdx b/docs/strategies.mdx index 18eeb3e0..80f08873 100644 --- a/docs/strategies.mdx +++ b/docs/strategies.mdx @@ -133,7 +133,7 @@ Splits at sentence boundaries (never mid-sentence) and enforces a hard token cap from graphrag_sdk.ingestion.chunking_strategies.sentence_token_cap import SentenceTokenCapChunking chunker = SentenceTokenCapChunking( - max_tokens=512, # max tokens per chunk (default: 512) + max_tokens=384, # max tokens per chunk (default: 384) overlap_sentences=2, # sentences shared between chunks (default: 2) encoding_name="cl100k_base", # tiktoken encoding (default: cl100k_base) ) @@ -148,7 +148,7 @@ from graphrag_sdk.ingestion.chunking_strategies.contextual_chunking import Conte chunker = ContextualChunking( llm=my_llm, - max_tokens=512, # token cap per chunk (default: 512) + max_tokens=384, # token cap per chunk (default: 384) overlap_sentences=2, # sentence overlap (default: 2) max_document_tokens=16_000, # truncation limit for the doc reference in prompts (default: 16000) ) @@ -177,7 +177,7 @@ Groups content by heading hierarchy into token-bounded chunks. Each chunk stores from graphrag_sdk.ingestion.chunking_strategies.structural_chunking import StructuralChunking chunker = StructuralChunking( - max_tokens=512, # max tokens per chunk (default: 512) + max_tokens=384, # max tokens per chunk (default: 384) overlap_sentences=2, # sentences shared between chunks (default: 2) ) ``` diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py index 9440deaf..9142b64e 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/chunking_strategies/structural_chunking.py @@ -177,11 +177,12 @@ def _flush_buffer() -> None: new_metadata["token_count"] = final_tokens new_metadata["breadcrumbs"] = list(el.breadcrumbs) if el.breadcrumbs else [] + # No ``uid``: the pipeline assigns deterministic chunk + # ids after chunking and overwrites whatever is set here. chunks.append( TextChunk( text=final_text, index=chunk_index, - uid=sc.uid, metadata=new_metadata, ) ) diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index 35632864..ca3fa964 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -9,6 +9,7 @@ import asyncio import hashlib +import os import logging from typing import Any @@ -65,10 +66,23 @@ def _assign_deterministic_chunk_uids(doc_info: DocumentInfo, chunks: TextChunks) Chunkers that deliberately assign their own meaningful UIDs are not special-cased: this runs after chunking and overwrites unconditionally, so identity is decided in exactly one place. + + Chunkers that enrich the text (``ContextualChunking`` prepends an + LLM-written summary) record the untouched source in + ``metadata["original_chunk"]``; that is what is hashed, so a differently + worded summary on the next run does not mint a new id for the same text. + + Because ``index`` is part of the key, inserting a paragraph early in an + edited document re-ids every chunk after it. That is deliberate — it keeps + ``doc:index`` a true position — but it means an *edited* file grows the + lexical layer under plain ``ingest()``; only a byte-identical file is a + no-op (see the unchanged-document short-circuit in ``run``). """ doc_key = doc_info.uid or doc_info.path or "" for chunk in chunks.chunks: - digest = hashlib.sha256(f"{doc_key}\x00{chunk.index}\x00{chunk.text}".encode()).hexdigest() + base = chunk.metadata.get("original_chunk") + text = base if isinstance(base, str) and base else chunk.text + digest = hashlib.sha256(f"{doc_key}\x00{chunk.index}\x00{text}".encode()).hexdigest() chunk.uid = f"chunk-{digest[:32]}" @@ -187,9 +201,34 @@ async def run( document_info=document_info or DocumentInfo(path=source), ) ctx.log("Using provided text (loader skipped)") + if document_info is None: + # Text mode with no caller-supplied identity: derive the + # Document id from the text so the same text ingested + # twice is the same document (and the chunk ids below, + # which include the document id, are stable too). + document.document_info = DocumentInfo( + uid=f"text-{hashlib.sha256(text.encode('utf-8')).hexdigest()[:16]}", + path=source, + metadata=document.document_info.metadata, + ) else: ctx.log("Step 1/9: Loading source") document = await self.loader.load(source, ctx) + # Loaders leave ``uid`` at its ``uuid4()`` default, so + # without a caller-supplied ``document_info`` every run of + # the same file was a new Document — and, since the chunk + # ids include the document id, a new set of chunks. Derive + # the id from the normalised source path, the same rule + # ``GraphRAG._resolve_document_id`` applies, so the + # pipeline is idempotent on its own and not only through + # the facade. ``update()`` always passes an explicit + # pending id and is unaffected. + if document_info is None: + document.document_info = DocumentInfo( + uid=os.path.normpath(source), + path=document.document_info.path or source, + metadata=document.document_info.metadata, + ) # When the caller supplies a ``document_info`` (e.g. for # stable-id ingestion or update()), prefer its uid/path # over whatever the loader produced. The loader-side @@ -207,56 +246,59 @@ async def run( }, ) - # Step 2: Chunk - ctx.log("Step 2/9: Chunking text") - chunks = await self.chunker.chunk_document(document, ctx) - - if not chunks.chunks: - ctx.log("No chunks produced, pipeline complete") - return IngestionResult(document_info=document.document_info) - - _assign_deterministic_chunk_uids(document.document_info, chunks) - - # Step 3: Build lexical graph (MANDATORY — not a strategy). - # Hash the loaded text so ``GraphRAG.update()`` can short-circuit - # when content is unchanged. SHA-256 hex; cost is negligible - # next to extraction. + # Hash the loaded text. Written to the Document node at the END of + # a successful run (step 9b) so ``GraphRAG.update()`` and the + # short-circuit below can recognise unchanged content. SHA-256 + # hex; cost is negligible next to extraction. # # Assumes the loader returns deterministic text for the same # source. Loaders that inject non-deterministic content # (timestamps, randomized ordering, run-id watermarks, etc.) # will produce a different hash on every run and the no-op - # short-circuit in update() will never fire — correct, just - # not optimal. + # short-circuit will never fire — correct, just not optimal. content_hash = hashlib.sha256(document.text.encode("utf-8")).hexdigest() - # Unchanged re-ingest short-circuit. Stable chunk UIDs made the - # lexical layer idempotent, but extraction still re-ran and, being - # LLM work, named a few entities differently each time: measured - # +3 to +16 entity nodes, +18 to +50 RELATES and +30 MENTIONED_IN - # per re-ingest of one unchanged document. Same rule update() - # uses for its no-op: a stored Document with this exact content - # hash means every chunk, entity and edge below is already there. - # Changed text (new hash) still takes the full path; a changed - # ontology or strategy with the same text is update()'s job. + # Unchanged re-ingest short-circuit — before chunking, so a + # no-op costs one graph lookup and nothing else (contextual and + # semantic chunkers make provider calls per chunk). Stable chunk + # UIDs made the lexical layer idempotent, but extraction still + # re-ran and, being LLM work, named a few entities differently + # each time: measured +3 to +16 entity nodes, +18 to +50 RELATES + # and +30 MENTIONED_IN per re-ingest of one unchanged document. + # Same rule update() uses for its no-op: a stored Document whose + # content_hash matches means every chunk, entity and edge is + # already there — the hash is only written once a run has + # completed, so a half-finished ingest is never mistaken for a + # finished one. Changed text (new hash) takes the full path; a + # changed ontology or strategy with the same text is update()'s + # job. doc_uid = document.document_info.uid if doc_uid: existing = await self.graph_store.get_document_record(doc_uid) if existing is not None and existing.content_hash == content_hash: ctx.log( f"Document '{doc_uid}' already ingested with identical content; " - f"skipping extraction. Call update() to re-extract." + f"skipping. Call update() to re-extract." ) return IngestionResult( document_info=document.document_info, - chunks_indexed=len(chunks.chunks), + chunks_indexed=0, metadata={"skipped_unchanged": True, "content_hash": content_hash}, ) + # Step 2: Chunk + ctx.log("Step 2/9: Chunking text") + chunks = await self.chunker.chunk_document(document, ctx) + + if not chunks.chunks: + ctx.log("No chunks produced, pipeline complete") + return IngestionResult(document_info=document.document_info) + + _assign_deterministic_chunk_uids(document.document_info, chunks) + + # Step 3: Build lexical graph (MANDATORY — not a strategy). ctx.log("Step 3/9: Building lexical graph (provenance chain)") - await self._build_lexical_graph( - document.document_info, chunks, ctx, content_hash=content_hash - ) + await self._build_lexical_graph(document.document_info, chunks, ctx) # Step 4: Extract entities & relationships ctx.log("Step 4/9: Extracting entities & relationships") @@ -331,6 +373,13 @@ async def _step_index_chunks() -> None: _step_index_chunks(), ) + # Step 9b: only now record the content hash. Writing it in step 3 + # meant a failure in extraction or the graph write left a + # Document that looked complete, and the next ingest skipped it + # (Copilot review on #309). A partial run therefore retries in + # full; only a finished run is recognised as unchanged. + await self._mark_content_hash(document.document_info.uid, content_hash) + total_rels = len(resolved.relationships) + mentions_written result = IngestionResult( document_info=document.document_info, @@ -359,13 +408,22 @@ async def _step_index_chunks() -> None: logger.debug("Pipeline failure details", exc_info=True) raise IngestionError(f"Pipeline failed: {exc}") from exc + async def _mark_content_hash(self, doc_uid: str, content_hash: str) -> None: + """Record ``content_hash`` on an existing Document node. + + Called as the last step of a successful ``run``. ``upsert_nodes`` + merges on id, so this only adds the property; the Document node and + its ``path`` / metadata were written by ``_build_lexical_graph``. + """ + await self.graph_store.upsert_nodes( + [GraphNode(id=doc_uid, label="Document", properties={"content_hash": content_hash})] + ) + async def _build_lexical_graph( self, doc_info: DocumentInfo, chunks: TextChunks, ctx: Context, - *, - content_hash: str | None = None, ) -> None: """Build the mandatory provenance chain. @@ -378,17 +436,15 @@ async def _build_lexical_graph( This is NON-OPTIONAL. The Zero-Loss Data principle requires that every piece of source material is traceable in the graph. - ``content_hash`` is the SHA-256 of the loaded source text. When - present it is written to the Document node so ``GraphRAG.update()`` - can short-circuit no-op updates without re-running extraction. + The Document's ``content_hash`` is *not* written here; ``run`` sets + it via :meth:`_mark_content_hash` once the whole pipeline has + succeeded. """ # Document node doc_props: dict[str, Any] = { "path": doc_info.path or "", **doc_info.metadata, } - if content_hash is not None: - doc_props["content_hash"] = content_hash doc_node = GraphNode( id=doc_info.uid, label="Document", diff --git a/graphrag_sdk/tests/test_ingestion_e2e.py b/graphrag_sdk/tests/test_ingestion_e2e.py index a36374da..110c0bad 100644 --- a/graphrag_sdk/tests/test_ingestion_e2e.py +++ b/graphrag_sdk/tests/test_ingestion_e2e.py @@ -11,6 +11,8 @@ """ from __future__ import annotations +import os + import pytest _MARKDOWN = """\ @@ -249,8 +251,10 @@ async def resolve(self, graph_data, ctx): for n in call[0][0] ] doc_nodes = [n for n in all_nodes if n.label == "Document"] - assert len(doc_nodes) == 1 + # step-3 Document write (path) + the end-of-run content_hash write + assert len(doc_nodes) == 2 assert doc_nodes[0].properties.get("path") == str(md_file) + assert {n.id for n in doc_nodes} == {os.path.normpath(str(md_file))} async def test_header_markup_stripped_from_chunk_text( self, ctx, tmp_path, mock_graph_store, mock_vector_store diff --git a/graphrag_sdk/tests/test_pipeline.py b/graphrag_sdk/tests/test_pipeline.py index 4657c325..5748ea70 100644 --- a/graphrag_sdk/tests/test_pipeline.py +++ b/graphrag_sdk/tests/test_pipeline.py @@ -227,14 +227,18 @@ async def test_pipeline_writes_content_hash(self, ctx, mock_graph_store, mock_ve await pipeline.run("test.txt", ctx) - # Find the Document node among all upsert_nodes calls. - doc_nodes: list[GraphNode] = [] - for call in mock_graph_store.upsert_nodes.call_args_list: - for n in call[0][0]: - if n.label == "Document": - doc_nodes.append(n) - assert len(doc_nodes) == 1, "expected exactly one Document upsert" - assert doc_nodes[0].properties.get("content_hash") == expected_hash + # The Document node is written in step 3 (path + metadata) and the + # content_hash is added by a second, final upsert once the run has + # completed — so a half-finished ingest never carries a hash. + calls = mock_graph_store.upsert_nodes.call_args_list + doc_calls = [ + (i, n) for i, call in enumerate(calls) for n in call[0][0] if n.label == "Document" + ] + assert len(doc_calls) == 2, "expected the Document upsert, then the hash upsert" + (i_first, first), (i_last, last) = doc_calls + assert "content_hash" not in first.properties + assert last.properties == {"content_hash": expected_hash} + assert i_last == len(calls) - 1, "hash must be the last node write of the run" async def test_pipeline_uses_provided_document_info_uid( self, ctx, mock_graph_store, mock_vector_store @@ -253,8 +257,9 @@ async def test_pipeline_uses_provided_document_info_uid( for n in call[0][0]: if n.label == "Document": doc_nodes.append(n) - assert len(doc_nodes) == 1 - assert doc_nodes[0].id == "my-stable-id" + # step-3 Document write + the end-of-run content_hash write + assert len(doc_nodes) == 2 + assert {n.id for n in doc_nodes} == {"my-stable-id"} assert doc_nodes[0].properties.get("path") == "docs/a.md" async def test_pipeline_remaps_mentions_through_resolver_remap( @@ -451,6 +456,72 @@ async def test_identical_content_skips_extraction_and_writes( assert result.metadata["skipped_unchanged"] is True assert result.nodes_created == 0 + async def test_unchanged_document_never_reaches_the_chunker( + self, ctx, mock_graph_store, mock_vector_store + ): + """The check runs before step 2: contextual / semantic chunkers make a + provider call per chunk, so a no-op must not chunk. ``chunks_indexed`` + reports the work actually done (none).""" + import hashlib + + chunker = StubChunker() + chunker.chunk_document = AsyncMock(wraps=chunker.chunk_document) + digest = hashlib.sha256(b"Alice works at Acme Corp.").hexdigest() + mock_graph_store.get_document_record = self._stored(digest) + pipeline = self._pipeline(mock_graph_store, mock_vector_store, StubExtractor()) + pipeline.chunker = chunker + + result = await pipeline.run("test.txt", ctx, document_info=DocumentInfo(uid="doc-1")) + + chunker.chunk_document.assert_not_called() + mock_vector_store.index_chunks.assert_not_called() + assert result.metadata["skipped_unchanged"] is True + assert result.chunks_indexed == 0 + + async def test_skip_fires_without_caller_supplied_document_info( + self, ctx, mock_graph_store, mock_vector_store + ): + """Direct ``IngestionPipeline.run(source)`` — no facade, no + ``document_info`` — must still recognise an unchanged document: the + pipeline derives the Document id from the source path itself.""" + import hashlib + + extractor = StubExtractor() + extractor.extract = AsyncMock(wraps=extractor.extract) + digest = hashlib.sha256(b"Alice works at Acme Corp.").hexdigest() + mock_graph_store.get_document_record = self._stored(digest) + + pipeline = self._pipeline(mock_graph_store, mock_vector_store, extractor) + result = await pipeline.run("./docs/../test.txt", ctx) + + mock_graph_store.get_document_record.assert_awaited_once_with("test.txt") + extractor.extract.assert_not_called() + assert result.metadata["skipped_unchanged"] is True + + async def test_content_hash_is_written_only_after_a_successful_run( + self, ctx, mock_graph_store, mock_vector_store + ): + """A failed extraction must not leave a Document that looks complete, + or the next ingest would skip it forever.""" + extractor = StubExtractor() + extractor.extract = AsyncMock(side_effect=RuntimeError("provider down")) + mock_graph_store.get_document_record = AsyncMock(return_value=None) + pipeline = self._pipeline(mock_graph_store, mock_vector_store, extractor) + + from graphrag_sdk.core.exceptions import IngestionError + + with pytest.raises(IngestionError): + await pipeline.run("test.txt", ctx, document_info=DocumentInfo(uid="doc-1")) + + written = [ + n + for call in mock_graph_store.upsert_nodes.call_args_list + for n in call[0][0] + if n.label == "Document" + ] + assert written, "the Document node itself is still written in step 3" + assert all("content_hash" not in n.properties for n in written) + async def test_changed_content_takes_the_full_path( self, ctx, mock_graph_store, mock_vector_store ): @@ -769,6 +840,74 @@ def test_prune_pattern_mismatch_sample_is_bounded(self, caplog): assert msg.count("('Company', 'Person')") <= 3 +class TestReingestIdempotencyEndToEnd: + """The behaviour the PR exists for, exercised through ``pipeline.run`` with + the defaults a direct caller gets — no ``document_info``, loader-provided + ``DocumentInfo`` with only ``path`` set.""" + + def _pipeline(self, mock_graph_store, mock_vector_store, text): + return IngestionPipeline( + loader=StubLoader(text), + chunker=StubChunker(), + extractor=StubExtractor(), + resolver=StubResolver(), + graph_store=mock_graph_store, + vector_store=mock_vector_store, + ontology=Ontology(), + ) + + @staticmethod + def _chunk_ids(mock_graph_store): + return sorted( + n.id + for call in mock_graph_store.upsert_nodes.call_args_list + for n in call[0][0] + if n.label == "Chunk" + ) + + async def test_two_runs_of_the_same_file_write_the_same_chunk_ids( + self, ctx, mock_graph_store, mock_vector_store + ): + mock_graph_store.get_document_record = AsyncMock(return_value=None) + text = "Alice works at Acme. Bob works at Beta. Carol runs Gamma." + first = self._pipeline(mock_graph_store, mock_vector_store, text) + await first.run("report.txt", ctx) + ids_a = self._chunk_ids(mock_graph_store) + doc_a = {n.id for c in mock_graph_store.upsert_nodes.call_args_list for n in c[0][0] if n.label == "Document"} + + mock_graph_store.upsert_nodes.reset_mock() + second = self._pipeline(mock_graph_store, mock_vector_store, text) + await second.run("report.txt", ctx) + ids_b = self._chunk_ids(mock_graph_store) + doc_b = {n.id for c in mock_graph_store.upsert_nodes.call_args_list for n in c[0][0] if n.label == "Document"} + + assert ids_a and ids_a == ids_b + assert doc_a == doc_b == {"report.txt"} + + async def test_different_files_with_the_same_text_get_different_chunk_ids( + self, ctx, mock_graph_store, mock_vector_store + ): + mock_graph_store.get_document_record = AsyncMock(return_value=None) + text = "Alice works at Acme. Bob works at Beta." + await self._pipeline(mock_graph_store, mock_vector_store, text).run("a.txt", ctx) + ids_a = self._chunk_ids(mock_graph_store) + mock_graph_store.upsert_nodes.reset_mock() + await self._pipeline(mock_graph_store, mock_vector_store, text).run("b.txt", ctx) + ids_b = self._chunk_ids(mock_graph_store) + assert ids_a and not set(ids_a) & set(ids_b) + + async def test_text_mode_without_document_info_is_stable_too( + self, ctx, mock_graph_store, mock_vector_store + ): + mock_graph_store.get_document_record = AsyncMock(return_value=None) + p = self._pipeline(mock_graph_store, mock_vector_store, "unused") + await p.run("ignored", ctx, text="Alice works at Acme. Bob works at Beta.") + ids_a = self._chunk_ids(mock_graph_store) + mock_graph_store.upsert_nodes.reset_mock() + await p.run("ignored", ctx, text="Alice works at Acme. Bob works at Beta.") + assert ids_a and ids_a == self._chunk_ids(mock_graph_store) + + class TestDeterministicChunkUids: """Bug #12 — re-ingesting an unchanged document duplicated its chunks. @@ -815,3 +954,26 @@ def test_different_document_yields_different_uid(self): def test_uids_are_unique_within_a_document(self): uids = self._chunks([f"chunk {i}" for i in range(50)]) assert len(set(uids)) == 50 + + def test_contextual_chunks_hash_the_original_text(self): + """``ContextualChunking`` prepends an LLM summary; the id must come from + ``metadata["original_chunk"]`` so a reworded summary keeps the id.""" + from graphrag_sdk.ingestion.pipeline import _assign_deterministic_chunk_uids + + def enriched(summary): + chunks = TextChunks( + chunks=[ + TextChunk( + text=f"{summary}\n\nThe lighthouse was built in 1896.", + index=0, + metadata={"original_chunk": "The lighthouse was built in 1896."}, + ) + ] + ) + _assign_deterministic_chunk_uids(DocumentInfo(uid="doc-1"), chunks) + return chunks.chunks[0].uid + + assert enriched("Context: about a lighthouse.") == enriched("Summary: a lighthouse's history.") + plain = TextChunks(chunks=[TextChunk(text="The lighthouse was built in 1896.", index=0)]) + _assign_deterministic_chunk_uids(DocumentInfo(uid="doc-1"), plain) + assert plain.chunks[0].uid == enriched("anything") From 5ae8cdd39f9b4e80e93ed7d2095309ffb788a365 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:18:40 +0300 Subject: [PATCH 08/10] =?UTF-8?q?fix(ingestion):=20follow-up=20review=20on?= =?UTF-8?q?=20#309=20=E2=80=94=20stable=20text-mode=20ids=20in=20the=20fac?= =?UTF-8?q?ade,=20URI-safe=20ids,=20import=20order?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * GraphRAG.ingest(text=...) without document_id used a fresh text- per call, contradicting the pipeline's new text-hash id and the CHANGELOG claim. _resolve_document_id now derives text-<16hex> from a SHA-256 of the text, so identical text is one document and a no-op the second time; pass document_id to store identical text as distinct documents. * os.path.normpath() rewrites URIs (https:// -> https:/, and resolves '..' inside a query string), merging distinct URLs into one Document id. New core.models.stable_document_id(source) normalises filesystem paths only and returns anything with a URI scheme verbatim; used by both IngestionPipeline.run and GraphRAG._resolve_document_id (the facade had the same defect on main). * Skip log line no longer says 'Call update()' — the pipeline has none; it names GraphRAG.update(). * CHANGELOG: 103 -> 157 extraction calls is +52 % (Gal's +35 % was a 53k-token corpus; both figures now given); text-mode wording made precise. docs/benchmark.mdx: 'before this release', not 'before 1.2'. * Import order in pipeline.py (ruff I001, the CI lint failure). Tests: text-mode id stable/distinct/explicit-wins; stable_document_id on paths and four URI forms; pipeline uses a URI verbatim as the Document id. Full suite 1116 passed, 41 skipped; ruff check + format clean. --- CHANGELOG.md | 16 ++++--- docs/benchmark.mdx | 2 +- graphrag_sdk/src/graphrag_sdk/api/main.py | 23 ++++++---- graphrag_sdk/src/graphrag_sdk/core/models.py | 19 +++++++++ .../src/graphrag_sdk/ingestion/pipeline.py | 17 ++++---- graphrag_sdk/tests/test_facade.py | 14 +++++++ graphrag_sdk/tests/test_pipeline.py | 42 +++++++++++++++++++ 7 files changed, 110 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 841d5f45..0519ea7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,8 +16,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 original chunk text, not the LLM-enriched one. Applies to `IngestionPipeline.run()` directly as well as through `GraphRAG.ingest()`: when no `document_info` is supplied the pipeline now derives a stable - Document id from the normalised source path (or, in text mode, from the - text). **Upgrade note:** graphs built before this change hold random chunk + Document id from the source (normalised filesystem path; URIs kept + verbatim), and in text mode from a hash + of the text. `GraphRAG.ingest(text=...)` without `document_id` uses the + same text hash (previously a fresh `text-` per call), so ingesting + identical text twice is now one document and a no-op the second time; pass + `document_id` to store identical text as distinct documents. + **Upgrade note:** graphs built before this change hold random chunk ids; the first re-ingest of an existing document adds one more copy of its chunk layer (matching nothing), and is stable from the second re-ingest on. Because position is part of the id, inserting a paragraph into an edited @@ -38,9 +43,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the documented `CallableChunking` example. Measured on the benchmark corpus: entity F1 0.574 vs 0.563 and relation F1 0.237 vs 0.223 against 768; with the current extraction prompt, exact relation F1 2.2× and answer accuracy - 27 → 32 % for the full stack. **Cost:** ~35 % more extraction LLM calls and - ~19 % more input tokens per ingest (103 → 157 calls on the 11-document - corpus), since the per-call instructions are re-sent once per chunk. Pass + 27 → 32 % for the full stack. **Cost:** more extraction LLM calls per + ingest — 103 → 157 (+52 %) on the 11-document benchmark corpus, +35 % on a + 53k-token corpus — and ~19 % more input tokens, since the per-call + instructions are re-sent once per chunk. Pass `max_tokens=512` to keep the old size. - Fixed vector-search ordering so chunk, entity, and relationship searches use similarity scores, with higher values indicating closer matches. diff --git a/docs/benchmark.mdx b/docs/benchmark.mdx index 40b8cbb7..6a80f033 100644 --- a/docs/benchmark.mdx +++ b/docs/benchmark.mdx @@ -74,7 +74,7 @@ model and the judge. | Generation temperature | 0.7 | Appendix H.2 | | Framework | GraphRAG-SDK 1.3.0 (PyPI) on FalkorDB | — | | Graph layout | one graph per corpus document | — | -| Chunking | `SentenceTokenCapChunking`, max_tokens 384 (was 512 before 1.2), overlap 2 sentences | SDK default | +| Chunking | `SentenceTokenCapChunking`, max_tokens 384 (was 512 before this release), overlap 2 sentences | SDK default | | Retrieval | `MultiPathRetrieval` — chunk_top_k 15, rel_top_k 15, max_entities 30, max_relationships 20, keyword_limit 10 | SDK default | | Embeddings | `text-embedding-3-large` @ 1024 dimensions | Declared below | | Text-to-Cypher | enabled | Declared below | diff --git a/graphrag_sdk/src/graphrag_sdk/api/main.py b/graphrag_sdk/src/graphrag_sdk/api/main.py index 5ce1d192..df6345db 100644 --- a/graphrag_sdk/src/graphrag_sdk/api/main.py +++ b/graphrag_sdk/src/graphrag_sdk/api/main.py @@ -38,6 +38,7 @@ RagResult, RetrieverResult, UpdateResult, + stable_document_id, ) from graphrag_sdk.core.providers import Embedder, LLMInterface from graphrag_sdk.discovery import SchemaExtensionProposal, suggest_extensions @@ -1419,7 +1420,7 @@ async def ingest( if isinstance(source, list) and document_id is not None: raise ValueError( "'document_id' cannot be set on batch ingest (list source). " - "Each file's id defaults to os.path.normpath(path); pass an " + "Each file's id defaults to its normalised path; pass an " "explicit document_id only for single-source calls." ) if text is not None and loader is not None: @@ -1496,21 +1497,25 @@ def _resolve_document_id( """Compute the stable Document node id for an ingest/update call. - explicit ``document_id`` → used verbatim - - file mode (source given, no id) → ``os.path.normpath(source)`` - - text mode (no id) → generated ``text-<8hex>`` + - file mode (source given, no id) → ``stable_document_id(source)``: + the normalised path, or the URI verbatim + - text mode (no id) → ``text-<16hex>`` derived from the text Path normalization collapses ``./``, ``../``, and double slashes so the same logical path always yields the same id, regardless of - how the caller spelled it. + how the caller spelled it; sources with a URI scheme are not + touched, since ``normpath`` would rewrite them. Text mode hashes the text (SHA-256, 64 + bits) so ingesting the same text twice is the same document — and + therefore a no-op on the second call, like a file — instead of a + fresh random ``text-`` per call. Two different texts collide + with probability ~2 in 10^11 at 10K ingests; pass ``document_id`` + when you need to ingest identical text as distinct documents. """ if document_id is not None: return document_id if text is None and source is not None: - return os.path.normpath(source) - # 64-bit suffix — at 32 bits (the original [:8]), 10K text-mode - # ingests in one session collide with ~12% probability. 64 bits - # pushes that to roughly 2 in 10^11 for the same volume. - return f"text-{uuid4().hex[:16]}" + return stable_document_id(source) + return f"text-{hashlib.sha256((text or '').encode('utf-8')).hexdigest()[:16]}" async def _ingest_single( self, diff --git a/graphrag_sdk/src/graphrag_sdk/core/models.py b/graphrag_sdk/src/graphrag_sdk/core/models.py index 382fd760..f2ac1328 100644 --- a/graphrag_sdk/src/graphrag_sdk/core/models.py +++ b/graphrag_sdk/src/graphrag_sdk/core/models.py @@ -5,6 +5,8 @@ from __future__ import annotations import logging +import os +import re from enum import Enum from typing import Any, Generic, Literal, TypeVar from uuid import uuid4 @@ -93,6 +95,23 @@ class TextChunks(DataModel): chunks: list[TextChunk] = Field(default_factory=list) +_URI_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://") + + +def stable_document_id(source: str) -> str: + """Document node id for a loader ``source`` when the caller gives none. + + Filesystem paths are normalised (``./``, ``../``, doubled slashes) so the + same file spelled two ways is one document. Anything with a URI scheme + (``https://``, ``s3://`` …) is returned verbatim: ``os.path.normpath`` would + collapse ``https://`` to ``https:/`` and resolve ``..`` inside the query + string, merging distinct URLs into one id. + """ + if _URI_SCHEME.match(source): + return source + return os.path.normpath(source) + + class DocumentInfo(DataModel): """Metadata about the source document.""" diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index ca3fa964..a3a94f5b 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -9,7 +9,6 @@ import asyncio import hashlib -import os import logging from typing import Any @@ -25,6 +24,7 @@ IngestionResult, Ontology, TextChunks, + stable_document_id, ) from graphrag_sdk.ingestion.chunking_strategies.base import ChunkingStrategy from graphrag_sdk.ingestion.extraction_strategies.base import ExtractionStrategy @@ -218,14 +218,14 @@ async def run( # without a caller-supplied ``document_info`` every run of # the same file was a new Document — and, since the chunk # ids include the document id, a new set of chunks. Derive - # the id from the normalised source path, the same rule - # ``GraphRAG._resolve_document_id`` applies, so the - # pipeline is idempotent on its own and not only through - # the facade. ``update()`` always passes an explicit - # pending id and is unaffected. + # the id from the source (normalised path; URIs verbatim), + # the same rule ``GraphRAG._resolve_document_id`` applies, + # so the pipeline is idempotent on its own and not only + # through the facade. ``update()`` always passes an + # explicit pending id and is unaffected. if document_info is None: document.document_info = DocumentInfo( - uid=os.path.normpath(source), + uid=stable_document_id(source), path=document.document_info.path or source, metadata=document.document_info.metadata, ) @@ -278,7 +278,8 @@ async def run( if existing is not None and existing.content_hash == content_hash: ctx.log( f"Document '{doc_uid}' already ingested with identical content; " - f"skipping. Call update() to re-extract." + f"skipping. Use GraphRAG.update() to re-extract with a new " + f"ontology or strategy." ) return IngestionResult( document_info=document.document_info, diff --git a/graphrag_sdk/tests/test_facade.py b/graphrag_sdk/tests/test_facade.py index fc0060eb..7e7d63dd 100644 --- a/graphrag_sdk/tests/test_facade.py +++ b/graphrag_sdk/tests/test_facade.py @@ -1069,6 +1069,20 @@ async def test_ingest_text_auto_generates_document_id(self, graphrag): result = await graphrag.ingest(text="some text") assert result is not None + def test_text_mode_document_id_is_derived_from_the_text(self): + """Same text → same id (so a second ingest is a no-op); different + text → different id; an explicit id always wins.""" + from graphrag_sdk import GraphRAG + + a = GraphRAG._resolve_document_id(None, "Alice works at Acme.", None) + b = GraphRAG._resolve_document_id(None, "Alice works at Acme.", None) + c = GraphRAG._resolve_document_id(None, "Bob works at Beta.", None) + assert a == b + assert a != c + assert a.startswith("text-") and len(a) == len("text-") + 16 + assert GraphRAG._resolve_document_id(None, "Alice works at Acme.", "mine") == "mine" + assert GraphRAG._resolve_document_id("./docs/../a.md", None, None) == "a.md" + async def test_ingest_single_still_works(self, graphrag, tmp_path): f = tmp_path / "single.txt" f.write_text("Single document.") diff --git a/graphrag_sdk/tests/test_pipeline.py b/graphrag_sdk/tests/test_pipeline.py index 5748ea70..58c04551 100644 --- a/graphrag_sdk/tests/test_pipeline.py +++ b/graphrag_sdk/tests/test_pipeline.py @@ -977,3 +977,45 @@ def enriched(summary): plain = TextChunks(chunks=[TextChunk(text="The lighthouse was built in 1896.", index=0)]) _assign_deterministic_chunk_uids(DocumentInfo(uid="doc-1"), plain) assert plain.chunks[0].uid == enriched("anything") + + +class TestStableDocumentId: + def test_paths_are_normalised(self): + from graphrag_sdk.core.models import stable_document_id + + assert stable_document_id("./docs/../a.md") == "a.md" + assert stable_document_id("docs//a.md") == "docs/a.md" + + def test_uris_are_kept_verbatim(self): + """``os.path.normpath`` would turn ``https://`` into ``https:/`` and + resolve ``..`` inside a query string, merging distinct URLs.""" + from graphrag_sdk.core.models import stable_document_id + + for uri in ( + "https://example.test/doc?path=a/../b", + "https://example.test/b", + "s3://bucket/key//with/./dots", + "file:///tmp/../etc/x", + ): + assert stable_document_id(uri) == uri + assert stable_document_id("https://example.test/doc?path=a/../b") != stable_document_id( + "https://example.test/b" + ) + + async def test_pipeline_uses_the_uri_verbatim_as_document_id( + self, ctx, mock_graph_store, mock_vector_store + ): + mock_graph_store.get_document_record = AsyncMock(return_value=None) + pipeline = IngestionPipeline( + loader=StubLoader("Alice works at Acme."), + chunker=StubChunker(), + extractor=StubExtractor(), + resolver=StubResolver(), + graph_store=mock_graph_store, + vector_store=mock_vector_store, + ontology=Ontology(), + ) + await pipeline.run("https://example.test/doc?path=a/../b", ctx) + mock_graph_store.get_document_record.assert_awaited_once_with( + "https://example.test/doc?path=a/../b" + ) From dedae862cc11c60e4fbb62d527f57971a6934a34 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:23:34 +0300 Subject: [PATCH 09/10] =?UTF-8?q?fix(ingestion):=20address=20review=20on?= =?UTF-8?q?=20#309=20=E2=80=94=20content=20hash=20only=20on=20complete=20w?= =?UTF-8?q?rites,=20update(force=3DTrue),=20identity=20merge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pipeline: content_hash is stamped only when every relationship, mention and chunk-index write reports the count it was given; a shortfall is returned in IngestionResult.metadata["incomplete_writes"] and warned, so the next ingest repairs instead of skipping. get_document_record is looked up with getattr so duck-typed stores still work. A caller-supplied DocumentInfo no longer clobbers the derived id with a random uuid when only path is set. Derived ids containing the __pending__ marker are rejected before any I/O. api: GraphRAG.update(force=True) re-extracts unchanged content; the pipeline's skip message now points at it. ingest()/update() docstrings describe normalised-path ids, text-<16hex> ids and the unchanged no-op. CHANGELOG: explicit breaking-change note for text-mode dedup; the deterministic-uid limitation across update() cutover is documented. --- CHANGELOG.md | 28 +- docs/graphrag-accuracy-benchmark.mdx | 2 +- docs/incremental-updates.mdx | 7 +- graphrag_sdk/.venv | 1 + graphrag_sdk/src/graphrag_sdk/api/main.py | 74 +++-- graphrag_sdk/src/graphrag_sdk/core/models.py | 25 ++ .../src/graphrag_sdk/ingestion/pipeline.py | 252 +++++++++++++----- graphrag_sdk/tests/conftest.py | 9 +- graphrag_sdk/tests/test_facade.py | 47 +++- graphrag_sdk/tests/test_pipeline.py | 231 +++++++++++++++- 10 files changed, 570 insertions(+), 106 deletions(-) create mode 120000 graphrag_sdk/.venv diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b908b13..bee0445c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,19 +41,43 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 same text hash (previously a fresh `text-` per call), so ingesting identical text twice is now one document and a no-op the second time; pass `document_id` to store identical text as distinct documents. + **Breaking:** code that called `ingest(text=...)` once per record and + relied on every call creating its own Document now gets one Document per + distinct text and a zero-result `skipped_unchanged` no-op for each + duplicate — pass a `document_id` per record to keep the old behaviour. + A caller-supplied `document_info` is merged field by field using + `model_fields_set`, so `DocumentInfo(path=...)` without an explicit `uid` + takes the derived id instead of the model's random default. Ids the + pipeline derives from a source path are checked for the reserved + `__pending__` marker (`ValueError`), the same rule `GraphRAG` applies to + explicit ids. **Upgrade note:** graphs built before this change hold random chunk ids; the first re-ingest of an existing document adds one more copy of its chunk layer (matching nothing), and is stable from the second re-ingest on. Because position is part of the id, inserting a paragraph into an edited document re-ids every later chunk; only byte-identical files are a no-op. + Chunks written by `update()` are keyed on its transient pending id (the + pending and live documents must not share chunk nodes during the + cutover); the `content_hash` the cutover records is what keeps a later + `ingest()` of that content a no-op. - Re-ingesting an unchanged document is now a true no-op. The pipeline hashes the loaded text and, if the stored Document carries the same `content_hash`, returns before chunking with `IngestionResult.metadata["skipped_unchanged"] = True` and `chunks_indexed = 0` — no chunker, NER, LLM or graph calls (measured: 33 provider calls and ~30 s → 0 and 0.01 s). The hash is written only after a - run completes, so a partially failed ingest is retried in full rather than - skipped. + run completes with every write reported in full: a failed extraction, a + `RELATES`/`MENTIONED_IN` edge the graph store dropped after a transient + error, or a chunk left without an embedding by a rate-limited embedder all + withhold the hash (`IngestionResult.metadata["incomplete_writes"]` lists + the shortfalls), so the next ingest repairs the document instead of + skipping it. Graph-store adapters without `get_document_record` keep + working — the pipeline skips the check instead of raising. +- `GraphRAG.update(..., force=True)` re-extracts a document whose content + hash is unchanged. With `ingest()` now skipping unchanged documents, this + is the supported way to re-chunk or re-extract existing text after + changing the ontology, chunker, extractor or model (for example to adopt + the new 384-token default below); `update_sync()` takes the same flag. ### Changed diff --git a/docs/graphrag-accuracy-benchmark.mdx b/docs/graphrag-accuracy-benchmark.mdx index 314d779f..eb10aaa6 100644 --- a/docs/graphrag-accuracy-benchmark.mdx +++ b/docs/graphrag-accuracy-benchmark.mdx @@ -103,7 +103,7 @@ model and the judge. | Generation temperature | 0.7 | Appendix H.2 | | Framework | GraphRAG-SDK 1.3.0 (PyPI) on FalkorDB | — | | Graph layout | one graph per corpus document | — | -| Chunking | `SentenceTokenCapChunking`, max_tokens 512, overlap 2 sentences | SDK default | +| Chunking | `SentenceTokenCapChunking`, max_tokens 512, overlap 2 sentences | SDK default at 1.3.0 (the default is now 384) | | Retrieval | `MultiPathRetrieval` — chunk_top_k 15, rel_top_k 15, max_entities 30, max_relationships 20, keyword_limit 10 | SDK default | | Embeddings | `text-embedding-3-large` @ 1024 dimensions | Declared deviation | | Text-to-Cypher | enabled (`enable_cypher=True`) | Declared deviation | diff --git a/docs/incremental-updates.mdx b/docs/incremental-updates.mdx index b92544f7..d151bb67 100644 --- a/docs/incremental-updates.mdx +++ b/docs/incremental-updates.mdx @@ -53,11 +53,12 @@ result = await rag.update( | Argument | Meaning | | --------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `source` | File path. Mutually exclusive with `text`. In file mode, `document_id` defaults to `os.path.normpath(source)`. | +| `source` | File path. Mutually exclusive with `text`. In file mode, `document_id` defaults to the normalised path (a URI source is used verbatim) — the same id `ingest(source)` derives. | | `text` | Raw text. Skips the loader. Requires an explicit `document_id`. | | `document_id` | Stable id of the `Document` node to update. Required for text mode. | | `loader` / `chunker` / `extractor` / `resolver` | Per-call strategy overrides, identical to `ingest()`. | | `if_missing` | `"error"` (default) raises `DocumentNotFoundError` if the id is unknown. `"ingest"` falls through to a fresh `ingest()` — upsert semantics. | +| `force` | `False` (default) returns `no_op=True` when the content hash is unchanged. `True` re-extracts anyway — the way to apply a new ontology, chunker or extractor to existing text, since `ingest()` also skips unchanged documents. | ### Returns — `UpdateResult` @@ -79,7 +80,7 @@ UpdateResult( | Scenario | What changes | | ----------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Content hash matches (touch-only edit) | **Nothing.** SHA-256 short-circuits the call to a single Cypher lookup. `no_op=True`. Use this for CRLF/formatter-only PRs. | +| Content hash matches (touch-only edit) | **Nothing.** SHA-256 short-circuits the call to a single Cypher lookup. `no_op=True`. Use this for CRLF/formatter-only PRs. Pass `force=True` to re-extract regardless. | | Real content change | New chunks written under a pending Document, then atomically cut over to the canonical id. Old chunks are deleted. Entities previously mentioned only by this document are removed. | | Entity still referenced by another doc | **Preserved.** Orphan cleanup is scoped to candidates from this document — never global. | | `RELATES` edge sourced only from old chunks | **Removed** as a stale fact (cleanup keyed on `source_chunk_ids`). | @@ -124,7 +125,7 @@ result = await rag.delete_document( | Argument | Meaning | | ------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `document_id` | The Document node id (e.g. `os.path.normpath(path)` for file-mode ingests). | +| `document_id` | The Document node id (the normalised path for file-mode ingests; `text-<16hex>` for text-mode ingests without an explicit id). | | `if_missing` | `"error"` (default) raises `DocumentNotFoundError`. `"ignore"` returns an empty result with zero counts — useful for CI deletes when the caller doesn't track which files were ever ingested. | ### Returns — `DeleteDocumentResult` diff --git a/graphrag_sdk/.venv b/graphrag_sdk/.venv new file mode 120000 index 00000000..764a44cd --- /dev/null +++ b/graphrag_sdk/.venv @@ -0,0 +1 @@ +/Users/naseemali/Documents/GitHub/GraphRAG-SDK/graphrag_sdk/.venv \ No newline at end of file diff --git a/graphrag_sdk/src/graphrag_sdk/api/main.py b/graphrag_sdk/src/graphrag_sdk/api/main.py index df6345db..bca0a21c 100644 --- a/graphrag_sdk/src/graphrag_sdk/api/main.py +++ b/graphrag_sdk/src/graphrag_sdk/api/main.py @@ -38,6 +38,7 @@ RagResult, RetrieverResult, UpdateResult, + ensure_no_pending_marker, stable_document_id, ) from graphrag_sdk.core.providers import Embedder, LLMInterface @@ -1355,11 +1356,24 @@ async def ingest( - **File mode** — pass ``source`` (single path or list of paths). The loader reads from disk; ``document_id`` is optional and, - when omitted, defaults to ``os.path.normpath(source)`` so the - path itself is the stable handle for ``update()`` later. + when omitted, defaults to the normalised path (a source with a + URI scheme such as ``https://`` or ``s3://`` is used verbatim) + so the path itself is the stable handle for ``update()`` later. - **Text mode** — pass ``text`` directly. Optionally pass - ``document_id`` to label the document; if omitted, an - identifier is generated. ``source`` and ``loader`` are rejected. + ``document_id`` to label the document; if omitted, the id is + ``text-<16hex>``, a SHA-256 prefix of the text. ``source`` and + ``loader`` are rejected. + + Re-ingesting a document whose text is byte-identical to what the + graph already holds is a no-op: the call returns before chunking + with ``chunks_indexed=0`` and ``metadata["skipped_unchanged"]=True`` + and makes no provider calls. Consequently, in text mode two calls + with the same text and no ``document_id`` address the *same* + Document — the second is the no-op — rather than one new document + per call as previously. Pass a distinct ``document_id`` per call to + keep identical texts as separate documents. To re-extract + unchanged content under a new ontology, chunker or extractor use + ``update(..., force=True)``. Note: Call :meth:`finalize` once after all sources are ingested to run @@ -1384,8 +1398,10 @@ async def ingest( source: File path (or list of paths) — file mode only. text: Raw text — text mode only. document_id: Stable identifier used as the Document node's - ``id``. In file mode, defaults to ``os.path.normpath(source)``. - In text mode, defaults to a generated ``text-<8hex>`` id. + ``id``. In file mode, defaults to the normalised path (URI + sources verbatim). In text mode, defaults to + ``text-<16hex>`` derived from the text. Must not contain + the reserved substring ``__pending__``. Pass an explicit value when you want a different identity scheme (e.g. content-hash, repo-relative path, slug). loader: Custom loader strategy. File mode only. @@ -1479,14 +1495,11 @@ def _check_no_pending_marker(document_id: str) -> None: leftover pending of ``foo`` — leading to either silent rollback of the user's real document or a destructive rollforward against a node that was never an actual pending. + + The rule itself lives in ``core.models.ensure_no_pending_marker`` so + ``IngestionPipeline.run()`` applies it to derived ids too. """ - if "__pending__" in document_id: - raise ValueError( - f"document_id '{document_id}' contains the reserved substring " - "'__pending__' which is used internally by the update() " - "state-machine cutover. Pick a different id (or rename the " - "source file) to avoid prefix-collision with pending nodes." - ) + ensure_no_pending_marker(document_id) @staticmethod def _resolve_document_id( @@ -1976,6 +1989,7 @@ async def update( resolver: ResolutionStrategy | None = None, cache_unchanged_chunks: bool = False, if_missing: Literal["error", "ingest"] = "error", + force: bool = False, ctx: Context | None = None, ) -> UpdateResult: """Re-sync a previously-ingested document into the graph. @@ -1989,7 +2003,10 @@ async def update( SHA-256 content-hash short-circuits no-op updates: if the new text matches the stored ``Document.content_hash``, no extraction runs and the call is essentially a single Cypher lookup. This is - the win for touch-only PRs (CRLF, formatter-only changes). + the win for touch-only PRs (CRLF, formatter-only changes). Pass + ``force=True`` to re-extract anyway — the only way to re-process + unchanged content after changing the ontology, chunker, extractor + or model, since ``ingest()`` skips unchanged documents too. State-machine cutover (crash-safe). Columns: ``pend`` = pending Document exists; @@ -2031,7 +2048,8 @@ async def update( text: Raw text. Skips the loader. Mutually exclusive with ``source``. document_id: Stable id of the Document node to update. In - file mode, defaults to ``os.path.normpath(source)`` so + file mode, defaults to the same id ``ingest(path)`` derives + (the normalised path; a URI source verbatim) so ``update(path)`` matches the corresponding ``ingest(path)`` with no extra plumbing. Required in text mode. loader / chunker / extractor / resolver: Per-call strategy @@ -2056,6 +2074,13 @@ async def update( if_missing: ``"error"`` (default) raises ``DocumentNotFoundError`` when the id is unknown. ``"ingest"`` falls through to ``ingest()`` for upsert semantics. + force: Re-extract even when the content hash is unchanged. + Default ``False`` returns ``UpdateResult(no_op=True)`` on + identical content; ``True`` runs the full replace so a new + ontology, chunker, extractor or model is applied to + existing text. Combine with ``cache_unchanged_chunks=False`` + (the default) — the cache would otherwise hand back the + previous extraction for every chunk. ctx: Execution context. Returns: @@ -2152,11 +2177,18 @@ async def update( ) if existing.content_hash == new_hash: - ctx.log(f"update: content hash matches for '{resolved_id}', no-op") - return UpdateResult( - document_info=DocumentInfo(uid=resolved_id, path=existing.path or doc_path), - no_op=True, - replaced_existing=True, + if not force: + ctx.log( + f"update: content hash matches for '{resolved_id}', no-op " + f"(pass force=True to re-extract unchanged content)" + ) + return UpdateResult( + document_info=DocumentInfo(uid=resolved_id, path=existing.path or doc_path), + no_op=True, + replaced_existing=True, + ) + ctx.log( + f"update: content hash matches for '{resolved_id}' but force=True; re-extracting" ) # ── Phase 2: snapshot entity candidates AND old chunk ids BEFORE @@ -3237,6 +3269,7 @@ def update_sync( resolver: ResolutionStrategy | None = None, cache_unchanged_chunks: bool = False, if_missing: Literal["error", "ingest"] = "error", + force: bool = False, ctx: Context | None = None, ) -> UpdateResult: """Synchronous update convenience method. @@ -3261,6 +3294,7 @@ def update_sync( resolver=resolver, cache_unchanged_chunks=cache_unchanged_chunks, if_missing=if_missing, + force=force, ctx=ctx, ) ) diff --git a/graphrag_sdk/src/graphrag_sdk/core/models.py b/graphrag_sdk/src/graphrag_sdk/core/models.py index f2ac1328..aa3c4409 100644 --- a/graphrag_sdk/src/graphrag_sdk/core/models.py +++ b/graphrag_sdk/src/graphrag_sdk/core/models.py @@ -112,6 +112,31 @@ def stable_document_id(source: str) -> str: return os.path.normpath(source) +# Separator ``GraphRAG.update()`` uses to build the transient id of the +# Document written during its crash-safe cutover (``__pending__<8hex>``). +# Reserved: a real Document id containing it would be picked up by the +# ``STARTS WITH "__pending__"`` recovery scan. +PENDING_ID_MARKER = "__pending__" + + +def ensure_no_pending_marker(document_id: str) -> None: + """Raise ``ValueError`` if ``document_id`` contains :data:`PENDING_ID_MARKER`. + + Applied to every Document id that is *not* a pending id — explicit ids in + ``GraphRAG.ingest()`` / ``update()`` / ``delete_document()`` and ids the + ingestion pipeline derives from a source path — so a file called + ``foo__pending__bar.txt`` can never be mistaken for an interrupted update + of ``foo`` and rolled back or rolled forward over the real document. + """ + if PENDING_ID_MARKER in document_id: + raise ValueError( + f"document_id '{document_id}' contains the reserved substring " + f"'{PENDING_ID_MARKER}' which is used internally by the update() " + "state-machine cutover. Pick a different id (or rename the " + "source file) to avoid prefix-collision with pending nodes." + ) + + class DocumentInfo(DataModel): """Metadata about the source document.""" diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index a3a94f5b..bde2aab6 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -24,6 +24,7 @@ IngestionResult, Ontology, TextChunks, + ensure_no_pending_marker, stable_document_id, ) from graphrag_sdk.ingestion.chunking_strategies.base import ChunkingStrategy @@ -39,6 +40,56 @@ _PATTERN_MISMATCH_SAMPLE_SIZE = 3 +def _has_explicit(info: DocumentInfo, field: str) -> bool: + """True when the caller set ``field`` on ``info`` (and to a non-empty value). + + ``DocumentInfo.uid`` has a ``uuid4()`` default, so a truthiness test cannot + tell "the caller chose this id" from "nobody did"; Pydantic's + ``model_fields_set`` can. + """ + return field in info.model_fields_set and bool(getattr(info, field)) + + +def _resolve_identity( + *, + loaded: DocumentInfo, + supplied: DocumentInfo | None, + derived_uid: str, + source: str, +) -> DocumentInfo: + """Merge the loader's ``DocumentInfo`` with the caller's into the identity + the run will write under. + + Precedence per field: caller's explicitly-set value, then the derived id + (for ``uid``) or the loader's value / ``source`` (for ``path``). Metadata is + the loader's overlaid with the caller's. The loader's own ``uid`` is never + used — no shipped loader sets one, so it is always the random default. + """ + if supplied is not None and _has_explicit(supplied, "uid"): + uid = supplied.uid + else: + uid = derived_uid + if supplied is not None and _has_explicit(supplied, "path"): + path = supplied.path + else: + path = loaded.path or source + metadata = {**loaded.metadata, **(supplied.metadata if supplied is not None else {})} + return DocumentInfo(uid=uid, path=path, metadata=metadata) + + +def _reported_short(reported: Any, expected: int) -> bool: + """True when a store reported writing fewer items than it was handed. + + ``GraphStore.upsert_relationships`` and ``VectorStore.index_chunks`` do + not raise on per-item failures — they log, skip and return a count — so + the count is the only signal that a relationship or an embedding is + missing. Stores that return nothing (``None``, a mock) are taken at their + word: the pipeline cannot tell and must not refuse to ever mark a run + complete against a duck-typed store. + """ + return isinstance(reported, int) and not isinstance(reported, bool) and reported < expected + + def _assign_deterministic_chunk_uids(doc_info: DocumentInfo, chunks: TextChunks) -> None: """Replace random chunk UIDs with a content-derived, stable identity. @@ -77,6 +128,17 @@ def _assign_deterministic_chunk_uids(doc_info: DocumentInfo, chunks: TextChunks) ``doc:index`` a true position — but it means an *edited* file grows the lexical layer under plain ``ingest()``; only a byte-identical file is a no-op (see the unchanged-document short-circuit in ``run``). + + ``GraphRAG.update()`` runs the pipeline under a transient + ``__pending__`` Document, so the chunks it writes are keyed on + that pending id, not the final one. This is load-bearing, not an + oversight: the pending and live documents coexist until the cutover, and + the cutover deletes every chunk ``PART_OF`` the live document — chunks + keyed on the final id would MERGE onto the live nodes and be deleted + along with them. After ``update()`` the Document carries the new + ``content_hash``, so a later ``ingest()`` of that content is a no-op and + never re-derives ids; only an ``ingest()`` of *edited* content re-mints + them (as it does for any edited file). """ doc_key = doc_info.uid or doc_info.path or "" for chunk in chunks.chunks: @@ -191,60 +253,51 @@ async def run( if ctx is None: ctx = Context() + # Identity the pipeline will use when the caller does not pin one + # (no ``document_info``, or one without an explicit ``uid``). Text + # mode hashes the text; file mode applies ``stable_document_id`` + # (normalised path, URIs verbatim). Computed up front so the + # reserved-marker check fails fast, before any I/O, and is not + # wrapped in ``IngestionError`` — same ``ValueError`` the facade + # raises for an explicit ``document_id``. + if text is not None: + derived_uid = f"text-{hashlib.sha256(text.encode('utf-8')).hexdigest()[:16]}" + else: + derived_uid = stable_document_id(source) + if document_info is None or not _has_explicit(document_info, "uid"): + # Only *derived* ids are checked: ``GraphRAG.update()`` passes + # its ``__pending__`` DocumentInfo explicitly and must + # get through. + ensure_no_pending_marker(derived_uid) + ctx.log("Pipeline starting") try: # Step 1: Load if text is not None: - document = DocumentOutput( - text=text, - document_info=document_info or DocumentInfo(path=source), - ) + document = DocumentOutput(text=text, document_info=DocumentInfo(path=source)) ctx.log("Using provided text (loader skipped)") - if document_info is None: - # Text mode with no caller-supplied identity: derive the - # Document id from the text so the same text ingested - # twice is the same document (and the chunk ids below, - # which include the document id, are stable too). - document.document_info = DocumentInfo( - uid=f"text-{hashlib.sha256(text.encode('utf-8')).hexdigest()[:16]}", - path=source, - metadata=document.document_info.metadata, - ) else: ctx.log("Step 1/9: Loading source") document = await self.loader.load(source, ctx) - # Loaders leave ``uid`` at its ``uuid4()`` default, so - # without a caller-supplied ``document_info`` every run of - # the same file was a new Document — and, since the chunk - # ids include the document id, a new set of chunks. Derive - # the id from the source (normalised path; URIs verbatim), - # the same rule ``GraphRAG._resolve_document_id`` applies, - # so the pipeline is idempotent on its own and not only - # through the facade. ``update()`` always passes an - # explicit pending id and is unaffected. - if document_info is None: - document.document_info = DocumentInfo( - uid=stable_document_id(source), - path=document.document_info.path or source, - metadata=document.document_info.metadata, - ) - # When the caller supplies a ``document_info`` (e.g. for - # stable-id ingestion or update()), prefer its uid/path - # over whatever the loader produced. The loader-side - # metadata is preserved. Each field falls back to the - # loader's value when the caller's is empty/missing, so - # a partially-populated DocumentInfo (e.g. only ``path`` - # set) cannot accidentally clobber the loader's uid. - if document_info is not None: - document.document_info = DocumentInfo( - uid=document_info.uid or document.document_info.uid, - path=document_info.path or document.document_info.path, - metadata={ - **document.document_info.metadata, - **document_info.metadata, - }, - ) + + # Loaders leave ``DocumentInfo.uid`` at its ``uuid4()`` default, + # so without a stable identity every run of the same file was a + # new Document — and, since the chunk ids include the document + # id, a new set of chunks. Resolve the identity here so the + # pipeline is idempotent on its own and not only through the + # facade: the caller's explicitly-set fields win, then the + # derived id / loader path, and metadata is merged (caller over + # loader). ``model_fields_set`` is what tells an explicit + # ``uid=`` apart from the random default — a + # ``DocumentInfo(path=...)`` must not clobber the derived id + # with a fresh uuid. + document.document_info = _resolve_identity( + loaded=document.document_info, + supplied=document_info, + derived_uid=derived_uid, + source=source, + ) # Hash the loaded text. Written to the Document node at the END of # a successful run (step 9b) so ``GraphRAG.update()`` and the @@ -268,18 +321,25 @@ async def run( # Same rule update() uses for its no-op: a stored Document whose # content_hash matches means every chunk, entity and edge is # already there — the hash is only written once a run has - # completed, so a half-finished ingest is never mistaken for a - # finished one. Changed text (new hash) takes the full path; a - # changed ontology or strategy with the same text is update()'s - # job. + # completed with every write reported in full, so a half-finished + # ingest is never mistaken for a finished one. Changed text (new + # hash) takes the full path; a changed ontology or strategy with + # the same text is ``GraphRAG.update(force=True)``'s job. + # + # ``graph_store`` is duck-typed (``Any`` — see ``__init__``); the + # pipeline only ever required ``upsert_nodes`` / + # ``upsert_relationships``. A store without ``get_document_record`` + # simply never short-circuits (the previous behaviour) rather than + # failing every ingest with ``AttributeError``. doc_uid = document.document_info.uid - if doc_uid: - existing = await self.graph_store.get_document_record(doc_uid) + get_record = getattr(self.graph_store, "get_document_record", None) + if doc_uid and get_record is not None: + existing = await get_record(doc_uid) if existing is not None and existing.content_hash == content_hash: ctx.log( f"Document '{doc_uid}' already ingested with identical content; " - f"skipping. Use GraphRAG.update() to re-extract with a new " - f"ontology or strategy." + f"skipping. Use GraphRAG.update(..., force=True) to re-extract " + f"unchanged content with a new ontology, chunker or extractor." ) return IngestionResult( document_info=document.document_info, @@ -297,9 +357,17 @@ async def run( _assign_deterministic_chunk_uids(document.document_info, chunks) + # Writes that report a shortfall instead of raising (see + # ``_reported_short``). Any entry here means the graph is not a + # faithful copy of this document, so the content hash is *not* + # recorded and the next ingest repairs it by re-running. + incomplete_writes: list[str] = [] + # Step 3: Build lexical graph (MANDATORY — not a strategy). ctx.log("Step 3/9: Building lexical graph (provenance chain)") - await self._build_lexical_graph(document.document_info, chunks, ctx) + lexical_shortfall = await self._build_lexical_graph(document.document_info, chunks, ctx) + if lexical_shortfall: + incomplete_writes.append(lexical_shortfall) # Step 4: Extract entities & relationships ctx.log("Step 4/9: Extracting entities & relationships") @@ -333,7 +401,11 @@ async def run( # Step 7: Write to graph (batched) ctx.log("Step 7/9: Writing to graph store") await self.graph_store.upsert_nodes(resolved.nodes) - await self.graph_store.upsert_relationships(resolved.relationships) + rels_written = await self.graph_store.upsert_relationships(resolved.relationships) + if _reported_short(rels_written, len(resolved.relationships)): + incomplete_writes.append( + f"relationships {rels_written}/{len(resolved.relationships)}" + ) # ╔══════════════════════════════════════════════════════════╗ # ║ LOAD-BEARING — DO NOT MAKE STEP 8 ASYNCHRONOUS WITH ║ @@ -361,25 +433,52 @@ async def run( # Tripwire: tests/test_integration.py:: # TestIncrementalUpdateInvariants:: # test_concurrent_updates_preserve_shared_entity - async def _step_mentions() -> int: + async def _step_mentions() -> tuple[int, str | None]: ctx.log("Step 8/9: Writing mentions (uncapped)") return await self._write_mentions(graph_data, ctx) - async def _step_index_chunks() -> None: + async def _step_index_chunks() -> str | None: ctx.log("Step 9/9: Embedding & indexing chunks") - await self.vector_store.index_chunks(chunks) + indexed = await self.vector_store.index_chunks(chunks) + if _reported_short(indexed, len(chunks.chunks)): + return f"chunks indexed {indexed}/{len(chunks.chunks)}" + return None - mentions_written, _ = await asyncio.gather( + (mentions_written, mentions_shortfall), index_shortfall = await asyncio.gather( _step_mentions(), _step_index_chunks(), ) + incomplete_writes.extend(s for s in (mentions_shortfall, index_shortfall) if s) # Step 9b: only now record the content hash. Writing it in step 3 # meant a failure in extraction or the graph write left a # Document that looked complete, and the next ingest skipped it # (Copilot review on #309). A partial run therefore retries in - # full; only a finished run is recognised as unchanged. - await self._mark_content_hash(document.document_info.uid, content_hash) + # full; only a finished run is recognised as unchanged. "Finished" + # includes the writes that swallow per-item errors: a RELATES + # edge dropped by a transient graph error or a chunk left without + # an embedding by a rate-limited embedder must not be certified + # complete, or it would never be repaired (galshubeli on #309). + result_metadata: dict[str, Any] = { + "merged_entities": resolved.merged_count, + "raw_nodes": len(graph_data.nodes), + "raw_relationships": len(graph_data.relationships), + "mention_edges_created": mentions_written, + } + if incomplete_writes: + result_metadata["incomplete_writes"] = incomplete_writes + ctx.log( + "Some writes were reported incomplete " + f"({'; '.join(incomplete_writes)}); content_hash not recorded — " + "the next ingest of this document re-runs in full" + ) + logger.warning( + "Ingest of '%s' left incomplete writes (%s); not marking content_hash", + document.document_info.uid, + "; ".join(incomplete_writes), + ) + else: + await self._mark_content_hash(document.document_info.uid, content_hash) total_rels = len(resolved.relationships) + mentions_written result = IngestionResult( @@ -387,12 +486,7 @@ async def _step_index_chunks() -> None: nodes_created=len(resolved.nodes), relationships_created=total_rels, chunks_indexed=len(chunks.chunks), - metadata={ - "merged_entities": resolved.merged_count, - "raw_nodes": len(graph_data.nodes), - "raw_relationships": len(graph_data.relationships), - "mention_edges_created": mentions_written, - }, + metadata=result_metadata, ) ctx.log( f"Pipeline complete: {result.nodes_created} nodes, " @@ -425,7 +519,7 @@ async def _build_lexical_graph( doc_info: DocumentInfo, chunks: TextChunks, ctx: Context, - ) -> None: + ) -> str | None: """Build the mandatory provenance chain. Creates: @@ -440,6 +534,9 @@ async def _build_lexical_graph( The Document's ``content_hash`` is *not* written here; ``run`` sets it via :meth:`_mark_content_hash` once the whole pipeline has succeeded. + + Returns ``None`` when every edge was reported written, else a short + description of the shortfall for ``run`` to record. """ # Document node doc_props: dict[str, Any] = { @@ -494,12 +591,16 @@ async def _build_lexical_graph( prev_chunk_id = chunk.uid await self.graph_store.upsert_nodes(chunk_nodes) - await self.graph_store.upsert_relationships(part_of_rels + next_chunk_rels) + lexical_rels = part_of_rels + next_chunk_rels + written = await self.graph_store.upsert_relationships(lexical_rels) ctx.log( f"Lexical graph: 1 Document, {len(chunk_nodes)} Chunks, " f"{len(part_of_rels)} PART_OF, {len(next_chunk_rels)} NEXT_CHUNK" ) + if _reported_short(written, len(lexical_rels)): + return f"lexical edges {written}/{len(lexical_rels)}" + return None def _prune(self, graph_data: GraphData, ontology: Ontology) -> GraphData: """Filter graph data to only include ontology-conforming nodes and relationships. @@ -644,17 +745,20 @@ def _remap_mentions(graph_data: GraphData, remap: dict[str, str]) -> GraphData: rewritten.append(EntityMention(entity_id=new_id, chunk_id=m.chunk_id)) return graph_data.model_copy(update={"mentions": rewritten}) - async def _write_mentions(self, graph_data: GraphData, ctx: Context) -> int: + async def _write_mentions(self, graph_data: GraphData, ctx: Context) -> tuple[int, str | None]: """Write MENTIONED_IN edges linking entities to their source chunks. Every entity connects to every chunk it was extracted from (uncapped). With global dedup controlling entity cardinality, uncapped mentions provide richer entity-chunk connectivity for retrieval. + + Returns ``(edges attempted, shortfall)`` where ``shortfall`` is ``None`` + when the store reported every edge written. """ mentions: list[EntityMention] = graph_data.mentions or [] if not mentions: - return 0 + return 0, None seen: set[tuple[str, str]] = set() mention_rels: list[GraphRelationship] = [] @@ -670,6 +774,8 @@ async def _write_mentions(self, graph_data: GraphData, ctx: Context) -> int: type="MENTIONED_IN", ) ) - await self.graph_store.upsert_relationships(mention_rels) + written = await self.graph_store.upsert_relationships(mention_rels) ctx.log(f"Wrote {len(mention_rels)} MENTIONED_IN edges (uncapped)") - return len(mention_rels) + if _reported_short(written, len(mention_rels)): + return len(mention_rels), f"mentions {written}/{len(mention_rels)}" + return len(mention_rels), None diff --git a/graphrag_sdk/tests/conftest.py b/graphrag_sdk/tests/conftest.py index 64976dfd..590c301b 100644 --- a/graphrag_sdk/tests/conftest.py +++ b/graphrag_sdk/tests/conftest.py @@ -222,8 +222,11 @@ def mock_graph_store(mock_connection: MagicMock) -> MagicMock: from graphrag_sdk.storage.graph_store import GraphStore store = MagicMock(spec=GraphStore) - store.upsert_nodes = AsyncMock(return_value=0) - store.upsert_relationships = AsyncMock(return_value=0) + # Report every item as written, like the real store does on success — + # the pipeline treats a short count as a partial failure and withholds + # the Document's content_hash. + store.upsert_nodes = AsyncMock(side_effect=lambda nodes: len(nodes)) + store.upsert_relationships = AsyncMock(side_effect=lambda rels: len(rels)) store.get_connected_entities = AsyncMock(return_value=[]) store.query_raw = AsyncMock(return_value=MagicMock(result_set=[])) store.delete_all = AsyncMock() @@ -241,7 +244,7 @@ def mock_vector_store(embedder: MockEmbedder) -> MagicMock: from graphrag_sdk.storage.vector_store import VectorStore store = MagicMock(spec=VectorStore) - store.index_chunks = AsyncMock(return_value=0) + store.index_chunks = AsyncMock(side_effect=lambda chunks: len(chunks.chunks)) store.search_chunks = AsyncMock(return_value=[]) store.search_entities = AsyncMock(return_value=[]) store.search_relationships = AsyncMock(return_value=[]) diff --git a/graphrag_sdk/tests/test_facade.py b/graphrag_sdk/tests/test_facade.py index 7e7d63dd..2598be75 100644 --- a/graphrag_sdk/tests/test_facade.py +++ b/graphrag_sdk/tests/test_facade.py @@ -1552,10 +1552,10 @@ async def _rollforward(*, pending_id, real_id, path, content_hash): g._graph_store.mark_document_pending_delete = AsyncMock(side_effect=_mark_delete) g._graph_store.has_pending_delete = AsyncMock(return_value=False) g._graph_store.rollforward_cutover = AsyncMock(side_effect=_rollforward) - g._graph_store.upsert_nodes = AsyncMock(return_value=0) - g._graph_store.upsert_relationships = AsyncMock(return_value=0) + g._graph_store.upsert_nodes = AsyncMock(side_effect=lambda nodes: len(nodes)) + g._graph_store.upsert_relationships = AsyncMock(side_effect=lambda rels: len(rels)) g._vector_store.ensure_indices = AsyncMock(return_value={}) - g._vector_store.index_chunks = AsyncMock(return_value=0) + g._vector_store.index_chunks = AsyncMock(side_effect=lambda chunks: len(chunks.chunks)) g._vector_store.backfill_entity_embeddings = AsyncMock(return_value=0) @@ -1633,6 +1633,47 @@ async def test_no_op_on_unchanged_content(self, graphrag): graphrag._graph_store.upsert_nodes.assert_not_awaited() graphrag._graph_store.rollforward_cutover.assert_not_awaited() + async def test_force_re_extracts_unchanged_content(self, graphrag): + """``ingest()`` skips unchanged documents and ``update()`` no-ops on + them, so ``force=True`` is the only way to apply a new ontology, + chunker or extractor to existing text (galshubeli on #309). It runs + the full replace flow — pending write, commit, cutover, cleanup.""" + import hashlib + + from graphrag_sdk.core.models import DocumentRecord + + text = "Stable content." + existing_hash = hashlib.sha256(text.encode("utf-8")).hexdigest() + _stub_graph_store_for_update( + graphrag, + existing_record={"path": "my-doc", "content_hash": existing_hash}, + candidates=["entity-1"], + cutover_chunks_deleted=1, + ) + # Only the live id resolves; the pipeline's own unchanged-content + # check looks up the *pending* id, which a real store has no record of. + live = DocumentRecord(path="my-doc", content_hash=existing_hash) + graphrag._graph_store.get_document_record = AsyncMock( + side_effect=lambda doc_id: live if doc_id == "my-doc" else None + ) + + result = await graphrag.update(text=text, document_id="my-doc", force=True) + + assert result.no_op is False + assert result.replaced_existing is True + assert result.chunks_deleted == 1 + graphrag._graph_store.upsert_nodes.assert_awaited() + graphrag._graph_store.mark_pending_committed.assert_awaited_once() + graphrag._graph_store.rollforward_cutover.assert_awaited_once() + graphrag._graph_store.delete_orphan_entities.assert_awaited_once_with(["entity-1"]) + + def test_update_sync_forwards_force(self, graphrag): + """Keep ``update_sync`` in step with ``update``.""" + import inspect + + assert "force" in inspect.signature(graphrag.update_sync).parameters + assert "force" in inspect.signature(graphrag.update).parameters + async def test_doc_not_found_default_raises(self, graphrag): """if_missing='error' (default) raises DocumentNotFoundError.""" _stub_graph_store_for_update(graphrag, existing_record=None) diff --git a/graphrag_sdk/tests/test_pipeline.py b/graphrag_sdk/tests/test_pipeline.py index 58c04551..5cceda9d 100644 --- a/graphrag_sdk/tests/test_pipeline.py +++ b/graphrag_sdk/tests/test_pipeline.py @@ -908,7 +908,236 @@ async def test_text_mode_without_document_info_is_stable_too( assert ids_a and ids_a == self._chunk_ids(mock_graph_store) -class TestDeterministicChunkUids: +class TestDocumentInfoIdentityMerge: + """A caller-supplied ``DocumentInfo`` is merged field by field, and only + fields the caller actually set win. ``DocumentInfo.uid`` has a ``uuid4()`` + default, so ``DocumentInfo(path=...)`` must not smuggle a random id past + the derived one (Copilot / CodeRabbit on #309).""" + + def _pipeline(self, mock_graph_store, mock_vector_store): + mock_graph_store.get_document_record = AsyncMock(return_value=None) + return IngestionPipeline( + loader=StubLoader("Alice works at Acme. Bob works at Beta."), + chunker=StubChunker(), + extractor=StubExtractor(), + resolver=StubResolver(), + graph_store=mock_graph_store, + vector_store=mock_vector_store, + ontology=Ontology(), + ) + + async def test_path_only_document_info_takes_the_derived_id( + self, ctx, mock_graph_store, mock_vector_store + ): + p = self._pipeline(mock_graph_store, mock_vector_store) + first = await p.run("./docs/../report.txt", ctx, document_info=DocumentInfo(path="docs/report.txt")) + second = await p.run("./docs/../report.txt", ctx, document_info=DocumentInfo(path="docs/report.txt")) + + assert first.document_info.uid == second.document_info.uid == "report.txt" + assert first.document_info.path == "docs/report.txt" + + async def test_explicit_uid_still_wins(self, ctx, mock_graph_store, mock_vector_store): + p = self._pipeline(mock_graph_store, mock_vector_store) + result = await p.run("report.txt", ctx, document_info=DocumentInfo(uid="pinned")) + assert result.document_info.uid == "pinned" + assert result.document_info.path == "report.txt", "path falls back to the loader's" + + async def test_text_mode_path_only_document_info_takes_the_text_hash( + self, ctx, mock_graph_store, mock_vector_store + ): + import hashlib + + p = self._pipeline(mock_graph_store, mock_vector_store) + text = "Alice works at Acme." + result = await p.run("label", ctx, text=text, document_info=DocumentInfo(path="label")) + assert result.document_info.uid == f"text-{hashlib.sha256(text.encode()).hexdigest()[:16]}" + + async def test_metadata_is_merged_caller_over_loader( + self, ctx, mock_graph_store, mock_vector_store + ): + class MetaLoader(StubLoader): + async def load(self, source, ctx): + return DocumentOutput( + text=self._text, + document_info=DocumentInfo(path=source, metadata={"a": 1, "b": 1}), + ) + + p = self._pipeline(mock_graph_store, mock_vector_store) + p.loader = MetaLoader("Alice works at Acme.") + result = await p.run("r.txt", ctx, document_info=DocumentInfo(metadata={"b": 2, "c": 3})) + assert result.document_info.metadata == {"a": 1, "b": 2, "c": 3} + + async def test_derived_id_with_pending_marker_is_rejected_before_any_io( + self, ctx, mock_graph_store, mock_vector_store + ): + """Direct pipeline callers bypass ``GraphRAG``'s guard; a file called + ``foo__pending__bar.txt`` would otherwise be picked up by + ``find_pending("foo")`` and rolled back as an interrupted update.""" + p = self._pipeline(mock_graph_store, mock_vector_store) + p.loader.load = AsyncMock(wraps=p.loader.load) + + with pytest.raises(ValueError, match="reserved substring '__pending__'"): + await p.run("foo__pending__bar.txt", ctx) + + p.loader.load.assert_not_called() + mock_graph_store.upsert_nodes.assert_not_called() + + async def test_explicit_pending_id_is_still_accepted( + self, ctx, mock_graph_store, mock_vector_store + ): + """``GraphRAG.update()`` passes its pending id explicitly and must get through.""" + p = self._pipeline(mock_graph_store, mock_vector_store) + result = await p.run("r.txt", ctx, document_info=DocumentInfo(uid="r.txt__pending__ab12cd34")) + assert result.document_info.uid == "r.txt__pending__ab12cd34" + + +class TestDuckTypedGraphStore: + """``graph_store`` is ``Any`` by design; a store that only implements the + write surface the pipeline always required must still work + (galshubeli on #309).""" + + async def test_store_without_get_document_record_ingests_without_the_skip( + self, ctx, mock_vector_store + ): + class MinimalStore: + def __init__(self): + self.nodes = [] + self.rels = [] + + async def upsert_nodes(self, nodes): + self.nodes.extend(nodes) + return len(nodes) + + async def upsert_relationships(self, rels): + self.rels.extend(rels) + return len(rels) + + store = MinimalStore() + p = IngestionPipeline( + loader=StubLoader("Alice works at Acme."), + chunker=StubChunker(), + extractor=StubExtractor(), + resolver=StubResolver(), + graph_store=store, + vector_store=mock_vector_store, + ontology=Ontology(), + ) + result = await p.run("r.txt", ctx) + assert result.chunks_indexed == 1 + assert "skipped_unchanged" not in result.metadata + assert any("content_hash" in n.properties for n in store.nodes if n.label == "Document") + + +class TestContentHashRequiresCompleteWrites: + """``upsert_relationships`` and ``index_chunks`` swallow per-item failures + and return a count. A run whose count came up short must not be stamped + with ``content_hash``, or the missing edges / embeddings would never be + repaired — every later ingest would skip the document (galshubeli on #309).""" + + def _pipeline(self, mock_graph_store, mock_vector_store, extractor=None): + mock_graph_store.get_document_record = AsyncMock(return_value=None) + return IngestionPipeline( + loader=StubLoader("Alice works at Acme. Bob works at Beta. Carol runs Gamma."), + chunker=StubChunker(), + extractor=extractor or StubExtractor(), + resolver=StubResolver(), + graph_store=mock_graph_store, + vector_store=mock_vector_store, + ontology=Ontology(), + ) + + @staticmethod + def _hash_written(mock_graph_store) -> bool: + return any( + "content_hash" in n.properties + for call in mock_graph_store.upsert_nodes.call_args_list + for n in call[0][0] + if n.label == "Document" + ) + + async def test_complete_run_records_the_hash(self, ctx, mock_graph_store, mock_vector_store): + result = await self._pipeline(mock_graph_store, mock_vector_store).run("r.txt", ctx) + assert self._hash_written(mock_graph_store) + assert "incomplete_writes" not in result.metadata + + async def test_short_relationship_count_withholds_the_hash( + self, ctx, mock_graph_store, mock_vector_store + ): + mock_graph_store.upsert_relationships = AsyncMock(side_effect=lambda rels: max(len(rels) - 1, 0)) + result = await self._pipeline(mock_graph_store, mock_vector_store).run("r.txt", ctx) + + assert not self._hash_written(mock_graph_store) + assert any(s.startswith("lexical edges") for s in result.metadata["incomplete_writes"]) + # The run itself still reports what it attempted. + assert result.chunks_indexed == 3 + + async def test_short_mention_count_withholds_the_hash( + self, ctx, mock_graph_store, mock_vector_store + ): + from graphrag_sdk.core.models import EntityMention + + class MentionExtractor(StubExtractor): + async def extract(self, chunks, ontology, ctx): + return GraphData( + nodes=[GraphNode(id="e1", label="Entity", properties={"name": "Alice"})], + relationships=[], + mentions=[EntityMention(entity_id="e1", chunk_id=c.uid) for c in chunks.chunks], + ) + + def _drop_one_mention(rels): + return len(rels) - 1 if rels and rels[0].type == "MENTIONED_IN" else len(rels) + + mock_graph_store.upsert_relationships = AsyncMock(side_effect=_drop_one_mention) + result = await self._pipeline( + mock_graph_store, mock_vector_store, extractor=MentionExtractor() + ).run("r.txt", ctx) + + assert not self._hash_written(mock_graph_store) + assert result.metadata["incomplete_writes"] == ["mentions 2/3"] + + async def test_unembedded_chunks_withhold_the_hash( + self, ctx, mock_graph_store, mock_vector_store + ): + """0 from ``index_chunks`` — every embedding call failed, or no + embedder — means the document is absent from chunk vector search.""" + mock_vector_store.index_chunks = AsyncMock(return_value=0) + result = await self._pipeline(mock_graph_store, mock_vector_store).run("r.txt", ctx) + + assert not self._hash_written(mock_graph_store) + assert result.metadata["incomplete_writes"] == ["chunks indexed 0/3"] + + async def test_stores_that_report_nothing_are_taken_at_their_word( + self, ctx, mock_graph_store, mock_vector_store + ): + """Duck-typed stores returning ``None`` give no shortfall signal, so the + pipeline must not refuse to ever mark such a run complete.""" + mock_graph_store.upsert_relationships = AsyncMock(return_value=None) + mock_vector_store.index_chunks = AsyncMock(return_value=None) + result = await self._pipeline(mock_graph_store, mock_vector_store).run("r.txt", ctx) + + assert self._hash_written(mock_graph_store) + assert "incomplete_writes" not in result.metadata + + async def test_next_ingest_after_a_short_run_takes_the_full_path( + self, ctx, mock_graph_store, mock_vector_store + ): + """The point of withholding the hash: the follow-up ingest repairs.""" + from graphrag_sdk.core.models import DocumentRecord + + mock_vector_store.index_chunks = AsyncMock(return_value=0) + p = self._pipeline(mock_graph_store, mock_vector_store) + await p.run("r.txt", ctx) + # The graph now holds the Document with no content_hash, as the run left it. + mock_graph_store.get_document_record = AsyncMock( + return_value=DocumentRecord(path="r.txt", content_hash=None) + ) + mock_vector_store.index_chunks = AsyncMock(side_effect=lambda chunks: len(chunks.chunks)) + mock_graph_store.upsert_nodes.reset_mock() + + result = await p.run("r.txt", ctx) + + assert "skipped_unchanged" not in result.metadata + assert self._hash_written(mock_graph_store) """Bug #12 — re-ingesting an unchanged document duplicated its chunks. ``TextChunk.uid`` defaulted to ``uuid4()``, so the lexical graph's From a1f05a39469a3be6f41407cef40e65253898d830 Mon Sep 17 00:00:00 2001 From: Naseem Ali <34807727+Naseem77@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:53:22 +0300 Subject: [PATCH 10/10] =?UTF-8?q?fix(ingestion):=20address=20review=20roun?= =?UTF-8?q?d=202=20on=20#309=20=E2=80=94=20gate=20update()=20cutover=20on?= =?UTF-8?q?=20complete=20writes,=20no-embedder=20is=20not=20a=20shortfall?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - update(): Phase 5 now reads pipeline_result.metadata["incomplete_writes"] and promotes the pending Document with content_hash=None when any write came up short, so a dropped RELATES edge or an unembedded chunk leaves the document eligible for repair instead of being certified complete. rollforward_cutover accepts content_hash=None and REMOVEs the property rather than setting it. Phase 0 crash recovery no longer treats a COMMITTED pending without a hash as corruption (path is still required); it rolls forward uncertified. - VectorStore.index_chunks returns None (not attempted) instead of 0 (all failed) when no embedder is configured, so graph-only deployments still record content_hash and get the unchanged-document skip. - pipeline: document why upsert_nodes is not gated on its count (it raises on real failures; the only silent drop is an id that sanitises to empty, which a re-run would drop identically). - Untrack graphrag_sdk/.venv (symlink to a local absolute path) and ignore it explicitly in .gitignore. - tests: restore the dropped `class TestDeterministicChunkUids:` header, replace passthrough lambdas with `len` (CodeQL), add coverage for the update() gate, Phase 0 no-hash rollforward, cutover REMOVE, and the no-embedder pipeline path. --- .gitignore | 1 + CHANGELOG.md | 11 ++- graphrag_sdk/.venv | 1 - graphrag_sdk/src/graphrag_sdk/api/main.py | 78 +++++++++++++------ .../src/graphrag_sdk/ingestion/pipeline.py | 12 ++- .../src/graphrag_sdk/storage/graph_store.py | 35 ++++++--- .../src/graphrag_sdk/storage/vector_store.py | 11 ++- graphrag_sdk/tests/conftest.py | 4 +- graphrag_sdk/tests/test_facade.py | 71 ++++++++++++++++- graphrag_sdk/tests/test_graph_store.py | 29 +++++++ graphrag_sdk/tests/test_pipeline.py | 22 +++++- graphrag_sdk/tests/test_vector_store.py | 7 +- 12 files changed, 231 insertions(+), 51 deletions(-) delete mode 120000 graphrag_sdk/.venv diff --git a/.gitignore b/.gitignore index 22a87d98..9ae19e3d 100644 --- a/.gitignore +++ b/.gitignore @@ -56,6 +56,7 @@ debug_*.py # Virtual environments .venv/ +graphrag_sdk/.venv .venv-*/ venv/ diff --git a/CHANGELOG.md b/CHANGELOG.md index bee0445c..ec8202c3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -71,8 +71,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 error, or a chunk left without an embedding by a rate-limited embedder all withhold the hash (`IngestionResult.metadata["incomplete_writes"]` lists the shortfalls), so the next ingest repairs the document instead of - skipping it. Graph-store adapters without `get_document_record` keep - working — the pipeline skips the check instead of raising. + skipping it. `update()` applies the same gate: its cutover promotes the + pending Document without a hash when the pipeline reported a shortfall + (and crash recovery of such a pending rolls forward uncertified rather + than refusing). Deployments without an embedder are not penalised — + `VectorStore.index_chunks` now returns `None` (nothing attempted) instead + of `0` (every embedding failed) when no embedder is configured, so + graph-only ingests still record the hash. Graph-store adapters without + `get_document_record` keep working — the pipeline skips the check instead + of raising. - `GraphRAG.update(..., force=True)` re-extracts a document whose content hash is unchanged. With `ingest()` now skipping unchanged documents, this is the supported way to re-chunk or re-extract existing text after diff --git a/graphrag_sdk/.venv b/graphrag_sdk/.venv deleted file mode 120000 index 764a44cd..00000000 --- a/graphrag_sdk/.venv +++ /dev/null @@ -1 +0,0 @@ -/Users/naseemali/Documents/GitHub/GraphRAG-SDK/graphrag_sdk/.venv \ No newline at end of file diff --git a/graphrag_sdk/src/graphrag_sdk/api/main.py b/graphrag_sdk/src/graphrag_sdk/api/main.py index bca0a21c..8370dbee 100644 --- a/graphrag_sdk/src/graphrag_sdk/api/main.py +++ b/graphrag_sdk/src/graphrag_sdk/api/main.py @@ -1913,34 +1913,40 @@ async def _phase0_recover_prior_operations(self, resolved_id: str, ctx: Context) f"update: detected COMMITTED pending '{prior_pending_id}' " f"from a prior crash — rolling forward" ) - # The pending node carries the "real" path/hash (set just - # before we crashed). Look those up before rollforward so the - # canonical Document ends up with the right metadata. + # The pending node carries the "real" path (set just before + # we crashed) and, if the pipeline run was complete, its + # hash. Look those up before rollforward so the canonical + # Document ends up with the right metadata. pending_record = await self._graph_store.get_document_record(prior_pending_id) - # A committed pending without persisted path metadata is a - # corruption signal — the pipeline must have completed step 7 - # (write-graph) for the marker to be set, so the path/hash - # MUST be there. Refuse to silently default to the canonical - # id (would write a non-filesystem path) or to "" hash (would - # break future no-op short-circuits forever). - # Both path AND content_hash are required for a COMMITTED - # pending — pipeline must have completed step 7 to write - # them. Refusing to fall back to ``""`` on either: a - # non-filesystem path is wrong, and an empty hash would - # permanently disable the no-op short-circuit on future - # updates (no real SHA-256 will ever match ``""``). - roll_hash = prior_hash or (pending_record.content_hash if pending_record else None) - if pending_record is None or not pending_record.path or not roll_hash: + # A committed pending without a persisted path is a + # corruption signal — the pipeline must have completed step 3 + # (lexical graph) for the marker to be set, so the path MUST + # be there. Refuse to silently default to the canonical id + # (would write a non-filesystem path). + if pending_record is None or not pending_record.path: raise DatabaseError( f"Phase 0 rollforward: COMMITTED pending " f"'{prior_pending_id}' has incomplete metadata " - f"(path={pending_record.path if pending_record else None!r}, " - f"hash={roll_hash!r}). Graph state is inconsistent — " - "possible corruption or partial write before the " - "commit marker. Refusing to proceed; manual " - "intervention required." + f"(path={pending_record.path if pending_record else None!r}). " + "Graph state is inconsistent — possible corruption or " + "partial write before the commit marker. Refusing to " + "proceed; manual intervention required." ) roll_path = pending_record.path + # A missing hash is NOT corruption: the pipeline withholds + # ``content_hash`` when a write came up short (see + # ``IngestionPipeline.run``), and the interrupted update() + # would have promoted the pending without one. Roll forward + # the same way — never fall back to ``""`` (no real SHA-256 + # matches it, so the no-op short-circuit would be disabled + # for good) and never invent a hash the run did not earn. + roll_hash = prior_hash or pending_record.content_hash or None + if roll_hash is None: + ctx.log( + f"update: COMMITTED pending '{prior_pending_id}' carries no " + "content_hash (prior run reported incomplete writes); " + "promoting it uncertified so the document stays eligible for repair" + ) # Belt-and-braces: if the pending doesn't carry cleanup # state (e.g. it was committed by pre-fix code, or by a # test simulation that wrote the marker directly), snapshot @@ -2006,7 +2012,11 @@ async def update( the win for touch-only PRs (CRLF, formatter-only changes). Pass ``force=True`` to re-extract anyway — the only way to re-process unchanged content after changing the ontology, chunker, extractor - or model, since ``ingest()`` skips unchanged documents too. + or model, since ``ingest()`` skips unchanged documents too. The + hash is recorded only when the pipeline reported every write in + full; if ``UpdateResult.metadata["incomplete_writes"]`` is present + the document is promoted without one, so the next ``ingest()`` or + ``update()`` re-runs and repairs it instead of skipping. State-machine cutover (crash-safe). Columns: ``pend`` = pending Document exists; @@ -2282,11 +2292,31 @@ async def update( ) # ── Phase 5: rollforward cutover (idempotent) ── + # The cutover is what stamps ``content_hash`` on the canonical + # Document, so it must honour the same complete-writes gate as + # ``ingest()``: if the pipeline reported a shortfall (a dropped + # RELATES edge, an unembedded chunk — see + # ``IngestionPipeline.run``), promote the pending without a hash so + # the document stays eligible for repair instead of being skipped + # as unchanged forever (galshubeli on #309). + incomplete_writes = pipeline_result.metadata.get("incomplete_writes") + cutover_hash: str | None = None if incomplete_writes else new_hash + if incomplete_writes: + ctx.log( + "update: some writes were reported incomplete " + f"({'; '.join(incomplete_writes)}); content_hash not recorded — " + "the next ingest/update of this document re-runs in full" + ) + logger.warning( + "update of '%s' left incomplete writes (%s); not marking content_hash", + resolved_id, + "; ".join(incomplete_writes), + ) chunks_deleted = await self._graph_store.rollforward_cutover( pending_id=pending_id, real_id=resolved_id, path=doc_path, - content_hash=new_hash, + content_hash=cutover_hash, ) # ── Phase 6: unified post-cutover cleanup (recoverable) ── diff --git a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py index bde2aab6..e8e09dcf 100644 --- a/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py +++ b/graphrag_sdk/src/graphrag_sdk/ingestion/pipeline.py @@ -84,8 +84,10 @@ def _reported_short(reported: Any, expected: int) -> bool: not raise on per-item failures — they log, skip and return a count — so the count is the only signal that a relationship or an embedding is missing. Stores that return nothing (``None``, a mock) are taken at their - word: the pipeline cannot tell and must not refuse to ever mark a run - complete against a duck-typed store. + word: ``index_chunks`` returns ``None`` when no embedder is configured + (nothing was attempted, so nothing is missing), and the pipeline cannot + tell and must not refuse to ever mark a run complete against a + duck-typed store. """ return isinstance(reported, int) and not isinstance(reported, bool) and reported < expected @@ -400,6 +402,12 @@ async def run( # Step 7: Write to graph (batched) ctx.log("Step 7/9: Writing to graph store") + # ``upsert_nodes`` is deliberately not gated on its count: it + # raises ``DatabaseError`` on a real write failure, and the only + # nodes it drops silently are those whose id sanitises to empty. + # A re-run would drop those identically, so counting them would + # withhold ``content_hash`` for this document forever rather than + # flag something the next ingest can repair. await self.graph_store.upsert_nodes(resolved.nodes) rels_written = await self.graph_store.upsert_relationships(resolved.relationships) if _reported_short(rels_written, len(resolved.relationships)): diff --git a/graphrag_sdk/src/graphrag_sdk/storage/graph_store.py b/graphrag_sdk/src/graphrag_sdk/storage/graph_store.py index c2cef098..5af87981 100644 --- a/graphrag_sdk/src/graphrag_sdk/storage/graph_store.py +++ b/graphrag_sdk/src/graphrag_sdk/storage/graph_store.py @@ -565,11 +565,17 @@ async def rollforward_cutover( pending_id: str, real_id: str, path: str, - content_hash: str, + content_hash: str | None, ) -> int: """Replay the cutover from a (possibly partial) COMMITTED state to FINAL. Idempotent — every operation is safe to re-run. + ``content_hash=None`` promotes the pending *without* certifying it: + the canonical Document ends up with no ``content_hash`` (any hash the + pending carried is removed), so the next ``ingest()``/``update()`` of + that content re-runs in full instead of short-circuiting. Callers + pass ``None`` when the pipeline reported incomplete writes. + Sequence: 0. Precondition: pending_id must still exist. On a successful replay (steps 1-2 ran, step 3 didn't) the pending is still @@ -619,17 +625,24 @@ async def rollforward_cutover( # 3. Promote pending → canonical id and remove the commit marker. # ``REMOVE p.ready_to_commit`` is the idiomatic way to drop a # property; on a replay where the rename already happened this - # MATCH finds nothing and the whole statement is a no-op. + # MATCH finds nothing and the whole statement is a no-op. With no + # hash to certify, ``content_hash`` is removed rather than set so + # the promoted Document never inherits a stale one. + params: dict[str, Any] = { + "pending_id": pending_id, + "real_id": real_id, + "path": path, + } + set_clause = "SET p.id = $real_id, p.path = $path" + remove_clause = "REMOVE p.ready_to_commit" + if content_hash is None: + remove_clause += ", p.content_hash" + else: + set_clause += ", p.content_hash = $hash" + params["hash"] = content_hash await self._conn.query( - "MATCH (p:Document {id: $pending_id}) " - "SET p.id = $real_id, p.path = $path, p.content_hash = $hash " - "REMOVE p.ready_to_commit", - { - "pending_id": pending_id, - "real_id": real_id, - "path": path, - "hash": content_hash, - }, + f"MATCH (p:Document {{id: $pending_id}}) {set_clause} {remove_clause}", + params, ) return chunks_removed diff --git a/graphrag_sdk/src/graphrag_sdk/storage/vector_store.py b/graphrag_sdk/src/graphrag_sdk/storage/vector_store.py index 14ecd2d1..84c9131f 100644 --- a/graphrag_sdk/src/graphrag_sdk/storage/vector_store.py +++ b/graphrag_sdk/src/graphrag_sdk/storage/vector_store.py @@ -181,7 +181,7 @@ async def drop_relates_vector_index(self) -> None: # ── Indexing ───────────────────────────────────────────────── - async def index_chunks(self, chunks: TextChunks) -> int: + async def index_chunks(self, chunks: TextChunks) -> int | None: """Embed and store vectors for all chunks. Uses batch embedding (``aembed_documents``) for efficiency, @@ -191,11 +191,16 @@ async def index_chunks(self, chunks: TextChunks) -> int: chunks: TextChunks collection to embed and index. Returns: - Number of chunks indexed. + Number of chunks indexed, or ``None`` when no embedder is + configured and indexing was not attempted. ``0`` means every + embedding was attempted and failed; the ingestion pipeline + treats that as an incomplete write and withholds the + Document's ``content_hash``, whereas ``None`` (nothing to do) + leaves the run complete. """ if not self._embedder: logger.warning("No embedder configured — skipping chunk indexing") - return 0 + return None if not chunks.chunks: return 0 diff --git a/graphrag_sdk/tests/conftest.py b/graphrag_sdk/tests/conftest.py index 590c301b..742fd959 100644 --- a/graphrag_sdk/tests/conftest.py +++ b/graphrag_sdk/tests/conftest.py @@ -225,8 +225,8 @@ def mock_graph_store(mock_connection: MagicMock) -> MagicMock: # Report every item as written, like the real store does on success — # the pipeline treats a short count as a partial failure and withholds # the Document's content_hash. - store.upsert_nodes = AsyncMock(side_effect=lambda nodes: len(nodes)) - store.upsert_relationships = AsyncMock(side_effect=lambda rels: len(rels)) + store.upsert_nodes = AsyncMock(side_effect=len) + store.upsert_relationships = AsyncMock(side_effect=len) store.get_connected_entities = AsyncMock(return_value=[]) store.query_raw = AsyncMock(return_value=MagicMock(result_set=[])) store.delete_all = AsyncMock() diff --git a/graphrag_sdk/tests/test_facade.py b/graphrag_sdk/tests/test_facade.py index 2598be75..9352a5ba 100644 --- a/graphrag_sdk/tests/test_facade.py +++ b/graphrag_sdk/tests/test_facade.py @@ -1552,8 +1552,8 @@ async def _rollforward(*, pending_id, real_id, path, content_hash): g._graph_store.mark_document_pending_delete = AsyncMock(side_effect=_mark_delete) g._graph_store.has_pending_delete = AsyncMock(return_value=False) g._graph_store.rollforward_cutover = AsyncMock(side_effect=_rollforward) - g._graph_store.upsert_nodes = AsyncMock(side_effect=lambda nodes: len(nodes)) - g._graph_store.upsert_relationships = AsyncMock(side_effect=lambda rels: len(rels)) + g._graph_store.upsert_nodes = AsyncMock(side_effect=len) + g._graph_store.upsert_relationships = AsyncMock(side_effect=len) g._vector_store.ensure_indices = AsyncMock(return_value={}) g._vector_store.index_chunks = AsyncMock(side_effect=lambda chunks: len(chunks.chunks)) g._vector_store.backfill_entity_embeddings = AsyncMock(return_value=0) @@ -1674,6 +1674,73 @@ def test_update_sync_forwards_force(self, graphrag): assert "force" in inspect.signature(graphrag.update_sync).parameters assert "force" in inspect.signature(graphrag.update).parameters + async def test_complete_writes_record_content_hash_on_cutover(self, graphrag): + """The cutover is where ``update()`` stamps the hash; a run with every + write reported in full certifies the document as complete.""" + import hashlib + + text = "Fresh content, fully written." + _stub_graph_store_for_update( + graphrag, existing_record={"path": "my-doc", "content_hash": "old-hash"} + ) + + result = await graphrag.update(text=text, document_id="my-doc") + + assert "incomplete_writes" not in result.metadata + kwargs = graphrag._graph_store.rollforward_cutover.await_args.kwargs + assert kwargs["content_hash"] == hashlib.sha256(text.encode("utf-8")).hexdigest() + + async def test_incomplete_writes_promote_without_content_hash(self, graphrag): + """``update()`` must honour the same complete-writes gate as + ``ingest()``: a shortfall reported by the pipeline (here: no chunk got + an embedding) promotes the pending *without* a hash, so the next + ingest/update repairs the document instead of skipping it as unchanged + (galshubeli on #309).""" + _stub_graph_store_for_update( + graphrag, existing_record={"path": "my-doc", "content_hash": "old-hash"} + ) + graphrag._vector_store.index_chunks = AsyncMock(return_value=0) + + result = await graphrag.update(text="Fresh content, half written.", document_id="my-doc") + + assert result.no_op is False + assert result.replaced_existing is True + assert any(s.startswith("chunks indexed 0/") for s in result.metadata["incomplete_writes"]) + # The cutover still happens — crash-safety is untouched — but uncertified. + graphrag._graph_store.rollforward_cutover.assert_awaited_once() + assert graphrag._graph_store.rollforward_cutover.await_args.kwargs["content_hash"] is None + + async def test_phase0_rolls_forward_committed_pending_without_hash(self, graphrag): + """A COMMITTED pending with a path but no ``content_hash`` is what an + interrupted update() with incomplete writes leaves behind — not + corruption. Phase 0 must replay it uncertified (``content_hash=None``), + never refuse, and never invent a hash.""" + from graphrag_sdk.core.models import DocumentRecord + + pending_id = "my-doc__pending__abc12345" + _stub_graph_store_for_update( + graphrag, + existing_record={"path": "my-doc", "content_hash": "old-hash"}, + prior_pending=("COMMITTED", pending_id, None), + ) + records = { + pending_id: DocumentRecord(path="my-doc", content_hash=None), + "my-doc": DocumentRecord(path="my-doc", content_hash="old-hash"), + } + graphrag._graph_store.get_document_record = AsyncMock( + side_effect=lambda doc_id: records.get(doc_id) + ) + + await graphrag.update(text="new", document_id="my-doc") + + # First cutover is the Phase 0 replay, second is this update's own. + assert graphrag._graph_store.rollforward_cutover.await_count == 2 + replay = graphrag._graph_store.rollforward_cutover.await_args_list[0].kwargs + assert replay["pending_id"] == pending_id + assert replay["real_id"] == "my-doc" + assert replay["path"] == "my-doc" + assert replay["content_hash"] is None + async def test_doc_not_found_default_raises(self, graphrag): """if_missing='error' (default) raises DocumentNotFoundError.""" _stub_graph_store_for_update(graphrag, existing_record=None) diff --git a/graphrag_sdk/tests/test_graph_store.py b/graphrag_sdk/tests/test_graph_store.py index 18e3ff64..f819973d 100644 --- a/graphrag_sdk/tests/test_graph_store.py +++ b/graphrag_sdk/tests/test_graph_store.py @@ -485,6 +485,35 @@ async def test_rollforward_cutover_runs_precondition_then_three_idempotent_ops( # FINAL state would still report as a pending Document. assert "REMOVE p.ready_to_commit" in rename_cypher + async def test_rollforward_without_hash_removes_content_hash( + self, graph_store, mock_connection + ): + """``content_hash=None`` promotes the pending uncertified: the hash + is REMOVEd (never set to ``""``/null-ish) so the canonical Document + stays eligible for repair on the next ingest/update.""" + results = [ + MagicMock(result_set=[[1]]), + MagicMock(result_set=[[2]]), + MagicMock(result_set=[]), + MagicMock(result_set=[]), + ] + mock_connection.query = AsyncMock(side_effect=results) + + chunks_removed = await graph_store.rollforward_cutover( + pending_id="docs/a.md__pending__abc12345", + real_id="docs/a.md", + path="docs/a.md", + content_hash=None, + ) + assert chunks_removed == 2 + + rename_cypher, rename_params = mock_connection.query.await_args_list[3][0] + assert "SET p.id = $real_id" in rename_cypher + assert "p.path = $path" in rename_cypher + assert "p.content_hash = $hash" not in rename_cypher + assert "REMOVE p.ready_to_commit, p.content_hash" in rename_cypher + assert "hash" not in rename_params + async def test_rollforward_aborts_if_pending_missing( self, graph_store, mock_connection ): diff --git a/graphrag_sdk/tests/test_pipeline.py b/graphrag_sdk/tests/test_pipeline.py index 5cceda9d..aed1e3f3 100644 --- a/graphrag_sdk/tests/test_pipeline.py +++ b/graphrag_sdk/tests/test_pipeline.py @@ -1098,8 +1098,9 @@ def _drop_one_mention(rels): async def test_unembedded_chunks_withhold_the_hash( self, ctx, mock_graph_store, mock_vector_store ): - """0 from ``index_chunks`` — every embedding call failed, or no - embedder — means the document is absent from chunk vector search.""" + """0 from ``index_chunks`` — every embedding call failed — means the + document is absent from chunk vector search. (No embedder at all is + ``None``, covered by ``test_no_embedder_is_not_a_shortfall``.)""" mock_vector_store.index_chunks = AsyncMock(return_value=0) result = await self._pipeline(mock_graph_store, mock_vector_store).run("r.txt", ctx) @@ -1138,6 +1139,23 @@ async def test_next_ingest_after_a_short_run_takes_the_full_path( assert "skipped_unchanged" not in result.metadata assert self._hash_written(mock_graph_store) + + async def test_no_embedder_is_not_a_shortfall(self, ctx, mock_graph_store, mock_connection): + """Graph-only deployments: ``VectorStore.index_chunks`` returns ``None`` + when no embedder is configured — nothing was attempted, so nothing is + missing — and the hash must still be recorded or the unchanged-document + skip would never fire for them (galshubeli on #309).""" + from graphrag_sdk.storage.vector_store import VectorStore + + store = VectorStore(mock_connection, embedder=None) + result = await self._pipeline(mock_graph_store, store).run("r.txt", ctx) + + assert self._hash_written(mock_graph_store) + assert "incomplete_writes" not in result.metadata + assert result.chunks_indexed == 3 + + +class TestDeterministicChunkUids: """Bug #12 — re-ingesting an unchanged document duplicated its chunks. ``TextChunk.uid`` defaulted to ``uuid4()``, so the lexical graph's diff --git a/graphrag_sdk/tests/test_vector_store.py b/graphrag_sdk/tests/test_vector_store.py index d52721a9..53a974c0 100644 --- a/graphrag_sdk/tests/test_vector_store.py +++ b/graphrag_sdk/tests/test_vector_store.py @@ -135,10 +135,13 @@ async def test_index_chunks_uses_unwind(self, vector_store, mock_connection, emb params = mock_connection.query.call_args[0][1] assert len(params["batch"]) == 2 - async def test_index_chunks_no_embedder(self, vector_store_no_embedder): + async def test_index_chunks_no_embedder(self, vector_store_no_embedder, mock_connection): + """No embedder → ``None`` (not attempted), distinct from ``0`` (all + failed) so the ingestion pipeline does not read it as a shortfall.""" chunks = TextChunks(chunks=[TextChunk(text="Hi", index=0)]) result = await vector_store_no_embedder.index_chunks(chunks) - assert result == 0 + assert result is None + mock_connection.query.assert_not_awaited() async def test_index_chunks_batch_fallback(self, vector_store, mock_connection, embedder): """When UNWIND batch fails, should fall back to individual queries."""