Skip to content

Fix incremental indexing reprocessing and document state handling #112

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
huytg2610 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix incremental indexing reprocessing and document state handling #112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
467f5b5
fix: make incremental indexing idempotent
huytg2610 Jul 16, 2026
c0b1eaf
fix: normalise Unicode to NFC before hashing text into ids
huytg2610 Jul 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 106 additions & 56 deletions minirag/minirag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@
StorageNameSpace,
QueryParam,
DocStatus,
DocProcessingStatus,
TextChunkSchema,
)


Expand Down Expand Up @@ -353,40 +355,55 @@ async def ainsert(
if isinstance(ids, str):
ids = [ids]

await self.apipeline_enqueue_documents(input, ids)
await self.apipeline_process_enqueue_documents(
split_by_character, split_by_character_only
)

# Perform additional entity extraction as per original ainsert logic
inserting_chunks = {
compute_mdhash_id(dp["content"], prefix="chunk-"): {
**dp,
"full_doc_id": doc_id,
}
for doc_id, status_doc in (
await self.doc_status.get_docs_by_status(DocStatus.PROCESSED)
).items()
for dp in self.chunking_func(
status_doc.content,
self.chunk_overlap_token_size,
self.chunk_token_size,
self.tiktoken_model_name,
extraction_failures = []
try:
await self.apipeline_enqueue_documents(input, ids)
documents_to_extract = await self.apipeline_process_enqueue_documents(
split_by_character, split_by_character_only
)
}

if inserting_chunks:
logger.info("Performing entity extraction on newly processed chunks")
await extract_entities(
inserting_chunks,
knowledge_graph_inst=self.chunk_entity_relation_graph,
entity_vdb=self.entities_vdb,
entity_name_vdb=self.entity_name_vdb,
relationships_vdb=self.relationships_vdb,
global_config=asdict(self),
)

await self._insert_done()
for doc_id, (status_doc, inserting_chunks) in (
documents_to_extract.items()
):
try:
logger.info(
"Performing entity extraction for %s newly processed chunks",
len(inserting_chunks),
)
await extract_entities(
inserting_chunks,
knowledge_graph_inst=self.chunk_entity_relation_graph,
entity_vdb=self.entities_vdb,
entity_name_vdb=self.entity_name_vdb,
relationships_vdb=self.relationships_vdb,
global_config=asdict(self),
)
except Exception as error:
await self._update_document_status(
doc_id,
status_doc,
DocStatus.FAILED,
len(inserting_chunks),
error=str(error),
)
extraction_failures.append((doc_id, error))
logger.exception("Entity extraction failed for document %s", doc_id)
continue

await self._update_document_status(
doc_id,
status_doc,
DocStatus.PROCESSED,
len(inserting_chunks),
)
finally:
await self._insert_done()

if extraction_failures:
failed_doc_ids = ", ".join(doc_id for doc_id, _ in extraction_failures)
raise RuntimeError(
f"Entity extraction failed for document(s): {failed_doc_ids}"
) from extraction_failures[0][1]

async def apipeline_enqueue_documents(
self, input: str | list[str], ids: list[str] | None = None
Expand Down Expand Up @@ -448,15 +465,39 @@ async def apipeline_enqueue_documents(
await self.doc_status.upsert(new_docs)
logger.info(f"Stored {len(new_docs)} new unique documents")

async def _update_document_status(
self,
doc_id: str,
status_doc: DocProcessingStatus,
status: DocStatus,
chunks_count: int,
error: str | None = None,
) -> None:
status_data = {
"status": status,
"chunks_count": chunks_count,
"content": status_doc.content,
"content_summary": status_doc.content_summary,
"content_length": status_doc.content_length,
"created_at": status_doc.created_at,
"updated_at": datetime.now().isoformat(),
}
if error is not None:
status_data["error"] = error
await self.doc_status.upsert({doc_id: status_data})

async def apipeline_process_enqueue_documents(
self,
split_by_character: str | None = None,
split_by_character_only: bool = False,
) -> None:
) -> dict[str, tuple[DocProcessingStatus, dict[str, TextChunkSchema]]]:
"""
Process pending documents by splitting them into chunks, processing
each chunk for entity and relation extraction, and updating the
document status.
Stage pending documents for graph extraction.

Documents remain in PROCESSING until entity and relation extraction
completes in ``ainsert``. The returned mapping contains only chunks
staged during this invocation, so already processed documents are not
extracted again.
"""
processing_docs, failed_docs, pending_docs = await asyncio.gather(
self.doc_status.get_docs_by_status(DocStatus.PROCESSING),
Expand All @@ -471,13 +512,14 @@ async def apipeline_process_enqueue_documents(
}
if not to_process_docs:
logger.info("No documents to process")
return
return {}

docs_batches = [
list(to_process_docs.items())[i : i + self.max_parallel_insert]
for i in range(0, len(to_process_docs), self.max_parallel_insert)
]
logger.info(f"Number of batches to process: {len(docs_batches)}")
documents_to_extract = {}

for batch_idx, docs_batch in enumerate(docs_batches):
for doc_id, status_doc in docs_batch:
Expand All @@ -493,25 +535,33 @@ async def apipeline_process_enqueue_documents(
self.tiktoken_model_name,
)
}
await asyncio.gather(
self.chunks_vdb.upsert(chunks),
self.full_docs.upsert({doc_id: {"content": status_doc.content}}),
self.text_chunks.upsert(chunks),
await self._update_document_status(
doc_id,
status_doc,
DocStatus.PROCESSING,
len(chunks),
)
await self.doc_status.upsert(
{
doc_id: {
"status": DocStatus.PROCESSED,
"chunks_count": len(chunks),
"content": status_doc.content,
"content_summary": status_doc.content_summary,
"content_length": status_doc.content_length,
"created_at": status_doc.created_at,
"updated_at": datetime.now().isoformat(),
}
}
)
logger.info("Document processing pipeline completed")
try:
await asyncio.gather(
self.chunks_vdb.upsert(chunks),
self.full_docs.upsert(
{doc_id: {"content": status_doc.content}}
),
self.text_chunks.upsert(chunks),
)
except Exception as error:
await self._update_document_status(
doc_id,
status_doc,
DocStatus.FAILED,
len(chunks),
error=str(error),
)
raise
documents_to_extract[doc_id] = (status_doc, chunks)

logger.info("Document staging pipeline completed")
return documents_to_extract

async def _insert_done(self):
tasks = []
Expand Down Expand Up @@ -608,4 +658,4 @@ async def _delete_by_entity_done(self):
if storage_inst is None:
continue
tasks.append(cast(StorageNameSpace, storage_inst).index_done_callback())
await asyncio.gather(*tasks)
await asyncio.gather(*tasks)
22 changes: 19 additions & 3 deletions minirag/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import logging
import os
import re
import unicodedata
from dataclasses import dataclass
from functools import wraps
from hashlib import md5
Expand Down Expand Up @@ -64,8 +65,15 @@ def compute_args_hash(*args, cache_type: str | None = None) -> str:


def clean_text(text: str) -> str:
"""Clean text by removing null bytes (0x00) and whitespace"""
return text.strip().replace("\x00", "")
"""Clean text by removing null bytes (0x00) and whitespace, and normalising
its Unicode composition.

See :func:`clean_str` for why the NFC pass matters. Here it also decides
document identity: ``doc_id`` is an md5 of this function's output, so
without it the same document typed on one keyboard and copied from another
hashes differently and is indexed twice.
"""
return unicodedata.normalize("NFC", text.strip().replace("\x00", ""))


def get_content_summary(content: str, max_length: int = 100) -> str:
Expand Down Expand Up @@ -179,7 +187,15 @@ def clean_str(input: Any) -> str:

result = html.unescape(input.strip())
# https://stackoverflow.com/questions/4324790/removing-control-characters-from-a-string-in-python
return re.sub(r"[\x00-\x1f\x7f-\x9f]", "", result)
result = re.sub(r"[\x00-\x1f\x7f-\x9f]", "", result)
# Accented scripts have more than one valid encoding of the same text: "ệ"
# is either U+1EC7 or "e" + U+0323 + U+0302. Both render identically, so the
# difference is invisible, but the strings compare unequal and hash
# differently — and these strings become identity keys (graph node ids and
# the md5 behind every `ent-` vector id). Vietnamese hits this constantly
# because sources disagree: IMEs emit NFC, the macOS filesystem emits NFD.
# Pure-ASCII text is unaffected, which is why this went unnoticed.
return unicodedata.normalize("NFC", result)


def is_float_regex(value):
Expand Down
Loading