LLM 能力集成：RAG 检索增强生成的架构设计与工程实践

LLM 能力集成：RAG 检索增强生成的架构设计与工程实践

一、大模型的知识盲区：为什么纯参数记忆不够用

大语言模型通过参数记忆存储了海量知识，但这种记忆有三个结构性缺陷：知识截止日期导致时效性不足、长尾领域知识覆盖不完整、以及无法引用具体来源导致可信度存疑。在企业落地场景中，这三个缺陷往往同时出现——当用户询问"公司最新的报销政策是什么"时，模型既不知道最新政策，也无法给出政策文档的出处。

RAG（Retrieval-Augmented Generation）的核心思路是：与其让模型记住一切，不如在推理时动态检索外部知识库，将检索结果注入上下文，让模型基于事实生成回答。这种"外挂知识库"的方式，既解决了知识时效性问题，又提供了可追溯的引用来源。

sequenceDiagram participant U as 用户 participant Q as 查询处理层 participant R as 检索引擎 participant K as 知识库 participant L as LLM participant P as 后处理层 U->>Q: 原始问题 Q->>Q: 查询改写与扩展 Q->>R: 优化后的查询向量 R->>K: 向量相似度检索 K-->>R: Top-K 候选文档 R->>R: 重排序与去重 R-->>Q: 检索结果 Q->>L: 问题 + 检索上下文 L-->>P: 生成回答 P->>P: 引用标注与事实校验 P-->>U: 带引用的回答

二、RAG 架构的核心组件与数据流

一个生产级 RAG 系统由五个核心组件构成，每个组件都有独立的优化空间。

查询处理层负责将用户的自然语言问题转化为适合检索的查询。直接用原始问题做向量检索往往效果不佳，因为用户提问的表述方式与文档的表述方式存在语义鸿沟。查询改写（Query Rewriting）通过 LLM 将口语化问题转化为更精确的检索查询，查询扩展（Query Expansion）则生成多个相关查询以提升召回率。

检索引擎是 RAG 的性能瓶颈所在。向量检索（ANN）擅长语义匹配，但在精确关键词匹配上表现较弱；关键词检索（BM25）则恰好相反。生产环境中通常采用混合检索策略，将向量检索与关键词检索的结果融合，通过 Reciprocal Rank Fusion（RRF）算法合并排序。

重排序层对检索结果进行精排。粗检索阶段追求召回率，返回的 Top-K 结果中可能包含大量噪声。重排序模型（如 BGE-Reranker、Cohere Rerank）对候选文档与查询的相关性做精细打分，过滤掉低相关性的文档，减少注入 LLM 上下文的噪声。

生成层将检索结果与用户问题组装为 Prompt，调用 LLM 生成回答。Prompt 的组装策略直接影响生成质量——检索结果的顺序、截断策略、以及是否区分"核心证据"与"补充上下文"，都需要精心设计。

后处理层负责引用标注和事实校验。将生成回答中的陈述与检索文档进行对齐，标注来源段落，同时对生成内容进行一致性检查，防止模型产生与检索事实矛盾的幻觉。

三、生产级 RAG 的代码实现

以下实现展示了 RAG 管线的核心逻辑，重点关注混合检索、重排序和 Prompt 组装三个关键环节。

