向量数据库

-- PostgreSQL + pgvector
CREATE EXTENSION vector;

CREATE TABLE items (
    id bigserial PRIMARY KEY,
    content text,
    embedding vector(1536)
);

-- 创建索引
CREATE INDEX ON items USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

-- 相似度查询
SELECT content, embedding <=> '[0.1, 0.2, ...]' AS distance
FROM items
ORDER BY distance
LIMIT 5;

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import numpy as np

# 初始化客户端
client = QdrantClient("localhost", port=6333)

# 创建集合
client.create_collection(
    collection_name="articles",
    vectors_config=VectorParams(size=768, distance=Distance.COSINE),
)

# 插入向量
points = [
    PointStruct(
        id=idx,
        vector=embedding.tolist(),
        payload={"title": title, "content": content}
    )
    for idx, (title, content, embedding) in enumerate(documents)
]

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

# 搜索相似文档
search_result = client.search(
    collection_name="articles",
    query_vector=query_embedding.tolist(),
    limit=5,
    query_filter={
        "must": [{"key": "category", "match": {"value": "technology"}}]
    }
)

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

# 连接配置
connections.connect(
    alias="default",
    host='localhost',
    port='19530'
)

# 定义schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=200),
    FieldSchema(name="timestamp", dtype=DataType.INT64)
]

schema = CollectionSchema(fields, "文章向量集合")

# 创建集合
collection = Collection("articles", schema)

# 创建索引（重要！）
index_params = {
    "metric_type": "IP",  # 内积
    "index_type": "IVF_FLAT",
    "params": {"nlist": 2048}
}

collection.create_index(
    field_name="embedding", 
    index_params=index_params
)

# 加载到内存
collection.load()

class VectorRAG:
    def __init__(self, vector_db, embedding_model, llm):
        self.vector_db = vector_db
        self.embedding_model = embedding_model
        self.llm = llm
    
    def add_documents(self, documents):
        """添加文档到向量数据库"""
        for doc in documents:
            # 1. 文本分块
            chunks = self.split_text(doc.content)
            
            # 2. 生成嵌入
            embeddings = self.embedding_model.encode(chunks)
            
            # 3. 存储到向量数据库
            self.vector_db.insert(
                embeddings=embeddings,
                metadatas=[{
                    "doc_id": doc.id,
                    "chunk_id": i,
                    "text": chunk
                } for i, chunk in enumerate(chunks)]
            )
    
    def query(self, question, top_k=5):
        """RAG查询流程"""
        # 1. 问题向量化
        query_embedding = self.embedding_model.encode(question)
        
        # 2. 向量检索
        results = self.vector_db.search(
            query_embedding, 
            top_k=top_k
        )
        
        # 3. 构建上下文
        context = "\n".join([r.metadata["text"] for r in results])
        
        # 4. 生成回答
        prompt = f"""基于以下信息回答问题：
        
        {context}
        
        问题：{question}
        """
        
        return self.llm.generate(prompt)

# 推荐：Chroma（轻量级、易上手）
import chromadb

client = chromadb.Client()
collection = client.create_collection("demo")
collection.add(
    embeddings=[[1.0, 2.0, 3.0]],
    documents=["示例文档"],
    ids=["doc1"]
)

# 推荐：Milvus + Docker Compose
version: '3.5'
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
  minio:
    image: minio/minio:latest
  milvus:
    image: milvusdb/milvus:latest
    depends_on:
      - etcd
      - minio

-- 推荐：pgvector（复用基础设施）
ALTER SYSTEM SET shared_preload_libraries = 'vector';
CREATE EXTENSION vector;

-- 优化配置
ALTER SYSTEM SET max_parallel_workers_per_gather = 4;
ALTER SYSTEM SET effective_cache_size = '4GB';