向量数据库一天快速入门教程¶
整理时间：2026年2月10日
目录¶
1. 什么是向量数据库？¶
1.1 向量数据库的定义¶
向量数据库是专门用于存储和检索高维向量数据的数据库系统。在AI和机器学习应用中，我们将文本、图像、音频等内容通过机器学习模型转换成向量（embeddings），这些向量包含了内容的语义信息。向量数据库能够在这些高维向量中进行高效的相似性搜索，找到与查询向量最相似的结果。
1.2 为什么需要向量数据库？¶
传统数据库的局限性： - 传统数据库（SQL、NoSQL）基于精确匹配和条件查询 - 无法理解语义相似性（"狗"和"puppy"无法关联） - 处理高维数据效率极低 - 大规模向量搜索需要逐一遍历，时间复杂度O(n)
向量数据库的优势： - 速度：在数十亿向量中毫秒级检索 - 规模：支持十亿级别向量高效管理 - 精度：近似最近邻搜索(ANN)达到95-99%召回率 - 语义理解：基于向量距离理解语义相似性 - 生产就绪：专为高吞吐量、低延迟应用设计
1.3 市场发展¶
据预测，向量数据库市场规模预计到2028年将达到43亿美元，这主要得益于生成式AI和RAG系统的快速发展。
2. 核心概念¶
2.1 向量（Vectors）¶
向量是由数字组成的数组（列表），用于表示数据的特征：
特点： - 维度：通常为384到1536维（现代 embedding 模型） - 每个维度代表数据的一个特征 - 相似的物品在向量空间中距离更近
2.2 Embedding（嵌入）¶
Embedding 是将文本、图像等内容转换为向量表示的过程：
常用Embedding模型： - OpenAI: text-embedding-ada-002, text-embedding-3-small/large - Sentence Transformers: all-MiniLM-L6-v2, paraphrase-multilingual-mpnet-base-v2 - Cohere: embed-english-v3.0, embed-multilingual-v3.0 - 开源: BGE, M3E, Chinese-roberta-wwm-ext
2.3 相似性搜索¶
向量数据库通过计算向量间的距离或相似度来找到最相似的结果：
常用距离度量： - 余弦相似度（Cosine Similarity）：[-1, 1]，越接近1越相似 - 欧氏距离（Euclidean Distance）：越小越相似 - 点积（Dot Product）：越大越相似
# 余弦相似度计算示例
import numpy as np
def cosine_similarity(v1, v2):
dot_product = np.dot(v1, v2)
norm_v1 = np.linalg.norm(v1)
norm_v2 = np.linalg.norm(v2)
return dot_product / (norm_v1 * norm_v2)
2.4 近似最近邻搜索（ANN）¶
由于精确搜索在海量数据中太慢，向量数据库使用近似最近邻（ANN）算法：
- 特点：牺牲少量精度换取大幅速度提升
- 典型召回率：95-99%（已足够满足大多数应用）
- 常用算法：HNSW、IVF、PQ、LSH
3. 主流向量数据库对比¶
3.1 主流产品概览¶
| 数据库 | 类型 | 开源 | 特点 | 适用场景 |
|---|---|---|---|---|
| Milvus | 专用向量DB | 是 | 开源之王，十亿级规模，分布式架构 | 大规模生产环境 |
| Pinecone | 专用向量DB | 否 | 全托管Serverless，零运维 | 快速上手，追求便捷 |
| Qdrant | 专用向量DB | 是 | 高性能，Rust开发，Rust/Python生态 | 需要高性能的场景 |
| Weaviate | 专用向量DB | 是 | 知识图谱集成，原生支持RAG | 语义搜索、推荐系统 |
| Chroma | 向量库 | 是 | 轻量级，Python优先 | 快速原型、个人项目 |
| pgvector | PostgreSQL扩展 | 是 | 与PostgreSQL集成 | 已有PostgreSQL的项目 |
| Elasticsearch | 全文搜索 | 是 | 向量搜索+KNN | ELK技术栈的项目 |
| OpenSearch | 全文搜索 | 是 | 向量搜索插件 | AWS生态 |
3.2 详细对比¶
Milvus 推荐生产环境¶
优势： - 完全开源（Apache 2.0） - 支持十亿级向量 - 成熟的分布式架构 - 丰富的生态系统（Attu, Birdwatcher） - 云原生（Kubernetes）
劣势： - 需要自行部署和运维 - 学习曲线较陡
安装：
Pinecone 推荐快速上手¶
优势： - 全托管Serverless，零运维 - 极致易用，5分钟上手 - 全球CDN，低延迟 - 稳定的性能表现
劣势： - 专有服务，供应商锁定 - 高用量时价格较贵 - 无法本地部署
Qdrant 推荐Rust/Python生态¶
优势： - 高性能（Rust开发） - 内存效率高 - 支持标量过滤 - 丰富的客户端库
劣势： - 相对较新 - 生态系统不如Milvus成熟
Weaviate 推荐RAG应用¶
优势： - 原生支持知识图谱 - 内置向量化模块 - 强大的GraphQL查询 - 模块化架构
劣势： - 资源消耗较大 - 复杂查询学习曲线陡
Chroma 推荐快速原型¶
优势： - 极其简单易用 - 纯Python实现 - 轻量级 - 适合学习和实验
劣势： - 功能相对简单 - 不适合生产环境大规模使用 - 持久化支持有限
安装：
pgvector 推荐PostgreSQL用户¶
优势： - 与PostgreSQL无缝集成 - 支持多种索引方式（IVFFlat、HNSW） - 事务支持 - 备份恢复简单
劣势： - PostgreSQL本身不是为向量优化 - 超大规模时性能不如专用向量DB
3.3 选择建议¶
学习/实验 → Chroma, pgvector
快速原型 → Pinecone, Weaviate
生产环境 → Milvus, Qdrant
已有PostgreSQL → pgvector
追求极致性能 → Qdrant, Milvus
4. 第一天学习计划¶
学习目标¶
- 理解向量数据库的核心概念
- 掌握至少一种向量数据库的安装和使用
- 构建一个简单的语义搜索应用
时间安排¶
| 时间段 | 内容 | 目标 |
|---|---|---|
| 09:00-10:00 | 学习核心概念 | 理解向量、Embedding、相似性搜索 |
| 10:00-11:00 | 搭建开发环境 | 安装Chroma或Qdrant，准备Python环境 |
| 11:00-12:00 | 第一个向量应用 | 实现基础的向量存储和搜索 |
| 14:00-15:30 | 深入学习索引 | 理解HNSW、IVF等索引技术 |
| 15:30-17:00 | RAG应用实践 | 构建一个简单的问答系统 |
| 17:00-18:00 | 总结与进阶 | 整理学习笔记，规划下一步学习 |
5. 实操教程¶
5.1 环境准备¶
# 创建虚拟环境
python -m venv vector-db-tutorial
source vector-db-tutorial/bin/activate # Linux/Mac
# 或
vector-db-tutorial\Scripts\activate # Windows
# 安装依赖
pip install chromadb sentence-transformers openai numpy qdrant-client langchain
5.2 基础示例：使用Chroma实现语义搜索¶
"""
第一个向量搜索应用 - 使用Chroma实现文档语义搜索
"""
import chromadb
from sentence_transformers import SentenceTransformer
# 1. 初始化Chroma数据库
client = chromadb.Client()
collection = client.create_collection("documents")
# 2. 准备示例文档
documents = [
"人工智能是计算机科学的一个分支，致力于创建能模仿人类智能的系统。",
"机器学习是AI的一个子领域，让计算机通过数据学习而无需明确编程。",
"深度学习是机器学习的一种方法，使用多层神经网络来学习数据的表示。",
"自然语言处理是AI领域，专注于计算机与人类语言的交互。",
"向量数据库是专门用于存储和搜索高维向量数据的数据库系统。",
"ChatGPT是由OpenAI开发的大型语言模型对话系统。",
"自动驾驶汽车使用AI技术来感知环境并做出驾驶决策。",
"推荐系统通过分析用户行为来推荐相关产品或内容。",
"计算机视觉让计算机能够从图像和视频中理解和提取信息。",
"强化学习是一种通过与环境交互来学习最优策略的机器学习方法。"
]
# 3. 使用Embedding模型生成向量
print("正在生成文档向量...")
model = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
embeddings = model.encode(documents)
# 4. 将文档和向量存入数据库
print("正在存储文档到向量数据库...")
collection.add(
documents=documents,
embeddings=embeddings.tolist(),
ids=[f"doc_{i}" for i in range(len(documents))]
)
print("✓ 成功存储 10 个文档到向量数据库\n")
# 5. 执行语义搜索
def search_documents(query, n_results=3):
"""搜索与查询语义最相似的文档"""
print(f"\n🔍 查询：{query}")
print("-" * 50)
# 生成查询向量
query_embedding = model.encode([query])
# 执行搜索
results = collection.query(
query_embeddings=query_embedding.tolist(),
n_results=n_results
)
# 显示结果
for i, (doc, distance) in enumerate(zip(
results['documents'][0],
results['distances'][0]
)):
similarity = 1 - distance # 转换为相似度
print(f"\n[{i+1}] 相关度：{similarity:.2%}")
print(f" 内容：{doc}")
# 测试搜索
print("=" * 50)
print("开始语义搜索测试")
print("=" * 50)
search_documents("什么是深度学习？")
search_documents("AI有哪些应用场景？")
search_documents("如何让计算机理解语言？")
5.3 使用Qdrant实现高级向量搜索¶
"""
使用Qdrant实现带过滤条件的向量搜索
"""
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct
from sentence_transformers import SentenceTransformer
# 1. 初始化Qdrant客户端
client = QdrantClient(":memory:") # 使用内存模式，也可以改为本地路径或远程地址
# 2. 创建Collection
client.create_collection(
collection_name="products",
vectors_config=VectorParams(
size=384, # 向量维度
distance=Distance.COSINE # 使用余弦距离
)
)
# 3. 准备带标签的产品数据
products = [
{"id": 1, "name": "iPhone 15 Pro", "category": "手机", "desc": "苹果最新旗舰智能手机"},
{"id": 2, "name": "Samsung Galaxy S24", "category": "手机", "desc": "三星高端安卓手机"},
{"id": 3, "name": "MacBook Pro 14", "category": "电脑", "desc": "苹果专业笔记本电脑"},
{"id": 4, "name": "Dell XPS 15", "category": "电脑", "desc": "戴尔高端Windows笔记本"},
{"id": 5, "name": "Sony WH-1000XM5", "category": "耳机", "desc": "索尼降噪无线耳机"},
{"id": 6, "name": "AirPods Pro", "category": "耳机", "desc": "苹果无线降噪耳机"},
{"id": 7, "name": "iPad Pro", "category": "平板", "desc": "苹果专业平板电脑"},
{"id": 8, "name": "Kindle Paperwhite", "category": "阅读器", "desc": "亚马逊电子阅读器"},
]
# 4. 生成向量并存储
print("正在处理产品数据...")
model = SentenceTransformer('all-MiniLM-L6-v2')
descriptions = [p["desc"] for p in products]
embeddings = model.encode(descriptions).tolist()
# 上传到Qdrant
client.upsert(
collection_name="products",
points=[
PointStruct(
id=product["id"],
vector=embedding,
payload={
"name": product["name"],
"category": product["category"],
"description": product["desc"]
}
)
for product, embedding in zip(products, embeddings)
]
)
print("✓ 已存储产品数据\n")
# 5. 执行搜索（带分类过滤）
def search_products(query, category=None, top_k=3):
"""搜索产品，支持按类别过滤"""
print(f"🔍 搜索：'{query}'" + (f" (类别: {category})" if category else ""))
print("-" * 60)
query_embedding = model.encode([query]).tolist()
# 构建过滤条件
filter_condition = None
if category:
filter_condition = {
"must": [
{
"key": "category",
"match": {"value": category}
}
]
}
# 执行搜索
results = client.search(
collection_name="products",
query_vector=query_embedding,
query_filter=filter_condition,
limit=top_k
)
# 显示结果
for hit in results:
print(f"📦 {hit.payload['name']} ({hit.payload['category']})")
print(f" 描述：{hit.payload['description']}")
print(f" 相似度：{hit.score:.2%}")
print()
# 测试搜索
print("=" * 60)
print("智能产品搜索演示")
print("=" * 60)
# 不带过滤的搜索
search_products("高端无线音频设备")
# 带类别过滤的搜索
search_products("便携办公设备", category="电脑")
search_samples = ["拍照好的手机", "适合阅读的设备"]
for query in search_samples:
search_products(query)
5.4 完整RAG应用示例¶
"""
简单的RAG（检索增强生成）应用示例
"""
import chromadb
from sentence_transformers import SentenceTransformer
from typing import List, Tuple
class SimpleRAG:
def __init__(self, documents: List[str]):
"""初始化RAG系统"""
print("🔧 初始化RAG系统...")
# 加载embedding模型
self.model = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
# 创建向量数据库
self.client = chromadb.Client()
self.collection = self.client.create_collection("knowledge_base")
# 存储文档
embeddings = self.model.encode(documents).tolist()
self.collection.add(
documents=documents,
embeddings=embeddings,
ids=[f"doc_{i}" for i in range(len(documents))]
)
print("✓ RAG系统初始化完成\n")
def retrieve(self, query: str, top_k: int = 3) -> List[str]:
"""检索相关文档"""
query_embedding = self.model.encode([query]).tolist()
results = self.collection.query(
query_embeddings=query_embedding,
n_results=top_k
)
return results['documents'][0]
def generate(self, query: str, context: List[str]) -> str:
"""生成回答"""
context_text = "\n\n".join(context)
prompt = f"""
基于以下上下文信息，回答用户的问题。如果上下文中没有相关信息，请说明无法从已知信息中回答。
上下文：
{context_text}
问题：{query}
回答：
"""
# 实际应用中，这里会调用LLM API
return f"（实际应用中，这里会调用LLM生成回答）\n参考上下文：{context[0][:100]}..."
def ask(self, query: str) -> Tuple[str, List[str]]:
"""问领主流程：检索 + 生成"""
print(f"❓ 问题：{query}")
print("-" * 50)
# 检索相关文档
relevant_docs = self.retrieve(query)
print(f"📚 检索到 {len(relevant_docs)} 篇相关文档")
for i, doc in enumerate(relevant_docs):
print(f" [{i+1}] {doc[:80]}...")
# 生成回答
answer = self.generate(query, relevant_docs)
print(f"\n💡 回答：{answer}")
return answer, relevant_docs
# 知识库数据
knowledge_base = [
"人工智能（AI）是计算机科学的一个分支，致力于创建能够模拟人类智能的系统。AI的应用领域包括机器学习、自然语言处理、计算机视觉等。",
"机器学习是AI的一个子领域，专注于开发能够从数据中学习的算法。常见的机器学习类型包括监督学习、无监督学习和强化学习。",
"深度学习是机器学习的一种方法，使用多层神经网络来学习数据的复杂表示。卷积神经网络（CNN）在图像识别方面表现出色，循环神经网络（RNN）适用于序列数据处理。",
"向量数据库是专门设计用于存储、索引和查询高维向量数据的数据库系统。它们是实现语义搜索和AI应用的关键基础设施。",
"RAG（检索增强生成）是一种结合检索和生成的技术。RAG系统首先从向量数据库中检索相关信息，然后将这些信息作为上下文提供给大语言模型，以生成更准确、更有根据的回答。",
"ChatGPT是由OpenAI开发的大型语言模型，基于GPT架构。它能够进行自然语言对话、回答问题、编写代码和创作内容。",
"Transformer架构是现代自然语言处理的核心技术。它使用自注意力机制来处理序列数据，已成为BERT、GPT等模型的基础。",
"LangChain是一个用于构建LLM应用的框架，它提供了链（Chain）、代理（Agent）和记忆（Memory）等抽象，帮助开发者快速构建复杂的AI应用。"
]
# 运行演示
print("=" * 60)
print(" 简单RAG系统演示")
print("=" * 60)
rag = SimpleRAG(knowledge_base)
print("=" * 60)
rag.ask("什么是深度学习？")
print("\n" + "=" * 60)
rag.ask("向量数据库有什么作用？")
print("\n" + "=" * 60)
rag.ask("RAG是如何工作的？")
6. 向量索引技术¶
6.1 为什么需要索引？¶
- 无索引：逐一遍历所有向量，O(n)复杂度
- 有索引：只搜索相关区域，O(log n)或近似常数时间
6.2 主流索引算法对比¶
| 索引 | 类型 | 搜索速度 | 内存占用 | 召回率 | 适用场景 |
|---|---|---|---|---|---|
| HNSW | 图索引 | 5星 | 4星 | 5星 | 通用场景，推荐首选 |
| IVF | 聚类索引 | 4星 | 5星 | 4星 | 大规模、低内存 |
| PQ | 量化索引 | 5星 | 5星 | 3星 | 超大规模 |
| LSH | 哈希索引 | 3星 | 3星 | 3星 | 近似搜索 |
6.3 HNSW（Hierarchical Navigable Small World）¶
原理： 构建多层图结构，高层快速定位，底层精确搜索
特点： - 搜索速度极快 - 召回率高 - 构建时间较长 - 内存占用较大
# 在Milvus中使用HNSW索引
from pymilvus import Collection, IndexType, MetricType
# 创建HNSW索引
index_params = {
"metric_type": MetricType.COSINE,
"index_type": IndexType.HNSW,
"params": {
"M": 16, # 每个节点的连接数
"efConstruction": 200 # 构建时的搜索宽度
}
}
collection.create_index(field_name="vector", index_params=index_params)
# 搜索时设置ef参数
search_params = {"ef": 64}
results = collection.search(data=query_vector, anns_field="vector", params=search_params, limit=10)
6.4 IVF（Inverted File Index）¶
原理： 先将向量聚类，搜索时只遍历相关聚类
特点： - 内存效率高 - 构建速度快 - 适合大规模数据 - 需要调优聚类数
# 在pgvector中使用IVFFlat索引
CREATE TABLE documents (
id SERIAL PRIMARY KEY,
embedding vector(384),
content TEXT
);
-- 创建IVFFlat索引
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
-- 查询时使用
SELECT id, content, embedding <=> query_vector AS distance
FROM documents
ORDER BY distance
LIMIT 10;
6.5 PQ（Product Quantization）¶
原理： 将向量分割成子向量，分别量化压缩
特点： - 极大减少内存占用（可减少10倍） - 支持超大规模数据 - 有精度损失 - 构建复杂
7. RAG应用实践¶
7.1 RAG架构深度解析¶
7.1.1 完整RAG系统架构¶
┌─────────────────────────────────────────────────────────────────────────────┐
│ RAG系统完整架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 离线处理管道（离线） │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │文档加载器 │──►│ 文本分块 │──►│向量化处理 │──►│ 向量索引存储 │ │ │
│ │ │ Loaders │ │Splitters │ │Embedding │ │Vector Store │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ │ │ │ │ │ │ │
│ │ PDF/DOCX 递归分割 OpenAI/ Pinecone/ │ │
│ │ HTML/MD Overlap Sentence Milvus/ │ │
│ │ Database Transformers Chroma │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 在线处理管道（实时） │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ 用户查询 │──►│查询向量化 │──►│ 相似度 │──►│ 获取上下文 │ │ │
│ │ │ Query │ │ Embed │ │ Search │ │ Context │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ │ │ │ │ │ │ │
│ │ "什么是深度 生成查询 向量数据库 原始文档 │ │
│ │ 学习？" 向量 检索 内容块 │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────────────┐ │ │
│ │ │ 提示词拼接（Prompt Engineering） │ │ │
│ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ 系统提示：你是一个专业的AI助手... │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ 上下文： │ │ │ │
│ │ │ │ 1. 深度学习是机器学习的一种方法... │ │ │ │
│ │ │ │ 2. 它使用多层神经网络... │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ 问题：什么是深度学习？ │ │ │ │
│ │ │ └─────────────────────────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌────────────────────────────────────────────────────────────────┐ │ │
│ │ │ LLM生成（Generation） │ │ │
│ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │
│ │ │ │ GPT-4 │ │ Claude 3 │ │ Llama 3 │ │ │ │
│ │ │ │ (OpenAI) │ │ (Anthropic) │ │ (Meta) │ │ │ │
│ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ │ 流式输出 │ │ │
│ │ │ Streaming │ │ │
│ │ └────────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌──────────────────┐ │ │
│ │ │ 用户回答 │ │ │
│ │ │ "深度学习是..." │ │ │
│ │ └──────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 可观测性层（Observability） │ │
│ │ LangSmith / OpenTelemetry / Prometheus + Grafana │ │
│ │ 追踪：检索延迟、生成延迟、Token消耗、召回率、幻觉率 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
7.1.2 RAG工作流程详解¶
Step 1: 用户输入
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
用户问题："深度学习与机器学习有什么关系？"
│
▼
Step 2: 查询向量化
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
查询 → Sentence-Transformers → [0.12, -0.34, 0.56, ...] (384维)
│
▼
Step 3: 向量检索
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
查询向量 → 向量数据库(K=5) → 返回最相似的5个文档块
│
├── 文档1: "机器学习是AI的一个子领域..." (相似度: 0.92)
├── 文档2: "深度学习是机器学习的一种方法..." (相似度: 0.89)
├── 文档3: "神经网络是深度学习的基础..." (相似度: 0.85)
├── 文档4: "AI的应用包括机器学习和深度学习..." (相似度: 0.82)
└── 文档5: "监督学习是机器学习的一种..." (相似度: 0.78)
│
▼
Step 4: 上下文组装
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
组装提示词：
"""
你是一个专业的AI助手。请根据以下上下文回答用户的问题。
相关上下文：
[1] 机器学习是人工智能的一个核心子领域，专注于开发能够从数据中学习和改进的算法...
[2] 深度学习是机器学习的一个高级分支，利用多层神经网络（深度神经网络）来学习数据的抽象表示...
[3] 神经网络是受人脑结构启发的计算模型，是深度学习的基础...
用户问题：深度学习与机器学习有什么关系？
"""
│
▼
Step 5: LLM生成
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
LLM处理提示词 → 生成最终回答
回答：
"深度学习与机器学习有密切的关系。简单来说：
1. **层级关系**：深度学习是机器学习的一个高级子领域，专门研究如何使用深层神经网络。
2. **技术演进**：机器学习最早使用传统算法（如决策树、SVM），而深度学习通过多层神经网络实现了更好的特征学习能力。
3. **依赖关系**：深度学习依赖于机器学习的基本概念，同时又扩展了机器学习的能力边界..."
7.2 LangChain核心组件详解¶
7.2.1 LangChain表达式语言（LCEL）- 2026年推荐标准¶
LCEL（LangChain Expression Language）是LangChain 1.0的核心特性，提供了声明式的管道组合语法：
"""
LangChain LCEL - 现代RAG链构建方式
参考来源: LangChain官方文档 (docs.langchain.com)
"""
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 1. 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的AI助手。请根据以下上下文回答用户的问题。
如果上下文中没有相关信息，请明确说明你不知道。
上下文：
{context}"""),
("human", "{question}")
])
# 2. 创建检索器
vectorstore = Chroma.from_documents(
documents=splits,
embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 3. 定义上下文格式化函数
def format_docs(docs):
"""将检索到的文档格式化为字符串"""
return "\n\n".join(doc.page_content for doc in docs)
# 4. 使用LCEL组合RAG链
rag_chain = (
{
# 并行执行：格式化文档 + 传递问题
"context": retriever | format_docs,
"question": RunnablePassthrough()
}
| prompt # 拼接提示词
| ChatOpenAI(model="gpt-4", temperature=0) # LLM生成
| StrOutputParser() # 输出解析
)
# 5. 流式输出（关键特性）
for chunk in rag_chain.stream("深度学习与机器学习有什么关系？"):
print(chunk, end="", flush=True)
LCEL的核心优势：
- 声明式管道语法（使用|操作符）
- 原生支持异步和流式输出
- 更好的可观测性（集成LangSmith）
- 内置并行处理能力
7.2.2 文档加载器（Document Loaders）¶
LangChain支持150+文档加载器：
"""
文档加载器完整示例
"""
from langchain_community.document_loaders import (
# 文本类
TextLoader, # 纯文本
CSVLoader, # CSV文件
JSONLoader, # JSON文件
MarkdownLoader, # Markdown
UnstructuredMarkdownLoader, # 结构化Markdown
# 文档类
PyPDFLoader, # PDF文件
Docx2txtLoader, # Word文档
UnstructuredExcelLoader, # Excel
# 网页类
WebBaseLoader, # 网页
SitemapLoader, # 网站地图
# 数据库类
SQLDatabaseLoader, # SQL数据库
NotionDBLoader, # Notion
# 云存储类
GoogleDriveLoader, # Google云盘
DropboxLoader, # Dropbox
)
# PDF文档加载示例
pdf_loader = PyPDFLoader("document.pdf")
pdf_documents = pdf_loader.load()
# 批量加载目录
directory_loader = DirectoryLoader(
"./docs/",
glob="**/*.pdf",
loader_cls=PyPDFLoader,
show_progress=True # 显示加载进度
)
all_documents = directory_loader.load()
# 带元数据的加载
def metadata_func(record: dict, metadata: dict) -> dict:
"""自定义元数据提取"""
metadata["source"] = record.get("source", "unknown")
metadata["author"] = record.get("author", "unknown")
metadata["created_date"] = record.get("created_date")
return metadata
loader = CSVLoader(
"data.csv",
metadata_columns=["source", "author", "created_date"],
metadata_func=metadata_func
)
7.2.3 文本分割器（Text Splitters）- 关键组件¶
文本分割是RAG性能的核心因素之一：
"""
文本分割器完整指南
"""
from langchain_text_splitters import (
# 推荐：递归字符分割（默认最佳选择）
RecursiveCharacterTextSplitter,
# 基于Token的分割
TokenTextSplitter,
# 结构化分割
MarkdownHeaderTextSplitter,
HTMLHeaderTextSplitter,
# 代码分割
LanguageTextSplitter,
)
# ==================== RecursiveCharacterTextSplitter ====================
# 最佳默认选择，按层次尝试分割符
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 块大小（字符数）
chunk_overlap=200, # 重叠大小（保持上下文）
length_function=len, # 长度计算函数
separators=[
"\n\n", # 段落级别（最高优先级）
"\n", # 句子级别
". ", # 英文句子
"。", "！", "？", # 中文标点
" ", # 单词级别
"" # 字符级别（最后手段）
],
keep_separator=True, # 保留分隔符
is_separator_regex=False # 是否使用正则表达式
)
# ==================== TokenTextSplitter ====================
# 基于Token数量分割（更精确控制）
token_splitter = TokenTextSplitter(
chunk_size=500, # 块大小（Token数）
chunk_overlap=50, # 重叠Token数
encoding_name="cl100k_base" # GPT-4使用的编码
)
# ==================== MarkdownHeaderTextSplitter ====================
# 保留文档结构
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "h1"), # 一级标题
("##", "h2"), # 二级标题
("###", "h3"), # 三级标题
],
strip_headers=False # 保留标题在内容中
)
# 使用示例
markdown_doc = """# 机器学习简介
## 什么是机器学习？
机器学习是人工智能的一个分支...
## 监督学习
监督学习是..."""
md_splits = markdown_splitter.split_text(markdown_doc)
# 结果：[{"content": "# 机器学习简介...", "metadata": {"h1": "机器学习简介"}},
# {"content": "## 什么是机器学习？...", "metadata": {"h2": "什么是机器学习？"}}, ...]
# ==================== LanguageTextSplitter ====================
# 代码专用分割
code_splitter = LanguageTextSplitter(
language="python",
chunk_size=500,
chunk_overlap=50
)
python_code = """
def calculate_sum(a, b):
return a + b
class Calculator:
def __init__(self):
self.result = 0
def add(self, value):
self.result += value
return self
"""
code_splits = code_splitter.split_text(python_code)
文本分割最佳实践：
"""
生产级文本分割配置
"""
class ProductionTextSplitter:
"""生产环境推荐的文本分割器"""
@staticmethod
def get_default_splitter():
"""获取默认分割器"""
return RecursiveCharacterTextSplitter(
chunk_size=1000, # 经验值：500-1500
chunk_overlap=200, # 10-20% 的chunk_size
length_function=lambda x: len(x), # 字符计数
separators=[
"\n\n", "\n", # 段落和行
"。！？；", # 中文标点
".!?;", # 英文标点
" ", # 空格
"" # 字符
]
)
@staticmethod
def get_code_splitter(language: str):
"""获取代码分割器"""
return LanguageTextSplitter(
language=language,
chunk_size=800,
chunk_overlap=100
)
@staticmethod
def get_semantic_splitter():
"""语义感知分割（高级）"""
# 可结合LLM进行更智能的分割
pass
# 分割策略选择指南
SPLITTER_SELECTION = {
"通用文本": "RecursiveCharacterTextSplitter",
"代码文件": "LanguageTextSplitter",
"技术文档": "MarkdownHeaderTextSplitter + Recursive",
"对话记录": "ConversationTextSplitter",
"法律文档": "MarkdownHeaderTextSplitter + 自定义规则"
}
7.3 高级RAG技术¶
7.3.1 HyDE（Hypothetical Document Embeddings）¶
原理：先生成假设性回答，用假设回答的向量进行检索，弥补短查询与长文档之间的语义差距。
"""
HyDE实现 - 假设文档嵌入
参考来源: arxiv.org/abs/2212.10496
"""
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
class HyDERetriever:
"""HyDE检索器"""
def __init__(self, vectorstore, llm, embeddings):
self.vectorstore = vectorstore
self.llm = llm
self.embeddings = embeddings
self.hyde_prompt = ChatPromptTemplate.from_template(
"请生成一个详细的段落来回答这个问题。不要给出答案，只需要描述一个假设的回答段落。\n\n"
"问题：{question}\n\n"
"假设回答段落："
)
def generate_hypothetical_doc(self, query: str) -> str:
"""生成假设文档"""
response = self.llm.invoke(
self.hyde_prompt.format(question=query)
)
return response.content
def retrieve(self, query: str, k: int = 5):
"""执行HyDE检索"""
# Step 1: 生成假设文档
hypothetical_doc = self.generate_hypothetical_doc(query)
# Step 2: 用假设文档的向量检索
hypothetical_embedding = self.embeddings.embed_query(hypothetical_doc)
# Step 3: 返回相似度最高的文档
return self.vectorstore.similarity_search_by_vector(
hypothetical_embedding,
k=k
)
# 使用示例
hyde_retriever = HyDERetriever(
vectorstore=vectorstore,
llm=ChatOpenAI(model="gpt-4"),
embeddings=OpenAIEmbeddings()
)
results = hyde_retriever.retrieve("深度学习与机器学习有什么关系？", k=5)
HyDE优势： - 提升检索召回率15-25% - 特别适合短查询场景 - 无需额外训练
7.3.2 Multi-Query Retrieval（多查询检索）¶
原理：生成多个查询变体，检索所有变体的结果并合并。
"""
Multi-Query检索实现
参考来源: LangChain官方文档
"""
from langchain.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
# 基础检索器
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 创建多查询检索器
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=base_retriever,
llm=ChatOpenAI(model="gpt-4", temperature=0),
prompt=ChatPromptTemplate.from_template(
"你是一个AI助手。请为这个问题生成3个不同的查询变体，"
"以帮助检索更相关的文档。每个查询应该用不同的措辞表达相同的意思。\n\n"
"原始问题：{question}\n\n"
"查询变体："
)
)
# 使用多查询检索器
results = multi_query_retriever.invoke("深度学习与机器学习有什么关系？")
# 自动生成的查询变体可能是：
# 1. "深度学习和机器学习有什么关联？"
# 2. "机器学习与深度学习的区别和联系"
# 3. "深度学习属于机器学习吗？"
7.3.3 混合搜索（Hybrid Search）¶
原理：结合向量搜索和关键词搜索（BM25），使用Reciprocal Rank Fusion合并结果。
"""
混合搜索实现 - 向量 + BM25
"""
from langchain.retrievers import BM25Retriever
from langchain.retrievers import MergerRetriever
from langchain_community.vectorstores import Chroma
# 1. 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 2. BM25关键词检索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# 3. 合并检索器（RRF - Reciprocal Rank Fusion）
hybrid_retriever = MergerRetriever(retrievers=[vector_retriever, bm25_retriever])
# 使用混合检索
results = hybrid_retriever.get_relevant_documents("深度学习教程")
# 高级：自定义RRF权重
class WeightedHybridRetriever:
"""加权混合检索器"""
def __init__(self, vector_retriever, bm25_retriever, vector_weight=0.7):
self.vector_retriever = vector_retriever
self.bm25_retriever = bm25_retriever
self.vector_weight = vector_weight
self.bm25_weight = 1 - vector_weight
def get_relevant_documents(self, query: str):
# 并行执行两种检索
vector_results = self.vector_retriever.get_relevant_documents(query)
bm25_results = self.bm25_retriever.get_relevant_documents(query)
# RRF合并
fused = self._reciprocal_rank_fusion(
[vector_results, bm25_results]
)
return fused[:10] # 返回Top 10
def _reciprocal_rank_fusion(self, result_lists, k=60):
"""RRF算法实现"""
scores = {}
for results in result_lists:
for rank, doc in enumerate(results, 1):
doc_id = doc.page_content[:50] # 简化的ID
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)
# 按分数排序
all_docs = []
for results in result_lists:
for doc in results:
doc_id = doc.page_content[:50]
if doc not in all_docs:
doc.score = scores[doc_id]
all_docs.append(doc)
all_docs.sort(key=lambda x: x.score, reverse=True)
return all_docs
7.3.4 Reranking（重排序）¶
原理：使用专门的排序模型对初步检索结果进行重新排序，提升精度。
"""
Reranking实现 - Cohere重排序
参考来源: Cohere官方文档
"""
from langchain_cohere import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.document_compressors import LLMChainExtractor
# 1. Cohere重排序器
cohere_reranker = CohereRerank(
top_n=5, # 返回Top 5
model="rerank-english-v3.0" # 或rerank-multilingual-v3.0
)
# 2. 创建压缩检索器
compression_retriever = ContextualCompressionRetriever(
base_compressor=cohere_reranker,
base_retriever=base_retriever
)
# 使用重排序检索
results = compression_retriever.invoke("深度学习与机器学习的关系")
# 3. Cross-Encoder重排序（开源方案）
from sentence_transformers import CrossEncoder
cross_encoder = CrossEncoder('ms-marco-MiniLM-L-6-v2')
def cross_encoder_rerank(query: str, documents: list, top_k: int = 5):
"""使用Cross-Encoder进行重排序"""
# 计算每文档与查询的相关性分数
doc_pairs = [(query, doc.page_content) for doc in documents]
scores = cross_encoder.predict(doc_pairs)
# 按分数排序
scored_docs = list(zip(documents, scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
return [doc for doc, score in scored_docs[:top_k]]
# 使用示例
initial_results = base_retriever.get_relevant_documents("深度学习")
reranked_results = cross_encoder_rerank("深度学习", initial_results, top_k=5)
重排序效果： - 精度提升20-35% - 延迟增加200-500ms - 建议对Top 20-50结果重排序
7.3.5 上下文压缩（Context Compression）¶
原理：从检索到的文档中提取最相关的部分，减少token使用。
"""
上下文压缩实现
参考来源: LangChain官方文档
"""
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import (
LLMChainExtractor, # LLM提取
EmbeddingsFilter, # 向量相似度过滤
DocumentCompressorPipeline # 管道组合
)
# 1. LLMChainExtractor - 使用LLM提取关键信息
llm = ChatOpenAI(model="gpt-4", temperature=0)
extractor = LLMChainExtractor.from_llm(llm)
compression_retriever_1 = ContextualCompressionRetriever(
base_compressor=extractor,
base_retriever=base_retriever
)
# 2. EmbeddingsFilter - 基于向量相似度过滤
from langchain_openai import OpenAIEmbeddings
embeddings_filter = EmbeddingsFilter(
embeddings=OpenAIEmbeddings(),
similarity_threshold=0.76 # 只保留相似度>0.76的文档
)
compression_retriever_2 = ContextualCompressionRetriever(
base_compressor=embeddings_filter,
base_retriever=base_retriever
)
# 3. 管道组合 - 多级压缩
pipeline_compressor = DocumentCompressorPipeline(
transformers=[
embeddings_filter, # 先过滤
extractor # 再提取
]
)
compression_retriever_3 = ContextualCompressionRetriever(
base_compressor=pipeline_compressor,
base_retriever=base_retriever
)
7.4 RAG开发框架对比与选择¶
| 框架 | 特点 | 适用场景 | 2026年状态 |
|---|---|---|---|
| LangChain | 生态最全，150+加载器，60+向量库 | 通用RAG，快速原型 | ✅ 活跃更新，LCEL标准化 |
| LlamaIndex | 索引优先，数据连接器丰富 | 复杂文档处理 | ✅ 活跃更新 |
| LangGraph | 低级API，Agent工作流 | 复杂多步骤流程 | ✅ LangChain 1.0核心 |
| Dify | 开源LLM应用平台 | 无代码/低代码 | ✅ 活跃开源 |
| LangFlow | 可视化工作流 | 学习/演示 | ✅ 活跃开发 |
7.5 使用LangChain构建生产级RAG¶
"""
生产级RAG系统完整实现
参考来源: LangChain官方生产部署指南
"""
import os
from typing import List, Dict, Any
from dataclasses import dataclass
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Pinecone
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, DirectoryLoader
# ==================== 配置类 ====================
@dataclass
class RAGConfig:
"""RAG系统配置"""
# LLM配置
llm_model: str = "gpt-4-turbo"
llm_temperature: float = 0.0
llm_max_tokens: int = 1000
# Embedding配置
embedding_model: str = "text-embedding-3-large"
embedding_dimensions: int = 3072
# 检索配置
retrieval_k: int = 5
retrieval_score_threshold: float = 0.7
# 分割配置
chunk_size: int = 1000
chunk_overlap: int = 200
# 向量库配置
vectorstore_type: str = "pinecone" # pinecone, chroma, milvus
# Pinecone配置
pinecone_index: str = "production-rag"
pinecone_namespace: str = "default"
# 缓存配置
enable_cache: bool = True
cache_ttl: int = 3600 # 秒
# ==================== 生产级RAG类 ====================
class ProductionRAG:
"""生产级RAG系统"""
def __init__(self, config: RAGConfig):
self.config = config
self._init_components()
def _init_components(self):
"""初始化所有组件"""
# LLM
self.llm = ChatOpenAI(
model=self.config.llm_model,
temperature=self.config.llm_temperature,
max_tokens=self.config.llm_max_tokens
)
# Embeddings
self.embeddings = OpenAIEmbeddings(
model=self.config.embedding_model,
dimensions=self.config.embedding_dimensions
)
# 文本分割器
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=self.config.chunk_size,
chunk_overlap=self.config.chunk_overlap,
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)
# 向量存储（延迟初始化）
self._vectorstore = None
@property
def vectorstore(self):
"""延迟加载向量存储"""
if self._vectorstore is None:
if self.config.vectorstore_type == "pinecone":
import pinecone
pinecone.init(
api_key=os.getenv("PINECONE_API_KEY"),
environment=os.getenv("PINECONE_ENVIRONMENT")
)
self._vectorstore = Pinecone.from_existing_index(
self.config.pinecone_index,
self.embeddings,
namespace=self.config.pinecone_namespace
)
elif self.config.vectorstore_type == "chroma":
self._vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=self.embeddings
)
return self._vectorstore
def load_documents(self, file_path: str = None, directory: str = None):
"""加载文档"""
if file_path:
loader = TextLoader(file_path, encoding="utf-8")
documents = loader.load()
elif directory:
loader = DirectoryLoader(
directory,
glob="**/*.txt",
loader_cls=TextLoader
)
documents = loader.load()
else:
raise ValueError("必须提供file_path或directory")
# 分割文档
splits = self.text_splitter.split_documents(documents)
print(f"✅ 加载并分割了 {len(splits)} 个文档块")
return splits
def index_documents(self, documents: List[Document]):
"""索引文档到向量数据库"""
if self.config.vectorstore_type == "pinecone":
Pinecone.from_documents(
documents=documents,
embedding=self.embeddings,
index_name=self.config.pinecone_index,
namespace=self.config.pinecone_namespace
)
elif self.config.vectorstore_type == "chroma":
Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory="./chroma_db"
)
print(f"✅ 索引完成")
def build_chain(self):
"""构建RAG链"""
# 提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的AI助手。请根据以下上下文回答用户的问题。
要求：
1. 只根据提供的上下文回答，不要编造信息
2. 如果上下文中没有相关信息，请明确说明
3. 回答要准确、简洁、有条理
上下文：
{context}"""),
("human", "{question}")
])
# 创建检索器
retriever = self.vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": self.config.retrieval_k,
"score_threshold": self.config.retrieval_score_threshold
}
)
# 格式化文档
def format_docs(docs: List[Document]) -> str:
return "\n\n".join(
f"[{i+1}] {doc.page_content}"
for i, doc in enumerate(docs)
)
# 构建LCEL链
self.rag_chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough()
}
| prompt
| self.llm
| StrOutputParser()
)
print("✅ RAG链构建完成")
def query(self, question: str, stream: bool = False):
"""查询"""
if not hasattr(self, 'rag_chain'):
self.build_chain()
if stream:
# 流式输出
for chunk in self.rag_chain.stream(question):
print(chunk, end="", flush=True)
else:
# 非流式输出
return self.rag_chain.invoke(question)
# ==================== 使用示例 ====================
if __name__ == "__main__":
# 配置
config = RAGConfig(
llm_model="gpt-4-turbo",
embedding_model="text-embedding-3-large",
retrieval_k=5,
vectorstore_type="chroma" # 开发环境用Chroma
)
# 初始化RAG
rag = ProductionRAG(config)
# 加载和索引文档
docs = rag.load_documents(directory="./data")
rag.index_documents(docs)
# 构建链
rag.build_chain()
# 查询示例
print("\n" + "="*60)
print("问答演示")
print("="*60)
response = rag.query("深度学习与机器学习有什么关系？")
print(f"\n回答：{response}")
7.6 RAG生产部署最佳实践¶
7.6.1 错误处理与重试¶
"""
RAG系统错误处理与重试机制
参考来源: LangChain官方文档
"""
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type
)
from langchain_openai import RateLimitError, APIError, Timeout
@retry(
stop=stop_after_attempt(3), # 最多重试3次
wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避
retry=retry_if_exception_type((RateLimitError, Timeout))
)
async def llm_with_retry(prompt: str):
"""带重试的LLM调用"""
return await llm.ainvoke(prompt)
class ResilientRAG:
"""带熔断和降级的RAG系统"""
def __init__(self):
self.primary_llm = ChatOpenAI(model="gpt-4")
self.fallback_llm = ChatOpenAI(model="gpt-3.5-turbo")
self.local_llm = Ollama(model="llama2")
async def generate(self, query: str):
try:
# 尝试主模型
return await self._call_llm(self.primary_llm, query)
except RateLimitError:
try:
# 降级到备用模型
return await self._call_llm(self.fallback_llm, query)
except Exception:
# 最后尝试本地模型
return await self.local_llm.ainvoke(query)
async def _call_llm(self, llm, query):
# 简化实现
return await llm.ainvoke(query)
7.6.2 可观测性集成（LangSmith）¶
"""
LangSmith可观测性集成
参考来源: LangSmith官方文档
"""
import os
# 设置环境变量
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_PROJECT"] = "production-rag"
from langchain_openai import ChatOpenAI
from langsmith import traceable
@traceable
class MonitoredRAG:
"""可追踪的RAG系统"""
@traceable
def retrieve(self, query: str):
"""追踪检索过程"""
return retriever.invoke(query)
@traceable
def generate(self, context: str, query: str):
"""追踪生成过程"""
return llm.invoke(prompt.format(context=context, question=query))
7.6.3 性能指标监控¶
"""
RAG性能指标监控
"""
class RAGMetrics:
"""RAG系统关键指标"""
# 检索指标
RETRIEVAL_METRICS = {
"precision@k": "检索到的相关文档比例",
"recall@k": "相关文档被检索到的比例",
"mrr": "第一个相关结果的平均倒数排名",
"ndcg": "归一化折扣累积增益"
}
# 生成指标
GENERATION_METRICS = {
"faithfulness": "回答与上下文的一致性",
"answer_relevance": "回答与问题的相关性",
"hallucination_rate": "幻觉率（生成非上下文内容的比例）",
"citation_accuracy": "引用准确性"
}
# 性能指标
PERFORMANCE_METRICS = {
"p50_latency": "P50延迟",
"p95_latency": "P95延迟",
"p99_latency": "P99延迟",
"throughput": "吞吐量（QPS）",
"cache_hit_rate": "缓存命中率"
}
# 目标指标（生产环境）
PRODUCTION_TARGETS = {
"p95_latency": "< 2秒",
"precision@10": "> 0.8",
"faithfulness": "> 0.9",
"hallucination_rate": "< 0.1",
"cache_hit_rate": "> 0.6"
}
7.7 RAG评估方法¶
7.7.1 RAG三元组评估¶
"""
RAG评估 - TruLens RAG三元组
参考来源: TruLens官方文档 (trulens.org)
"""
from trulens_eval import Tru, Feedback, OpenAI
from trulens_eval.feedback import Groundedness
# 初始化Tru
tru = Tru()
# 定义反馈函数
f_context_relevance = Feedback(
OpenAI().context_relevance_with_cot_reasons,
name="上下文相关性"
).on_input()
f_groundedness = Feedback(
Groundedness(OpenAI()).groundedness_measure_with_cot_reasons,
name="可靠性（基于上下文）"
).on_input_output()
f_answer_relevance = Feedback(
OpenAI().relevance,
name="回答相关性"
).on_input_output()
# 评估RAG应用
rag_evaluator = App(
rag_chain,
app_id="production-rag",
feedbacks=[f_context_relevance, f_groundedness, f_answer_relevance]
)
# 运行评估
with tru.Record():
response = rag_evaluator.invoke("深度学习与机器学习有什么关系？")
# 查看评估结果
tru.get_leaderboard()
7.7.2 RAGAS评估框架¶
"""
RAGAS评估实现
参考来源: RAGAS官方GitHub
"""
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall
)
from datasets import Dataset
# 准备评估数据
eval_data = {
'question': [
"深度学习与机器学习有什么关系？",
"什么是向量数据库？",
"RAG系统由哪些组件组成？"
],
'answer': [
"深度学习是机器学习的一个分支...",
"向量数据库是专门存储高维向量的数据库...",
"RAG系统包括检索器和生成器..."
],
'contexts': [
["机器学习是AI的子领域...", "深度学习使用神经网络..."],
["向量数据库存储向量...", "向量用于表示..."],
["RAG包括检索和生成...", "组件包括..."]
],
'ground_truths': [
"深度学习是机器学习的子集，使用多层神经网络",
"向量数据库是存储和检索向量数据的专用数据库",
"RAG由检索器、向量存储和生成器组成"
]
}
dataset = Dataset.from_dict(eval_data)
# 运行评估
result = evaluate(
dataset=dataset,
metrics=[
faithfulness, # 可靠性
answer_relevancy, # 回答相关性
context_precision, # 上下文精确度
context_recall # 上下文召回率
]
)
print(result)
# 输出：
# {
# 'faithfulness': 0.85,
# 'answer_relevancy': 0.92,
# 'context_precision': 0.88,
# 'context_recall': 0.80
# }
7.8 RAG性能优化技巧总结¶
| 优化方向 | 具体技术 | 预期效果 |
|---|---|---|
| 检索优化 | HyDE | +15-25%召回率 |
| Multi-Query | +20-30%召回率 | |
| 混合搜索 | +25-40%精度 | |
| Reranking | +20-35%精度 | |
| 上下文优化 | 上下文压缩 | -50-80%Token使用 |
| 智能分块 | 提升相关性 | |
| 成本优化 | 语义缓存 | -60-70%成本 |
| 模型路由 | -40-60%成本 | |
| 性能优化 | 流式输出 | 改善用户体验 |
| 异步处理 | 提升吞吐量 |
8. 最佳实践与性能优化¶
8.1 Embedding最佳实践¶
"""
Embedding最佳实践示例
"""
from sentence_transformers import SentenceTransformer
import numpy as np
def get_embedding_model(model_name="paraphrase-multilingual-mpnet-base-v2"):
"""获取最优的embedding模型"""
return SentenceTransformer(model_name)
# 批处理优化
def embed_documents_batch(documents, model, batch_size=32):
"""批量处理文档，提升效率"""
embeddings = []
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
embeddings.extend(model.encode(batch))
return np.array(embeddings)
# 文本预处理
def preprocess_text(text):
"""文本预处理，提高embedding质量"""
text = text.replace("\n", " ")
text = " ".join(text.split())
max_length = 5000
if len(text) > max_length:
text = text[:max_length]
return text
8.2 索引优化¶
# Milvus HNSW索引配置
INDEX_PARAMS = {
"metric_type": "COSINE",
"index_type": "HNSW",
"params": {
"M": 16,
"efConstruction": 200
}
}
# 搜索时配置
SEARCH_PARAMS = {
"ef": 64
}
8.3 性能调优checklist¶
- 批量插入数据（比单条插入快10-100倍）
- 合理设置索引参数
- 使用内存缓存热点数据
- 监控慢查询并优化
- 定期清理无效数据
- 使用连接池
- 合理设置超时时间
- 实施读写分离
8.4 常见陷阱与解决方案¶
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 搜索结果不准确 | Embedding模型不匹配 | 选择合适的模型 |
| 检索速度慢 | 索引配置不当 | 调整HNSW/IVF参数 |
| 内存不足 | 向量维度太高 | 降低维度或使用PQ压缩 |
| 召回率低 | 索引类型不适合 | 切换到HNSW |
| 重复结果 | 缺乏去重机制 | 添加去重逻辑 |
9. 常见问题与解决方案¶
Q1: 如何选择合适的Embedding模型？¶
- 英文为主：all-MiniLM-L6-v2（轻量，速度快）
- 多语言：paraphrase-multilingual-mpnet-base-v2
- 高精度：E5-large-v2
Q2: 向量维度越高越好吗？¶
不是！需要权衡：
维度 vs 效果 vs 性能
│
384├─────────────────────────────────────► 推荐起点
│
768├─────────────────────────────────────► 高精度场景
│
1536├─────────────────────────────────────► 大规模预训练模型
│
2000+├───────────────────────────────────► 需要PCA降维
建议： 从 384 维开始，根据效果调整。
Q3: 如何处理大规模数据？¶
分片策略：
# 按数据范围分片
shard1 = milvus.create_collection("shard_1", partition_tag="2024")
shard2 = milvus.create_collection("shard_2", partition_tag="2025")
Q4: 如何保证向量搜索的准确性？¶
评估指标： - 召回率（Recall@K） - 准确率（Precision@K） - mAP（Mean Average Precision）
10. 进阶学习资源¶
10.1 官方文档¶
| 资源 | 链接 |
|---|---|
| Milvus | https://milvus.io/docs |
| Pinecone | https://docs.pinecone.io |
| Qdrant | https://qdrant.tech/documentation |
| Weaviate | https://weaviate.io/developers |
| Chroma | https://docs.trychroma.com |
| pgvector | https://github.com/pgvector/pgvector |
10.2 学习路径¶
Day 1: 基础概念 + 环境搭建
└─► Day 2-3: 掌握一种向量数据库（推荐Chroma入门，Milvus深入）
└─► Day 4-5: 学习RAG应用（LangChain/LlamaIndex）
└─► Day 6-7: 性能优化 + 生产部署
10.3 推荐阅读¶
- 入门文章
- 《向量数据库：从0到1》 - Daily Dose of DS
-
《A Beginner's Guide to Vector Databases》 - Pinecone Blog
-
技术深度
- 《ANN算法详解：HNSW vs IVF》 - Milvus Blog
- 《向量索引的那些事儿》 - Medium
10.4 社区与交流¶
- Discord: Milvus Community, Qdrant Community
- GitHub: 关注各项目的示例代码
- 知乎: 搜索"向量数据库"话题
- Reddit: r/LocalLLaMA, r/MachineLearning
附录：完整代码仓库结构¶
vector-db-tutorial/
│
├── 1_first_vector_search.py # 基础语义搜索示例
├── 2_advanced_search_qdrant.py # 高级搜索（带过滤）
├── 3_simple_rag_app.py # 简单RAG应用
├── 4_rag_with_langchain.py # LangChain RAG
│
├── data/
│ ├── documents.txt # 示例文档
│ └── knowledge_base.txt # 知识库
│
├── vector_db/ # Chroma持久化存储
│
├── requirements.txt # 依赖列表
│
└── README.md # 项目说明
安装依赖¶
chromadb>=0.4.0
sentence-transformers>=2.2.0
qdrant-client>=1.7.0
openai>=1.0.0
langchain>=0.1.0
langchain-community>=0.0.0
numpy>=1.24.0
祝学习愉快！🚀