54 lines
No EOL
2 KiB
Python
54 lines
No EOL
2 KiB
Python
"""
|
|
Shared ChromaDB client singleton to prevent initialization conflicts
|
|
"""
|
|
import chromadb
|
|
from chromadb.config import Settings as ChromaSettings
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
class ChromaDBSingleton:
|
|
"""Singleton ChromaDB client to prevent 'different settings' errors"""
|
|
_instance: Optional['ChromaDBSingleton'] = None
|
|
_client: Optional[chromadb.PersistentClient] = None
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def get_client(self, chroma_db_path: str) -> chromadb.PersistentClient:
|
|
"""Get or create ChromaDB client with consistent settings"""
|
|
if self._client is None:
|
|
try:
|
|
# Use consistent settings across all services
|
|
settings = ChromaSettings(anonymized_telemetry=False)
|
|
self._client = chromadb.PersistentClient(
|
|
path=chroma_db_path,
|
|
settings=settings
|
|
)
|
|
print(f"Created shared ChromaDB client at {chroma_db_path}")
|
|
except Exception as e:
|
|
print(f"Error creating shared ChromaDB client: {e}")
|
|
# Try with minimal settings if the above fails
|
|
try:
|
|
self._client = chromadb.PersistentClient(path=chroma_db_path)
|
|
print(f"Created ChromaDB client with default settings at {chroma_db_path}")
|
|
except Exception as e2:
|
|
print(f"Failed to create ChromaDB client: {e2}")
|
|
raise e2
|
|
return self._client
|
|
|
|
@classmethod
|
|
def reset(cls):
|
|
"""Reset the singleton (useful for testing or reinitialization)"""
|
|
if cls._instance and cls._instance._client:
|
|
# Close existing client if possible
|
|
try:
|
|
cls._instance._client.reset()
|
|
except:
|
|
pass
|
|
cls._instance = None
|
|
cls._client = None
|
|
|
|
# Global singleton instance
|
|
chroma_singleton = ChromaDBSingleton() |