ChatGLM本地部署与RAG:私有化知识库问答助理落地

发布时间:2026/11/2 21:40:50
ChatGLM本地部署与RAG:私有化知识库问答助理落地

简介:这份压缩包面向希望上手私有化智能问答系统的开发者与学习者,围绕基于深度学习的文件上传、知识库检索与对话生成展开,结合 Gradio 界面、FAISS 向量检索和 Sentence-Transformers 文本嵌入,可用于搭建本地可运行、数据不出内网的问答助理示例。包内共18个文件,以 py 源码、txt 配置与说明、yaml 环境配置为主,另含 whl 依赖包、sh 脚本、docx/xlsx/pdf 示例文档及 pem 证书文件,压缩包约336KB,便于快速查看项目结构与运行依赖。已有140人学习下载,适合作为智能问答、知识库查询方向的入门实践参考。读者可从中了解界面交互、文档向量化、相似度检索与回答生成的基本链路,并参考示例文档和配置组织方式完成本地调试与二次开发。

1. 从一台不能联网的机器说起:私有化问答助理真正要先定的三件事

某公司内网有几十台服务器,资料分散在共享盘、工单系统和几个 Word 目录里,问一句话要翻半小时。需求很朴素:做一个本地部署的智能问答助理,机器不接外网,资料不出机房。这类项目表面上是「选个大模型 + 套个聊天界面」,实施时真正卡住人的通常是三件事:模型在这张显卡上能否达到可接受的生成速度;知识库更新后多久能被检索到;答案错了能否追溯到原文。这套示例项目按 ChatGLM 本地推理 + 检索增强的路线走了一遍,从硬件清单、切分参数到验收指标都有具体落点,适合要交毕设/课设的学生,也适合被派去做内部知识助手的技术同学直接改。

2. 总体架构设计:四层结构与模型服务的职责边界

2.1 四层划分:谁负责理解,谁负责查资料

很多人第一版会把「检索 + 拼提示词 + 调模型」写在一个chat()函数里,两百行跑通,然后进入长期维护困难期——换个模型要改业务代码,改个切分参数要重启整个服务,出了问题不知道是检索召回错了还是模型在编造答案。我的做法是横向切四层,每层只做一件事,并且能单独替换、单独评测。

层职责典型组件变更频率
接入层会话鉴权、前端交互、SSE 推流Web 前端 + 网关低
编排层会话管理、问题改写、上下文装配、提示词模板业务服务(Python)中
检索层解析、切分、向量化、召回、重排索引服务 + 向量库高
模型层文本生成、向量编码ChatGLM 推理服务 + Embedding 服务低

这样切分后,好处很直接:模型层以 OpenAI 兼容协议对外暴露,编排层只认 HTTP 接口,后面想从 6B 换到更大参数量的版本,编排层一行不改;检索层可以离线跑评测集,把召回率单独拉出来看,不必每次都用「回答得好不好」这种主观标准去反推问题出在哪。

还有一个容易忽略的点是故障隔离。检索层挂了,编排层可以降级成纯模型对话并明确提示「本次未检索到资料」;模型服务重启时,索引和会话数据都还在磁盘上,用户刷新页面就能继续。单进程脚本做不到这个。

2.2 接口契约:三个接口定死前后端

接口不在多,够用且稳定就行。整个系统对外只需要三个接口:会话创建与历史拉取、流式问答、知识库管理。问答接口是整个项目最核心的接口约定,参数一旦定下来,前端、压测脚本、评测脚本都能复用。

# api/routes.py from fastapi import APIRouter from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field router = APIRouter() class ChatRequest(BaseModel): session_id: str = Field(..., description="会话ID,为空则新建") kb_id: str = Field(..., description="知识库ID,支持多库隔离") question: str = Field(..., max_length=1000) top_k: int = Field(5, ge=1, le=10, description="送入提示词的片段数") stream: bool = True @router.post("/api/chat") async def chat(req: ChatRequest): # 编排层只做装配与调度,不在这里写检索细节和模型细节 ctx = await build_context(req.session_id, req.kb_id, req.question, req.top_k) if not req.stream: return await generate_once(ctx) # SSE 推流:先推引用,再推正文,前端可先渲染来源卡片 return StreamingResponse(generate_stream(ctx), media_type="text/event-stream")

