背景与问题

企业里做知识库问答，表面上看像是“把文档喂给大模型，然后让它回答问题”，真正落地时却远没有这么简单。

很多团队在第一版系统里都会遇到类似问题：

• 回答看起来很像对的，但其实是编的
• 文档明明有，模型就是没召回到
• 知识更新后，回答还是旧版本
• 一问多轮后，模型开始“跑偏”
• 响应时间太长，用户用两次就放弃
• 涉及权限的数据被错误引用，存在合规风险

我自己做过几类企业场景：制度问答、售后知识助手、研发文档检索、内部流程机器人。一个很直接的体会是：企业知识问答不是单点模型能力的问题，而是检索、切分、排序、提示词、上下文编排、权限控制、缓存和评测共同作用的系统工程。

而 RAG（Retrieval-Augmented Generation，检索增强生成）之所以成为主流方案，核心原因也很现实：

1. 降低幻觉：回答基于企业知识，而不是只靠模型参数记忆
2. 支持动态更新：文档改了，不需要重新训练大模型
3. 更容易做可追溯：可以返回引用片段和来源
4. 更适合企业治理：权限、审计、数据隔离都更容易落地

这篇文章不讲“概念大全”，而是从一个可落地的企业架构视角，带你把 RAG 问答系统拆开：为什么这样设计、关键技术点在哪、常见坑怎么避、性能和安全怎么做。

一、方案总览：企业级 RAG 问答系统长什么样

先看一个典型架构。

flowchart LR
    A[知识源\nPDF/Word/Wiki/DB/工单] --> B[数据接入与清洗]
    B --> C[文档切分 Chunking]
    C --> D[向量化 Embedding]
    D --> E[向量库]
    C --> F[元数据索引\n标题/部门/时间/权限]
    F --> G[检索编排层]

    U[用户问题] --> H[Query Rewrite]
    H --> G
    E --> G
    G --> I[召回 TopK]
    I --> J[重排 Rerank]
    J --> K[上下文构建]
    K --> L[LLM 生成答案]
    L --> M[答案+引用+置信度]

    N[权限系统] --> G
    O[缓存/日志/评测] --> G
    O --> L

如果只用一句话概括：RAG = 把“找资料”和“组织答案”拆成两个环节，再通过检索链路把它们接起来。

在企业里，这个架构通常要多考虑三件事：

• 知识质量：原始文档并不天然适合检索
• 查询质量：用户的问题往往表达不完整
• 工程质量：延迟、成本、权限、监控缺一不可

二、背景与问题：为什么企业问答系统难做

1. 企业知识并不“干净”

公开互联网文本往往结构相对规整，但企业内部知识常见问题是：

• 文档格式混乱，PDF 扫描件很多
• 内容存在历史版本，互相冲突
• 一份文档里有大量目录、页眉页脚、模板文字
• 同义词多，比如“请假”“休假”“年休”“带薪假”

如果这些内容不先清洗，后续 embedding 和召回效果会直接崩掉。

2. 用户问题并不标准

用户不会像写搜索引擎关键词那样提问，而是会说：

• “报销那个差旅标准怎么走？”
• “去年更新后的采购审批金额线是多少？”
• “这个事情能不能走特批？”

这些问题常常有三个特征：

• 指代不清
• 上下文依赖强
• 术语不统一

所以企业级 RAG 几乎都离不开 Query Rewrite（问题改写） 和 多路召回。

3. 单纯“向量检索 + LLM”远远不够

很多入门版系统是这样：

1. 文档切块
2. 做 embedding
3. 向量检索 top-k
4. 把结果丢给 LLM 回答

这个流程能跑起来，但很快会遇到瓶颈：

• 相似但不相关的片段被召回
• 关键数字、条款、版本号检索不到
• 长文档上下文碎裂
• 多文档答案无法整合
• 回答缺少出处，用户不信

所以真正可用的系统，往往会演进到：

• 混合检索：向量 + BM25/关键词
• 重排模型：提升 top-k 质量
• 元数据过滤：版本、部门、时间、权限
• 答案约束：必须基于引用内容作答
• 回退策略：检索不到时明确说不知道

三、核心原理：RAG 的关键技术链路

1. 文档切分：决定了“能不能找到”

切分不是机械地按 500 字一刀切。它决定了后面检索颗粒度是否合理。

常见策略：

