向量数据库

向量数据库是 RAG 系统的核心组件，用于存储和检索文档的向量表示。本章将介绍常用的向量数据库及其使用方法。

什么是向量数据库？

向量数据库 专门用于存储高维向量（嵌入），并支持高效的相似度搜索。在 RAG 系统中，它存储文档块的嵌入向量，根据用户查询检索最相关的文档。

为什么需要向量数据库？

• 高效相似度搜索：传统数据库无法高效计算向量相似度
• 语义检索：基于语义而非关键词匹配
• 大规模数据：支持数十亿级向量的快速检索

常用向量数据库

1. Chroma

轻量级、易用的本地向量数据库：

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建嵌入
embeddings = OpenAIEmbeddings()

# 从文档创建向量存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    collection_name="my_collection"
)

# 持久化存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 检索
results = vectorstore.similarity_search("query", k=3)

2. Pinecone

云端向量数据库，支持大规模部署：

from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
import pinecone

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

# 创建向量存储
vectorstore = PineconeVectorStore.from_documents(
    documents=splits,
    embedding=embeddings,
    index_name="my-index"
)

# 检索
results = vectorstore.similarity_search("query", k=3)

3. Weaviate

开源向量搜索引擎：

from langchain_weaviate import WeaviateVectorStore
import weaviate

# 连接 Weaviate
client = weaviate.Client("http://localhost:8080")

# 创建向量存储
vectorstore = WeaviateVectorStore(
    client=client,
    index_name="MyIndex",
    text_key="text",
    embedding=embeddings
)

# 添加文档
vectorstore.add_documents(documents)

4. Milvus

国产开源向量数据库，支持多种索引：

from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
vectorstore = Milvus.from_documents(
    documents=splits,
    embedding=embeddings,
    connection_args={"host": "localhost", "port": "19530"}
)

# 检索
results = vectorstore.similarity_search("query", k=3)

5. Qdrant

Rust 编写的向量数据库，性能高：

from langchain_qdrant import Qdrant
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
vectorstore = Qdrant.from_documents(
    documents=splits,
    embedding=embeddings,
    location=":memory:",  # 内存模式，或使用 url 连接到远程
    collection_name="my_collection"
)

6. FAISS

Facebook 的向量检索库，适合本地测试：

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

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

# 保存到磁盘
vectorstore.save_local("faiss_index")

# 加载
loaded_vectorstore = FAISS.load_local("faiss_index", embeddings)

向量数据库操作

创建索引

# 方式1：从文档创建
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_name="my_docs"
)

# 方式2：手动添加
vectorstore = Chroma.from_texts(
    texts=["text1", "text2", "text3"],
    embedding=embeddings
)

相似度搜索

# 基本搜索
results = vectorstore.similarity_search("query", k=3)

# 带相似度分数
results = vectorstore.similarity_search_with_score("query", k=3)

# 查看分数
for doc, score in results:
    print(f"内容: {doc.page_content[:50]}...")
    print(f"相似度分数: {score}\n")

删除和更新

# 删除集合
vectorstore.delete_collection()

# 删除指定文档
vectorstore.delete(ids=["doc_id1", "doc_id2"])

# 更新文档
vectorstore.update_document("doc_id", new_document)

检索器配置

as_retriever

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 类型
    search_kwargs={
        "k": 3,                # 返回数量
        "score_threshold": 0.8  # 相似度阈值
    }
)

# 使用检索器
docs = retriever.invoke("query")

搜索类型

类型	说明	适用场景
similarity	相似度搜索	默认场景
similarity_threshold	阈值过滤	精确匹配
mmr	最大边际相关	多样性结果
similarity_score_threshold	分数阈值	质量控制

嵌入模型

OpenAI Embeddings

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"  # 或 text-embedding-3-large
)

# 文本嵌入
vector = embeddings.embed_query("Hello world")
print(f"维度: {len(vector)}")

本地 Embeddings

from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

其他 Embeddings

# Google
from langchain_google_genai import GoogleGenerativeAIEmbeddings

# Anthropic
from langchain_anthropic import AnthropicEmbeddings

# Cohere
from langchain_cohere import CohereEmbeddings

实际应用示例

1. 文档问答系统

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.prompts import ChatPromptTemplate

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

# 2. 分割
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = splitter.split_documents(documents)

# 3. 嵌入存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(splits, embeddings)

# 4. 创建问答链
prompt = ChatPromptTemplate.from_template("""
基于以下文档回答问题：

{context}

问题：{input}
""")

qa_chain = create_stuff_documents_chain(model, prompt)
retrieval_chain = create_retrieval_chain(vectorstore.as_retriever(), qa_chain)

# 5. 问答
result = retrieval_chain.invoke({"input": "你的问题"})
print(result["answer"])

2. 本地知识库

# 使用 FAISS 本地存储
from langchain_community.vectorstores import FAISS

# 创建
vectorstore = FAISS.from_documents(documents, embeddings)

# 保存
vectorstore.save_local("knowledge_base")

# 加载
vectorstore = FAISS.load_local("knowledge_base", embeddings)

# 检索
results = vectorstore.similarity_search("查询内容")

3. 多租户系统

# 为每个租户创建独立的向量存储
def get_tenant_vectorstore(tenant_id: str):
    return Chroma.from_documents(
        documents=get_tenant_documents(tenant_id),
        embedding=embeddings,
        collection_name=f"tenant_{tenant_id}"
    )

# 使用
for tenant in ["tenant_a", "tenant_b"]:
    vs = get_tenant_vectorstore(tenant)
    results = vs.similarity_search(user_query)

性能优化

索引类型

# Pinecone 索引配置
index = pinecone.Index("my-index", 
    dimension=1536,
    metric="cosine",
    pod_type="p1"
)

批处理

# 批量添加文档
from langchain_community.vectorstores import Chroma

# 分批添加
batch_size = 100
for i in range(0, len(documents), batch_size):
    batch = documents[i:i+batch_size]
    vectorstore.add_documents(batch)

缓存嵌入

from langchain.embeddings import CacheBackedEmbeddings

# 创建缓存嵌入
cache_embedder = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=embeddings,
    bytes_store=InMemoryByteStore(),
    namespace="embeddings"
)

常见问题

1. 检索结果不准确

• 检查文档质量
• 调整 chunk_size
• 尝试不同的嵌入模型

2. 性能问题

• 使用索引加速
• 增加服务器资源
• 使用更小的嵌入模型

3. 数据持久化

• 定期备份向量数据库
• 检查磁盘空间

下一步

现在你已经掌握了向量数据库的使用，接下来学习：

• Tools 工具 - 为代理添加工具能力
• 回调与监控 - LangSmith 调试和监控