import hashlib from dataclasses import dataclass, field from typing import Optional @dataclass class Document: """知识库文档结构""" doc_id: str content: str metadata: dict = field(default_factory=dict) score: float = 0.0 @dataclass class RAGConfig: """RAG 管线配置""" vector_top_k: int = 20 # 向量检索召回数 keyword_top_k: int = 20 # 关键词检索召回数 rerank_top_k: int = 5 # 重排序后保留数 chunk_size: int = 512 # 文档分块大小 chunk_overlap: int = 64 # 分块重叠字符数 enable_query_rewrite: bool = True enable_rerank: bool = True class HybridRetriever: """混合检索器：融合向量检索与关键词检索""" def __init__(self, vector_store, keyword_index, config: RAGConfig): self.vector_store = vector_store self.keyword_index = keyword_index self.config = config async def retrieve(self, query: str, query_embedding: list[float]) -> list[Document]: # 并行执行两种检索 vector_results = await self.vector_store.search( query_embedding, top_k=self.config.vector_top_k ) keyword_results = self.keyword_index.search( query, top_k=self.config.keyword_top_k ) # Reciprocal Rank Fusion 合并排序 return self._rrf_merge(vector_results, keyword_results) def _rrf_merge(self, vector_docs: list[Document], keyword_docs: list[Document], k: int = 60) -> list[Document]: """RRF 算法：避免单一检索方式的排序偏差""" scores: dict[str, float] = {} doc_map: dict[str, Document] = {} for rank, doc in enumerate(vector_docs): scores[doc.doc_id] = scores.get(doc.doc_id, 0) + 1.0 / (k + rank + 1) doc_map[doc.doc_id] = doc for rank, doc in enumerate(keyword_docs): scores[doc.doc_id] = scores.get(doc.doc_id, 0) + 1.0 / (k + rank + 1) if doc.doc_id not in doc_map: doc_map[doc.doc_id] = doc # 按融合分数降序排列 sorted_ids = sorted(scores, key=scores.get, reverse=True) return [doc_map[doc_id] for doc_id in sorted_ids] class RAGPipeline: """RAG 完整管线""" def __init__(self, retriever: HybridRetriever, reranker, llm_client, config: RAGConfig): self.retriever = retriever self.reranker = reranker self.llm = llm_client self.config = config async def query(self, question: str) -> dict: # Step 1: 查询改写（可选） search_query = question if self.config.enable_query_rewrite: search_query = await self._rewrite_query(question) # Step 2: 混合检索 query_embedding = await self._embed(search_query) candidates = await self.retriever.retrieve(search_query, query_embedding) # Step 3: 重排序（可选但强烈推荐） if self.config.enable_rerank and self.reranker: candidates = await self.reranker.rerank( query=search_query, documents=candidates, top_k=self.config.rerank_top_k ) else: candidates = candidates[:self.config.rerank_top_k] # Step 4: 组装 Prompt 并生成 context = self._build_context(candidates) answer = await self._generate(question, context) # Step 5: 引用标注 citations = self._extract_citations(answer, candidates) return {"answer": answer, "citations": citations, "sources": candidates} def _build_context(self, docs: list[Document]) -> str: """组装检索上下文：区分核心证据与补充材料""" if not docs: return "未检索到相关文档。" parts = [] for i, doc in enumerate(docs, 1): # 按相关性分数区分层级 label = "核心证据" if doc.score > 0.8 else "补充参考" parts.append(f"[{label} {i}] (来源: {doc.metadata.get('source', 'unknown')})\n{doc.content}") return "\n\n".join(parts) async def _generate(self, question: str, context: str) -> str: prompt = f"""基于以下检索到的文档内容回答问题。如果文档中没有相关信息，请明确说明。 检索上下文： {context} 问题：{question} 要求： 1. 仅基于检索到的文档内容回答，不要编造信息 2. 引用具体来源时标注 [来源 N] 3. 如果文档不足以回答问题，明确指出缺失的信息""" return await self.llm.complete(prompt)

四、RAG 架构的边界与权衡

RAG 并非万能方案，其适用性取决于具体场景的知识特性。

检索噪声与上下文窗口的矛盾。注入过多检索结果会挤占上下文窗口，降低模型的推理质量；注入过少则可能遗漏关键信息。经验值是检索结果控制在 2000-4000 token 以内，但这需要根据模型上下文长度和问题复杂度动态调整。

知识库更新的实时性。向量索引的构建是批量操作，从文档更新到索引生效存在延迟。对于要求秒级更新的场景，需要引入增量索引机制，但增量索引会带来向量空间的一致性问题。

多跳推理的局限性。标准 RAG 是单次检索-生成，无法处理需要多步推理的复杂问题。例如"比较 A 产品和 B 产品在安全合规方面的差异"需要分别检索两个产品的信息再综合比较。解决方案是引入迭代检索（Iterative Retrieval），但每增加一轮检索，延迟和成本都会线性增长。

五、总结

RAG 架构通过"检索-增强-生成"的三段式设计，有效缓解了大模型的知识时效性和可信度问题。混合检索与重排序是提升检索质量的关键，Prompt 组装策略直接影响生成效果。但 RAG 也有明确的边界：检索噪声、索引延迟、多跳推理局限，都是工程落地中需要正视的问题。

落地路线建议：第一，从简单的向量检索 + BM25 混合方案起步，验证基础效果后再引入重排序；第二，建立检索质量的评估基准，使用 Recall@K 和 MRR 指标量化检索效果；第三，针对多跳场景，优先考虑查询分解而非迭代检索，以控制延迟和成本。