Class: Pikuri::VectorDb::Indexer

Inherits:
Object
  • Object
show all
Defined in:
lib/pikuri/vector_db/indexer.rb

Overview

The composing piece of the vectordb pipeline: walks the configured source, enumerates indexable files (skipping DENYLIST + dot-files), extracts text via FileType.read_as_text, chunks via the Chunker, embeds via the Embedder, and Backend#upserts the result.

Whole-corpus entry points: #index_all! (unconditional), #reindex! (nuke-and-reload), #index_if_empty! (boot-time — InMemory always indexes, Chroma only on first boot / after a manual reindex). One-file-at-a-time, driven by the Watcher daemon: #reindex_file!, #remove_file!, and #reconcile_plan (the boot sweep — a plan, not an action).

Each indexed file emits one [i/total] INFO line through LOGGER; indexing against a local llama.cpp embedder takes minutes with the user blocked on boot, so progress visibility is load-bearing (hosts with a richer channel reroute PIKURI_LOG_VECTORDB). WARN lines surface skips (image/binary, missing path, no extractable text).

Chunk identity

Chunk.id is "source:index" (relative path + ordinal-within-file) — readable, deterministic, stable across reindexes. Incremental reindex replaces by source (not by content-addressed id), so the file's content hash lives in Chunk.metadata[:hash] (same value on every chunk of one file) — that's what lets #reconcile_plan read one chunk per source and still know whether the file changed. Chunk.source is relative to the source root: short, citation-friendly, survives moving the corpus; absolute paths would tie the backend to a machine layout.

Errors mid-indexing

An embedder failure mid-run propagates and aborts (caller is internal pikuri code, not the LLM — "errors are loud"), leaving the backend partial. Recourse is #reindex!. InMemory resets on restart anyway; only Chroma persists partial state, and a fresh reindex recovers cleanly.

Constant Summary collapse

LOGGER = 
Pikuri.logger_for('VectorDb::Indexer')
DENYLIST =

Basenames skipped during the walk — cruft that accumulates inside a notes folder put under source: (a cloned repo, a Python venv, a build dir). Conservative; configurable ignore rules deferred (see ideas/vectordb-deferrals.md).

