背景与问题
很多团队第一次做企业知识库问答,直觉上会这么想:把文档喂给大模型,再包一层聊天界面,不就行了?
但真正落地后,问题往往比想象中多:
- 文档很多,来源杂:PDF、Word、Confluence、飞书、工单系统、数据库导出
- 大模型“知道得很多”,但不知道你公司的内部知识
- 直接把长文档塞进上下文,成本高、延迟高,还容易漏信息
- 业务同学觉得“偶尔答错一次也能接受”,可一旦进入生产环境,错一次可能就是一次事故
- 系统上线后,最难的不是“能回答”,而是:
怎么持续衡量它答得好不好?为什么不好?该优化哪一层?
这也是 RAG(Retrieval-Augmented Generation,检索增强生成)在企业场景里最有价值的地方:
让大模型先查企业知识,再基于检索结果回答。
但我想先泼一盆冷水:RAG 不是“加个向量库”就结束了。
一个能上线、能持续优化的企业知识库问答系统,至少要回答这几个问题:
- 文档如何切分、清洗、索引?
- 检索是纯向量,还是关键词 + 向量混合?
- 怎么减少“看起来很像对,其实不对”的幻觉?
- 怎么做离线评测和线上观测?
- 怎么控制权限、成本、延迟和数据安全?
这篇文章我会从架构设计 + 可运行代码 + 评测方法三个层面,带你搭一套中级开发者能真正上手的方案。
方案全景:从“能跑”到“能上线”
先给出一个比较实用的分层架构。它不追求最复杂,但足够覆盖大多数企业内部问答场景。
flowchart TD
A[企业数据源<br/>PDF/Word/Wiki/数据库] --> B[数据清洗与切分]
B --> C[Embedding 向量化]
B --> D[关键词索引 BM25]
C --> E[向量数据库]
D --> F[倒排索引]
G[用户问题] --> H[Query 改写]
H --> I[混合检索]
E --> I
F --> I
I --> J[重排序 Reranker]
J --> K[上下文构建]
K --> L[大模型生成答案]
L --> M[引用与溯源]
L --> N[效果评测与日志分析]
这套架构里,真正影响效果的关键链路是:
数据质量 → 切分策略 → 检索召回 → 重排序 → Prompt 约束 → 评测反馈
很多项目效果不好,不是模型不行,而是前面几个环节已经把“正确答案”丢掉了。
核心原理
1. RAG 的本质:先缩小搜索空间,再让模型做语言组织
大模型擅长:
- 理解问题
- 汇总信息
- 生成自然语言答案
但它不擅长:
- 实时记住你公司的最新制度
- 稳定、精确地从海量私有文档里找证据
- 对“哪个版本的文件是最新”做严格判断
所以 RAG 的目标不是让模型“更聪明”,而是让它在有限、可信的上下文里回答。
可以把它理解成两步:
- 检索:从知识库中找到最相关的几段内容
- 生成:要求模型仅依据这些内容作答,并给出引用
如果检索没找到对的片段,后面的生成再强也救不回来。
这也是为什么在企业场景里,检索质量通常比模型参数量更重要。
2. 为什么企业知识库要做混合检索
纯向量检索很强,但它不是万能的。
例如用户问:
- “Q3 报销上限是多少?”
- “合同审批单 SOP v2 的负责人是谁?”
- “错误码 E1024 怎么处理?”
这些问题里经常有:
- 专有名词
- 缩写
- 版本号
- 错误码
- 产品型号
这类词汇往往关键词匹配比语义相似更稳定。
所以我的建议通常是:默认从混合检索起步,而不是纯向量检索。
一个典型策略:
- 向量检索:找语义接近的段落
- BM25/关键词检索:找字面匹配强的段落
- Reranker:统一排序,保留最有用的 Top-K
sequenceDiagram
participant U as 用户
participant Q as Query改写器
participant V as 向量检索
participant B as BM25检索
participant R as 重排序器
participant L as 大模型
U->>Q: 输入问题
Q->>V: 改写后问题
Q->>B: 原问题/关键词问题
V-->>R: TopN 语义候选
B-->>R: TopN 关键词候选
R-->>L: TopK 高相关上下文
L-->>U: 带引用的答案
3. 切分策略决定了“能不能检到”
这是很多团队最容易忽视的部分。
常见切分误区
误区一:固定 1000 字切分
看起来简单,但容易把一个完整的知识点拆散。
误区二:完全按段落切分
对于排版不规范的文档,段落长度可能极不均衡。
误区三:切分时不保留标题层级
后面即使检索到了段落,也不知道它属于哪个章节,模型理解会差很多。
更实用的切分原则
我更推荐一种“结构优先 + 长度兜底”的策略:
- 先按文档结构切:标题、章节、列表、表格
- 再按长度切:例如 300~800 中文字符
- 增加 overlap:例如 80~150 字
- 保留 metadata:
- 文档标题
- 章节路径
- 来源链接
- 更新时间
- 权限标签
举个例子,一段 chunk 最终最好长这样:
标题:费用报销制度
章节:3. 差旅报销 / 3.2 住宿标准
内容:自 2024 年 1 月起,华东区出差住宿标准上限为...
来源:wiki://finance/reimburse-v3
更新时间:2024-01-15
权限:finance_internal
这样做的好处是:
即使只检索到一个 chunk,模型也能知道它在什么语境下。
4. 重排序是“最后一道保险”
混合检索召回后,候选片段可能有几十条。
这时候不能直接把所有片段都塞给大模型,不然:
- 成本上升
- 噪音变多
- 真正关键的信息被淹没
重排序(Reranker)的作用,就是对 query 和候选片段做更细粒度相关性判断。
简单理解:
- 向量检索:像“先粗筛”
- 重排序:像“再精排”
经验上,如果你已经有一定召回率,但答案仍然“似是而非”,那大概率该补的是重排序,而不是先换更大的生成模型。
方案对比与取舍分析
1. 三种常见架构
| 方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 纯大模型长上下文 | 实现简单 | 成本高、易漏、更新难 | 文档量小、PoC |
| 纯向量 RAG | 快速入门 | 对术语/编号不稳定 | 通用知识问答 |
| 混合检索 + Reranker + RAG | 效果更稳,可评测可优化 | 架构更复杂 | 企业生产环境 |
我的建议是:
- PoC 阶段:纯向量 RAG 就够了
- 试点阶段:加入混合检索
- 生产阶段:补齐重排序、引用、权限控制、离线评测
2. 容量估算思路
做企业知识库系统,最怕到了上线前才发现容量没估。
一个简单估算方式:
文档切分量估算
假设:
- 原始文档总量:10 万页
- 平均每页可清洗出 1500 中文字符
- 每个 chunk:500 字,overlap 100 字
粗略估算 chunk 数:
每页 chunk 数 ≈ 1500 / (500 - 100) ≈ 3.75
总 chunk 数 ≈ 10 万 * 3.75 = 37.5 万
存储估算
如果 embedding 维度是 1024,每个 float 用 4 字节:
单 chunk 向量大小 ≈ 1024 * 4 = 4096 字节 ≈ 4KB
总向量存储 ≈ 37.5 万 * 4KB ≈ 1.5GB
再加 metadata、索引、冗余副本,通常按 3~5 倍预估更稳妥。
吞吐估算
关键指标通常看:
- 每秒查询数 QPS
- 平均检索耗时
- 重排序耗时
- 大模型生成耗时
- P95 / P99 延迟
一般企业内部问答,真正的瓶颈往往不在向量检索,而在:
- 重排序模型
- 生成模型调用
- 过长上下文导致的 token 消耗
实战代码(可运行)
下面用 Python 做一个最小可运行版本。
它会演示:
- 文档切分
- 向量化
- 简化版检索
- 构造 Prompt
- 做一个基础评测
为了让示例容易跑起来,这里我用本地 Python 代码模拟 embedding 和检索,不依赖复杂外部服务。
真实生产环境中,你可以替换成:
- embedding 服务:OpenAI / Azure / 通义 / 智谱 / 私有模型
- 向量库:FAISS / Milvus / pgvector / Elasticsearch
- 重排序:bge-reranker / jina-reranker / 私有 cross-encoder
1. 安装依赖
pip install numpy scikit-learn
2. 构建一个最小 RAG 检索示例
import re
import math
from typing import List, Dict, Tuple
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
documents = [
{
"id": "doc1",
"title": "费用报销制度",
"section": "3.2 住宿标准",
"content": "自2024年1月起,华东区员工出差住宿报销标准上限为每晚600元,超出部分原则上自行承担,特殊情况需总监审批。",
"source": "wiki://finance/reimburse-v3",
"updated_at": "2024-01-15",
"permission": "finance_internal",
},
{
"id": "doc2",
"title": "费用报销制度",
"section": "2.1 交通标准",
"content": "员工出差优先选择高铁二等座。若出行时间超过6小时,经理级及以上可申请飞机经济舱。",
"source": "wiki://finance/reimburse-v3",
"updated_at": "2024-01-15",
"permission": "finance_internal",
},
{
"id": "doc3",
"title": "合同审批SOP v2",
"section": "1.3 审批责任人",
"content": "合同审批单SOP v2的流程负责人为法务运营负责人。涉及采购合同的,需增加采购总监会签。",
"source": "wiki://legal/contract-sop-v2",
"updated_at": "2024-03-20",
"permission": "legal_internal",
},
{
"id": "doc4",
"title": "错误码处理手册",
"section": "E1024",
"content": "错误码E1024通常表示数据库连接池耗尽。建议优先检查连接泄漏、最大连接数配置,以及慢SQL堆积情况。",
"source": "wiki://tech/error-code-manual",
"updated_at": "2024-02-11",
"permission": "tech_internal",
},
]
def normalize_text(text: str) -> str:
text = text.lower().strip()
text = re.sub(r"\s+", " ", text)
return text
def build_corpus(docs: List[Dict]) -> List[str]:
corpus = []
for d in docs:
merged = f"{d['title']} {d['section']} {d['content']}"
corpus.append(normalize_text(merged))
return corpus
class SimpleRAG:
def __init__(self, docs: List[Dict]):
self.docs = docs
self.corpus = build_corpus(docs)
self.vectorizer = TfidfVectorizer(token_pattern=r"(?u)\b\w+\b")
self.doc_vectors = self.vectorizer.fit_transform(self.corpus)
def keyword_score(self, query: str, doc: Dict) -> float:
query_terms = set(re.findall(r"\w+", normalize_text(query)))
doc_text = normalize_text(f"{doc['title']} {doc['section']} {doc['content']}")
score = 0
for t in query_terms:
if t in doc_text:
score += 1
return score
def search(self, query: str, top_k: int = 3) -> List[Tuple[Dict, float]]:
q = normalize_text(query)
qv = self.vectorizer.transform([q])
sim_scores = (self.doc_vectors @ qv.T).toarray().reshape(-1)
results = []
for idx, doc in enumerate(self.docs):
kw = self.keyword_score(query, doc)
hybrid_score = float(sim_scores[idx]) * 0.7 + kw * 0.3
results.append((doc, hybrid_score))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
def build_prompt(self, query: str, retrieved_docs: List[Tuple[Dict, float]]) -> str:
context_blocks = []
for i, (doc, score) in enumerate(retrieved_docs, start=1):
block = (
f"[参考资料{i}]\n"
f"标题:{doc['title']}\n"
f"章节:{doc['section']}\n"
f"内容:{doc['content']}\n"
f"来源:{doc['source']}\n"
)
context_blocks.append(block)
context = "\n".join(context_blocks)
prompt = f"""你是企业知识库问答助手。
请严格基于参考资料回答,不要编造不存在的信息。
如果参考资料不足以回答,请明确说“根据现有资料无法确认”。
回答时尽量简洁,并在末尾列出引用来源。
用户问题:
{query}
参考资料:
{context}
"""
return prompt
def mock_generate_answer(query: str, retrieved_docs: List[Tuple[Dict, float]]) -> str:
# 这里用规则模拟大模型回答,方便示例可运行
top_doc = retrieved_docs[0][0]
q = query.lower()
if "住宿" in q or "报销上限" in q:
return f"根据现有资料,华东区员工出差住宿报销标准上限为每晚600元,特殊情况需总监审批。来源:{top_doc['source']}"
if "负责人" in q:
return f"根据现有资料,合同审批单SOP v2的流程负责人为法务运营负责人。来源:{top_doc['source']}"
if "e1024" in q:
return f"错误码E1024通常表示数据库连接池耗尽,建议检查连接泄漏、最大连接数配置和慢SQL堆积。来源:{top_doc['source']}"
return "根据现有资料无法确认。"
if __name__ == "__main__":
rag = SimpleRAG(documents)
query = "华东区出差住宿报销上限是多少?"
retrieved = rag.search(query, top_k=3)
print("=== 检索结果 ===")
for doc, score in retrieved:
print(doc["id"], round(score, 4), doc["title"], doc["section"])
print("\n=== Prompt ===")
print(rag.build_prompt(query, retrieved))
print("=== Answer ===")
print(mock_generate_answer(query, retrieved))
运行结果会看到什么
你会看到:
- 检索结果里,
费用报销制度 / 3.2 住宿标准排在前面 - Prompt 中包含引用资料
- 最终答案引用了来源
这个示例虽然简单,但已经说明了 RAG 的主流程。
后续你可以逐步替换掉各个模块,而不是一开始就堆满复杂组件。
3. 增加一个基础效果评测脚本
上线前,至少要先做一组离线评测集。
最简单的思路是:准备若干个“问题—标准答案—期望命中文档”的样本。
from typing import List, Dict
eval_set = [
{
"question": "华东区出差住宿报销上限是多少?",
"gold_answer": "每晚600元",
"gold_doc_id": "doc1",
},
{
"question": "合同审批单SOP v2的负责人是谁?",
"gold_answer": "法务运营负责人",
"gold_doc_id": "doc3",
},
{
"question": "错误码E1024是什么意思?",
"gold_answer": "数据库连接池耗尽",
"gold_doc_id": "doc4",
},
]
def contains_answer(pred: str, gold: str) -> bool:
return gold in pred
def evaluate(rag: SimpleRAG, dataset: List[Dict]):
hit_at_1 = 0
answer_acc = 0
for item in dataset:
retrieved = rag.search(item["question"], top_k=3)
pred_top1_doc = retrieved[0][0]["id"]
pred_answer = mock_generate_answer(item["question"], retrieved)
if pred_top1_doc == item["gold_doc_id"]:
hit_at_1 += 1
if contains_answer(pred_answer, item["gold_answer"]):
answer_acc += 1
print(f"问题: {item['question']}")
print(f"Top1文档: {pred_top1_doc}, 期望文档: {item['gold_doc_id']}")
print(f"预测答案: {pred_answer}")
print("-" * 50)
total = len(dataset)
print(f"Hit@1: {hit_at_1 / total:.2%}")
print(f"Answer Accuracy: {answer_acc / total:.2%}")
if __name__ == "__main__":
rag = SimpleRAG(documents)
evaluate(rag, eval_set)
这类评测很朴素,但非常重要。
因为它帮你把问题拆成两个维度:
- 检索对了吗? 看 Hit@K / Recall@K
- 生成对了吗? 看答案准确率、引用一致性
如果检索都错了,别急着调 Prompt。
如果检索对了但答案还错,再看生成约束、上下文构造和模型本身。
效果评测实践:别只看“感觉还行”
很多团队做评测时,容易陷入两个极端:
- 要么完全凭人工感受
- 要么追求非常复杂的自动评分体系,结果迟迟落不了地
我的建议是:先建立“够用”的评测闭环,再逐步精细化。
1. 先定义评测分层
第一层:检索评测
目标:判断知识有没有被找到。
常见指标:
- Hit@K:标准文档是否出现在前 K 条
- Recall@K:相关片段召回率
- MRR:正确文档排名是否靠前
第二层:生成评测
目标:判断回答有没有说对。
常见指标:
- 答案准确率
- 引用正确率
- 幻觉率
- 拒答正确率
第三层:业务评测
目标:判断对业务是否有用。
常见指标:
- 首问解决率
- 人工转接率
- 用户追问率
- 用户满意度
- 平均响应时间
2. 评测集怎么构造更靠谱
评测集不要只让技术同学拍脑袋出题。
更有效的方法是从真实场景抽样:
- 从搜索日志、工单、IM 群问答里抽问题
- 按主题分桶:
- 制度类
- 流程类
- 故障类
- 定义类
- 对比类
- 标注标准答案与证据文档
- 标注问题难度:
- 单文档可答
- 多文档聚合
- 时效性强
- 无答案应拒答
一个成熟一点的评测集,通常要覆盖:
- 高频问题
- 长尾问题
- 模糊提问
- 错别字提问
- 带版本号/错误码的问题
- 权限不足的问题
- 故意诱导模型编造的问题
3. 一套实用的错误归因方法
当回答效果不佳时,我一般按下面顺序排查:
flowchart TD
A[答案错误] --> B{检索到正确文档了吗?}
B -- 否 --> C[数据清洗/切分/索引问题]
B -- 是 --> D{正确片段排位靠前吗?}
D -- 否 --> E[召回融合或重排序问题]
D -- 是 --> F{Prompt约束是否清晰?}
F -- 否 --> G[生成策略问题]
F -- 是 --> H{答案是否超出证据?}
H -- 是 --> I[幻觉与拒答策略问题]
H -- 否 --> J[评测样本或标注问题]
这张图非常适合做团队内复盘。
因为它逼着大家把“模型回答不好”拆成可修复的问题,而不是一句“LLM 不稳定”就结束。
常见坑与排查
1. 检索命中了,但答案还是错
这是最常见的坑。
典型原因
- 上下文太长,关键信息被淹没
- 多个相似片段互相干扰
- Prompt 没明确要求“仅依据资料回答”
- 没要求输出“无法确认”
排查建议
- 把 Top10 检索结果打印出来,人眼先看一遍
- 只喂 Top3 和只喂 Top8 分别测试
- 给每个 chunk 增加标题和章节信息
- 明确 Prompt 中的拒答规则
2. 专有名词、错误码、版本号效果差
典型原因
- 纯向量检索对稀有 token 不敏感
- 分词不理想
- 文档里同义写法很多
排查建议
- 加 BM25 或 Elasticsearch 关键词召回
- 给 query 做规则归一化:
sop v2↔SOPv2e1024↔E-1024
- 建术语词典和别名表
3. 新文档更新后查不到
典型原因
- 增量索引没跑
- metadata 版本字段不一致
- embedding 任务失败但没有告警
排查建议
- 给数据同步链路做状态机和重试
- 每次索引完成记录文档数、chunk 数、失败数
- 建立“抽样回查”任务:随机检查新文档是否可检索
4. 多文档汇总类问题回答很差
例如:
- “差旅报销和招待报销的审批路径有什么区别?”
- “新旧制度相比,变化点有哪些?”
典型原因
- 检索只拿到单文档片段
- Prompt 没要求模型做对比归纳
- 上下文构造没有按主题分组
排查建议
- 这类问题单独打标签评测
- 引入 query 分类:
- 事实型
- 列表型
- 对比型
- 汇总型
- 对比型问题单独使用模板化 Prompt
安全/性能最佳实践
企业知识库问答一旦上线,安全和性能不是加分项,而是必选项。
1. 权限控制要前置到检索层
最危险的做法是:
- 先全库检索
- 再在生成前做“显示过滤”
这会导致高风险问题:
模型虽然不显示内容,但已经“看过”不该看的文档。
正确做法是:
- 文档入库时打权限标签
- 检索时就按用户身份过滤
- 生成时只使用已授权上下文
一个简单权限模型可以是:
classDiagram
class User {
+user_id
+departments
+roles
+data_scopes
}
class Document {
+doc_id
+title
+permission_tags
+updated_at
}
class Retriever {
+search(query, user_context)
}
User --> Retriever : 携带身份信息
Retriever --> Document : 按权限过滤检索
2. 敏感信息脱敏与审计
知识库里常见敏感信息包括:
- 客户信息
- 合同金额
- 身份证号 / 手机号 / 邮箱
- 内部 API 密钥
- 运维账号信息
建议至少做三件事:
- 入库前脱敏:规则 + NER 识别
- 输出前审查:命中敏感策略就拒答或打码
- 全链路审计:记录谁查了什么、命中了哪些文档、模型返回了什么
3. 降低延迟的实用手段
在真实项目里,用户对“3 秒以内”和“8 秒以上”的感知差别非常大。
可优先做这些优化:
- 检索与关键词查询并行
- 缩短 chunk 数量,减少上下文 token
- 先用轻量模型做 query 改写
- 重排序只对 TopN 候选做,不要全量做
- 对高频问题做缓存
- 对固定知识点做 FAQ 兜底
一个经验值:
- 检索:100~300ms
- 重排序:100~500ms
- 生成:1~3s
如果生成耗时始终是大头,不要先优化向量库,优先考虑:
- 缩上下文
- 换更快模型
- 控制输出长度
- 使用流式输出
4. 成本控制不要等到月账单来了才做
RAG 成本主要来自:
- embedding 批量构建
- 重排序调用
- 生成模型 token 消耗
建议从第一天就记录:
- 每次问答输入 token / 输出 token
- 平均检索条数
- 平均上下文长度
- 每类问题的命中成本
- 缓存命中率
我踩过一个坑:为了“保险”,把 Top10 全塞进模型。
结果效果没明显提升,账单先涨了几倍。后来把真正进 Prompt 的片段压到 Top3~Top5,效果反而更稳。
一个更接近生产的落地建议
如果你是中级开发者,准备带团队或自己主导一个知识库问答项目,我建议按下面路线推进,而不是一步到位:
阶段一:做出可用 PoC
目标:
- 能回答
- 有引用
- 支持基础文档入库
最小组件:
- 文档切分
- 向量检索
- Prompt 约束
- 简单评测集
阶段二:提高稳定性
目标:
- 降低错答率
- 提升专有名词命中率
- 能定位问题
新增组件:
- BM25 混合检索
- 重排序
- 离线评测
- 检索日志分析
阶段三:进入生产
目标:
- 安全可控
- 可观测
- 可持续优化
新增组件:
- 权限过滤
- 敏感信息审查
- 增量索引
- A/B 测试
- 线上反馈闭环
- 告警与审计
总结
企业知识库问答系统,真正的挑战从来不只是“接一个大模型接口”。
要做得稳,核心是把它当成一套完整的信息系统来设计:
- RAG 解决的是私有知识接入问题
- 混合检索 + 重排序 解决的是召回和排序问题
- Prompt 约束 + 引用 解决的是可控回答问题
- 离线评测 + 线上观测 解决的是持续优化问题
- 权限 + 脱敏 + 审计 解决的是企业上线问题
如果你现在就要开始,我给三个最实用的建议:
- 先做混合检索,不要迷信纯向量
- 先建小而准的评测集,再谈大规模优化
- 先保证“找得到且能引用”,再追求“回答得像人”
最后给一个边界条件判断:
如果你的知识库内容非常少、变化不频繁、问题也高度固定,那么复杂 RAG 架构未必值得,FAQ + 搜索可能更划算。
但如果你的文档在持续增长、跨部门知识很多、且需要权限和评测,那就值得按本文这套路线逐步演进。
别急着一开始就做“最强架构”。
先让系统可验证、可排错、可迭代,才是企业场景里真正能活下来的方案。