1.1 什么是RAG？¶
In [ ]:
Copied!
# 让我们通过一个简单的例子理解RAG
# 场景：你是一家公司的客服
customer_question = "公司的年假政策是什么？"
print(f"客户问题: {customer_question}")
print("\n" + "="*50)
# 没有RAG的情况
print("❌ 传统LLM（没有RAG）:")
print(" LLM回答: 我不知道贵公司的具体政策...")
print(" 问题: LLM不知道公司内部信息")
print("\n" + "="*50)
# 有RAG的情况
print("✅ RAG系统:")
# 1. 检索相关文档
retrieved_docs = [
"员工手册第3章：年假规定",
"HR政策文档：员工可享受15天年假"
]
print(f" 检索到的文档: {retrieved_docs}")
# 2. 基于文档生成答案
print(" RAG回答: 根据员工手册第3章，员工可享受15天年假...")
print(" 优势: 基于真实文档，答案准确可靠")
# 让我们通过一个简单的例子理解RAG
# 场景：你是一家公司的客服
customer_question = "公司的年假政策是什么？"
print(f"客户问题: {customer_question}")
print("\n" + "="*50)
# 没有RAG的情况
print("❌ 传统LLM（没有RAG）:")
print(" LLM回答: 我不知道贵公司的具体政策...")
print(" 问题: LLM不知道公司内部信息")
print("\n" + "="*50)
# 有RAG的情况
print("✅ RAG系统:")
# 1. 检索相关文档
retrieved_docs = [
"员工手册第3章：年假规定",
"HR政策文档：员工可享受15天年假"
]
print(f" 检索到的文档: {retrieved_docs}")
# 2. 基于文档生成答案
print(" RAG回答: 根据员工手册第3章，员工可享受15天年假...")
print(" 优势: 基于真实文档，答案准确可靠")
练习1.1：RAG vs 其他技术¶
题目：判断以下场景应该使用RAG、Fine-tuning还是Prompt Engineering？
- 需要AI用特定格式输出JSON
- 回答公司内部文档的问题
- 提升通用问答能力
答案：
- 场景1 → Prompt Engineering（格式要求）
- 场景2 → RAG（准确性和实时性）
- 场景3 → Fine-tuning（输出质量）
1.2 RAG的5大核心组件¶
In [ ]:
Copied!
# RAG系统的5大核心组件
class RAGComponents:
"""RAG核心组件演示"""
def __init__(self):
print("初始化RAG系统的5大核心组件...\n")
def component_1_document_loader(self):
"""组件1：文档加载器"""
print("📄 组件1：文档加载器 (Document Loaders)")
print(" 作用: 从各种数据源加载文档")
print(" 支持格式: PDF, TXT, Word, 网页等")
print(" 示例: SimpleDirectoryReader, PyPDFLoader")
print()
def component_2_text_splitter(self):
"""组件2：文本分块器"""
print("✂️ 组件2：文本分块器 (Text Splitters)")
print(" 作用: 将长文档切分成可管理的块")
print(" 参数: chunk_size, chunk_overlap")
print(" 示例: SentenceSplitter, RecursiveCharacterTextSplitter")
print()
def component_3_embedding_model(self):
"""组件3：嵌入模型"""
print("🔢 组件3：嵌入模型 (Embedding Models)")
print(" 作用: 将文本转换为向量")
print(" 示例: OpenAI embeddings, BGE, E5")
print(" 输出: 768/1536维向量")
print()
def component_4_vector_store(self):
"""组件4：向量数据库"""
print("🗄️ 组件4：向量数据库 (Vector Stores)")
print(" 作用: 存储和检索向量")
print(" 示例: Chroma, Qdrant, Milvus")
print(" 算法: HNSW, IVF")
print()
def component_5_llm(self):
"""组件5：大语言模型"""
print("🤖 组件5：大语言模型 (LLMs)")
print(" 作用: 基于检索文档生成答案")
print(" 示例: GPT-4, Claude, Qwen")
print(" 参数: temperature, max_tokens")
print()
# 展示所有组件
rag = RAGComponents()
rag.component_1_document_loader()
rag.component_2_text_splitter()
rag.component_3_embedding_model()
rag.component_4_vector_store()
rag.component_5_llm()
# RAG系统的5大核心组件
class RAGComponents:
"""RAG核心组件演示"""
def __init__(self):
print("初始化RAG系统的5大核心组件...\n")
def component_1_document_loader(self):
"""组件1：文档加载器"""
print("📄 组件1：文档加载器 (Document Loaders)")
print(" 作用: 从各种数据源加载文档")
print(" 支持格式: PDF, TXT, Word, 网页等")
print(" 示例: SimpleDirectoryReader, PyPDFLoader")
print()
def component_2_text_splitter(self):
"""组件2：文本分块器"""
print("✂️ 组件2：文本分块器 (Text Splitters)")
print(" 作用: 将长文档切分成可管理的块")
print(" 参数: chunk_size, chunk_overlap")
print(" 示例: SentenceSplitter, RecursiveCharacterTextSplitter")
print()
def component_3_embedding_model(self):
"""组件3：嵌入模型"""
print("🔢 组件3：嵌入模型 (Embedding Models)")
print(" 作用: 将文本转换为向量")
print(" 示例: OpenAI embeddings, BGE, E5")
print(" 输出: 768/1536维向量")
print()
def component_4_vector_store(self):
"""组件4：向量数据库"""
print("🗄️ 组件4：向量数据库 (Vector Stores)")
print(" 作用: 存储和检索向量")
print(" 示例: Chroma, Qdrant, Milvus")
print(" 算法: HNSW, IVF")
print()
def component_5_llm(self):
"""组件5：大语言模型"""
print("🤖 组件5：大语言模型 (LLMs)")
print(" 作用: 基于检索文档生成答案")
print(" 示例: GPT-4, Claude, Qwen")
print(" 参数: temperature, max_tokens")
print()
# 展示所有组件
rag = RAGComponents()
rag.component_1_document_loader()
rag.component_2_text_splitter()
rag.component_3_embedding_model()
rag.component_4_vector_store()
rag.component_5_llm()
练习1.2：组件匹配¶
题目：将以下功能与对应的组件匹配
功能 → 组件
- 将PDF转换为可处理的文本 → ?
- 计算两个文本的相似度 → ?
- 生成自然语言答案 → ?
- 将长文档切分 → ?
- 存储百万级向量 → ?
答案： 1 → 文档加载器 2 → 嵌入模型 3 → LLM 4 → 文本分块器 5 → 向量数据库
1.3 简单RAG系统演示¶
让我们构建一个最简单的RAG系统来理解工作流程
In [ ]:
Copied!
# 简单RAG系统演示（不需要真实API调用）
class SimpleRAGDemo:
"""简化的RAG系统演示"""
def __init__(self):
# 模拟知识库
self.knowledge_base = [
"Python是一种高级编程语言，由Guido van Rossum于1991年创建。",
"Python的特点是语法简洁、易学易用。",
"Python广泛应用于Web开发、数据分析、人工智能等领域。"
]
def retrieve(self, query, top_k=2):
"""步骤1：检索（简化版，基于关键词）"""
print(f"\n🔍 步骤1：检索相关文档")
print(f" 查询: {query}")
# 简化的关键词匹配
scores = []
for i, doc in enumerate(self.knowledge_base):
# 计算查询词在文档中出现的次数
score = sum(1 for word in query.lower().split() if word in doc.lower())
scores.append((i, doc, score))
print(f" 文档{i+1} 相关度: {score}")
# 排序并返回top_k
scores.sort(key=lambda x: x[2], reverse=True)
retrieved = [doc for i, doc, score in scores[:top_k] if score > 0]
print(f"\n ✅ 检索到 {len(retrieved)} 个相关文档")
return retrieved
def generate(self, query, retrieved_docs):
"""步骤2：生成答案"""
print(f"\n🤖 步骤2：生成答案")
if not retrieved_docs:
return "抱歉，知识库中没有找到相关信息。"
# 构建上下文
context = "\n".join([f"- {doc}" for doc in retrieved_docs])
# 简化的答案生成（实际应该调用LLM）
answer = f"""基于检索到的信息：
{context}
答案总结：{query}的相关信息如上所述。"""
print(f" ✅ 答案生成完成")
return answer
def query(self, question):
"""完整的查询流程"""
print("="*60)
print(f"❓ 用户问题: {question}")
print("="*60)
# 检索
retrieved = self.retrieve(question)
# 生成
answer = self.generate(question, retrieved)
print("\n" + "="*60)
print("📝 最终答案:")
print(answer)
print("="*60)
return answer
# 测试系统
demo = SimpleRAGDemo()
# 测试问题
questions = [
"Python是什么？",
"Python有哪些特点？",
"Python可以用来做什么？"
]
for q in questions:
demo.query(q)
print()
# 简单RAG系统演示（不需要真实API调用）
class SimpleRAGDemo:
"""简化的RAG系统演示"""
def __init__(self):
# 模拟知识库
self.knowledge_base = [
"Python是一种高级编程语言，由Guido van Rossum于1991年创建。",
"Python的特点是语法简洁、易学易用。",
"Python广泛应用于Web开发、数据分析、人工智能等领域。"
]
def retrieve(self, query, top_k=2):
"""步骤1：检索（简化版，基于关键词）"""
print(f"\n🔍 步骤1：检索相关文档")
print(f" 查询: {query}")
# 简化的关键词匹配
scores = []
for i, doc in enumerate(self.knowledge_base):
# 计算查询词在文档中出现的次数
score = sum(1 for word in query.lower().split() if word in doc.lower())
scores.append((i, doc, score))
print(f" 文档{i+1} 相关度: {score}")
# 排序并返回top_k
scores.sort(key=lambda x: x[2], reverse=True)
retrieved = [doc for i, doc, score in scores[:top_k] if score > 0]
print(f"\n ✅ 检索到 {len(retrieved)} 个相关文档")
return retrieved
def generate(self, query, retrieved_docs):
"""步骤2：生成答案"""
print(f"\n🤖 步骤2：生成答案")
if not retrieved_docs:
return "抱歉，知识库中没有找到相关信息。"
# 构建上下文
context = "\n".join([f"- {doc}" for doc in retrieved_docs])
# 简化的答案生成（实际应该调用LLM）
answer = f"""基于检索到的信息：
{context}
答案总结：{query}的相关信息如上所述。"""
print(f" ✅ 答案生成完成")
return answer
def query(self, question):
"""完整的查询流程"""
print("="*60)
print(f"❓ 用户问题: {question}")
print("="*60)
# 检索
retrieved = self.retrieve(question)
# 生成
answer = self.generate(question, retrieved)
print("\n" + "="*60)
print("📝 最终答案:")
print(answer)
print("="*60)
return answer
# 测试系统
demo = SimpleRAGDemo()
# 测试问题
questions = [
"Python是什么？",
"Python有哪些特点？",
"Python可以用来做什么？"
]
for q in questions:
demo.query(q)
print()
1.4 技术栈选择练习¶
In [ ]:
Copied!
# 技术栈选择决策树
def choose_tech_stack():
"""交互式技术栈选择工具"""
print("🤔 RAG技术栈选择助手\n")
# 问题1
deployment = input("Q1: 部署方式? (1.云API 2.私有部署): ")
# 问题2
budget = input("Q2: 预算情况? (1.有预算 2.预算有限): ")
# 问题3
scale = input("Q3: 数据规模? (1.<10万文档 2.>10万文档): ")
# 推荐方案
print("\n" + "="*50)
print("💡 推荐技术栈:")
print("="*50)
# 框架
print("\n📦 框架: LlamaIndex")
print(" 理由: 专注RAG，易于上手")
# 向量库
if scale == "1":
print("\n🗄️ 向量库: Chroma")
print(" 理由: 轻量简单，适合中小规模")
else:
print("\n🗄️ 向量库: Milvus")
print(" 理由: 高性能可扩展，适合大规模")
# LLM
if deployment == "1" and budget == "1":
print("\n🤖 LLM: GPT-4")
print(" 理由: 质量最高，适合生产")
elif deployment == "1":
print("\n🤖 LLM: GPT-3.5-turbo")
print(" 理由: 性价比高，速度快")
else:
print("\n🤖 LLM: Qwen-72B")
print(" 理由: 可私有部署，中文优化")
print("\n" + "="*50)
# 取消注释以运行交互式选择
# choose_tech_stack()
# 预设示例
print("示例配置：快速学习场景")
print("- 框架: LlamaIndex")
print("- 向量库: Chroma")
print("- LLM: GPT-3.5-turbo")
print("- 嵌入: OpenAI text-embedding-3-small")
# 技术栈选择决策树
def choose_tech_stack():
"""交互式技术栈选择工具"""
print("🤔 RAG技术栈选择助手\n")
# 问题1
deployment = input("Q1: 部署方式? (1.云API 2.私有部署): ")
# 问题2
budget = input("Q2: 预算情况? (1.有预算 2.预算有限): ")
# 问题3
scale = input("Q3: 数据规模? (1.<10万文档 2.>10万文档): ")
# 推荐方案
print("\n" + "="*50)
print("💡 推荐技术栈:")
print("="*50)
# 框架
print("\n📦 框架: LlamaIndex")
print(" 理由: 专注RAG，易于上手")
# 向量库
if scale == "1":
print("\n🗄️ 向量库: Chroma")
print(" 理由: 轻量简单，适合中小规模")
else:
print("\n🗄️ 向量库: Milvus")
print(" 理由: 高性能可扩展，适合大规模")
# LLM
if deployment == "1" and budget == "1":
print("\n🤖 LLM: GPT-4")
print(" 理由: 质量最高，适合生产")
elif deployment == "1":
print("\n🤖 LLM: GPT-3.5-turbo")
print(" 理由: 性价比高，速度快")
else:
print("\n🤖 LLM: Qwen-72B")
print(" 理由: 可私有部署，中文优化")
print("\n" + "="*50)
# 取消注释以运行交互式选择
# choose_tech_stack()
# 预设示例
print("示例配置：快速学习场景")
print("- 框架: LlamaIndex")
print("- 向量库: Chroma")
print("- LLM: GPT-3.5-turbo")
print("- 嵌入: OpenAI text-embedding-3-small")
1.5 知识检查¶
自测题¶
RAG的核心价值是什么？
- a) 降低成本
- b) 减少幻觉、知识更新、可解释性 ✅
- c) 提高生成速度
以下哪个不是RAG的组件？
- a) 文档加载器
- b) 数据库连接池 ✅
- c) 嵌入模型
Hit Rate衡量什么？
- a) 检索速度
- b) 检索到相关文档的比例 ✅
- c) 生成质量
编码练习¶
任务：完成下面的RAG流程函数
def rag_pipeline(query, knowledge_base, llm):
"""
简单的RAG流程
Args:
query: 用户问题
knowledge_base: 文档列表
llm: 语言模型函数
Returns:
答案
"""
# TODO: 实现RAG流程
# 1. 检索相关文档
# 2. 构建提示词
# 3. 生成答案
pass
答案：
def rag_pipeline(query, knowledge_base, llm):
# 1. 检索（简化版）
relevant_docs = [doc for doc in knowledge_base
if any(word in doc.lower()
for word in query.lower().split())]
# 2. 构建提示词
context = "\n".join(relevant_docs)
prompt = f"基于以下文档回答问题：\n{context}\n问题：{query}"
# 3. 生成
answer = llm(prompt)
return answer
本章总结¶
✅ 你学到了：
- RAG的定义和核心价值
- RAG的5大核心组件
- RAG vs Fine-tuning vs Prompt Engineering的区别
- 如何选择合适的技术栈
🎯 下一步：
- 第2章：环境搭建与工具准备
- 动手搭建开发环境
📚 扩展阅读：