引言
在大语言模型（LLM）蓬勃发展的今天，如何让AI更准确地回答特定领域的问题成为了一个关键挑战。RAG（Retrieval-Augmented Generation，检索增强生成）技术应运而生，它通过结合外部知识库和生成模型，显著提升了AI回答的准确性和时效性。
本文将深入探讨RAG的核心原理，重点解析向量检索和上下文注入两大关键技术，并提供实践指导。
一、RAG是什么？
1.1 核心思想
RAG的核心思想非常直观：在生成答案之前，先从知识库中检索相关信息，然后将这些信息作为上下文提供给大语言模型，让模型基于这些"参考资料"来生成更准确的回答。
这就像是让AI在开卷考试而不是闭卷考试——它可以查阅资料后再作答。
1.2 为什么需要RAG？
传统LLM面临几个关键问题：
- 知识时效性：模型的知识截止于训练时间，无法获取最新信息
- 幻觉问题：模型可能生成看似合理但实际错误的内容
- 专业领域知识不足：通用模型对特定领域的深度知识有限
- 成本问题：频繁微调大模型成本高昂
RAG通过外部知识检索优雅地解决了这些问题，无需重新训练模型。
二、向量检索：RAG的核心引擎
2.1 什么是向量检索？
向量检索是RAG系统的第一步，也是最关键的一步。它的任务是从海量文档中快速找出与用户问题最相关的内容。
文本向量化
文本向量化（Embedding）是将文本转换为高维向量的过程：
"什么是机器学习？" → [0.12, -0.34, 0.56, ..., 0.89] # 维度通常为384-1536
向量的特点：
- 语义相似的文本，向量距离更近
- 向量可以进行数学运算（相似度计算）
- 降维后可视化（理解语义空间）
常用的Embedding模型
- OpenAI text-embedding-3-small/large：性能强大，支持多语言
- sentence-transformers：开源方案，适合中文
- BGE系列：国内优秀的开源模型
- m3e：专门针对中文优化
2.2 向量检索的工作流程
用户问题 → Embedding模型 → 查询向量 → 向量数据库 → Top-K 相似文档
步骤详解：
文档预处理：
- 文档切片（Chunking）：将长文档分割成适当大小的片段（通常300-1000 tokens）
- 向量化：使用Embedding模型将每个片段转换为向量
- 存储：将向量及元数据存入向量数据库
查询处理：
- 用户问题同样经过Embedding模型转换为查询向量
- 在向量数据库中进行相似度搜索
- 返回Top-K个最相关的文档片段
2.3 相似度计算方法
余弦相似度（最常用）
import numpy as np
def cosine_similarity(vec1, vec2):
"""计算两个向量的余弦相似度"""
dot_product = np.dot(vec1, vec2)
norm_product = np.linalg.norm(vec1) * np.linalg.norm(vec2)
return dot_product / norm_product
# 示例
query_vec = np.array([0.5, 0.3, 0.8])
doc_vec = np.array([0.6, 0.2, 0.9])
similarity = cosine_similarity(query_vec, doc_vec)
print(f"相似度: {similarity:.3f}") # 输出：0.989
优点：不受向量长度影响，只关注方向
欧氏距离
def euclidean_distance(vec1, vec2):
"""计算欧氏距离（距离越小越相似）"""
return np.linalg.norm(vec1 - vec2)
点积
def dot_product_similarity(vec1, vec2):
"""点积相似度"""
return np.dot(vec1, vec2)
2.4 向量数据库选择
| 数据库 | 特点 | 适用场景 |
|---|
| Pinecone | 云服务，易用性强 | 快速原型开发 |
| Milvus | 开源，性能强大 | 大规模生产环境 |
| Weaviate | 支持多模态 | 复杂查询需求 |
| Chroma | 轻量级，易部署 | 小型项目、本地开发 |
| FAISS | Facebook开源，速度快 | 研究和实验 |
2.5 优化向量检索的技巧
技巧1：混合检索（Hybrid Search）
结合关键词检索和向量检索：
# 伪代码示例
def hybrid_search(query, alpha=0.5):
# 向量检索得分
vector_results = vector_search(query)
# 关键词检索得分（BM25）
keyword_results = bm25_search(query)
# 加权融合
final_scores = alpha * vector_results + (1-alpha) * keyword_results
return top_k(final_scores)
技巧2：重排序（Reranking）
使用更强大的模型对初步检索结果重新排序：
def rerank(query, initial_results):
"""使用交叉编码器重排序"""
cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
pairs = [(query, doc) for doc in initial_results]
scores = cross_encoder.predict(pairs)
# 按新得分重新排序
return sort_by_scores(initial_results, scores)
技巧3：查询扩展
扩展用户查询以提高召回率：
def query_expansion(query):
"""生成查询的多个变体"""
expanded_queries = [
query,
f"关于{query}的详细解释",
f"{query}是什么意思",
f"如何理解{query}"
]
return expanded_queries
三、上下文注入：让LLM"看见"外部知识
3.1 上下文注入的原理
上下文注入是将检索到的文档作为提示（Prompt）的一部分，提供给LLM。这个过程就像给AI提供"参考资料"。
基本结构
系统指令 + 检索到的上下文 + 用户问题 → LLM → 生成答案
3.2 Prompt工程最佳实践
模板示例1：基础RAG Prompt
def create_rag_prompt(query, context_docs):
prompt = f"""你是一个专业的AI助手。请基于以下参考资料回答用户的问题。
参考资料：
{format_context(context_docs)}
重要提示：
1. 只基于上述参考资料回答问题
2. 如果参考资料中没有相关信息，请明确说明
3. 引用参考资料时请注明来源
用户问题：{query}
请提供准确、详细的回答："""
return prompt
def format_context(docs):
"""格式化上下文文档"""
formatted = []
for i, doc in enumerate(docs, 1):
formatted.append(f"[文档{i}]\n{doc['content']}\n来源：{doc['source']}\n")
return "\n".join(formatted)
模板示例2：带引用的高级Prompt
def create_advanced_rag_prompt(query, context_docs):
prompt = f"""# 角色
你是一个严谨的知识问答助手。
# 任务
基于提供的参考资料回答用户问题，并标注信息来源。
# 参考资料
{format_numbered_context(context_docs)}
# 回答要求
1. **准确性**：确保答案完全基于参考资料
2. **引用标注**：使用[1][2]标注信息来源
3. **完整性**：综合所有相关资料给出全面回答
4. **诚实性**：如果资料不足，明确说明局限性
# 用户问题
{query}
# 你的回答
"""
return prompt
def format_numbered_context(docs):
"""带编号的上下文格式化"""
formatted = []
for i, doc in enumerate(docs, 1):
formatted.append(f"[{i}] {doc['content']}\n(来源: {doc['source']})\n")
return "\n".join(formatted)
3.3 上下文窗口管理
问题：上下文过长
当检索到的文档过多或过长时，可能超出LLM的上下文窗口限制。
解决方案
方案1：智能截断
def truncate_context(docs, max_tokens=2000):
"""智能截断上下文"""
truncated = []
current_tokens = 0
for doc in docs:
doc_tokens = count_tokens(doc['content'])
if current_tokens + doc_tokens <= max_tokens:
truncated.append(doc)
current_tokens += doc_tokens
else:
# 截断最后一个文档
remaining = max_tokens - current_tokens
doc['content'] = truncate_to_tokens(doc['content'], remaining)
truncated.append(doc)
break
return truncated
方案2：分层检索
def hierarchical_retrieval(query, k1=10, k2=3):
"""两阶段检索：先召回，再精选"""
# 第一阶段：快速召回更多文档
candidates = vector_search(query, top_k=k1)
# 第二阶段：使用更强模型精选最相关的
final_docs = rerank(query, candidates, top_k=k2)
return final_docs
方案3：文档摘要
async def summarize_docs(docs, llm):
"""对长文档进行摘要"""
summaries = []
for doc in docs:
if len(doc['content']) > 1000:
summary = await llm.summarize(doc['content'])
doc['content'] = summary
summaries.append(doc)
return summaries
3.4 上下文质量优化
技巧1：去重
def deduplicate_docs(docs, similarity_threshold=0.9):
"""移除相似度过高的重复文档"""
unique_docs = []
for doc in docs:
is_duplicate = False
for existing in unique_docs:
if cosine_similarity(doc['embedding'], existing['embedding']) > similarity_threshold:
is_duplicate = True
break
if not is_duplicate:
unique_docs.append(doc)
return unique_docs
技巧2：相关性过滤
def filter_by_relevance(docs, min_score=0.7):
"""过滤掉相关性低的文档"""
return [doc for doc in docs if doc['score'] >= min_score]
技巧3：多样性采样
def diversify_results(docs, top_k=5):
"""确保结果的多样性"""
selected = [docs[0]] # 选择最相关的
for doc in docs[1:]:
if len(selected) >= top_k:
break
# 计算与已选文档的最大相似度
max_sim = max([cosine_similarity(doc['embedding'], s['embedding'])
for s in selected])
# 如果不太相似，则添加
if max_sim < 0.85:
selected.append(doc)
return selected
四、完整RAG系统实现
4.1 系统架构
┌─────────────┐
│ 用户查询 │
└──────┬──────┘
│
▼
┌─────────────────┐
│ 查询处理模块 │ ← 查询改写、扩展
└──────┬──────────┘
│
▼
┌─────────────────┐
│ 向量检索引擎 │ ← 向量数据库
└──────┬──────────┘
│
▼
┌─────────────────┐
│ 重排序模块 │ ← 提高精确度
└──────┬──────────┘
│
▼
┌─────────────────┐
│ 上下文构建 │ ← Prompt工程
└──────┬──────────┘
│
▼
┌─────────────────┐
│ LLM生成 │ ← 生成答案
└──────┬──────────┘
│
▼
┌─────────────────┐
│ 后处理与验证 │ ← 事实检查
└──────┬──────────┘
│
▼
┌─────────────────┐
│ 返回结果 │
└─────────────────┘
4.2 Python实现示例
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA
class RAGSystem:
def __init__(self, documents):
"""初始化RAG系统"""
# 1. 文档处理
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?"]
)
# 2. Embedding模型
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 3. 向量数据库
self.vectorstore = self._build_vectorstore(documents)
# 4. LLM
self.llm = OpenAI(temperature=0)
# 5. 检索器
self.retriever = self.vectorstore.as_retriever(
search_type="mmr", # 最大边际相关性
search_kwargs={
"k": 4,
"fetch_k": 20,
"lambda_mult": 0.5
}
)
def _build_vectorstore(self, documents):
"""构建向量存储"""
# 切分文档
chunks = self.text_splitter.split_documents(documents)
# 创建向量数据库
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
persist_directory="./chroma_db"
)
return vectorstore
def query(self, question):
"""执行RAG查询"""
# 创建问答链
qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff",
retriever=self.retriever,
return_source_documents=True,
chain_type_kwargs={
"prompt": self._create_prompt()
}
)
# 执行查询
result = qa_chain({"query": question})
return {
"answer": result["result"],
"sources": result["source_documents"]
}
def _create_prompt(self):
"""创建Prompt模板"""
from langchain.prompts import PromptTemplate
template = """基于以下参考资料回答问题。如果资料中没有答案，请说"我不知道"。
参考资料：
{context}
问题：{question}
详细回答："""
return PromptTemplate(
template=template,
input_variables=["context", "question"]
)
# 使用示例
from langchain.document_loaders import TextLoader
# 加载文档
loader = TextLoader("knowledge_base.txt")
documents = loader.load()
# 创建RAG系统
rag = RAGSystem(documents)
# 查询
result = rag.query("什么是机器学习？")
print(f"回答：{result['answer']}")
print(f"参考文档数量：{len(result['sources'])}")
4.3 高级优化：多查询RAG
class AdvancedRAG:
def multi_query_retrieval(self, question):
"""生成多个查询角度"""
# 使用LLM生成问题的不同表述
variations = self.llm.generate_variations(question, num=3)
all_docs = []
for variation in variations:
docs = self.retriever.get_relevant_documents(variation)
all_docs.extend(docs)
# 去重和排序
unique_docs = self.deduplicate(all_docs)
ranked_docs = self.rerank(question, unique_docs)
return ranked_docs[:5]
def self_query_with_metadata(self, question):
"""基于元数据的自查询"""
# 从问题中提取过滤条件
metadata_filter = self.extract_metadata_filter(question)
# 在向量搜索中应用过滤
docs = self.vectorstore.similarity_search(
question,
filter=metadata_filter,
k=5
)
return docs
五、实践案例与应用场景
5.1 企业知识库问答
场景：企业内部有大量文档（产品手册、政策文档、FAQ等）
实现要点：
# 示例：企业知识库RAG
class EnterpriseRAG:
def __init__(self):
self.vectorstore = Chroma(
collection_name="company_docs",
embedding_function=embeddings
)
def add_document(self, doc, metadata):
"""添加文档并包含元数据"""
chunks = self.split_document(doc)
for chunk in chunks:
self.vectorstore.add_texts(
texts=[chunk],
metadatas=[{
"department": metadata["department"],
"doc_type": metadata["doc_type"],
"last_updated": metadata["date"],
"access_level": metadata["access_level"]
}]
)
def query_with_access_control(self, question, user_level):
"""带权限控制的查询"""
results = self.vectorstore.similarity_search(
question,
filter={"access_level": {"$lte": user_level}},
k=5
)
return results
5.2 客服智能问答
场景：自动回答客户常见问题
实现要点：
5.3 学术研究助手
场景：帮助研究人员查找和总结文献
实现要点：
六、评估与优化
6.1 评估指标
检索质量指标
def calculate_retrieval_metrics(retrieved_docs, relevant_docs):
"""计算检索指标"""
retrieved_ids = set([doc['id'] for doc in retrieved_docs])
relevant_ids = set([doc['id'] for doc in relevant_docs])
# 召回率 (Recall)
recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)
# 精确率 (Precision)
precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)
# F1分数
f1 = 2 * (precision * recall) / (precision + recall)
# MRR (Mean Reciprocal Rank)
for i, doc in enumerate(retrieved_docs, 1):
if doc['id'] in relevant_ids:
mrr = 1 / i
break
return {
"recall": recall,
"precision": precision,
"f1": f1,
"mrr": mrr
}
生成质量指标
- 答案准确性：与标准答案的相似度
- 幻觉率：生成内容中不基于参考资料的比例
- 完整性：是否完整回答了问题
- 引用准确性：引用是否正确
6.2 常见问题与解决方案
| 问题 | 原因 | 解决方案 |
|---|
| 检索不到相关文档 | Embedding模型不合适 | 更换或微调Embedding模型 |
| 答案包含幻觉 | 上下文不足或Prompt不当 | 优化Prompt，增加"仅基于资料回答"约束 |
| 响应速度慢 | 检索或生成耗时长 | 使用更快的向量数据库，减少检索文档数 |
| 答案质量不稳定 | 检索结果质量波动 | 增加重排序步骤，提高检索精确度 |
6.3 持续优化策略
- A/B测试：对比不同检索策略和Prompt的效果
- 用户反馈循环：收集用户评价，优化系统
- 定期评估：建立测试集，定期评估系统性能
- 模型更新：跟踪最新的Embedding和LLM模型
七、未来趋势与展望
7.1 多模态RAG
支持图像、音频等多种模态的检索和生成。
7.2 自适应RAG
根据问题类型自动选择最佳检索策略。
7.3 知识图谱增强
结合结构化知识图谱提升推理能力。
7.4 实时RAG
支持流式检索和增量生成，提升用户体验。
总结
RAG技术通过向量检索和上下文注入两大核心机制，成功地将外部知识与大语言模型结合，显著提升了AI系统的准确性和实用性。
关键要点回顾
- 向量检索是基础：选择合适的Embedding模型和向量数据库至关重要
- 上下文注入是关键：精心设计的Prompt能大幅提升答案质量
- 优化是持续的：通过混合检索、重排序、元数据过滤等技术不断改进
- 评估要全面：关注检索和生成两个阶段的指标
实践建议
- 从简单开始：先实现基础RAG，再逐步优化
- 重视数据质量：高质量的文档是RAG成功的前提
- 持续迭代：基于用户反馈和评估结果不断改进
- 选择合适的工具栈：根据实际需求选择Embedding模型、向量数据库和LLM
RAG技术正在快速发展，掌握其原理与实践，将帮助你构建更智能、更可靠的AI应用。
