专栏第12篇：前面三篇文章分别讲了RAG的离线阶段（文档处理）、在线阶段（混合检索与重排序）和评估优化。但在真实的企业环境中，RAG不是一段Jupyter Notebook脚本，而是一个需要处理并发请求、支持知识库热更新、保证低延迟、还要能流式输出的完整系统。这篇文章把我搞建企业级RAG系统的架构设计、核心模块和踩坑经验分享出来。

目录

一、从Demo到生产：RAG系统面临的真实挑战

很多教程里的RAG是这样的：加载几个PDF，建个向量库，用户提问时检索一下，把结果塞给LLM。这在笔记本上跑通没问题，但放到生产环境会暴露出各种问题。

维度Demo级RAG企业级RAG
并发单用户串行多用户同时请求
知识更新手动重新建索引自动热更新、不停机
延迟几秒钟可接受首字延迟要控制在1秒内
输出一次性返回流式输出、实时渲染
容错出错就报错优雅降级、超时保护
来源追溯可有可无必须准确、可点击

二、系统架构全景

知识库更新层

生成层

检索层

查询层

用户请求

查询重写

意图路由

Dense向量检索

BM25关键词检索

RRF融合排序

上下文组装

LLM生成

SSE流式输出

文档变更

MD5指纹检测

增量索引

原子切换

系统分为四层：

  • 查询层：把用户问题转化为更适合检索的形式，并判断意图
  • 检索层：Dense向量 + BM25双路召回，RRF融合排序
  • 生成层：上下文组装、LLM生成、流式输出
  • 知识库更新层：监控文档变化、增量更新索引、原子切换

三、查询层：不只是"把问题丢给检索器"

用户的问题往往不能直接用于检索，需要经过两层处理。

3.1 查询重写（Query Rewriting）

用户输入可能口语化、有歧义、或者缺少上下文。查询重写用LLM把问题转化成更利于检索的形式。

def rewrite_query(original_query: str, chat_history: list) -> str:
    """
    把用户的口语化问题重写为适合向量检索的查询
    """
    prompt = f"""基于对话历史，把用户的问题改写成适合向量检索的查询。
要求：
1. 补充指代和省略的上下文
2. 去除口语化表达
3. 保留关键技术术语

对话历史：{chat_history}
用户问题：{original_query}
"""
    return llm.invoke(prompt)

示例：

  • 原问题：“那个怎么配？”
  • 重写后：“Python虚拟环境的配置步骤是什么？”

3.2 意图路由（Intent Router）

不是所有问题都需要走RAG检索。系统需要判断用户的意图，选择不同的处理路径：

意图类型处理方式示例
知识查询走RAG检索 + LLM生成“怎么配置环境变量？”
工具调用调用外部API/工具“帮我查一下今天的天气”
闲聊/问候直接回复，不走检索“你好”
class IntentRouter:
    """意图路由器：决定用户请求走哪条处理通道"""
    
    ROUTES = {
        "knowledge": "知识库查询",
        "tool": "工具调用", 
        "chat": "直接对话"
    }
    
    def route(self, query: str, chat_history: list) -> str:
        prompt = f"""判断用户意图，只能输出以下之一：knowledge/tool/chat
        
用户问题：{query}
"""
        intent = llm.invoke(prompt).strip().lower()
        
        # 兜底：如果模型输出异常，默认走知识库
        if intent not in self.ROUTES:
            intent = "knowledge"
            
        return intent

⚠️ 重要：路由失败必须有兜底策略。如果LLM判断出错或超时，默认降级到知识库查询，而不是报错。

四、检索层：双引擎混合召回的工程实现

检索层的核心是"双路召回 + RRF融合"，这在第10篇已经详细讲过原理。这里重点讲工程实现层面的优化。

4.1 检索流程

class RetrievalEngine:
    """检索引擎：Dense向量 + BM25 双路召回"""
    
    def __init__(self, vector_store, bm25_index):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        
    def retrieve(self, query: str, top_k: int = 6) -> list:
        # 1. 向量检索
        vector_results = self.vector_store.similarity_search(query, k=top_k)
        
        # 2. BM25检索
        bm25_results = self.bm25_index.search(query, top_k=top_k)
        
        # 3. RRF融合
        fused_results = reciprocal_rank_fusion(
            [vector_results, bm25_results],
            k=60
        )
        
        return fused_results[:top_k]

