Class: Pikuri::VectorDb::Backend::Qdrant

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

Overview

Thin Faraday HTTP client against a self-hosted Qdrant server (1.x REST, 1.8+), the recommended persistent backend (see pikuri-vectordb/DESIGN.md for the Chroma-vs-Qdrant survey). Same duck-typed Pikuri::VectorDb::Backend protocol as InMemory / Chroma. Hand-rolled for the same reasons as Chroma; the 1.x REST surface is stable, so the track-by-hand cost is lower. Bring your own via +Qdrant.new(host:, port:, collection:)+, or let Server::Qdrant.ensure_running supervise a container — same server-vs-client split as Server::Chroma / Chroma. Endpoints used are all under /collections/{name} (+exists+, PUT to create, points upsert, points/search, points/count, points/delete, points/scroll, DELETE).

Point ids: chunk id → derived UUID

Qdrant point ids must be unsigned ints or UUIDs, so the Chunk's readable "source:offset" id can't be the point id. Each is derived as the first 16 bytes of SHA1(chunk.id) UUID-formatted: deterministic, so re-upserting the same chunk id replaces the old point (collision probability negligible at any corpus size). The real chunk id rides in the payload's reserved id key and is restored out, so callers never see the UUID.

Contracts

  • Payload: Qdrant has no separate document store, so everything beyond the vector lives in the point payload — RESERVED_PAYLOAD_KEYS carry the fixed fields, metadata is merged alongside with the same stringify/symbolize normalization as Chroma (identical chunk across backends).
  • Cosine, no conversion: created distance: 'Cosine'; unlike Chroma (distance, 1 - d), Qdrant's score already is a similarity, passed through untouched. (Same convention that makes mem0's ranker correct on Qdrant and inverted on pgvector — pikuri-memory/DESIGN.md.)
  • Vector-dim diverges from InMemory like Chroma, with a wrinkle: Qdrant fixes the dim at collection creation, so #upsert creates it with the first batch's size; later mismatches are HTTP 4xx → RuntimeError.
  • No payload indexes (deliberate): every filtered call is an exact scan that works unindexed, milliseconds at pikuri's 10³–10⁵ scale.

Constant Summary collapse

MANIFEST_PAGE_SIZE =

Rows per /scroll page in #sources_with_hashes — caps the boot manifest read (cf. Chroma::MANIFEST_PAGE_SIZE).

1_000
RESERVED_PAYLOAD_KEYS =

Payload keys reserved for the Chunk's fixed fields; user metadata with these keys would collide, so the Indexer never sets them.

%w[id source text].freeze

Instance Method Summary collapse

Constructor Details

#initialize(host:, port:, collection:, connection: nil) ⇒ Qdrant

Parameters:

  • host (String)
  • port (Integer)
  • collection (String) —

    Qdrant collection name (engine-specific, so it lives here, not on VectorDb::Extension — cf. Chroma).

  • connection (Faraday::Connection, nil) (defaults to: nil) —

    DI point for tests.

Raises:

  • (ArgumentError) —

    on empty host or empty collection.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 65

def initialize(host:, port:, collection:, connection: nil)
  raise ArgumentError, 'host must be non-empty' if host.nil? || host.to_s.empty?
  raise ArgumentError, 'collection must be non-empty' if collection.nil? || collection.to_s.empty?

  @host = host
  @port = port
  @collection = collection
  @known_exists = false
  @connection = connection || Faraday.new(url: "http://#{host}:#{port}") do |f|
    f.request :json
    f.response :json
    f.adapter Faraday.default_adapter
  end
end

Instance Method Details

#count ⇒ Integer

Returns current chunk count (exact). Zero before the first #upsert.

Returns:

  • (Integer) —

    current chunk count (exact). Zero before the first #upsert.

Raises:

  • (RuntimeError) —

    on HTTP failure.



176
177
178
179
180
181
182
183
184
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 176

def count
  return 0 unless collection_exists?

  body = request_json(:post, "#{collection_path}/points/count", { exact: true })
  n = body.is_a?(Hash) ? body.dig('result', 'count') : nil
  raise "Backend::Qdrant: count response missing result.count (got #{body.inspect})" unless n.is_a?(Integer)

  n
end

#delete_all ⇒ void

This method returns an undefined value.

Drop the collection. Next #upsert re-creates from scratch (with that batch's vector dim) — the v1 nuke-and-reload reindex path. No-op if no collection was ever created; 404 on the DELETE is treated as "already gone" — idempotent.

Raises:

  • (RuntimeError) —

    on unexpected HTTP failure.



161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 161

def delete_all
  return nil unless collection_exists?

  response = @connection.delete(collection_path)
  unless [200, 404].include?(response.status)
    raise "Backend::Qdrant: DELETE #{collection_path} returned " \
          "HTTP #{response.status}: #{response.body.inspect}"
  end
  @known_exists = false
  nil
end

#delete_by_source(source) ⇒ void

This method returns an undefined value.

Remove every chunk whose source matches, via a payload-filtered points/delete. The scoped counterpart to #delete_all. No-op when the collection doesn't exist yet.

Parameters:

Raises:

  • (RuntimeError) —

    on HTTP failure.



194
195
196
197
198
199
200
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 194

def delete_by_source(source)
  return nil unless collection_exists?

  request_json(:post, "#{collection_path}/points/delete?wait=true",
               { filter: source_filter(source) })
  nil
end

#query(vector:, top_k:) ⇒ Array<Backend::Result>

