LangChain框架精通：构建生产级AI应用

本文集文档发布于灏天文库

2.3 向量存储

文档摘要

2.3 向量存储 — LangChain框架精通核心组件 本节导读：学习向量存储的核心概念和实现方案，掌握FAISS、Chroma、Pinecone等主流向量数据库的使用方法，构建高效的语义检索系统。 学习目标 理解向量存储的基本原理和架构设计 掌握FAISS本地向量库的配置和使用 熟练使用Chroma云原生向量数据库 了解Pinecone云向量服务的高级功能 实现增量更新和批量操作 优化向量检索性能和准确性 核心概念 向量存储是LangChain中最关键的组件之一，它负责将文本转换为高维向量并进行高效检索。理解向量存储的工作原理对构建高效的RAG系统至关重要。 向量存储的架构原理 !

2.3 向量存储 — LangChain框架精通核心组件

本节导读：学习向量存储的核心概念和实现方案，掌握FAISS、Chroma、Pinecone等主流向量数据库的使用方法，构建高效的语义检索系统。

学习目标

  • 理解向量存储的基本原理和架构设计
  • 掌握FAISS本地向量库的配置和使用
  • 熟练使用Chroma云原生向量数据库
  • 了解Pinecone云向量服务的高级功能
  • 实现增量更新和批量操作
  • 优化向量检索性能和准确性

核心概念

向量存储是LangChain中最关键的组件之一，它负责将文本转换为高维向量并进行高效检索。理解向量存储的工作原理对构建高效的RAG系统至关重要。

向量存储的架构原理

![向量存储架构示意图：文档分割、嵌入模型、向量索引、相似度检索的完整流程]

向量存储的核心工作流程包括：

  1. 文档预处理：将文档分割为小块
  2. 嵌入生成：使用嵌入模型将文本转换为向量
  3. 向量索引：建立高效的向量索引结构
  4. 相似度检索：计算查询向量与存储向量的相似度

向量存储的核心组件

  • 嵌入模型(Embedding Model)：将文本转换为向量的神经网络模型
  • 向量索引(Vector Index)：用于快速检索向量数据结构
  • 相似度计算(Similarity Search)：计算向量间相似度的方法
  • 元数据管理(Metadata)：存储向量的元数据信息

环境准备 / 前置知识

基础依赖安装

# 向量存储基础包
pip install langchain langchain-community

# FAISS本地向量库
pip install faiss-cpu  # CPU版本
pip install faiss-gpu  # GPU版本(需要CUDA支持)

# Chroma向量数据库
pip install chromadb

# Pinecone云向量服务
pip install pinecone-client

# OpenAI嵌入模型
pip install openai

# HuggingFace嵌入模型
pip install sentence-transformers
​

核心导入模块

import os
from typing import List, Dict, Any, Optional
import numpy as np

# LangChain向量存储
from langchain.vectorstores import FAISS, Chroma, Pinecone
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader

# 其他工具
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
​

分步实战

步骤 1：基础向量存储配置

首先配置基础的向量存储环境，包括嵌入模型和文本分割器：

import os
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader

# 设置API密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 创建嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002",
    chunk_size=1000
)

# 创建文本分割器
text_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
)

# 加载文档
loader = TextLoader("sample.txt")
documents = loader.load()
texts = text_splitter.split_documents(documents)

print(f"原始文档数: {len(documents)}")
print(f"分割后文本数: {len(texts)}")
print(f"文本示例: {texts[0][:200]}...")
​

步骤 2：使用FAISS本地向量库

FAISS(Facebook AI Similarity Search)是一个高效的相似性搜索库，适合本地部署：

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# 创建嵌入模型
embeddings = OpenAIEmbeddings()

# 创建FAISS向量存储
faiss_store = FAISS.from_documents(texts, embeddings)

# 保存本地索引
faiss_store.save_local("faiss_index")

# 从本地加载索引
faiss_store = FAISS.load_local("faiss_index", embeddings)

# 相似性搜索
query = "什么是机器学习"
results = faiss_store.similarity_search(query, k=3)

for i, doc in enumerate(results):
    print(f"结果 {i+1}: {doc.page_content[:200]}...")
    print(f"相关性分数: {doc.metadata.get('score', 'N/A')}")
    print()

