学习时间：2-3 周
难度：⭐⭐⭐☆☆
完成本教程后，你将能够：
- ✅ 理解 RAG 的核心原理和架构
- ✅ 掌握 Embedding（文本向量化）技术
- ✅ 实现文档切分与元数据管理
- ✅ 使用向量数据库（ChromaDB）
- ✅ 构建完整的知识库问答系统
- ✅ 应用高级 RAG 技术（查询改写、重排序）
RAG (Retrieval-Augmented Generation) = 检索增强生成
传统 LLM 的问题：
- ❌ 知识有截止日期（训练数据的时间限制）
- ❌ 容易产生幻觉（编造不存在的信息）
- ❌ 无法访问私有数据（企业内部文档）
- ❌ 难以追溯信息来源
RAG 的解决方案：
- ✅ 从外部知识库检索相关信息
- ✅ 将检索到的文档作为上下文
- ✅ LLM 基于真实文档生成答案
- ✅ 可以标注答案来源，增加可信度
┌─────────────────────────────────────────────────────────────┐
│ 离线阶段（一次性） │
└─────────────────────────────────────────────────────────────┘
📄 原始文档（PDF, Word, Markdown）
↓
📄 文档加载 & 清洗
↓
✂️ 文档切分（Chunking）
↓
🔢 向量化（Embedding）
↓
💾 存储到向量数据库
┌─────────────────────────────────────────────────────────────┐
│ 在线阶段（每次查询） │
└─────────────────────────────────────────────────────────────┘
❓ 用户提问
↓
🔢 问题向量化
↓
🔍 向量相似度搜索（检索相关文档）
↓
📋 构建 Prompt（问题 + 检索到的文档）
↓
🤖 LLM 生成答案
↓
💡 返回答案 + 来源引用
| 特性 | RAG | 微调（Fine-tuning） | 长 Prompt |
|---|---|---|---|
| 成本 | 低 | 高（需要GPU训练） | 中（Token多） |
| 知识更新 | 实时（更新知识库） | 需要重新训练 | 实时 |
| 准确性 | 高（基于真实文档） | 高 | 中 |
| 可追溯性 | ✅ 可标注来源 | ❌ 无法追溯 | |
| 私有数据 | ✅ 支持 | ✅ 支持 | ✅ 支持 |
| 维护成本 | 低 | 高 | 低 |
最佳实践： RAG 适合大多数企业知识库场景，微调适合特定领域任务。
| 特性 | Java | Python |
|---|---|---|
| 向量化 | 需要外部服务 | OpenAI SDK 直接调用 |
| 向量数据库 | LangChain4j + Chroma | ChromaDB / Qdrant 原生支持 |
| 文档解析 | Apache POI / Tika | pypdf / python-docx |
| 科学计算 | 依赖 DL4J | NumPy / SciPy 原生 |
| 框架成熟度 | LangChain4j（较新） | LangChain（成熟） |
Embedding = 将文本转换为向量（数字数组）
"人工智能正在改变世界" → [0.23, -0.45, 0.67, ..., 0.12]
↑
1536 维向量核心思想： 语义相似的文本，在向量空间中距离更近。
"我喜欢吃苹果" → 向量A
"我喜欢吃水果" → 向量B ← 与A相似度 0.85
"今天天气很好" → 向量C ← 与A相似度 0.15参考代码：examples/03-rag/01_basic_embedding.py
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def get_embedding(text: str) -> list[float]:
"""获取文本的向量表示"""
text = text.replace("\n", " ") # 清理换行符
response = client.embeddings.create(
input=[text],
model="text-embedding-3-small" # 1536维，性价比高
)
return response.data[0].embedding
# 示例
text = "人工智能正在改变世界"
embedding = get_embedding(text)
print(f"文本：{text}")
print(f"向量维度：{len(embedding)}") # 1536
print(f"前5个维度：{embedding[:5]}")输出：
文本：人工智能正在改变世界
向量维度：1536
前5个维度：[0.0234, -0.0456, 0.0678, -0.0123, 0.0890]
| 模型 | 维度 | 价格 | 适用场景 |
|---|---|---|---|
| text-embedding-3-small | 1536 | $0.02/1M tokens | 日常应用、开发测试 |
| text-embedding-3-large | 3072 | $0.13/1M tokens | 高精度要求 |
| text-embedding-ada-002 | 1536 | $0.10/1M tokens | 旧版本（不推荐） |
推荐： 优先使用 text-embedding-3-small，性价比最高。
import numpy as np
def cosine_similarity(vec1: list[float], vec2: list[float]) -> float:
"""
计算余弦相似度
公式：cos(θ) = (A · B) / (||A|| * ||B||)
返回值：-1 到 1，越接近 1 越相似
"""
a = np.array(vec1)
b = np.array(vec2)
dot_product = np.dot(a, b) # 点积
norm_a = np.linalg.norm(a) # A的模
norm_b = np.linalg.norm(b) # B的模
return dot_product / (norm_a * norm_b)
# 示例
text1 = "我喜欢吃苹果"
text2 = "我喜欢吃水果"
text3 = "今天天气很好"
emb1 = get_embedding(text1)
emb2 = get_embedding(text2)
emb3 = get_embedding(text3)
sim_12 = cosine_similarity(emb1, emb2)
sim_13 = cosine_similarity(emb1, emb3)
print(f"相似度（1-2）：{sim_12:.4f}") # 0.85+ (高度相关)
print(f"相似度（1-3）：{sim_13:.4f}") # 0.15- (无关)def euclidean_distance(vec1: list[float], vec2: list[float]) -> float:
"""
计算欧式距离
公式：d = sqrt(Σ(ai - bi)²)
返回值：0 到 ∞，越接近 0 越相似
"""
a = np.array(vec1)
b = np.array(vec2)
return np.linalg.norm(a - b)为什么余弦相似度更常用？
- 不受向量长度影响，只关心方向
- 数值范围固定（-1 到 1），易于解释
- 在高维空间表现更好
def semantic_search(query: str, documents: list[str], top_k: int = 3):
"""
语义搜索：找到与查询最相似的文档
Args:
query: 查询文本
documents: 文档列表
top_k: 返回前K个结果
Returns:
[(文档, 相似度分数), ...]
"""
# 1. 获取查询向量
query_embedding = get_embedding(query)
# 2. 批量获取文档向量
doc_embeddings = get_embeddings_batch(documents)
# 3. 计算相似度
similarities = []
for i, doc_embedding in enumerate(doc_embeddings):
similarity = cosine_similarity(query_embedding, doc_embedding)
similarities.append((documents[i], similarity))
# 4. 排序并返回 Top K
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_k]
# 示例
documents = [
"Python 是一种高级编程语言，适合数据科学和机器学习",
"Java 是一种面向对象的编程语言，广泛用于企业开发",
"机器学习是人工智能的一个分支，让计算机从数据中学习",
"深度学习使用神经网络处理复杂的模式识别任务",
]
results = semantic_search("适合 AI 开发的编程语言", documents, top_k=2)
for doc, score in results:
print(f"[{score:.4f}] {doc}")输出：
[0.8521] Python 是一种高级编程语言，适合数据科学和机器学习
[0.7234] 机器学习是人工智能的一个分支，让计算机从数据中学习
❌ 错误做法： 逐个调用 API
# 慢且浪费
embeddings = [get_embedding(doc) for doc in documents] # 100个文档 = 100次API调用✅ 正确做法： 批量调用
def get_embeddings_batch(texts: list[str]) -> list[list[float]]:
"""批量获取 Embedding（最多 2048 个文本）"""
texts = [text.replace("\n", " ") for text in texts]
response = client.embeddings.create(
input=texts, # 一次传入多个文本
model="text-embedding-3-small"
)
return [data.embedding for data in response.data]
# 100个文档 = 1次API调用（快10倍+）
embeddings = get_embeddings_batch(documents)性能对比：
- 逐个调用：100个文档 ≈ 10-15秒
- 批量调用：100个文档 ≈ 1-2秒
OpenAI Embedding 天然支持多语言：
# 中英文语义相同的文本，向量相似
text_cn = "你好世界"
text_en = "Hello world"
emb_cn = get_embedding(text_cn)
emb_en = get_embedding(text_en)
similarity = cosine_similarity(emb_cn, emb_en)
print(f"中英相似度：{similarity:.4f}") # 0.70+ (语义相同)问题： LLM 的上下文窗口有限制
- GPT-4o-mini：128K tokens
- 完整文档可能有 100 页、50K+ tokens
- 但我们只需要其中相关的 1-2 页
解决方案： 将文档切分成小块（Chunks）
┌──────────────────────────────┐
│ 100页技术文档（50K tokens） │
└──────────────────────────────┘
↓ 切分
┌────────┬────────┬────────┬────────┐
│ 块1 │ 块2 │ 块3 │ ... │
│ 500字 │ 500字 │ 500字 │ │
└────────┴────────┴────────┴────────┘
↓ 检索
只取相关的 2-3 块
↓
传给 LLM（1500字，3K tokens）
参考代码：examples/03-rag/02_document_chunking.py
class SimpleTextSplitter:
"""简单文本切分器"""
def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
"""
Args:
chunk_size: 每块的字符数
chunk_overlap: 重叠字符数（避免语义断裂）
"""
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
def split_text(self, text: str) -> list[str]:
"""切分文本"""
chunks = []
start = 0
while start < len(text):
end = start + self.chunk_size
chunk = text[start:end]
# 尝试在句号处切分
if end < len(text):
last_period = max(chunk.rfind('。'), chunk.rfind('.'))
if last_period != -1 and last_period > self.chunk_size * 0.5:
chunk = chunk[:last_period + 1]
end = start + last_period + 1
chunks.append(chunk.strip())
start = end - self.chunk_overlap # 重叠部分
return chunks
# 示例
text = """Python 是一种高级编程语言。它由 Guido van Rossum 创建。
Python 的设计哲学强调代码的可读性。Python 支持多种编程范式。"""
splitter = SimpleTextSplitter(chunk_size=50, chunk_overlap=10)
chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks, 1):
print(f"块{i}: {chunk}")输出：
块1: Python 是一种高级编程语言。它由 Guido van Rossum 创建。
块2: Rossum 创建。Python 的设计哲学强调代码的可读性。
块3: 可读性。Python 支持多种编程范式。
注意重叠部分： 块2包含了块1的结尾，避免语义被切断。
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 优先在段落、句子边界切分
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", "。", ".", " ", ""] # 优先级
)
chunks = splitter.split_text(text)切分优先级：
- 段落边界（
\n\n） - 行边界（
\n） - 句子边界（
。.） - 单词边界（空格）
- 字符边界（最后手段）
| 参数 | 推荐值 | 说明 |
|---|---|---|
| chunk_size | 500-1500 字符 | 太小：上下文不足 太大：噪音过多 |
| chunk_overlap | 10-20% | chunk_size 的 10-20% |
不同场景的建议：
| 场景 | chunk_size | chunk_overlap | 原因 |
|---|---|---|---|
| 技术文档 | 800-1200 | 150-200 | 需要完整的代码示例 |
| 对话记录 | 400-600 | 50-100 | 对话轮次较短 |
| 法律文件 | 1000-1500 | 200-300 | 条款需要完整上下文 |
| 新闻文章 | 500-800 | 100-150 | 段落结构清晰 |
from dataclasses import dataclass
from typing import Dict
@dataclass
class Document:
"""文档块"""
content: str
metadata: Dict
# 示例
doc = Document(
content="Python 是一种高级编程语言...",
metadata={
'source': 'python_tutorial.pdf', # 来源文件
'page': 1, # 页码
'chapter': '第一章：Python 简介', # 章节
'author': '张三', # 作者
'created_at': '2024-01-01', # 创建时间
'chunk_index': 0, # 当前是第几块
'chunk_count': 10 # 总共多少块
}
)为什么需要元数据？
- 来源追溯 - 告诉用户答案来自哪个文档
- 过滤检索 - 只搜索特定章节或作者的文档
- 相关性排序 - 优先返回最新或权威的内容
- 调试追踪 - 快速定位问题文档
class SimpleTextSplitter:
def create_documents(
self,
text: str,
metadata: Dict = None
) -> list[Document]:
"""将文本切分为 Document 列表"""
if metadata is None:
metadata = {}
chunks = self.split_text(text)
documents = []
for i, chunk in enumerate(chunks):
doc_metadata = metadata.copy()
doc_metadata['chunk_index'] = i
doc_metadata['chunk_count'] = len(chunks)
documents.append(Document(
content=chunk,
metadata=doc_metadata
))
return documents
# 使用示例
splitter = SimpleTextSplitter(chunk_size=500, chunk_overlap=50)
text = open('python_tutorial.txt', 'r', encoding='utf-8').read()
docs = splitter.create_documents(
text,
metadata={
'source': 'python_tutorial.txt',
'category': 'programming',
'author': '张三'
}
)
print(f"切分为 {len(docs)} 个文档块")
for doc in docs[:2]:
print(doc)问题： 文档数量大时，内存存不下所有向量
1万个文档 × 1536维 × 4字节/float = 61 MB ✅ 可以
10万个文档 = 614 MB ⚠️ 勉强
100万个文档 = 6.1 GB ❌ 内存不够
向量数据库的作用：
- ✅ 持久化存储 - 向量存在磁盘，不占内存
- ✅ 高效检索 - ANN（近似最近邻）算法，毫秒级搜索
- ✅ 元数据过滤 - 支持条件查询
- ✅ 分布式扩展 - 支持海量数据
| 数据库 | 类型 | 适用场景 | 特点 |
|---|---|---|---|
| ChromaDB | 嵌入式 | 开发、小规模（<10万） | 轻量、易用、无需部署 |
| Qdrant | 独立服务 | 生产、中等规模（10万-100万） | 高性能、分布式 |
| Milvus | 分布式 | 大规模（100万+） | GPU 加速、云原生 |
| Pinecone | 云服务 | 全托管 | 免运维、按量付费 |
| pgvector | PostgreSQL插件 | 已有PG数据库 | SQL兼容、事务支持 |
本教程使用 ChromaDB： 零配置，适合学习和快速开发。
参考代码：examples/03-rag/03_vector_database.py
pip install chromadbimport chromadb
# 1. 初始化客户端
client = chromadb.PersistentClient(path="./my_vectordb")
# 2. 创建集合（Collection）
collection = client.get_or_create_collection(
name="my_documents",
metadata={"description": "我的文档集合"}
)
# 3. 添加文档
collection.add(
documents=[
"Python 是一种编程语言",
"机器学习是AI的分支",
"向量数据库用于存储向量"
],
embeddings=[
get_embedding("Python 是一种编程语言"),
get_embedding("机器学习是AI的分支"),
get_embedding("向量数据库用于存储向量")
],
metadatas=[
{"source": "doc1", "category": "programming"},
{"source": "doc2", "category": "AI"},
{"source": "doc3", "category": "database"}
],
ids=["id1", "id2", "id3"]
)
# 4. 检索
query = "什么是编程语言？"
query_embedding = get_embedding(query)
results = collection.query(
query_embeddings=[query_embedding],
n_results=2 # 返回前2个
)
print("检索结果：")
for doc, metadata in zip(results['documents'][0], results['metadatas'][0]):
print(f" [{metadata['source']}] {doc}")输出：
检索结果：
[doc1] Python 是一种编程语言
[doc2] 机器学习是AI的分支
class KnowledgeBase:
"""知识库管理器"""
def __init__(
self,
collection_name: str = "kb",
persist_dir: str = "./chroma_db"
):
self.collection_name = collection_name
# 初始化 ChromaDB
self.chroma_client = chromadb.PersistentClient(path=persist_dir)
self.collection = self.chroma_client.get_or_create_collection(
name=collection_name
)
print(f"✓ 知识库初始化完成：{collection_name}")
print(f" 当前文档数：{self.collection.count()}")
def add_documents(self, documents: list[Document]):
"""添加文档到知识库"""
if not documents:
return
# 提取内容和元数据
contents = [doc.content for doc in documents]
metadatas = [doc.metadata for doc in documents]
# 生成ID
start_id = self.collection.count()
ids = [f"doc_{start_id + i}" for i in range(len(documents))]
# 批量生成向量
embeddings = get_embeddings_batch(contents)
# 添加到向量库
self.collection.add(
documents=contents,
embeddings=embeddings,
metadatas=metadatas,
ids=ids
)
print(f"✅ 已添加 {len(documents)} 个文档")
def search(
self,
query: str,
top_k: int = 3,
filter_metadata: dict = None
) -> list[tuple]:
"""检索相关文档"""
# 生成查询向量
query_embedding = get_embedding(query)
# 执行检索
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
where=filter_metadata, # 元数据过滤
include=["documents", "metadatas", "distances"]
)
# 转换为 (文档, 元数据, 相似度) 格式
retrieval_results = []
for doc, metadata, distance in zip(
results['documents'][0],
results['metadatas'][0],
results['distances'][0]
):
similarity = 1 - distance # 距离转相似度
retrieval_results.append((doc, metadata, similarity))
return retrieval_results
def clear(self):
"""清空知识库"""
self.chroma_client.delete_collection(self.collection_name)
self.collection = self.chroma_client.create_collection(
name=self.collection_name
)
# 使用示例
kb = KnowledgeBase(collection_name="my_kb")
# 添加文档
docs = [
Document("Python 是一种编程语言", {"category": "programming"}),
Document("机器学习是AI的分支", {"category": "AI"}),
]
kb.add_documents(docs)
# 检索
results = kb.search("编程语言", top_k=2)
for doc, metadata, similarity in results:
print(f"[{similarity:.3f}] {doc}")# 只搜索特定分类的文档
results = kb.search(
query="Python",
top_k=5,
filter_metadata={"category": "programming"} # 只返回编程类别
)
# 复杂过滤条件
results = kb.search(
query="Python",
filter_metadata={
"$and": [
{"category": "programming"},
{"year": {"$gte": 2020}} # 2020年及以后
]
}
)参考代码：examples/03-rag/04_complete_rag_system.py
class RAGSystem:
"""RAG 问答系统"""
def __init__(
self,
knowledge_base: KnowledgeBase,
model: str = "gpt-4o-mini"
):
self.kb = knowledge_base
self.model = model
def ask(
self,
question: str,
top_k: int = 3,
show_sources: bool = True
) -> tuple[str, list]:
"""
提问并获取答案
Returns:
(答案, 检索到的文档列表)
"""
# 1. 检索相关文档
retrieved_docs = self.kb.search(question, top_k=top_k)
if not retrieved_docs:
return "抱歉，我没有找到相关信息。", []
# 2. 构建上下文
context = self._build_context(retrieved_docs)
# 3. 构建 Prompt
prompt = self._build_prompt(question, context)
# 4. 调用 LLM 生成答案
answer = self._generate_answer(prompt)
return answer, retrieved_docs
def _build_context(self, docs: list) -> str:
"""构建上下文"""
context_parts = []
for i, (doc, metadata, similarity) in enumerate(docs, 1):
context_parts.append(f"[文档 {i}]\n{doc}")
return "\n\n".join(context_parts)
def _build_prompt(self, question: str, context: str) -> str:
"""构建 Prompt"""
prompt = f"""你是一个专业的知识助手。请基于以下文档回答用户的问题。
**重要规则：**
1. 只使用提供的文档中的信息回答
2. 如果文档中没有相关信息，明确告知用户
3. 回答要准确、简洁、有条理
4. 可以引用文档编号（如"根据文档1..."）
**文档内容：**
{context}
**用户问题：**
{question}
**回答：**"""
return prompt
def _generate_answer(self, prompt: str) -> str:
"""调用 LLM 生成答案"""
response = openai_client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "你是一个专业的知识助手，基于提供的文档准确回答问题。"
},
{"role": "user", "content": prompt}
],
temperature=0.3, # 降低随机性
max_tokens=500
)
return response.choices[0].message.content# 步骤 1：准备知识内容
knowledge_content = {
"Python 编程": """
Python 是一种高级编程语言，由 Guido van Rossum 于 1991 年创建。
Python 的主要特点包括：
1. 简洁易学：语法简单，适合初学者
2. 功能强大：拥有丰富的标准库和第三方库
3. 应用广泛：Web 开发、数据分析、机器学习、自动化等
""",
"机器学习基础": """
机器学习是人工智能的一个分支，使计算机能够从数据中学习。
机器学习的三种主要类型：
1. 监督学习：使用标注数据训练模型
2. 无监督学习：从无标注数据中发现模式
3. 强化学习：通过与环境交互学习最优策略
"""
}
# 步骤 2：初始化知识库
kb = KnowledgeBase(collection_name="rag_demo")
kb.clear()
# 步骤 3：切分并添加文档
splitter = SimpleTextSplitter(chunk_size=300, chunk_overlap=50)
all_documents = []
for title, content in knowledge_content.items():
docs = splitter.create_documents(
content,
metadata={'source': title}
)
all_documents.extend(docs)
kb.add_documents(all_documents)
# 步骤 4：初始化 RAG 系统
rag = RAGSystem(kb, model="gpt-4o-mini")
# 步骤 5：提问
answer, sources = rag.ask("Python 有哪些特点？", top_k=2)
print(f"💡 答案：\n{answer}")
print(f"\n📚 参考来源：")
for doc, metadata, similarity in sources:
print(f" [{metadata['source']}] (相似度: {similarity:.3f})")输出：
💡 答案：
根据文档1，Python 的主要特点包括：
1. 简洁易学：语法简单，适合初学者
2. 功能强大：拥有丰富的标准库和第三方库
3. 应用广泛：适用于 Web 开发、数据分析、机器学习、自动化等领域
📚 参考来源：
[Python 编程] (相似度: 0.872)
[机器学习基础] (相似度: 0.654)
# 模糊、容易编造信息
prompt = f"""回答问题：{question}
参考：{context}"""问题：
- LLM 可能忽略文档，使用自己的知识
- 没有明确要求引用来源
- 没有处理"文档中没有答案"的情况
prompt = f"""你是一个专业的知识助手。请**严格基于**以下文档回答问题。
**硬性规则：**
1. ⚠️ 只使用文档中的信息，不要使用文档外的知识
2. ⚠️ 如果文档中没有相关信息，回复"文档中没有相关信息"
3. ✅ 回答时引用文档编号（如"根据文档1..."）
4. ✅ 保持简洁、准确、有条理
**文档内容：**
{context}
**用户问题：**
{question}
**你的回答：**"""def ask(self, question: str, top_k: int = 3):
# 检索文档
docs = self.kb.search(question, top_k=top_k)
# 情况1：没有找到任何文档
if not docs:
return "抱歉，我的知识库中没有相关信息。"
# 情况2：相似度太低
if docs[0][2] < 0.7: # 最高相似度 < 0.7
return "抱歉，我找到的信息可能不够相关。您可以换个方式提问。"
# 情况3：正常生成答案
context = self._build_context(docs)
prompt = self._build_prompt(question, context)
answer = self._generate_answer(prompt)
return answer参考代码：examples/03-rag/05_advanced_rag.py
| 问题 | 影响 | 示例 |
|---|---|---|
| 查询表述多样 | 召回率低 | "Python怎么用" vs "Python的使用方法" |
| 向量检索不精确 | 精确率低 | 返回语义相似但不相关的文档 |
| 缺少来源标注 | 可信度低 | 用户不知道答案从哪来 |
核心思想： 生成多个查询变体，增加召回率
def rewrite_query(original_query: str) -> list[str]:
"""
将一个查询改写为多个变体
Args:
original_query: 原始查询
Returns:
查询变体列表
"""
prompt = f"""请将以下查询改写为3个不同的表述方式，保持语义不变。
原始查询：{original_query}
要求：
1. 每个改写独立一行
2. 使用不同的词汇和句式
3. 保持查询意图不变
改写结果："""
response = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
rewritten = response.choices[0].message.content.strip().split('\n')
return [original_query] + [q.strip() for q in rewritten if q.strip()]
# 示例
query = "Python 适合做什么？"
queries = rewrite_query(query)
print("原始查询：", query)
print("\n改写后：")
for i, q in enumerate(queries[1:], 1):
print(f"{i}. {q}")输出：
原始查询： Python 适合做什么？
改写后：
1. Python 的主要应用场景有哪些？
2. Python 编程语言适用于什么领域？
3. 使用 Python 可以开发什么类型的项目？
多查询融合检索：
def search_with_rewrite(query: str, kb: KnowledgeBase, top_k: int = 5):
"""使用查询改写提高召回率"""
# 1. 生成查询变体
queries = rewrite_query(query)
# 2. 对每个查询执行检索
all_results = []
for q in queries:
results = kb.search(q, top_k=top_k)
all_results.extend(results)
# 3. 去重并按相似度排序
unique_docs = {}
for doc, metadata, similarity in all_results:
doc_id = metadata.get('id', doc[:50])
if doc_id not in unique_docs or similarity > unique_docs[doc_id][2]:
unique_docs[doc_id] = (doc, metadata, similarity)
# 4. 返回 Top K
sorted_results = sorted(
unique_docs.values(),
key=lambda x: x[2],
reverse=True
)
return sorted_results[:top_k]效果：
- 召回率提升 15-30%
- 适合口语化、模糊查询
问题： 向量相似度不等于真实相关性
初始检索（向量相似度）：
1. 文档A [0.85] - 包含"Python"，但讲的是历史
2. 文档B [0.83] - 包含"编程"，但讲的是Java
3. 文档C [0.82] - 直接回答"Python适合做什么"
重排序后（语义相关性）：
1. 文档C [最相关] - 直接回答问题
2. 文档A [相关]
3. 文档B [部分相关]
实现：
def rerank_results(
query: str,
documents: list,
top_k: int = 3
) -> list:
"""
使用 LLM 对检索结果进行重排序
Args:
query: 查询
documents: 初始检索结果
top_k: 最终返回数量
Returns:
重排序后的结果
"""
# 1. 构建重排序 Prompt
docs_text = "\n\n".join([
f"[文档 {i+1}]\n{doc}"
for i, (doc, _, _) in enumerate(documents)
])
prompt = f"""请为以下文档评估与查询的相关性，并按相关性排序。
查询：{query}
文档：
{docs_text}
要求：
1. 输出格式：文档编号,相关性分数(0-10)
2. 按相关性从高到低排序
3. 只输出排序结果，不要解释
排序结果："""
response = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
# 2. 解析排序结果
lines = response.choices[0].message.content.strip().split('\n')
reranked = []
for line in lines[:top_k]:
try:
doc_num = int(line.split(',')[0]) - 1
score = float(line.split(',')[1])
reranked.append((*documents[doc_num], score))
except:
continue
return reranked
# 使用示例
# 第一阶段：向量检索（召回更多）
initial_results = kb.search(query, top_k=10)
# 第二阶段：LLM 重排序（精确排序）
final_results = rerank_results(query, initial_results, top_k=3)效果：
- 精确率提升 10-20%
- 成本增加（多一次 LLM 调用）
核心思想： 向量检索 + 关键词检索
def hybrid_search(
query: str,
kb: KnowledgeBase,
top_k: int = 5,
alpha: float = 0.7 # 向量检索权重
) -> list:
"""
混合检索：结合向量相似度和关键词匹配
Args:
alpha: 向量检索权重（0-1），1-alpha 为关键词权重
"""
# 1. 向量检索
vector_results = kb.search(query, top_k=top_k * 2)
# 2. 关键词检索（简单实现）
keywords = query.split()
keyword_scores = {}
for doc, metadata, _ in vector_results:
score = sum(1 for kw in keywords if kw in doc)
doc_id = metadata.get('id', doc[:50])
keyword_scores[doc_id] = score / len(keywords)
# 3. 融合分数
hybrid_results = []
for doc, metadata, vec_score in vector_results:
doc_id = metadata.get('id', doc[:50])
kw_score = keyword_scores.get(doc_id, 0)
# 加权融合
final_score = alpha * vec_score + (1 - alpha) * kw_score
hybrid_results.append((doc, metadata, final_score))
# 4. 排序并返回
hybrid_results.sort(key=lambda x: x[2], reverse=True)
return hybrid_results[:top_k]让 LLM 在答案中标注来源：
def _build_prompt_with_citations(self, question: str, context: str) -> str:
"""构建带引用的 Prompt"""
prompt = f"""你是一个专业的知识助手。请基于以下文档回答问题。
**重要要求：**
1. 只使用文档中的信息
2. 在答案中用[文档X]标注信息来源
3. 每个关键信息都要标注来源
**示例：**
问题：Python 有哪些特点？
答案：Python 具有简洁易学的特点[文档1]，拥有丰富的标准库[文档2]。
**文档内容：**
{context}
**用户问题：**
{question}
**你的回答：**"""
return prompt
# 生成的答案示例
"""
Python 的主要特点包括：
1. 简洁易学[文档1]
2. 功能强大，拥有丰富的标准库[文档1]
3. 应用广泛，适用于 Web 开发、数据分析等[文档1]
"""class AdvancedRAGSystem:
"""高级 RAG 系统"""
def ask(
self,
question: str,
use_rewrite: bool = True,
use_rerank: bool = True,
top_k: int = 3
) -> tuple[str, list]:
"""
高级问答
Args:
use_rewrite: 是否使用查询改写
use_rerank: 是否使用重排序
"""
# 1. 查询改写（可选）
if use_rewrite:
queries = rewrite_query(question)
initial_results = []
for q in queries:
results = self.kb.search(q, top_k=top_k * 2)
initial_results.extend(results)
# 去重
initial_results = self._deduplicate(initial_results)
else:
initial_results = self.kb.search(question, top_k=top_k * 2)
# 2. 重排序（可选）
if use_rerank:
final_results = rerank_results(
question,
initial_results,
top_k=top_k
)
else:
final_results = initial_results[:top_k]
# 3. 生成答案（带引用）
context = self._build_context(final_results)
prompt = self._build_prompt_with_citations(question, context)
answer = self._generate_answer(prompt)
return answer, final_results需求： 为公司内部技术文档构建问答系统
功能：
- 支持 Markdown、PDF 文档
- 按项目/模块分类检索
- 标注答案来源（文件名 + 行号）
实现步骤：
# 1. 加载文档
from pathlib import Path
def load_markdown_files(directory: str) -> list[Document]:
"""加载目录下所有 Markdown 文件"""
docs = []
for md_file in Path(directory).rglob("*.md"):
content = md_file.read_text(encoding='utf-8')
# 按二级标题切分
sections = content.split('\n## ')
for i, section in enumerate(sections):
if i == 0: # 跳过文件头
continue
title = section.split('\n')[0]
body = '\n'.join(section.split('\n')[1:])
docs.append(Document(
content=f"## {title}\n{body}",
metadata={
'source': str(md_file),
'section': title,
'project': md_file.parent.name
}
))
return docs
# 2. 构建知识库
kb = KnowledgeBase(collection_name="tech_docs")
docs = load_markdown_files("./docs")
kb.add_documents(docs)
# 3. 按项目检索
results = kb.search(
"如何配置数据库连接？",
filter_metadata={"project": "backend"}
)需求： 基于客服FAQ构建自动问答
功能：
- 处理同义问题（"退款" = "退钱" = "申请退款"）
- 多轮对话支持
- 置信度评分
实现要点：
class CustomerServiceRAG:
"""客服 RAG 系统"""
def __init__(self, kb: KnowledgeBase):
self.kb = kb
self.conversation_history = []
def ask(self, question: str) -> dict:
"""
客服问答
Returns:
{
'answer': str,
'confidence': float, # 置信度 0-1
'sources': list,
'need_human': bool # 是否需要人工介入
}
"""
# 1. 查询改写（处理同义词）
queries = self._expand_synonyms(question)
# 2. 检索
results = []
for q in queries:
results.extend(self.kb.search(q, top_k=3))
results = self._deduplicate(results)[:3]
# 3. 置信度评分
confidence = results[0][2] if results else 0
# 4. 判断是否需要人工
need_human = confidence < 0.75 or not results
if need_human:
return {
'answer': "抱歉，这个问题我不太确定，为您转接人工客服。",
'confidence': confidence,
'sources': [],
'need_human': True
}
# 5. 生成答案
context = self._build_context(results)
answer = self._generate_answer(question, context)
return {
'answer': answer,
'confidence': confidence,
'sources': [r[1] for r in results],
'need_human': False
}
def _expand_synonyms(self, query: str) -> list[str]:
"""同义词扩展"""
synonyms = {
'退款': ['退钱', '申请退款', '要求退款'],
'发货': ['什么时候发货', '几天发货', '发货时间'],
'质量': ['质量问题', '产品问题', '有问题']
}
queries = [query]
for key, values in synonyms.items():
if key in query:
queries.extend([query.replace(key, v) for v in values])
return queries需求： 帮助研究人员快速理解论文
功能：
- PDF 解析
- 引用追踪
- 多论文对比
# 1. PDF 解析
import pypdf
def load_pdf(pdf_path: str) -> list[Document]:
"""加载 PDF 并按页切分"""
reader = pypdf.PdfReader(pdf_path)
docs = []
for page_num, page in enumerate(reader.pages, 1):
text = page.extract_text()
docs.append(Document(
content=text,
metadata={
'source': pdf_path,
'page': page_num,
'total_pages': len(reader.pages)
}
))
return docs
# 2. 引用追踪
def extract_citations(text: str) -> list[str]:
"""提取论文引用"""
import re
# 匹配 [1], [2-5] 格式
citations = re.findall(r'\[(\d+(?:-\d+)?)\]', text)
return citations
# 3. 多论文对比
def compare_papers(question: str, paper_ids: list[str], kb: KnowledgeBase):
"""对比多篇论文对某个问题的观点"""
results_by_paper = {}
for paper_id in paper_ids:
results = kb.search(
question,
filter_metadata={'paper_id': paper_id},
top_k=2
)
results_by_paper[paper_id] = results
# 生成对比报告
prompt = f"""请对比以下论文对"{question}"的观点：
{_format_comparison(results_by_paper)}
总结它们的异同点。"""
return generate_answer(prompt)# 保留丰富的元数据
Document(
content="Python 的特点包括...",
metadata={
'source': 'python_tutorial.pdf',
'page': 12,
'chapter': '第2章：Python基础',
'section': '2.1 语言特性',
'author': '张三',
'created_at': '2024-01-01',
'version': '2.0',
'category': 'programming',
'tags': ['python', 'tutorial', 'beginner']
}
)# 元数据不足
Document(
content="Python 的特点包括...",
metadata={} # 丢失所有上下文信息
)"""你是一个专业的知识助手。请**严格基于**以下文档回答问题。
**硬性规则：**
1. ⚠️ 只使用文档中的信息，不要使用文档外的知识
2. ⚠️ 如果文档中没有相关信息，明确告知用户
3. ✅ 回答要准确、简洁、有条理
4. ✅ 用[文档X]标注信息来源
**文档内容：**
{context}
**用户问题：**
{question}
**你的回答：**"""要点：
⚠️ 使用警告符号强调约束- ✅ 使用示例说明期望格式
- 📋 明确处理边界情况
# 模糊、容易编造信息
"""回答问题：{question}
参考：{context}"""# ✅ 好：先召回多，再精确排序
# 第一阶段：向量检索（召回）
initial_results = kb.search(query, top_k=10)
# 第二阶段：LLM 重排序（精确）
final_results = rerank_results(query, initial_results, top_k=3)# ❌ 差：直接检索少量
results = kb.search(query, top_k=3) # 可能遗漏相关文档# ✅ 好：过滤低相关结果
def search_with_threshold(query: str, kb: KnowledgeBase, threshold: float = 0.7):
results = kb.search(query, top_k=5)
return [r for r in results if r[2] >= threshold]# 避免返回的文档过于相似
def diverse_search(query: str, kb: KnowledgeBase, top_k: int = 5):
"""返回多样化的检索结果"""
results = kb.search(query, top_k=top_k * 3)
diverse_results = [results[0]] # 第一个一定是最相关的
for result in results[1:]:
# 检查与已选结果的相似度
if all(
cosine_similarity(result[3], selected[3]) < 0.9
for selected in diverse_results
):
diverse_results.append(result)
if len(diverse_results) >= top_k:
break
return diverse_resultsclass RAGSystem:
def ask(self, question: str):
try:
# 1. 输入验证
if not question or len(question.strip()) < 3:
return "请输入有效的问题（至少3个字符）"
# 2. 检索
results = self.kb.search(question, top_k=3)
# 3. 检查检索质量
if not results:
return "抱歉，我的知识库中没有相关信息。"
if results[0][2] < 0.65: # 相似度阈值
return "抱歉，我找到的信息可能不够相关。您可以：\n1. 换个方式提问\n2. 提供更多上下文"
# 4. 生成答案
answer = self._generate_answer_with_retry(question, results)
return answer
except Exception as e:
logger.error(f"RAG error: {e}")
return "抱歉，系统遇到问题，请稍后再试。"
def _generate_answer_with_retry(self, question, results, max_retries=3):
"""带重试的答案生成"""
for attempt in range(max_retries):
try:
return self._generate_answer(question, results)
except openai.RateLimitError:
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # 指数退避
else:
raisefrom functools import lru_cache
@lru_cache(maxsize=100)
def get_embedding_cached(text: str) -> tuple:
"""缓存 Embedding 结果"""
embedding = get_embedding(text)
return tuple(embedding) # list 不能作为缓存键，转为 tupleimport asyncio
async def search_async(queries: list[str], kb: KnowledgeBase):
"""并行检索多个查询"""
tasks = [
asyncio.create_task(kb.search_async(q))
for q in queries
]
results = await asyncio.gather(*tasks)
return results# ✅ 快：批量向量化
embeddings = get_embeddings_batch(documents) # 1次API调用
# ❌ 慢：逐个向量化
embeddings = [get_embedding(doc) for doc in documents] # N次API调用# text-embedding-3-small: $0.02/1M tokens
# text-embedding-3-large: $0.13/1M tokens (贵6.5倍)
# 大多数场景用 small 就够了
model = "text-embedding-3-small"# 向量只生成一次，持久化存储
# 不要每次启动都重新生成# ❌ 差：检索太多
results = kb.search(query, top_k=20) # 传给LLM太多上下文，浪费tokens
# ✅ 好：只检索必要的
results = kb.search(query, top_k=3) # 通常3-5个就够# 优化文档内容，去除冗余
def clean_document(text: str) -> str:
"""清理文档，减少 token 消耗"""
# 移除多余空白
text = re.sub(r'\s+', ' ', text)
# 移除重复标点
text = re.sub(r'([。！？])\1+', r'\1', text)
return text.strip()A: 检查以下几点：
-
文档切分是否合理
# chunk_size 太大 → 噪音多 # chunk_size 太小 → 上下文不足 # 建议：500-1500 字符
-
查询表述是否清晰
# ❌ 差："Python" # ✅ 好："Python 有哪些主要特点？"
-
相似度阈值
# 过滤低相关结果 if results[0][2] < 0.7: return "未找到足够相关的信息"
-
使用查询改写
# 生成多个查询变体 queries = rewrite_query(original_query)
A: 优化 Prompt
# ✅ 强约束 Prompt
prompt = """**硬性规则：**
1. ⚠️ 只使用文档中的信息，不要使用文档外的知识
2. ⚠️ 如果文档中没有，明确说"文档中没有相关信息"
3. ⚠️ 不要编造任何信息
**文档内容：**
{context}
**问题：**
{question}"""
# 同时降低 temperature
temperature=0.1 # 减少创造性A: 分层检索
# 策略1：章节 + 段落两层结构
# 1. 先检索相关章节
chapter_results = search_chapters(query, top_k=2)
# 2. 在章节内检索段落
paragraph_results = []
for chapter in chapter_results:
paragraphs = search_paragraphs(query, chapter, top_k=3)
paragraph_results.extend(paragraphs)
# 策略2：滑动窗口
# chunk_size=1000, chunk_overlap=200
# 相邻块有重叠，避免信息断裂A: 批量处理
# ✅ 快：批量调用（10倍+提速）
embeddings = get_embeddings_batch(documents)
# ❌ 慢：逐个调用
embeddings = [get_embedding(doc) for doc in documents]
# 注意：批量最多 2048 个文本
# 如果超过，需要分批
def batch_embed_large(documents: list[str], batch_size: int = 2000):
all_embeddings = []
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
embeddings = get_embeddings_batch(batch)
all_embeddings.extend(embeddings)
return all_embeddingsA: 关键指标
# 1. 召回率（Recall）
# 相关文档中有多少被检索到
recall = len(检索到的相关文档) / len(所有相关文档)
# 2. 精确率（Precision）
# 检索到的文档中有多少是相关的
precision = len(检索到的相关文档) / len(所有检索到的文档)
# 3. MRR（Mean Reciprocal Rank）
# 第一个相关文档的排名倒数
# 例如：第1个是相关的 → MRR = 1
# 第3个是相关的 → MRR = 1/3
# 4. 答案质量（人工评估）
# - 准确性：答案是否正确
# - 完整性：是否遗漏关键信息
# - 可追溯性：是否标注了来源A: OpenAI Embedding 天然支持多语言
# 中英文混合检索
docs_cn = ["Python 是一种编程语言"]
docs_en = ["Python is a programming language"]
kb.add_documents(docs_cn + docs_en)
# 中文查询可以检索到英文文档（反之亦然）
results = kb.search("什么是 Python？")
# 可能返回中文和英文文档A: 增量更新
# 方案1：追加新文档
new_docs = load_new_documents()
kb.add_documents(new_docs)
# 方案2：删除旧文档，添加新版本
# ChromaDB 支持按 ID 删除
kb.collection.delete(ids=["old_doc_1", "old_doc_2"])
kb.add_documents(updated_docs)
# 方案3：定期重建
# 如果文档结构变化大，推荐完全重建
kb.clear()
all_docs = load_all_documents()
kb.add_documents(all_docs)- 理解 RAG 的核心原理和工作流程
- 能够使用 OpenAI API 生成 Embedding
- 会计算余弦相似度
- 实现简单的语义搜索
- 完成
01_basic_embedding.py示例
- 掌握文档切分策略（chunk_size、chunk_overlap）
- 理解元数据的作用
- 能够使用 ChromaDB 存储和检索向量
- 实现完整的知识库管理器
- 完成
02_document_chunking.py和03_vector_database.py
- 构建端到端 RAG 问答系统
- 掌握 Prompt 工程技巧
- 理解高级技术（查询改写、重排序）
- 能够处理边界情况和错误
- 完成所有示例代码和至少 1 个实战项目
任务： 实现一个代码片段搜索引擎
# 数据集
code_snippets = [
"def hello():\n print('Hello World')",
"class Person:\n def __init__(self, name):\n self.name = name",
"import pandas as pd\ndf = pd.read_csv('data.csv')",
"for i in range(10):\n print(i)",
]
# TODO: 实现搜索函数
def search_code(query: str, top_k: int = 2):
"""搜索最相关的代码片段"""
pass
# 测试
results = search_code("如何定义类")
# 应该返回包含 class 的代码任务： 构建客服 FAQ 问答系统
faqs = [
("如何退款？", "退款需要在订单页面点击退款按钮..."),
("发货需要多久？", "通常3-5个工作日内发货..."),
("支持哪些支付方式？", "支持微信、支付宝、信用卡..."),
]
# TODO:
# 1. 构建知识库
# 2. 实现问答函数
# 3. 处理同义问题（"退钱" = "退款"）任务： 构建 Markdown 文档问答系统
要求：
- 加载目录下所有
.md文件 - 按二级标题切分
- 支持按文件名过滤
- 答案中标注来源（文件名 + 章节）
任务： 对比多个技术方案文档
# 场景：对比 "Python vs Java" 在不同文档中的描述
# 要求：
# 1. 从多个文档中检索相关段落
# 2. 生成对比表格
# 3. 总结异同点-
LangChain RAG
pip install langchain
- 更高级的抽象
- 丰富的文档加载器
- 内置的检索优化
-
其他向量数据库
- Qdrant - 生产级性能
- Milvus - 大规模分布式
- Pinecone - 云原生托管
-
高级技术
- HyDE (Hypothetical Document Embeddings)
- Self-Query (自动提取元数据过滤条件)
- Parent Document Retriever (检索小块，返回大块)
- 个人知识库 - 管理笔记、文章、书籍摘要
- 代码搜索引擎 - 搜索项目代码和文档
- 客服机器人 - 基于公司FAQ的自动问答
- 论文阅读助手 - PDF解析 + 知识图谱
- Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks
- Dense Passage Retrieval for Open-Domain Question Answering
开始学习： 从 examples/03-rag/01_basic_embedding.py 开始，按顺序运行示例 📖
有问题？ 查看 examples/03-rag/README.md 中的详细说明和故障排除指南。