向量数据库教程¶

import faiss
import numpy as np

# 创建向量维度
dimension = 128

# 创建索引
index = faiss.IndexFlatL2(dimension)

# 生成示例数据
vectors = np.random.random((1000, dimension)).astype('float32')

# 添加向量
index.add(vectors)

print(f"索引中的向量数量: {index.ntotal}")

# 查询
query = np.random.random((1, dimension)).astype('float32')
distances, indices = index.search(query, k=5)

print(f"最近的5个向量: {indices}")
print(f"对应距离: {distances}")

# 1. Flat索引（精确搜索）
index_flat = faiss.IndexFlatL2(dimension)

# 2. IVF索引（倒排文件索引）
quantizer = faiss.IndexFlatL2(dimension)
index_ivf = faiss.IndexIVFFlat(quantizer, dimension, 100)

# 3. HNSW索引（图索引）
index_hnsw = faiss.IndexHNSWFlat(dimension, 32)

# 4. PQ索引（乘积量化）
index_pq = faiss.IndexPQ(dimension, 8, 8)

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 创建向量库
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    texts=["文档1内容", "文档2内容", "文档3内容"],
    embedding=embeddings
)

# 检索
docs = vectorstore.similarity_search("查询内容", k=3)

# 保存/加载
vectorstore.save_local("faiss_index")
vectorstore = FAISS.load_local("faiss_index", embeddings)

import chromadb

# 创建客户端
client = chromadb.Client()

# 创建集合
collection = client.create_collection("my_collection")

# 添加文档
collection.add(
    documents=["文档1", "文档2", "文档3"],
    ids=["id1", "id2", "id3"],
    metadatas=[
        {"source": "网页"},
        {"source": "PDF"},
        {"source": "数据库"}
    ]
)

# 查询
results = collection.query(
    query_texts=["查询内容"],
    n_results=3
)

print(results)

import chromadb

# 持久化客户端
client = chromadb.PersistentClient(path="./chroma_db")

# 或使用指定目录
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建向量库
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

# 检索
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke("查询内容")

from pinecone import Pinecone, ServerlessSpec

# 初始化
pc = Pinecone(api_key="your-api-key")

# 创建索引
pc.create_index(
    name="my-index",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

# 连接索引
index = pc.Index("my-index")

# 插入向量
index.upsert(
    vectors=[
        {"id": "vec1", "values": [0.1]*1536, "metadata": {"text": "文档1"}},
        {"id": "vec2", "values": [0.2]*1536, "metadata": {"text": "文档2"}}
    ]
)

# 查询
results = index.query(
    vector=[0.1]*1536,
    top_k=3,
    include_metadata=True
)

from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings

# 创建向量库
vectorstore = PineconeVectorStore(
    index_name="my-index",
    embedding=OpenAIEmbeddings(),
    text_key="text"
)

# 检索
docs = vectorstore.similarity_search("查询内容", k=3)

import weaviate

# 创建客户端
client = weaviate.Client("http://localhost:8080")

# 创建类
client.schema.create_class({
    "class": "Document",
    "vectorizer": "text2vec-transformers",
    "properties": [
        {
            "name": "content",
            "dataType": ["text"]
        },
        {
            "name": "source",
            "dataType": ["string"]
        }
    ]
})

# 添加数据
client.data_object.create(
    class_name="Document",
    data_object={
        "content": "这是一篇关于AI的文章",
        "source": "网站"
    }
)

# 向量搜索
result = client.query.get("Document", ["content", "source"]) \
    .with_near_text({"concepts": ["人工智能"]}) \
    .with_limit(5) \
    .do()

from langchain_weaviate import WeaviateVectorStore
from langchain_openai import OpenAIEmbeddings

# 创建向量库
vectorstore = WeaviateVectorStore.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    weaviate_url="http://localhost:8080"
)

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

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 连接Milvus
connections.connect("default", host="localhost", port="19530")

# 定义字段
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=200)
]

# 创建集合
schema = CollectionSchema(fields, description="文档集合")
collection = Collection("documents", schema)

# 插入数据
import numpy as np
data = [
    np.random.random((100, 128)).tolist(),
    [f"文档{i}" for i in range(100)]
]
collection.insert(data)

# 创建索引
index_params = {
    "metric_type": "L2",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)

# 搜索
collection.load()
results = collection.search(
    data=np.random.random((1, 128)).tolist(),
    anns_field="embedding",
    param={"metric_type": "L2", "params": {"nprobe": 16}},
    limit=5,
    output_fields=["text"]
)

# 1. 使用余弦相似度而非L2
# 适合文本等高维稀疏数据

# 2. 调整搜索参数
results = index.search(
    query,
    k=10,
    params={"nprobe": 32}  # IVF索引的搜索范围
)

# 3. 使用批量查询
results = index.search(batch_queries, k=10)

# 1. 使用PQ压缩
index = faiss.IndexPQ(dimension, 8, 8)

# 2. 使用IVF+PQ
index = faiss.IndexIVFPQ(quantizer, dimension, 100, 8, 8)

# 3. 分片存储
# 对于超大规模数据，使用分布式方案