# 向量搜索
query_vector = embeddings.embed_query(query)
results = faiss_store.similarity_search_by_vector(query_vector, k=3)

# 添加新文档
new_texts = ["机器学习是人工智能的一个分支", "深度学习是机器学习的子集"]
faiss_store.add_texts(new_texts)
​

步骤 3：使用Chroma向量数据库

Chroma是一个开源的向量数据库，支持云原生部署：

import chromadb
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 初始化Chroma客户端
client = chromadb.Client()

# 创建Chroma向量存储
chroma_store = Chroma.from_documents(
    documents=texts,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

# 持久化存储
chroma_store.persist()

# 加载现有存储
chroma_store = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings()
)

# 相似性搜索
query = "什么是深度学习"
results = chroma_store.similarity_search(query, k=3)

# 元数据搜索
results = chroma_store.similarity_search(
    query=query,
    k=3,
    filter={"source": "sample.txt"}
)

# 使用集合管理
collection = client.create_collection("langchain_docs")
chroma_store = Chroma(
    collection_name="langchain_docs",
    embedding_function=OpenAIEmbeddings(),
    client=client
)

# 批量操作
batch_texts = ["文本1", "文本2", "文本3"]
batch_metadatas = [{"source": "batch1"}] * 3
batch_ids = ["doc1", "doc2", "doc3"]

chroma_store.add_texts(
    texts=batch_texts,
    metadatas=batch_metadatas,
    ids=batch_ids
)
​

步骤 4：使用Pinecone云向量服务

Pinecone是一个全托管的向量数据库服务，适合大规模部署：

import pinecone
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

# 初始化Pinecone
pinecone.init(
    api_key="your-pinecone-api-key",
    environment="your-environment"
)

# 创建Pinecone向量存储
pinecone_store = Pinecone.from_documents(
    documents=texts,
    embedding=OpenAIEmbeddings(),
    index_name="langchain-index"
)

# 连接现有索引
index = pinecone.Index("langchain-index")
pinecone_store = Pinecone(
    index=index,
    embedding=OpenAIEmbeddings()
)

# 相似性搜索
query = "什么是神经网络"
results = pinecone_store.similarity_search(query, k=3)

# 带过滤的搜索
results = pinecone_store.similarity_search(
    query=query,
    k=3,
    filter={"category": "ai"}
)

# 向量搜索
query_vector = OpenAIEmbeddings().embed_query(query)
results = pinecone_store.similarity_search_by_vector(query_vector, k=3)

# 删除向量
pinecone_store.delete(["doc1", "doc2"])

# 更新向量
pinecone_store.add_texts(
    texts=["更新后的文本"],
    ids=["doc1"],
    metadatas={"updated": True}
)
​

步骤 5：高级向量存储操作

掌握向量存储的高级操作，包括自定义嵌入和查询优化：

from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS

# 使用HuggingFace嵌入模型
huggingface_embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2",
    model_kwargs={'device': 'cpu'}
)

# 创建自定义向量存储
custom_store = FAISS.from_documents(
    documents=texts,
    embedding=huggingface_embeddings
)

# 自定义相似度搜索
def custom_similarity_search(query: str, k: int = 3):
    query_vector = huggingface_embeddings.embed_query(query)
    results = custom_store.similarity_search_by_vector(query_vector, k=k)
    
    # 添加自定义处理逻辑
    processed_results = []
    for doc in results:
        processed_results.append({
            'content': doc.page_content,
            'metadata': doc.metadata,
            'score': len(doc.page_content) / 1000  # 自定义评分
        })
    
    return processed_results

# 使用自定义搜索
results = custom_similarity_search("什么是自然语言处理")

# 批量处理
batch_texts = ["批量文本1", "批量文本2", "批量文本3"]
batch_embeddings = huggingface_embeddings.embed_documents(batch_texts)

# 添加批量向量
custom_store.add_embeddings(
    text_embeddings=list(zip(batch_texts, batch_embeddings)),
    metadatas=[{"source": "batch"}] * len(batch_texts)
)
​

完整示例

完整的RAG系统实现

下面是一个完整的RAG系统实现，使用FAISS作为向量存储：

import os
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# 配置环境
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

