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 指标量化检索效果；第三，针对多跳场景，优先考虑查询分解而非迭代检索，以控制延迟和成本。