%w[
  .git
  node_modules
  __pycache__
  venv
  target
  build
  dist
  out
  vendor
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(backend:, source:, embedder:, chunker:) ⇒ Indexer

A single source, not an array: with multiple roots two files both named cooking.md would derive identical relative source values, hence identical "source:offset" ids, and Backend#upsert's replace-by-id would silently let the second overwrite the first. Multi-source is deferred — see ideas/vectordb-deferrals.md.

Parameters:

  • backend (#upsert, #query, #delete_all, #count) —

    any Backend.

  • source (String, Pathname) —

    path to index (a file directly, a directory recursively). Tilde-expanded.

  • embedder (#embed) —

    embed(Array<String>) -> Array<Array<Float>>.

  • chunker (#chunk) —

    chunk(String) -> Array<String>.



74
75
76
77
78
79
 
# File 'lib/pikuri/vector_db/indexer.rb', line 74

def initialize(backend:, source:, embedder:, chunker:)
  @backend  = backend
  @source   = Pathname.new(source).expand_path
  @embedder = embedder
  @chunker  = chunker
end

Instance Attribute Details

#source ⇒ Pathname (readonly)

Returns the configured source, tilde-expanded (a file or a directory tree). The Watcher reads this to decide what to watch.

Returns:

  • (Pathname) —

    the configured source, tilde-expanded (a file or a directory tree). The Watcher reads this to decide what to watch.



83
84
85
 
# File 'lib/pikuri/vector_db/indexer.rb', line 83

def source
  @source
end

Instance Method Details

#index_all! ⇒ Integer

Walk every source, index every reachable non-denylisted file. Returns the total chunk count emitted into the backend across this invocation.

Returns:

  • (Integer) 


99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/pikuri/vector_db/indexer.rb', line 99

def index_all!
  files = enumerate_files
  if files.empty?
    LOGGER.warn("no indexable files found under source: #{@source}")
    return 0
  end

  LOGGER.info("indexing #{files.length} file(s) from #{@source}")
  started = Time.now
  total_chunks = 0
  files.each_with_index do |(root, path), i|
    total_chunks += index_file(root: root, path: path, i: i + 1, total: files.length)
  end
  LOGGER.info(format(
                'done: %d file(s), %d chunks, %.1fs',
                files.length, total_chunks, Time.now - started
              ))
  total_chunks
end

#index_if_empty! ⇒ Integer

Index only if the backend is currently empty. The boot-time entry point — InMemory backends always re-index (RAM-only); Chroma backends only re-index on first boot or after a manual reindex!.

Returns:

  • (Integer) —

    total chunks indexed (0 if backend was non-empty and the indexer was skipped).



136
137
138
139
140
141
142
143
 
# File 'lib/pikuri/vector_db/indexer.rb', line 136

def index_if_empty!
  existing = @backend.count
  if existing.positive?
    LOGGER.info("backend already has #{existing} chunk(s); skipping boot index")
    return 0
  end
  index_all!
end

#reconcile_plan ⇒ Hash{Symbol => Array<Pathname>}

The boot reconciliation sweep, as a plan not an action: walk the tree, hash every file, diff against Backend#sources_with_hashes, return the work list. The Watcher feeds it through its single queue so the sweep and live events share one last-intent-wins path (and teardown can interrupt between files). Uniform across backends with no is_a?: InMemory reports an empty manifest at boot (RAM reset) so every file reads as new; Chroma reports its persisted manifest so only changed files return — which also closes the downtime gap (changes made while no Watcher ran are invisible to filesystem events but caught by the diff).

Returns:

  • (Hash{Symbol => Array<Pathname>}) —

    {reindex:, remove:} — files new-or-changed, and files (as +root+/source paths) gone from disk.



210
211
212
213
214
215
216
217
218
 
# File 'lib/pikuri/vector_db/indexer.rb', line 210

def reconcile_plan
  on_disk = {} # source (String) => Pathname
  enumerate_files.each { |_root, p| on_disk[relative_source(p)] = p }
  indexed = @backend.sources_with_hashes

  reindex = on_disk.select { |source, path| indexed[source] != file_hash(path) }.values
  remove  = (indexed.keys - on_disk.keys).map { |source| root.join(source) }
  { reindex: reindex, remove: remove }
end

#reindex! ⇒ Integer

backend.delete_all followed by #index_all!. The v1 nuke-and-reload reindex path.

Returns:

  • (Integer) —

    total chunks indexed.



123
124
125
126
127
 
# File 'lib/pikuri/vector_db/indexer.rb', line 123

def reindex!
  LOGGER.info('reindex: clearing backend')
  @backend.delete_all
  index_all!
end

#reindex_file!(path) ⇒ Integer

Re-index a single file in place: extract → chunk → embed → Backend#replace_source (the atomic unit the Watcher drives on a modify/add event). The embed happens before the backend write, so an embedder outage raises here leaving the prior chunks untouched (see Backend::Chroma#replace_source). A file that no longer yields text (emptied, now-binary, vanished) is treated as a removal — its stale chunks drop and nothing is written, so no orphans linger.

Parameters:

  • path (String, Pathname) —

    the file; need not exist (vanished ⇒ removal).

Returns:

  • (Integer) —

    chunks now stored (0 if removed).

Raises:

  • (RuntimeError) —

    if the embedder or backend fails — the Watcher logs it and moves on.



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/pikuri/vector_db/indexer.rb', line 158

def reindex_file!(path)
  path   = Pathname.new(path).expand_path
  source = relative_source(path)

  texts = begin
    chunk_texts_for(path)
  rescue ArgumentError, Errno::ENOENT, RuntimeError => e
    LOGGER.info("reindex #{source}: unindexable (#{e.message}); removing from index")
    @backend.delete_by_source(source)
    return 0
  end

  if texts.empty?
    LOGGER.info("reindex #{source}: no indexable text; removing from index")
    @backend.delete_by_source(source)
    return 0
  end

  # embed_and_build is *outside* the rescue: an embedder outage
  # raises straight out, before replace_source touches the
  # backend, so the prior chunks survive (embed-before-delete).
  chunks, vectors = embed_and_build(source: source, path: path, chunk_texts: texts)
  @backend.replace_source(source: source, chunks: chunks, vectors: vectors)
  LOGGER.info("reindex #{source}: #{chunks.length} chunk(s)")
  chunks.length
end

#remove_file!(path) ⇒ void

This method returns an undefined value.

Drop a file's chunks from the index — the Watcher's response to a delete (or move-away) event. Idempotent: removing a source that isn't indexed is a no-op.

Parameters:

  • path (String, Pathname) —

    the (now-absent) file.



191
192
193
194
195
196
 
# File 'lib/pikuri/vector_db/indexer.rb', line 191

def remove_file!(path)
  source = relative_source(Pathname.new(path).expand_path)
  @backend.delete_by_source(source)
  LOGGER.info("removed #{source} from index")
  nil
end

#root ⇒ Pathname

The directory whose tree is indexed — the anchor for every relative Chunk#source. A directory source is its own root; a single-file source roots at its parent (citation is just the basename).

Returns:

  • (Pathname) 


90
91
92
 
# File 'lib/pikuri/vector_db/indexer.rb', line 90

def root
  @source.directory? ? @source : @source.parent
end