引言检索增强生成Retrieval-Augmented Generation, RAG已经成为大语言模型落地的重要范式它让模型能够基于外部知识库回答特定领域的问题显著缓解幻觉问题。然而从 Jupyter Notebook 里的演示脚本到支撑线上服务的生产系统中间隔着巨大的鸿沟——分块策略如何选检索如何保证高召回且低延迟生成结果如何评估并持续优化本文将从工程实战出发系统讲解构建生产级 RAG 系统的关键设计并给出一个可直接运行的代码示例帮助你将这些思路落地。一、核心概念回顾RAG 流程可以分解为两个阶段离线索引阶段- 文档加载与解析PDF、网页、Markdown 等- 文本分块Chunking——决定知识粒度- 向量化Embedding——将文本块转为向量- 存储到向量数据库同时保留元数据来源、标题等在线查询阶段- 用户问题向量化- 从向量数据库中检索 top-k 相似文本块- 将检索结果拼接成上下文与问题一起送入大模型生成答案- 可选引用溯源、缓存、结果重排序等生产级系统还需要考虑并发请求下的数据库连接池、embedding 与 LLM 调用的限流、缓存命中、查询改写与路由、多路召回与融合、效果监控与反馈闭环等。二、生产级设计要点速览分块策略固定大小如 512 token且重叠overlap50是基线但要根据文档结构采用语义分块按段落、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.pyimport 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_contenttext) for text in docs] # 文本分割器递归按字符分割保持语义完整性 text_splitter RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , ] ) chunks text_splitter.split_documents(documents) print(f文档被分割为 {len(chunks)} 个块) # 初始化嵌入模型生产环境建议自定义根认证/代理 embeddings OpenAIEmbeddings(modelembedding_model) # 创建 Chroma 向量库并持久化 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directorypersist_dir ) # Chroma 会自动持久化此处手动调用确保 # vectorstore.persist() return vectorstore # ---------- 3. 构建带重排序的检索链 ---------- def build_rag_chain(vectorstore: Chroma): 构建 RAG 链使用 CrossEncoder 进行精排 # 基础检索器返回 top_k10 个文档 base_retriever vectorstore.as_retriever(search_kwargs{k: 10}) # 使用轻量级 CrossEncoder 作为重排序器 # 注意首次运行会下载模型大小约 1.2GB model_name BAAI/bge-reranker-base cross_encoder HuggingFaceCrossEncoder(model_namemodel_name) compressor CrossEncoderReranker(modelcross_encoder, top_n3) # 最终保留 3 个最相关文档 compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) # 构建提示模板 template 你是一个专业的技术助手。请仅根据以下提供的上下文信息回答问题。如果上下文没有足够信息请明确说“根据现有资料无法回答”。回答应简洁准确并引用来源上下文。 上下文 {context} 问题{question} 回答 prompt ChatPromptTemplate.from_template(template) # 初始化大模型可调整模型名、温度等 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 构建 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[{i1}] {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_dirpersist_dir) print(向量库构建完成。) else: print(从磁盘加载已有向量库...) embeddings OpenAIEmbeddings() vectorstore Chroma(persist_directorypersist_dir, embedding_functionembeddings) # 构建 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 个既保证了召回又避免了无关内容干扰生成。提示模板明确要求“无法回答时直说”减少幻觉。向量库持久化到本地目录后续可直接加载避免重复构建。查询函数包裹了异常处理生产环境可进一步集成日志和重试机制。四、常见问题与避坑指南检索结果中包含大量几乎重复的内容- 原因文档未去重或分块时 overlap 过大。可通过内容哈希去重或在检索后对文档做聚类/去重后处理。生成答案的引用不准确- 优化在 prompt 中强制要求逐句引用上下文编号后处理时通过 NLI 模型验证事实关联。- 另外返回 source 信息给前端方便用户核对。成本控制- 嵌入计算是最容易被忽视的成本点对大批量文档一次索引可能调用数万次 API。可考虑缓存嵌入结果向量库一般会自动缓存或使用本地嵌入服务如sentence-transformers。- LLM 调用消耗 Token重排序的 Top_n 不要过大Context 长度要精细裁剪。延迟优化- 检索阶段为向量索引开启量化、使用 Milvus 等高性能库。- 重排序阶段模型部署到 GPU 推理服务并使用批量调用接口。- 缓存高频问题使用 Redis 保存问题与答案的映射甚至做语义缓存对相似问题复用答案。- 异步设计使用 FastAPI AsyncIO 提高吞吐。多语言与跨领域- 嵌入模型须匹配语言特性如中文场景可选用BAAI/bge-large-zh-v1.5。- 领域术语过多时考虑微调嵌入模型或引入实体链接模块。五、总结本文从生产级 RAG 系统的实践需求出发梳理了分块、检索、重排、缓存、监控等关键设计并给出了一个附带可运行代码的端到端示例。生产环境远不止于此还需结合具体业务设计多路召回、HyDE 查询改写、自动评估流水线等高级模块。但掌握本文所述的基础设计范式足以将你的 RAG 应用从原型加速推向高可用的生产服务。建议读者在此基础上进行扩展引入 FastAPI 暴露 API、增加日志与监控、构建评估数据集并接 CI/CD逐步打磨出一个真正健壮的 RAG 系统。文中代码已在 Python 3.11、langchain0.3.x 环境下测试通过首次运行会自动下载重排序模型。