k-NN query by cosine similarity. Returns at most top_k Results descending by score. Qdrant's +score+ is already a cosine similarity (higher = better), so it passes through unconverted — same scale as InMemory.

Parameters:

  • vector (Array<Float>)
  • top_k (Integer)

Returns:

Raises:

  • (ArgumentError) —

    on non-positive top_k.

  • (RuntimeError) —

    on HTTP failure.



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 122

def query(vector:, top_k:)
  raise ArgumentError, "top_k must be positive (got #{top_k})" if top_k <= 0

  # Never upserted → no collection → semantic answer is
  # "no hits" (same short-circuit as Chroma).
  return [] unless collection_exists?

  body = request_json(:post, "#{collection_path}/points/search", {
                        vector: vector,
                        limit: top_k,
                        with_payload: true
                      })

  hits = body['result'] || []
  hits.map do |hit|
    payload = hit['payload'] || {}
    chunk_meta = {}
    payload.each do |k, v|
      next if RESERVED_PAYLOAD_KEYS.include?(k)

      chunk_meta[k.to_sym] = v
    end

    chunk = Chunk.new(
      id: payload['id'] || '', source: payload['source'] || '',
      text: payload['text'] || '', metadata: chunk_meta
    )
    Result.new(chunk: chunk, score: hit['score'].to_f)
  end
end

#replace_source(source:, chunks:, vectors:) ⇒ void

This method returns an undefined value.

Replace all chunks for one source: delete the old set, then upsert the new one. The incremental-reindex unit. Two HTTP calls, so not transactional — same divergence from InMemory, same mitigation, as documented on Chroma#replace_source.

Parameters:

  • source (String) —

    the Chunk#source being replaced.

  • chunks (Array<Chunk>) —

    the new chunk set.

  • vectors (Array<Array<Float>>) —

    parallel to chunks.

Raises:

  • (ArgumentError) —

    on empty input or length mismatch.

  • (RuntimeError) —

    on HTTP failure.



214
215
216
217
218
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 214

def replace_source(source:, chunks:, vectors:)
  delete_by_source(source)
  upsert(chunks: chunks, vectors: vectors)
  nil
end

#source_indexed?(source) ⇒ Boolean

Is source in the corpus? A payload-filtered exact count — O(1) transport regardless of corpus size, never the full #sources_with_hashes manifest. See the Backend protocol yardoc.

Parameters:

Returns:

  • (Boolean) —

    true if at least one chunk has this source.

Raises:

  • (RuntimeError) —

    on HTTP failure.



267
268
269
270
271
272
273
274
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 267

def source_indexed?(source)
  return false unless collection_exists?

  body = request_json(:post, "#{collection_path}/points/count",
                      { filter: source_filter(source), exact: true })
  n = body.is_a?(Hash) ? body.dig('result', 'count') : nil
  n.is_a?(Integer) && n.positive?
end

#sources_with_hashes ⇒ Hash{String => String, nil}

The boot-sweep reference: source → stored content hash, one payload row per file (filter offset == 0), projected to +source+/+hash+, vectors excluded, paged by Qdrant's own scroll cursor (+next_page_offset+ — no client-side offset arithmetic, unlike Chroma). Assumes the manifest isn't mutating mid-read; the Watcher drives it from its single worker thread.

Returns:

  • (Hash{String => String, nil}) —

    source → content hash. Empty when the collection doesn't exist yet.

Raises:

  • (RuntimeError) —

    on HTTP failure.



230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 230

def sources_with_hashes
  return {} unless collection_exists?

  result = {}
  cursor = nil
  loop do
    request = {
      filter: { must: [{ key: 'offset', match: { value: 0 } }] },
      with_payload: %w[source hash],
      with_vector: false,
      limit: MANIFEST_PAGE_SIZE
    }
    request[:offset] = cursor unless cursor.nil?

    body = request_json(:post, "#{collection_path}/points/scroll", request)
    points = body.is_a?(Hash) ? (body.dig('result', 'points') || []) : []
    points.each do |point|
      payload = point['payload']
      next unless payload.is_a?(Hash) && payload['source']

      result[payload['source']] = payload['hash']
    end

    cursor = body.is_a?(Hash) ? body.dig('result', 'next_page_offset') : nil
    break if cursor.nil?
  end
  result
end

#upsert(chunks:, vectors:) ⇒ void

This method returns an undefined value.

Insert-or-replace by chunk.id (via the derived point UUID — see the class header). Parallel arrays of equal length; raises on empty input or length mismatch (same contract as InMemory). Creates the collection on first use, fixing the vector dim to this batch's; mismatched dims later surface as RuntimeError from a 4xx response.

Parameters:

  • chunks (Array<Chunk>)
  • vectors (Array<Array<Float>>)

Raises:

  • (ArgumentError) —

    on empty input or length mismatch.

  • (RuntimeError) —

    on HTTP failure.



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/pikuri/vector_db/backend/qdrant.rb', line 93

def upsert(chunks:, vectors:)
  raise ArgumentError, 'upsert called with empty chunks/vectors' if chunks.empty?
  if chunks.size != vectors.size
    raise ArgumentError, "size mismatch: #{chunks.size} chunks vs #{vectors.size} vectors"
  end

  ensure_collection!(dim: vectors.first.size)

  points = chunks.each_with_index.map do |c, i|
    payload = { 'id' => c.id, 'source' => c.source, 'text' => c.text }
    c.metadata.each { |k, v| payload[k.to_s] = v }
    { id: point_id(c.id), vector: vectors[i], payload: payload }
  end

  request_json(:put, "#{collection_path}/points?wait=true", { points: points })
  nil
end