背景与问题
很多团队第一次做企业内部知识库问答,直觉上会觉得:把文档丢进向量库,再接一个大模型,不就完了吗?
但真正上线后,问题往往不是“能不能回答”,而是:
- 回答不稳定,同一个问题上午和下午答案不一样
- 文档明明存在,却检索不到
- 检索到了很多“看起来相关、其实没用”的片段
- 模型生成时“脑补”制度细节,风险很高
- 用户一多,延迟飙升,成本失控
- 权限隔离做不好,容易把 A 部门文档答给 B 部门
我自己做这类系统时,最大的感受是:RAG 的难点从来不只是模型,而是整个链路的工程设计。
对中级开发者来说,真正需要掌握的是这几个层面:
- 文档如何被切分、清洗、建索引
- 查询如何被改写、召回、重排
- 答案如何带引用生成,并可控地拒答
- 系统如何在权限、性能、成本之间平衡
这篇文章我会从企业场景出发,带你搭一个可运行的最小版本,再讲清楚为什么很多“看似合理”的做法会在生产环境里翻车。
核心原理
RAG(Retrieval-Augmented Generation,检索增强生成)可以简单理解为:
先从企业知识库里找资料,再让大模型基于这些资料回答问题。
它通常分成两条主链路:
- 离线链路:文档采集 → 清洗 → 切分 → 向量化 → 建立索引
- 在线链路:用户提问 → 查询理解/改写 → 检索 → 重排 → 生成答案 → 引用与审计
为什么企业知识库不能只靠“向量检索”
仅用向量检索,在企业场景中经常不够,原因很实际:
- 制度、流程、产品文档里有很多关键词强约束
- 版本号、工单号、系统名、接口名这类信息,BM25 这类关键词检索更稳
- 用户提问常常很短,例如“报销截止时间”“VPN 申请在哪”,语义信息不足
- 只靠向量,容易把“语义相近但不属于同一制度”的文档召回进来
所以企业里更常见的是混合检索:
- 关键词检索:查准
- 向量检索:查全
- 重排模型:把真正最相关的内容放前面
一个更贴近生产的 RAG 架构
flowchart LR
A[企业文档源\nWiki/FAQ/PDF/工单/数据库] --> B[文档接入与清洗]
B --> C[切分 Chunking]
C --> D1[关键词索引 BM25]
C --> D2[向量索引 Vector DB]
D1 --> E[混合召回]
D2 --> E
Q[用户问题] --> F[查询改写/权限过滤]
F --> E
E --> G[重排 Rerank]
G --> H[上下文构造]
H --> I[LLM 生成答案]
I --> J[返回答案+引用+置信度]
这个架构的关键,不是模块多,而是每一层都在解决一个具体问题:
- 清洗:降低脏数据干扰
- 切分:让检索粒度合适
- 混合召回:减少“查不到”和“查偏了”
- 重排:提升前几条结果质量
- 引用:让答案可追溯
- 权限过滤:保证“只能答他有权看的”
方案对比与取舍分析
方案一:纯大模型直答
优点:
- 开发快
- 不需要知识库建设
缺点:
- 幻觉严重
- 无法绑定企业私有知识
- 不能追溯依据
- 合规性弱
适用场景:
- 通用问答
- 原型验证,不适合正式企业知识问答
方案二:纯向量 RAG
优点:
- 实现成本适中
- 能接入私有文档
缺点:
- 短查询效果不稳定
- 关键词强约束场景容易失真
- 多版本文档容易召回混乱
适用场景:
- 中小规模知识库
- 文档结构相对规整
方案三:混合检索 + 重排 + 权限过滤
优点:
- 更适合企业真实数据
- 检索稳定性更高
- 便于做审计、引用、拒答
缺点:
- 工程复杂度更高
- 调参与监控成本增加
适用场景:
- 生产环境
- 多部门、多文档源、强权限隔离
我的建议很明确:如果你准备做上线系统,直接按“混合检索 + 重排 + 权限过滤”的思路设计。
不要先偷懒做纯向量版,然后再补权限和重排,后面会改得很痛苦。
架构设计:从离线到在线
1. 离线索引链路
离线阶段决定了在线效果上限。
文档接入
企业知识通常来自:
- Confluence / Wiki
- PDF / Word
- FAQ 页面
- 工单系统
- 数据库中的制度、产品说明、操作手册
接入时重点关注:
- 文档标题
- 正文内容
- 更新时间
- 来源链接
- 部门/权限标签
- 文档版本
文档清洗
常见清洗动作:
- 去掉页眉页脚、导航栏、版权声明
- 修正 OCR 乱码
- 合并被错误拆开的段落
- 保留标题层级
- 补充 metadata,如
doc_id、department、updated_at
切分策略
切分不是越小越好,也不是越大越好。
经验上:
- 过小:上下文不足,答案碎片化
- 过大:检索不精确,token 成本高
企业文档通常可以从这几种策略开始:
- 按标题层级切分:最适合制度和手册
- 固定长度 + overlap:实现简单
- 语义切分:适合 FAQ、说明文档
我一般建议:
- 初版先用 300~800 中文字一个 chunk
- overlap 保持在 50~120 字
- 保留标题路径,比如:
报销制度 > 差旅报销 > 交通票据要求
2. 在线问答链路
在线阶段更看重准确率、延迟和安全。
sequenceDiagram
participant U as 用户
participant API as 问答服务
participant ACL as 权限服务
participant RET as 检索服务
participant RR as 重排服务
participant LLM as 大模型
U->>API: 提问
API->>ACL: 校验用户可访问文档范围
ACL-->>API: 可访问标签/文档集合
API->>RET: 混合检索(问题+权限过滤)
RET-->>API: TopK 候选片段
API->>RR: 重排候选片段
RR-->>API: 高相关上下文
API->>LLM: 问题+上下文+回答约束
LLM-->>API: 答案+引用
API-->>U: 返回结果
查询改写
用户的问题并不总是“适合检索”的。例如:
- “这个怎么提?”
- “最新的政策是什么?”
- “我离职前还有什么要处理?”
系统需要做一层轻量查询改写,把口语化问题扩成适合检索的形式:
- 补足业务实体
- 展开同义词
- 区分“制度问答”和“操作问答”
- 必要时结合对话历史
但要注意:查询改写不能太激进。
改写过头,反而会把用户本意带偏。
容量估算:别等上线了才发现顶不住
做架构时,建议至少估这几项:
1. 文档规模
假设:
- 10 万篇文档
- 平均每篇切成 8 个 chunk
- 总计 80 万个 chunk
如果每个向量 1536 维,float32 存储大致:
- 1536 × 4 bytes ≈ 6 KB / chunk
- 80 万 chunk ≈ 4.8 GB 向量数据
- 再加 metadata、索引结构,通常至少要预留 10~20 GB
这还不算关键词索引和副本。
2. 查询并发
假设:
- 峰值 QPS 20
- 每次查询:
- 向量检索 1 次
- BM25 检索 1 次
- 重排 20 条候选
- LLM 生成 1 次
瓶颈通常不在检索,而在:
- 重排模型
- LLM 推理
- 权限过滤查询
所以典型优化顺序是:
- 缓存高频问题
- 降低候选数和上下文长度
- 让重排模型更轻
- 做异步索引更新
- 再考虑更贵的推理资源
实战代码(可运行)
下面给一个可运行的最小示例。
它不依赖真实大模型,而是用 Python 实现一个简化版 RAG 流程:
- 文档切分
- TF-IDF 向量化
- 混合检索(关键词 + 向量)
- 权限过滤
- 返回带引用的答案
安装依赖:
pip install scikit-learn numpy
代码如下:
from dataclasses import dataclass
from typing import List, Dict, Tuple
import math
import re
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
@dataclass
class Chunk:
chunk_id: str
doc_id: str
title: str
content: str
department: str
source: str
class SimpleRAG:
def __init__(self, chunks: List[Chunk]):
self.chunks = chunks
self.vectorizer = TfidfVectorizer(token_pattern=r"(?u)\b\w+\b")
self.texts = [f"{c.title} {c.content}" for c in chunks]
self.tfidf_matrix = self.vectorizer.fit_transform(self.texts)
def keyword_score(self, query: str, text: str) -> float:
query_terms = [t.strip().lower() for t in re.split(r"\s+", query) if t.strip()]
text_lower = text.lower()
if not query_terms:
return 0.0
hits = sum(1 for term in query_terms if term in text_lower)
return hits / len(query_terms)
def retrieve(
self,
query: str,
user_department: str,
top_k: int = 5,
alpha: float = 0.7
) -> List[Tuple[Chunk, float]]:
# 权限过滤
allowed = [
(idx, c) for idx, c in enumerate(self.chunks)
if c.department == user_department or c.department == "public"
]
if not allowed:
return []
indices = [idx for idx, _ in allowed]
filtered_texts = [self.texts[idx] for idx in indices]
filtered_matrix = self.tfidf_matrix[indices]
# 向量相似度
query_vec = self.vectorizer.transform([query])
vec_scores = cosine_similarity(query_vec, filtered_matrix)[0]
# 关键词分数
kw_scores = [
self.keyword_score(query, text)
for text in filtered_texts
]
# 混合打分
results = []
for i, idx in enumerate(indices):
score = alpha * vec_scores[i] + (1 - alpha) * kw_scores[i]
results.append((self.chunks[idx], float(score)))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
def answer(self, query: str, user_department: str) -> Dict:
candidates = self.retrieve(query, user_department=user_department, top_k=3)
if not candidates or candidates[0][1] < 0.05:
return {
"answer": "我没有在当前权限范围内找到足够可靠的资料,建议换个更具体的问题,或联系知识库管理员。",
"citations": []
}
top_chunks = [c for c, _ in candidates]
context = "\n".join(
[f"[{i+1}] {c.title}: {c.content}" for i, c in enumerate(top_chunks)]
)
# 这里用简化模板模拟生成结果
answer = (
f"根据检索到的知识库内容,和“{query}”最相关的信息如下:\n"
f"{top_chunks[0].content}\n\n"
"如果你要正式执行,请以引用文档原文为准。"
)
citations = [
{
"title": c.title,
"source": c.source,
"department": c.department
}
for c in top_chunks
]
return {
"answer": answer,
"citations": citations,
"debug_context": context
}
if __name__ == "__main__":
chunks = [
Chunk(
chunk_id="c1",
doc_id="d1",
title="差旅报销制度",
content="员工差旅费用应在出差结束后 15 个自然日内提交报销申请,逾期需主管说明原因。",
department="finance",
source="https://kb.local/finance/travel-expense"
),
Chunk(
chunk_id="c2",
doc_id="d1",
title="差旅报销制度",
content="高铁票、机票行程单、酒店发票为差旅报销的基础凭证,缺失时需补充说明。",
department="finance",
source="https://kb.local/finance/travel-expense"
),
Chunk(
chunk_id="c3",
doc_id="d2",
title="VPN 使用说明",
content="员工可在 IT 服务台提交 VPN 申请,审批通过后由系统自动发送客户端配置说明。",
department="public",
source="https://kb.local/it/vpn"
),
Chunk(
chunk_id="c4",
doc_id="d3",
title="离职流程",
content="员工发起离职后,需要在最后工作日前完成资产归还、账号交接和权限回收确认。",
department="hr",
source="https://kb.local/hr/offboarding"
),
]
rag = SimpleRAG(chunks)
queries = [
("报销截止时间", "finance"),
("VPN 怎么申请", "finance"),
("离职前要做什么", "finance"),
]
for q, dept in queries:
result = rag.answer(q, dept)
print("=" * 60)
print("问题:", q)
print("答案:", result["answer"])
print("引用:", result["citations"])
运行后你会看到:
- finance 用户能查到财务制度
- finance 用户也能查到 public 文档
- finance 用户查不到 hr 私有离职流程
这个例子虽然简化,但已经体现了企业 RAG 的几个核心点:
- 不是所有文档都能查
- 不是只靠一个相似度分数
- 答案要带引用
- 找不到时要敢于拒答
关键设计细节:真正影响效果的地方
1. Chunk 设计比模型选择更先影响结果
很多人上来就纠结 embedding 模型选哪个,但我见过更多问题出在 chunk 上:
- 一个 chunk 混了多个主题
- 标题被丢掉了
- 表格被抽成无意义碎片
- 文档版本混杂,没有生效时间
如果你只能优先做一件事,我会建议先把 chunk 做对:
- 保留标题路径
- 保留生效时间/版本
- 表格转成结构化文本
- 同一 chunk 不跨多个制度主题
2. 重排比盲目增大 TopK 更有效
很多团队召回效果不好,就把 TopK 从 5 拉到 20、50、100。
看起来“召回更多了”,但实际问题是:
- 无关内容更多
- LLM 上下文更乱
- token 成本更高
- 答案更容易拼错信息
更稳的做法通常是:
- 混合召回取 20~50 条
- 用轻量 rerank 压到前 3~8 条
- 再构造上下文给模型
3. 拒答能力是企业系统的基本能力
企业场景里,“不知道”比“编一个”更重要。
提示词里最好明确约束:
- 只能根据给定上下文回答
- 没有依据就明确说明未找到
- 不得补全制度细节
- 引用来源必须来自检索结果
常见坑与排查
这一节我尽量讲得接地气一点,因为这些坑真的是上线时最常见的。
坑一:检索结果看起来相关,但答案还是错
现象:
- 检索出的文档主题没错
- 但模型回答抓错了细节,比如“15 天”说成“30 天”
常见原因:
- 上下文中混入了多个版本
- chunk 过大,关键信息被淹没
- prompt 没要求“优先引用明确数字/规则”
排查方法:
- 打印最终传给 LLM 的上下文
- 检查是否存在互相冲突的片段
- 看引用是否来自旧版本文档
- 缩小上下文,只保留前 2~3 个高质量 chunk 做对比
坑二:明明文档存在,却总是检索不到
常见原因:
- 切分把关键信息拆散了
- 文档清洗时把标题或表格丢了
- 查询词和文档词不一致,没有同义词扩展
- 权限过滤过严,在线被排除了
排查方法:
- 先绕过 LLM,只看 retrieval topK
- 用原始问题、改写问题分别测
- 检查 metadata 是否正确写入索引
- 对比有权限和无权限用户的结果差异
坑三:系统越跑越慢
常见原因:
- 每次请求都重新计算不必要的特征
- 候选召回太多,重排太重
- 上下文太长,生成耗时高
- 没有缓存高频问题
排查方法:
把链路耗时拆开记日志:
- query rewrite
- retrieval
- rerank
- prompt build
- llm inference
只看总耗时没意义,必须知道慢在哪。
坑四:回答越“聪明”,风险越高
这是我踩过的一个真实坑:为了让回答更自然,我们把系统提示词写得太开放,结果模型会根据常识补全公司制度里没写的内容。
解决思路:
- 把目标从“更会说”改成“更可靠”
- 明确区分“知识库事实”与“模型解释”
- 高风险场景输出原文摘要 + 引用,不做自由发挥
安全/性能最佳实践
企业知识库问答,安全和性能不是附加项,而是主功能。
安全最佳实践
1. 检索前做权限过滤,而不是回答后再遮罩
正确顺序应该是:
- 确定用户身份
- 获取可访问文档范围
- 在这个范围内检索
- 再生成答案
如果先全库检索,再在结果层做隐藏,很容易泄漏“存在性信息”。
2. 文档级和字段级都要考虑
有些系统只做文档级权限,但企业里常见情况是:
- 文档可见,但部分字段敏感
- 同一条记录里有手机号、邮箱、合同金额等敏感信息
必要时要在索引前做脱敏或字段裁剪。
3. 保留审计日志
至少记录:
- 谁问了什么
- 检索到了哪些文档
- 最终用了哪些引用
- 模型输出了什么
- 是否触发拒答/风控
这样出了问题,才能复盘。
性能最佳实践
1. 做分层缓存
典型缓存层:
- 查询改写结果缓存
- 检索结果缓存
- 高频问答结果缓存
- 文档向量缓存
但注意:带权限的结果缓存必须包含用户权限维度,不然容易串权限。
2. 控制上下文长度
上下文不是越长越好。
建议优先保留:
- 命中分高的 chunk
- 最新版本
- 标题明确、规则明确的片段
删除:
- 重复片段
- 噪声页眉页脚
- 和问题关系弱的背景介绍
3. 异步索引更新
不要每次文档变更都阻塞在线服务。
更常见的做法是:
- 文档更新进入消息队列
- 异步清洗、切分、向量化
- 索引版本化切换
- 在线服务平滑读新版本
flowchart TD
A[文档变更事件] --> B[消息队列]
B --> C[清洗与切分 Worker]
C --> D[向量化 Worker]
D --> E[构建新索引版本]
E --> F[灰度切换]
F --> G[在线问答服务]
4. 给高风险问题单独策略
比如:
- 财务制度
- 法务条款
- 人事政策
- 安全操作规程
这些场景建议:
- 降低生成自由度
- 必须带引用
- 置信度低时直接拒答
- 必要时只返回“相关文档列表 + 摘要”
一套更稳的落地建议
如果你准备在团队里真正推进一个企业 RAG 项目,我建议按下面顺序做:
第一阶段:先把链路跑通
目标:
- 文档能接入
- 能切 chunk
- 能检索
- 能回答并带引用
此阶段不要过早追求“模型最强”,重点是可观察性:
- 能看到每次召回了什么
- 能看到最终上下文是什么
- 能看到为什么答错
第二阶段:把效果做稳
重点做三件事:
- 混合检索
- 重排
- 版本与权限治理
这一步对实际体验提升最大。
第三阶段:把成本和延迟打下来
重点优化:
- 缓存
- 上下文裁剪
- 更轻的 rerank 模型
- 分层服务拆分
第四阶段:做运营闭环
真正好用的企业知识库,最后都离不开运营:
- 收集未命中问题
- 标注错误答案
- 补齐高频 FAQ
- 定期清理过期文档
RAG 不是“一次搭好永远可用”,而是一个持续迭代的系统。
总结
企业内部知识库问答系统,表面上是“大模型应用”,本质上更像一个检索、权限、数据治理和生成控制的组合工程。
如果你要抓住最核心的落地原则,我建议记住这几条:
- 不要只做纯向量检索,企业场景优先考虑混合检索
- chunk 质量往往比换模型更影响结果
- 先做权限过滤,再做检索和生成
- 重排比一味增大 TopK 更有效
- 要敢于拒答,别让模型瞎补
- 从第一天起就做好日志、引用和审计
最后给一个很务实的边界判断:
- 如果你的知识库规模小、权限简单,可以先做轻量 RAG
- 如果涉及多部门、敏感制度、强审计要求,就必须按生产架构来设计
- 如果你发现问题总集中在“检索不到”或“答非所问”,先别急着换大模型,先回头看文档清洗、切分和召回链路
把这些基础打牢,RAG 才不是一个“能演示”的 Demo,而是一个真正能在企业里长期运行的系统。