当前位置：首页 > news >正文

RAG技术实战：从原理到代码，构建企业知识库问答系统

news 2026/8/31 4:04:07

前阵子做企业知识库问答时，我一直在想一个问题：大模型明明那么能“聊”，为什么面对公司内部文档还是经常一本正经地胡说八道？

原因并不复杂：模型的知识来自训练数据，它有截止时间，也无法接触到企业私有文档。如果直接把“最近三个月项目复盘”“新版报销制度”扔给模型，它大概率会编一个看起来很像那么回事的答案。

后来我彻底想通了一个方案：与其逼模型记住知识，不如让模型学会查资料。也就是本文要聊的核心——RAG。

这篇文章会围绕“从0到1学习AI应用开发”的主线，把 RAG 的原理、基础实现、高级检索技巧和工程落地细节全部串起来。无论你是刚开始接触 AI 应用开发，还是已经写过几个 OpenAI 调用脚本，都能从这篇文章里找到可以上手的干货。

1. RAG 是什么：从知识库问答说起

1.1 大模型直接问答的三个短板

在进入 RAG 之前，先看一个非常典型的场景。

假设你是一家电商公司的后端工程师，公司希望你做一个内部客服助手，能回答关于售后规则、仓库发货时效、商品上下架流程等问题。这些内容分散在几十篇 Word 和在线文档里，而且每个月都会更新。

如果直接用现成的大模型对话能力实现，会遇到三个很实际的问题：

知识截止：模型训练数据有截止时间，公司文档里的新规则它完全不知道。
幻觉问题：当模型不知道答案时，它会倾向于“编造”，而不是承认自己不会。
私有数据隔离：企业内部数据不可能上传给外部大模型训练，但模型又需要这些数据才能回答。

微调好像是另一个选项，但文档每周都在变，总不能每周都重新训练一版模型。

1.2 检索增强生成：给模型来一场开卷考试

RAG 的出现，恰好解决了上面这个矛盾。

RAG 的全称是 Retrieval-Augmented Generation，翻译过来是检索增强生成。它的核心思路不复杂：在模型回答之前，先从外部知识库中检索出与问题最相关的资料，再把问题和资料一起交给模型生成回答。

类比一下就很好理解了。

传统大模型直接回答，像一场闭卷考试，模型只能靠记忆里的知识作答，不会的就瞎编。

RAG 则是开卷考试。模型拿到问题后，先把知识库翻一遍，找到相关的段落，然后照着资料来回答。资料里有，就答；资料里没有，就明确说不知道。

用一句话总结：RAG 不是要让模型记住更多知识，而是让模型学会按需查资料。

这带来几个明显的好处：

知识库可以随时更新，不需要重新训练模型。
模型回答有据可依，幻觉问题大幅减少。
企业内部数据可以保留在自己的知识库里，只需要在调用推理服务时把相关片段拼进 Prompt。

1.3 RAG 与微调怎么选

很多初学者会把 RAG 和微调放在一起比较，其实它们解决的是不同层面的问题。

对比维度	RAG	微调
核心原理	引入外部检索结果作为上下文，辅助模型回答	在模型权重上继续训练，把新知识写进参数
知识更新	替换知识库文档即可，成本低	需要重新训练和验证，周期长
训练成本	不需要训练模型，只需要构建索引	依赖 GPU 资源和训练数据，成本高
幻觉控制	有明确检索片段作为依据，可控性较强	依然可能产生幻觉，依赖模型记忆
适用场景	知识库问答、私有数据检索、动态文档问答	特定写作风格、领域术语强化、工具调用能力

实际项目里，两者不是互斥的。很多团队会先用 RAG 解决知识库问答，再针对高频场景微调一个特定风格的模型，最后把两者结合起来使用。对于新手来说，RAG 是投入产出比最高的起步方案。

2. RAG 系统全流程拆解

尽管不同框架对 RAG 的实现细节略有差异，但整体流程大体可以分为三个阶段：索引阶段、检索阶段、生成阶段。

2.1 索引阶段：把文档变成向量

索引阶段是离线的准备工作，目标是把原始文档处理成机器可以快速检索的形式。

整个流程可以拆成四步：

文档加载：读取 PDF、Word、Markdown、TXT、HTML 等格式的原始内容，并解析成纯文本。
文本清洗：去掉页眉页脚、乱码、特殊符号、无意义的空行，保留正文核心内容。
切块：把长文本切成若干个小片段。这一步非常关键，切片太大检索不精准，太小又丢失上下文。
向量化与入库：调用 Embedding 模型，将每个文本片段编码成一个特征向量，然后写入向量数据库。