4.2 工程优化：并发检索

向量检索和BM25检索互不依赖，可以并行执行，降低整体延迟：

import asyncio

async def retrieve_parallel(self, query: str, top_k: int = 6):
    # 两个检索任务并行执行
    vector_task = asyncio.create_task(
        self.vector_store.asimilarity_search(query, k=top_k)
    )
    bm25_task = asyncio.create_task(
        self.bm25_index.asearch(query, top_k=top_k)
    )
    
    vector_results, bm25_results = await asyncio.gather(
        vector_task, bm25_task
    )
    
    return reciprocal_rank_fusion(
        [vector_results, bm25_results], k=60
    )[:top_k]

4.3 工程优化：检索缓存

高频查询可以缓存结果，避免重复计算Embedding：

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_retrieve(query_hash: str):
    """用查询文本的hash作为缓存key"""
    return retrieval_engine.retrieve(query_hash)

4.4 检索结果后处理

融合排序后，还需要过滤低质量结果：

def post_filter(results: list, min_score: float = 0.01) -> list:
    """过滤RRF分数过低的文档"""
    return [r for r in results if r.score >= min_score]

五、生成层：检索与生成的职责分离

这是架构设计中最关键的一个决策：把检索和生成分成两个独立模块。

5.1 为什么要分离？

方面耦合在一起分离后
优化维度互相牵制检索优化不影响生成，生成优化不影响检索
测试难以单独测试可以独立测试检索召回率和生成质量
迭代改一处可能影响另一处模块独立迭代
复用检索逻辑和生成逻辑绑死检索模块可被其他场景复用

5.2 模块职责划分

+---------------------+     +---------------------+
|     检索模块        |     |     生成模块        |
|     (rag_core)      | --> |  (llm_generator)    |
+---------------------+     +---------------------+
| - 查询重写          |     | - 上下文组装        |
| - 多路检索          |     | - 提示词构建        |
| - RRF融合           |     | - LLM调用           |
| - 结果过滤          |     | - 流式输出          |
+---------------------+     +---------------------+

5.3 上下文组装

检索模块返回文档列表，生成模块负责把文档组装成LLM可用的上下文：

def build_context(query: str, documents: list, max_tokens: int = 3000) -> str:
    """
    把检索结果组装成上下文，控制总长度
    """
    context_parts = []
    total_length = 0
    
    for doc in documents:
        # 估算token数（粗略：1个汉字约1个token）
        doc_text = f"[来源：{doc.metadata.get('source', '未知')}]\n{doc.page_content}\n\n"
        doc_length = len(doc_text)
        
        if total_length + doc_length > max_tokens:
            break
            
        context_parts.append(doc_text)
        total_length += doc_length
    
    context = "".join(context_parts)
    
    return f"""基于以下参考资料回答问题：

{context}

用户问题：{query}

请基于参考资料回答，如果资料中没有相关信息，请明确说明。"""

5.4 来源追溯

生成回答时，必须保留文档来源信息，让用户知道答案来自哪里：

def generate_with_sources(query: str, documents: list):
    # 给每篇文档编号
    for i, doc in enumerate(documents, 1):
        doc.metadata["cite_id"] = f"[{i}]"
    
    # 构建带编号的上下文
    context = build_context(query, documents)
    
    # LLM生成（流式）
    for chunk in llm.stream(context):
        yield chunk
    
    # 返回来源列表
    sources = [
        {
            "cite_id": doc.metadata["cite_id"],
            "source": doc.metadata.get("source", "未知"),
            "score": doc.metadata.get("score", 0)
        }
        for doc in documents
    ]
    yield {"type": "sources", "data": sources}

六、流式输出：SSE与用户体验

企业级RAG不能等LLM全部生成完再返回——用户看到"加载中"超过3秒就会焦虑。流式输出（Streaming）是必选项。

6.1 SSE流式响应

