这是RAG系列的最后一篇。前面6篇我们从原理讲到高级架构,从切分讲到评估,你的RAG应该已经能在本地跑出不错的效果了。但"本地跑得好"跟"线上稳定运行"之间隔着十万八千里。

我见过太多RAG项目,演示的时候很完美,上线一周就翻车:文档更新了但向量库没更新、检索延迟突然从200ms飙升到5秒、Embedding API限流导致整个服务不可用、新增一个客户的数据后检索结果混了……

今天这篇我把自己在生产环境踩过的坑整理成一份部署指南,照着做你的RAG能少走半年弯路。

架构设计:RAG服务的标准分层

生产级RAG不是"一个Python脚本跑个Flask",而是分层架构:

<span leaf="">┌──────────────────────────────────┐</span><br><span leaf="">│ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;接入层 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │ &nbsp;API网关、认证、限流</span><br><span leaf="">├──────────────────────────────────┤</span><br><span leaf="">│ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;编排层 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │ &nbsp;查询路由、对话管理</span><br><span leaf="">├──────────────────────────────────┤</span><br><span leaf="">│ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;检索层 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │ &nbsp;混合检索、Rerank</span><br><span leaf="">├──────────────────────────────────┤</span><br><span leaf="">│ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;生成层 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │ &nbsp;LLM调用、流式输出</span><br><span leaf="">├──────────────────────────────────┤</span><br><span leaf="">│ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;存储层 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │ &nbsp;向量库、文档库、缓存</span><br><span leaf="">├──────────────────────────────────┤</span><br><span leaf="">│ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;数据管道 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │ &nbsp;文档入库、增量更新</span><br><span leaf="">└──────────────────────────────────┘</span><br>

每一层都可以独立扩展和替换。别把所有逻辑写在一个文件里——后面改一个地方牵一发动全身。

Java类比:这就像你搭一个Spring Boot微服务架构——API网关(接入层)→业务Service(编排层)→数据访问层(检索层)→外部服务调用(生成层)→数据库(存储层)。分层解耦是工程基本功,RAG也一样。

增量索引:文档更新不用重建

生产环境的文档是动态变化的——每天都有新文档入库、旧文档更新、错误文档删除。你不能每次变更都重建整个向量库。

增量更新流程