top_k默认 5 是有依据的:切分块控制在 500 字上下时,5 个片段的资料预算大约 1800~2200 token,正好卡在 4K 上下文模型的安全区间内;调到 10 会挤掉历史对话的空间,反而让多轮追问变差。kb_id作为必填参数而不是从会话里推断,是为了支持「同一用户切不同知识库提问」这种真实场景,权限校验也挂在这个字段上。

2.3 一次问答的时序与上下文装配

把一次问答拆成固定步骤,后面所有优化都有地方下手:取最近 3 轮对话 → 用问题 + 上一轮问题做检索改写(把「它怎么配置」补全成「XX 服务怎么配置」)→ 向量召回 top 20 → 重排取 top 5 → 按优先级拼接提示词 → 流式生成 → 落库并回填引用。

上下文预算必须提前算清楚,不能等模型报超长再改。我给一个实测可用的分配比例:

组成预算(token)超出时的处理
系统提示词300固定,不裁剪
资料片段1800按重排分从低到高丢弃
历史对话600保留最近 3 轮,更早的做摘要压缩
用户问题200超长直接截断并提示
生成输出512由服务端max_tokens限制

这套预算在 4096 上下文下留了余量,换成 8K 模型时按比例放大资料片段即可。经验上,多轮对话变差多半不是模型记性不好,而是历史把资料挤掉了——第 4 轮开始答案质量下滑,八成是这里出了问题。

2.4 目录结构与配置分层

工程结构按层落目录,配置分三层(默认 → 环境 → 启动参数),避免把显卡号、模型路径写死在代码里:

assistant/ ├── api/ # 接入层:路由、鉴权、SSE ├── core/ # 编排层:会话、改写、上下文、提示词 ├── retrieval/ # 检索层:parser / splitter / embed / search / rerank ├── client/ # 模型层客户端:chat、embedding 封装 ├── configs/ │ ├── default.yaml # 默认配置,入库 │ └── local.yaml # 本机覆盖,不入库 ├── data/ # 原始文档、索引、sqlite 会话库 └── tests/ # 评测集与回归脚本
# configs/default.yaml model: base_url: "http://127.0.0.1:8000/v1" # 本地推理服务,OpenAI 兼容 name: "chatglm-local" max_tokens: 512 temperature: 0.2 # 知识问答要稳,不要发散 retrieval: chunk_size: 500 chunk_overlap: 80 recall_k: 20 rerank_k: 5 score_threshold: 0.35 # 低于此分视为无答案,直接拒答

temperature设 0.2 而不是 0,是因为全 0 在部分推理后端上容易出现重复片段;0.2 在事实性和表达自然度之间比较平衡。score_threshold是最容易被忽略的一个参数,没有它,系统会在资料完全没有覆盖的问题上硬答,而这恰恰是验收时最容易被追问的地方。

3. 模型本地化:ChatGLM 的显存账与推理服务参数

3.1 显存账与量化档位:先算清楚再买卡

本地部署第一个决定不是选哪个模型,而是手里有什么卡。把账算在前面,能省掉大量返工。下面是我按 batch=1、4K 上下文估的区间,实际以下运行时用nvidia-smi盯峰值显存为准——权重之外还有 KV cache 和框架自身的开销,通常会比「模型文件大小」多出 2~4GB。

规模精度权重占用推理峰值(估算)适配显卡
6BFP16约 12GB14~16GB24GB 单卡,充裕
6BINT8约 7GB9~11GB16GB 卡可跑
6BINT4约 4GB6~8GB12GB 卡可跑,速度略降
更大参数量FP1618GB+24GB+24GB 卡吃紧,需 32GB 或双卡

课程设计或毕设阶段,6B + INT4 是最省心的组合:一张消费级卡就能跑,量化后中文表达能力下降有限,答辩演示完全够用。真要做成部门在用的东西,再按并发数往上加——并发是显存杀手,10 路并发的 KV cache 开销可能比权重本身还大。

3.2 推理服务启动参数:把可变项全部外置

推理服务独立进程启动,用 OpenAI 兼容协议对外,这样编排层换成任何同协议的后端都不用改代码:

# 本地推理服务启动(INT4 量化,4K 上下文) python -m vllm.entrypoints.openai.api_server \ --model /data/models/chatglm-local \ --served-model-name chatglm-local \ --quantization gptq \ --dtype float16 \ --max-model-len 4096 \ --gpu-memory-utilization 0.85 \ --max-num-seqs 8 \ --port 8000

