梦栈
30.71M · 2026-03-13
RAG(Retrieval-Augmented Generation,检索增强生成)是一种让大语言模型能够访问外部知识库的技术,通过在生成答案前先检索相关文档,从根本上解决了大模型的三大痛点。
用户:今天的新闻头条是什么?
纯LLM回答:抱歉,我的知识截止到2024年1月,无法回答当前新闻...
问题:大模型在训练完成后,知识就"冻结"了,无法获取最新信息。
用户:我们公司的休假政策是什么?
纯LLM回答:根据劳动法规定,员工每年享有5天带薪年假...
(实际上:你们公司可能是10天,模型是"猜"的)
问题:当模型不知道答案时,它会基于统计规律"编造"一个听起来合理的答案。
用户:帮我分析一下我们公司2024年Q3的销售数据
纯LLM回答:抱歉,我无法访问您公司的内部数据...
问题:私有数据、专业领域知识无法被纳入预训练语料。
RAG的核心思想非常直观:把相关信息直接"塞"到Prompt里。
用户问题:我们公司的休假政策是什么?
步骤1:检索(Retrieval)
→ 从公司知识库检索相关文档
→ 找到:"员工手册第5章:每位员工享有10天带薪年假"
步骤2:增强(Augmentation)
→ 把检索到的文档加入Prompt
→ 构造增强后的Prompt:
"""
参考文档:员工手册第5章:每位员工享有10天带薪年假
问题:我们公司的休假政策是什么?
请基于参考文档回答。
"""
步骤3:生成(Generation)
→ 大模型基于增强后的Prompt生成答案
→ 回答:"根据员工手册第5章,每位员工享有10天带薪年假"
优势:
RAG系统通常包含四个核心组件:
┌─────────────────────────────────────────────────┐
│ 用户问题 │
│ "明天北京天气怎么样?" │
└───────────────────┬─────────────────────────────┘
↓
┌───────────────────────────────────────────────────┐
│ 1. Query重写与理解(Query Rewriter) │
│ - 改写问题使其更适合检索 │
│ - 提取关键词 │
│ - 生成多个查询变体 │
└───────────────────┬───────────────────────────────┘
↓
┌───────────────────────────────────────────────────┐
│ 2. 检索器(Retriever) │
│ - 从向量数据库检索相关文档 │
│ - 返回Top-K个最相关片段 │
└───────────────────┬───────────────────────────────┘
↓
┌───────────────────────────────────────────────────┐
│ 3. 重排序器(Reranker) │
│ - 对检索结果进行精细排序 │
│ - 过滤掉不相关的文档 │
└───────────────────┬───────────────────────────────┘
↓
┌───────────────────────────────────────────────────┐
│ 4. 生成器(Generator) │
│ - 大语言模型基于检索到的文档生成答案 │
│ - 标注引用来源 │
└───────────────────┬───────────────────────────────┘
↓
┌─────────┐
│ 答案 │
└─────────┘
在RAG系统运行之前,需要先构建知识库。
大文档需要切分成小片段,原因:
常见分块策略:
def fixed_size_chunking(text, chunk_size=512, overlap=50):
"""
按固定字符数分块
参数:
- chunk_size: 每块大小(字符数)
- overlap: 块之间的重叠(避免语义被切断)
"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap # 重叠部分
return chunks
# 示例
text = "人工智能是...(省略5000字)"
chunks = fixed_size_chunking(text, chunk_size=512, overlap=50)
# 结果:10个chunk,每个约512字符,相邻chunk重叠50字符
优点:简单、快速 缺点:可能在句子中间切断,破坏语义
def semantic_chunking(text, max_chunk_size=512):
"""
按语义单元分块(段落、句子)
"""
paragraphs = text.split('nn') # 按段落分
chunks = []
current_chunk = ""
for para in paragraphs:
if len(current_chunk) + len(para) <= max_chunk_size:
current_chunk += para + "nn"
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para + "nn"
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
优点:保持语义完整性 缺点:可能产生大小不均的chunk
# 按优先级尝试不同的分隔符
separators = ["nn", "n", "。", ".", " "]
# 先按段落分,如果段落太大再按句子分,如此递归
优点:平衡了大小和语义 缺点:实现较复杂
将文本chunk转换为数值向量,使计算机能够"理解"语义相似度。
| 模型 | 维度 | 语言 | 适用场景 |
|---|---|---|---|
| OpenAI text-embedding-3-small | 1536 | 多语言 | 通用场景,性价比高 |
| OpenAI text-embedding-3-large | 3072 | 多语言 | 高精度需求 |
| bge-large-zh | 1024 | 中文优化 | 中文场景首选 |
| bge-m3 | 1024 | 多语言 | 多语言混合场景 |
| Cohere embed-multilingual | 768 | 100+语言 | 全球化产品 |
# 示例:将文本转为向量
chunk1 = "今天天气很好"
chunk2 = "今日天气不错"
chunk3 = "量子计算机的原理"
embedding1 = embedding_model.encode(chunk1)
# → [0.23, 0.45, -0.12, ..., 0.78] (1024维向量)
embedding2 = embedding_model.encode(chunk2)
# → [0.25, 0.43, -0.10, ..., 0.76] (语义相近,向量也相近)
embedding3 = embedding_model.encode(chunk3)
# → [-0.54, 0.12, 0.89, ..., -0.32] (语义不同,向量差异大)
语义相似度计算:
这就是余弦相似度(Cosine Similarity),值范围为[-1, 1]:
# 计算相似度
similarity(embedding1, embedding2) = 0.95 # 很相似
similarity(embedding1, embedding3) = 0.12 # 不相关
存储和检索向量的专用数据库。
主流向量数据库对比:
| 数据库 | 类型 | 特点 | 适用场景 |
|---|---|---|---|
| Pinecone | 云服务 | 全托管,性能强 | 快速上线,不想管理基础设施 |
| Weaviate | 开源/云 | 功能全面,支持混合搜索 | 需要高级功能(过滤、图谱) |
| Milvus | 开源 | 性能强,规模大 | 大规模数据(百万级以上) |
| Chroma | 开源 | 轻量级,易上手 | 小项目、快速原型 |
| Qdrant | 开源/云 | Rust实现,高性能 | 注重性能和可靠性 |
| Faiss | 开源库 | Meta开发,算法多 | 学术研究、自建系统 |
向量数据库的核心操作:
# 1. 创建索引
import chromadb
client = chromadb.Client()
collection = client.create_collection(name="company_docs")
# 2. 插入向量
collection.add(
embeddings=[embedding1, embedding2, embedding3],
documents=[chunk1, chunk2, chunk3],
ids=["doc1", "doc2", "doc3"]
)
# 3. 检索(查询)
query = "今天天气如何"
query_embedding = embedding_model.encode(query)
results = collection.query(
query_embeddings=[query_embedding],
n_results=2 # 返回Top-2
)
# 返回:
# [
# {"id": "doc1", "document": "今天天气很好", "distance": 0.05},
# {"id": "doc2", "document": "今日天气不错", "distance": 0.07}
# ]
检索器负责从向量数据库中找到与用户问题最相关的文档片段。
原理:用Embedding模型将问题和文档都转为向量,通过向量相似度检索。
query = "什么是Transformer的注意力机制?"
query_embedding = embedding_model.encode(query)
# 在向量数据库中搜索
results = vector_db.search(query_embedding, top_k=5)
优点:
缺点:
原理:传统关键词搜索(BM25、TF-IDF)。
# BM25算法
from rank_bm25 import BM25Okapi
corpus = ["文档1", "文档2", "文档3"]
tokenized_corpus = [doc.split() for doc in corpus]
bm25 = BM25Okapi(tokenized_corpus)
query = "Transformer 注意力"
scores = bm25.get_scores(query.split())
# 返回每个文档的BM25分数
优点:
缺点:
原理:结合稠密和稀疏检索,取长补短。
# 1. 分别执行两种检索
dense_results = dense_retrieval(query, top_k=10)
sparse_results = sparse_retrieval(query, top_k=10)
# 2. 融合结果(Reciprocal Rank Fusion)
def reciprocal_rank_fusion(results1, results2, k=60):
scores = {}
for rank, doc_id in enumerate(results1):
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
for rank, doc_id in enumerate(results2):
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
# 按分数排序
final_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return final_results[:5] # 返回Top-5
实际效果对比:
| 场景 | 稠密检索 | 稀疏检索 | 混合检索 |
|---|---|---|---|
| "什么是注意力机制?" | 90分 | ️ 70分 | 95分 |
| "Attention mechanism" | 85分 | ️ 65分 | 92分 |
| "产品ID: A1234-B56" | ️ 40分 | 95分 | 98分 |
| "如何优化推理速度" | 88分 | ️ 72分 | 93分 |
结论:混合检索在大多数场景下表现最佳。
用户的原始问题往往不是最佳检索查询,需要改写优化。
思路:让大模型先生成一个"假想答案",用这个答案去检索。
# 用户问题
query = "如何优化Transformer的推理速度?"
# Step 1: 生成假想答案
hypothetical_answer = llm.generate(
f"请详细回答:{query}"
)
# 输出:"优化Transformer推理速度的方法包括:
# 1. 使用KV Cache减少重复计算
# 2. 量化模型参数到INT8或INT4
# 3. 采用FlashAttention算法..."
# Step 2: 用假想答案去检索(而不是原问题)
results = vector_db.search(
embedding_model.encode(hypothetical_answer),
top_k=5
)
为什么有效?
思路:生成多个变体查询,分别检索后合并结果。
# 原问题
query = "Transformer如何处理长文本?"
# 生成多个变体
queries = llm.generate(
f"为以下问题生成3个不同角度的变体:{query}"
)
# 输出:
# 1. "Transformer模型在处理长序列时会遇到什么问题?"
# 2. "如何扩展Transformer的上下文窗口长度?"
# 3. "哪些技术可以让Transformer支持更长的输入?"
# 分别检索
all_results = []
for q in queries:
results = vector_db.search(embedding_model.encode(q), top_k=3)
all_results.extend(results)
# 去重合并
final_results = deduplicate_and_rank(all_results, top_k=5)
思路:添加同义词、相关术语。
query = "AI Agent"
# 扩展同义词
expanded_query = query + " 智能体 autonomous agent ReAct"
# 用扩展后的查询检索
results = vector_db.search(embedding_model.encode(expanded_query), top_k=5)
检索器返回的Top-K结果可能不够精确,重排序器对结果进行二次排序。
检索器的局限:
重排序器的优势:
Bi-Encoder(检索器常用):
Query → Encoder → Query Vector
→ 计算相似度
Doc → Encoder → Doc Vector
Cross-Encoder(重排序器常用):
[Query + Doc] → Encoder → 相关性分数
from sentence_transformers import CrossEncoder
# 1. 检索阶段(Bi-Encoder,快速)
retrieval_results = vector_db.search(query_embedding, top_k=100)
# 2. 重排序阶段(Cross-Encoder,精确)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-12-v2')
# 计算每个文档与问题的相关性分数
pairs = [[query, doc] for doc in retrieval_results]
scores = reranker.predict(pairs)
# 按分数重新排序
reranked_results = sorted(
zip(retrieval_results, scores),
key=lambda x: x[1],
reverse=True
)[:5] # 取Top-5
| 模型 | 大小 | 语言 | 性能 | 速度 |
|---|---|---|---|---|
| bge-reranker-large | 560M | 中英文 | ⭐⭐⭐⭐⭐ | 中等 |
| bge-reranker-base | 279M | 中英文 | ⭐⭐⭐⭐ | 快 |
| cohere-rerank-multilingual | API | 100+语言 | ⭐⭐⭐⭐⭐ | API |
| cross-encoder/ms-marco-MiniLM | 66M | 英文 | ⭐⭐⭐ | 很快 |
效果提升示例:
原始检索Top-5(只看向量相似度):
1. 相关性:60%
2. 相关性:55%
3. 相关性:50%
4. 相关性:95% ← 最相关的排在第4!
5. 相关性:45%
重排序后Top-5:
1. 相关性:95% ← 正确排序
2. 相关性:60%
3. 相关性:55%
4. 相关性:50%
5. 相关性:45%
生成器是大语言模型,负责基于检索到的文档生成最终答案。
基础RAG Prompt模板:
prompt_template = """
你是一个专业的AI助手。请根据以下参考文档回答用户问题。
【重要规则】
1. 答案必须基于参考文档,不要编造信息
2. 如果文档中没有相关信息,请明确说明"根据提供的文档无法回答"
3. 引用文档时请标注来源
参考文档:
{retrieved_documents}
用户问题:
{user_question}
请给出答案:
"""
示例:
参考文档:
[文档1] 员工手册第5章:每位员工享有10天带薪年假,工作满5年后增加到15天。
[文档2] 请假流程:需提前3天向直属主管提交申请,经批准后生效。
用户问题:
我工作2年了,有多少天年假?
AI回答:
根据员工手册第5章,每位员工享有10天带薪年假。由于您工作了2年,
尚未满足"工作满5年"的条件,因此您当前享有10天带薪年假。
如需请假,请参考请假流程:提前3天向直属主管提交申请。
【来源】员工手册第5章、请假流程文档
prompt = """
参考文档:{documents}
用户问题:{question}
请按以下步骤回答:
1. 首先,确定问题的关键点
2. 然后,从文档中找到相关信息
3. 最后,综合信息给出答案
你的回答:
"""
prompt = """
你需要综合以下多个文档的信息来回答问题。
文档A:{doc_a}
文档B:{doc_b}
文档C:{doc_c}
问题:{question}
请注意:
- 如果文档之间有冲突,请指出并说明
- 如果需要综合多个文档,请明确说明
"""
prompt = """
参考文档:{documents}
问题:{question}
回答要求:
- 如果文档中有明确答案,给出确定的回答
- 如果文档中信息不完整,说明"部分信息缺失"
- 如果文档完全无关,说明"无法根据提供的文档回答"
你的回答:
"""
让模型标注答案来源,提高可信度。
prompt = """
参考文档:
[1] 文档标题A:内容...
[2] 文档标题B:内容...
[3] 文档标题C:内容...
问题:{question}
回答格式要求:
答案:【你的答案】
来源:[引用的文档编号]
示例:
答案:根据公司政策,员工享有10天年假。
来源:[1]
"""
问题:用户问题中可能包含元数据过滤条件。
用户问题:"2023年Q4的销售报告"
传统RAG:直接检索 → 可能返回2022年、2024年的报告
自查询RAG:
1. 提取过滤条件:year=2023, quarter=Q4, type=销售报告
2. 在向量检索时应用过滤器
3. 只在符合条件的文档中检索
实现:
# 1. 让LLM提取元数据
extraction_prompt = f"""
从以下问题中提取结构化信息:
问题:{user_question}
输出JSON格式:
{{
"semantic_query": "语义查询部分",
"filters": {{
"year": ...,
"quarter": ...,
"type": ...
}}
}}
"""
metadata = llm.extract(extraction_prompt)
# 2. 应用过滤器检索
results = vector_db.search(
query_embedding=embedding_model.encode(metadata["semantic_query"]),
filters=metadata["filters"],
top_k=5
)
问题:有些文档有层次结构(章节、段落)。
方案:
# 层级1:检索相关文档
doc_results = vector_db.search_documents(query_embedding, top_k=3)
# 层级2:在选中的文档内检索段落
paragraph_results = []
for doc in doc_results:
paras = vector_db.search_paragraphs(
query_embedding,
document_id=doc.id,
top_k=2
)
paragraph_results.extend(paras)
问题:新信息应该比旧信息更重要。
import datetime
def time_weighted_score(similarity_score, document_date):
"""
结合相似度和时间新鲜度的混合评分
"""
days_old = (datetime.now() - document_date).days
time_decay = math.exp(-days_old / 365) # 一年衰减到约37%
final_score = 0.7 * similarity_score + 0.3 * time_decay
return final_score
问题:有些问题需要多次检索才能回答。
问题:"GPT-4的作者之前发表过哪些著名论文?"
步骤1:检索 "GPT-4的作者" → 找到 "OpenAI团队,主要贡献者包括..."
步骤2:检索 "某某研究员的论文" → 找到论文列表
步骤3:综合回答
实现(结合ReAct Agent):
def multi_hop_rag(question):
thoughts = []
context = []
# 迭代式检索
for step in range(max_steps):
# 生成下一步的查询
next_query = llm.generate(
f"问题:{question}n已知信息:{context}n下一步应该查询什么?"
)
# 检索
results = vector_db.search(next_query)
context.append(results)
# 判断是否足够回答
is_sufficient = llm.check(
f"问题:{question}n已知:{context}n信息是否足够?"
)
if is_sufficient:
break
# 生成最终答案
answer = llm.generate(
f"问题:{question}n参考信息:{context}n请给出答案:"
)
return answer
如何衡量RAG系统的好坏?
指标:
# 示例:计算Recall@5
def recall_at_k(retrieved_docs, relevant_docs, k=5):
"""
retrieved_docs: 检索到的文档ID列表(按相关性排序)
relevant_docs: 真正相关的文档ID集合
"""
retrieved_top_k = set(retrieved_docs[:k])
relevant_set = set(relevant_docs)
overlap = retrieved_top_k & relevant_set
recall = len(overlap) / len(relevant_set)
return recall
# 示例
retrieved = ["doc1", "doc3", "doc7", "doc2", "doc9"]
relevant = ["doc1", "doc2", "doc5"]
recall = recall_at_k(retrieved, relevant, k=5)
# 结果:2/3 = 0.67(检索到了doc1和doc2,漏了doc5)
指标:
自动评估(使用LLM作为评判):
def evaluate_faithfulness(answer, retrieved_docs):
"""
评估答案是否忠实于文档
"""
eval_prompt = f"""
参考文档:{retrieved_docs}
AI回答:{answer}
问题:AI的回答是否完全基于参考文档,没有编造信息?
评分标准:
1分:回答包含文档中没有的信息
2分:回答大部分基于文档,但有少量推测
3分:回答完全基于文档
给出评分(1-3)和理由:
"""
score = llm.evaluate(eval_prompt)
return score
指标:
RAGAS(RAG Assessment)是专门用于评估RAG系统的框架。
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_relevancy,
context_recall
)
# 准备评估数据
data = {
"question": ["问题1", "问题2", ...],
"answer": ["AI回答1", "AI回答2", ...],
"contexts": [["检索doc1", "检索doc2"], ...],
"ground_truth": ["正确答案1", "正确答案2", ...]
}
# 运行评估
result = evaluate(
dataset=data,
metrics=[
faithfulness,
answer_relevancy,
context_relevancy,
context_recall
]
)
print(result)
# 输出:
# {
# "faithfulness": 0.92,
# "answer_relevancy": 0.88,
# "context_relevancy": 0.85,
# "context_recall": 0.79
# }
什么时候用RAG,什么时候用微调?
| 对比维度 | RAG | 微调(Fine-tuning) |
|---|---|---|
| 适用场景 | 需要实时更新的知识 | 需要改变模型行为/风格 |
| 成本 | 低(无需训练) | 高(需要GPU训练) |
| 更新速度 | 实时(更新知识库即可) | 慢(需重新训练) |
| 知识准确性 | 高(来自真实文档) | 中(可能产生幻觉) |
| 响应速度 | 慢(需要检索) | 快(直接生成) |
| 适用知识类型 | 事实性知识、文档内容 | 任务模式、领域语言风格 |
| 可解释性 | 强(可追溯来源) | 弱(黑盒) |
实际案例:
| 任务 | 推荐方案 | 原因 |
|---|---|---|
| 企业知识库问答 | RAG | 文档频繁更新,需要引用来源 |
| 客服对话风格 | 微调 | 需要学习特定的对话模式 |
| 医疗诊断辅助 | RAG | 需要基于最新医学文献 |
| 代码生成 | 微调 | 需要学习特定编程风格 |
| 法律文档分析 | RAG + 微调 | 结合:微调学习法律术语,RAG检索具体案例 |
最佳实践:RAG + 微调组合
1. 微调:让模型学习领域知识和对话风格
2. RAG:提供具体、最新的事实信息
3. 结果:既有专业性,又有准确性
可能原因:
解决方案:
可能原因:
解决方案:
可能原因:
解决方案:
瓶颈分析:
| 步骤 | 耗时占比 | 优化方案 |
|---|---|---|
| Embedding查询 | 5-10% | 使用小模型(384维vs1024维) |
| 向量检索 | 10-20% | 优化索引(HNSW、IVF) |
| 重排序 | 20-30% | 只重排Top-10而非Top-100 |
| LLM生成 | 40-60% | 使用更快的模型(Haiku vs Sonnet) |
优化示例:
# 优化前:2000ms
results = vector_db.search(query_emb, top_k=100) # 200ms
reranked = reranker.rerank(results) # 500ms
answer = llm.generate(prompt) # 1300ms
# 优化后:800ms
results = vector_db.search(query_emb, top_k=20) # 100ms
reranked = reranker.rerank(results[:10]) # 150ms
answer = faster_llm.generate(prompt) # 550ms
成本分解:
单次RAG查询成本:
- Embedding(query):$0.0001
- 向量数据库查询:$0.0001
- Reranking:$0.0001
- LLM生成(2K tokens):$0.01
----------------------------------------
总计:约 $0.01 / 次
优化策略:
# 智能路由示例
def should_use_rag(question):
"""
判断问题是否需要RAG
"""
# 简单问题直接回答
if is_simple_question(question): # 如"你好"
return False
# 通用知识问题,模型内部知识足够
if is_general_knowledge(question): # 如"什么是Python"
return False
# 需要特定/实时信息,使用RAG
return True
传统RAG局限:只能基于文本相似度检索,无法理解实体关系。
Graph RAG:构建知识图谱,理解实体关系。
问题:"北京和上海之间有直达高铁吗?"
传统RAG:检索包含"北京"和"上海"的文档
Graph RAG:
1. 识别实体:北京(城市)、上海(城市)
2. 查询关系:(北京)-[直达高铁]->(上海)
3. 返回:是,京沪高铁
结合Agent能力:让RAG系统能主动规划、多轮检索。
# 传统RAG:一次检索
results = retrieve(question)
answer = generate(results)
# Agentic RAG:多轮交互
agent = RAGAgent()
while not agent.is_done():
thought = agent.think() # 规划下一步
action = agent.act() # 检索或生成
observation = agent.observe() # 评估结果
不仅检索文本,还检索图片、视频、音频。
问题:"这个产品的外观是什么样的?"
检索结果:
- 文本:产品描述文档
- 图片:产品照片
- 视频:产品展示视频
多模态LLM综合生成答案
根据用户历史和偏好个性化检索。
# 考虑用户上下文
user_context = {
"role": "前端工程师",
"history": ["之前问过React相关问题"],
"preference": "喜欢代码示例"
}
# 个性化检索
results = retrieve(
question=question,
user_context=user_context,
boost_fields=["code_examples"] # 提升代码示例权重
)
RAG(检索增强生成)通过"检索相关文档 + 增强Prompt"的方式,让大模型能够:
核心组件:
关键技术:
实践建议:
RAG已成为大模型应用的标配技术,掌握RAG是构建生产级AI应用的关键能力!