RAG 系统设计：企业级检索增强生成架构
检索增强生成（RAG）是企业 AI 最重要的架构模式之一——它让 LLM 能够基于企业私有数据生成可靠的、有据可查的回答。本文系统性地讲解 RAG 的架构设计，涵盖文档分块策略、向量检索优化、上下文组装、生成质量保障四大核心环节，并提供可直接应用于生产环境的设计模式和代码示例。
Coomia发布于 2026年2月1日16 分钟阅读
分享本文Twitter / X
RAG 系统设计：企业级检索增强生成架构
“系列：S13 AI 工程 · 第 2 篇 | 难度：高级 | 阅读时间：18 分钟
#TL;DR
检索增强生成（RAG）是企业 AI 最重要的架构模式之一——它让 LLM 能够基于企业私有数据生成可靠的、有据可查的回答。本文系统性地讲解 RAG 的架构设计，涵盖文档分块策略、向量检索优化、上下文组装、生成质量保障四大核心环节，并提供可直接应用于生产环境的设计模式和代码示例。
#1. 为什么企业需要 RAG
大语言模型（LLM）拥有强大的语言理解和生成能力，但在企业场景中面临三个根本性限制：
- 知识截止：模型的训练数据有时间边界，无法获知最新信息
- 知识缺失：模型不具备企业私有数据（内部文档、业务规则、客户信息等）
- 幻觉风险：当被问及不了解的内容时，模型会"编造"看似合理的答案
RAG 通过在生成前先检索相关文档，将模型的输出"锚定"在真实数据上，从而系统性地解决这三个问题。
Code
用户查询 → 检索相关文档 → 组装上下文 → LLM 生成 → 输出验证 → 响应
#2. RAG 架构全景
#2.1 核心组件
一个生产级 RAG 系统包含以下核心组件：
Code
┌─────────────────────────────────────────────────────┐
│ RAG Pipeline │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────┐ │
│ │ Document │→ │ Chunking │→ │ Embedding│→ │Vector│ │
│ │ Ingestion │ │ Engine │ │ Service │ │ Store│ │
│ └──────────┘ └──────────┘ └──────────┘ └──────┘ │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────┐ │
│ │ Query │→ │ Retriever│→ │ Context │→ │ LLM │ │
│ │ Processor│ │ │ │ Assembler│ │ │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────┘ │
└─────────────────────────────────────────────────────┘
#2.2 离线管道与在线管道
RAG 系统天然分为两条管道：
- 离线管道（Indexing Pipeline）：负责文档处理、分块、向量化和索引构建，通常以批量或增量方式运行
- 在线管道（Query Pipeline）：负责查询处理、检索、上下文组装和生成，要求低延迟高可用
Python
from dataclasses import dataclass, field
from enum import Enum
class PipelineMode(Enum):
OFFLINE = "offline" # 索引管道
ONLINE = "online" # 查询管道
@dataclass
class RAGConfig:
"""RAG 系统配置"""
# 分块配置
chunk_size: int = 512
chunk_overlap: int = 64
chunking_strategy: str = "semantic" # semantic | fixed | recursive
# 检索配置
top_k: int = 10
rerank_top_k: int = 5
similarity_threshold: float = 0.7
# 生成配置
model_name: str = "gpt-4"
max_output_tokens: int = 2048
temperature: float = 0.1
# 质量保障
enable_citation: bool = True
enable_hallucination_check: bool = True
enable_answer_relevance_check: bool = True
#3. 文档处理与分块策略
#3.1 文档解析
企业文档格式多样——PDF、Word、PPT、Excel、HTML、Markdown、代码文件等。文档解析是 RAG 管道的第一步，也是最容易被低估的环节。
Python
from abc import ABC, abstractmethod
from pathlib import Path
class DocumentParser(ABC):
"""文档解析器基类"""
@abstractmethod
def parse(self, file_path: Path) -> list[DocumentSection]:
"""解析文档为结构化段落"""
...
class PDFParser(DocumentParser):
"""PDF 解析器 — 保留结构信息"""
def parse(self, file_path: Path) -> list[DocumentSection]:
sections = []
# 1. 提取文本和布局信息
pages = self._extract_pages_with_layout(file_path)
for page in pages:
# 2. 识别标题、段落、表格、图片
elements = self._detect_elements(page)
for element in elements:
sections.append(DocumentSection(
content=element.text,
element_type=element.type, # heading, paragraph, table, etc.
page_number=page.number,
metadata={
"source": str(file_path),
"bbox": element.bounding_box,
"font_size": element.font_size,
}
))
return sections
#3.2 分块策略对比
分块（Chunking）是 RAG 系统中最关键的设计决策之一。不同策略适用于不同场景：
| 策略 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 固定长度分块 | 按字符/token 数切分 | 简单、可预测 | 可能切断语义 | 格式统一的文档 |
| 递归分块 | 按层级分隔符递归切分 | 保留段落结构 | 块大小不均匀 | Markdown/代码 |
| 语义分块 | 按语义相似度切分 | 语义完整性最佳 | 计算成本高 | 知识库问答 |
| 文档结构分块 | 按文档结构（标题、段落）切分 | 保留文档层次 | 依赖解析质量 | 技术文档 |
Python
class SemanticChunker:
"""基于语义相似度的分块器"""
def __init__(self, embedding_model, similarity_threshold: float = 0.5):
self.embedding_model = embedding_model
self.threshold = similarity_threshold
def chunk(self, sentences: list[str]) -> list[Chunk]:
"""将句子列表按语义边界分组"""
if not sentences:
return []
# 1. 为每个句子生成嵌入
embeddings = self.embedding_model.encode(sentences)
# 2. 计算相邻句子的余弦相似度
similarities = []
for i in range(len(embeddings) - 1):
sim = cosine_similarity(embeddings[i], embeddings[i + 1])
similarities.append(sim)
# 3. 在相似度低于阈值处切分
chunks = []
current_chunk_sentences = [sentences[0]]
for i, sim in enumerate(similarities):
if sim < self.threshold:
# 语义断裂点，创建新块
chunks.append(Chunk(
text="\n".join(current_chunk_sentences),
metadata={"start_sentence": i - len(current_chunk_sentences) + 1}
))
current_chunk_sentences = [sentences[i + 1]]
else:
current_chunk_sentences.append(sentences[i + 1])
# 处理最后一个块
if current_chunk_sentences:
chunks.append(Chunk(text="\n".join(current_chunk_sentences)))
return chunks
#3.3 分块的元数据策略
每个文档块都应携带丰富的元数据，这对后续的检索、过滤和引用至关重要：
Python
@dataclass
class ChunkMetadata:
"""文档块元数据"""
# 来源信息
source_document: str # 原始文档路径/URI
document_title: str # 文档标题
section_title: str | None # 所在章节标题
page_number: int | None # 页码
# 位置信息
chunk_index: int # 在文档中的序号
total_chunks: int # 文档总块数
char_offset_start: int # 在原文中的起始位置
char_offset_end: int # 在原文中的结束位置
# 时间信息
document_created_at: str # 文档创建时间
document_updated_at: str # 文档更新时间
indexed_at: str # 索引时间
# 分类信息
document_type: str # 文档类型（policy, manual, report...）
access_level: str # 访问级别
department: str # 所属部门
# 上下文窗口
parent_chunk_id: str | None # 父块 ID（用于层次检索）
sibling_chunk_ids: list[str] # 相邻块 ID（用于上下文扩展）
#4. 向量检索架构
#4.1 嵌入模型选择
嵌入模型（Embedding Model）将文本转换为高维向量。选择合适的嵌入模型需要考虑：
- 语言支持：中文、英文、多语言
- 维度：影响存储成本和检索速度
- 质量：在目标领域的语义表示能力
- 速度：推理延迟和吞吐量
| 模型 | 维度 | 中文支持 | 特点 |
|---|---|---|---|
| text-embedding-3-large | 3072 | 良好 | OpenAI 最新，支持维度截断 |
| bge-large-zh-v1.5 | 1024 | 优秀 | BAAI 中文专优 |
| multilingual-e5-large | 1024 | 良好 | 多语言统一表示 |
| nomic-embed-text-v1.5 | 768 | 一般 | 开源、可本地部署 |
#4.2 向量索引类型
Python
class VectorIndexConfig:
"""向量索引配置"""
@staticmethod
def hnsw_config() -> dict:
"""HNSW — 高召回、高内存"""
return {
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {
"M": 16, # 每个节点的最大连接数
"ef_construction": 200, # 构建时搜索宽度
"ef_search": 128, # 查询时搜索宽度
},
"use_case": "实时查询，数据量 < 10M",
}
@staticmethod
def ivf_pq_config() -> dict:
"""IVF_PQ — 大规模、低内存"""
return {
"index_type": "IVF_PQ",
"metric_type": "L2",
"params": {
"nlist": 4096, # 聚类中心数
"m": 16, # PQ 子空间数
"nbits": 8, # 每个子空间的位数
"nprobe": 64, # 查询时扫描的聚类数
},
"use_case": "海量数据，召回率可适当牺牲",
}
#4.3 混合检索策略
纯向量检索在某些场景下效果不佳（如精确匹配产品编号、日期等）。混合检索结合向量搜索和关键词搜索的优势：
Python
class HybridRetriever:
"""混合检索器：向量 + 关键词"""
def __init__(self, vector_store, keyword_store, alpha: float = 0.7):
self.vector_store = vector_store
self.keyword_store = keyword_store
self.alpha = alpha # 向量得分权重
def retrieve(self, query: str, top_k: int = 10) -> list[RetrievalResult]:
# 1. 向量检索
vector_results = self.vector_store.search(
query_embedding=self.embed(query),
top_k=top_k * 2, # 过检索
)
# 2. 关键词检索（BM25）
keyword_results = self.keyword_store.search(
query=query,
top_k=top_k * 2,
)
# 3. 归一化分数
vector_scores = self._normalize_scores(vector_results)
keyword_scores = self._normalize_scores(keyword_results)
# 4. 加权融合（Reciprocal Rank Fusion 或线性加权）
fused = self._reciprocal_rank_fusion(
vector_results=vector_scores,
keyword_results=keyword_scores,
k=60,
)
return fused[:top_k]
def _reciprocal_rank_fusion(self, vector_results, keyword_results, k=60):
"""RRF 融合算法"""
scores = {}
for rank, result in enumerate(vector_results):
scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1)
for rank, result in enumerate(keyword_results):
scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1)
sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [RetrievalResult(id=doc_id, score=score) for doc_id, score in sorted_results]
#5. 查询处理与增强
#5.1 查询理解
用户查询往往不够精确或包含歧义。查询处理层负责优化查询以提升检索质量：
Python
class QueryProcessor:
"""查询处理器"""
def __init__(self, llm_client):
self.llm = llm_client
async def process(self, raw_query: str) -> ProcessedQuery:
"""处理用户查询"""
# 1. 查询分类
intent = await self._classify_intent(raw_query)
# 2. 查询改写
rewritten = await self._rewrite_query(raw_query, intent)
# 3. 查询扩展（生成多个检索查询）
expanded_queries = await self._expand_query(rewritten)
# 4. 提取过滤条件
filters = await self._extract_filters(raw_query)
return ProcessedQuery(
original=raw_query,
rewritten=rewritten,
expanded=expanded_queries,
intent=intent,
filters=filters,
)
async def _expand_query(self, query: str) -> list[str]:
"""HyDE：生成假设性文档作为检索查询"""
prompt = f"""Given the question: "{query}"
Generate 3 different versions of this question that capture different aspects
and would help retrieve relevant documents. Return as JSON array."""
response = await self.llm.complete(prompt)
return json.loads(response)
#5.2 多步检索（Multi-hop Retrieval）
复杂问题需要从多个文档中组合信息。多步检索通过迭代方式逐步收集所需上下文：
Python
class MultiHopRetriever:
"""多步检索器：处理需要跨文档推理的复杂查询"""
def __init__(self, retriever, llm, max_hops: int = 3):
self.retriever = retriever
self.llm = llm
self.max_hops = max_hops
async def retrieve(self, query: str) -> list[RetrievalResult]:
collected_context = []
current_query = query
for hop in range(self.max_hops):
# 1. 检索当前查询的相关文档
results = self.retriever.retrieve(current_query, top_k=5)
collected_context.extend(results)
# 2. 判断是否需要继续检索
assessment = await self._assess_completeness(
query, collected_context
)
if assessment.is_sufficient:
break
# 3. 生成后续查询
current_query = await self._generate_followup_query(
original_query=query,
current_context=collected_context,
missing_info=assessment.missing_information,
)
return self._deduplicate(collected_context)
#6. 上下文组装与提示工程
#6.1 上下文窗口管理
LLM 的上下文窗口有限（即使是 128K 的模型，也需要为输出保留空间）。上下文组装需要精心安排检索到的文档块：
Python
class ContextAssembler:
"""上下文组装器"""
def __init__(self, max_context_tokens: int = 8000):
self.max_tokens = max_context_tokens
def assemble(
self,
query: str,
retrieved_chunks: list[Chunk],
system_prompt: str,
) -> AssembledContext:
"""组装 LLM 输入上下文"""
# 1. 按相关性排序（已由检索器完成）
# 2. 去重和合并相邻块
merged = self._merge_adjacent_chunks(retrieved_chunks)
# 3. Token 预算分配
system_tokens = self._count_tokens(system_prompt)
query_tokens = self._count_tokens(query)
available = self.max_tokens - system_tokens - query_tokens - 200 # 缓冲
# 4. 贪心填充（优先填充高相关性的块）
selected_chunks = []
used_tokens = 0
for chunk in merged:
chunk_tokens = self._count_tokens(chunk.text)
if used_tokens + chunk_tokens <= available:
selected_chunks.append(chunk)
used_tokens += chunk_tokens
# 5. 按文档原始顺序重排（保持阅读连贯性）
selected_chunks.sort(key=lambda c: (c.metadata.source_document, c.metadata.chunk_index))
return AssembledContext(
chunks=selected_chunks,
total_tokens=used_tokens,
coverage_ratio=len(selected_chunks) / len(retrieved_chunks),
)
#6.2 提示模板设计
Python
RAG_PROMPT_TEMPLATE = """你是一个企业知识助手。基于以下检索到的文档片段回答用户的问题。
## 规则
1. 只基于提供的文档回答，不要使用你的训练知识
2. 如果文档中没有足够的信息，明确说"根据现有文档，我无法回答这个问题"
3. 每个关键陈述都必须引用来源文档，格式为 [来源: 文档名, 页码]
4. 如果不同文档有矛盾信息，指出矛盾并说明各文档的说法
## 检索到的文档
{context}
## 用户问题
{query}
## 回答
"""
#7. 生成质量保障
#7.1 幻觉检测
Python
class HallucinationDetector:
"""幻觉检测器：验证生成内容是否有文档支撑"""
def __init__(self, nli_model):
self.nli_model = nli_model # Natural Language Inference 模型
def detect(self, generated_answer: str, source_chunks: list[Chunk]) -> list[HallucinationFlag]:
flags = []
# 将回答拆分为独立声明
claims = self._extract_claims(generated_answer)
for claim in claims:
# 对每个声明，检查是否被源文档支持
support_scores = []
for chunk in source_chunks:
score = self.nli_model.predict_entailment(
premise=chunk.text,
hypothesis=claim.text,
)
support_scores.append(score)
max_support = max(support_scores) if support_scores else 0
if max_support < 0.5:
flags.append(HallucinationFlag(
claim=claim.text,
confidence=1 - max_support,
suggestion="该声明在源文档中未找到充分支撑",
))
return flags
#7.2 答案相关性评估
Python
class AnswerRelevanceEvaluator:
"""评估生成的答案与用户问题的相关性"""
async def evaluate(self, query: str, answer: str) -> float:
"""返回 0-1 的相关性分数"""
prompt = f"""Rate the relevance of the following answer to the question.
Score from 0 (completely irrelevant) to 1 (perfectly relevant).
Question: {query}
Answer: {answer}
Return only the numeric score."""
score = float(await self.llm.complete(prompt))
return min(max(score, 0), 1)
#7.3 引用验证
每个引用都应可追溯到具体的源文档和位置，这是企业级 RAG 的刚需：
Python
class CitationVerifier:
"""引用验证器"""
def verify(self, answer: str, chunks: list[Chunk]) -> VerificationResult:
citations = self._extract_citations(answer)
verified = []
unverified = []
for citation in citations:
# 在源文档中查找引用的内容
match = self._find_in_chunks(citation.text, chunks)
if match and match.similarity > 0.85:
verified.append(VerifiedCitation(
citation=citation,
source_chunk=match.chunk,
similarity=match.similarity,
))
else:
unverified.append(citation)
return VerificationResult(
verified=verified,
unverified=unverified,
verification_rate=len(verified) / len(citations) if citations else 1.0,
)
#8. 生产部署考量
#8.1 性能优化
- 嵌入缓存：缓存常见查询的嵌入向量
- 预热索引：将热门文档的索引加载到内存
- 异步管道：文档处理和索引构建异步执行
- 批量推理：嵌入计算批量处理以利用 GPU 并行
#8.2 容错设计
- 检索降级：向量检索超时时自动降级到关键词检索
- 模型降级：主模型不可用时切换到备用模型
- 缓存回退：返回缓存中的相似历史答案
- 熔断机制：避免故障级联传播
#8.3 监控指标
| 指标类别 | 具体指标 | 目标值 |
|---|---|---|
| 延迟 | 端到端 P95 延迟 | < 3s |
| 质量 | 答案相关性 | > 0.85 |
| 质量 | 幻觉率 | < 5% |
| 质量 | 引用准确率 | > 95% |
| 可用性 | 系统可用率 | 99.9% |
| 成本 | 每次查询成本 | < $0.05 |
#9. 进阶模式
#9.1 Agentic RAG
将 RAG 与 Agent 能力结合，让系统自主决定何时检索、检索什么、如何组合信息：
Python
class AgenticRAG:
"""Agent 增强的 RAG 系统"""
def __init__(self, retriever, llm, tools):
self.retriever = retriever
self.llm = llm
self.tools = tools # 计算器、数据库查询、API 调用等
async def answer(self, query: str) -> AgentResponse:
plan = await self._plan(query)
results = []
for step in plan.steps:
if step.action == "retrieve":
docs = self.retriever.retrieve(step.query)
results.append(("context", docs))
elif step.action == "compute":
output = await self.tools.execute(step.tool, step.args)
results.append(("computation", output))
elif step.action == "synthesize":
answer = await self._synthesize(query, results)
results.append(("answer", answer))
return AgentResponse(answer=results[-1][1], trace=results)
#9.2 GraphRAG
结合知识图谱进行结构化推理，特别适合需要关系推理的场景。
#9.3 自适应 RAG
根据查询的复杂度动态调整检索策略和模型选择，在质量和成本间取得最优平衡。
#Key Takeaways
- RAG 是企业 AI 的核心架构模式——通过检索"锚定"生成，系统性解决 LLM 的知识缺失和幻觉问题
- 分块策略决定检索质量——语义分块 > 递归分块 > 固定长度分块，但需根据文档类型灵活选择
- 混合检索是生产最佳实践——向量检索处理语义匹配，关键词检索处理精确匹配
- 查询处理不可忽视——查询改写和扩展可显著提升检索召回率
- 质量保障是企业级的刚需——幻觉检测、引用验证、答案相关性评估缺一不可
- 生产部署需要全面的容错和监控——延迟、质量、成本三个维度同步优化
#Next Article
下一篇 S13-03: 向量嵌入管道 将深入讲解嵌入模型的选择、微调、部署和管道化运维，这是 RAG 系统的核心基础设施。
Tags: #RAG #检索增强生成 #LLM #向量检索 #文档分块 #幻觉检测 #企业AI #AI架构