几个参数值得单独说:--gpu-memory-utilization 0.85留 15% 给 CUDA 上下文和显存碎片,设到 0.95 常常在跑了几十分钟后 OOM;--max-num-seqs 8是并发上限,超过这个数请求会排队而不是把显存撑爆,8 路对内部工具基本够;--max-model-len一定要显式指定,很多默认值给到 8K 甚至更高,KV cache 按这个长度预分配,显存瞬间就没了。

如果没有 GPU,INT4 的 6B 模型在 CPU 上也能跑,但速度大概是每秒几个字,只适合本地调试接口逻辑,不适合演示问答体验。这一点在开题时就要和需求方对齐,别等到验收现场才发现。

3.3 流式输出与并发控制:让等待变得可接受

本地单卡的实际生成速度通常在每秒 20~40 字之间,一段 200 字的回答需要 5~10 秒。用户能忍受 5 秒的逐字输出,但忍不了 5 秒白屏。所以流式不是可选项,是必需品:

# core/generate.py import asyncio, json from client.llm import chat_stream SEM = asyncio.Semaphore(6) # 全局并发闸门,低于推理服务的 max-num-seqs async def generate_stream(ctx): async with SEM: yield sse_pack("refs", ctx.refs) # 先推引用,前端提前渲染来源 first = True try: async for delta in chat_stream(ctx.messages, timeout=60): if first: log_first_token(ctx.trace_id) # 记录首字延迟,验收要用 first = False yield sse_pack("delta", {"text": delta}) yield sse_pack("done", {"trace_id": ctx.trace_id}) except asyncio.TimeoutError: # 超时不让用户干等,给可执行的提示 yield sse_pack("error", {"msg": "生成超时,请缩短问题或稍后重试"})

并发闸门放在编排层而不是只靠推理服务自己排队,好处是超出的请求在业务层就能拿到明确的排队反馈,而不是连接挂在那里几十秒。trace_id从请求进来就生成,日志、引用、延迟指标全挂在上面,出问题时按这一个 ID 就能把整条链路捞出来。

3.4 提示词模板与多轮上下文裁剪

提示词的质量直接决定答案格式是否稳定。我的模板固定三件事:资料编号、拒答出口、引用要求。

PROMPT = """你是一个内部知识助理。请严格依据下面编号的资料回答问题。 规则: 1. 只使用资料中的信息,不要引入外部知识; 2. 每句结论后必须标注来源编号,如 [1][2]; 3. 资料中没有的信息,直接回答“资料中未涉及该问题”,不要猜测; 4. 用中文回答,条目化,不超过 300 字。 资料: {context} 对话历史: {history} 问题:{question} """

「资料中未涉及」这个出口必须写死在提示词里,并且配合 2.4 里的score_threshold做双重保险——提示词是软约束,阈值是硬拦截。实测只靠提示词,模型仍有 10%~20% 的概率编造;加上召回分数阈值之后,无答案场景的胡编率能压到个位数。历史裁剪按轮次而非 token 做,简单可靠:保留最近 3 轮,更早的合并成一句「此前讨论过 XX 的配置问题」放进历史区,比整段丢弃对多轮追问更友好。

4. 检索链路设计:文档切分、向量化与召回重排

4.1 文档解析与切分:块大小决定答案上限

切分是检索质量的天花板。切得太碎,一段话被拆成两半,模型拿到半截资料只能编;切得太大,一个块里混了三个主题,向量被平均掉,召回就不准。我一般按 500 字、80 字重叠起步,再按文档结构做二次修正:

# retrieval/splitter.py import re def split_by_heading(text: str, max_len: int = 500, overlap: int = 80): # 先按标题层级切,保证一个块不跨章节,语义完整性优先 sections = re.split(r"\n(?=#{1,3}\s)", text) chunks = [] for sec in sections: sec = sec.strip() if not sec: continue if len(sec) <= max_len: chunks.append(sec) # 短章节整体保留,不做无意义切分 continue start = 0 while start < len(sec): end = start + max_len chunks.append(sec[start:end]) start = end - overlap # 重叠区保证跨块句子不被截断 return chunks

