DocumentVersionManager, DocVersion, ContentDiff

core.document_versioning.DocumentVersionManager and supporting dataclasses

Overview

DocumentVersionManager tracks every version of every document ingested into a RAG system and performs diff-based re-indexing: content hashing determines whether a document actually changed before triggering an expensive remove-and-reindex cycle, avoiding unnecessary vector-DB churn for no-op updates.

[!NOTE] Availability and coupling Like evaluation, this module is imported into core's top-level namespace inside a try/except ImportError guard. Unlike every other module in the package, add_or_update() performs a deferred, function-local import of LoadedDocument using the absolute path zerda.rag.core.rag_system (not a relative from .rag_system import ...). This means DocumentVersionManager.add_or_update() will raise ImportError if core is used as a standalone top-level package rather than installed as part of the zerda.rag namespace — see Architecture.

Data classes

DocVersion

@dataclass
class DocVersion:
    doc_id:       str
    version:      int
    content_hash: str
    chunk_hashes: List[str]
    created_at:   float = field(default_factory=time.time)
    metadata:     Dict[str, Any] = field(default_factory=dict)
    text_preview: str = ""

An immutable snapshot of a document at a specific version number. version starts at 1 and increments per document on each content-changing update.

Methods

• to_dict() -> Dict[str, Any] — full field dump for JSON persistence.
• from_dict(d) -> DocVersion (classmethod) — reconstructs via cls(**d).

ContentDiff

@dataclass
class ContentDiff:
    doc_id:          str
    old_version:     int
    new_version:     int
    is_new:          bool
    content_changed: bool
    added_chunks:    List[int]
    removed_chunks:  List[int]
    unchanged_count: int
    change_ratio:    float

The result of comparing two DocVersions.

__repr__

• New document: ContentDiff('{doc_id}': NEW document, {n} chunks)
• Unchanged: ContentDiff('{doc_id}': UNCHANGED)
• Changed: ContentDiff('{doc_id}': v{old}→v{new} | +{added} -{removed} ={unchanged} [{ratio:.0%} changed])

Internal chunking for diff purposes

_chunk_texts (module-level)

def _chunk_texts(text: str, chunk_size: int = 512, overlap: int = 128) -> List[str]

[!WARNING] This is not the RAG system's real chunker This function exists solely to hash text for change-detection purposes and is completely independent of whatever chunker object you pass to RAGSystem. It uses a simple sentence-boundary-aware split (looking backward from end for . followed by \n, Arabic comma ،, Arabic question mark ؟, or \n again) with the given chunk_size/overlap. Chunk boundaries computed here will generally not match the boundaries your real chunker produces, so added_chunks/removed_chunks counts describe changes at the granularity of this internal hasher, not your production chunker.

_sha256 (module-level)

hashlib.sha256(text.encode("utf-8")).hexdigest().

DocumentVersionManager

Exact signature

class DocumentVersionManager:
    def __init__(
        self,
        rag_system: Any,
        persist_path: Optional[str] = None,
        chunk_size: int = 512,
        chunk_overlap: int = 128,
    ) -> None: ...

Parameters

Initialization flow

1. Stores rag_system, chunk_size, chunk_overlap.
2. If persist_path given, creates the directory (mkdir(parents=True, exist_ok=True)) and converts to a Path.
3. Initializes an empty in-memory _store: Dict[str, List[DocVersion]].
4. Calls _load_from_disk() to hydrate _store from any existing *.json files under persist_path (skips and logs a warning on any corrupt/unreadable file rather than failing construction).

Diff algorithm

_make_version (internal)

Chunks the text via _chunk_texts, SHA-256-hashes each chunk and the full text, and determines the next version number (last_version + 1, or 1 if this is the first version for doc_id).

_diff (internal)

def _diff(self, old: Optional[DocVersion], new: DocVersion) -> ContentDiff

Three cases:

1. old is None (brand-new document): is_new=True, content_changed=True, added_chunks = range(len(new.chunk_hashes)), removed_chunks = [], change_ratio = 1.0.
2. old.content_hash == new.content_hash (byte-identical full text): content_changed=False, no added/removed chunks, change_ratio = 0.0. This is a full-text hash comparison, not a chunk-level one — even if chunk boundaries would differ, identical full text short-circuits to "unchanged".
3. Otherwise (content differs): computes chunk-hash set differences —
   • added_chunks = indices of new chunks whose hash isn't in the old hash set.
   • removed_chunks = indices of old chunks whose hash isn't in the new hash set.
   • unchanged_count = |old_hashes ∩ new_hashes|.
   • change_ratio = (len(added) + len(removed)) / max(len(old), len(new), 1), rounded to 4 decimals.

add_or_update

def add_or_update(self, doc_id: str, text: str, metadata: Optional[Dict[str, Any]] = None) -> ContentDiff

Parameters

Returns

