Langchain-Chatchat文档解析资源隔离方案设计-编程实验室

Langchain-Chatchat文档解析资源隔离方案设计

在企业对数据隐私要求日益严苛的今天，将敏感文档上传至公有云进行智能处理已不再可接受。金融、医疗、法律等行业尤其如此——一份合同、一张病历、一个财务报表，都可能涉及核心机密。而与此同时，知识库问答系统的需求却在快速增长：员工希望快速检索内部制度，医生需要辅助查阅诊疗指南，律师期待高效比对判例条文。

这种矛盾催生了一个明确的技术方向：本地化部署 + 强隔离能力的知识库问答系统。

Langchain-Chatchat 正是这一趋势下的代表性开源框架。它结合了 LangChain 的灵活架构与中文语义理解优化，支持完全离线运行。但当多个用户同时使用时，问题也随之而来：如何确保张三上传的财报不会被李四的问题检索到？如何避免王五的大型PDF解析占用全部内存导致他人服务卡顿？

答案在于——文档解析资源的精细化隔离。

我们不妨从一个真实场景切入：某律师事务所部署了一套基于 Langchain-Chatchat 的案件辅助查询系统。三位律师 A、B、C 同时登录并创建独立会话，分别上传各自手头的案件材料。系统必须做到：

A 提问“该案件是否适用诉讼时效中断？”时，只能检索他自己上传的卷宗；
B 解析一份500页的判决书不应影响 C 的实时问答响应速度；
任何一方都无法访问或感知其他人的文件存在。

这不仅仅是功能需求，更是合规底线。

要实现这一点，不能仅靠逻辑判断或权限控制，而是要在存储、内存、上下文三个维度上构建物理级隔离机制。而这正是本文要深入探讨的设计思路。

Langchain 作为整个系统的底层支撑，其模块化设计理念为资源隔离提供了天然便利。它的核心思想是将 AI 应用拆解为可组合的“链”（Chains），每个组件如DocumentLoader、TextSplitter、EmbeddingModel和VectorStore都可以独立配置和替换。

以最典型的问答流程为例：

用户输入问题；
系统通过Retriever在向量数据库中查找相似文本片段；
将原始问题与检索结果拼接成 Prompt；
交给本地 LLM 推理生成回答。

这个过程的关键在于：只要Retriever绑定的是专属向量库，就能天然实现知识边界隔离。

来看一段关键代码：

from langchain.chains import RetrievalQA from langchain.vectorstores import Chroma from langchain.embeddings import HuggingFaceEmbeddings from langchain.llms import CTransformers embeddings = HuggingFaceEmbeddings(model_name="shibing624/text2vec-base-chinese") vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) llm = CTransformers(model="llama-2-7b-chat.ggmlv3.q4_0.bin", model_type="llama") qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), return_source_documents=True )

这段代码看似简单，但其中隐藏着一个重要的设计启示：vectorstore是可以动态创建和销毁的。如果我们为每一个用户会话生成一个独立的vectorstore实例，并将其路径指向专属目录，那么自然就实现了数据层面的硬隔离。

这也正是 Chatchat 系统能够支撑多租户模式的基础所在。

Chatchat 原名 Langchain-ChatGLM，现已发展为通用型本地知识库问答平台。它不仅继承了 LangChain 的灵活性，还针对中文场景做了大量优化：默认集成中文分词策略、预置高性能中文 Embedding 模型（如text2vec-large-chinese）、支持 PDF/Word/Excel/TXT 等多种格式解析。

更重要的是，它的整体架构本身就是面向本地部署设计的：

前端采用 Vue 构建交互界面；
后端基于 FastAPI 提供 RESTful 接口；
所有中间数据（原始文档、分块文本、向量索引）均落盘于本地文件系统；
支持接入 llama.cpp、ChatGLM、Qwen、Baichuan 等主流本地模型。

这意味着，整个处理链条无需依赖外部网络，彻底规避了数据外泄风险。

但在高并发或多用户环境下，仅仅“本地化”还不够。我们必须进一步解决资源争用问题。试想以下几种典型困境：

多个用户同时上传大文件，磁盘 I/O 阻塞；
多个会话加载相同的 Embedding 模型造成内存重复占用；
共享向量库导致跨用户信息泄露；
长时间运行后临时文件堆积，磁盘空间耗尽。