overlap给 80 是折中:太小,跨块的句子在检索时匹配不上;太大,索引体积和重复召回都上去了。另外,文档解析阶段必须做清洗——PDF 的页眉页脚、双栏排版的错行、表格被拉平成一行,这三类问题不处理,后面所有参数调优都是白费。表格建议单独抽成「表头 + 行」的自然语言描述再入库,比直接把表格文本塞进块里召回率高得多。

4.2 向量化与索引构建:批量、归一化、带元数据

Embedding 走本地部署,和生成模型分两个进程,互不抢显存。入库时把「向量 + 原文 + 来源 + 版本号」绑在一起,元数据比向量本身更重要:

# retrieval/index.py def build_index(docs, embedder, kb_id: str, version: str): for batch in chunked(docs, 32): # 批大小 32,兼顾吞吐与显存 vecs = embedder.encode([d.text for d in batch], normalize_embeddings=True) for d, v in zip(batch, vecs): store.add( kb_id=kb_id, version=version, # 版本号用于失效旧索引 doc_id=d.doc_id, page=d.page, text=d.text, vector=v, )

normalize_embeddings=True是为了让内积等价于余弦相似度,省掉检索时的重复归一化;version字段解决的是「资料更新了但答案还是旧的」这类常见问题;page保留页码,回答里的[1]才能点开看到原文第几页,这是答辩时最有说服力的一个细节。

4.3 混合检索与召回重排:关键词和语义各管一半

纯向量检索在专有名词、编号、缩写上很容易失效——「工单系统 V2.3 的接口超时阈值」这种问题,关键词匹配往往比语义匹配更准。所以我用向量召回 + 关键词召回各取 20 条,做排名融合,再上重排模型:

def hybrid_search(query, kb_id, top_k=20): vec_hits = vector_store.search(query, kb_id, top_k) kw_hits = bm25_store.search(query, kb_id, top_k) # RRF 融合:按名次而非原始分数加权,避免两路分数量纲不一致 scores = {} for rank, hit in enumerate(vec_hits): scores[hit.id] = scores.get(hit.id, 0) + 1 / (60 + rank) for rank, hit in enumerate(kw_hits): scores[hit.id] = scores.get(hit.id, 0) + 1 / (60 + rank) merged = sorted(scores.items(), key=lambda x: -x[1])[:top_k] return rerank(query, [load_chunk(i) for i, _ in merged]) # 重排取最终 5 条

RRF 里的常数 60 是常用经验值,作用是压平头部名次的差距,让两路结果都能进来。融合之后一定要过一遍重排模型,把 top 20 压到 5 条——重排模型对「问题与片段是否真的相关」的判断比向量相似度准得多,这一步通常能把准确率再拉高 10 个点以上。

4.4 无答案兜底与引用溯源:敢说不知道才可信

一个知识助理的价值,一半在于答得对,另一半在于答不出来时不说错。重排分数低于阈值就返回固定话术,不要交给模型自由发挥:

if not refs or max(r.score for r in refs) < settings.score_threshold: return {"answer": "资料中未涉及该问题,建议补充相关文档后再试。", "refs": []}

阈值需要按自己的语料标定:拿 30~50 条「资料里确实没有」的问题跑一遍,看分数的分布,取能挡掉 90% 的那条线。引用溯源则要求模型输出里的[n]必须能在 refs 里找到对应项,装配阶段做一次校验,找不到的就从答案里剔除该标记,并记一条告警日志——这类日志积累起来,就是下次优化切分和提示词的直接依据。

5. 常见问题与排查:本地部署最容易翻车的五处

5.1 环境与资源类:跑得起来和跑得稳是两回事

现象一:模型能加载,问答跑十几轮后突然 OOM 崩掉。原因通常是 KV cache 按最大上下文长度预分配,多轮对话不断累积历史,显存占用一路上涨直到触顶。 解决办法:推理服务显式限制--max-model-len,编排层强制历史裁剪,并且把--gpu-memory-utilization压到 0.85 以下,留出碎片空间。压测时至少连续跑 50 轮再判断稳不稳。

现象二:本地调试秒回,部署到内网机器首字延迟十几秒。原因多半在检索链路是同步阻塞的——向量检索、重排、拼装串行执行,加上并发没限流,几个请求一起来就互相排队。 解决办法:检索与重排改成异步并发,向量库启动时预热,给生成加并发信号量,并在日志里单独记「检索耗时」和「首字耗时」两个指标,别混在一起看。

5.2 检索质量类:答案不对,八成不是模型的锅

