Spaces:
Running
Running
| """ | |
| 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() | |