在过去 18 个月里，RAG 已经悄无声息地完成了一次本质跃迁：从最初"搜索 + LLM"的简单拼接模式，进化成了一套完整、分层、体系化的工程化学科。本文将带你从 0 到 1 建立对 RAG 工程化体系的完整认知。
一、RAG 到底是什么？它解决了什么核心问题？
RAG（Retrieval-Augmented Generation，检索增强生成） 是一种将大语言模型与外部知识检索系统相结合的技术架构。它的核心思想是：在生成回答之前，先从外部知识库中检索相关信息，然后将这些信息作为上下文提供给大模型，从而生成更准确、更可靠的回答。
1.1 RAG 解决的核心痛点
大语言模型虽然能力强大，但存在三个致命的缺陷：
1.2 为什么"接个向量库"远远不够？
很多开发者对 RAG 的认知还停留在"写个检索逻辑，调用一下大模型 API"的阶段。但实际上，生产级 RAG 系统需要解决以下问题：
数据提取层面
复杂文档格式（PDF、Word、扫描件）的精准解析
表格、公式、图片内嵌内容的提取
多语言混合文档的处理
文本分块层面
如何划分块的大小？太大导致上下文稀释，太小丢失语义
段落边界、语义完整性如何保持
递归分块 vs 语义分块的选择
嵌入模型层面
中文场景需要专门的中文嵌入模型
不同嵌入模型对检索准确率的影响巨大
长文档需要支持更长上下文的嵌入模型
检索策略层面
纯向量检索的局限性
混合检索（关键词 + 向量）的重要性
查询改写、意图扩展的必要性
重排序（Rerank）的价值
上下文管理层面
Token 窗口限制
超长上下文的摘要策略
对话历史的处理
效果评估层面
如何量化 RAG 系统的好坏
持续优化的闭环机制
二、RAG 七层架构详解
┌─────────────────────────────────────────────────────────────────────────────┐
│ RAG 系统七层架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第七层：效果评估层 (Evaluation) │ │
│ │ Ragas / TruLens / Giskard → 三级评估体系 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第六层：编排框架层 (Orchestration) │ │
│ │ LangChain / LlamaIndex / Haystack → 流程串联 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第五层：大语言模型层 (LLMs) │ │
│ │ GPT-4 / Claude / DeepSeek / Qwen → 答案生成 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第四层：开源模型接入层 (Open LLMs) │ │
│ │ Ollama / HuggingFace / Groq → 模型灵活性 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第三层：向量数据库层 (Vector Store) │ │
│ │ Milvus / Qdrant / Weaviate / ChromaDB / Pinecone │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第二层：文本嵌入层 (Embeddings) │ │
│ │ BGE / Voyage AI / OpenAI Embeddings → 语义向量化 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 第一层：数据提取层 (Data Extraction) │ │
│ │ FireCrawl / LlamaParse / Docling → 内容解析 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘2.1 第一层：数据提取层
数据质量决定了 RAG 效果的天花板。即使后续环节再完美，如果源数据就是错误或混乱的，结果也不可能正确。
核心工具选型
2.2 第二层：文本嵌入层
嵌入模型的质量直接决定了检索环节的准确率。
中文场景推荐
# 嵌入模型选型配置
embedding_config:
# 中文首选：BGE 系列
bge_large_zh:
model_name: "BAAI/bge-large-zh-v1.5"
dimension: 1024
max_seq_length: 512
use_case: "通用中文场景"
bge_m3:
model_name: "BAAI/bge-m3"
dimension: 1024
max_seq_length: 8192
use_case: "长文档场景"
# 备选方案
voyage_multilingual:
provider: "Voyage AI"
model: "voyage-multilingual-2"
use_case: "多语言场景"2.3 第三层：向量数据库层
向量数据库是 RAG 系统的记忆中枢，负责存储和检索向量化的知识。
2.4 第四-六层：模型与编排层
这三层决定了 RAG 系统的智能程度和灵活性。推荐组合：
# 生产级 RAG 配置示例
rag_config:
# 编排框架
orchestration:
framework: "langchain" # 或 llama_index / haystack
# 大语言模型
llm:
primary: "gpt-4o-mini" # 通用场景
fallback: "deepseek-v3" # 降级选项
long_context: "claude-3-5-sonnet" # 长文档场景
# 检索配置
retrieval:
top_k: 5
hybrid_search: true # 混合检索
rerank: true # 重排序
query_expansion: true # 查询扩展三、文档解析与分块策略
3.1 递归字符分块
递归分块是当前最常用的分块策略，通过递归地按特定分隔符划分文本，直到块大小满足要求。
from typing import List, Optional
from langchain.text_splitter import RecursiveCharacterTextSplitter
class DocumentChunker:
"""
文档分块器：支持递归分块和语义分块两种策略
"""
def __init__(
self,
chunk_size: int = 1000,
chunk_overlap: int = 200,
separators: Optional[List[str]] = None
):
"""
初始化分块器
Args:
chunk_size: 每个块的 token 数量（近似）
chunk_overlap: 块之间的重叠 token 数，保证语义连续性
separators: 分隔符列表，按优先级排序
"""
if separators is None:
# 默认分隔符：先按段落，再按句子，最后按单词
separators = ["\n\n", "\n", ". ", " ", ""]
self.splitter = RecursiveCharacterTextSplitter(
separators=separators,
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=self._count_tokens, # 自定义 token 计数
is_separator_regex=False,
)
def _count_tokens(self, text: str) -> int:
"""估算 token 数量（中文约 2 字符 = 1 token）"""
# 简单估算：中文 * 1.5 + 英文 / 4
chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
other_chars = len(text) - chinese_chars
return int(chinese_chars * 1.5 + other_chars / 4)
def chunk_text(self, text: str, metadata: Optional[dict] = None) -> List[dict]:
"""
将文本分割成块
Args:
text: 输入文本
metadata: 元数据（如文档标题、来源等）
Returns:
块列表，每个块包含内容和元数据
"""
# 使用 langchain 的分块器
from langchain.schema import Document
langchain_docs = [Document(page_content=text, metadata=metadata or {})]
chunks = self.splitter.split_documents(langchain_docs)
# 转换为标准格式
return [
{
"content": chunk.page_content,
"metadata": {
**chunk.metadata,
"chunk_id": idx,
"total_chunks": len(chunks)
}
}
for idx, chunk in enumerate(chunks)
]
def chunk_documents(
self,
documents: List[dict]
) -> List[dict]:
"""
批量分块处理多个文档
Args:
documents: 文档列表，每个文档包含 content 和可选的 metadata
Returns:
所有文档的分块列表
"""
all_chunks = []
for doc in documents:
chunks = self.chunk_text(
text=doc.get("content", ""),
metadata=doc.get("metadata", {})
)
all_chunks.extend(chunks)
return all_chunks3.2 语义分块
语义分块根据文本的语义相似性自动划分边界，比固定长度分块效果更好。
from typing import List, Optional
import numpy as np
class SemanticChunker:
"""
语义分块器：基于语义相似性自动划分文档边界
核心思想：
1. 将文本切分成较小的句子单元
2. 计算相邻句子之间的语义相似度
3. 相似度骤降处即为语义边界
4. 将连续的句子聚合为块
"""
def __init__(
self,
embed_model, # 嵌入模型实例
min_chunk_size: int = 3, # 最少句子数
max_chunk_size: int = 10, # 最多句子数
similarity_threshold: float = 0.5, # 相似度阈值
buffer_size: int = 1 # 边界处保留的相邻句子数
):
self.embed_model = embed_model
self.min_chunk_size = min_chunk_size
self.max_chunk_size = max_chunk_size
self.similarity_threshold = similarity_threshold
self.buffer_size = buffer_size
def _split_into_sentences(self, text: str) -> List[str]:
"""简单句子分割（实际项目可使用更复杂的 NLP 工具）"""
import re
# 按常见句末标点分割
sentences = re.split(r'[。！？\n]+', text)
return [s.strip() for s in sentences if s.strip()]
def _compute_similarity(self, text1: str, text2: str) -> float:
"""计算两个文本的语义相似度"""
emb1 = self.embed_model.embed_query(text1)
emb2 = self.embed_model.embed_query(text2)
# 余弦相似度
dot_product = np.dot(emb1, emb2)
norm1 = np.linalg.norm(emb1)
norm2 = np.linalg.norm(emb2)
return dot_product / (norm1 * norm2 + 1e-8)
def _find_breaking_points(
self,
sentences: List[str]
) -> List[int]:
"""
找到语义边界点
策略：
1. 计算相邻句子的相似度
2. 相似度低于阈值视为语义断裂
3. 强制在 max_chunk_size 处截断
"""
if len(sentences) <= 1:
return []
breaking_points = []
for i in range(len(sentences) - 1):
similarity = self._compute_similarity(
sentences[i],
sentences[i + 1]
)
# 语义断裂 或 达到最大块大小
if similarity < self.similarity_threshold:
breaking_points.append(i + 1)
elif (i + 1) % self.max_chunk_size == 0 and i + 1 < len(sentences) - 1:
breaking_points.append(i + 1)
return sorted(set(breaking_points))
def chunk_text(self, text: str, metadata: Optional[dict] = None) -> List[dict]:
"""执行语义分块"""
sentences = self._split_into_sentences(text)
if len(sentences) < self.min_chunk_size:
# 文本太短，直接返回整个文本
return [{
"content": text,
"metadata": {
**(metadata or {}),
"chunk_id": 0,
"total_chunks": 1,
"chunking_method": "semantic"
}
}]
breaking_points = self._find_breaking_points(sentences)
breaking_points = [0] + breaking_points + [len(sentences)]
chunks = []
for i in range(len(breaking_points) - 1):
start, end = breaking_points[i], breaking_points[i + 1]
chunk_sentences = sentences[start:end]
chunks.append({
"content": "".join(chunk_sentences),
"metadata": {
**(metadata or {}),
"chunk_id": i,
"total_chunks": len(breaking_points) - 1,
"chunking_method": "semantic",
"num_sentences": len(chunk_sentences)
}
})
return chunks四、检索策略：混合检索与重排序
4.1 为什么需要混合检索？
纯向量检索虽然能捕获语义相似性，但存在以下问题：
对专有名词、型号、代码等精确匹配能力弱
无法处理拼写变体
长尾查询效果不稳定
关键词检索（BM25/TF-IDF）正好弥补这些缺陷。
from typing import List, Dict, Any, Optional
import numpy as np
class HybridRetriever:
"""
混合检索器：结合向量检索和关键词检索的优势
检索流程：
1. 同时执行向量检索和关键词检索
2. 对结果进行标准化分数融合
3. 可选：使用重排序模型进一步优化
"""
def __init__(
self,
vector_store, # 向量数据库
bm25_index, # BM25 索引
embed_model, # 嵌入模型
reranker: Optional[Any] = None, # 重排序模型
vector_weight: float = 0.7, # 向量检索权重
keyword_weight: float = 0.3, # 关键词检索权重
top_k: int = 20, # 初步召回数量
final_k: int = 5 # 最终返回数量
):
self.vector_store = vector_store
self.bm25_index = bm25_index
self.embed_model = embed_model
self.reranker = reranker
self.vector_weight = vector_weight
self.keyword_weight = keyword_weight
self.top_k = top_k
self.final_k = final_k
def _vector_search(self, query: str) -> List[Dict[str, Any]]:
"""向量语义检索"""
query_embedding = self.embed_model.embed_query(query)
results = self.vector_store.similarity_search_by_vector(
embedding=query_embedding,
k=self.top_k
)
return [
{
"id": r["id"],
"content": r["content"],
"score": r.get("score", 0.0),
"source": "vector"
}
for r in results
]
def _keyword_search(self, query: str) -> List[Dict[str, Any]]:
"""关键词 BM25 检索"""
results = self.bm25_index.search(query, k=self.top_k)
return [
{
"id": r["id"],
"content": r["content"],
"score": r.get("score", 0.0),
"source": "keyword"
}
for r in results
]
def _normalize_scores(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Min-Max 分数标准化"""
if not results:
return results
scores = [r["score"] for r in results]
min_score, max_score = min(scores), max(scores)
if max_score == min_score:
# 所有分数相同，设为 1.0
return [{**r, "normalized_score": 1.0} for r in results]
return [
{**r, "normalized_score": (r["score"] - min_score) / (max_score - min_score)}
for r in results
]
def _fusion_scores(
self,
vector_results: List[Dict[str, Any]],
keyword_results: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""RRF（Reciprocal Rank Fusion）分数融合"""
# 构建 id -> 结果 的映射
all_results = {}
for rank, result in enumerate(vector_results):
result["fusion_score"] = (
self.vector_weight * result["normalized_score"] +
(1 / (rank + 1)) * 0.1 # RRF 调整
)
all_results[result["id"]] = result
for rank, result in enumerate(keyword_results):
result["fusion_score"] = (
self.keyword_weight * result["normalized_score"] +
(1 / (rank + 1)) * 0.1 # RRF 调整
)
if result["id"] in all_results:
all_results[result["id"]]["fusion_score"] += result["fusion_score"]
all_results[result["id"]]["source"] = "hybrid"
else:
all_results[result["id"]] = result
# 按融合分数排序
sorted_results = sorted(
all_results.values(),
key=lambda x: x["fusion_score"],
reverse=True
)
return sorted_results[:self.top_k]
def _rerank(self, query: str, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""使用重排序模型优化结果顺序"""
if not self.reranker:
return results[:self.final_k]
# 准备重排序输入
doc_texts = [r["content"] for r in results]
reranked = self.reranker.rank(query=query, documents=doc_texts)
# 更新分数
for i, idx in enumerate(reranked["indices"]):
results[idx]["rerank_score"] = reranked["scores"][i]
# 按重排序分数排序
return sorted(
results,
key=lambda x: x.get("rerank_score", x["fusion_score"]),
reverse=True
)[:self.final_k]
def retrieve(self, query: str) -> List[Dict[str, Any]]:
"""
执行混合检索
Args:
query: 用户查询
Returns:
检索结果列表
"""
# 1. 并行执行向量检索和关键词检索
vector_results = self._vector_search(query)
keyword_results = self._keyword_search(query)
# 2. 标准化分数
vector_results = self._normalize_scores(vector_results)
keyword_results = self._normalize_scores(keyword_results)
# 3. 融合分数
fused_results = self._fusion_scores(vector_results, keyword_results)
# 4. 重排序（如配置）
if self.reranker:
final_results = self._rerank(query, fused_results)
else:
final_results = fused_results[:self.final_k]
return final_results4.2 查询改写与扩展
用户的自然语言查询往往不够精确，需要通过查询改写提升检索效果。
class QueryExpander:
"""
查询扩展器：通过大模型生成多个相关查询，提升召回率
策略：
1. 同义词扩展
2. 概念分解
3. 假设性问题生成
"""
def __init__(self, llm, max_queries: int = 3):
self.llm = llm
self.max_queries = max_queries
def expand(self, query: str) -> List[str]:
"""生成扩展查询列表"""
prompt = f"""给定用户查询，生成 {self.max_queries} 个不同的搜索查询，
帮助从不同角度检索相关信息。
用户查询：{query}
要求：
1. 生成多样化但相关的查询
2. 包含同义词和专业术语
3. 可以包含更具体或更宽泛的变体
4. 每个查询一行，直接输出，不要编号
扩展查询："""
response = self.llm.invoke(prompt)
# 解析响应
queries = [q.strip() for q in response.split('\n') if q.strip()]
return [query] + queries[:self.max_queries] # 保留原始查询五、上下文窗口管理与三级应答机制
5.1 Token 限制与摘要策略
当检索内容超过模型上下文窗口时，需要采用摘要策略。
from typing import List, Dict, Any, Optional
import tiktoken
class ContextWindowManager:
"""
上下文窗口管理器：处理超长上下文场景
策略：
1. Token 计数与截断
2. 基于相关性的选择
3. 摘要压缩
"""
def __init__(
self,
model_name: str = "gpt-4o",
max_tokens: int = 128000, # GPT-4o 支持 128K
reserved_tokens: int = 4000 # 保留给生成内容的空间
):
self.model_name = model_name
self.max_tokens = max_tokens
self.reserved_tokens = reserved_tokens
self.available_tokens = max_tokens - reserved_tokens
# 初始化 tokenizer
self.encoding = tiktoken.encoding_for_model(model_name)
def count_tokens(self, text: str) -> int:
"""计算文本的 token 数量"""
return len(self.encoding.encode(text))
def truncate_to_limit(
self,
texts: List[str],
max_tokens: Optional[int] = None
) -> List[str]:
"""
智能截断：在 token 限制内尽可能保留内容
策略：
1. 按顺序添加文本
2. 如果添加后超限，截断当前文本或跳过
3. 优先保留靠前的相关内容
"""
limit = max_tokens or self.available_tokens
result = []
current_tokens = 0
for text in texts:
text_tokens = self.count_tokens(text)
if current_tokens + text_tokens <= limit:
result.append(text)
current_tokens += text_tokens
else:
# 尝试截断当前文本
remaining_tokens = limit - current_tokens
if remaining_tokens > 100: # 至少保留 100 tokens
truncated = self._truncate_text(text, remaining_tokens)
result.append(truncated)
# 超出限制，停止添加
break
return result
def _truncate_text(self, text: str, max_tokens: int) -> str:
"""截断文本到指定 token 数"""
tokens = self.encoding.encode(text)
truncated_tokens = tokens[:max_tokens]
return self.encoding.decode(truncated_tokens)
def summarize_long_content(
self,
content: str,
llm,
target_tokens: int = 2000
) -> str:
"""
对超长内容进行摘要
使用"map-reduce"风格摘要：
1. 将内容分成小块
2. 每块生成摘要
3. 合并摘要再次摘要
"""
chunk_size = 5000 # 每次处理 5000 字符
chunks = [
content[i:i+chunk_size]
for i in range(0, len(content), chunk_size)
]
if len(chunks) == 1:
# 内容不长，直接摘要
return self._summarize_single(content, llm, target_tokens)
# Map: 每块摘要
chunk_summaries = []
for chunk in chunks:
summary = self._summarize_single(
chunk,
llm,
target_tokens // len(chunks)
)
chunk_summaries.append(summary)
# Reduce: 合并摘要
combined = "\n\n".join(chunk_summaries)
return self._summarize_single(combined, llm, target_tokens)
def _summarize_single(
self,
content: str,
llm,
target_tokens: int
) -> str:
"""单次摘要"""
prompt = f"""请用简洁的语言总结以下内容，保留关键信息：
{content}
摘要（不超过 {target_tokens} tokens）："""
response = llm.invoke(prompt)
return response.content5.2 三级应答机制
根据检索置信度决定回答策略，实现精准路由。
from enum import Enum
from dataclasses import dataclass
from typing import Optional, List, Dict, Any
class ConfidenceLevel(Enum):
"""置信度级别"""
HIGH = "high" # 检索到高相关上下文
MEDIUM = "medium" # 检索到中等相关上下文
LOW = "low" # 检索结果不理想
NONE = "none" # 未检索到相关内容
@dataclass
class ResponseContext:
"""应答上下文"""
confidence: ConfidenceLevel
retrieved_docs: List[Dict[str, Any]]
context_tokens: int
needs_generation: bool
fallback_strategy: Optional[str] = None
class ThreeTierResponseRouter:
"""
三级应答路由器：根据置信度选择最佳应答策略
┌──────────────────────────────────────────────────────────────┐
│ 用户查询 │
└──────────────────────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────────┐
│ 置信度评估 │
│ - 检索文档数 │
│ - 平均相似度分数 │
│ - 答案完整性 │
└──────────────────────────────────────────────────────────────┘
↓
┌─────────────────────┼─────────────────────┐
↓ ↓ ↓
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ HIGH │ │ MEDIUM │ │ LOW/NONE │
│ 直接生成 │ │ 增强生成 │ │ 降级处理 │
└──────────────┘ └──────────────┘ └──────────────┘
↓ ↓ ↓
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ RAG + 严格 │ │ RAG + 宽松 │ │ 拒答/转人工 │
│ 引用控制 │ │ 引用控制 │ │ /知识库推荐 │
└──────────────┘ └──────────────┘ └──────────────┘
"""
def __init__(
self,
retriever, # 检索器
llm, # 大语言模型
context_manager: ContextWindowManager, # 上下文管理器
# 置信度阈值
high_threshold: float = 0.85,
medium_threshold: float = 0.60,
# 其他配置
min_docs_required: int = 2,
max_context_docs: int = 5
):
self.retriever = retriever
self.llm = llm
self.context_manager = context_manager
self.high_threshold = high_threshold
self.medium_threshold = medium_threshold
self.min_docs_required = min_docs_required
self.max_context_docs = max_context_docs
def _evaluate_confidence(
self,
query: str,
retrieved_docs: List[Dict[str, Any]]
) -> ResponseContext:
"""
评估检索置信度
综合考虑：
1. 检索到的文档数量
2. 文档的平均相似度分数
3. 文档内容与查询的相关性
"""
if not retrieved_docs:
return ResponseContext(
confidence=ConfidenceLevel.NONE,
retrieved_docs=[],
context_tokens=0,
needs_generation=False,
fallback_strategy="no_context"
)
# 计算平均相似度
scores = [doc.get("score", 0) for doc in retrieved_docs]
avg_score = sum(scores) / len(scores)
max_score = max(scores)
# 综合评估
# 考虑：文档数量、平均分数、最高分数
confidence_score = (avg_score * 0.4 + max_score * 0.6)
if confidence_score >= self.high_threshold and len(retrieved_docs) >= self.min_docs_required:
confidence = ConfidenceLevel.HIGH
elif confidence_score >= self.medium_threshold:
confidence = ConfidenceLevel.MEDIUM
else:
confidence = ConfidenceLevel.LOW
# 计算上下文 token 数
context_texts = [doc["content"] for doc in retrieved_docs]
context_tokens = self.context_manager.count_tokens(
" ".join(context_texts)
)
return ResponseContext(
confidence=confidence,
retrieved_docs=retrieved_docs,
context_tokens=context_tokens,
needs_generation=True,
fallback_strategy=None
)
def _build_prompt(
self,
query: str,
context_docs: List[Dict[str, Any]],
confidence: ConfidenceLevel
) -> str:
"""
构建不同置信度级别的 prompt
策略：
- HIGH: 严格基于上下文，使用引用
- MEDIUM: 基于上下文但允许一定灵活性
- LOW: 提示不确定性，建议其他渠道
"""
context_text = "\n\n".join([
f"[文档 {i+1}]\n{doc['content']}"
for i, doc in enumerate(context_docs)
])
if confidence == ConfidenceLevel.HIGH:
return f"""你是一个专业的问答助手。请严格基于以下上下文回答用户问题。
上下文：
{context_text}
用户问题：{query}
要求：
1. 只使用上下文中的信息回答
2. 对于每个关键信息，使用 [文档X] 格式标注来源
3. 如果上下文中没有相关信息，请明确说明"根据现有资料，我无法回答这个问题"
回答："""
elif confidence == ConfidenceLevel.MEDIUM:
return f"""你是一个专业的问答助手。请主要基于以下上下文回答用户问题，允许适当补充相关背景知识。
上下文：
{context_text}
用户问题：{query}
要求：
1. 优先使用上下文中的信息
2. 可以适当补充上下文未涵盖但明显相关的知识
3. 对于上下文未明确的信息，请标注"根据补充知识..."
4. 如果上下文完全不相关，请明确说明
回答："""
else: # LOW or NONE
return f"""你是一个专业的问答助手。
用户问题：{query}
注意：当前检索到的资料与问题的相关性不高。请：
1. 坦诚告知用户当前资料有限
2. 提供你已有的通用知识作为参考
3. 建议用户通过哪些渠道获取更准确的信息
4. 避免编造具体的数据或细节
请基于你的通用知识给出一个参考性回答："""
def _generate_response(
self,
query: str,
context: ResponseContext
) -> Dict[str, Any]:
"""生成最终响应"""
if context.confidence == ConfidenceLevel.NONE:
return {
"answer": "抱歉，我目前没有找到与您问题相关的信息。建议您：\n1. 尝试重新表述问题\n2. 咨询相关领域的专业人员\n3. 查看官方文档获取更多帮助",
"confidence": "none",
"source_documents": [],
"needs_human_review": True,
"suggestions": [
"转人工客服",
"查看知识库",
"提交工单"
]
}
# 限制上下文文档数量
docs_to_use = context.retrieved_docs[:self.max_context_docs]
# 构建 prompt
prompt = self._build_prompt(query, docs_to_use, context.confidence)
# 生成回答
response = self.llm.invoke(prompt)
return {
"answer": response.content,
"confidence": context.confidence.value,
"source_documents": [
{
"content": doc["content"][:200] + "...", # 截断显示
"score": doc.get("score", 0),
"metadata": doc.get("metadata", {})
}
for doc in docs_to_use
],
"needs_human_review": context.confidence == ConfidenceLevel.LOW,
"context_used_tokens": context.context_tokens
}
def answer(self, query: str) -> Dict[str, Any]:
"""
主入口：回答用户问题
流程：
1. 检索相关文档
2. 评估置信度
3. 根据置信度选择策略生成回答
"""
# 1. 检索
retrieved_docs = self.retriever.retrieve(query)
# 2. 评估置信度
context = self._evaluate_confidence(query, retrieved_docs)
# 3. 生成回答
return self._generate_response(query, context)六、完整 RAG Pipeline 代码
"""
完整的 RAG Pipeline 实现
集成：数据加载 → 分块 → 嵌入 → 存储 → 检索 → 重排 → 生成
"""
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from pathlib import Path
import yaml
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_community.retrievers import BM25Retriever
from rank_bm25 import BM25Okapi
@dataclass
class RAGConfig:
"""RAG 系统配置"""
# 嵌入模型
embedding_model: str = "bge-large-zh-v1.5"
embedding_dimension: int = 1024
# 大语言模型
llm_model: str = "gpt-4o-mini"
llm_temperature: float = 0.0
# 分块配置
chunk_size: int = 1000
chunk_overlap: int = 200
# 检索配置
retrieval_top_k: int = 20
final_top_k: int = 5
hybrid_alpha: float = 0.7 # 向量检索权重
# 置信度阈值
high_confidence_threshold: float = 0.85
medium_confidence_threshold: float = 0.60
class CompleteRAGPipeline:
"""
完整的 RAG 处理流水线
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 文档加载 │ -> │ 文档分块 │ -> │ 嵌入向量化 │
└─────────────┘ └─────────────┘ └─────────────┘
│
↓
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 生成回答 │ <- │ 置信度路由 │ <- │ 混合检索 │ <- │ 向量存储 │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
"""
def __init__(self, config: RAGConfig):
self.config = config
# 初始化组件
self._init_embedding()
self._init_llm()
self._init_chunkers()
# 存储组件
self.vector_store: Optional[Chroma] = None
self.bm25_retriever: Optional[BM25Retriever] = None
self.docs: List[Any] = []
def _init_embedding(self):
"""初始化嵌入模型"""
self.embeddings = OpenAIEmbeddings(
model="text-embedding-3-large",
dimensions=self.config.embedding_dimension
)
def _init_llm(self):
"""初始化大语言模型"""
self.llm = ChatOpenAI(
model=self.config.llm_model,
temperature=self.config.llm_temperature
)
def _init_chunkers(self):
"""初始化分块器"""
self.chunker = DocumentChunker(
chunk_size=self.config.chunk_size,
chunk_overlap=self.config.chunk_overlap
)
def load_documents(self, paths: List[str]) -> List[Dict[str, Any]]:
"""
加载文档
支持格式：PDF, TXT, MD
"""
from langchain.schema import Document
all_docs = []
for path in paths:
p = Path(path)
if p.suffix == ".pdf":
loader = PyPDFLoader(str(p))
else:
loader = TextLoader(str(p))
pages = loader.load()
for page in pages:
all_docs.append({
"content": page.page_content,
"metadata": {
**page.metadata,
"source": str(p)
}
})
self.docs = all_docs
return all_docs
def process_documents(self) -> List[Dict[str, Any]]:
"""
处理文档：分块 + 嵌入
Returns:
处理后的块列表
"""
if not self.docs:
raise ValueError("请先调用 load_documents 加载文档")
# 分块
chunks = self.chunker.chunk_documents(self.docs)
# 准备向量数据库数据
texts = [chunk["content"] for chunk in chunks]
metadatas = [chunk["metadata"] for chunk in chunks]
# 创建向量存储
self.vector_store = Chroma.from_texts(
texts=texts,
embedding=self.embeddings,
metadatas=metadatas,
persist_directory="./chroma_db"
)
# 创建 BM25 索引
tokenized_corpus = [text.split() for text in texts]
self.bm25 = BM25Okapi(tokenized_corpus)
self.bm25_corpus = texts
return chunks
def query(self, question: str) -> Dict[str, Any]:
"""
查询入口
Args:
question: 用户问题
Returns:
包含答案和来源的响应
"""
if not self.vector_store:
raise ValueError("请先调用 process_documents 处理文档")
# 创建混合检索器
retriever = HybridRetriever(
vector_store=self.vector_store,
bm25_index=self,
embed_model=self.embeddings,
top_k=self.config.retrieval_top_k,
final_k=self.config.final_top_k,
vector_weight=self.config.hybrid_alpha,
keyword_weight=1 - self.config.hybrid_alpha
)
# 创建上下文管理器
context_manager = ContextWindowManager(
model_name=self.config.llm_model
)
# 创建路由器
router = ThreeTierResponseRouter(
retriever=retriever,
llm=self.llm,
context_manager=context_manager,
high_threshold=self.config.high_confidence_threshold,
medium_threshold=self.config.medium_confidence_threshold
)
# 获取回答
return router.answer(question)
# BM25 接口适配
def search(self, query: str, k: int = 10) -> List[Dict[str, Any]]:
"""BM25 搜索接口"""
tokenized_query = query.split()
scores = self.bm25.get_scores(tokenized_query)
top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:k]
return [
{
"id": idx,
"content": self.bm25_corpus[idx],
"score": scores[idx]
}
for idx in top_indices
]
# 使用示例
if __name__ == "__main__":
# 加载配置
config = RAGConfig()
# 初始化流水线
rag = CompleteRAGPipeline(config)
# 加载并处理文档
rag.load_documents(["./docs/faq.pdf"])
chunks = rag.process_documents()
print(f"处理完成，共 {len(chunks)} 个块")
# 查询
result = rag.query("产品的退换货政策是什么？")
print(f"置信度: {result['confidence']}")
print(f"回答: {result['answer']}")七、评估体系：Ragas 指标
# Ragas 评估指标配置
ragas_config:
metrics:
# 忠实度（Faithfulness）：生成答案与上下文的吻合程度
faithfulness:
enabled: true
threshold: 0.8
# 答案相关性（Answer Relevancy）：答案与问题的相关程度
answer_relevancy:
enabled: true
threshold: 0.75
# 上下文召回率（Context Recall）：检索到的上下文覆盖正确答案的比例
context_recall:
enabled: true
threshold: 0.85
# 上下文精准度（Context Precision）：检索到的上下文的相关性排序质量
context_precision:
enabled: true
threshold: 0.80
evaluation_sets:
# 测试集路径
test_data: "./data/eval_testset.json"
# 自动生成测试集配置
auto_generate:
enabled: true
sample_count: 50
llm_model: "gpt-4o-mini"八、总结
RAG 工程化是一个系统工程，不是简单的"接个向量库"就能完成的。从数据提取到效果评估，每个环节都需要精心设计和持续优化。
核心要点回顾：
数据质量是天花板：脏数据再好的算法也救不回来
分块策略影响检索效果：递归分块适合通用场景，语义分块适合结构化文档
嵌入模型决定检索下限：中文场景务必使用 BGE 等中文优化模型
混合检索 + 重排序是黄金组合：弥补单一检索方式的不足
三级应答机制提升用户体验：精准路由，避免"答不上来硬答"
评估体系是生产级系统的标配：没有量化指标就无法优化
技术选型建议：
RAG 的竞争，本质上是工程化体系的竞争。掌握完整的 RAG 技术栈，理解每一层的核心逻辑与选型策略，是 AI 时代开发者的核心竞争力。
评论区