现象三:回答看着有模有样,但关键数字和原文对不上。原因一般有三层:切分块太大导致一个块混了多个主题、没有重排导致相关度低的块被塞进提示词、提示词里没有强制引用要求。 解决办法:把块大小降到 400~500 字并加 50~80 重叠;引入重排模型;提示词里写死「每句结论标注来源编号」,装配后做引用校验。

现象四:资料明明更新了,问出来还是旧答案。原因是索引与文件版本没有绑定,或者检索结果做了进程内缓存但没有失效机制。 解决办法:入库时写入version和文件摘要值,重建索引后切换版本指针;缓存键带上kb_id + version,版本一变自然全部失效。这件事不提前设计,后期补起来非常麻烦。

5.3 生成与交互类:体验问题也是质量问题

现象五:同一份资料,用户换了三种问法,只有一种能答上来。原因是用户口语化提问和文档书面表述之间存在语义鸿沟,单靠原始问题做向量化召回不够。 解决办法:加一步问题改写,用最近一轮对话补全指代(「它」「这个」),同时生成 2~3 个同义查询做多路召回再融合。改写这一步成本很低,对召回率的提升却很直接,属于性价比最高的优化之一。

6. 验收与答辩:用一套指标证明这套问答助理真的能用

验收最忌讳现场随便问几个问题,答得好就说行。评审老师或需求方追问一句「你怎么证明它在大部分情况下是对的」,就答不上来了。我习惯在开发阶段就把评测集攒起来:从真实文档里挑 60~100 条问题,其中 70% 是资料里有答案的,30% 是资料里明确没有的,每条标注标准答案要点和应该命中的文档页码。有了这份集子,后面所有调参都有依据,答辩时也有东西可讲。

指标不要贪多,六个就够,覆盖质量、速度、稳定性三条线:

指标定义合格线
召回命中率 Recall@5标准答案所在块进入前 5 的比例≥ 0.85
答案要点覆盖率人工抽检 50 条,要点覆盖完整比例≥ 0.85
正确拒答率30% 无答案问题中,明确拒答的比例≥ 0.90
引用准确率答案中[n]能正确对应原文的比例≥ 0.95
首字延迟 P95从请求到第一个字的时间≤ 1.5s
知识库更新生效时间文档入库到可被检索≤ 10min

跑评测不要靠手点,写个脚本一次性跑完,输出逐条结果,方便定位是哪一类问题在拖后腿:

# tests/eval.py import csv, asyncio from core.chat import answer_once async def run(eval_file="tests/eval_set.csv"): rows, hit, reject_ok = [], 0, 0 for item in load(eval_file): r = await answer_once(item.kb_id, item.question) recalled = any(p == item.gold_page for p in r.refs_pages) rejected = "未涉及" in r.answer hit += recalled reject_ok += (rejected == item.is_no_answer) # 该拒的拒了、该答的答了才算对 rows.append({"q": item.question, "recall": recalled, "answer": r.answer, "first_token_ms": r.latency_ms}) print(f"Recall@5={hit/len(rows):.2f} 拒答判断准确率={reject_ok/len(rows):.2f}") csv.DictWriter(open("tests/result.csv", "w", newline=""), fieldnames=rows[0].keys()).writeheader() csv.DictWriter(open("tests/result.csv", "a", newline=""), fieldnames=rows[0].keys()).writerows(rows)

答辩讲解的顺序也有讲究:先讲约束——不联网、资料不出机房、单卡可用,这三条决定了很多设计取舍;再讲架构分层和数据流,说清哪一层负责哪件事;然后现场只跑三条 case,一条资料里有明确答案的(展示引用跳转)、一条需要跨文档综合的(展示多片段拼接)、一条资料里没有的(展示正确拒答),这三条把系统的能力边界展示得比跑十条都清楚;最后落到指标表,讲清楚每条指标怎么测、当前值多少。老师最容易追问的是「阈值怎么定的」「切分大小为什么是 500」,这两个问题都要有实测曲线或对比数据支撑,不能只说凭经验。

从这个项目之后,我每次做检索类的东西,都会在写第一行业务代码之前先把评测集攒出来,哪怕只有三十条。没有评测集,所有调参都是凭感觉调参,答辩现场被问一句「换个问法还行吗」就只能当场碰运气。希望这套结构和指标能帮到你,省下我当初绕的那几圈。

本文还有配套的精品资源,点击获取