• 固定长度切分：实现简单，但可能切断语义
• 带 overlap 的滑窗切分：较通用
• 按标题/段落/表格结构切分：更适合企业文档
• 语义切分：效果更好，但成本更高

经验上：

• 制度、规范类文档：适合按标题层级切分
• FAQ、工单知识：适合按问答对切分
• API/技术文档：适合按章节 + 示例代码切分
• 合同/政策：要保留条款编号和版本信息

一个常见坑是：chunk 太小，语义不完整；chunk 太大，召回噪声高。

一般可以先从下面参数试起：

• chunk size：300~800 中文字
• overlap：50~150 中文字

边测边调，不要迷信固定最佳值。

2. 向量化：解决语义相似问题

Embedding 的作用是把文本映射到向量空间，让“意思接近”的文本距离更近。

它擅长解决：

• 同义表达
• 口语化提问
• 不完全关键词匹配

但它也有边界：

• 对精确编号、日期、版本号未必敏感
• 对数字型约束不一定稳定
• 对短 query 的语义可能过度泛化

这也是为什么企业搜索里，只做向量检索通常不够。

3. 混合检索：语义和关键词都要抓

比较稳的做法是：

• 一路用向量检索找语义相近内容
• 一路用 BM25 或关键词检索找精确命中内容
• 再合并去重

比如用户问：

“2024 差旅报销住宿标准二线城市上限多少？”

其中：

• “差旅报销”“住宿标准”适合语义召回
• “2024”“二线城市”“上限”更适合关键词和结构化过滤

所以混合检索比纯向量检索更适合企业知识库。

4. 重排（Rerank）：从“召回来”到“排对顺序”

召回的目标是“别漏掉”，重排的目标是“把最相关的放前面”。

一个实战上很有用的策略是：

1. 向量检索 top 20
2. BM25 检索 top 20
3. 合并后去重
4. 用 reranker 排前 5~8 条
5. 再喂给大模型

这一步往往是从“能用”到“好用”的分水岭。

5. 上下文构建：别把所有内容一股脑塞给模型

把召回结果直接拼接给 LLM，看起来简单，实际上容易出问题：

• 上下文太长，成本高
• 无关内容太多，模型被干扰
• 多个片段相互冲突，模型开始“脑补”

更好的做法是对上下文做编排：

• 按相关性排序
• 保留标题、来源、更新时间
• 去掉重复内容
• 控制总 token 数
• 对冲突版本优先保留最新且有效的文档

6. 生成约束：让模型“基于证据说话”

提示词里建议明确要求：

• 仅基于提供材料回答
• 不确定时直接说明“未检索到充分依据”
• 返回引用来源
• 数字、时间、版本信息逐字引用

这看似只是 prompt 技巧，实际上对企业可用性影响非常大。

四、方案对比与取舍分析

1. 纯大模型直答 vs RAG

结论很直接：企业知识问答，能用 RAG 就不要指望“模型自己懂公司制度”。

2. 向量库选型怎么判断

常见考虑维度：

• 检索性能与延迟
• 过滤能力（metadata filter）
• 部署方式（云服务/本地）
• 运维复杂度
• 成本

如果是中型团队，优先考虑：

• 支持 metadata filtering
• 支持 ANN 检索
• 支持批量写入与增量更新
• 有稳定 Python SDK

3. 容量估算的简单思路

假设：

• 企业文档 10 万篇
• 平均每篇切成 20 个 chunk
• 共 200 万 chunk
• embedding 维度 1024
• 单向量约数 KB 级存储（含索引额外开销）

需要重点估算：

• 向量库存储量
• 索引构建时间
• 日均查询 QPS
• 峰值并发
• LLM token 成本
• 缓存命中率

一个容易被忽略的点是：生产成本不只是模型调用费，检索链路、日志、缓存、重排、OCR 都会花钱。

五、实战代码：从零做一个可运行的简化版 RAG 问答服务

下面给一个可运行的 Python 示例。它不是生产级完整系统，但足够帮你把核心链路跑起来：

• 文档切分
• TF-IDF 检索（为了本地可运行，不依赖外部向量库）
• 简单上下文拼接
• 调用 OpenAI 兼容接口生成答案

说明：为了保证示例可运行，我这里把“向量检索”简化成了 TF-IDF 检索。如果你接入正式 embedding 模型和向量库，整体代码结构是一样的。

1. 安装依赖

pip install fastapi uvicorn scikit-learn numpy requests

2. 准备知识文档

