简介基于大模型实现的知识库问答源代码工程适合对LLM应用开发、RAG检索增强生成感兴趣的开发者可快速搭建本地知识库问答助手适用于企业制度查询、科研文献管理、个人文档问答等场景。压缩包共73个文件约17.3MB以Python源码、pickle数据、Word文档及readme为主。py文件实现模型接入、文本加载、检索与对话链等核心模块pickle用于缓存和向量索引docx为内置知识库样例图片展示了处理流程。项目已有216人学习浏览。目录结构清晰包含agent、loader、database、models、chains等模块内置ChatGLM与MOSS两类大模型调用接口并同时提供网页、命令行、接口三种演示方式方便不同环境快速体验。还附带中文文本分割器、PDF与图像加载工具、nltk语料支持以及说明文档能覆盖从文件上传、文本拆分、向量化存储到检索问答的完整流程。读者既可基于73个文件直接运行测试也可修改模型配置与知识库内容扩展为定制化问答系统是学习企业级RAG落地的实用参考。1. 基于大模型的知识库问答先绕过“伪需求”再谈代码知识库问答听起来是个老话题但加上“基于大模型”之后整个实现路径完全变了。过去做问答靠规则模板加关键词检索召回的是一堆文档链接现在接入大模型之后要的是用自然语言问、拿自然语言答答案还要带着出处和推理痕迹。这个题目里的“源代码”三个字很关键它意味着你拿到的不只是一个 Demo而是一条完整链路文档加载、文本切分、Embedding 入库、向量检索、上下文拼装、调大模型接口、流式返回、前端展示。这条链路看着长但每一环都有固定套路踩坑点也高度一致。我见过太多人把精力耗在“选哪个大模型”上结果跑通之后发现答案质量差、检索结果驴唇不对马嘴反而开始怀疑大模型本身。真实情况是知识库问答的瓶颈八成不在模型而在检索这一侧。Embedding 选型、切分粒度、向量索引参数、召回阈值、重排策略每一个都直接影响最终答案能不能看。所以这篇文章不打算只贴一份可运行的源码而是把这份源码拆开讲清楚哪些参数是必调的哪些环节是容易翻车的哪些代码看一眼就应该删掉重写。适合正在搭本地知识库问答系统的开发者也适合想评估这项技术值不值得投入的架构师。读完你能达到的状态是拿到一份基于大模型的知识库问答源码能说清楚每一层在做什么知道改哪里能提分遇到毛病知道去哪排查。2. 检索引擎与生成模型的结合点RAG 架构下的关键选型2.1 为什么问答系统必须引入向量检索而不是继续用 BM25传统的关键词检索解决不了同义改写问题。用户问“怎么办理异地就医备案”文档里写的是“跨省就医直接结算手续”两句话没有共享任何关键词BM25 召回直接落空。向量检索的核心思路是把文本映射成稠密向量语义相近的句子在向量空间里距离也近于是上面这个例子能稳定命中。但向量检索不是万能的它对数字、型号、精确名称这类需要逐字匹配的查询反而容易出错。常见做法是 BM25 与向量检索双路召回再融合但大部分团队的第一版都从纯向量检索起步因为工程简单效果也够看。# 双路召回的最小实现BM25 走关键词向量走语义最后按分融合 import jieba from rank_bm25 import BM25Okapi from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-m3) # 中文场景常用 def recall(question: str, corpus_texts: list[str], top_k: int 10): # BM25 需要分词中文用 jieba 切一遍 tokenized_corpus [list(jieba.cut(t)) for t in corpus_texts] bm25 BM25Okapi(tokenized_corpus) bm25_scores bm25.get_scores(list(jieba.cut(question))) # 向量检索用余弦相似度bge-m3 的向量默认没有归一化 q_emb model.encode(question) doc_embs model.encode(corpus_texts) vec_scores (doc_embs q_emb) / (doc_embs.norm(dim1) * q_emb.norm()) # 分数融合先各自归一化再加权alpha 控制语义占比 alpha 0.7 return vec_scores * alpha bm25_scores * (1 - alpha)这套双路召回代码的真实价值不在算法新颖而在于它给了你一个可调节的旋钮alpha 偏大则语义优先适合口语化提问偏小则关键词优先适合型号代码类查询。实际项目中我会把它暴露成配置文件里的一个字段按业务域来调。融合之前两个分数必须各自归一化到同一量纲否则一路分数过大另一路等于白做。2.2 大模型在本地的接入方式与上下文组装顺序有了候选文档下一步是拼 Prompt 送进大模型。这一层的关键不是调 API而是控制上下文的质量。常见错误是把所有召回的文档一股脑塞进去导致 Prompt 超过模型上下文窗口或者噪声盖过答案。我一般会做两件事按相关性截断只保留前 top_k 条同时给每条文档带上来源元数据让模型在回答末尾标注出处。def build_prompt(question: str, retrieved: list[dict]) - str: context_blocks [] for i, item in enumerate(retrieved): # 每条上下文带编号和元数据便于模型引用 src item.get(source, unknown) context_blocks.append(f[{i1}] (来源: {src})\n{item[text]}) context \n\n.join(context_blocks) prompt ( 你是企业知识库助手。请严格依据以下资料回答问题。\n 如果资料中没有明确依据直接说“资料中未找到相关内容”不要编造。\n 回答末尾按 [序号] 标注依据来源。\n\n f资料\n{context}\n\n f问题{question}\n 回答 ) return promptPrompt 里“不要编造”这句话不能只是口头约束要配合两个机制落地。第一是在代码里禁止系统级推理避免模型自己脑补来源第二是控制温度参数知识库问答场景 temperature 一般设在 0.1 到 0.3 之间太高模型容易自由发挥。另外来源元数据一定是从切分阶段就保留的不能在检索之后才补否则定位不到原文页码。代码上这段没什么玄学但顺序错了后面排查问题会非常痛苦。3. 文本切分与 Embedding 入库背后的“黑匣子”参数3.1 切分粒度决定你能召回什么粒度的答案文本切分是知识库问答里最容易被低估的环节。切得太碎语义被割裂一句话的上下文丢了主语切得太整向量表示的语义被稀释跟问题的匹配度下降。我在正式项目里默认用两级切分第一级按 Markdown 标题或文档逻辑块切开第二级对超长块按固定窗口滑动切分。每个切块要保留一个唯一 ID 和来源信息入库之后才能追溯。from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on [ (##, section), (###, subsection), ] splitter MarkdownHeaderTextSplitter(headers_to_split_onheaders_to_split_on) chunks splitter.split_text(markdown_doc) # 经标题切分后仍有超长块继续按字符滑窗切 from langchain.text_splitter import RecursiveCharacterTextSplitter rec_splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap64, separators[\n\n, \n, 。, , , ., !, ?, , ] ) final_chunks [] for chunk in chunks: if len(chunk.page_content) 512: sub_chunks rec_splitter.split_text(chunk.page_content) final_chunks.extend(sub_chunks) else: final_chunks.append(chunk)chunk_size 取多少没有绝对标准我自己的经验是配合 Embedding 模型的 max_seq_length 来定。比如 bge-m3 最长支持 8192 token但实际 512 token 的片段已经能表达完整语义且检索更精准。chunk_overlap 用来补偿切分边界断句的问题64 到 128 个字符之间比较稳妥。注意句子级分隔符要放在列表偏后的位置这样长句不会被硬切短文本也不会过度拆分。3.2 Embedding 模型选型与向量库索引参数之间的联系Embedding 模型决定了向量空间里“相似”的定义。中文场景下我常用 BGE 系列在检索任务上有专门调优多语言或英文为主的场景则考虑 Cohere 或 OpenAI 的 embedding 接口。选型时先厘清一个指标检索命中率。拿一批真实问题去本地建好的库里测试召回 Top5 的准确率不达标的模型直接换不要舍不得重跑一遍入库。# FAISS 索引构建的两种典型配置对比 import faiss dimension 1024 # 以 bge-m3 为例实际按模型输出维度填写 # 精确检索准确但慢适合数据量小于 10 万的起步阶段 index_exact faiss.IndexFlatIP(dimension) faiss.normalize_L2(embeddings) # 内积前必须归一化否则结果无意义 index_exact.add(embeddings) # 近似检索百万级数据用 HNSW召回速度与精度取平衡 index_approx faiss.IndexHNSWFlat(dimension, 32) # 32 是邻居数 index_approx.hnsw.efSearch 64 # 检索时探索的节点数越大越准但越慢 index_approx.hnsw.efConstruction 128 # 建库时的探索深度影响索引质量 index_approx.add(embeddings)IndexFlatIP 是暴力内积检索数据量小的时候精确度最高。IndexHNSWFlat 是图结构索引适合百万级向量。注意 faiss 做内积检索前必须先对向量做 L2 归一化否则算出来的不是余弦相似度哪怕模型再准也会召回牛头不对马嘴。efSearch 与 efConstruction 两个参数比较费解前者控制查询时的贪心探索宽度后者控制建图质量建库时改得越大索引越慢但准确率略升。我建议先固定 efSearch 为 64数据量再大再去调 efConstruction不要一上来就贪。4. 把问答主流程写成“源代码”从检索到生成的全链路工程化4.1 一个可运行的服务端骨架接口设计、异步调用与超时兜底知识库问答的源码质量看三个地方就够接口是否支持流式返回、检索是否做缓存、大模型调用是否设置了超时。真实环境里大模型接口经常慢如果不做超时控制用户那边一直转圈服务端线程被占满最后表现为整个系统卡死。以下是一个浓缩后的服务端骨架代码省略了细节但保留了核心逻辑。from flask import Flask, request, Response, jsonify import openai import json from functools import lru_cache app Flask(__name__) lru_cache(maxsize256) def search_cache(query: str): 检索结果缓存避免同一问题反复打向量库 return retrieve(query) app.route(/chat, methods[POST]) def chat(): data request.get_json() question data.get(question, ).strip() if not question: return jsonify({error: question不能为空}), 400 # 第一步检索 docs search_cache(question) # 第二步组装 Prompt prompt build_prompt(question, docs) # 第三步流式调用大模型超时设 60 秒 def generate(): try: stream openai.ChatCompletion.create( modelqwen-plus if use_qwen else gpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.2, streamTrue, timeout60, ) for chunk in stream: delta chunk.choices[0].delta if delta and delta.content: yield fdata: {delta.content}\n\n except openai.error.Timeout: yield data: 大模型响应超时请稍后重试\n\n except Exception as e: yield fdata: 服务异常: {str(e)}\n\n return Response(generate(), mimetypetext/event-stream)这个接口设计里有三个细节值得抄。第一是 search_cache 做 LRU 缓存相同问题第二次进来直接走缓存大幅降低向量库查询压力。第二是流式返回用户等待首字时间从几秒降到几百毫秒体验提升明显。第三是超时兜底生成器内部捕获异常并输出错误信息避免前端连接悬挂。生产环境我还会加一层基于 Redis 的分布式缓存LRU 只适合单机小规模。4.2 多轮对话的上下文管理与指代消解处理知识库问答一旦面对真实用户就逃不开多轮对话。用户先说“帮我查一下报销流程”下一句说“那需要什么材料”如果直接把第二句拿去检索结果往往为空因为这句缺少主语。解决思路有两条一是把多轮历史拼进检索 query 中做改写二是维护一个“当前主题”的状态变量。工程上最省力的做法是调用一次大模型做 query 改写但注意这会增加一次模型调用延迟。我在项目中默认采用轻量规则加模型修正的混合方案。def rewrite_query(history: list[dict], current_question: str) - str: # 规则一当前问题包含疑问词但没有明确主语补上前一轮问题的主语 if any(w in current_question for w in [这个, 那个, 它, 他, 她]) and history: last_user_question history[-1][user] return f{last_user_question}{current_question} # 规则二问题长度太短少于4字主动拼历史最后一句 if len(current_question) 4 and history: return f{history[-1][user]}{current_question} # 其余情况走大模型改写一次调用控制在 0.5 秒内 if random.random() 0.3: # 生产环境改为主动判断 prompt f请将以下对话中的问题改写为独立完整的问题\n历史{history}\n当前{current_question} return llm(prompt) return current_question多轮对话里最常被忽视的是“指代消解后检索但回答仍用原始问题”。我在代码里把改写后的 query 只用于检索生成回答时仍使用原始问题这样模型回答时不会被改写引入的冗余信息干扰。另外历史对话不能无限堆积通常只保留最近三轮太久远的反而会引入话题漂移。规则触发条件一定要写清楚第四种情况里没有主语、历史为空的时候就不该改写。5. 避坑指南基于大模型的知识库问答最容易翻车的四个环节5.1 切分后上下文丢失现象是“模型回答牛头不对马嘴”现象检索回来的文本片段看起来和问题相关但模型答出来的内容明显缺少上下文信息比如把“该方案”指代成另一份文档里的内容。原因切分时把段落截断前文的主语和限定条件留在了上一个 chunk 里。解决强制 chunk_overlap 覆盖一句话的长度并且切分边界尽量选择段落末尾。我的经验值是 overlap 至少 64 字符宁可多存一点重复内容也不要丢失关键限制条件。5.2 Embedding 维度不对导致的“库白建了”现象建库时正常检索时报错“dimension mismatch”或者检索结果为空。原因换过 Embedding 模型新旧模型输出维度不同但 FAISS 索引没有重建。解决任何模型替换之后必须全量重建索引不存在“原地升级”。代码里我把维度做成配置项建库前先判断 index 文件是否与新模型匹配不匹配就直接抛异常提示重建这是一个成本极低但极其管用的防御。5.3 相似度阈值定得太高导致大量问题召回为空现象一部分问题检索返回结果数量为 0系统直接回答“未找到相关内容”。原因余弦相似度阈值设得过高比如 0.8然而实际有效答案的相似度普遍在 0.6 到 0.75 之间。解决先在库上统计所有真实问题召回的相似度分布取一个覆盖 80% 以上有效答案的阈值作为默认值。代码里不要把阈值写死在检索函数里而是放在配置文件中方便按业务域调整。5.4 不加重排导致“多路召回数量虚高、精准度不足”现象Top10 里只有前两条相关其余全是陪跑模型被无关信息干扰。原因向量检索只保证了粗粒度语义相关没有对候选结果做精确排序。解决加入一个轻量级重排层用交叉编码器对候选文档与问题进行二次打分。下面是一个实现要点。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-v2-m3) def rerank(question: str, docs: list[dict], top_n: int 3): pairs [(question, d[text]) for d in docs] scores reranker.predict(pairs) # 按重排分数降序取前 N ranked sorted(zip(docs, scores), keylambda x: x[1], reverseTrue) return ranked[:top_n]重排层的作用不是替代向量召回而是把向量召回的结果从“大致相关”精修到“确实相关”。交叉编码器的计算成本高不能直接对全库跑只能在向量召回之后用。实践中我设定重排候选池大小为 10重排后取前 3 条送给大模型。这一步之后回答质量通常会有肉眼可见的提升。6. 把知识库问答源代码用起来最小启动命令与离线验证清单拿到这份源码之后别急着改代码先用最小命令跑通一条链路。假设仓库目录结构包含embed.py建库脚本、query.py检索脚本、server.py服务入口启动顺序如下先装依赖再建库然后启动服务。我习惯把建库与服务分开因为知识库更新频率低建库是离线任务服务是线上任务。# 安装依赖建议用 Python 3.10 的虚拟环境 pip install sentence-transformers faiss-cpu flask openai0.28 python-dotenv # 第一步离线建库把 docs 目录下的文档全部入向量库 python embed.py --corpus ./docs --output ./index.bin # 第二步启动问答服务默认监听 8000 端口 python server.py --index ./index.bin --port 8000跑通之后做一轮离线验证我自己的习惯是准备 20 到 50 条真实问题覆盖三种类型事实查询报销流程是什么、对比查询A方案和B方案的区别、推理查询如果延迟增加了可能是什么原因。对每一类问题分别记录检索命中的 source 是否准确、最终回答是否有依据。如果事实查询命中率低于 80%优先检查切分粒度和 Embedding 模型如果推理查询答得不完整优先加长上下文窗口或调整 rerank 的 top_n。最后分享一个我在多个项目里验证过的习惯把这个源码仓库当成一个持续迭代的活项目而不是一次性交付物。每次业务文档更新重新跑一遍 embedding 增量入库每条用户的“回答不满意”反馈记录到单独的文件里每月抽一批语料做一次命中率回归。这个习惯比换模型、调参数更能长期提升系统质量。希望帮到你。本文还有配套的精品资源点击获取