拓冰网站定制 
Langchain-Chatchat 知识库缓存层源码解析ThreadSafeObject 与 CachePool 的线程安全设计【免费下载链接】Langchain-ChatchatLangchain-Chatchat原Langchain-ChatGLM基于 Langchain 与 ChatGLM, Qwen 与 Llama 等语言模型的 RAG 与 Agent 应用 | Langchain-Chatchat (formerly langchain-ChatGLM), local knowledge based LLM (like ChatGLM, Qwen and Llama) RAG and Agent app with langchain项目地址: https://gitcode.com/GitHub_Trending/la/Langchain-ChatchatLangchain-Chatchat前身 Langchain-ChatGLM的本地知识库模块在 FAISS 向量库、嵌入模型之上构建了一套可缓存、可并发访问的资源管理层其核心正是 markdown_docs/server/knowledge_base/kb_cache/base.md 所记录的ThreadSafeObject与CachePool两个基础类。本文以该文档为主线结合当前仓库源码逐层拆解这两个类在知识库检索、文档入库、临时文件对话等场景中的线程安全机制、LRU 缓存策略与加载状态同步方案帮助你理解知识库向量库为什么可以在多线程/多请求下安全复用以及如何基于这套抽象扩展自己的缓存对象。一、为什么知识库层需要一套线程安全的缓存对象Langchain-Chatchat 的 API 服务如 chat_routes.py、kb_routes.py会同时处理多个对话与知识库操作请求。知识库检索链路中存在两类典型共享资源FAISS 向量库vector store一个知识库对应一份落盘索引index.faiss在服务进程内常驻缓存多个请求会对同一份向量库并发执行相似度检索、增删文档嵌入模型embeddings模型加载开销大、显存/内存占用高只应被加载一次并被所有知识库共享。如果每个请求都重新读盘加载向量库或对共享对象不加同步地写操作就会出现严重的性能问题与数据竞争。因此knowledge_base/kb_cache目录给出了统一的解法把裸资源包装成线程安全的缓存对象ThreadSafeObject再用线程安全的缓存池CachePool统一管理它们的生命周期、数量上限与淘汰策略。该目录由两个模块构成faiss_cache.md 有对应说明文件职责kb_cache/base.py定义通用抽象ThreadSafeObject、CachePoolkb_cache/faiss_cache.py面向 FAISS 的具体落地ThreadSafeFaiss、KBFaissPool、MemoFaissPool及两个模块级单例二、ThreadSafeObject给任意对象穿上锁 加载状态ThreadSafeObject位于 kb_cache/base.py#L15-L61是对被缓存对象本身的一层线程安全封装。它不关心对象具体是什么类型——可以是 FAISS 向量库、嵌入模型也可以是任何自定义业务对象。2.1 实例属性一览属性类型说明_objAny被封装的实际对象默认为None任何类型_keystr或tuple对象的唯一标识键。知识库 FAISS 场景下使用(kb_name, vector_name)元组比拼接字符串更规范_poolCachePool该对象所属的缓存池默认为None用于在acquire时联动 LRU_lockthreading.RLock可重入锁确保对_obj的并发访问安全_loadedthreading.Event加载状态事件用于加载完成通知与等待__init__kb_cache/base.py#L16-L23会依次完成上述五个属性的初始化。其中选用RLock可重入锁而非普通Lock是刻意的设计向量库的加载/写入方法存在嵌套调用同一把锁的可能可重入锁允许同一线程多次安全进入临界区避免自死锁。2.2 acquire推荐使用方式acquirekb_cache/base.py#L33-L44是一个基于contextmanager的上下文管理器它把加锁 → 使用 → 释放封装成with语句contextmanager def acquire(self, owner: str , msg: str ) - Generator[None, None, FAISS]: owner owner or fthread {threading.get_native_id()} try: self._lock.acquire() if self._pool is not None: self._pool._cache.move_to_end(self.key) # 命中即刷新 LRU 顺序 logger.debug(f{owner} 开始操作{self.key}。{msg}) yield self._obj finally: logger.debug(f{owner} 结束操作{self.key}。{msg}) self._lock.release()其要点有三owner与msgowner默认取当前线程原生 IDthreading.get_native_id()用于日志中追踪哪个线程正在操作哪个资源msg可附带业务说明。仅当log_verbose开启时 debug 日志才有意义。LRU 联动若对象已挂入缓存池_pool非空每次 acquire 都会把自身的_key通过move_to_end移到OrderedDict末尾配合CachePool的淘汰逻辑形成最近最常使用LRU语义。yield self._objwith块内拿到的是被封装对象本体可安全执行读写无论块内是否抛异常finally都会释放锁不会死锁。仓库中的真实调用范式随处可见例如 file_chat.py#L105-L106with memo_faiss_pool.load_vector_store(kb_nameid).acquire() as vs: vs.add_documents(documents)2.3 加载状态机start_loading / finish_loading / wait_for_loading这三者共同构成延迟加载 就绪等待机制kb_cache/base.py#L46-L53底层依赖threading.Eventstart_loading()调用self._loaded.clear()将对象标记为未加载完成。通常在真正开始从磁盘/远端加载资源之前调用finish_loading()调用self._loaded.set()通知所有等待者加载完成可以安全使用wait_for_loading()调用self._loaded.wait()若尚未set调用线程会阻塞直至加载完成。这套状态机的价值在于第一个请求触发加载时其它并发请求不必在锁上傻等整个 I/O 过程而是先拿到对象引用后阻塞在事件等待上一旦加载线程finish_loading所有等待者同时被唤醒。结合CachePool.get的实现可以保证从池中取出的对象永远处于可用状态。2.4 obj 属性getter/setter 与线程安全约定property def obj(self): return self._obj obj.setter def obj(self, val: Any): self._obj val读取用属性obj写入务必走obj的 setterkb_cache/base.py#L55-L61。加载线程在拿到向量库/嵌入模型后通过item.obj vector_store把资源塞进对象然后再调用finish_loading()发布就绪信号——这种先赋值、后置位的顺序保证了等待线程不会读到半初始化状态。2.5 key 与reprkey属性kb_cache/base.py#L29-L31返回_key是所有日志与缓存定位的基础__repr__kb_cache/base.py#L25-L27输出形如ThreadSafeObject: key: example_key, obj: example_object子类可覆写以携带更多上下文例如ThreadSafeFaiss的__repr__额外打印了文档数量docs_count。三、CachePool带数量上限与 FIFO 淘汰的线程安全缓存池CachePool位于 kb_cache/base.py#L64-L102以OrderedDict为存储载体统一管理多个ThreadSafeObject实例。3.1 内部结构与初始化属性类型说明_cache_numint缓存容量上限-1表示不限制数量_cacheOrderedDict有序缓存字典键为对象_key值为对象本身atomicthreading.RLock保护缓存字典结构这一层操作的锁__init__(cache_num-1)kb_cache/base.py#L65-L68默认不限制容量。采用OrderedDict的意义在于同时支持两种淘汰语义按插入序淘汰最早对象FIFO、以及通过move_to_end实现的 LRU 刷新。3.2 五个核心操作方法keys()kb_cache/base.py#L70-L71返回当前所有键的列表快照。典型用途是探活——file_chat.py#L144 在文件对话前先检查knowledge_id in memo_faiss_pool.keys()若临时知识库不存在则返回 404 提示用户先上传文件。_check_count()kb_cache/base.py#L73-L76私有方法当_cache_num 0时循环执行self._cache.popitem(lastFalse)即从头部最早插入移除直到数量回到上限内。这是缓存容量兜底的核心。get(key)kb_cache/base.py#L78-L81命中则先调用cache.wait_for_loading()阻塞等待该对象加载完成再返回未命中返回None。这一步保证了调用方拿到的缓存对象一定已就绪def get(self, key: str) - ThreadSafeObject: if cache : self._cache.get(key): cache.wait_for_loading() return cacheset(key, obj)kb_cache/base.py#L83-L86写入新对象并立即调用_check_count()触发容量检查返回对象本身方便链式调用。pop(keyNone)kb_cache/base.py#L88-L92指定key则移除并返回对应对象不存在返回None而非抛异常不指定则用popitem(lastFalse)弹出最早插入的对象。上层用它完成显式卸载例如上传新临时文档时清理旧的临时向量库file_chat.py#L87-L88、unload_vector_store释放知识库向量库等。3.3 acquire从池中安全取对象CachePool.acquire(key, owner, msg)kb_cache/base.py#L94-L102是上层最常用的入口之一def acquire(self, key, owner: str , msg: str ): cache self.get(key) if cache is None: raise RuntimeError(f请求的资源 {key} 不存在) elif isinstance(cache, ThreadSafeObject): self._cache.move_to_end(key) return cache.acquire(ownerowner, msgmsg) else: return cache它先经get完成等待加载完成再委托给对象的acquire上下文管理器若资源不存在则抛出RuntimeError(请求的资源 {key} 不存在)。因此调用方拿到结果后只需with ... as vs:即可安全使用无需关心加解锁细节。3.4 嵌入模型的并发加载文档记载的扩展职责markdown_docs/server/knowledge_base/kb_cache/base.md还记载了缓存层早期承载的嵌入向量管理职责CachePool提供load_kb_embeddings(kb_name, embed_device, default_embed_model)——先从知识库仓储查询该库绑定的嵌入模型名未绑定则回退到默认嵌入模型若模型属于在线模型由list_online_embed_models判定则返回EmbeddingsFunAdapter否则委托embeddings_pool的load_embeddings按本地模型加载。同时记载了一个继承自CachePool的EmbeddingsPool类其load_embeddings(model, device)曾负责按模型名分流text-embedding-ada-002走OpenAIEmbeddings、含bge-的模型按zh/en语言标识走HuggingFaceBgeEmbeddings并设置对应 query instruction、其余默认HuggingFaceEmbeddings。需要注意的是当前仓库中的 kb_cache/base.py 已收敛为纯粹的ThreadSafeObjectCachePool两个基础类嵌入模型适配器的创建被集中到了chatchat.server.utils的get_Embeddings/get_default_embedding等工具函数中faiss_cache.py#L9-L10 即从该处导入向量库加载时统一通过它们构造 Embeddings。阅读文档时应将这一部分理解为线程安全缓存思想同样适用于嵌入模型这一类重量级共享资源的早期设计印证。四、两级锁 事件缓存层的并发模型是怎么串起来的整个缓存层实际上由三个不同粒度的同步原语协同原语保护对象语义CachePool.atomicRLock缓存字典结构本身增删、容量控制池级互斥ThreadSafeObject._lockRLock被封装对象的读写操作对象级互斥ThreadSafeObject._loadedEvent对象加载完成状态就绪通知/等待以最复杂的KBFaissPool.load_vector_storefaiss_cache.py#L91-L142为例可以看到池锁先行占位、对象锁内初始化、事件最后置位的完整流程先self.atomic.acquire()抢占池锁防止多个线程同时发现缓存未命中并重复加载同一向量库用(kb_name, vector_name)复合键查缓存未命中则构造ThreadSafeFaiss并self.set(...)占位在with item.acquire(msg初始化)中释放池锁并开始真正加载若磁盘index.faiss存在则FAISS.load_local(...)开启normalize_L2True、allow_dangerous_deserializationTrue否则按create参数决定是新建空库并save_local落盘还是抛RuntimeError(fknowledge base {kb_name} not exist.)用item.obj vector_store写入资源最后item.finish_loading()唤醒所有在get/wait_for_loading上等待的线程。异常路径同样有兜底一旦初始化失败会检查池锁是否还由本线程持有locked标志是则释放避免锁泄漏随后统一抛出向量库 {kb_name} 加载失败。。五、落地到 FAISSThreadSafeFaiss、KBFaissPool 与 MemoFaissPool5.1 ThreadSafeFaiss带文档统计与落盘能力的子类faiss_cache.py#L27-L51 在ThreadSafeObject之上扩展了三个方法docs_count()返回self._obj.docstore._dict的长度即当前向量库内的文档数save(path, create_pathTrue)在acquire临界区内先建目录再调save_local落盘clear()在acquire临界区内收集全部 doc id 后执行vs.delete(ids)并断言清空成功。此外该模块对 FAISS 的InMemoryDocstore.search做了 monkey-patchfaiss_cache.py#L14-L24使按 id 取回文档时能把 id 写回Document.metadata[id]便于上层跟踪文档与向量的对应关系。5.2 两个 FaissPool 子类与两个全局单例_FaissPool提供空库创建工具new_vector_store/new_temp_vector_store先用一个占位 Document 初始化 FAISS 索引再删除它从而得到无文档但结构完整的空向量库normalize_L2True。KBFaissPoolfaiss_cache.py#L90-L142面向持久化知识库。键为(kb_name, vector_name)其中vector_name默认为嵌入模型名把:替换成_如bge-large-zh类模型名库文件按get_vs_path(kb_name, vector_name)落盘于知识库目录下的vector_store/{vector_name}/。知识库的增删文档、检索、清库都经由此池见 faiss_kb_service.py如load_vector_store在 faiss_kb_service.py#L31-L36do_clear_vs在池锁内pop复合键并清空磁盘目录。知识库文档摘要向量库summary_vector_store也复用了同一个kb_faiss_pool见 kb_summary/base.py#L44-L50。MemoFaissPoolfaiss_cache.py#L145-L169面向临时内存向量库文件对话上传场景键直接为临时目录 id向量库不落盘。模块尾部定义了供全局共享的两个单例faiss_cache.py#L172-L173kb_faiss_pool KBFaissPool(cache_numSettings.kb_settings.CACHED_VS_NUM) memo_faiss_pool MemoFaissPool(cache_numSettings.kb_settings.CACHED_MEMO_VS_NUM)对应容量上限配置定义于 settings.py 的 KBSettings配置项默认值含义CACHED_VS_NUM1常驻缓存的知识库 FAISS 向量库数量上限CACHED_MEMO_VS_NUM10临时内存向量库数量上限服务于文件对话容量超限时由CachePool._check_count自动淘汰最早添加的对象保证服务不会因缓存无限增长而耗尽内存。六、真实调用链缓存抽象在检索与文档写入中如何被消费6.1 持久化知识库检索链路FaissKBService.do_searchfaiss_kb_service.py#L66-L79展示了一条最小可复制的用法先self.load_vector_store()拿到ThreadSafeFaiss再用with ... .acquire() as vs进入临界区执行相似度检索def do_search(self, query, top_k, score_threshold...): with self.load_vector_store().acquire() as vs: retriever get_Retriever(ensemble).from_vectorstore(vs, top_ktop_k, score_thresholdscore_threshold) docs retriever.get_relevant_documents(query) return docs文档入库do_add_doc与删除do_delete_doc同样在acquire内操作同一份向量库并在操作后save_local(self.vs_path)写回磁盘可通过not_refresh_vs_cache跳过写盘以批量提速。6.2 文件对话的临时向量库链路文件对话使用独立的内存池memo_faiss_pool完整生命周期为上传解析upload_temp_docsfile_chat.py#L76-L110如有prev_id先memo_faiss_pool.pop(prev_id)清理上一批随后with memo_faiss_pool.load_vector_store(kb_nameid).acquire() as vs: vs.add_documents(documents)把切分后的文档向量化进内存库对话检索file_chat内部file_chat.py#L183-L186用memo_faiss_pool.acquire(knowledge_id)取出向量库做向量相似度检索similarity_search_with_score_by_vector临时文档检索接口search_temp_docskb_doc_api.py#L39-L45也复用同一套acquire语义。6.3 自定义缓存对象的最小示例若要把自定义资源接入该缓存体系参照ThreadSafeFaiss的写法即可from chatchat.server.knowledge_base.kb_cache.base import ThreadSafeObject, CachePool class MyResource: def heavy_load(self): ... class ThreadSafeMyResource(ThreadSafeObject): def __init__(self, key, poolNone): super().__init__(keykey, objNone, poolpool) class MyResourcePool(CachePool): def load(self, key): self.atomic.acquire() try: cache self.get(key) if cache is None: item ThreadSafeMyResource(key, poolself) self.set(key, item) with item.acquire(msg初始化): self.atomic.release() # 代价高昂的加载/构建过程 item.obj MyResource().heavy_load() item.finish_loading() return self.get(key) except Exception: # 依据锁是否已释放决定是否需要兜底释放 raise核心纪律只有三条先占位再加载、用item.obj ...写入资源、最后finish_loading()发布就绪——这样所有并发调用方都能安全地拿到完整初始化后的对象。七、设计要点与注意事项小结回顾整个kb_cache层可以总结出值得复用的几点工程经验双层互斥池级atomic锁只管缓存容器对象级_lock管内容两者都选用可重入锁容忍嵌套临界区。例如KBFaissPool.load_vector_store中先占池锁、进入item.acquire后再释放池锁正依赖 RLock 的持有语义不会错乱。加载与使用解耦把加载中状态建模为threading.Event让等待方在事件上阻塞而非长时间霸占互斥锁兼顾正确性与并发吞吐。有序字典承担双重淘汰职责OrderedDict既支持按插入序 FIFO 淘汰popitem(lastFalse)又支持按访问序刷新 LRUmove_to_end一套数据结构同时满足容量上限与热点保护。以with为使用契约任何从acquire取得资源的地方都应在with块内完成操作由finally保证锁释放杜绝死锁从CachePool.acquire取不存在的键会抛RuntimeError上层需据此做好资源探活如 file_chat.py#L144-L150 在检索前先检查键是否存在。容量与内存强相关向量库与嵌入模型都是大内存对象务必通过CACHED_VS_NUM/CACHED_MEMO_VS_NUM控制常驻数量避免缓存淘汰策略形同虚设导致内存超限。围绕该缓存层的回归测试可参考 tests/kb_vector_db/test_faiss_kb.py。若要继续深入下一篇可直接阅读记录具体 FAISS 池落盘逻辑的 markdown_docs/server/knowledge_base/kb_cache/faiss_cache.md以及其对应源码 kb_cache/faiss_cache.py。【免费下载链接】Langchain-ChatchatLangchain-Chatchat原Langchain-ChatGLM基于 Langchain 与 ChatGLM, Qwen 与 Llama 等语言模型的 RAG 与 Agent 应用 | Langchain-Chatchat (formerly langchain-ChatGLM), local knowledge based LLM (like ChatGLM, Qwen and Llama) RAG and Agent app with langchain项目地址: https://gitcode.com/GitHub_Trending/la/Langchain-Chatchat创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Langchain-Chatchat 知识库缓存层源码解析：ThreadSafeObject 与 CachePool 的线程安全设计
Langchain-Chatchat 知识库缓存层源码解析ThreadSafeObject 与 CachePool 的线程安全设计【免费下载链接】Langchain-ChatchatLangchain-Chatchat原Langchain-ChatGLM基于 Langchain 与 ChatGLM, Qwen 与 Llama 等语言模型的 RAG 与 Agent 应用 | Langchain-Chatchat (formerly langchain-ChatGLM), local knowledge based LLM (like ChatGLM, Qwen and Llama) RAG and Agent app with langchain项目地址: https://gitcode.com/GitHub_Trending/la/Langchain-ChatchatLangchain-Chatchat前身 Langchain-ChatGLM的本地知识库模块在 FAISS 向量库、嵌入模型之上构建了一套可缓存、可并发访问的资源管理层其核心正是 markdown_docs/server/knowledge_base/kb_cache/base.md 所记录的ThreadSafeObject与CachePool两个基础类。本文以该文档为主线结合当前仓库源码逐层拆解这两个类在知识库检索、文档入库、临时文件对话等场景中的线程安全机制、LRU 缓存策略与加载状态同步方案帮助你理解知识库向量库为什么可以在多线程/多请求下安全复用以及如何基于这套抽象扩展自己的缓存对象。一、为什么知识库层需要一套线程安全的缓存对象Langchain-Chatchat 的 API 服务如 chat_routes.py、kb_routes.py会同时处理多个对话与知识库操作请求。知识库检索链路中存在两类典型共享资源FAISS 向量库vector store一个知识库对应一份落盘索引index.faiss在服务进程内常驻缓存多个请求会对同一份向量库并发执行相似度检索、增删文档嵌入模型embeddings模型加载开销大、显存/内存占用高只应被加载一次并被所有知识库共享。如果每个请求都重新读盘加载向量库或对共享对象不加同步地写操作就会出现严重的性能问题与数据竞争。因此knowledge_base/kb_cache目录给出了统一的解法把裸资源包装成线程安全的缓存对象ThreadSafeObject再用线程安全的缓存池CachePool统一管理它们的生命周期、数量上限与淘汰策略。该目录由两个模块构成faiss_cache.md 有对应说明文件职责kb_cache/base.py定义通用抽象ThreadSafeObject、CachePoolkb_cache/faiss_cache.py面向 FAISS 的具体落地ThreadSafeFaiss、KBFaissPool、MemoFaissPool及两个模块级单例二、ThreadSafeObject给任意对象穿上锁 加载状态ThreadSafeObject位于 kb_cache/base.py#L15-L61是对被缓存对象本身的一层线程安全封装。它不关心对象具体是什么类型——可以是 FAISS 向量库、嵌入模型也可以是任何自定义业务对象。2.1 实例属性一览属性类型说明_objAny被封装的实际对象默认为None任何类型_keystr或tuple对象的唯一标识键。知识库 FAISS 场景下使用(kb_name, vector_name)元组比拼接字符串更规范_poolCachePool该对象所属的缓存池默认为None用于在acquire时联动 LRU_lockthreading.RLock可重入锁确保对_obj的并发访问安全_loadedthreading.Event加载状态事件用于加载完成通知与等待__init__kb_cache/base.py#L16-L23会依次完成上述五个属性的初始化。其中选用RLock可重入锁而非普通Lock是刻意的设计向量库的加载/写入方法存在嵌套调用同一把锁的可能可重入锁允许同一线程多次安全进入临界区避免自死锁。2.2 acquire推荐使用方式acquirekb_cache/base.py#L33-L44是一个基于contextmanager的上下文管理器它把加锁 → 使用 → 释放封装成with语句contextmanager def acquire(self, owner: str , msg: str ) - Generator[None, None, FAISS]: owner owner or fthread {threading.get_native_id()} try: self._lock.acquire() if self._pool is not None: self._pool._cache.move_to_end(self.key) # 命中即刷新 LRU 顺序 logger.debug(f{owner} 开始操作{self.key}。{msg}) yield self._obj finally: logger.debug(f{owner} 结束操作{self.key}。{msg}) self._lock.release()其要点有三owner与msgowner默认取当前线程原生 IDthreading.get_native_id()用于日志中追踪哪个线程正在操作哪个资源msg可附带业务说明。仅当log_verbose开启时 debug 日志才有意义。LRU 联动若对象已挂入缓存池_pool非空每次 acquire 都会把自身的_key通过move_to_end移到OrderedDict末尾配合CachePool的淘汰逻辑形成最近最常使用LRU语义。yield self._objwith块内拿到的是被封装对象本体可安全执行读写无论块内是否抛异常finally都会释放锁不会死锁。仓库中的真实调用范式随处可见例如 file_chat.py#L105-L106with memo_faiss_pool.load_vector_store(kb_nameid).acquire() as vs: vs.add_documents(documents)2.3 加载状态机start_loading / finish_loading / wait_for_loading这三者共同构成延迟加载 就绪等待机制kb_cache/base.py#L46-L53底层依赖threading.Eventstart_loading()调用self._loaded.clear()将对象标记为未加载完成。通常在真正开始从磁盘/远端加载资源之前调用finish_loading()调用self._loaded.set()通知所有等待者加载完成可以安全使用wait_for_loading()调用self._loaded.wait()若尚未set调用线程会阻塞直至加载完成。这套状态机的价值在于第一个请求触发加载时其它并发请求不必在锁上傻等整个 I/O 过程而是先拿到对象引用后阻塞在事件等待上一旦加载线程finish_loading所有等待者同时被唤醒。结合CachePool.get的实现可以保证从池中取出的对象永远处于可用状态。2.4 obj 属性getter/setter 与线程安全约定property def obj(self): return self._obj obj.setter def obj(self, val: Any): self._obj val读取用属性obj写入务必走obj的 setterkb_cache/base.py#L55-L61。加载线程在拿到向量库/嵌入模型后通过item.obj vector_store把资源塞进对象然后再调用finish_loading()发布就绪信号——这种先赋值、后置位的顺序保证了等待线程不会读到半初始化状态。2.5 key 与reprkey属性kb_cache/base.py#L29-L31返回_key是所有日志与缓存定位的基础__repr__kb_cache/base.py#L25-L27输出形如ThreadSafeObject: key: example_key, obj: example_object子类可覆写以携带更多上下文例如ThreadSafeFaiss的__repr__额外打印了文档数量docs_count。三、CachePool带数量上限与 FIFO 淘汰的线程安全缓存池CachePool位于 kb_cache/base.py#L64-L102以OrderedDict为存储载体统一管理多个ThreadSafeObject实例。3.1 内部结构与初始化属性类型说明_cache_numint缓存容量上限-1表示不限制数量_cacheOrderedDict有序缓存字典键为对象_key值为对象本身atomicthreading.RLock保护缓存字典结构这一层操作的锁__init__(cache_num-1)kb_cache/base.py#L65-L68默认不限制容量。采用OrderedDict的意义在于同时支持两种淘汰语义按插入序淘汰最早对象FIFO、以及通过move_to_end实现的 LRU 刷新。3.2 五个核心操作方法keys()kb_cache/base.py#L70-L71返回当前所有键的列表快照。典型用途是探活——file_chat.py#L144 在文件对话前先检查knowledge_id in memo_faiss_pool.keys()若临时知识库不存在则返回 404 提示用户先上传文件。_check_count()kb_cache/base.py#L73-L76私有方法当_cache_num 0时循环执行self._cache.popitem(lastFalse)即从头部最早插入移除直到数量回到上限内。这是缓存容量兜底的核心。get(key)kb_cache/base.py#L78-L81命中则先调用cache.wait_for_loading()阻塞等待该对象加载完成再返回未命中返回None。这一步保证了调用方拿到的缓存对象一定已就绪def get(self, key: str) - ThreadSafeObject: if cache : self._cache.get(key): cache.wait_for_loading() return cacheset(key, obj)kb_cache/base.py#L83-L86写入新对象并立即调用_check_count()触发容量检查返回对象本身方便链式调用。pop(keyNone)kb_cache/base.py#L88-L92指定key则移除并返回对应对象不存在返回None而非抛异常不指定则用popitem(lastFalse)弹出最早插入的对象。上层用它完成显式卸载例如上传新临时文档时清理旧的临时向量库file_chat.py#L87-L88、unload_vector_store释放知识库向量库等。3.3 acquire从池中安全取对象CachePool.acquire(key, owner, msg)kb_cache/base.py#L94-L102是上层最常用的入口之一def acquire(self, key, owner: str , msg: str ): cache self.get(key) if cache is None: raise RuntimeError(f请求的资源 {key} 不存在) elif isinstance(cache, ThreadSafeObject): self._cache.move_to_end(key) return cache.acquire(ownerowner, msgmsg) else: return cache它先经get完成等待加载完成再委托给对象的acquire上下文管理器若资源不存在则抛出RuntimeError(请求的资源 {key} 不存在)。因此调用方拿到结果后只需with ... as vs:即可安全使用无需关心加解锁细节。3.4 嵌入模型的并发加载文档记载的扩展职责markdown_docs/server/knowledge_base/kb_cache/base.md还记载了缓存层早期承载的嵌入向量管理职责CachePool提供load_kb_embeddings(kb_name, embed_device, default_embed_model)——先从知识库仓储查询该库绑定的嵌入模型名未绑定则回退到默认嵌入模型若模型属于在线模型由list_online_embed_models判定则返回EmbeddingsFunAdapter否则委托embeddings_pool的load_embeddings按本地模型加载。同时记载了一个继承自CachePool的EmbeddingsPool类其load_embeddings(model, device)曾负责按模型名分流text-embedding-ada-002走OpenAIEmbeddings、含bge-的模型按zh/en语言标识走HuggingFaceBgeEmbeddings并设置对应 query instruction、其余默认HuggingFaceEmbeddings。需要注意的是当前仓库中的 kb_cache/base.py 已收敛为纯粹的ThreadSafeObjectCachePool两个基础类嵌入模型适配器的创建被集中到了chatchat.server.utils的get_Embeddings/get_default_embedding等工具函数中faiss_cache.py#L9-L10 即从该处导入向量库加载时统一通过它们构造 Embeddings。阅读文档时应将这一部分理解为线程安全缓存思想同样适用于嵌入模型这一类重量级共享资源的早期设计印证。四、两级锁 事件缓存层的并发模型是怎么串起来的整个缓存层实际上由三个不同粒度的同步原语协同原语保护对象语义CachePool.atomicRLock缓存字典结构本身增删、容量控制池级互斥ThreadSafeObject._lockRLock被封装对象的读写操作对象级互斥ThreadSafeObject._loadedEvent对象加载完成状态就绪通知/等待以最复杂的KBFaissPool.load_vector_storefaiss_cache.py#L91-L142为例可以看到池锁先行占位、对象锁内初始化、事件最后置位的完整流程先self.atomic.acquire()抢占池锁防止多个线程同时发现缓存未命中并重复加载同一向量库用(kb_name, vector_name)复合键查缓存未命中则构造ThreadSafeFaiss并self.set(...)占位在with item.acquire(msg初始化)中释放池锁并开始真正加载若磁盘index.faiss存在则FAISS.load_local(...)开启normalize_L2True、allow_dangerous_deserializationTrue否则按create参数决定是新建空库并save_local落盘还是抛RuntimeError(fknowledge base {kb_name} not exist.)用item.obj vector_store写入资源最后item.finish_loading()唤醒所有在get/wait_for_loading上等待的线程。异常路径同样有兜底一旦初始化失败会检查池锁是否还由本线程持有locked标志是则释放避免锁泄漏随后统一抛出向量库 {kb_name} 加载失败。。五、落地到 FAISSThreadSafeFaiss、KBFaissPool 与 MemoFaissPool5.1 ThreadSafeFaiss带文档统计与落盘能力的子类faiss_cache.py#L27-L51 在ThreadSafeObject之上扩展了三个方法docs_count()返回self._obj.docstore._dict的长度即当前向量库内的文档数save(path, create_pathTrue)在acquire临界区内先建目录再调save_local落盘clear()在acquire临界区内收集全部 doc id 后执行vs.delete(ids)并断言清空成功。此外该模块对 FAISS 的InMemoryDocstore.search做了 monkey-patchfaiss_cache.py#L14-L24使按 id 取回文档时能把 id 写回Document.metadata[id]便于上层跟踪文档与向量的对应关系。5.2 两个 FaissPool 子类与两个全局单例_FaissPool提供空库创建工具new_vector_store/new_temp_vector_store先用一个占位 Document 初始化 FAISS 索引再删除它从而得到无文档但结构完整的空向量库normalize_L2True。KBFaissPoolfaiss_cache.py#L90-L142面向持久化知识库。键为(kb_name, vector_name)其中vector_name默认为嵌入模型名把:替换成_如bge-large-zh类模型名库文件按get_vs_path(kb_name, vector_name)落盘于知识库目录下的vector_store/{vector_name}/。知识库的增删文档、检索、清库都经由此池见 faiss_kb_service.py如load_vector_store在 faiss_kb_service.py#L31-L36do_clear_vs在池锁内pop复合键并清空磁盘目录。知识库文档摘要向量库summary_vector_store也复用了同一个kb_faiss_pool见 kb_summary/base.py#L44-L50。MemoFaissPoolfaiss_cache.py#L145-L169面向临时内存向量库文件对话上传场景键直接为临时目录 id向量库不落盘。模块尾部定义了供全局共享的两个单例faiss_cache.py#L172-L173kb_faiss_pool KBFaissPool(cache_numSettings.kb_settings.CACHED_VS_NUM) memo_faiss_pool MemoFaissPool(cache_numSettings.kb_settings.CACHED_MEMO_VS_NUM)对应容量上限配置定义于 settings.py 的 KBSettings配置项默认值含义CACHED_VS_NUM1常驻缓存的知识库 FAISS 向量库数量上限CACHED_MEMO_VS_NUM10临时内存向量库数量上限服务于文件对话容量超限时由CachePool._check_count自动淘汰最早添加的对象保证服务不会因缓存无限增长而耗尽内存。六、真实调用链缓存抽象在检索与文档写入中如何被消费6.1 持久化知识库检索链路FaissKBService.do_searchfaiss_kb_service.py#L66-L79展示了一条最小可复制的用法先self.load_vector_store()拿到ThreadSafeFaiss再用with ... .acquire() as vs进入临界区执行相似度检索def do_search(self, query, top_k, score_threshold...): with self.load_vector_store().acquire() as vs: retriever get_Retriever(ensemble).from_vectorstore(vs, top_ktop_k, score_thresholdscore_threshold) docs retriever.get_relevant_documents(query) return docs文档入库do_add_doc与删除do_delete_doc同样在acquire内操作同一份向量库并在操作后save_local(self.vs_path)写回磁盘可通过not_refresh_vs_cache跳过写盘以批量提速。6.2 文件对话的临时向量库链路文件对话使用独立的内存池memo_faiss_pool完整生命周期为上传解析upload_temp_docsfile_chat.py#L76-L110如有prev_id先memo_faiss_pool.pop(prev_id)清理上一批随后with memo_faiss_pool.load_vector_store(kb_nameid).acquire() as vs: vs.add_documents(documents)把切分后的文档向量化进内存库对话检索file_chat内部file_chat.py#L183-L186用memo_faiss_pool.acquire(knowledge_id)取出向量库做向量相似度检索similarity_search_with_score_by_vector临时文档检索接口search_temp_docskb_doc_api.py#L39-L45也复用同一套acquire语义。6.3 自定义缓存对象的最小示例若要把自定义资源接入该缓存体系参照ThreadSafeFaiss的写法即可from chatchat.server.knowledge_base.kb_cache.base import ThreadSafeObject, CachePool class MyResource: def heavy_load(self): ... class ThreadSafeMyResource(ThreadSafeObject): def __init__(self, key, poolNone): super().__init__(keykey, objNone, poolpool) class MyResourcePool(CachePool): def load(self, key): self.atomic.acquire() try: cache self.get(key) if cache is None: item ThreadSafeMyResource(key, poolself) self.set(key, item) with item.acquire(msg初始化): self.atomic.release() # 代价高昂的加载/构建过程 item.obj MyResource().heavy_load() item.finish_loading() return self.get(key) except Exception: # 依据锁是否已释放决定是否需要兜底释放 raise核心纪律只有三条先占位再加载、用item.obj ...写入资源、最后finish_loading()发布就绪——这样所有并发调用方都能安全地拿到完整初始化后的对象。七、设计要点与注意事项小结回顾整个kb_cache层可以总结出值得复用的几点工程经验双层互斥池级atomic锁只管缓存容器对象级_lock管内容两者都选用可重入锁容忍嵌套临界区。例如KBFaissPool.load_vector_store中先占池锁、进入item.acquire后再释放池锁正依赖 RLock 的持有语义不会错乱。加载与使用解耦把加载中状态建模为threading.Event让等待方在事件上阻塞而非长时间霸占互斥锁兼顾正确性与并发吞吐。有序字典承担双重淘汰职责OrderedDict既支持按插入序 FIFO 淘汰popitem(lastFalse)又支持按访问序刷新 LRUmove_to_end一套数据结构同时满足容量上限与热点保护。以with为使用契约任何从acquire取得资源的地方都应在with块内完成操作由finally保证锁释放杜绝死锁从CachePool.acquire取不存在的键会抛RuntimeError上层需据此做好资源探活如 file_chat.py#L144-L150 在检索前先检查键是否存在。容量与内存强相关向量库与嵌入模型都是大内存对象务必通过CACHED_VS_NUM/CACHED_MEMO_VS_NUM控制常驻数量避免缓存淘汰策略形同虚设导致内存超限。围绕该缓存层的回归测试可参考 tests/kb_vector_db/test_faiss_kb.py。若要继续深入下一篇可直接阅读记录具体 FAISS 池落盘逻辑的 markdown_docs/server/knowledge_base/kb_cache/faiss_cache.md以及其对应源码 kb_cache/faiss_cache.py。【免费下载链接】Langchain-ChatchatLangchain-Chatchat原Langchain-ChatGLM基于 Langchain 与 ChatGLM, Qwen 与 Llama 等语言模型的 RAG 与 Agent 应用 | Langchain-Chatchat (formerly langchain-ChatGLM), local knowledge based LLM (like ChatGLM, Qwen and Llama) RAG and Agent app with langchain项目地址: https://gitcode.com/GitHub_Trending/la/Langchain-Chatchat创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考