Method acquire

app/knowledge_base/kb_cache/base.py:30–40

(self, owner: str = "", msg: str = "")

    @contextmanager
    def acquire(self, owner: str = "", msg: str = "") -> Generator[None, None, FAISS]:
        owner = owner or f"thread {threading.get_native_id()}"
        try:
            self._lock.acquire()
            if self._pool is not None:
                self._pool._cache.move_to_end(self.key)
            logger.info(f"{owner} 开始操作：{self.key}。{msg}")
            yield self._obj
        finally:
            logger.info(f"{owner} 结束操作：{self.key}。{msg}")
            self._lock.release()

    def start_loading(self):
        self._loaded.clear()