ResearchRAG / app /database.py
riezqidr's picture
chore: add production essentials
4397111
Raw
History Blame Contribute Delete
6.08 kB
"""
ChromaDB client + embedding singleton for ResearchRAG.
Collections per user:
{name}_{hash} β†’ child chunks (small, embedded, searched)
{name}_{hash}_parent β†’ parent chunks (large, LLM context, looked up by ID)
"""
import chromadb
from chromadb.config import Settings as ChromaSettings
from sentence_transformers import SentenceTransformer
from app.config import get_settings
from app.logger import logger
import hashlib
import threading
_client = None
_collections: dict[str, chromadb.Collection] = {}
_embedder = None
_embedder_lock = threading.Lock()
# ─── Embedder ─────────────────────────────────────────────────────────────────
def get_embedder() -> SentenceTransformer:
global _embedder
# Same race as the reranker: concurrent threadpool workers could each see
# None and each load a ~1.1 GB model. Double-checked locking bounds it to one.
if _embedder is None:
with _embedder_lock:
if _embedder is None:
cfg = get_settings()
logger.info("embedder_loading model=%s", cfg.embedding_model)
_embedder = SentenceTransformer(cfg.embedding_model)
return _embedder
# ─── Collection naming ────────────────────────────────────────────────────────
def _normalize_user_id(user_id: str | None) -> str:
if not user_id:
return "default"
return user_id.strip().lower() or "default"
def _collection_name(user_id: str | None, role: str = "child") -> str:
cfg = get_settings()
normalized = _normalize_user_id(user_id)
if normalized == "default":
base = cfg.chroma_collection
else:
suffix = hashlib.md5(normalized.encode()).hexdigest()[:12]
base = f"{cfg.chroma_collection}_{suffix}"
if role == "parent":
return f"{base}_parent"
return base
# ─── Client init ──────────────────────────────────────────────────────────────
def _get_client() -> chromadb.PersistentClient:
global _client
if _client is None:
cfg = get_settings()
_client = chromadb.PersistentClient(
path = cfg.chroma_path,
settings = ChromaSettings(anonymized_telemetry=False),
)
return _client
def _get_or_create(name: str) -> chromadb.Collection:
client = _get_client()
col = client.get_or_create_collection(
name = name,
metadata = {"hnsw:space": "cosine"},
)
return col
# ─── Public collection accessors ─────────────────────────────────────────────
def init_chroma(user_id: str | None = None):
global _collections
normalized = _normalize_user_id(user_id)
child_name = _collection_name(user_id, "child")
parent_name = _collection_name(user_id, "parent")
_collections[normalized] = _get_or_create(child_name)
_collections[f"{normalized}_parent"] = _get_or_create(parent_name)
get_embedder()
print(
f"[ChromaDB] Ready β€” {child_name} ({_collections[normalized].count()} child chunks) "
f"| {parent_name} ({_collections[f'{normalized}_parent'].count()} parent chunks)"
)
def get_collection(user_id: str | None = None) -> chromadb.Collection:
"""Returns the CHILD collection (for embedding + search)."""
normalized = _normalize_user_id(user_id)
if normalized not in _collections:
init_chroma(user_id)
return _collections[normalized]
def get_parent_collection(user_id: str | None = None) -> chromadb.Collection:
"""Returns the PARENT collection (for LLM context lookup)."""
normalized = _normalize_user_id(user_id)
key = f"{normalized}_parent"
if key not in _collections:
init_chroma(user_id)
return _collections[key]
def get_parents_by_ids(
parent_ids: list[str],
user_id: str | None = None,
) -> list[dict]:
"""
Fetch parent chunks by their IDs from the parent collection.
Returns list of {id, text, metadata} dicts.
"""
if not parent_ids:
return []
col = get_parent_collection(user_id)
unique = list(dict.fromkeys(parent_ids)) # deduplicate, preserve order
try:
results = col.get(ids=unique, include=["documents", "metadatas"])
except Exception:
return []
out = []
for pid, doc, meta in zip(
results.get("ids", []),
results.get("documents", []),
results.get("metadatas", []),
):
out.append({"id": pid, "text": doc, "metadata": meta})
return out
# ─── Utilities ────────────────────────────────────────────────────────────────
def embed_documents(texts: list[str]) -> list[list[float]]:
"""Embed passages for indexing. e5 models need the 'passage: ' prefix."""
prefix = get_settings().embedding_passage_prefix
return get_embedder().encode(
[f"{prefix}{t}" for t in texts],
normalize_embeddings=True,
show_progress_bar=False,
).tolist()
def embed_query(text: str) -> list[float]:
"""Embed a search query. e5 models need the 'query: ' prefix."""
prefix = get_settings().embedding_query_prefix
return get_embedder().encode(
f"{prefix}{text}",
normalize_embeddings=True,
).tolist()
# Backward-compat alias β€” all remaining callers embed documents.
embed_texts = embed_documents
def make_doc_id(source: str, chunk_index: int) -> str:
"""Stable unique ID for a chunk."""
raw = f"{source}::{chunk_index}"
return hashlib.md5(raw.encode()).hexdigest()