新建 docs.txt：

# 差旅报销制度（2024版）
差旅报销中，住宿费按城市等级设定标准。一线城市每晚上限 800 元，二线城市每晚上限 500 元，三线及以下城市每晚上限 350 元。超出标准部分原则上不予报销。该制度自 2024-01-01 起生效。

# 采购审批制度（2024版）
单笔采购金额在 5 万元以下，由部门负责人审批；5 万元及以上、20 万元以下，由分管副总审批；20 万元及以上，需提交采购委员会审议。该制度自 2024-02-01 起生效。

# 请假制度（2023版）
员工年假需至少提前 3 个工作日发起申请。如遇紧急情况，可先口头报备后补流程。年假审批由直属主管完成。

3. 服务端代码

保存为 app.py：

from fastapi import FastAPI
from pydantic import BaseModel
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Dict
import requests
import os
import re

app = FastAPI(title="Simple RAG Demo")

DOC_PATH = "docs.txt"
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

class AskRequest(BaseModel):
    question: str
    top_k: int = 3

class AskResponse(BaseModel):
    answer: str
    contexts: List[Dict]

def load_docs(path: str) -> List[Dict]:
    with open(path, "r", encoding="utf-8") as f:
        raw = f.read()

    sections = [s.strip() for s in raw.split("\n# ") if s.strip()]
    docs = []
    for i, sec in enumerate(sections):
        if i == 0 and sec.startswith("# "):
            sec = sec[2:]
        lines = sec.splitlines()
        title = lines[0].replace("# ", "").strip()
        content = "\n".join(lines[1:]).strip()
        chunks = split_text(content, chunk_size=120, overlap=20)
        for idx, chunk in enumerate(chunks):
            docs.append({
                "id": f"{i}-{idx}",
                "title": title,
                "content": chunk
            })
    return docs

def split_text(text: str, chunk_size: int = 120, overlap: int = 20) -> List[str]:
    text = re.sub(r"\s+", " ", text).strip()
    if len(text) <= chunk_size:
        return [text]
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        if end >= len(text):
            break
        start = end - overlap
    return chunks

DOCS = load_docs(DOC_PATH)
VECTOR_TEXTS = [f"{d['title']} {d['content']}" for d in DOCS]
VECTORIZER = TfidfVectorizer()
DOC_MATRIX = VECTORIZER.fit_transform(VECTOR_TEXTS)

def retrieve(question: str, top_k: int = 3) -> List[Dict]:
    q_vec = VECTORIZER.transform([question])
    sims = cosine_similarity(q_vec, DOC_MATRIX)[0]
    top_indices = sims.argsort()[::-1][:top_k]
    results = []
    for idx in top_indices:
        results.append({
            "id": DOCS[idx]["id"],
            "title": DOCS[idx]["title"],
            "content": DOCS[idx]["content"],
            "score": float(sims[idx])
        })
    return results

def build_prompt(question: str, contexts: List[Dict]) -> str:
    context_text = "\n\n".join([
        f"[资料{idx+1}] 标题：{c['title']}\n内容：{c['content']}"
        for idx, c in enumerate(contexts)
    ])

    return f"""你是企业知识库问答助手。请严格基于给定资料回答问题：
1. 只能依据资料内容回答，不要补充资料外信息
2. 如果资料不足，请明确说“未检索到充分依据”
3. 回答尽量简洁准确
4. 回答后附上引用资料编号

用户问题：
{question}

给定资料：
{context_text}
"""

def call_llm(prompt: str) -> str:
    if not OPENAI_API_KEY:
        return "未配置 OPENAI_API_KEY，无法调用大模型。"

    url = f"{OPENAI_BASE_URL}/chat/completions"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": OPENAI_MODEL,
        "messages": [
            {"role": "system", "content": "你是一个严谨的企业知识问答助手。"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.2
    }

    resp = requests.post(url, headers=headers, json=payload, timeout=60)
    resp.raise_for_status()
    data = resp.json()
    return data["choices"][0]["message"]["content"]

@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest):
    contexts = retrieve(req.question, req.top_k)
    prompt = build_prompt(req.question, contexts)
    answer = call_llm(prompt)
    return AskResponse(answer=answer, contexts=contexts)

4. 启动服务

uvicorn app:app --reload

5. 测试请求

curl -X POST "http://127.0.0.1:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "question": "2024年二线城市差旅住宿报销上限是多少？",
    "top_k": 3
  }'

