向量数据库详解

# OpenAI Embeddings
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536
)

# 单个文本
vector = embeddings.embed_query("什么是向量数据库？")
print(f"维度: {len(vector)}")

# 批量文本
texts = ["文本1", "文本2", "文本3"]
vectors = embeddings.embed_documents(texts)

# 本地模型
from langchain_community.embeddings import HuggingFaceEmbeddings

local_embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={'device': 'cuda'},  # GPU 加速
    encode_kwargs={'normalize_embeddings': True}
)

import numpy as np

def cosine_similarity(a, b):
    """余弦相似度：范围 [-1, 1]，越接近 1 越相似"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 示例
vec1 = np.array([1, 2, 3])
vec2 = np.array([1.1, 2.1, 2.9])

similarity = cosine_similarity(vec1, vec2)
print(f"余弦相似度: {similarity}")

def euclidean_distance(a, b):
    """欧氏距离：越小越相似"""
    return np.linalg.norm(a - b)

# 示例
distance = euclidean_distance(vec1, vec2)
print(f"欧氏距离: {distance}")

def dot_product(a, b):
    """点积：越大越相似"""
    return np.dot(a, b)

# 示例
dot = dot_product(vec1, vec2)
print(f"点积: {dot}")

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")

# 创建 Index
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([
    ("id1", [0.1, 0.2, ...], {"category": "tech"}),
    ("id2", [0.3, 0.4, ...], {"category": "news"})
])

# 搜索
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True
)

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

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

# 定义 Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535)
]

schema = CollectionSchema(fields, "my_collection")

# 创建 Collection
collection = Collection("my_collection", schema)

# 插入数据
data = [
    [[0.1, 0.2, ...], [0.3, 0.4, ...]],  # embeddings
    ["文本1", "文本2"]                     # texts
]
collection.insert(data)

# 创建索引
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 200}
}
collection.create_index("embedding", index_params)

# 搜索
results = collection.search(
    data=[[0.1, 0.2, ...]],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 50}},
    limit=5
)

import chromadb
from chromadb.config import Settings

# 创建客户端
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

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

# 添加文档
collection.add(
    documents=["文档1内容", "文档2内容"],
    embeddings=[[0.1, 0.2, ...], [0.3, 0.4, ...]],
    metadatas=[{"source": "doc1"}, {"source": "doc2"}],
    ids=["id1", "id2"]
)

# 查询
results = collection.query(
    query_embeddings=[[0.1, 0.2, ...]],
    n_results=5
)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

# 连接
client = QdrantClient(url="http://localhost:6333")

# 创建 Collection
client.create_collection(
    collection_name="my_collection",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

# 插入数据
client.upsert(
    collection_name="my_collection",
    points=[
        PointStruct(id=1, vector=[0.1, 0.2, ...], payload={"text": "文档1"}),
        PointStruct(id=2, vector=[0.3, 0.4, ...], payload={"text": "文档2"})
    ]
)

# 搜索（带过滤）
results = client.search(
    collection_name="my_collection",
    query_vector=[0.1, 0.2, ...],
    query_filter={
        "must": [
            {"key": "category", "match": {"value": "tech"}}
        ]
    },
    limit=5
)

import chromadb
from chromadb.config import Settings
from langchain_openai import OpenAIEmbeddings

# 1. 创建客户端
client = chromadb.PersistentClient(path="./chroma_db")

# 2. 创建 Collection
collection = client.get_or_create_collection("documents")

# 3. 初始化 Embeddings
embeddings = OpenAIEmbeddings()

# 4. 添加文档
def add_documents(texts, metadatas=None):
    vectors = embeddings.embed_documents(texts)
    collection.add(
        documents=texts,
        embeddings=vectors,
        metadatas=metadatas or [{}] * len(texts),
        ids=[f"doc_{i}" for i in range(len(texts))]
    )

# 5. 搜索
def search(query, n_results=5):
    query_vector = embeddings.embed_query(query)
    return collection.query(
        query_embeddings=[query_vector],
        n_results=n_results
    )

# 使用
add_documents(
    texts=["RAG 是检索增强生成", "向量数据库存储向量"],
    metadatas=[{"topic": "AI"}, {"topic": "DB"}]
)

results = search("什么是 RAG？")
print(results)

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import numpy as np

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

# 定义 Schema
fields = [
    FieldSchema("id", DataType.INT64, is_primary=True),
    FieldSchema("content", DataType.VARCHAR, max_length=65535),
    FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=1536)
]

schema = CollectionSchema(fields, "documents")
collection = Collection("documents", schema)

# 插入数据
def insert_documents(contents):
    from langchain_openai import OpenAIEmbeddings
    emb = OpenAIEmbeddings()

    vectors = emb.embed_documents(contents)

    data = [
        [i for i in range(len(contents))],  # ids
        contents,                            # contents
        vectors                              # embeddings
    ]

    collection.insert(data)
    collection.flush()

# 创建索引
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 200}
}
collection.create_index("embedding", index_params)
collection.load()

# 搜索
def search(query, top_k=5):
    from langchain_openai import OpenAIEmbeddings
    emb = OpenAIEmbeddings()

    query_vector = emb.embed_query(query)

    results = collection.search(
        data=[query_vector],
        anns_field="embedding",
        param={"metric_type": "COSINE", "params": {"ef": 50}},
        limit=top_k,
        output_fields=["content"]
    )

    return results

import os
from pinecone import Pinecone, ServerlessSpec
from langchain_openai import OpenAIEmbeddings

# 初始化
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

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

index = pc.Index("rag-index")

# 批量插入
def upsert_documents(documents):
    emb = OpenAIEmbeddings()

    vectors = []
    for i, doc in enumerate(documents):
        vector = emb.embed_query(doc["content"])
        vectors.append({
            "id": f"doc_{i}",
            "values": vector,
            "metadata": {"content": doc["content"], **doc.get("metadata", {})}
        })

    index.upsert(vectors)

# 搜索
def search(query, top_k=5, filter=None):
    emb = OpenAIEmbeddings()
    query_vector = emb.embed_query(query)

    results = index.query(
        vector=query_vector,
        top_k=top_k,
        include_metadata=True,
        filter=filter
    )

    return results

# 1. 选择合适的索引类型

# 小规模（<10K）：Flat
index_params = {"index_type": "FLAT"}

# 中等规模（10K-1M）：IVF
index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "COSINE",
    "params": {"nlist": 100}
}

# 大规模（>1M）：HNSW
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 200}
}

# 内存受限：IVF + PQ
index_params = {
    "index_type": "IVF_PQ",
    "params": {
        "nlist": 100,
        "m": 8  # PQ 子空间数
    }
}

# 1. 调整 ef 参数
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 100}},  # 增加 ef 提高精度
    limit=10
)

# 2. 使用过滤减少搜索空间
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "COSINE"},
    limit=10,
    expr="category == 'tech'"  # 只搜索 tech 类别
)

# 3. 批量查询
results = collection.search(
    data=[query_vector1, query_vector2, query_vector3],  # 批量
    anns_field="embedding",
    param={"metric_type": "COSINE"},
    limit=10
)

# 1. 使用量化减少内存
index_params = {
    "index_type": "IVF_PQ",
    "params": {
        "nlist": 100,
        "m": 16,           # PQ 参数
        "nbits": 8         # 每子空间位数
    }
}

# 2. 定期压缩
collection.compact()
collection.flush()

# 3. 分区存储
# 按类别分区
tech_collection = Collection("tech_docs", schema)
news_collection = Collection("news_docs", schema)