<span><span leaf="">import</span></span><span leaf="">&nbsp;hashlib</span><br><span><span leaf="">from</span></span><span leaf="">&nbsp;datetime&nbsp;</span><span><span leaf="">import</span></span><span leaf="">&nbsp;datetime</span><br><br><span><span><span leaf="">class</span></span><span leaf="">&nbsp;</span><span><span leaf="">DocumentManager</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""文档增量管理器"""</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">__init__</span></span><span><span leaf="">(self, vectorstore, doc_registry)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; self.vectorstore = vectorstore</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; self.registry = doc_registry &nbsp;</span><span><span leaf=""># 文档注册表(数据库)</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">_get_doc_hash</span></span><span><span leaf="">(self, content: str)</span></span><span leaf="">&nbsp;-&gt; str:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""计算文档内容哈希"""</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;hashlib.md5(content.encode()).hexdigest()</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">add_document</span></span><span><span leaf="">(self, doc_path: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""新增文档"""</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; content = self._load_document(doc_path)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; doc_hash = self._get_doc_hash(content)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 检查是否已存在</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; existing = self.registry.get(doc_path)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;existing&nbsp;</span><span><span leaf="">and</span></span><span leaf="">&nbsp;existing[</span><span><span leaf="">"hash"</span></span><span leaf="">] == doc_hash:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print(</span><span><span leaf="">f"文档</span><span><span leaf="">{doc_path}</span></span><span leaf="">未变化,跳过"</span></span><span leaf="">)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 切分 + 向量化 + 入库</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; chunks = self.splitter.split_documents(content)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; ids = self.vectorstore.add_documents(chunks)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 更新注册表</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; self.registry.set(doc_path, {</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"hash"</span></span><span leaf="">: doc_hash,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"chunk_ids"</span></span><span leaf="">: ids,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"updated_at"</span></span><span leaf="">: datetime.now().isoformat(),</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; })</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">update_document</span></span><span><span leaf="">(self, doc_path: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""更新文档:删旧的→加新的"""</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; existing = self.registry.get(doc_path)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;existing:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 删除旧chunks</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">for</span></span><span leaf="">&nbsp;chunk_id&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;existing[</span><span><span leaf="">"chunk_ids"</span></span><span leaf="">]:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.vectorstore.delete([chunk_id])</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 重新入库</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; self.add_document(doc_path)</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">delete_document</span></span><span><span leaf="">(self, doc_path: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""删除文档"""</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; existing = self.registry.get(doc_path)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;existing:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">for</span></span><span leaf="">&nbsp;chunk_id&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;existing[</span><span><span leaf="">"chunk_ids"</span></span><span leaf="">]:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.vectorstore.delete([chunk_id])</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.registry.delete(doc_path)</span><br>

定时同步任务

<span><span leaf="">from</span></span><span leaf="">&nbsp;apscheduler.schedulers.background&nbsp;</span><span><span leaf="">import</span></span><span leaf="">&nbsp;BackgroundScheduler</span><br><br><span leaf="">scheduler = BackgroundScheduler()</span><br><br><span><span leaf=""># 每小时扫描一次文档目录</span></span><br><span leaf="">scheduler.add_job(</span><br><span leaf="">&nbsp; &nbsp; scan_and_sync_documents,</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">"interval"</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; hours=</span><span><span leaf="">1</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; args=[DOCUMENT_DIR, doc_manager],</span><br><span leaf="">)</span><br><br><span leaf="">scheduler.start()</span><br>

关键注意点

  1. 文档版本管理:给每个文档存版本号或哈希值,避免重复入库

  2. 原子性:删旧→加新要在一个事务里,中间挂了不能出现"旧删了新没加"的状态

  3. 向量ID映射:chunk在向量库里的ID要跟文档注册表关联,否则删不了

  4. 批量操作:大量文档更新时用批量写入,别一条条加

监控告警:出了问题马上知道

RAG系统上线后,这几个指标必须监控:

关键指标

| 指标

|

告警阈值

|

说明

检索延迟P95

|

> 500ms

|

向量库或Rerank出问题

| |

生成延迟P95

|

> 5s

|

模型限流或网络问题

| |

检索空结果率

|

> 10%

|

Embedding模型故障或知识库缺失

| |

Faithfulness得分

|

< 0.7

|

模型在编造答案

| |

Embedding API错误率

|

> 5%

|

API Key过期或限流

| |

向量库查询QPS

|

接近上限

|

需要扩容

|

监控实现

<span><span leaf="">import</span></span><span leaf="">&nbsp;time</span><br><span><span leaf="">import</span></span><span leaf="">&nbsp;logging</span><br><span><span leaf="">from</span></span><span leaf="">&nbsp;functools&nbsp;</span><span><span leaf="">import</span></span><span leaf="">&nbsp;wraps</span><br><br><span leaf="">logger = logging.getLogger(</span><span><span leaf="">"rag_monitor"</span></span><span leaf="">)</span><br><br><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">monitor_rag</span></span><span><span leaf="">(func)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""RAG调用监控装饰器"""</span></span><br><span><span leaf="">&nbsp; &nbsp; @wraps(func)</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">wrapper</span></span><span><span leaf="">(*args, **kwargs)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; start = time.time()</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">try</span></span><span leaf="">:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; result = func(*args, **kwargs)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; duration = time.time() - start</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 记录指标</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; logger.info(json.dumps({</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"function"</span></span><span leaf="">: func.__name__,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"duration_ms"</span></span><span leaf="">: round(duration *&nbsp;</span><span><span leaf="">1000</span></span><span leaf="">),</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"status"</span></span><span leaf="">:&nbsp;</span><span><span leaf="">"success"</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"num_results"</span></span><span leaf="">: len(result)&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;isinstance(result, list)&nbsp;</span><span><span leaf="">else</span></span><span leaf="">&nbsp;</span><span><span leaf="">1</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }))</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf=""># 延迟告警</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;duration &gt;&nbsp;</span><span><span leaf="">2.0</span></span><span leaf="">:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; logger.warning(</span><span><span leaf="">f"⚠️&nbsp;</span><span><span leaf="">{func.__name__}</span></span><span leaf="">&nbsp;耗时</span><span><span leaf="">{duration:</span><span><span leaf="">.1</span></span><span leaf="">f}</span></span><span leaf="">s,超过2秒阈值"</span></span><span leaf="">)</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;result</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">except</span></span><span leaf="">&nbsp;Exception&nbsp;</span><span><span leaf="">as</span></span><span leaf="">&nbsp;e:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; duration = time.time() - start</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; logger.error(json.dumps({</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"function"</span></span><span leaf="">: func.__name__,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"duration_ms"</span></span><span leaf="">: round(duration *&nbsp;</span><span><span leaf="">1000</span></span><span leaf="">),</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"status"</span></span><span leaf="">:&nbsp;</span><span><span leaf="">"error"</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">"error"</span></span><span leaf="">: str(e),</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }))</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">raise</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;wrapper</span><br><br><span><span leaf=""># 使用</span></span><br><span><span leaf="">@monitor_rag</span></span><br><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">search_knowledge_base</span></span><span><span leaf="">(query: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;hybrid_retriever.invoke(query)</span><br>

成本控制:别让Token烧光你的预算

RAG的成本主要来自三块:Embedding调用、LLM生成、向量库存储。

成本优化策略

1. Embedding缓存

相同文本不重复调Embedding API:

<span><span leaf="">from</span></span><span leaf="">&nbsp;functools&nbsp;</span><span><span leaf="">import</span></span><span leaf="">&nbsp;lru_cache</span><br><br><span><span leaf="">@lru_cache(maxsize=10000)</span></span><br><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">get_embedding</span></span><span><span leaf="">(text: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;embedding_model.embed_query(text)</span><br>

2. 上下文压缩

检索到的文档可能很长,压缩后再送模型:

<span><span leaf="">from</span></span><span leaf="">&nbsp;langchain.retrievers&nbsp;</span><span><span leaf="">import</span></span><span leaf="">&nbsp;ContextualCompressionRetriever</span><br><span><span leaf="">from</span></span><span leaf="">&nbsp;langchain_community.document_compressors&nbsp;</span><span><span leaf="">import</span></span><span leaf="">&nbsp;LLMChainExtractor</span><br><br><span leaf="">compressor = LLMChainExtractor.from_llm(llm=cheap_model)</span><br><span leaf="">compression_retriever = ContextualCompressionRetriever(</span><br><span leaf="">&nbsp; &nbsp; base_compressor=compressor,</span><br><span leaf="">&nbsp; &nbsp; base_retriever=retriever,</span><br><span leaf="">)</span><br>

3. 模型分级

不是所有问题都需要GPT-4o:

<span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">get_model_for_query</span></span><span><span leaf="">(query: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">"""根据问题复杂度选择模型"""</span></span><br><span leaf="">&nbsp; &nbsp; simple_keywords = [</span><span><span leaf="">"是什么"</span></span><span leaf="">,&nbsp;</span><span><span leaf="">"多少"</span></span><span leaf="">,&nbsp;</span><span><span leaf="">"什么时候"</span></span><span leaf="">,&nbsp;</span><span><span leaf="">"在哪里"</span></span><span leaf="">]</span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;any(kw&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;query&nbsp;</span><span><span leaf="">for</span></span><span leaf="">&nbsp;kw&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;simple_keywords):</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;</span><span><span leaf="">"gpt-4o-mini"</span></span><span leaf="">&nbsp;&nbsp;</span><span><span leaf=""># 简单问题用便宜模型</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;</span><span><span leaf="">"gpt-4o"</span></span><span leaf="">&nbsp;&nbsp;</span><span><span leaf=""># 复杂问题用好模型</span></span><br>

4. 检索结果数量控制

Top-K不是越大越好——5条够用的别拉10条,每多一条就多几百Token:

<span><span leaf=""># 根据问题类型调整K值</span></span><br><span><span><span leaf="">def</span></span><span leaf="">&nbsp;</span><span><span leaf="">get_k_for_query</span></span><span><span leaf="">(query: str)</span></span><span leaf="">:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">if</span></span><span leaf="">&nbsp;</span><span><span leaf="">"对比"</span></span><span leaf="">&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;query&nbsp;</span><span><span leaf="">or</span></span><span leaf="">&nbsp;</span><span><span leaf="">"分析"</span></span><span leaf="">&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;query:</span><br><span leaf="">&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;</span><span><span leaf="">8</span></span><span leaf="">&nbsp;&nbsp;</span><span><span leaf=""># 需要多条文档对比</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">return</span></span><span leaf="">&nbsp;</span><span><span leaf="">4</span></span><span leaf="">&nbsp;&nbsp;</span><span><span leaf=""># 简单问答4条足够</span></span><br>

多租户架构:不同客户的数据隔离

如果你的RAG服务SaaS化,多个客户共用一套系统,数据隔离是刚需。

方案1:命名空间隔离

在同一个向量库中用元数据字段区分不同租户:

<span><span leaf=""># 入库时标记租户</span></span><br><span leaf="">chunks = splitter.split_documents(doc)</span><br><span><span leaf="">for</span></span><span leaf="">&nbsp;chunk&nbsp;</span><span><span leaf="">in</span></span><span leaf="">&nbsp;chunks:</span><br><span leaf="">&nbsp; &nbsp; chunk.metadata[</span><span><span leaf="">"tenant_id"</span></span><span leaf="">] =&nbsp;</span><span><span leaf="">"client_001"</span></span><br><span leaf="">vectorstore.add_documents(chunks)</span><br><br><span><span leaf=""># 检索时过滤</span></span><br><span leaf="">results = vectorstore.similarity_search(</span><br><span leaf="">&nbsp; &nbsp; query=</span><span><span leaf="">"年假政策"</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; k=</span><span><span leaf="">5</span></span><span leaf="">,</span><br><span leaf="">&nbsp; &nbsp; filter={</span><span><span leaf="">"tenant_id"</span></span><span leaf="">:&nbsp;</span><span><span leaf="">"client_001"</span></span><span leaf="">}, &nbsp;</span><span><span leaf=""># 只搜这个客户的数据</span></span><br><span leaf="">)</span><br>

Chroma、Milvus、Qdrant都支持元数据过滤。

方案2:独立Collection/索引

每个租户一个独立的向量库:

<span><span leaf=""># 租户A的向量库</span></span><br><span leaf="">vectorstore_a = Milvus(collection_name=</span><span><span leaf="">"client_001"</span></span><span leaf="">, ...)</span><br><br><span><span leaf=""># 租户B的向量库</span></span><br><span leaf="">vectorstore_b = Milvus(collection_name=</span><span><span leaf="">"client_002"</span></span><span leaf="">, ...)</span><br>

隔离更彻底但运维成本更高,适合对数据安全要求极高的客户。

部署方案

方案1:Docker Compose(小团队推荐)

<span><span leaf=""># docker-compose.yml</span></span><br><span><span leaf="">version:</span></span><span leaf="">&nbsp;</span><span><span leaf="">'3.8'</span></span><br><span><span leaf="">services:</span></span><br><span leaf="">&nbsp;&nbsp;</span><span><span leaf="">rag-api:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">build:</span></span><span leaf="">&nbsp;</span><span><span leaf="">.</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">ports:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">"8000:8000"</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">environment:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">OPENAI_API_KEY=${OPENAI_API_KEY}</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">MILVUS_HOST=milvus</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">depends_on:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">milvus</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">redis</span></span><br><br><span leaf="">&nbsp;&nbsp;</span><span><span leaf="">milvus:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">image:</span></span><span leaf="">&nbsp;</span><span><span leaf="">milvusdb/milvus:v2.4-latest</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">ports:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">"19530:19530"</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">volumes:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">milvus_data:/var/lib/milvus</span></span><br><br><span leaf="">&nbsp;&nbsp;</span><span><span leaf="">redis:</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">image:</span></span><span leaf="">&nbsp;</span><span><span leaf="">redis:7-alpine</span></span><br><span leaf="">&nbsp; &nbsp;&nbsp;</span><span><span leaf="">ports:</span></span><br><span leaf="">&nbsp; &nbsp; &nbsp;&nbsp;</span><span><span leaf="">-</span></span><span leaf="">&nbsp;</span><span><span leaf="">"6379:6379"</span></span><br><br><span><span leaf="">volumes:</span></span><br><span leaf="">&nbsp;&nbsp;</span><span><span leaf="">milvus_data:</span></span><br>

方案2:Kubernetes(大团队推荐)

适合需要自动扩缩容、多实例部署的场景。关键组件:

  • RAG API服务:Deployment + HPA自动扩缩容

  • Milvus集群:分布式部署,读写分离

  • Redis:缓存层

  • Prometheus + Grafana:监控

方案3:Serverless(低频场景)

用云函数+托管向量库,按调用计费:

  • API层:AWS Lambda / 阿里云函数计算

  • 向量库:Pinecone / Zilliz Cloud(托管Milvus)

  • LLM:API调用

适合调用量不大(每天<1万次)、不需要7x24运行的场景。

本篇要点

| 要点

|

说明

分层架构

|

接入/编排/检索/生成/存储/数据管道六层

| |

增量索引

|

文档哈希做差量检测,别全量重建

| |

监控告警

|

检索P95<500ms,生成P95<5s,空结果率<10%

| |

成本控制

|

Embedding缓存+上下文压缩+模型分级+K值控制

| |

多租户

|

命名空间隔离(轻量)或独立Collection(彻底)

| |

部署方案

|

Docker Compose(小团队)/K8s(大团队)/Serverless(低频)

| |

上线Checklist

|

12项,逐条检查再发版

|

RAG生产上线Checklist

上线前逐条检查:

  • 增量索引机制已实现,文档更新不用重建

  • 检索延迟P95 < 500ms,生成延迟P95 < 5s

  • 监控告警已配好:延迟、错误率、空结果率

  • Embedding API有降级方案(主模型限流时切备用)

  • 上下文长度控制:SummarizationMiddleware或截断中间件

  • 多租户数据隔离已验证

  • RAGAS评估流水线已搭建,基线分数已记录

  • 文档注册表已实现,可追踪每个chunk的来源

  • 缓存层已配好,重复查询不重复调模型

  • API Key从环境变量读取,未硬编码

  • 错误处理完善:检索失败有兜底、模型超时有重试

  • 日志记录完整:每次查询的query、检索结果、生成答案、耗时

前7篇回顾

7篇基础篇,从"RAG到底在干什么"到"生产部署Checklist",RAG的核心流程走了一遍:

  1. 基础篇(第1篇):RAG原理、三代架构演进、与微调和长上下文的对比

  2. 数据处理篇(第2篇):6种切分策略、PDF解析、chunk_size调参

  3. 向量化篇(第3篇):Embedding模型选型、向量数据库对比

  4. 检索优化篇(第4篇):混合检索、Rerank重排序、上下文压缩

  5. 高级架构篇(第5篇):查询改写、多跳检索、GraphRAG、Agentic RAG

  6. 评估篇(第6篇):RAGAS指标、黄金测试集、A/B测试、Bad Case分析

  7. 部署篇(第7篇):增量索引、监控告警、成本控制、多租户

学RAG最有效的方式是拿真实业务数据做一遍。从最基础的Naive RAG开始,遇到问题再针对性升级——先加混合检索,再加Rerank,再考虑查询改写。别一上来就GraphRAG + Agentic RAG,99%的项目用不着。

后面还有3篇前沿专题:GraphRAG知识图谱检索、多模态RAG、Agentic RAG实战。基础打牢之后再看这些,你会理解得更深。


系列如果对你有帮助,转发给同样在做RAG的朋友。你们的支持是我持续输出的动力。

你的RAG上线了吗?遇到过什么生产事故?评论区聊聊。

觉得有用就点个在看,下一篇讲GraphRAG——跨文档推理的终极方案。