mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-10-02 02:13:46 +00:00
fix(index): invalidate chunks when chunkers change
This commit is contained in:
parent
1f67a6ce29
commit
1ec6536b32
9 changed files with 182 additions and 6 deletions
|
|
@ -1,6 +1,8 @@
|
|||
"""Abstract base for file chunkers."""
|
||||
|
||||
from abc import abstractmethod
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from ..base_component import BaseComponent
|
||||
|
|
@ -17,6 +19,20 @@ class BaseFileChunker(BaseComponent):
|
|||
super().__init__(**kwargs)
|
||||
self.supported_extensions: list[str] = supported_extensions or []
|
||||
|
||||
def config_fingerprint(self) -> str:
|
||||
"""Return a stable fingerprint for the chunker's derived output shape."""
|
||||
payload = self._config_fingerprint_payload()
|
||||
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
||||
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
def _config_fingerprint_payload(self) -> dict:
|
||||
"""Return the structured state that should invalidate persisted chunks."""
|
||||
return {
|
||||
"type": type(self).__name__,
|
||||
"backend": self.backend,
|
||||
"supported_extensions": list(self.supported_extensions),
|
||||
}
|
||||
|
||||
@abstractmethod
|
||||
async def chunk(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]:
|
||||
"""Chunk a file into (node, chunks)."""
|
||||
|
|
|
|||
|
|
@ -38,6 +38,18 @@ class DefaultFileChunker(BaseFileChunker):
|
|||
self.chunk_byte_size = max(100, chunk_byte_size)
|
||||
self.overlap_byte_size = max(4, overlap_byte_size)
|
||||
|
||||
def _config_fingerprint_payload(self) -> dict:
|
||||
payload = super()._config_fingerprint_payload()
|
||||
payload.update(
|
||||
{
|
||||
"encoding": self.encoding,
|
||||
"invalid_encoding_policy": self.invalid_encoding_policy,
|
||||
"chunk_byte_size": self.chunk_byte_size,
|
||||
"overlap_byte_size": self.overlap_byte_size,
|
||||
},
|
||||
)
|
||||
return payload
|
||||
|
||||
@staticmethod
|
||||
def _parse_front_matter(text: str) -> tuple[FileFrontMatter, str]:
|
||||
"""Parse YAML front matter delimited by ---, return (front_matter, remaining)."""
|
||||
|
|
|
|||
|
|
@ -34,6 +34,17 @@ class JsonFileChunker(BaseFileChunker):
|
|||
self.chunk_chars = max(256, chunk_chars)
|
||||
self.min_element_size = max(64, int(self.chunk_chars * 0.05))
|
||||
|
||||
def _config_fingerprint_payload(self) -> dict:
|
||||
payload = super()._config_fingerprint_payload()
|
||||
payload.update(
|
||||
{
|
||||
"encoding": self.encoding,
|
||||
"chunk_chars": self.chunk_chars,
|
||||
"min_element_size": self.min_element_size,
|
||||
},
|
||||
)
|
||||
return payload
|
||||
|
||||
class Node:
|
||||
"""AST node for JSON tree construction."""
|
||||
|
||||
|
|
|
|||
|
|
@ -46,6 +46,19 @@ class JsonlFileChunker(BaseFileChunker):
|
|||
self.max_lines_per_chunk = max(1, max_lines_per_chunk) if max_lines_per_chunk is not None else None
|
||||
self.mode = mode
|
||||
|
||||
def _config_fingerprint_payload(self) -> dict:
|
||||
payload = super()._config_fingerprint_payload()
|
||||
payload.update(
|
||||
{
|
||||
"encoding": self.encoding,
|
||||
"max_chars": self.max_chars,
|
||||
"max_overlap_chars": self.max_overlap_chars,
|
||||
"max_lines_per_chunk": self.max_lines_per_chunk,
|
||||
"mode": self.mode,
|
||||
},
|
||||
)
|
||||
return payload
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Size helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -123,6 +123,20 @@ class MarkdownFileChunker(DefaultFileChunker):
|
|||
self.include_frontmatter_in_metadata = include_frontmatter_in_metadata
|
||||
self.include_frontmatter_keys_in_metadata = list(include_frontmatter_keys_in_metadata or [])
|
||||
|
||||
def _config_fingerprint_payload(self) -> dict:
|
||||
payload = super()._config_fingerprint_payload()
|
||||
payload.update(
|
||||
{
|
||||
"encoding": self.encoding,
|
||||
"chunk_byte_size": self.chunk_byte_size,
|
||||
"embed_toc": self.embed_toc,
|
||||
"max_ast_sections": self.max_ast_sections,
|
||||
"include_frontmatter_in_metadata": self.include_frontmatter_in_metadata,
|
||||
"include_frontmatter_keys_in_metadata": list(self.include_frontmatter_keys_in_metadata),
|
||||
},
|
||||
)
|
||||
return payload
|
||||
|
||||
async def chunk(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]:
|
||||
file_path = Path(path)
|
||||
rel_path = self.to_workspace_relative(path)
|
||||
|
|
|
|||
|
|
@ -83,8 +83,6 @@ class FaissLocalFileStore(LocalFileStore):
|
|||
self.hnsw_m = hnsw_m
|
||||
self.hnsw_ef_construction = hnsw_ef_construction
|
||||
self.async_reindex = async_reindex
|
||||
self.faiss_path = self.component_metadata_path / f"faiss_index_{self.name}_{self.store_version}.bin"
|
||||
self.faiss_idmap_path = self.component_metadata_path / f"faiss_idmap_{self.name}_{self.store_version}.json"
|
||||
self._faiss_index = None # faiss.Index | None
|
||||
self._id_map: list[str] = [] # row -> chunk_id
|
||||
self._id_to_row: dict[str, int] = {} # chunk_id -> row (live entries only)
|
||||
|
|
@ -100,6 +98,14 @@ class FaissLocalFileStore(LocalFileStore):
|
|||
self._index_writes = 0 # bumped on every index mutation; used to re-arm the flag
|
||||
self._closing = False # set during _close() to stop spawning background reindexes
|
||||
|
||||
@property
|
||||
def faiss_path(self):
|
||||
return self.component_metadata_path / f"faiss_index_{self.name}_{self._store_version_suffix()}.bin"
|
||||
|
||||
@property
|
||||
def faiss_idmap_path(self):
|
||||
return self.component_metadata_path / f"faiss_idmap_{self.name}_{self._store_version_suffix()}.json"
|
||||
|
||||
@staticmethod
|
||||
def _import_faiss():
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -3,11 +3,13 @@
|
|||
import asyncio
|
||||
import base64
|
||||
import datetime
|
||||
import hashlib
|
||||
import heapq
|
||||
import json
|
||||
import time
|
||||
from collections.abc import Iterable
|
||||
from contextlib import suppress
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
|
@ -17,7 +19,7 @@ from ..embedding_store import BaseEmbeddingStore
|
|||
from ..file_graph import BaseFileGraph
|
||||
from ..keyword_index import BaseKeywordIndex
|
||||
from ..tag_index import BaseTagIndex
|
||||
from ...enumeration import LinkScopeEnum
|
||||
from ...enumeration import ComponentEnum, LinkScopeEnum
|
||||
from ...schema import FileChunk, FileLink, FileNode
|
||||
from ...utils import batch_cosine_similarity
|
||||
from ...utils.async_utils import complete_in_thread
|
||||
|
|
@ -71,7 +73,6 @@ class LocalFileStore(BaseFileStore):
|
|||
self.encoding = encoding
|
||||
self.store_version = store_version
|
||||
self.file_chunks: dict[str, FileChunk] = {}
|
||||
self.chunks_path = self.component_metadata_path / f"file_chunks_{self.name}_{self.store_version}.jsonl.zst"
|
||||
self._embedding_backfill_task: asyncio.Task | None = None
|
||||
self._embedding_backfill_pending = False
|
||||
self._embedding_rebuild_pending = bool(embedding_rebuild_required)
|
||||
|
|
@ -81,6 +82,32 @@ class LocalFileStore(BaseFileStore):
|
|||
self._tag_indexed_file_count = 0
|
||||
self._closing = False
|
||||
|
||||
def _store_version_suffix(self) -> str:
|
||||
"""Return the on-disk version suffix for the active chunker configuration."""
|
||||
fingerprint = self._chunker_fingerprint()
|
||||
return f"{self.store_version}_{fingerprint}" if fingerprint else self.store_version
|
||||
|
||||
def _chunker_fingerprint(self) -> str:
|
||||
"""Fingerprint the bound chunker configuration so derived chunks rotate on change."""
|
||||
if self.app_context is None:
|
||||
return ""
|
||||
chunkers = self.app_context.components.get(ComponentEnum.FILE_CHUNKER, {})
|
||||
if not chunkers:
|
||||
return ""
|
||||
payload = {
|
||||
name: chunker.config_fingerprint()
|
||||
for name, chunker in sorted(chunkers.items())
|
||||
if hasattr(chunker, "config_fingerprint")
|
||||
}
|
||||
if not payload:
|
||||
return ""
|
||||
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
||||
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
@property
|
||||
def chunks_path(self) -> Path:
|
||||
return self.component_metadata_path / f"file_chunks_{self.name}_{self._store_version_suffix()}.jsonl.zst"
|
||||
|
||||
# -- lifecycle ------------------------------------------------------------
|
||||
|
||||
async def _start(self) -> None:
|
||||
|
|
|
|||
|
|
@ -64,11 +64,17 @@ class ZvecLocalFileStore(LocalFileStore):
|
|||
self._zvec = self._import_zvec()
|
||||
self.hnsw_m = hnsw_m
|
||||
self.hnsw_ef_construction = hnsw_ef_construction
|
||||
self.zvec_path = self.component_metadata_path / f"zvec_index_{self.name}_{self.store_version}"
|
||||
self.zvec_sidecar_path = self.component_metadata_path / f"zvec_sidecar_{self.name}_{self.store_version}.json"
|
||||
self._collection = None # zvec.Collection | None
|
||||
self._indexed_ids: set[str] = set() # chunk ids currently present in the collection
|
||||
|
||||
@property
|
||||
def zvec_path(self):
|
||||
return self.component_metadata_path / f"zvec_index_{self.name}_{self._store_version_suffix()}"
|
||||
|
||||
@property
|
||||
def zvec_sidecar_path(self):
|
||||
return self.component_metadata_path / f"zvec_sidecar_{self.name}_{self._store_version_suffix()}.json"
|
||||
|
||||
@staticmethod
|
||||
def _import_zvec():
|
||||
try:
|
||||
|
|
|
|||
71
tests/unit/test_chunk_store_fingerprint.py
Normal file
71
tests/unit/test_chunk_store_fingerprint.py
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
"""Regression tests for chunker-sensitive file-store persistence."""
|
||||
|
||||
import asyncio
|
||||
|
||||
from reme.components.application_context import ApplicationContext
|
||||
from reme.components.file_chunker import MarkdownFileChunker
|
||||
from reme.components.file_store import LocalFileStore
|
||||
from reme.enumeration import ComponentEnum
|
||||
from reme.schema import FileChunk, FileNode
|
||||
from reme.utils.jsonl_zst import write_jsonl_zst
|
||||
|
||||
|
||||
def _store_for_chunker(workspace, *, chunk_byte_size: int) -> LocalFileStore:
|
||||
context = ApplicationContext(workspace_dir=str(workspace))
|
||||
chunker = MarkdownFileChunker(chunk_byte_size=chunk_byte_size, app_context=context)
|
||||
context.components[ComponentEnum.FILE_CHUNKER] = {"default": chunker}
|
||||
return LocalFileStore(
|
||||
name="default",
|
||||
embedding_store="",
|
||||
app_context=context,
|
||||
)
|
||||
|
||||
|
||||
def test_chunk_store_path_changes_when_chunker_configuration_changes(tmp_path):
|
||||
first = _store_for_chunker(tmp_path, chunk_byte_size=1000)
|
||||
second = _store_for_chunker(tmp_path, chunk_byte_size=2000)
|
||||
|
||||
assert first.chunks_path != second.chunks_path
|
||||
assert first.chunks_path.name.startswith("file_chunks_default_v1_")
|
||||
assert second.chunks_path.name.startswith("file_chunks_default_v1_")
|
||||
|
||||
|
||||
def test_chunk_store_path_is_stable_for_equivalent_chunker_configuration(tmp_path):
|
||||
first = _store_for_chunker(tmp_path, chunk_byte_size=1000)
|
||||
second = _store_for_chunker(tmp_path, chunk_byte_size=1000)
|
||||
|
||||
assert first.chunks_path == second.chunks_path
|
||||
|
||||
|
||||
def test_changed_chunker_does_not_load_legacy_chunks_and_clears_stale_graph(tmp_path):
|
||||
class Graph:
|
||||
def __init__(self):
|
||||
self.nodes = [FileNode(path="note.md", st_mtime=1, chunk_ids=["legacy"])]
|
||||
self.cleared = False
|
||||
|
||||
async def get_nodes(self):
|
||||
return self.nodes
|
||||
|
||||
async def clear(self):
|
||||
self.cleared = True
|
||||
self.nodes = []
|
||||
|
||||
async def run():
|
||||
store = _store_for_chunker(tmp_path, chunk_byte_size=2000)
|
||||
legacy_path = store.component_metadata_path / "file_chunks_default_v1.jsonl.zst"
|
||||
legacy_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
write_jsonl_zst(legacy_path, [FileChunk(id="legacy", path="note.md", text="old chunk").model_dump_json()])
|
||||
|
||||
graph = Graph()
|
||||
store.file_graph = graph
|
||||
store.keyword_index = None
|
||||
store.tag_index = None
|
||||
|
||||
assert legacy_path != store.chunks_path
|
||||
await store.load()
|
||||
|
||||
assert store.file_chunks == {}
|
||||
assert graph.cleared is True
|
||||
assert legacy_path.exists()
|
||||
|
||||
asyncio.run(run())
|
||||
Loading…
Add table
Reference in a new issue