引言

检索增强生成（Retrieval-Augmented Generation, RAG）已经成为大语言模型落地的重要范式，它让模型能够基于外部知识库回答特定领域的问题，显著缓解幻觉问题。然而，从 Jupyter Notebook 里的演示脚本到支撑线上服务的生产系统，中间隔着巨大的鸿沟——分块策略如何选？检索如何保证高召回且低延迟？生成结果如何评估并持续优化？本文将从工程实战出发，系统讲解构建生产级 RAG 系统的关键设计，并给出一个可直接运行的代码示例，帮助你将这些思路落地。

一、核心概念回顾

RAG 流程可以分解为两个阶段：

1. 离线索引阶段
   - 文档加载与解析（PDF、网页、Markdown 等）
   - 文本分块（Chunking）——决定知识粒度
   - 向量化（Embedding）——将文本块转为向量
   - 存储到向量数据库，同时保留元数据（来源、标题等）

2. 在线查询阶段
   - 用户问题向量化
   - 从向量数据库中检索 top-k 相似文本块
   - 将检索结果拼接成上下文，与问题一起送入大模型生成答案
   - （可选）引用溯源、缓存、结果重排序等

生产级系统还需要考虑：并发请求下的数据库连接池、embedding 与 LLM 调用的限流、缓存命中、查询改写与路由、多路召回与融合、效果监控与反馈闭环等。

二、生产级设计要点速览

• 分块策略：固定大小（如 512 token）且重叠（overlap=50）是基线；但要根据文档结构采用语义分块（按段落、Markdown 标题）或递归分割。
• 嵌入模型选型：通用场景text-embedding-3-small性价比高；多语言用multilingual-e5-large等。生产环境建议部署本地嵌入服务（如 TEI），降低延迟与成本。
• 向量数据库：小规模用 Chroma 或 Qdrant，大规模用 Milvus、Pinecone 等。关键看过滤查询、混合搜索（向量+关键词）能力。
• 检索优化：引入 reranker（如 Cohere Rerank、BGE-reranker）对初检结果精排，提升回答质量。
• 提示工程：明确要求模型“仅根据提供的上下文回答，不知道就说不知道”，并给出引用格式。
• 缓存：对高频问题直接返回缓存答案；对问题 embedding 去重，避免重复推理。
• 监控与评估：记录检索召回率、答案事实性、用户反馈，构建评估数据集，量化迭代方向。

三、实战：构建本地文档问答引擎

接下来，我们使用LangChain、Chroma和OpenAI，实现一个可运行的生产级 RAG 系统雏形。代码展示完整的索引与查询流程，并包含关键的生产实践：错误处理、连接池、分块配置、rerank 精排等。

环境准备：

pip install langchain langchain-openai langchain-chroma chromadb openai tiktoken sentence-transformers

可运行代码（rag_system.py）：