向量可以理解成“文本的数学表示”。语义越接近的两段文本，它们的向量在高维空间中的距离越近。之后检索时，只需要计算向量距离，就能快速找出最相关的文本片段。

2.2 检索阶段：从知识库中找答案

索引构建好之后，就进入在线检索阶段。

用户输入一个问题后，系统会做两件事：

用同一个 Embedding 模型，把用户问题编码成向量。
在向量数据库中执行相似度搜索，找到与问题向量最接近的若干个知识片段，并按照相似度排序。

这个过程通常使用余弦相似度、内积或欧氏距离作为衡量标准。为了兼顾效率和精度，生产环境一般会配合倒排索引、HNSW 等算法一起使用。

2.3 生成阶段：让模型基于证据回答

检索到候选片段后，系统会把问题与片段拼接成 Prompt，交给大模型生成最终回答。

一个典型 Prompt 结构如下：

请根据下面的知识库片段回答用户问题。 如果知识库中没有相关信息，请直接说明“知识库中没有找到相关内容”，不要编造答案。 知识库片段： [片段1] [片段2] [片段3] 用户问题： [用户问题]

这里有两个设计要点：

上下文约束：告诉模型只能基于给定片段回答，不能自由发挥，从提示词层面降低幻觉。
失败兜底：明确要求模型在资料不足时承认不知道，避免给出错误答案。

整个 RAG 流程可以用一句话概括：先查后答，带资料作答。

3. 环境准备与项目结构

在开始写代码之前，先把运行环境准备好。本文的示例使用 Python 完成，重点演示 RAG 工作原理，不依赖重量级框架。这样做的目的是让你先看清每个环节到底做了什么，之后再接触 LangChain、LlamaIndex、Dify 时会更容易理解。

3.1 运行环境与依赖

建议使用 Python 3.9 及以上版本。

基础依赖库如下：

sentence-transformers：加载 Embedding 模型和重排序模型。
faiss-cpu：向量索引与相似度检索。
numpy：向量运算。
openai：调用 OpenAI 兼容接口的大模型服务。
rank-bm25：实现 BM25 关键词检索。
jieba：中文分词，配合 BM25 使用。

安装命令：

pip install sentence-transformers faiss-cpu numpy openai rank-bm25 jieba

需要注意的是：

不同的 Python 版本、操作系统对 faiss-cpu 的 wheel 支持略有差异，如果安装失败，可以先升级 pip 再重试。
Embedding 模型首次运行时会从模型仓库下载权重，需要确保网络可达。如果下载超时，可以提前把模型下载到本地，再通过本地路径加载。
openai 库版本建议使用 1.x，因为 1.x 之后客户端初始化方式统一为OpenAI(base_url=..., api_key=...)。如果你还在使用 0.x 版本，参数名会略有不同。

3.2 项目目录设计

本文会按照下面这个目录组织代码：

rag-demo/ ├── data/ │ └── employee_handbook.txt ├── build_index.py ├── query.py ├── advanced_retrieval.py └── requirements.txt

data/：存放知识库原始文档。
build_index.py：读取文档、切块、向量化、构建索引。
query.py：检索 + Prompt 组装 + 大模型生成。
advanced_retrieval.py：高级检索能力演示，包括混合检索和重排序。

3.3 本地模型服务准备思路

本文示例中的生成环节会调用一个 OpenAI 兼容的大模型接口。如果你还没有可用的 API，可以考虑本地部署。

目前主流的本地推理服务，例如Ollama、vLLM、llama.cpp 的 server 模式，大多提供了 OpenAI 兼容的/v1接口。你只需要把base_url指向本地服务地址，即可用 openai 客户端库完成调用。

本文示例的地址为http://localhost:8000/v1，请根据你本地实际部署的服务地址进行修改。

4. 基础 RAG 实战：第一个知识库问答程序

这一节会从零开始搭建一个最小可运行的 RAG 程序。为了让流程更直观，我会用它做一个“员工手册问答助手”。

4.1 准备知识库数据

在data/employee_handbook.txt中放入以下内容：

公司实行五天工作制，工作时间为上午9:00至下午18:00，午休时间12:00至13:00。 员工入职满一年后可享受年假，年假天数按工龄计算。满一年为5天，满三年为7天，满五年为10天。 年假需提前三天在OA系统中提交申请，审批通过后方可休假。 新员工的试用期为三个月，试用期内离职需提前三天提交书面申请。 员工每个月有一次调休机会，调休申请需在周五前提交，并经部门负责人审批。 办公用品申领统一通过行政服务台处理，紧急需求可在工作日9:00-17:00电话联系行政。

