fix: add thread-safe write lock to ChromaDB operations
This commit is contained in:
+14
-6
@@ -1,24 +1,32 @@
|
|||||||
"""ChromaDB 数据库层."""
|
"""ChromaDB 数据库层."""
|
||||||
|
import threading
|
||||||
import chromadb
|
import chromadb
|
||||||
from chromadb.api.models.Collection import Collection
|
from chromadb.api.models.Collection import Collection
|
||||||
|
|
||||||
|
|
||||||
class VectorDB:
|
class VectorDB:
|
||||||
"""向量数据库封装."""
|
"""线程安全的向量数据库封装."""
|
||||||
|
|
||||||
def __init__(self, persist_dir: str = "./data"):
|
def __init__(self, persist_dir: str = "./data"):
|
||||||
self.client = chromadb.PersistentClient(path=persist_dir)
|
self.client = chromadb.PersistentClient(path=persist_dir)
|
||||||
|
self._write_lock = threading.Lock()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def write_lock(self) -> threading.Lock:
|
||||||
|
"""获取写锁, 供外部在 add/delete/update 操作时使用."""
|
||||||
|
return self._write_lock
|
||||||
|
|
||||||
def get_or_create_collection(self, name: str) -> Collection:
|
def get_or_create_collection(self, name: str) -> Collection:
|
||||||
"""获取或创建 collection."""
|
"""获取或创建 collection."""
|
||||||
return self.client.get_or_create_collection(name=name)
|
return self.client.get_or_create_collection(name=name)
|
||||||
|
|
||||||
def delete_collection(self, name: str) -> None:
|
def delete_collection(self, name: str) -> None:
|
||||||
"""删除 collection."""
|
"""删除 collection (线程安全)."""
|
||||||
try:
|
with self._write_lock:
|
||||||
self.client.delete_collection(name=name)
|
try:
|
||||||
except ValueError:
|
self.client.delete_collection(name=name)
|
||||||
pass # collection 不存在则忽略
|
except ValueError:
|
||||||
|
pass # collection 不存在则忽略
|
||||||
|
|
||||||
def close(self) -> None:
|
def close(self) -> None:
|
||||||
"""释放数据库连接."""
|
"""释放数据库连接."""
|
||||||
|
|||||||
+13
-11
@@ -206,12 +206,13 @@ class DocumentIngestor:
|
|||||||
for i, c in enumerate(chunks)
|
for i, c in enumerate(chunks)
|
||||||
]
|
]
|
||||||
|
|
||||||
self.collection.add(
|
with self.db.write_lock:
|
||||||
ids=ids,
|
self.collection.add(
|
||||||
embeddings=embeddings,
|
ids=ids,
|
||||||
documents=texts,
|
embeddings=embeddings,
|
||||||
metadatas=metadatas,
|
documents=texts,
|
||||||
)
|
metadatas=metadatas,
|
||||||
|
)
|
||||||
|
|
||||||
return len(chunks)
|
return len(chunks)
|
||||||
|
|
||||||
@@ -226,10 +227,11 @@ class DocumentIngestor:
|
|||||||
def _remove_by_source(self, file_name: str) -> None:
|
def _remove_by_source(self, file_name: str) -> None:
|
||||||
"""按 source_file 删除已有 chunks."""
|
"""按 source_file 删除已有 chunks."""
|
||||||
try:
|
try:
|
||||||
existing = self.collection.get(
|
with self.db.write_lock:
|
||||||
where={"source_file": file_name}
|
existing = self.collection.get(
|
||||||
)
|
where={"source_file": file_name}
|
||||||
if existing and existing["ids"]:
|
)
|
||||||
self.collection.delete(ids=existing["ids"])
|
if existing and existing["ids"]:
|
||||||
|
self.collection.delete(ids=existing["ids"])
|
||||||
except Exception:
|
except Exception:
|
||||||
pass # collection 为空时 get 可能抛异常
|
pass # collection 为空时 get 可能抛异常
|
||||||
|
|||||||
+8
-7
@@ -75,14 +75,15 @@ class Searcher:
|
|||||||
return sorted(sources)
|
return sorted(sources)
|
||||||
|
|
||||||
def delete_by_source(self, file_name: str) -> bool:
|
def delete_by_source(self, file_name: str) -> bool:
|
||||||
"""按文件名删除文档."""
|
"""按文件名删除文档 (线程安全)."""
|
||||||
try:
|
try:
|
||||||
existing = self.collection.get(
|
with self.db.write_lock:
|
||||||
where={"source_file": file_name}
|
existing = self.collection.get(
|
||||||
)
|
where={"source_file": file_name}
|
||||||
if existing and existing["ids"]:
|
)
|
||||||
self.collection.delete(ids=existing["ids"])
|
if existing and existing["ids"]:
|
||||||
return True
|
self.collection.delete(ids=existing["ids"])
|
||||||
|
return True
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
return False
|
return False
|
||||||
|
|||||||
Reference in New Issue
Block a user