向量数据库完全指南【2026】- 原理、选型与实战

from openai import OpenAI

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding

text = "向量数据库是AI应用的核心基础设施"
embedding = get_embedding(text)
print(f"向量维度: {len(embedding)}")  # 输出: 向量维度: 1536

# HNSW参数配置示例
hnsw_params = {
    "M": 16,           # 每个节点的最大连接数
    "ef_construction": 200,  # 构建时的搜索宽度
    "ef_search": 100   # 查询时的搜索宽度
}

性能优先：Qdrant > Milvus > Weaviate > Chroma
成本优先：Chroma > Qdrant > Milvus > Pinecone
易用性：Chroma > Pinecone > Weaviate > Milvus

import chromadb
from chromadb.utils import embedding_functions

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-api-key",
    model_name="text-embedding-3-small"
)

client = chromadb.PersistentClient(path="./chroma_db")

collection = client.get_or_create_collection(
    name="documents",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine"}
)

documents = [
    "向量数据库是存储和检索高维向量的专用数据库",
    "HNSW算法通过分层图结构实现高效的近似最近邻搜索",
    "RAG技术结合检索和生成，提升大模型的回答质量",
    "Embedding模型将文本转换为稠密的数值向量表示"
]

collection.add(
    documents=documents,
    ids=[f"doc_{i}" for i in range(len(documents))]
)

results = collection.query(
    query_texts=["什么是向量搜索"],
    n_results=2
)

print("搜索结果：")
for doc, distance in zip(results['documents'][0], results['distances'][0]):
    print(f"  相似度: {1-distance:.4f} | {doc[:50]}...")

from pinecone import Pinecone, ServerlessSpec
from openai import OpenAI

pc = Pinecone(api_key="your-pinecone-api-key")
openai_client = OpenAI()

index_name = "document-search"

if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

index = pc.Index(index_name)

def get_embedding(text: str) -> list[float]:
    response = openai_client.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    )
    return response.data[0].embedding

documents = [
    {"id": "1", "text": "向量数据库支持语义搜索", "category": "database"},
    {"id": "2", "text": "HNSW是高效的索引算法", "category": "algorithm"},
    {"id": "3", "text": "RAG提升大模型回答质量", "category": "llm"}
]

vectors = []
for doc in documents:
    embedding = get_embedding(doc["text"])
    vectors.append({
        "id": doc["id"],
        "values": embedding,
        "metadata": {"text": doc["text"], "category": doc["category"]}
    })

index.upsert(vectors=vectors)

query = "如何提升AI回答的准确性"
query_embedding = get_embedding(query)

results = index.query(
    vector=query_embedding,
    top_k=2,
    include_metadata=True
)

print("搜索结果：")
for match in results.matches:
    print(f"  得分: {match.score:.4f} | {match.metadata['text']}")

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue

client = QdrantClient(path="./qdrant_db")

client.recreate_collection(
    collection_name="articles",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

points = [
    PointStruct(
        id=1,
        vector=get_embedding("向量数据库入门指南"),
        payload={"title": "向量数据库入门", "category": "tutorial", "views": 1000}
    ),
    PointStruct(
        id=2,
        vector=get_embedding("HNSW算法深度解析"),
        payload={"title": "HNSW算法解析", "category": "algorithm", "views": 500}
    ),
    PointStruct(
        id=3,
        vector=get_embedding("RAG应用实战案例"),
        payload={"title": "RAG实战", "category": "tutorial", "views": 2000}
    )
]

client.upsert(collection_name="articles", points=points)

results = client.search(
    collection_name="articles",
    query_vector=get_embedding("学习向量数据库"),
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="tutorial"))
        ]
    ),
    limit=2
)

print("过滤搜索结果：")
for result in results:
    print(f"  得分: {result.score:.4f} | {result.payload['title']}")

用户问题 → Embedding → 向量检索 → 相关文档 → LLM生成 → 回答

from openai import OpenAI
import chromadb
from chromadb.utils import embedding_functions

class SimpleRAG:
    def __init__(self):
        self.openai = OpenAI()
        self.chroma = chromadb.PersistentClient(path="./rag_db")
        self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key="your-api-key",
            model_name="text-embedding-3-small"
        )
        self.collection = self.chroma.get_or_create_collection(
            name="knowledge_base",
            embedding_function=self.embedding_fn
        )
    
    def add_documents(self, documents: list[str], ids: list[str] = None):
        if ids is None:
            ids = [f"doc_{i}" for i in range(len(documents))]
        self.collection.add(documents=documents, ids=ids)
    
    def query(self, question: str, top_k: int = 3) -> str:
        results = self.collection.query(
            query_texts=[question],
            n_results=top_k
        )
        
        context = "\n".join(results['documents'][0])
        
        response = self.openai.chat.completions.create(
            model="gpt-4-turbo",
            messages=[
                {"role": "system", "content": f"基于以下上下文回答问题：\n\n{context}"},
                {"role": "user", "content": question}
            ]
        )
        
        return response.choices[0].message.content

rag = SimpleRAG()

knowledge = [
    "向量数据库是专门用于存储和检索高维向量的数据库系统。",
    "HNSW算法通过构建分层的小世界图来实现高效的近似最近邻搜索。",
    "Pinecone是一个完全托管的云向量数据库服务。",
    "Chroma是一个轻量级的开源向量数据库，适合本地开发。"
]
rag.add_documents(knowledge)

answer = rag.query("什么是向量数据库？有哪些常用的产品？")
print(answer)

# HNSW参数优化建议
hnsw_config = {
    "M": 16,                    # 小数据集可降低到8
    "ef_construction": 200,     # 构建质量，越高越好但越慢
    "ef_search": 100            # 查询精度，根据召回率要求调整
}

# IVF参数优化建议
ivf_config = {
    "nlist": 1024,              # 聚类数，建议为sqrt(N)到4*sqrt(N)
    "nprobe": 16                # 搜索簇数，越高召回率越高
}

# 批量插入而非逐条插入
batch_size = 100
for i in range(0, len(documents), batch_size):
    batch = documents[i:i+batch_size]
    collection.add(
        documents=[d["text"] for d in batch],
        ids=[d["id"] for d in batch],
        metadatas=[d["metadata"] for d in batch]
    )

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_cached_embedding(text: str) -> tuple:
    embedding = get_embedding(text)
    return tuple(embedding)