这是一段简单的纯文本知识库。现实项目中，这里可以替换成企业制度 PDF、产品帮助文档、技术架构文档等。

4.2 文档加载与切块

创建build_index.py，先写一个带有重叠机制的文本切块函数。

import re def load_text(path): with open(path, "r", encoding="utf-8") as f: return f.read() def split_text(text, chunk_size=200, chunk_overlap=50): # 先按空行拆成段落 paragraphs = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] chunks = [] buffer = "" for para in paragraphs: if len(buffer) + len(para) > chunk_size and buffer: # 上一段已达到长度，作为完整片段保存 chunks.append(buffer.strip()) # 保留末尾 chunk_overlap 长度的内容，避免上下文断裂 overlap_text = buffer[-chunk_overlap:] if chunk_overlap > 0 else "" buffer = overlap_text + "\n" + para else: buffer += "\n" + para if buffer.strip(): chunks.append(buffer.strip()) return chunks

为什么切块时要考虑重叠？

因为一段长文本如果被硬性截断，很可能把原本完整的业务规则拦腰切开。比如“新员工试用期为三个月”和“试用期内离职需提前三天申请”如果被分到两个不相邻的片段里，检索时可能只召回前半句，导致答案不完整。

重叠设置会让片段与片段之间保有公共内容，在一定程度上缓解这个问题。

4.3 构建向量索引

继续在build_index.py中补充索引构建逻辑。

import json import os import faiss import numpy as np from sentence_transformers import SentenceTransformer EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5" DATA_PATH = "data/employee_handbook.txt" INDEX_DIR = "rag_index" def build_index(): text = load_text(DATA_PATH) chunks = split_text(text) print(f"切块完成，共获得 {len(chunks)} 个片段") # 加载中文 Embedding 模型 model = SentenceTransformer(EMBEDDING_MODEL) # 对片段做向量化，并归一化处理 vectors = model.encode(chunks, normalize_embeddings=True) dim = vectors.shape[1] # 使用内积索引 + 归一化向量，等价于余弦相似度检索 index = faiss.IndexFlatIP(dim) index.add(np.array(vectors, dtype=np.float32)) # 保存索引和原始片段 os.makedirs(INDEX_DIR, exist_ok=True) faiss.write_index(index, os.path.join(INDEX_DIR, "faiss.index")) with open(os.path.join(INDEX_DIR, "chunks.json"), "w", encoding="utf-8") as f: json.dump(chunks, f, ensure_ascii=False, indent=2) print(f"索引构建完成，向量维度：{dim}") if __name__ == "__main__": build_index()

这里需要解释几个关键点。

Embedding 模型选择：示例使用BAAI/bge-small-zh-v1.5，这是中文场景下综合表现不错的模型，体积小、推理快，适合入门和中小规模知识库。如果希望效果更好，可以换用bge-base-zh-v1.5或bge-large-zh-v1.5。

为什么用归一化向量 + 内积：余弦相似度的计算方式，本质上等价于把两个向量先归一化再做内积。通过normalize_embeddings=True对向量做归一化，再用faiss.IndexFlatIP，可以少算一步余弦距离，检索性能更好。

IndexFlatIP是暴力精确检索，数据量不大时效果最好。数据量变大后，可以换成IndexHNSWFlat或各类支持近似检索的索引。

4.4 检索与生成

索引构建完成后，创建query.py完成检索和回答生成。

import json import faiss import numpy as np from openai import OpenAI from sentence_transformers import SentenceTransformer EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5" INDEX_DIR = "rag_index" # 加载模型与索引 embedding_model = SentenceTransformer(EMBEDDING_MODEL) index = faiss.read_index(f"{INDEX_DIR}/faiss.index") with open(f"{INDEX_DIR}/chunks.json", "r", encoding="utf-8") as f: chunks = json.load(f) # OpenAI 兼容客户端 client = OpenAI( base_url="http://localhost:8000/v1", # 本地推理服务地址 api_key="EMPTY", # 本地服务通常不校验 Key ) def search(query, top_k=3): query_vec = embedding_model.encode([query], normalize_embeddings=True) scores, ids = index.search(np.array(query_vec, dtype=np.float32), top_k) results = [] for score, idx in zip(scores[0], ids[0]): if idx < 0: continue results.append({"chunk": chunks[idx], "score": float(score)}) return results def generate_answer(query, top_k=3): results = search(query, top_k) context = "\n\n".join( f"[知识库片段{i + 1}] {item['chunk']}" for i, item in enumerate(results) ) prompt = f"""你是一个企业知识库问答助手。请根据下面的知识库片段回答用户问题。 要求： 1. 如果知识库片段足够回答，请基于片段内容作答； 2. 如果知识库片段不足以回答，请直接回答“知识库中没有找到足够信息”； 3. 请使用原文表达，不要编造不存在的制度。 知识库片段： {context} 用户问题：{query} """ response = client.chat.completions.create( model="qwen2.5:7b-instruct", # 替换为你本地部署的模型名 messages=[{"role": "user", "content": prompt}], temperature=0.2, ) return response.choices[0].message.content if __name__ == "__main__": question = "年假怎么申请？" print("问题：", question) print("回答：", generate_answer(question)) print("\n召回片段：") for item in search(question): print("-", round(item["score"], 4), item["chunk"][:50])

