AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build an incremental document indexer

Build an Incremental Document Indexer

Re-indexing your entire corpus every time a document changes is wasteful and creates operational risk. Incremental indexing tracks which documents have changed by hashing their content, then only re-embeds and re-indexes what is new or updated.

What you are building

Create an IncrementalIndexer that:

Tracks document state in a persistent JSON file: {doc_id: {"hash": str, "chunk_ids": list[str], "indexed_at": str}}.
Detects changes by comparing content hash to stored hash. A document is dirty if its hash changed or is absent from state.
Handles upserts: for a changed document, delete old chunk vectors, re-chunk, embed, and add new chunks.
Handles deletes: when a document is removed, delete its chunk vectors and remove from state.
Reports indexing results via an IndexSyncResult dataclass.
Is idempotent: calling sync twice with the same input makes no API calls on the second run.

Why this matters

Without incremental indexing, every CI deploy triggers a full re-index that may take hours and cost thousands of API calls. With it, a corpus of 100K documents where 50 changed today only costs indexing those 50.

Constraints

State file is JSON (human-readable for debugging).
Chunker and embed+add functions are injected (testable in isolation).
Timestamps in ISO 8601 format.

Rag Systems / medium / Step 5 of 6

Practice stage

General drill

Hint

Keep the solution explicit and reviewable.

Next drill

Implement retrieval with metadata filtering

Success criteria

Make the solution explicit, debuggable, and easy to explain.

Review checklist

Review where the boundary is, what gets validated, and what would be hard to debug later.

from __future__ import annotations

import hashlib
import json
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable

@dataclass
class DocumentInput:
    id: str
    content: str
    metadata: dict = field(default_factory=dict)

@dataclass
class IndexSyncResult:
    added: int = 0
    updated: int = 0
    skipped: int = 0
    deleted: int = 0
    errors: list[str] = field(default_factory=list)

class IncrementalIndexer:
    def __init__(
        self,
        state_file: str,
        chunk_fn: Callable[[str, str], list[dict]],
        index_add_fn: Callable[[list[dict]], None],
        index_delete_fn: Callable[[list[str]], None],
    ):
        self.state_file = state_file
        self.chunk_fn = chunk_fn
        self.index_add_fn = index_add_fn
        self.index_delete_fn = index_delete_fn
        self._state: dict = self._load_state()

def _load_state(self) -> dict:
        # TODO: load JSON from state_file; return {} if not found
        raise NotImplementedError

def _save_state(self) -> None:
        # TODO: write self._state to state_file as JSON
        raise NotImplementedError

def _content_hash(self, content: str) -> str:
        # TODO: return SHA-256 hex digest
        raise NotImplementedError

def upsert(self, doc: DocumentInput) -> str:
        """Index or re-index a document if content changed. Returns 'added', 'updated', or 'skipped'."""
        raise NotImplementedError

def delete(self, doc_id: str) -> bool:
        """Remove a document from the index and state. Returns True if it existed."""
        raise NotImplementedError

def sync(self, documents: list[DocumentInput]) -> IndexSyncResult:
        """Sync a full list of documents. Documents not in input list will be deleted."""
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.