麻瓜教程

Appearance

向量存储 ​

向量存储（Vector Stores）是 LangChain 中的重要组件，它用于存储和检索文本的向量表示，是实现语义搜索和知识检索的关键组件。本节将详细介绍 LangChain 1.2 中的向量存储使用方法。

安装基础依赖 ​

在开始使用向量存储之前，需要安装以下基础依赖：

bash
pip install langchain langchain-openai python-dotenv langchain-community

什么是向量存储？ ​

向量存储是一种特殊的数据库，用于存储和检索文本的向量表示。向量存储使用相似度搜索算法，能够根据语义相似性快速检索相关文本，是实现语义搜索和知识检索的关键组件。

向量存储的优势 ​

与传统的关键词搜索相比，向量存储具有以下优势：

  1. 语义理解：能够理解文本的语义含义，而不仅仅是关键词匹配
  2. 相关性排序：根据语义相似性对结果进行排序
  3. 高效检索：能够快速检索大量文本
  4. 多模态支持：可以存储和检索不同类型的数据，如文本、图像等

常用的向量存储 ​

LangChain 支持多种向量存储，以下是一些常用的向量存储：

向量存储类型特点依赖包
FAISS本地高效的相似度搜索库faiss-cpu 或 faiss-gpu
Chroma本地轻量级向量存储chromadb
Pinecone云端托管的向量数据库pinecone-client
Weaviate本地/云端开源的向量搜索引擎weaviate-client
Milvus本地/云端高性能向量数据库pymilvus
Redis本地/云端内存数据库，支持向量搜索redis

使用 FAISS ​

FAISS（Facebook AI Similarity Search）是 Facebook 开发的一个高效的相似度搜索库，适用于本地部署。

安装依赖 ​

bash
pip install faiss-cpu  # CPU 版本
# 或
pip install faiss-gpu  # GPU 版本（需要 CUDA 支持）

初始化 FAISS ​

python
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(api_key="your-api-key")

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

使用 FAISS 进行搜索 ​

python
# 相似度搜索
query = "What is LangChain?"
docs = vectorstore.similarity_search(query, k=3)
print(f"Found {len(docs)} documents")
for i, doc in enumerate(docs):
    print(f"Document {i+1}:")
    print(doc.page_content)
    print()

# 带分数的相似度搜索
docs_with_scores = vectorstore.similarity_search_with_score(query, k=3)
print(f"Found {len(docs_with_scores)} documents")
for i, (doc, score) in enumerate(docs_with_scores):
    print(f"Document {i+1} (score: {score}):")
    print(doc.page_content)
    print()

保存和加载 FAISS ​

python
# 保存向量存储
vectorstore.save_local("faiss_index")

# 加载向量存储
loaded_vectorstore = FAISS.load_local("faiss_index", embeddings)

# 使用加载的向量存储进行搜索
docs = loaded_vectorstore.similarity_search(query, k=3)
print(f"Found {len(docs)} documents")

使用 Chroma ​

Chroma 是一个轻量级的向量存储，适用于本地部署和开发环境。

安装依赖 ​

bash
pip install chromadb

初始化 Chroma ​

python
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(api_key="your-api-key")

# 创建向量存储
vectorstore = Chroma.from_documents(texts, embeddings)

使用 Chroma 进行搜索 ​

python
# 相似度搜索
query = "What is LangChain?"
docs = vectorstore.similarity_search(query, k=3)
print(f"Found {len(docs)} documents")
for i, doc in enumerate(docs):
    print(f"Document {i+1}:")
    print(doc.page_content)
    print()

# 带分数的相似度搜索
docs_with_scores = vectorstore.similarity_search_with_score(query, k=3)
print(f"Found {len(docs_with_scores)} documents")
for i, (doc, score) in enumerate(docs_with_scores):
    print(f"Document {i+1} (score: {score}):")
    print(doc.page_content)
    print()

保存和加载 Chroma ​

python
# Chroma 会自动保存数据到本地目录
# 可以指定保存路径
vectorstore = Chroma.from_documents(
    texts, 
    embeddings, 
    persist_directory="chroma_db"
)

# 持久化数据
vectorstore.persist()

# 加载向量存储
loaded_vectorstore = Chroma(
    persist_directory="chroma_db",
    embedding_function=embeddings
)