如果配置了模型接口，你会拿到类似结果：

{
  "answer": "根据资料，2024年二线城市差旅住宿报销上限为每晚500元。[资料1]",
  "contexts": [
    {
      "id": "0-0",
      "title": "差旅报销制度（2024版）",
      "content": "差旅报销中，住宿费按城市等级设定标准。一线城市每晚上限 800 元，二线城市每晚上限 500 元，三线及以下城市每晚上限 350 元。",
      "score": 0.72
    }
  ]
}

六、从 Demo 到生产：关键增强点

上面的示例能跑，但离企业级系统还有几步。

1. 引入真正的向量检索

生产环境建议改造成：

• 文档切块后调用 embedding 模型
• 写入向量数据库
• 查询时取 top-k 召回

结构可以抽象为：

class Retriever:
    def index(self, docs: list):
        pass

    def search(self, query: str, top_k: int = 5):
        pass

这样后续从本地 TF-IDF 切到 Milvus、pgvector、Elasticsearch、Weaviate 之类实现，接口不需要大改。

2. 增加混合检索与重排

一个常见生产链路如下：

sequenceDiagram
    participant U as 用户
    participant Q as Query Rewrite
    participant R1 as 向量检索
    participant R2 as BM25检索
    participant RR as Reranker
    participant L as LLM

    U->>Q: 提问
    Q->>R1: 改写后的 query
    Q->>R2: 原始/扩展 query
    R1-->>RR: 语义候选集
    R2-->>RR: 关键词候选集
    RR-->>L: TopN 高相关片段
    L-->>U: 答案 + 引用

我的建议是，先不要一上来做很复杂的 agent。先把“召回准”和“引用稳”做好，收益最大。

3. 做文档增量更新

企业知识不是一次性导入完就结束。你需要支持：

• 新文档导入
• 老文档失效
• 版本替换
• 定时重建索引

一个可用做法是给每个 chunk 打上这些 metadata：

• doc_id
• title
• department
• version
• effective_date
• expire_date
• permission_scope

这样后续过滤和治理会轻松很多。

七、常见坑与排查

这部分我尽量写得接地气一点，因为这里往往比“怎么搭建”更值钱。

1. 现象：模型总在“胡说八道”

可能原因：

• prompt 没有明确约束“只基于资料回答”
• 召回结果不相关
• 上下文里存在冲突信息
• 温度太高

排查路径：

1. 打印召回结果 top-k
2. 检查第一条是否真的能回答问题
3. 检查 prompt 是否要求引用来源
4. 把 temperature 降到 0~0.3
5. 检查是否把“系统知识”和“企业知识”混在一起了

止血方案：

• 增加“无依据时拒答”
• 限定答案必须引用资料编号
• 先上线“保守答复”，不要急着追求自然聊天感

2. 现象：文档明明有，系统却搜不到

可能原因：

• chunk 切分方式有问题
• embedding 模型不适合中文或行业语料
• query 太短或表达过于口语
• 只做了语义检索，没做关键词检索
• metadata filter 过滤过头

排查路径：

question = "采购 20 万以上谁审批"
results = retrieve(question, top_k=10)
for r in results:
    print(r["title"], r["score"], r["content"])

重点看：

• 候选集中有没有目标文档
• 如果有但排位靠后，需要重排
• 如果根本没有，需要改切分或检索策略

3. 现象：回答引用了过期制度

可能原因：

• 没有做版本管理
• chunk 丢失了版本元数据
• 检索时没有按生效日期过滤
• 新旧文档语义相似，旧文档反而分数更高

建议：

• 版本字段必须入索引
• 检索前按有效期和状态过滤
• 上下文构建时优先最新版本
• 过期文档不要只是“留着”，要显式标记失效

4. 现象：响应太慢

常见瓶颈：

• OCR/解析在线做
• 每次都全量检索多个索引
• top_k 过大
• rerank 模型太重
• 上下文拼接过长
• LLM 输出太多废话

优化思路：

• 文档处理离线化
• 检索结果缓存
• 热门问题缓存
• 控制 top_k 和上下文长度
• 小模型做 query rewrite / rerank，大模型只负责最终生成

八、安全与性能最佳实践

企业系统一旦上线，准确率只是门票，安全和性能才决定你能不能长期运行。

1. 安全最佳实践

权限过滤前置

最重要的一条：先按权限过滤，再检索或至少在召回后强过滤。