ContentDiff describing what changed (or didn't).

Algorithm

1. Look up the current latest DocVersion for doc_id (or None if new).
2. Build the candidate new version (_make_version) and compute the ContentDiff (_diff).
3. Log the diff (via its __repr__).
4. If content is unchanged (diff.content_changed is False): log a debug "skipping re-index" line and return the diff without touching the RAG system or the version store at all.
5. If not new, attempt self.rag.remove_document(doc_id) — failures here are caught, logged as a warning, and do not abort the update (the code proceeds to re-index anyway).
6. Re-index: dynamically import LoadedDocument (see availability note above), wrap text in a LoadedDocument(page_content=text, doc_id=doc_id, metadata=metadata), and call self.rag.add_documents([doc]). Any exception here is logged and re-raised — this step, unlike the removal step, is not tolerant of failure.
7. Append the new DocVersion to the in-memory store and persist it to disk (if persist_path was set).
8. Return the diff.

Raises

Side effects

• Calls rag_system.remove_document() and rag_system.add_documents().
• Mutates the in-memory version store.
• Writes a JSON file to persist_path (if configured).
• Logs extensively at INFO/DEBUG/WARNING/ERROR.

Complexity

O(text length) for hashing/chunking, plus whatever rag_system.remove_document/add_documents cost.

Examples

Basic — first ingest:

from zerda.rag.core import DocumentVersionManager

manager = DocumentVersionManager(rag_system=rag, persist_path="./versions")
diff = manager.add_or_update("policy_v1", "Smoking is prohibited in all buildings.")
print(diff)   # ContentDiff('policy_v1': NEW document, 1 chunks)

Update — content genuinely changed:

diff = manager.add_or_update(
    "policy_v1",
    "Smoking is prohibited in all buildings and gardens.",
)
print(diff)   # ContentDiff('policy_v1': v1→v2 | +1 -1 =0 [100% changed])

No-op update — identical text:

diff = manager.add_or_update("policy_v1", "Smoking is prohibited in all buildings and gardens.")
print(diff)   # ContentDiff('policy_v1': UNCHANGED)
# rag.add_documents() is NOT called in this case.

Error handling — re-index failure propagates:

class BrokenRAG:
    def remove_document(self, doc_id): return 0
    def add_documents(self, docs): raise ConnectionError("vector db down")

manager = DocumentVersionManager(rag_system=BrokenRAG())
try:
    manager.add_or_update("doc1", "some text")
except ConnectionError:
    print("Re-indexing failed — version was NOT recorded")

Production-ready usage — periodic sync job:

for doc_id, text in fetch_latest_documents():
    diff = manager.add_or_update(doc_id, text, metadata={"source": "cms"})
    if diff.content_changed and not diff.is_new:
        logger.info("Re-indexed changed doc %s (%.0f%% changed)",
                    doc_id, diff.change_ratio * 100)

history

def history(self, doc_id: str) -> List[DocVersion]

Returns all stored versions for a document, oldest first (a list() copy of the internal store's value — mutating the returned list does not affect internal state, though the DocVersion objects themselves are not deep-copied).

latest

def latest(self, doc_id: str) -> Optional[DocVersion]

Returns the most recent version, or None if the document was never indexed.

rollback

def rollback(self, doc_id: str, version: int, original_texts: Dict[str, str]) -> ContentDiff

Parameters

Raises

Algorithm

Finds the target DocVersion, validates the caller supplied its original text, then calls add_or_update(doc_id, text, target.metadata) — meaning a rollback is implemented as a new version (e.g. rolling back v3 to v1's content creates v4 with v1's content), not a literal history rewind.

Example

history = manager.history("policy")
old_text = "Smoking is prohibited in all buildings."  # caller must have this saved
manager.rollback("policy", version=1, original_texts={"policy": old_text})

delete

def delete(self, doc_id: str) -> bool

Removes all version history for doc_id and de-indexes it via rag_system.remove_document() (failures there are caught and logged, not raised). Deletes the persisted JSON file if present (fp.unlink(missing_ok=True)). Returns True if the document was tracked and removed, False if it wasn't tracked at all.

summary

def summary(self) -> Dict[str, Any]

Returns {"total_documents": n, "documents": {doc_id: {"versions", "latest", "last_updated", "chunk_count", "preview"}}} — a quick overview of every tracked document's latest state (preview truncated to 80 characters).

__repr__

DocumentVersionManager(docs={n}, persist={'yes'|'no'})

Thread safety

No locking anywhere in this module — _store is a plain dict and disk writes are not atomic (Path.write_text is not guarded against concurrent writers). Not documented or designed for concurrent use from multiple threads/processes against the same persist_path.

Best practices

• Store full document text yourself (e.g. alongside your document source of truth) if you plan to use rollback() — DocumentVersionManager only retains hashes and short previews, not full historical text.
• Use persist_path in any long-running process — without it, all version history is lost on restart.
• Don't assume the internal chunk-hash diff granularity matches your real chunker's boundaries; treat added_chunks/removed_chunks/change_ratio as an approximate change signal, not an exact diff of what will be re-indexed.

Common mistakes

• Calling rollback() without having preserved the original text out-of-band — it will raise ValueError.
• Using DocumentVersionManager as a standalone top-level core package without the zerda.rag namespace installed — add_or_update() will fail on its internal LoadedDocument import.
• Expecting remove_document() failures during add_or_update() to abort the operation — they're logged and swallowed; only the subsequent add_documents() failure is fatal.

Related APIs

• RAGSystem.add_documents, RAGSystem.remove_document — the methods this class drives.
• LoadedDocument — the wrapper used internally for re-indexing.