4.5 运行与验证

先运行索引构建：

python build_index.py

预期输出：

切块完成，共获得 6 个片段 索引构建完成，向量维度：512

然后运行查询：

python query.py

预期输出会包含知识库检索结果和模型生成的回答。

这里有个细节要注意：检索结果和生成模型是两套模型。Embedding 模型负责找资料，生成模型负责写答案。实际项目中，两者可以是不同的模型，甚至 Embedding 模型可以单独部署一个服务。

到这里，你已经完成了一个最简 RAG 系统。它虽然简单，但完整覆盖了“文档切块 -> 向量化 -> 索引存储 -> 相似度检索 -> 上下文增强生成”的全流程。

5. 高级检索实战：召回质量优化

基础 RAG 能跑通，但离生产环境还有相当距离。最典型的瓶颈不在“生成”，而在“检索”。如果检索阶段找回来的片段不准确，后面大模型再强也无济于事。

这一节会从五个方向提升高级检索能力：混合检索、重排序、查询改写、多轮对话、Agentic RAG。

5.1 基础向量检索的局限

纯向量检索存在几个先天问题：

语义相近但信息不完整：向量模型会把语义相似度算得很接近，但语义接近不代表包含正确答案。
关键词精确匹配能力弱：比如产品型号 “RAG-2024-X9”，语义向量可能把它和“最新一代产品”算得很接近，但无法精确命中用户想搜索的型号。
TOP-K 排序不稳定：向量检索的 top 1 至 top 5 往往质量参差不齐，一些噪声片段混入其中会让生成效果变差。

所以，生产级 RAG 系统通常要做多层召回和多路融合，而不是单一依赖向量检索。

5.2 混合检索：关键词 + 向量

混合检索的思路很简单：同时使用一种基于关键词的检索引擎和一种基于向量的检索引擎，再把两路结果融合起来。

经典关键词检索算法是 BM25。BM25 对精确词匹配非常敏感，非常适合查询中包含产品型号、人名、专有名词等场景。

下面这段代码演示了 BM25 检索与向量检索如何配合。

import jieba import numpy as np from rank_bm25 import BM25Okapi def build_bm25_index(chunks): # 中文场景先分词，再构造 BM25 索引 tokenized_corpus = [list(jieba.cut(chunk)) for chunk in chunks] bm25 = BM25Okapi(tokenized_corpus) return bm25 def bm25_search(bm25, query, top_k=3): tokenized_query = list(jieba.cut(query)) scores = bm25.get_scores(tokenized_query) top_ids = np.argsort(scores)[::-1][:top_k] return [(int(idx), float(scores[idx])) for idx in top_ids]

接下来是 RRF 融合算法。RRF 全称是 Reciprocal Rank Fusion，它的核心思想是：不直接比较两路检索的分数，而是按排名位置计算融合分。

def rrf_fusion(vector_hits, bm25_hits, k=60): fused = {} # vector_hits 结构为 [(index, score), ...] for rank, (idx, score) in enumerate(vector_hits): fused[idx] = fused.get(idx, 0) + 1.0 / (k + rank + 1) # bm25_hits 结构为 [(index, score), ...] for rank, (idx, score) in enumerate(bm25_hits): fused[idx] = fused.get(idx, 0) + 1.0 / (k + rank + 1) # 按融合分数降序排列 ranked = sorted(fused.items(), key=lambda x: x[1], reverse=True) return ranked[:3]

RRF 的好处是无需对两路分数做标准化。向量相似度和 BM25 得分的数值范围完全不同，直接相加会让某一方主导结果。RRF 只看排名，天然规避了这个问题。

5.3 重排序：用精排模型提升 Top-K 质量