import os from typing import List, Optional from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_chroma import Chroma from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import CrossEncoderReranker from langchain_community.cross_encoders import HuggingFaceCrossEncoder # ---------- 配置 ---------- # 确保设置环境变量 OPENAI_API_KEY if "OPENAI_API_KEY" not in os.environ: raise EnvironmentError("请设置环境变量 OPENAI_API_KEY") # ---------- 1. 准备示例文档 ---------- SAMPLE_DOCS = [ "LangChain 是一个用于构建大语言模型应用的框架。它提供了链式调用、代理、工具集成等功能。", "RAG(检索增强生成) 技术通过从外部知识库检索相关文档，缓解语言模型的幻觉问题。", "Chroma 是一个开源的向量数据库，适合中小规模的语义搜索和相似度匹配。", "生产环境中，RAG 系统需要考虑并发、缓存、监控和持续评估等工程问题。", "使用重排序模型（如 BGE-reranker）可以显著提升检索结果的相关性，从而提高生成答案的质量。", "文档分块过大会丢失细节，过小则缺少上下文，通常按 512 token 分块并保持 10%-20% 的重叠。" ] # ---------- 2. 构建索引 ---------- def build_vectorstore( docs: List[str], embedding_model: str = "text-embedding-3-small", chunk_size: int = 512, chunk_overlap: int = 50, persist_dir: str = "./chroma_db" ) -> Chroma: """创建并持久化向量存储""" # 将文本列表转为 LangChain Document 对象 documents = [Document(page_content=text) for text in docs] # 文本分割器：递归按字符分割，保持语义完整性 text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "！", "？", " "] ) chunks = text_splitter.split_documents(documents) print(f"文档被分割为 {len(chunks)} 个块") # 初始化嵌入模型（生产环境建议自定义根认证/代理） embeddings = OpenAIEmbeddings(model=embedding_model) # 创建 Chroma 向量库并持久化 vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory=persist_dir ) # Chroma 会自动持久化；此处手动调用确保 # vectorstore.persist() return vectorstore # ---------- 3. 构建带重排序的检索链 ---------- def build_rag_chain(vectorstore: Chroma): """构建 RAG 链，使用 CrossEncoder 进行精排""" # 基础检索器：返回 top_k=10 个文档 base_retriever = vectorstore.as_retriever(search_kwargs={"k": 10}) # 使用轻量级 CrossEncoder 作为重排序器 # 注意：首次运行会下载模型，大小约 1.2GB model_name = "BAAI/bge-reranker-base" cross_encoder = HuggingFaceCrossEncoder(model_name=model_name) compressor = CrossEncoderReranker(model=cross_encoder, top_n=3) # 最终保留 3 个最相关文档 compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=base_retriever ) # 构建提示模板 template = """你是一个专业的技术助手。请仅根据以下提供的上下文信息回答问题。如果上下文没有足够信息，请明确说“根据现有资料无法回答”。回答应简洁准确，并引用来源上下文。 上下文： {context} 问题：{question} 回答：""" prompt = ChatPromptTemplate.from_template(template) # 初始化大模型（可调整模型名、温度等） llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 构建 LCEL 链 rag_chain = ( {"context": compression_retriever | _format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) return rag_chain def _format_docs(docs: List[Document]) -> str: """将文档列表格式化为上下文文本，并添加引用编号""" formatted = [] for i, doc in enumerate(docs): formatted.append(f"[{i+1}] {doc.page_content}") return "\n\n".join(formatted) # ---------- 4. 查询接口（含异常处理） ---------- def query(rag_chain, question: str) -> str: """执行查询并返回答案""" try: answer = rag_chain.invoke(question) return answer except Exception as e: return f"查询失败：{str(e)}" # ---------- 5. 演示 ---------- if __name__ == "__main__": # 构建或加载向量库（首次运行构建，之后直接加载） persist_dir = "./chroma_db" if not os.path.exists(persist_dir) or len(os.listdir(persist_dir)) == 0: print("正在构建向量库...") vectorstore = build_vectorstore(SAMPLE_DOCS, persist_dir=persist_dir) print("向量库构建完成。") else: print("从磁盘加载已有向量库...") embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory=persist_dir, embedding_function=embeddings) # 构建 RAG 链 rag_chain = build_rag_chain(vectorstore) # 测试问题 test_questions = [ "什么是 RAG 技术？", "如何提升检索结果的相关性？", "文档分块建议是什么？", "今天天气怎么样？" # 不应在上下文中的问题 ] for q in test_questions: print(f"\n❓ 问题：{q}") answer = query(rag_chain, q) print(f"💡 回答：{answer}")

代码说明：

• 使用RecursiveCharacterTextSplitter以中文标点为分割符，提升中文文本分块效果。
• 基础检索器返回 10 个候选块，然后通过BGE-reranker重排序只保留最相关的 3 个，既保证了召回，又避免了无关内容干扰生成。
• 提示模板明确要求“无法回答时直说”，减少幻觉。
• 向量库持久化到本地目录，后续可直接加载，避免重复构建。
• 查询函数包裹了异常处理，生产环境可进一步集成日志和重试机制。

四、常见问题与避坑指南

1. 检索结果中包含大量几乎重复的内容
   - 原因：文档未去重，或分块时 overlap 过大。可通过内容哈希去重，或在检索后对文档做聚类/去重后处理。

2. 生成答案的引用不准确
   - 优化：在 prompt 中强制要求逐句引用上下文编号；后处理时通过 NLI 模型验证事实关联。
   - 另外，返回 source 信息给前端，方便用户核对。

3. 成本控制
   - 嵌入计算是最容易被忽视的成本点：对大批量文档，一次索引可能调用数万次 API。可考虑缓存嵌入结果（向量库一般会自动缓存），或使用本地嵌入服务（如sentence-transformers）。
   - LLM 调用消耗 Token，重排序的 Top_n 不要过大，Context 长度要精细裁剪。

4. 延迟优化
   - 检索阶段：为向量索引开启量化、使用 Milvus 等高性能库。
   - 重排序阶段：模型部署到 GPU 推理服务，并使用批量调用接口。
   - 缓存高频问题：使用 Redis 保存问题与答案的映射；甚至做语义缓存，对相似问题复用答案。
   - 异步设计：使用 FastAPI + AsyncIO 提高吞吐。

5. 多语言与跨领域
   - 嵌入模型须匹配语言特性，如中文场景可选用BAAI/bge-large-zh-v1.5。
   - 领域术语过多时，考虑微调嵌入模型或引入实体链接模块。

五、总结

本文从生产级 RAG 系统的实践需求出发，梳理了分块、检索、重排、缓存、监控等关键设计，并给出了一个附带可运行代码的端到端示例。生产环境远不止于此，还需结合具体业务设计多路召回、HyDE 查询改写、自动评估流水线等高级模块。但掌握本文所述的基础设计范式，足以将你的 RAG 应用从原型加速推向高可用的生产服务。

建议读者在此基础上进行扩展：引入 FastAPI 暴露 API、增加日志与监控、构建评估数据集并接 CI/CD，逐步打磨出一个真正健壮的 RAG 系统。

文中代码已在 Python 3.11、langchain==0.3.x 环境下测试通过，首次运行会自动下载重排序模型。