检索增强生成（RAG）
检索增强生成（Retrieval-Augmented Generation，简称 RAG）是近年来自然语言处理领域最重要的技术突破之一。它通过将信息检索与文本生成相结合，让大语言模型能够利用外部知识库生成更准确、更有依据的回答，有效解决了模型幻觉和知识过时的问题。
为什么需要 RAG
大语言模型虽然在理解和生成自然语言方面表现出色，但存在几个根本性的局限：
知识截止问题
大语言模型的知识来自训练数据，训练完成后知识就被"冻结"在特定时间点。这意味着模型无法回答关于最新事件的问题。比如，如果你问一个 2023 年训练的模型关于 2024 年发生的新闻事件，模型要么承认不知道，要么可能编造一个看起来合理但完全错误的答案。
这种局限性在企业应用中尤为明显。企业的产品信息、政策文档、技术规范都在持续更新，如果模型无法获取最新信息，其回答就失去了实际价值。
领域知识不足
通用大语言模型虽然拥有广泛的知识，但在特定专业领域往往缺乏深度。医疗诊断、法律咨询、金融分析等专业领域需要高度专业的知识，这些知识可能：
- 存在于私有文档中，从未公开
- 分布在专业数据库或知识库中
- 需要特定领域专家的理解和解读
仅依靠模型内置的通用知识，很难提供专业、准确的服务。
幻觉问题
大语言模型最令人头疼的问题是"幻觉"——模型会自信地生成看似合理但实际错误的信息。这是因为模型本质上是在进行概率预测，而非事实检索。当模型不确定时，它会倾向于生成一个听起来合理的答案，而不是承认"我不知道"。
幻觉问题在关键应用场景中可能带来严重后果。医疗问答系统给出错误的诊断建议、法律咨询系统提供不合规的法律意见，这些都会造成实际损失。
缺乏可解释性
传统大语言模型无法为其回答提供来源引用。用户要么完全信任模型的输出，要么需要自己验证每个事实。这种"黑箱"特性降低了模型输出的可信度，也难以满足企业对合规性和可审计性的要求。
RAG 的核心思想
RAG 的核心思想很直观：与其让模型"记住"所有知识，不如在需要时"查找"相关知识。这就像开卷考试和闭卷考试的区别——开卷考试允许查阅资料，通常能给出更准确的答案。
RAG 系统的工作流程分为四个核心阶段：
数据摄入：将外部知识库（如文档、数据库、网页）转换为可检索的形式。这包括文档切分、向量化、存储到向量数据库。
检索：当用户提出问题时，系统从知识库中检索相关的文档片段。检索过程基于语义相似度，能够理解问题的"意图"而非简单的关键词匹配。
增强：将检索到的相关文档与用户问题组合，形成增强的提示词。这个提示词包含了回答问题所需的背景知识。
生成：大语言模型基于增强后的提示词生成回答。由于有了具体的知识上下文，模型能够给出更准确、更有依据的答案。
这个流程带来的核心优势是：模型不需要"记住"所有知识，只需要在推理时"查找"相关知识即可。知识库可以随时更新，模型始终保持最新；知识来源可追溯，答案更加可信；专业领域的私有知识也能被有效利用。
RAG 系统架构
一个完整的 RAG 系统由多个组件构成，每个组件都有其特定的职责和设计考量。
知识库与文档处理
知识库是 RAG 系统的"记忆"，存储着系统可以检索的所有知识。知识库的内容可以是：
- 企业内部文档（政策手册、技术文档、会议记录）
- 产品文档和用户手册
- 客户问答历史
- 行业规范和标准
- 网页和新闻文章
原始文档通常需要经过处理才能被有效检索。文档处理包括几个关键步骤：
文档清洗：去除无关内容（如 HTML 标签、页眉页脚、导航栏），统一文本格式，处理特殊字符和编码问题。
文档切分：将长文档切分成适当大小的片段（chunk）。切分策略会显著影响检索效果：
- 固定长度切分：按字符数或词数切分，简单但可能打断语义完整性
- 语义切分：按段落、章节等语义单位切分，保持内容连贯性
- 递归切分：先按大单位切分，如果片段仍然过长则继续细分
切分的粒度需要权衡：太小的片段可能丢失上下文信息，太大的片段可能包含过多无关内容。通常每个片段包含 200-500 个 token 是比较合理的选择。
向量嵌入
向量嵌入是将文本转换为数值向量的过程。嵌入向量能够捕捉文本的语义信息，语义相近的文本在向量空间中距离更近。
嵌入模型的选择影响检索质量。常用的嵌入模型包括：
OpenAI text-embedding 系列：效果优秀，使用简单，但需要 API 调用费用。
BGE 系列：北京智源人工智能研究院开源的中文嵌入模型，在中文场景表现优异。
M3E 系列：多语言嵌入模型，支持中英文混合场景。
Sentence-Transformers：支持多种预训练模型，可本地部署。
from sentence_transformers import SentenceTransformer
# 加载中文嵌入模型
model = SentenceTransformer('BAAI/bge-base-zh-v1.5')
# 生成嵌入向量
texts = ["自然语言处理是人工智能的重要分支", "深度学习改变了NLP的研究范式"]
embeddings = model.encode(texts)
print(f"嵌入向量维度: {embeddings.shape}")
# 嵌入向量维度: (2, 768)
嵌入向量通常是高维的（768 维或 1024 维），每个维度代表文本在某个语义特征上的"得分"。两个嵌入向量之间的相似度可以用余弦相似度计算：
余弦相似度范围为 ，值越接近 1 表示两个向量越相似。
向量数据库
向量数据库是专门用于存储和检索向量数据的数据库。与传统数据库不同，向量数据库的核心操作是"近似最近邻搜索"（Approximate Nearest Neighbor, ANN）——给定一个查询向量，快速找到数据库中与之最相似的向量。
主流的向量数据库各有特点：
FAISS（Facebook AI Similarity Search）：Meta 开源的高效向量检索库，支持多种索引结构和量化方法。适合本地部署和大规模向量检索，但不支持数据持久化和分布式部署。
Chroma：轻量级向量数据库，支持嵌入式部署，适合快速原型开发和小规模应用。
Pinecone：全托管的云向量数据库，无需运维，支持自动扩展，但需要付费使用。
Weaviate：开源向量数据库，支持多种数据类型和语义搜索，提供 GraphQL API。
Milvus：开源分布式向量数据库，支持大规模向量检索，适合企业级应用。
import chromadb
from chromadb.config import Settings
# 创建本地向量数据库
client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./chroma_db"
))
# 创建集合
collection = client.create_collection("documents")
# 添加文档
documents = [
"Python 是一种广泛使用的编程语言",
"机器学习是人工智能的核心技术",
"自然语言处理让计算机理解人类语言"
]
collection.add(
documents=documents,
ids=["doc1", "doc2", "doc3"]
)
# 查询
results = collection.query(
query_texts=["编程语言有哪些"],
n_results=2
)
print(results['documents'])
# 返回最相关的文档
检索策略
检索是 RAG 系统最关键的环节之一。检索质量直接决定了生成答案的质量。
语义检索：使用查询向量与文档向量的相似度进行检索。语义检索能够理解查询的意图，找到语义相关的文档，即使没有关键词匹配。比如查询"如何提高睡眠质量"能够检索到"改善失眠的方法"相关文档。
关键词检索：传统的基于关键词的检索方法，如 BM25。关键词检索对于精确匹配（如专有名词、产品型号）更有效。
混合检索：结合语义检索和关键词检索的优点。先用两种方法分别检索，然后融合结果并重新排序。混合检索在各种场景下都能取得稳定的效果。
def hybrid_search(query, semantic_results, keyword_results, alpha=0.5):
"""混合检索：结合语义和关键词检索结果"""
# 语义检索得分
semantic_scores = {doc: score for doc, score in semantic_results}
# 关键词检索得分
keyword_scores = {doc: score for doc, score in keyword_results}
# 合并文档
all_docs = set(semantic_scores.keys()) | set(keyword_scores.keys())
# 计算混合得分
hybrid_scores = []
for doc in all_docs:
sem_score = semantic_scores.get(doc, 0)
kw_score = keyword_scores.get(doc, 0)
# 归一化后加权
combined = alpha * sem_score + (1 - alpha) * kw_score
hybrid_scores.append((doc, combined))
# 按得分排序
hybrid_scores.sort(key=lambda x: x[1], reverse=True)
return hybrid_scores
重排序：在初步检索后，使用更精确的模型对结果重新排序。重排序模型通常是交叉编码器，能够同时考虑查询和文档的内容，给出更准确的相关性评分。
生成阶段
检索到相关文档后，需要将其与用户问题组合，形成增强的提示词。提示词的设计会影响模型的回答质量。
一个好的提示词模板应该：
- 清晰地告诉模型使用提供的上下文回答问题
- 指示模型不要使用上下文中没有的信息
- 要求模型在不确定时承认不知道
def build_prompt(question, context_docs):
"""构建增强提示词"""
context = "\n\n".join([f"文档 {i+1}: {doc}"
for i, doc in enumerate(context_docs)])
prompt = f"""你是一个专业的问答助手。请根据以下提供的上下文信息回答用户的问题。
上下文信息：
{context}
用户问题：{question}
请基于上下文信息回答问题。如果上下文中没有足够的信息回答问题，请明确说明"根据提供的信息无法回答该问题"。回答时请注明信息来源。"""
return prompt
从零实现 RAG 系统
理解了 RAG 的核心概念后，让我们从零开始实现一个完整的 RAG 系统。这个实现将涵盖所有关键组件，帮助你深入理解 RAG 的工作原理。
安装依赖
pip install sentence-transformers chromadb numpy
实现向量数据库
首先实现一个简单的内存向量数据库，支持添加文档和相似度检索：
import numpy as np
from typing import List, Tuple
class SimpleVectorDB:
"""简单的内存向量数据库"""
def __init__(self):
self.documents = [] # 存储文档
self.embeddings = [] # 存储对应的嵌入向量
def add(self, documents: List[str], embeddings: np.ndarray):
"""添加文档和嵌入向量"""
self.documents.extend(documents)
if len(self.embeddings) == 0:
self.embeddings = embeddings
else:
self.embeddings = np.vstack([self.embeddings, embeddings])
def cosine_similarity(self, query_embedding: np.ndarray) -> np.ndarray:
"""计算查询向量与所有文档的余弦相似度"""
# 归一化
query_norm = query_embedding / np.linalg.norm(query_embedding)
docs_norm = self.embeddings / np.linalg.norm(
self.embeddings, axis=1, keepdims=True
)
# 计算余弦相似度
similarities = np.dot(docs_norm, query_norm)
return similarities
def search(self, query_embedding: np.ndarray, top_k: int = 5) -> List[Tuple[str, float]]:
"""检索最相关的文档"""
similarities = self.cosine_similarity(query_embedding)
# 获取 top-k 索引
top_indices = np.argsort(similarities)[::-1][:top_k]
# 返回文档和相似度
results = [
(self.documents[i], similarities[i])
for i in top_indices
]
return results
实现文档切分
文档切分是确保检索质量的关键步骤：
from typing import List
class TextSplitter:
"""文本切分器"""
def __init__(self, chunk_size: int = 200, overlap: int = 50):
"""
chunk_size: 每个片段的最大字符数
overlap: 相邻片段之间的重叠字符数
"""
self.chunk_size = chunk_size
self.overlap = overlap
def split_text(self, text: str) -> List[str]:
"""将文本切分成片段"""
# 按段落分割
paragraphs = text.split('\n\n')
chunks = []
current_chunk = ""
for para in paragraphs:
para = para.strip()
if not para:
continue
# 如果当前片段加上新段落不超过限制
if len(current_chunk) + len(para) <= self.chunk_size:
current_chunk += para + "\n\n"
else:
# 保存当前片段
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para + "\n\n"
# 保存最后一个片段
if current_chunk:
chunks.append(current_chunk.strip())
# 添加重叠
if self.overlap > 0 and len(chunks) > 1:
overlapped_chunks = []
for i, chunk in enumerate(chunks):
if i > 0:
# 从上一个片段末尾取重叠部分
prev_chunk = chunks[i-1]
overlap_text = prev_chunk[-self.overlap:]
chunk = overlap_text + "\n" + chunk
overlapped_chunks.append(chunk)
chunks = overlapped_chunks
return chunks
def split_documents(self, documents: List[str]) -> List[str]:
"""切分多个文档"""
all_chunks = []
for doc in documents:
all_chunks.extend(self.split_text(doc))
return all_chunks
实现完整的 RAG 系统
将所有组件组合起来，实现完整的 RAG 系统：
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Optional
import numpy as np
class RAGSystem:
"""完整的 RAG 系统"""
def __init__(self, embedding_model: str = "BAAI/bge-base-zh-v1.5"):
# 初始化嵌入模型
print(f"加载嵌入模型: {embedding_model}")
self.embedder = SentenceTransformer(embedding_model)
# 初始化向量数据库
self.vector_db = SimpleVectorDB()
# 初始化文本切分器
self.splitter = TextSplitter(chunk_size=300, overlap=50)
# 存储原始文档（用于引用）
self.source_docs = {}
def index_documents(self, documents: List[str], source_ids: Optional[List[str]] = None):
"""索引文档到向量数据库"""
if source_ids is None:
source_ids = [f"doc_{i}" for i in range(len(documents))]
# 切分文档
all_chunks = []
chunk_to_source = {}
for doc_id, doc in zip(source_ids, documents):
chunks = self.splitter.split_text(doc)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc_id}_chunk_{i}"
all_chunks.append(chunk)
chunk_to_source[chunk_id] = {
'source_id': doc_id,
'chunk_index': i,
'text': chunk
}
self.source_docs[chunk_id] = doc_id
# 生成嵌入向量
print(f"正在索引 {len(all_chunks)} 个文档片段...")
embeddings = self.embedder.encode(all_chunks)
# 存储到向量数据库
self.vector_db.add(all_chunks, embeddings)
self.chunk_to_source = chunk_to_source
print(f"索引完成！共 {len(all_chunks)} 个片段")
def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
"""检索相关文档"""
# 生成查询向量
query_embedding = self.embedder.encode([query])[0]
# 检索
results = self.vector_db.search(query_embedding, top_k)
# 格式化结果
retrieved = []
for text, score in results:
chunk_id = None
for cid, info in self.chunk_to_source.items():
if info['text'] == text:
chunk_id = cid
break
retrieved.append({
'text': text,
'score': float(score),
'source': self.source_docs.get(chunk_id, 'unknown')
})
return retrieved
def build_prompt(self, query: str, context_docs: List[Dict]) -> str:
"""构建增强提示词"""
context_text = "\n\n".join([
f"[文档 {i+1}] (来源: {doc['source']}, 相关度: {doc['score']:.3f})\n{doc['text']}"
for i, doc in enumerate(context_docs)
])
prompt = f"""你是一个专业的知识问答助手。请根据以下提供的参考文档回答用户问题。
参考文档：
{context_text}
用户问题：{query}
回答要求：
1. 请基于参考文档中的信息回答问题
2. 如果参考文档中没有相关信息，请明确说明"根据提供的参考文档无法回答该问题"
3. 回答时请注明信息来源于哪个文档
4. 回答要准确、简洁、有条理
请开始回答："""
return prompt
def query(self, question: str, top_k: int = 3) -> Dict:
"""回答问题"""
# 检索相关文档
retrieved_docs = self.retrieve(question, top_k)
# 构建提示词
prompt = self.build_prompt(question, retrieved_docs)
return {
'question': question,
'context': retrieved_docs,
'prompt': prompt
}
# 使用示例
if __name__ == "__main__":
# 创建 RAG 系统
rag = RAGSystem()
# 示例文档
documents = [
"""Python 是一种广泛使用的高级编程语言，由 Guido van Rossum 于 1991 年首次发布。
Python 的设计哲学强调代码的可读性和简洁性，它的语法允许程序员用更少的代码行表达概念。
Python 支持多种编程范式，包括面向对象、命令式、函数式和过程式编程。
Python 有一个大型标准库，提供了丰富的功能模块。""",
"""机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习并改进。
机器学习算法使用历史数据作为输入来预测新的输出值。
常见的机器学习方法包括监督学习、无监督学习和强化学习。
深度学习是机器学习的一个子领域，使用多层神经网络进行学习。""",
"""自然语言处理（NLP）是人工智能和语言学的交叉领域，致力于让计算机理解和处理人类语言。
NLP 的应用包括机器翻译、情感分析、问答系统、文本摘要等。
现代 NLP 系统主要基于深度学习技术，如 Transformer 架构和预训练语言模型。
BERT、GPT 等预训练模型在 NLP 任务中取得了突破性进展。"""
]
# 索引文档
rag.index_documents(documents, ['python_intro', 'ml_intro', 'nlp_intro'])
# 查询
result = rag.query("什么是自然语言处理？")
print("=" * 60)
print("问题:", result['question'])
print("=" * 60)
print("\n检索到的相关文档：")
for i, doc in enumerate(result['context']):
print(f"\n文档 {i+1} (相关度: {doc['score']:.4f}):")
print(f"来源: {doc['source']}")
print(f"内容: {doc['text'][:100]}...")
print("\n" + "=" * 60)
print("生成的提示词：")
print(result['prompt'])
这个实现展示了 RAG 系统的核心组件和工作流程。在实际应用中，可以将这个系统与大语言模型 API（如 OpenAI、通义千问等）结合，实现完整的问答功能。
使用 LangChain 构建 RAG 应用
LangChain 是目前最流行的 LLM 应用开发框架，它提供了丰富的组件来构建 RAG 系统。
安装依赖
pip install langchain langchain-community langchain-openai chromadb
构建 RAG 链
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import os
# 设置 API Key
os.environ["OPENAI_API_KEY"] = "your-api-key"
# 1. 加载文档
loader = TextLoader("documents/knowledge_base.txt", encoding="utf-8")
documents = loader.load()
# 2. 文档切分
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100,
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"文档已切分为 {len(chunks)} 个片段")
# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
# 4. 创建检索器
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3}
)
# 5. 创建 LLM
llm = ChatOpenAI(
model_name="gpt-3.5-turbo",
temperature=0
)
# 6. 自定义提示词模板
prompt_template = """你是一个专业的问答助手。请根据以下上下文信息回答问题。
上下文：
{context}
问题：{question}
请基于上下文回答问题。如果上下文中没有相关信息，请说明"根据提供的信息无法回答"。
回答："""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
# 7. 创建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True,
chain_type_kwargs={"prompt": PROMPT}
)
# 8. 问答
def ask_question(question: str):
"""提问并获取答案"""
result = qa_chain({"query": question})
print(f"问题：{question}")
print(f"\n回答：{result['result']}")
print("\n参考来源：")
for i, doc in enumerate(result['source_documents']):
print(f" [{i+1}] {doc.metadata.get('source', 'unknown')}")
print(f" {doc.page_content[:100]}...")
# 使用示例
ask_question("什么是机器学习？")
使用对话式 RAG
对于多轮对话场景，需要维护对话历史：
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
# 创建对话记忆
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 创建对话式 RAG 链
conversational_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
memory=memory,
return_source_documents=True
)
# 多轮对话
def chat_with_rag(question: str):
"""多轮对话"""
result = conversational_chain({"question": question})
print(f"用户：{question}")
print(f"助手：{result['answer']}")
print()
return result['answer']
# 测试多轮对话
chat_with_rag("什么是深度学习？")
chat_with_rag("它有哪些应用？") # 系统会记住上一轮的上下文
chat_with_rag("和传统机器学习有什么区别？")
高级 RAG 技术
随着 RAG 技术的发展，出现了许多改进传统 RAG 的方法。
查询优化
用户的原始问题可能表述不清或缺少关键信息，需要优化后再检索：
class QueryOptimizer:
"""查询优化器"""
def __init__(self, llm):
self.llm = llm
def expand_query(self, query: str) -> List[str]:
"""查询扩展：生成多个相关查询"""
prompt = f"""请将以下问题扩展为3个相关的搜索查询，以便获取更全面的信息。
原问题：{query}
请输出3个查询，每行一个："""
response = self.llm.predict(prompt)
queries = [q.strip() for q in response.strip().split('\n') if q.strip()]
queries.append(query) # 包含原始查询
return queries
def rewrite_query(self, query: str) -> str:
"""查询改写：使查询更适合检索"""
prompt = f"""请将以下问题改写为更适合搜索引擎的查询语句，保留核心关键词。
原问题：{query}
改写后的查询："""
return self.llm.predict(prompt).strip()
def multi_query_retrieval(query: str, retriever, optimizer, top_k: int = 3):
"""多查询检索"""
# 扩展查询
queries = optimizer.expand_query(query)
# 对每个查询检索
all_docs = []
for q in queries:
docs = retriever.get_relevant_documents(q)
all_docs.extend(docs)
# 去重并排序
unique_docs = {}
for doc in all_docs:
content = doc.page_content
if content not in unique_docs:
unique_docs[content] = doc
return list(unique_docs.values())[:top_k]
重排序
使用交叉编码器对检索结果重新排序，提高相关性：
from sentence_transformers import CrossEncoder
class Reranker:
"""重排序器"""
def __init__(self, model_name: str = "BAAI/bge-reranker-base"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: List[str], top_k: int = 5) -> List[Tuple[str, float]]:
"""对文档重新排序"""
# 构建查询-文档对
pairs = [(query, doc) for doc in documents]
# 计算相关性分数
scores = self.model.predict(pairs)
# 排序
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return ranked[:top_k]
def retrieval_with_reranking(query: str, retriever, reranker, top_k: int = 10, final_k: int = 3):
"""带重排序的检索"""
# 初步检索
docs = retriever.get_relevant_documents(query)
doc_texts = [doc.page_content for doc in docs]
# 重排序
reranked = reranker.rerank(query, doc_texts, top_k=final_k)
# 返回重排序后的文档
return [doc for doc, score in reranked]
混合检索
结合语义检索和关键词检索的优势：
from rank_bm25 import BM25Okapi
import jieba
class HybridRetriever:
"""混合检索器"""
def __init__(self, documents: List[str], embedder, vector_db):
self.documents = documents
self.embedder = embedder
self.vector_db = vector_db
# 构建 BM25 索引
tokenized_docs = [list(jieba.cut(doc)) for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
def retrieve(self, query: str, top_k: int = 5, alpha: float = 0.5):
"""
混合检索
alpha: 语义检索的权重，1-alpha 为关键词检索权重
"""
# 语义检索
query_embedding = self.embedder.encode([query])[0]
semantic_results = self.vector_db.search(query_embedding, top_k=top_k*2)
semantic_scores = {doc: score for doc, score in semantic_results}
# 关键词检索
tokenized_query = list(jieba.cut(query))
bm25_scores = self.bm25.get_scores(tokenized_query)
keyword_results = [(self.documents[i], bm25_scores[i])
for i in range(len(self.documents))]
keyword_scores = {doc: score for doc, score in keyword_results}
# 归一化分数
max_sem = max(semantic_scores.values()) if semantic_scores else 1
max_kw = max(keyword_scores.values()) if keyword_scores else 1
# 合并分数
all_docs = set(semantic_scores.keys()) | set(keyword_scores.keys())
hybrid_scores = []
for doc in all_docs:
sem_score = semantic_scores.get(doc, 0) / max_sem
kw_score = keyword_scores.get(doc, 0) / max_kw
combined = alpha * sem_score + (1 - alpha) * kw_score
hybrid_scores.append((doc, combined))
# 排序返回
hybrid_scores.sort(key=lambda x: x[1], reverse=True)
return hybrid_scores[:top_k]
自适应检索
根据问题类型选择不同的检索策略：
class AdaptiveRetriever:
"""自适应检索器"""
def __init__(self, llm, retrievers: Dict[str, object]):
self.llm = llm
self.retrievers = retrievers # 不同类型的检索器
def classify_query(self, query: str) -> str:
"""分类查询类型"""
prompt = f"""请判断以下问题的类型，从以下选项中选择一个：
- factual: 事实性问题，需要精确信息
- conceptual: 概念性问题，需要解释说明
- procedural: 过程性问题，需要步骤指导
- conversational: 对话性问题，需要上下文理解
问题：{query}
问题类型："""
response = self.llm.predict(prompt).strip().lower()
for qtype in ['factual', 'conceptual', 'procedural', 'conversational']:
if qtype in response:
return qtype
return 'factual'
def retrieve(self, query: str, top_k: int = 5):
"""自适应检索"""
query_type = self.classify_query(query)
# 根据查询类型选择检索策略
if query_type == 'factual':
# 事实性问题：使用精确检索
return self.retrievers['bm25'].retrieve(query, top_k)
elif query_type == 'conceptual':
# 概念性问题：使用语义检索
return self.retrievers['semantic'].retrieve(query, top_k)
elif query_type == 'procedural':
# 过程性问题：使用混合检索
return self.retrievers['hybrid'].retrieve(query, top_k)
else:
# 对话性问题：使用上下文增强检索
return self.retrievers['contextual'].retrieve(query, top_k)
RAG 系统评估
评估 RAG 系统的质量是持续改进的关键。
检索质量评估
检索质量决定了生成答案的上限：
from typing import List, Dict
def evaluate_retrieval(
queries: List[str],
ground_truth: List[List[str]],
retriever,
k_values: List[int] = [1, 3, 5, 10]
) -> Dict[str, float]:
"""
评估检索质量
ground_truth: 每个查询对应的正确文档 ID 列表
"""
metrics = {f"Recall@{k}": [] for k in k_values}
metrics.update({f"Precision@{k}": [] for k in k_values})
metrics["MRR"] = []
for query, relevant_docs in zip(queries, ground_truth):
retrieved = retriever.get_relevant_documents(query)
retrieved_ids = [doc.metadata.get('id') for doc in retrieved]
for k in k_values:
# Recall@K
relevant_retrieved = len(set(retrieved_ids[:k]) & set(relevant_docs))
recall = relevant_retrieved / len(relevant_docs) if relevant_docs else 0
metrics[f"Recall@{k}"].append(recall)
# Precision@K
precision = relevant_retrieved / k if k > 0 else 0
metrics[f"Precision@{k}"].append(precision)
# MRR (Mean Reciprocal Rank)
for i, doc_id in enumerate(retrieved_ids):
if doc_id in relevant_docs:
metrics["MRR"].append(1 / (i + 1))
break
else:
metrics["MRR"].append(0)
# 计算平均值
return {k: sum(v) / len(v) for k, v in metrics.items()}
生成质量评估
评估生成答案的质量：
def evaluate_generation(
questions: List[str],
generated_answers: List[str],
reference_answers: List[str],
llm
) -> Dict[str, float]:
"""评估生成质量"""
metrics = {
"faithfulness": [], # 忠实度
"relevance": [], # 相关性
"completeness": [] # 完整性
}
for question, generated, reference in zip(questions, generated_answers, reference_answers):
# 使用 LLM 评估
prompt = f"""请评估以下答案的质量。
问题：{question}
生成的答案：{generated}
参考答案：{reference}
请从以下维度评分（1-5分）：
1. 忠实度：答案是否基于提供的信息，没有幻觉
2. 相关性：答案是否针对问题
3. 完整性：答案是否完整回答了问题
请以 JSON 格式返回评分：
{{"faithfulness": X, "relevance": X, "completeness": X}}"""
response = llm.predict(prompt)
# 解析评分并添加到指标中
# 这里简化处理，实际应该解析 JSON
return {k: sum(v) / len(v) for k, v in metrics.items()}
端到端评估
使用 Ragas 等工具进行自动化评估：
# 需要安装：pip install ragas
from ragas import evaluate
from datasets import Dataset
def evaluate_rag_pipeline(
questions: List[str],
answers: List[str],
contexts: List[List[str]],
ground_truths: List[str]
):
"""使用 Ragas 评估 RAG 管道"""
data = {
"question": questions,
"answer": answers,
"contexts": contexts,
"ground_truth": ground_truths
}
dataset = Dataset.from_dict(data)
# 评估指标包括：
# - faithfulness: 答案是否忠实于上下文
# - answer_relevancy: 答案是否相关
# - context_recall: 检索是否覆盖了所有相关信息
# - context_precision: 检索的精确度
results = evaluate(dataset)
return results
RAG 应用场景
RAG 技术在众多场景中都有广泛应用。
企业知识库问答
构建基于企业内部文档的智能问答系统：
class EnterpriseKnowledgeQA:
"""企业知识库问答系统"""
def __init__(self, knowledge_dir: str):
self.rag = RAGSystem()
self.load_knowledge(knowledge_dir)
def load_knowledge(self, directory: str):
"""加载知识库文档"""
import os
documents = []
doc_ids = []
for filename in os.listdir(directory):
if filename.endswith('.txt') or filename.endswith('.md'):
filepath = os.path.join(directory, filename)
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
documents.append(content)
doc_ids.append(filename)
self.rag.index_documents(documents, doc_ids)
def answer(self, question: str) -> Dict:
"""回答问题"""
return self.rag.query(question)
客户服务机器人
结合对话历史和知识库的智能客服：
class CustomerServiceBot:
"""客服机器人"""
def __init__(self):
self.rag = RAGSystem()
self.conversation_history = []
def chat(self, user_input: str) -> str:
"""对话"""
# 添加用户输入到历史
self.conversation_history.append({"role": "user", "content": user_input})
# 结合历史构建查询
query = self._build_contextual_query()
# 检索相关文档
result = self.rag.query(query)
# 生成回复（这里应该调用 LLM）
response = self._generate_response(result)
# 添加回复到历史
self.conversation_history.append({"role": "assistant", "content": response})
return response
def _build_contextual_query(self) -> str:
"""构建上下文查询"""
# 使用最近的对话历史
recent_history = self.conversation_history[-4:]
context = " ".join([turn["content"] for turn in recent_history])
return context
技术文档助手
帮助开发者查询技术文档：
class TechDocAssistant:
"""技术文档助手"""
def __init__(self, doc_sources: Dict[str, str]):
"""
doc_sources: {名称: 文档路径或URL}
"""
self.rag = RAGSystem()
self.load_docs(doc_sources)
def load_docs(self, sources: Dict[str, str]):
"""加载技术文档"""
for name, source in sources.items():
# 加载文档内容
if source.startswith('http'):
content = self._fetch_url(source)
else:
with open(source, 'r', encoding='utf-8') as f:
content = f.read()
self.rag.index_documents([content], [name])
def search_api(self, query: str) -> str:
"""搜索 API 用法"""
result = self.rag.query(query)
return result
def debug_help(self, error_message: str) -> str:
"""调试帮助"""
query = f"如何解决这个错误：{error_message}"
return self.rag.query(query)
RAG 最佳实践
基于实践经验，总结 RAG 系统的设计和优化要点：
文档处理
- 切分粒度：根据问题类型调整切分大小。简单事实查询用小片段，复杂分析用大片段。
- 元数据保留：保存文档来源、时间戳、作者等元数据，便于追溯和过滤。
- 增量更新：支持知识库的增量更新，避免全量重建索引。
检索优化
- 混合检索：结合语义检索和关键词检索，提高召回率。
- 查询理解：对用户查询进行意图识别和改写，提高检索准确性。
- 结果去重：合并相似或重复的检索结果，避免信息冗余。
生成优化
- 提示词设计：明确指示模型使用上下文信息，避免幻觉。
- 引用标注：要求模型在回答中标注信息来源，提高可解释性。
- 长度控制：根据应用场景控制回答长度，保持简洁。
系统优化
- 缓存机制：对常见查询结果进行缓存，提高响应速度。
- 批处理：支持批量文档索引和查询，提高吞吐量。
- 监控告警：建立完善的监控体系，及时发现和处理问题。
总结
检索增强生成是连接大语言模型与外部知识的关键技术，有效解决了模型知识过时、领域知识不足和幻觉等问题。本章介绍了：
- RAG 核心概念：知识截止、幻觉问题，以及 RAG 如何解决这些问题
- 系统架构：文档处理、向量嵌入、向量数据库、检索策略、生成阶段
- 从零实现：完整的 RAG 系统实现，深入理解各组件
- LangChain 应用：使用成熟框架快速构建 RAG 应用
- 高级技术：查询优化、重排序、混合检索、自适应检索
- 评估方法：检索质量评估、生成质量评估、端到端评估
- 应用场景：企业知识库、客服机器人、技术文档助手
- 最佳实践：文档处理、检索优化、生成优化、系统优化
RAG 技术正在快速发展，从简单的检索-生成架构演进到智能体驱动的复杂系统。掌握 RAG 技术对于构建实用的 AI 应用至关重要，它让大语言模型能够真正落地于企业场景，发挥更大的价值。