SSE（Server-Sent Events）是一种服务器向客户端推送数据的技术，天然适合流式场景：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/chat")
async def chat(request: ChatRequest):
    async def event_stream():
        # 1. 查询重写
        rewritten_query = rewrite_query(request.query, request.history)
        
        # 2. 检索
        documents = await retrieval_engine.retrieve_parallel(rewritten_query)
        
        # 3. 生成（流式）
        async for chunk in llm.astream(
            build_context(request.query, documents)
        ):
            # SSE格式：data: {...}\n\n
            yield f"data: {json.dumps({'type': 'content', 'text': chunk})}\n\n"
        
        # 4. 来源信息
        sources = extract_sources(documents)
        yield f"data: {json.dumps({'type': 'sources', 'data': sources})}\n\n"
        
        # 5. 结束标记
        yield f"data: {json.dumps({'type': 'done'})}\n\n"
    
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream"
    )

6.2 前端逐帧渲染

前端收到SSE流后，需要逐帧渲染，而不是等全部内容：

const eventSource = new EventSource('/chat');

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    
    if (data.type === 'content') {
        // 逐字追加到回答区域
        appendText(data.text);
    } else if (data.type === 'sources') {
        // 渲染来源卡片
        renderSourceCards(data.data);
    } else if (data.type === 'done') {
        eventSource.close();
    }
};

6.3 图片即时渲染

如果知识库中有图片（如架构图、流程图），流式输出时需要特殊处理：

# 检测到Markdown图片标记时，发送特殊事件
if contains_image_markdown(chunk):
    yield f"data: {json.dumps({'type': 'image', 'url': extract_image_url(chunk)})}\n\n"

前端收到 type: "image" 事件时，直接渲染成 <img> 标签，而不是纯文本。

七、知识库热更新：不停机更新

企业知识库是动态变化的——文档会新增、修改、删除。系统必须支持不停机热更新。

7.1 更新检测：MD5指纹

为每篇文档计算MD5指纹，对比前后状态判断是否需要更新：

import hashlib