# 使用加载的向量存储进行搜索
docs = loaded_vectorstore.similarity_search(query, k=3)
print(f"Found {len(docs)} documents")

使用 Pinecone ​

Pinecone 是一个托管的向量数据库，适用于生产环境。

安装依赖 ​

bash
pip install pinecone-client

初始化 Pinecone ​

python
import pinecone
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Pinecone
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

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

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(api_key="your-api-key")

# 创建索引
index_name = "langchain-demo"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,  # OpenAI 嵌入模型的维度
        metric="cosine"
    )

# 创建向量存储
vectorstore = Pinecone.from_documents(
    texts, 
    embeddings, 
    index_name=index_name
)

使用 Pinecone 进行搜索 ​

python
# 相似度搜索
query = "What is LangChain?"
docs = vectorstore.similarity_search(query, k=3)
print(f"Found {len(docs)} documents")
for i, doc in enumerate(docs):
    print(f"Document {i+1}:")
    print(doc.page_content)
    print()

# 带分数的相似度搜索
docs_with_scores = vectorstore.similarity_search_with_score(query, k=3)
print(f"Found {len(docs_with_scores)} documents")
for i, (doc, score) in enumerate(docs_with_scores):
    print(f"Document {i+1} (score: {score}):")
    print(doc.page_content)
    print()

向量存储的高级配置 ​

自定义嵌入模型 ​

python
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 初始化自定义嵌入模型
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

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

# 相似度搜索
query = "What is LangChain?"
docs = vectorstore.similarity_search(query, k=3)
print(f"Found {len(docs)} documents")

使用元数据过滤 ​

python
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档并添加元数据
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 添加元数据
for i, doc in enumerate(texts):
    doc.metadata["section"] = f"Section {i+1}"
    doc.metadata["source"] = "document.txt"

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(api_key="your-api-key")

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

# 使用元数据过滤进行搜索
query = "What is LangChain?"
filter_dict = {"section": "Section 1"}
docs = vectorstore.similarity_search(query, k=3, filter=filter_dict)
print(f"Found {len(docs)} documents")
for i, doc in enumerate(docs):
    print(f"Document {i+1} (section: {doc.metadata['section']}):")
    print(doc.page_content)
    print()

向量存储的最佳实践 ​

1. 选择合适的向量存储 ​

根据应用的规模和需求，选择合适的向量存储：

  • 开发环境：使用 FAISS 或 Chroma
  • 生产环境：使用 Pinecone、Weaviate 或 Milvus

2. 合理设置文档分割参数 ​

根据文档的长度和复杂度，合理设置文档分割参数：

  • chunk_size：每个文档块的大小
  • chunk_overlap：文档块之间的重叠部分

3. 选择合适的嵌入模型 ​

根据应用的需求和预算，选择合适的嵌入模型：

  • 高质量：使用 OpenAI 的嵌入模型
  • 成本效益：使用 Hugging Face 的开源模型

4. 优化向量存储的参数 ​

根据应用的需求，优化向量存储的参数：

  • k：返回的文档数量
  • filter：元数据过滤条件
  • score_threshold：相似度阈值

5. 定期更新向量存储 ​

对于动态变化的文档集合，定期更新向量存储，确保检索结果的准确性。

示例：构建基于向量存储的问答系统 ​

python
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains import RetrievalQA

# 加载文档
loader = TextLoader("path/to/document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(api_key="your-api-key")

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

# 初始化聊天模型
chat_model = ChatOpenAI(api_key="your-api-key")

# 创建检索链
qa_chain = RetrievalQA.from_chain_type(
    llm=chat_model,
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)

# 运行检索链
response = qa_chain.invoke({"query": "What is LangChain?"})
print(response["result"])

# 交互循环
print("Q&A System: Hello! Ask me anything about the document.")
while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "quit", "bye"]:
        print("Q&A System: Goodbye!")
        break
    response = qa_chain.invoke({"query": user_input})
    print(f"Q&A System: {response['result']}")

总结 ​

向量存储是 LangChain 中的重要组件，它用于存储和检索文本的向量表示，是实现语义搜索和知识检索的关键组件。通过本文的介绍，您应该已经了解了 LangChain 1.2 中常用的向量存储及其使用方法。

在实际应用中，您可以根据具体需求选择合适的向量存储，配置适当的参数，构建功能强大的 LLM 应用。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1