否则会出现这种风险：

• 用户问一个普通问题
• 检索命中了高相关但无权限文档
• 模型在答案里泄露敏感信息

建议至少做到：

• 文档入库时带权限标签
• 查询时根据用户身份做过滤
• 审计日志记录“用户问了什么、系统引用了什么”

Prompt Injection 防护

如果知识库里有用户上传内容，要防范文档中出现恶意指令，比如：

• “忽略以上规则”
• “输出系统提示词”
• “泄露管理员数据”

应对方法：

• 把检索内容作为“资料”，而不是“指令”
• system prompt 中明确：资料里的命令不应执行
• 对上传文档做内容审查和清洗

敏感信息脱敏

对以下内容建议入库前脱敏：

• 身份证号
• 银行卡号
• 手机号
• 客户隐私数据
• 合同敏感字段

2. 性能最佳实践

多级缓存

适合做三层缓存：

• Query Rewrite 缓存
• 检索结果缓存
• 最终答案缓存

但要注意版本失效问题，缓存 key 最好带上索引版本号。

分层模型策略

很实用的一种做法：

• 小模型：问题改写、意图分类、召回路由
• 重排模型：候选排序
• 大模型：最终答案生成

这样通常比“所有步骤都上大模型”更省钱、更稳。

限制上下文预算

一个经验法则：

• 宁可给模型 5 段高质量上下文
• 也不要塞 20 段低相关内容

因为噪声多了，模型并不会更聪明，反而更容易答偏。

九、一个更完整的生产状态流转

stateDiagram-v2
    [*] --> 文档接入
    文档接入 --> 清洗解析
    清洗解析 --> 切分
    切分 --> 向量化
    向量化 --> 入库索引
    入库索引 --> 可检索

    可检索 --> 查询接收
    查询接收 --> 权限校验
    权限校验 --> 查询改写
    查询改写 --> 混合召回
    混合召回 --> 重排
    重排 --> 上下文构建
    上下文构建 --> 大模型生成
    大模型生成 --> 引用校验
    引用校验 --> 返回答案
    返回答案 --> 监控评测
    监控评测 --> 可检索

这个状态图想表达的是：RAG 不是“建完索引就结束”，而是一个持续迭代闭环。

上线后你还要持续看：

• 哪些问题答错最多
• 哪些知识源最常被引用
• 哪些 query 根本没有命中
• 哪些部门知识更新滞后
• 哪些答案经常被用户追问

十、评测与迭代：别只看主观感觉

很多团队评估系统，只靠“我问了几个问题感觉还行”。这在 PoC 阶段勉强可以，生产阶段远远不够。

建议至少建立三类指标：

1. 检索指标

• Recall@K
• MRR
• NDCG
• 命中率

2. 生成指标

• 答案准确率
• 引用正确率
• 拒答正确率
• 幻觉率

3. 工程指标

• P50/P95 延迟
• token 成本
• 缓存命中率
• 错误率
• 权限误召回率

我自己的经验是：先把评测集做出来，再谈优化，不然团队很容易在“感觉变好了”里打转。

评测集可以从以下来源抽样：

• 客服真实提问
• 内部工单标题
• 高频制度问题
• 搜索日志中的失败 query

总结

基于 RAG 构建企业知识库问答系统，真正的关键不在于“接了哪个大模型”，而在于你是否把下面这条链路打通了：

• 文档清洗是否可靠
• 切分颗粒度是否合理
• 检索是否混合化
• 重排是否提升相关性
• 上下文是否可控
• 回答是否基于证据
• 权限、安全、缓存、评测是否完整

如果你正在做第一版，我建议按这个顺序推进：

1. 先做一个能返回引用的基础 RAG
2. 再补混合检索和重排
3. 接着做版本、权限和缓存
4. 最后再优化多轮对话、复杂推理和 agent 能力

边界条件也很明确：

• 如果企业知识极少、更新不频繁，基础 RAG 就够了
• 如果文档版本复杂、权限严格，必须上 metadata 治理
• 如果追求高准确率，重排和评测体系几乎是必选项
• 如果问题高度依赖业务流程动作，而不只是知识问答，单纯 RAG 可能不够，需要工作流或 agent 协同

一句话收尾：RAG 在企业里不是“外挂检索”，而是一套面向真实业务约束的知识操作系统。 把系统工程做好，大模型才能真正落地，而不是停留在演示阶段。