sharmaaryan commited on
Commit
a8bc5d0
Β·
1 Parent(s): 283b271

Embedded Qdrant vector store (no external cluster)

Browse files
.dockerignore CHANGED
@@ -19,13 +19,16 @@ backend/.venv/
19
  .ruff_cache/
20
 
21
  # Heavy data NOT needed at runtime (offline ingestion inputs + eval outputs).
22
- # data/duckdb/ and data/bm25_index.pkl are intentionally NOT ignored β€” the
23
- # runtime needs them and the Dockerfile COPYs them explicitly.
 
24
  data/raw/
25
  data/processed/
26
  data/*.json
27
  data/eval_results_*.json
28
  data/smoke_results.json
 
 
29
 
30
  # Repo meta
31
  .git/
 
19
  .ruff_cache/
20
 
21
  # Heavy data NOT needed at runtime (offline ingestion inputs + eval outputs).
22
+ # data/duckdb/, data/bm25_index.pkl, and data/qdrant/ (the embedded vector store)
23
+ # are intentionally NOT ignored β€” the runtime needs them and the Dockerfile COPYs
24
+ # them explicitly.
25
  data/raw/
26
  data/processed/
27
  data/*.json
28
  data/eval_results_*.json
29
  data/smoke_results.json
30
+ # The embedded store's runtime lock β€” never bake it in; Qdrant recreates it.
31
+ data/qdrant/.lock
32
 
33
  # Repo meta
34
  .git/
.gitattributes CHANGED
@@ -34,3 +34,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
34
  *.zst filter=lfs diff=lfs merge=lfs -text
35
  *tfevents* filter=lfs diff=lfs merge=lfs -text
36
  *.duckdb filter=lfs diff=lfs merge=lfs -text
 
 
34
  *.zst filter=lfs diff=lfs merge=lfs -text
35
  *tfevents* filter=lfs diff=lfs merge=lfs -text
36
  *.duckdb filter=lfs diff=lfs merge=lfs -text
37
+ *.sqlite filter=lfs diff=lfs merge=lfs -text
Dockerfile CHANGED
@@ -33,10 +33,14 @@ WORKDIR /app
33
  COPY --from=builder --chown=appuser:appuser /app/backend /app/backend
34
  COPY --chown=appuser:appuser data/duckdb /app/data/duckdb
35
  COPY --chown=appuser:appuser data/bm25_index.pkl /app/data/bm25_index.pkl
 
 
 
36
 
37
  ENV PATH="/app/backend/.venv/bin:$PATH" \
38
  PYTHONUNBUFFERED=1 \
39
  PYTHONIOENCODING=utf-8 \
 
40
  PORT=8000
41
  USER appuser
42
  EXPOSE 8000
 
33
  COPY --from=builder --chown=appuser:appuser /app/backend /app/backend
34
  COPY --chown=appuser:appuser data/duckdb /app/data/duckdb
35
  COPY --chown=appuser:appuser data/bm25_index.pkl /app/data/bm25_index.pkl
36
+ # Embedded Qdrant vector store β€” baked in like DuckDB/BM25 so the image is fully
37
+ # self-contained (no external cluster to idle-wipe, no QDRANT_URL/KEY secret).
38
+ COPY --chown=appuser:appuser data/qdrant /app/data/qdrant
39
 
40
  ENV PATH="/app/backend/.venv/bin:$PATH" \
41
  PYTHONUNBUFFERED=1 \
42
  PYTHONIOENCODING=utf-8 \
43
+ QDRANT_PATH=./data/qdrant \
44
  PORT=8000
45
  USER appuser
46
  EXPOSE 8000
backend/src/finrag/config.py CHANGED
@@ -11,6 +11,14 @@ class Settings(BaseSettings):
11
  anthropic_api_key: str | None = None
12
  gemini_api_key: str | None = None
13
  cohere_api_key: str
 
 
 
 
 
 
 
 
14
  qdrant_url: str = "http://localhost:6333"
15
  qdrant_api_key: str | None = None
16
  duckdb_path: str = "./data/duckdb/finrag.duckdb"
 
11
  anthropic_api_key: str | None = None
12
  gemini_api_key: str | None = None
13
  cohere_api_key: str
14
+ # Vector store location. Two mutually exclusive modes:
15
+ # β€’ qdrant_path set β†’ EMBEDDED Qdrant (in-process, on-disk at this path).
16
+ # The public deploy uses this: the vector store is baked into the Docker
17
+ # image alongside DuckDB + BM25, so there is no external cluster to idle-
18
+ # wipe and no QDRANT_URL/KEY secret. Relative paths resolve from REPO_ROOT.
19
+ # β€’ qdrant_path empty β†’ REMOTE Qdrant at qdrant_url (local docker-compose in
20
+ # dev, or a managed cluster). The original behaviour, unchanged.
21
+ qdrant_path: str | None = None
22
  qdrant_url: str = "http://localhost:6333"
23
  qdrant_api_key: str | None = None
24
  duckdb_path: str = "./data/duckdb/finrag.duckdb"
backend/src/finrag/ingestion/embed.py CHANGED
@@ -11,6 +11,7 @@ Idempotent: re-running overwrites existing points by ID (deterministic hash).
11
  from __future__ import annotations
12
 
13
  import json
 
14
  import time
15
  from collections.abc import Iterable, Iterator
16
  from pathlib import Path
@@ -21,7 +22,7 @@ from cohere.errors import TooManyRequestsError
21
  from qdrant_client import QdrantClient
22
  from qdrant_client.models import Distance, PointStruct, VectorParams
23
 
24
- from finrag.config import settings
25
  from finrag.ingestion.parse import PROCESSED_DIR, Chunk
26
 
27
  # ── Constants ─────────────────────────────────────────────────────────────
@@ -48,10 +49,50 @@ COHERE_RETRY_INITIAL_BACKOFF_SECONDS = 30.0
48
  COHERE_MAX_RETRIES = 5
49
 
50
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
51
  # ── Helpers ───────────────────────────────────────────────────────────────
52
  T = TypeVar("T")
53
 
54
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
55
  def _batched(seq: Iterable[T], n: int) -> Iterator[list[T]]:
56
  """Yield lists of size `n` from `seq`. Last batch may be shorter."""
57
  buf: list[T] = []
@@ -149,10 +190,15 @@ def _chunk_to_point(chunk: Chunk, vector: list[float]) -> PointStruct:
149
 
150
 
151
  # ── Core ──────────────────────────────────────────────────────────────────
152
- def embed_and_upsert(chunks: list[Chunk]) -> int:
153
- """Embed all chunks and upsert into Qdrant. Returns count of points written."""
 
 
 
 
 
154
  co = cohere.ClientV2(api_key=settings.cohere_api_key)
155
- qdrant = QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
156
  _ensure_collection(qdrant, COLLECTION_NAME, EMBED_DIM)
157
 
158
  qdrant_buf: list[PointStruct] = []
@@ -191,20 +237,24 @@ def embed_and_upsert(chunks: list[Chunk]) -> int:
191
 
192
  # ── CLI ───────────────────────────────────────────────────────────────────
193
  def main() -> None:
 
194
  chunks = _read_all_chunks(PROCESSED_DIR)
195
  print(f"Loaded {len(chunks)} chunks from {PROCESSED_DIR}\n")
196
  if not chunks:
197
  print("No chunks found. Run `finrag.ingestion.parse` first.")
198
  return
199
 
200
- written = embed_and_upsert(chunks)
 
201
 
202
- # Verify final state via Qdrant's own count
203
- qdrant = QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
204
  info = qdrant.get_collection(COLLECTION_NAME)
 
205
  print(
206
  f"\nDone. {written} points written this run. "
207
- f"Collection '{COLLECTION_NAME}' contains {info.points_count} total."
 
208
  )
209
 
210
 
 
11
  from __future__ import annotations
12
 
13
  import json
14
+ import sys
15
  import time
16
  from collections.abc import Iterable, Iterator
17
  from pathlib import Path
 
22
  from qdrant_client import QdrantClient
23
  from qdrant_client.models import Distance, PointStruct, VectorParams
24
 
25
+ from finrag.config import REPO_ROOT, settings
26
  from finrag.ingestion.parse import PROCESSED_DIR, Chunk
27
 
28
  # ── Constants ─────────────────────────────────────────────────────────────
 
49
  COHERE_MAX_RETRIES = 5
50
 
51
 
52
+ # ── Clients ───────────────────────────────────────────────────────────────
53
+ def make_qdrant_client() -> QdrantClient:
54
+ """Build the Qdrant client for both ingestion and retrieval.
55
+
56
+ Embedded mode (settings.qdrant_path set): an in-process, on-disk Qdrant β€” no
57
+ server, nothing external to wipe. This is what the public image ships, with
58
+ the store baked in next to DuckDB/BM25. Otherwise: a remote client at
59
+ settings.qdrant_url (dev docker-compose or a managed cluster).
60
+
61
+ Note on embedded mode: the on-disk store is locked to a single client per
62
+ process, so the lru_cached retrieval client (one per FastAPI worker) is the
63
+ intended access pattern; don't open a second concurrent client on the path.
64
+ """
65
+ if settings.qdrant_path:
66
+ path = Path(settings.qdrant_path)
67
+ if not path.is_absolute():
68
+ path = REPO_ROOT / path
69
+ path.mkdir(parents=True, exist_ok=True)
70
+ return QdrantClient(path=str(path))
71
+ return QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
72
+
73
+
74
  # ── Helpers ───────────────────────────────────────────────────────────────
75
  T = TypeVar("T")
76
 
77
 
78
+ def _force_utf8_stdout() -> None:
79
+ """Make progress output (which uses ↳/⚠ glyphs) safe on any console.
80
+
81
+ Windows consoles default to cp1252, which can't encode those characters and
82
+ raises UnicodeEncodeError mid-run β€” crashing the upload after the collection
83
+ is created but before any points land (a silent empty-collection state). The
84
+ Docker image avoids this via PYTHONIOENCODING=utf-8; this makes a bare local
85
+ `python -m finrag.ingestion.embed` match it without needing the env var.
86
+ """
87
+ for stream in (sys.stdout, sys.stderr):
88
+ reconfigure = getattr(stream, "reconfigure", None)
89
+ if reconfigure is not None:
90
+ try:
91
+ reconfigure(encoding="utf-8")
92
+ except (ValueError, OSError): # redirected/non-reconfigurable stream
93
+ pass
94
+
95
+
96
  def _batched(seq: Iterable[T], n: int) -> Iterator[list[T]]:
97
  """Yield lists of size `n` from `seq`. Last batch may be shorter."""
98
  buf: list[T] = []
 
190
 
191
 
192
  # ── Core ──────────────────────────────────────────────────────────────────
193
+ def embed_and_upsert(chunks: list[Chunk], qdrant: QdrantClient | None = None) -> int:
194
+ """Embed all chunks and upsert into Qdrant. Returns count of points written.
195
+
196
+ Accepts an optional client so the CLI can build one and reuse it for the
197
+ post-run count β€” embedded (on-disk) mode locks the store to a single client,
198
+ so opening a second one would fail.
199
+ """
200
  co = cohere.ClientV2(api_key=settings.cohere_api_key)
201
+ qdrant = qdrant or make_qdrant_client()
202
  _ensure_collection(qdrant, COLLECTION_NAME, EMBED_DIM)
203
 
204
  qdrant_buf: list[PointStruct] = []
 
237
 
238
  # ── CLI ───────────────────────────────────────────────────────────────────
239
  def main() -> None:
240
+ _force_utf8_stdout()
241
  chunks = _read_all_chunks(PROCESSED_DIR)
242
  print(f"Loaded {len(chunks)} chunks from {PROCESSED_DIR}\n")
243
  if not chunks:
244
  print("No chunks found. Run `finrag.ingestion.parse` first.")
245
  return
246
 
247
+ qdrant = make_qdrant_client()
248
+ written = embed_and_upsert(chunks, qdrant=qdrant)
249
 
250
+ # Verify final state via Qdrant's own count (reusing the same client, so
251
+ # embedded-mode's single-client lock isn't violated).
252
  info = qdrant.get_collection(COLLECTION_NAME)
253
+ target = "embedded " + str(settings.qdrant_path) if settings.qdrant_path else settings.qdrant_url
254
  print(
255
  f"\nDone. {written} points written this run. "
256
+ f"Collection '{COLLECTION_NAME}' contains {info.points_count} total "
257
+ f"({target})."
258
  )
259
 
260
 
backend/src/finrag/retrieval/vector.py CHANGED
@@ -20,7 +20,7 @@ from qdrant_client.models import (
20
  )
21
 
22
  from finrag.config import settings
23
- from finrag.ingestion.embed import COHERE_MODEL, COLLECTION_NAME
24
 
25
 
26
  # ── Public response model ─────────────────────────────────────────────────
@@ -55,10 +55,10 @@ def get_cohere_client() -> cohere.ClientV2:
55
 
56
  @lru_cache(maxsize=1)
57
  def get_qdrant_client() -> QdrantClient:
58
- return QdrantClient(
59
- url=settings.qdrant_url,
60
- api_key=settings.qdrant_api_key,
61
- )
62
 
63
 
64
  # ── Query embedding ───────────────────────────────────────────────────────
 
20
  )
21
 
22
  from finrag.config import settings
23
+ from finrag.ingestion.embed import COHERE_MODEL, COLLECTION_NAME, make_qdrant_client
24
 
25
 
26
  # ── Public response model ─────────────────────────────────────────────────
 
55
 
56
  @lru_cache(maxsize=1)
57
  def get_qdrant_client() -> QdrantClient:
58
+ # Embedded (on-disk) or remote, decided by settings.qdrant_path β€” see
59
+ # make_qdrant_client. lru_cache makes this the single client per worker that
60
+ # embedded mode requires.
61
+ return make_qdrant_client()
62
 
63
 
64
  # ── Query embedding ───────────────────────────────────────────────────────
data/qdrant/collection/finrag_chunks/storage.sqlite ADDED
@@ -0,0 +1,3 @@
 
 
 
 
1
+ version https://git-lfs.github.com/spec/v1
2
+ oid sha256:2e17a57ab1ede9a19ed389e0878ea1b011ba1c15b30676e2a563ac2b3493786e
3
+ size 47828992
data/qdrant/meta.json ADDED
@@ -0,0 +1 @@
 
 
1
+ {"collections": {"finrag_chunks": {"vectors": {"size": 1024, "distance": "Cosine", "hnsw_config": null, "quantization_config": null, "on_disk": null, "datatype": null, "multivector_config": null}, "shard_number": null, "sharding_method": null, "replication_factor": null, "write_consistency_factor": null, "on_disk_payload": null, "hnsw_config": null, "wal_config": null, "optimizers_config": null, "quantization_config": null, "sparse_vectors": null, "strict_mode_config": null, "metadata": null}}, "aliases": {}}