def compute_doc_fingerprint(doc_path: str) -> str:
    """计算文档内容的MD5指纹"""
    with open(doc_path, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()

class KnowledgeBaseMonitor:
    """监控知识库文件变化"""
    
    def __init__(self, kb_dir: str):
        self.kb_dir = kb_dir
        self.fingerprints = {}  # 路径 -> MD5
    
    def scan_changes(self) -> dict:
        """扫描变更：新增、修改、删除"""
        current_files = set()
        changes = {"added": [], "modified": [], "deleted": []}
        
        for doc_path in list_documents(self.kb_dir):
            current_files.add(doc_path)
            current_md5 = compute_doc_fingerprint(doc_path)
            
            if doc_path not in self.fingerprints:
                changes["added"].append(doc_path)
            elif self.fingerprints[doc_path] != current_md5:
                changes["modified"].append(doc_path)
            
            self.fingerprints[doc_path] = current_md5
        
        # 检测删除
        for old_path in set(self.fingerprints.keys()) - current_files:
            changes["deleted"].append(old_path)
            del self.fingerprints[old_path]
        
        return changes

7.2 增量索引

向量库支持真正的增量更新，BM25 由于内部需要重新计算所有文档的频率统计，不支持多单文档的增量添加，必须全量重建。因此实际策略是：只解决向量库的增量问题，BM25 一律全量重建，但结合原子切换保证重建期间不影响查询：

def incremental_update(changes: dict, vector_store, all_docs: list):
    """向量库增量更新；BM25 必须全量重建"""
    
    # 向量库：支持增量删除 + 增量添加
    for doc_path in changes["deleted"] + changes["modified"]:
        doc_id = get_doc_id(doc_path)
        vector_store.delete(doc_id)
    
    for doc_path in changes["added"] + changes["modified"]:
        docs = load_and_split(doc_path)
        vector_store.add_documents(docs)
    
    # BM25：不支持增量，必须用全量文档重建
    # all_docs 是当前知识库的完整文档列表
    new_bm25_index = BM25Index(all_docs)
    return new_bm25_index  # 调用方通过原子切换更新活跃索引

7.3 并发安全与索引切换

索引重建期间，不能影响正在进行的查询。可以用 asyncio.Lock（普通互斥锁） 保证同时只有一个更新任务在运行，读操作不加锁——靠的是先在锁内构建好新索引，最后一次性替换对象引用。Python GIL 保证引用赋値是原子操作，切换前的查询读旧索引，切换后读新，不会出现中间状态。

class RAGCore:
    def __init__(self):
        self._index = None
        self._bm25_index = None
        self._init_lock = asyncio.Lock()  # 更新锁（互斥）
    
    async def search(self, query: str):
        """查询不加锁，直接读当前引用"""
        index = self._index  # 先拿到引用，即使后续更新也不影响本次查询
        return await index.query(query)
    
    async def incremental_update(self, new_docs: list):
        """更新加互斥锁：避免并发重建"""
        async with self._init_lock:
            # 1. 锁内构建新索引（耗时操作）
            new_bm25 = BM25Index(all_docs)               # BM25 全量重建
            new_index = await build_vector_index(new_docs)  # 向量增量刷新
            
            # 2. 原子替换引用（GIL 保证切换原子性）
            self._bm25_index = new_bm25
            self._index = new_index

💡 关键点：读操作不加锁，重建期间新查询不会被阻塞，读到的是旧索引。如果需要差异化对并发更新的控制（比如限制并发更新次数、排队），可以升级为读写锁（aiorwlock）。

八、踩过的5个坑

坑1：路由失败没有兜底，直接报错

问题：LLM路由判断异常时，系统直接抛错，用户体验极差。

解决：路由必须设置默认降级策略——异常时自动走知识库查询通道。

坑2：BM25索引和向量库数据不一致

问题：增量更新时只更新了向量库，忘了同步更新BM25索引，导致两套索引返回的结果完全不同。

解决：增量更新必须同时更新两套索引，用一个统一变更日志保证一致性。

坑3：SSE流没有结束标记，前端一直转圈

问题：SSE流输出完后没有发送 done 事件，前端不知道何时关闭连接。

解决：SSE流最后必须发送明确的结束标记，前端收到后立即关闭EventSource。

坑4：知识库指纹用文件修改时间，被Git操作误触发

问题：最初用文件的 mtime（修改时间）判断变更，但Git clone/pull会批量修改时间，导致不必要的全量重建。

解决：改用内容MD5指纹，只检测真实内容变化，忽略文件系统时间戳变化。

坑5：上下文长度超限，LLM忽略后面的文档

问题：检索返回6篇文档，每篇都很长，拼接后超过LLM的上下文窗口，导致后面的文档被截断。

解决：上下文组装时严格控制token预算，超长文档做摘要，确保所有检索结果都能进入上下文。

九、总结

搭建一个企业级RAG系统，核心不只是把检索和生成串起来，还要解决工程层面的挑战：

  1. 查询层：查询重写让问题更精准，意图路由让系统更智能，失败时自动兜底
  2. 检索层：Dense + BM25双路并行召回，RRF融合排序，缓存高频查询降低延迟
  3. 生成层：检索和生成职责分离，独立优化、独立迭代，上下文长度可控
  4. 流式输出：SSE实现首字秒出，图片即时渲染提升体验，必须有结束标记
  5. 知识库热更新：MD5指纹检测真实变更，增量索引减少开销，原子切换保证不停机

💡 企业级RAG checklist：

  • 路由有兜底策略，异常时默认走知识库
  • 检索双路并行，单路失败不影响整体
  • 流式输出有明确的开始和结束标记
  • 知识库支持热更新，更新期间服务不中断
  • 上下文长度可控，不超LLM窗口限制

参考资源：

  • FastAPI StreamingResponse 文档
  • SSE（Server-Sent Events）规范
  • RRF（Reciprocal Rank Fusion）论文
# AI
Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

cover

Windows 下部署 New API 中转站全过程实录（含 6 个真实踩坑）

avatar MCP技术社区

Google能索引，AI却抓不到

很多技术团队有个错觉：只要Google能正常收录，内容就没问题。但现实是，你的页面在Google搜索结果里好好的，在ChatGPT、Perplexity、Claude的答案里却像从未存在过一样。

avatar MCP技术社区
cover

英伟达AI刷爆ARC-AGI-3！AVO Harness架构让Agent从30%飙升到100%

avatar MCP技术社区