这些问题的本质，都是缺乏有效的资源边界管理。

为此，我们需要一套完整的文档解析资源隔离机制。其核心目标很明确：让每个用户的文档处理过程像在一个“沙箱”中独立运行。

具体来说，可以从三个层级入手：

存储隔离：每人一库，互不相见

最基础也是最关键的一步，就是为每个会话分配独立的存储路径。这包括：

文档存储：/data/sessions/{session_id}/docs/
分块文本：/data/sessions/{session_id}/texts/
向量数据库：/vectorstores/{session_id}

这样，即使两个用户上传了同名文件（例如“合同模板.docx”），也会因路径不同而完全隔离。操作系统级别的文件权限也可进一步加固安全，防止越权读取。

内存隔离：进程或上下文级分离

虽然共享 LLM 可提升资源利用率，但文档解析阶段应尽量避免内存冲突。理想做法是：

使用异步任务队列（如 Celery）调度解析工作；
每个任务在独立线程或轻量进程中执行，避免 GIL 锁竞争；
对于超高安全要求场景，甚至可考虑为每个 session 启动独立容器（如 Docker 或 gVisor sandbox）。

此外，Embedding 模型虽可缓存复用，但也需设置最大并发限制，防止单一用户耗尽显存。

上下文隔离：会话绑定，精准定位

所有操作必须与session_id强绑定。从前端发起请求开始，每一个 API 调用都应携带当前会话标识，后端据此查找对应的向量库实例。

例如，在初始化RetrievalQA时：

qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=session_vectorstore.as_retriever() )

这里的session_vectorstore必须来自当前session_id的上下文环境，绝不允许混用。

为了实现上述机制，我们可以设计一个SessionManager类来统一管理生命周期：

import os import time from uuid import uuid4 from langchain.vectorstores import FAISS from langchain.embeddings import HuggingFaceEmbeddings class SessionManager: def __init__(self, base_dir="./data/sessions"): self.base_dir = base_dir self.sessions = {} def create_session(self): session_id = str(uuid4()) session_path = os.path.join(self.base_dir, session_id) # 创建独立目录结构 os.makedirs(os.path.join(session_path, "docs"), exist_ok=True) os.makedirs(os.path.join(session_path, "texts"), exist_ok=True) vector_path = os.path.join(session_path, "vectorstore") # 初始化专属向量库 embeddings = HuggingFaceEmbeddings(model_name="shibing624/text2vec-base-chinese") vectorstore = FAISS(embedding_function=embeddings) self.sessions[session_id] = { "path": session_path, "vectorstore": vectorstore, "created_at": time.time(), "doc_count": 0, "last_active": time.time() } return session_id def get_vectorstore(self, session_id): if session_id not in self.sessions: raise ValueError("Invalid session ID") self.sessions[session_id]["last_active"] = time.time() # 更新活跃时间 return self.sessions[session_id]["vectorstore"] def cleanup_expired_sessions(self, ttl=86400): # 默认24小时 now = time.time() expired = [ sid for sid, info in self.sessions.items() if now - info["last_active"] > ttl ] for sid in expired: del self.sessions[sid] # 可选：删除对应目录 # shutil.rmtree(os.path.join(self.base_dir, sid))

这个类不仅负责创建会话，还能跟踪活跃状态，并提供自动清理接口。配合定时任务，可有效回收闲置资源，防止系统长期运行后性能下降。

在一个典型的企业部署架构中，这套机制是如何运作的呢？

+------------------+ +----------------------------+ | Web Frontend |<----->| FastAPI Backend (Chatchat)| +------------------+ +-------------+--------------+ | +---------------------------+-+----------------------------+ | | | +----------v----------+ +------------v-------------+ +----------v----------+ | Session-Specific | | Isolated Vector Storage | | Local LLM Runtime | | Document Parsing | | (per session: FAISS/Chroma)| | (e.g., llama.cpp) | +----------+----------+ +----------------------------+ +---------------------+ | +-----v------+ | User File | | Upload & Preprocessing | +------------+

前端接收用户上传文件后，后端根据当前session_id将其保存至专属目录，并触发后台异步解析任务。解析完成后，向量库序列化落盘，支持后续断点续用。提问时，系统仅从此会话对应的向量库中检索，确保知识边界清晰。

这种设计带来了几个显著优势：