class VectorRAGSystem:
    def __init__(self, documents_path: str):
        self.documents_path = documents_path
        self.vector_store = None
        self.qa_chain = None
        self.embeddings = OpenAIEmbeddings()
        
    def load_and_process_documents(self):
        """加载和处理文档"""
        # 加载PDF文档
        loader = PyPDFLoader(self.documents_path)
        documents = loader.load()
        
        # 文档分割
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len
        )
        texts = text_splitter.split_documents(documents)
        
        print(f"处理完成 - 原始文档: {len(documents)}, 文本块: {len(texts)}")
        return texts
    
    def create_vector_store(self, texts: list):
        """创建向量存储"""
        self.vector_store = FAISS.from_documents(texts, self.embeddings)
        print("向量存储创建完成")
        
    def create_qa_chain(self):
        """创建问答链"""
        prompt_template = """使用以下上下文来回答问题。如果不知道答案，就说不知道不要随意编造。

上下文：
{context}

问题：{question}

回答："""
        
        prompt = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )
        
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=OpenAI(),
            chain_type="stuff",
            retriever=self.vector_store.as_retriever(),
            chain_type_kwargs={"prompt": prompt},
            return_source_documents=True
        )
        
    def query(self, question: str):
        """查询问题"""
        if self.qa_chain is None:
            raise ValueError("QA链未初始化")
        
        result = self.qa_chain({"query": question})
        return result
    
    def save_vector_store(self, path: str):
        """保存向量存储"""
        if self.vector_store is None:
            raise ValueError("向量存储未创建")
        self.vector_store.save_local(path)
        
    def load_vector_store(self, path: str):
        """加载向量存储"""
        self.vector_store = FAISS.load_local(path, self.embeddings)

# 使用示例
if __name__ == "__main__":
    # 初始化系统
    rag_system = VectorRAGSystem("documents/sample.pdf")
    
    # 加载和处理文档
    texts = rag_system.load_and_process_documents()
    
    # 创建向量存储
    rag_system.create_vector_store(texts)
    
    # 创建问答链
    rag_system.create_qa_chain()
    
    # 查询示例
    question = "什么是LangChain的主要特性？"
    result = rag_system.query(question)
    
    print(f"问题: {question}")
    print(f"回答: {result['result']}")
    print(f"来源文档数: {len(result['source_documents'])}")
​

多向量存储对比实现

import time
import matplotlib.pyplot as plt
from langchain.vectorstores import FAISS, Chroma, Pinecone
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter

class VectorStoreComparison:
    def __init__(self, texts: list):
        self.texts = texts
        self.embeddings = OpenAIEmbeddings()
        self.results = {}
        
    def benchmark_faiss(self):
        """测试FAISS性能"""
        start_time = time.time()
        faiss_store = FAISS.from_documents(self.texts, self.embeddings)
        build_time = time.time() - start_time
        
        start_time = time.time()
        results = faiss_store.similarity_search("什么是机器学习", k=3)
        search_time = time.time() - start_time
        
        self.results['FAISS'] = {
            'build_time': build_time,
            'search_time': search_time,
            'result_count': len(results)
        }
        
        return self.results['FAISS']
    
    def benchmark_chroma(self):
        """测试Chroma性能"""
        import chromadb
        
        start_time = time.time()
        chroma_store = Chroma.from_documents(self.texts, self.embeddings)
        build_time = time.time() - start_time
        
        start_time = time.time()
        results = chroma_store.similarity_search("什么是机器学习", k=3)
        search_time = time.time() - start_time
        
        self.results['Chroma'] = {
            'build_time': build_time,
            'search_time': search_time,
            'result_count': len(results)
        }
        
        return self.results['Chroma']
    
    def benchmark_pinecone(self):
        """测试Pinecone性能"""
        import pinecone
        
        start_time = time.time()
        pinecone_store = Pinecone.from_documents(self.texts, self.embeddings, index_name="test-index")
        build_time = time.time() - start_time
        
        start_time = time.time()
        results = pinecone_store.similarity_search("什么是机器学习", k=3)
        search_time = time.time() - start_time
        
        self.results['Pinecone'] = {
            'build_time': build_time,
            'search_time': search_time,
            'result_count': len(results)
        }
        
        return self.results['Pinecone']
    
    def plot_comparison(self):
        """绘制性能对比图"""
        if not self.results:
            print("请先运行性能测试")
            return
        
        stores = list(self.results.keys())
        build_times = [self.results[store]['build_time'] for store in stores]
        search_times = [self.results[store]['search_time'] for store in stores]
        
        plt.figure(figsize=(12, 6))
        
        plt.subplot(1, 2, 1)
        plt.bar(stores, build_times, color=['blue', 'green', 'red'])
        plt.title('构建时间对比')
        plt.ylabel('时间(秒)')
        
        plt.subplot(1, 2, 2)
        plt.bar(stores, search_times, color=['blue', 'green', 'red'])
        plt.title('搜索时间对比')
        plt.ylabel('时间(秒)')
        
        plt.tight_layout()
        plt.savefig('vectorstore_comparison.png')
        print("性能对比图已保存为 vectorstore_comparison.png")

