企业级RAG知识库完整复盘：从技术选型到生产架构
一、背景：为什么要做生产级RAG系统
大模型应用越来越热，知识库、智能客服、文档问答等企业需求随处可见。但潜在合作方的第一个问题永远是："你有没有做过的案例？"
一个能跑通的 Demo 说服不了任何人——Chroma + OpenAI API 五分钟搭起来的东西，不足以称为作品。因此需要按真实企业需求的标准，从头搭建一套完整的 RAG 知识库，覆盖文档处理、检索优化、多租户、权限、性能、成本等全部环节。
💡
核心认知：企业项目不是 Demo 的放大版，而是另一个物种。
二、技术选型
2.1 常见 Demo 方案的局限
典型 Demo 技术栈：前端 Vue3 + Element Plus，后端 FastAPI，AI 框架 LangChain，向量数据库 Chroma，大模型 OpenAI GPT-4。能跑，但仅限于演示。企业场景一来，每个环节都出问题：
•
数据安全：很多企业文档不能出境，OpenAI 用不了，需准备国产模型或本地部署方案。
•
数据量：Chroma 处理几千条还行，几十万份文档上去检索延迟直接爆炸。
•
并发：中型企业几十人同时查询，系统不能排队。
•
权限：不同部门看不同文档，不同角色不同操作，Demo 里根本没有考虑。
•
运维：文档要更新，模型要监控，成本要控制，不是跑起来就完事。
2.2 最终选型
| | |
| | |
| | |
| | |
| | |
| | |
三、文档处理：四个核心坑与解法
坑 1：扫描件 PDF
企业文档中大量扫描件是纯图片而非文字型 PDF，必须先 OCR。中文 OCR 对正式文档还行，但一行错一个字，检索效果就大打折扣。
解法：PaddleOCR 做中文识别，对关键字段（合同金额、日期、当事人名称）做二次校验。
坑 2：表格数据
合同中的金额表格、赔偿标准表，常规 PDF 解析器直接当乱码丢弃。用户问"违约金怎么算"，AI 根本找不到那条数据——数据在表格里，解析器压根没读出来。
解法：MinerU 解析 PDF，能识别表格并保留结构。表格转成 Markdown 格式存入向量库，检索时和文字同等对待。
坑 3：文档层级结构
法律文书有"条→款→项"的层级，合同有"第一章→第一节→第一条"。RecursiveCharacterTextSplitter 按字符数切分，用户问"第三条第二款是什么"，检索出来的片段牛头不对马嘴。
解法：自定义 Chunker，按文档层级结构切分，每个 chunk 保留父级标题路径：
按文档层级结构切分的自定义 Chunker
Python
class HierarchicalChunker:
"""按文档层级结构切分"""
def split(self, document: str) -> List[Chunk]:
sections = self._parse_hierarchy(document)
chunks = []
坑 4：批量处理性能
一份文档解析 + Embedding 约 15 秒，2000 份同步处理需要 8 个多小时。
解法：Celery 异步任务队列，10 个 Worker 并行，WebSocket 实时推送进度。2000 份文档 25 分钟搞定。
四、检索质量优化：从 67% 到 92%
Demo 跑出来准确率约 67%——问 10 个问题，6-7 个能命中相关内容，离可用差得远。通过四步优化逐步提升到 92%。
第一步：混合检索（67% → 78%）
纯向量检索对精确查询很弱，例如"民法典第 1043 条"，向量相似度不如关键词匹配。同时跑向量检索和 BM25 关键词检索，然后按权重合并结果：
混合检索：BM25 + 向量检索
Python
from langchain.retrievers import EnsembleRetriever
vector_retriever = milvus.as_retriever(search_kwargs={"k": 20})
bm25_retriever = BM25Retriever.from_documents(docs, k=20)
hybrid_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.3, 0.7]
)
第二步：查询重写（78% → 85%）
用户输入和文档表述差距大，例如"那个赔偿怎么算" vs 文档里的"劳动合同解除的经济补偿金计算标准"。用 LLM 把用户的口语问题转成正式检索词，必要时拆成多个子查询：
查询重写 Prompt
Python
rewrite_prompt = """将用户的自然语言问题改写为更适合检索的表述：
1. 使用正式术语
2. 补充隐含上下文
3. 可以拆分为多个子查询（用 | 分隔）"""
# "那个赔偿怎么算"
# → "劳动合同解除经济补偿金计算标准 | 离职赔偿金计算方法"
第三步：Reranking（85% → 90%）
混合检索召回 20 条，但 Embedding 的排序不一定准。BGE-Reranker 做 Cross-Encoder 重排序，从 20 条精选 Top 5。这一步最直接——准确率从 85% 跳到 90%。
BGE-Reranker 重排序
Python
from FlagEmbedding import FlagReranker
第四步：多轮检索（90% → 92%）
复杂问题一步检索搞不定，例如"解除合同后竞业限制补偿金怎么算"——需要先检索"解除合同"条款，再检索"竞业限制"条款。LangGraph 做多步检索状态机，把"分解→检索→生成"串成有状态的工作流：
LangGraph 多轮检索工作流
Python
from langgraph.graph import StateGraph
workflow = StateGraph(AgentState)
workflow.add_node("decompose", decompose_query)
四步优化总结
| | | |
| | | |
| | | |
| | | |
五、生产架构设计
技术选型和检索优化是基本功。一个系统算不算"生产级"，看的是多租户、权限、成本控制和监控这些能力。
5.1 多租户
不同企业用同一套系统，数据必须隔离。检索时带上租户过滤条件，Milvus 的标量过滤在向量检索前就完成数据隔离：
租户级数据隔离
Python
results = milvus.search(
expr=f'tenant_id == "{tenant_id}"',
data=[query_embedding],
limit=10
)
tenant_config 表存储每个租户的独立配置：llm_api_key、model_name、daily_token_limit、embedding_model、storage_quota。租户启动时加载自己的配置，互不干扰。每个租户可以用不同的 LLM、Embedding 模型和用量上限。
5.2 权限控制
文档级别的访问控制：管理员看所有，普通员工只看自己部门的。检索时根据用户角色过滤可访问的文档，权限隔离在向量检索层面完成：
基于角色的权限过滤
Python
5.3 成本控制
API 调用不是免费的。每个租户的每日 Token 用量、费用追踪、80% 预警、超预算自动降级——这些都不是 Demo 会考虑的，但上线以后必须做。
具体做法：每次 LLM 调用完成后，记录 tenant_id、model_name、prompt_tokens、completion_tokens、cost 到 token_usage 表。费用按模型单价实时计算——通义千问和 GPT-4 的单价差好几倍，不能用同一个公式。
每天凌晨跑定时任务，汇总每个租户当天的 Token 消耗和费用。预警分两级：
•
80% 预警：当天用量达到预算的 80%，自动发通知给租户管理员。
•
100% 熔断：当天用量超出预算，自动切换到本地模型处理后续请求，同时发送告警。月初自动重置计数器。
成本预警与熔断核心逻辑
Python
5.4 监控
Prometheus + Grafana 监控核心指标，每个都设了告警阈值：
| | |
| | |
| | |
| | |
| | |
Sentry 收集错误日志。不是"看起来没问题"，是"有数据证明没问题"。
💡
以上多租户、权限、成本控制、监控，没有一个是 Demo 阶段会考虑的问题。但它们恰恰决定了一个系统能不能真正投入使用。
六、经验总结
Demo 是一回事，跑起来是另一回事。文档处理、权限、多租户、监控、成本控制——每个模块单独看都不难，合在一起调试时全是问题。花在联调上的时间比写代码还多。但恰恰是这些"麻烦事"才值钱，因为大部分教程不写这些。
检索优化是迭代出来的，不是一步到位的。先上混合检索，再逐步加 Reranking 和多轮检索。每步都用真实测试集验证。
国产模型 + 云端 API 的混合方案，目前最务实。敏感数据走本地，普通查询走云端。成本和效果平衡得最好。
全栈能力 = 独立交付能力。前端、后端、AI 引擎、部署运维一手搞定，才能把一个想法变成能跑的系统。