召回阶段追求“全”，排序阶段追求“准”。常见做法是：先用向量检索 + 关键词检索召回足够多的候选，再用一个轻量级 Cross-Encoder 模型对候选做细粒度打分，最后按精排分数取 Top-K。

Cross-Encoder 和普通 Embedding 模型的区别在于：Embedding 模型会把问题和文档分别编码成向量再计算相似度；Cross-Encoder 则会同时接收两个句子，输出一个相关性分数，精度更高，但速度更慢。

示例代码如下：

from sentence_transformers import CrossEncoder # 加载重排序模型 reranker = CrossEncoder("BAAI/bge-reranker-base") def rerank_candidates(query, candidates, top_k=3): # candidates: 候选片段列表 pairs = [(query, doc) for doc in candidates] scores = reranker.predict(pairs) # 按重排分数降序 sorted_pairs = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return sorted_pairs[:top_k]

一个典型的优化流程是：

向量检索召回 20 个候选片段。
BM25 检索召回 20 个候选片段。
使用 RRF 融合，取前 10 个候选。
使用 Cross-Encoder 对 10 个候选精排。
取前 3 个片段交给大模型生成。

这个模式在业界非常常用，也是 RAG 最值得投入优化的地方。

5.4 查询改写：让检索理解真实意图

用户输入往往不够规范，比如：

只输入“年假”
输入“我入职一年半能休几天”
输入“最多能调休几次”

这些问题在语义上是模糊的，直接拿去检索效果有限。此时可以先用大模型做查询改写，把用户问题扩写成更适合检索的表述。

示例 Prompt：

你是查询改写专家。用户向知识库提出一个问题，但它可能指代不明或过度省略。 请将问题改写成更适合检索的独立查询，只输出改写后的查询，不要输出其他内容。 原问题：{query} 改写后的查询：

执行流程变成：

用户提问。
大模型将问题改写为“年假申请条件及年假天数计算规则”。
用改写后的内容进行混合检索。
将原始问题和检索片段交给大模型生成回答。

查询改写不仅能改善模糊查询，还能处理多轮对话。用户说了“那试用期呢？”，改写模型会结合上轮上下文，补全成“新员工试用期离职流程是什么”，从而避免检索阶段丢失指代信息。

5.5 Agentic RAG：检索过程的下一步演进

再往上一层，是当前很热门的 Agentic RAG。

传统 RAG 是“一次检索 + 一次生成”。它的流程固定：问题进来，检索，生成，结束。但现实问题往往需要多步推理。

Agentic RAG 的思路是把大模型从“答案生成器”升级为“检索调度员”。模型可以根据问题自主决定：

先检索哪个知识库；
当前结果是否足够回答；
是否需要继续追问；
是否需要调用其他工具或数据库。

举个例子，用户问“上个月退款订单里面，哪些物流没有及时更新？”这已经不是单纯的知识库检索能解决的，它涉及订单库、物流接口、文档政策等多个数据源。Agentic RAG 会让模型拆解任务，先查订单系统，再调用物流查询工具，最后结合售后政策生成答案。

对入门者来说，先不要急着上 Agentic RAG。把基础 RAG 的召回与重排序做好，效果往往已经提升 80%。Agentic RAG 适合在基础流程稳定后，作为进一步扩展无人值守场景的方向。

6. 常见问题与排查思路

RAG 系统一旦效果不好，最容易让人一头雾水。下面整理了一份高频问题排查表。

问题现象	常见原因	解决思路
检索结果与问题明显无关	切块太大导致片段语义混杂；Embedding 模型效果不足	缩小 chunk_size；换成 bge 系列中文模型
回答仍然出现编造内容	检索到的上下文不足，但 Prompt 没有强制约束	强化 Prompt 限制；增加“未找到内容”兜底；提高检索阈值
向量维度不一致报错	构建索引和查询时使用了不同的 Embedding 模型	统一模型并重新构建索引
中文检索效果差	使用了面向英文优化的 Embedding 模型	换成 BAAI/bge 系列中文模型
本地推理服务调用失败	base_url 或 model 名称与部署服务不匹配	确认服务地址、模型名和 API Key 校验方式
FAISS 索引加载报错	索引路径不对，或构建索引的代码没有先运行	检查 index 文件是否存在；重新执行 build_index.py
BM25 结果看起来不合理	中文文本没有分词	使用 jieba 对文档和查询同时分词
模型推理速度很慢	模型过大或 GPU 资源不足	使用量化版本、换更小的模型，或增加缓存
知识库更新后回答仍是旧内容	索引没有重新构建	建立定时任务或触发机制，文档变更后自动重建索引