# 使用示例
if __name__ == "__main__":
    # 创建测试文本
    test_texts = ["机器学习是人工智能的一个分支"] * 10
    
    # 运行对比测试
    comparison = VectorStoreComparison(test_texts)
    
    faiss_result = comparison.benchmark_faiss()
    chroma_result = comparison.benchmark_chroma()
    pinecone_result = comparison.benchmark_pinecone()
    
    print("FAISS结果:", faiss_result)
    print("Chroma结果:", chroma_result)
    print("Pinecone结果:", pinecone_result)
    
    # 绘制对比图
    comparison.plot_comparison()
​

常见问题 FAQ

Q1：FAISS和Chroma如何选择？

A：FAISS适合本地部署、无需外部依赖、性能优秀但需要手动维护索引；Chroma适合需要持久化、云原生部署、自动管理索引的场景。小规模项目推荐FAISS，生产环境推荐Chroma或Pinecone。

Q2：如何优化向量存储的检索准确性？

A：可以通过调整文本分割策略、使用高质量的嵌入模型、增加上下文重叠、添加元数据过滤等方式来优化检索准确性。建议使用text-embedding-3-large等高质量嵌入模型。

Q3：向量存储如何处理大量数据？

A：对于大量数据，建议使用分片技术、增量更新、定期重建索引等方式。也可以考虑使用Pinecone等云服务来处理大规模向量数据。

Q4：如何降低向量存储的内存使用？

A：可以通过量化技术（如FAISS的IVF量化）、使用更小的嵌入维度、删除不需要的向量、使用本地数据库等方式降低内存使用。

Q5：向量存储的安全性如何保证？

A：可以通过API密钥管理、访问控制、数据加密、定期备份等方式来保证安全性。建议使用HTTPS连接，并限制访问权限。

最佳实践与避坑

最佳实践

  1. 合理的文本分割：根据文档类型选择合适的分割策略
  2. 嵌入模型选择：根据需求选择合适的嵌入模型和维度
  3. 元数据管理：为每个向量添加有意义的元数据
  4. 性能监控：定期监控向量存储的性能指标
  5. 备份策略：定期备份向量存储数据

常见陷阱

  1. 文本分割过小：导致语义不完整，影响检索质量
  2. 嵌入维度过高：增加内存和计算负担
  3. 忽略元数据：导致后续过滤和查询困难
  4. 不进行版本管理：难以回滚和调试
  5. 过度依赖单一存储方式：缺乏灵活性和可扩展性

本节小结

本节详细介绍了LangChain中向量存储的核心概念和实现方法，包括：

  • 向量存储原理：理解从文本到向量的转换和检索机制
  • FAISS使用：本地向量存储的配置和优化
  • Chroma应用：云原生向量数据库的使用
  • Pinecone集成：云端向量服务的应用
  • 性能优化：如何提升检索准确性和性能
  • 完整实践：从基础应用到高级实现的完整示例

掌握向量存储是构建高效RAG系统的关键基础。在下一节中，我们将继续学习记忆管理组件，进一步构建完整的LangChain应用体系。

延伸阅读

  • 官方文档：LangChain向量存储模块文档
  • 相关章节：本教程 2.4 节记忆管理
  • 推荐资源：《向量数据库原理与实践》
  • 开源项目：Chroma官方示例库
  • 技术博客：FAISS索引优化指南

关键词：向量存储,LangChain,FAISS,Chroma,Pinecone,RAG,嵌入模型,相似性搜索
难度：进阶
预计阅读：45分钟

发布者: 作者: 转发
评论区 (0)
U
0/500