向量数据库深度调研

CREATE TABLE items (
    id bigserial PRIMARY KEY,
    -- 推荐：大多数 AI 场景使用 halfvec 平衡性能与成本
    embedding halfvec(1536)
);

-- 插入操作（PGVector 会自动处理类型转换）
INSERT INTO items (embedding) VALUES ('[1.1, 2.2, 3.3, ...]');

-- 创建 HNSW 索引
-- m: 每层最大连接数（默认 16，建议 16-64），"路口"的分岔数。分岔越多搜索越快，但路也越宽（内存占用↑）。
-- ef_construction: 构建时搜索宽度（默认 64，建议 100-200）。建路时的探索范围。范围越大路网质量越好，但修路越慢。
CREATE INDEX ON items USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- ef_search: 查询时搜索宽度（默认 40，建议 100-200）。
SET hnsw.ef_search = 100;

-- 创建 IVFFlat 索引
-- ⚠️ 必须表中已有数据（建议 >10万行）才能计算聚类中心
-- lists: 把数据划分成多少个“格子”。rows < 1M: lists = rows / 1000；rows >= 1M: lists = sqrt(rows)。
CREATE INDEX ON items USING ivfflat (embedding vector_l2_ops)
WITH (lists = 100);

-- probes: 探针数，每次查询要翻找最近的几个“格子”（查询时 SET ivfflat.probes）。
SET ivfflat.probes = 10;  -- 建议 sqrt(lists)

-- 混合查询：既要“长得像”，又要“满足条件”
SELECT * FROM items WHERE category_id = 123 ORDER BY embedding <-> '[3,1,2]' LIMIT 5;

-- 宽松顺序（Relaxed Order）：为了凑齐人数，允许稍微牺牲一点排序的严格性（性能更好）
SET hnsw.iterative_scan = relaxed_order;

-- 严格顺序（Strict Order）：必须严格按距离排序，哪怕要扫描更多数据
SET hnsw.iterative_scan = strict_order;
-- SET ivfflat.iterative_scan = relaxed_order;  -- IVFFlat 索引的迭代扫描模式

-- 使用物化 CTE 在宽松顺序下获取高频查询的严格排序
WITH relaxed_results AS MATERIALIZED (
    SELECT id, embedding <-> '[1,2,3]' AS distance
    FROM items WHERE category_id = 123
    ORDER BY distance LIMIT 5
) SELECT * FROM relaxed_results ORDER BY distance + 0;  -- +0 for PG17+

-- 混合搜查令：结合“车牌号”与“素描画”
SELECT id, content,
    -- 综合嫌疑指数 = 车牌匹配度(30%) + 长相相似度(70%)
    ts_rank(to_tsvector('english', content), query) * 0.3  -- [车牌] 关键词匹配得分
    + (1 - (embedding <=> '[...]')) * 0.7                  -- [素描] 向量相似度得分
    AS final_score
FROM items, plainto_tsquery('english', 'machine learning') query
WHERE to_tsvector('english', content) @@ query  -- [初筛] 必须包含关键线索
ORDER BY final_score DESC
LIMIT 10;

-- 查看索引使用情况
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM source_embeddings
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

-- 调整 HNSW 搜索参数
SET hnsw.ef_search = 100;  -- 提升召回率

-- 批量数据导入后重建索引
REINDEX INDEX CONCURRENTLY idx_source_embedding_hnsw;

-- 清理碎片
VACUUM ANALYZE source_embeddings;

-- 标准创建（推荐）
CREATE INDEX ON items USING vchordrq (embedding vector_l2_ops);

-- 极客模式：通过 TOML 风格配置微调参数
CREATE INDEX ON items USING vchordrq (embedding vector_cosine_ops)
WITH (options = $$
    residual_quantization = true  -- [照片增强] 不仅存缩略图，还保留了和原图的差异细节，越看越清
    [build.internal]
    lists = [2000]                -- [城市规划] 强制划分为 2000 个行政区（若不填则 AI 自动规划）
    spherical_centroids = true    -- [球面投影] 适合 Cosine 距离，像在地球仪表面划分区域而不是平面地图
    build_threads = 8             -- [施工队] 8 个工人同时干活
$$);

-- 查询时调整搜索范围（即搜寻多少个相邻行政区）
SET vchordrq.probes TO '10';
SELECT * FROM items ORDER BY embedding <-> '[3,1,2]' LIMIT 10;

-- 依赖 vchord
CREATE EXTENSION IF NOT EXISTS vchord CASCADE;
-- 1. 数据表还是原来的配方（使用 vector 类型）
CREATE TABLE items (embedding vector(1536));

-- 2. 只需要在创建索引时，悄悄将“引擎”名从 hnsw 改为 vchordrq
-- CREATE INDEX ON items USING hnsw ...  <-- 旧引擎
CREATE INDEX ON items USING vchordrq ...  -- <-- 新引擎

/**
 * 离线地图模式 (Disk-Based Index)
 * 适合场景：内存有限，但有一块极速 SSD
 */
CREATE INDEX ON items USING vchordg (embedding vector_cosine_ops)
WITH (options = $$
    bits = 2                -- [地图缩放] 2x 缩放。值越小越省地，但地图越模糊（易指错路）；1 = 极省空间，2 = 兼顾准确；RaBitQ 量化比率，默认 2
    m = 32                  -- [交通枢纽] 规定每个路口最多连接 32 条路。路越多越精准，但路网越复杂；每顶点最大邻居数，默认 32
    ef_construction = 64    -- [勘测范围] 造地图时，先探索周边 64 个路标来确定最佳路线。探索越广，地图质量越高；构建时的搜索范围，默认 64
    alpha = [1.0, 1.2]      -- [绕路容忍度] 允许在建图时稍微绕点路（1.0-1.2倍），以发现潜在的捷径，防止陷入局部死胡同；剪枝时的 alpha 值，默认 [1.0, 1.2]
$$);

-- 启用预过滤
SET vchordrq.prefilter = on;

-- 适用于严格且低成本的过滤条件
-- 1% 选择率时可获得 200% QPS 提升
-- 10% 选择率时可获得 5% QPS 提升

from pymilvus import MilvusClient

# 使用 Milvus Lite 进行本地开发
client = MilvusClient("./milvus_demo.db")

# 创建 Collection
client.create_collection(
    collection_name="papers",
    dimension=1536,
    metric_type="COSINE"
)

# 插入数据
client.insert(
    collection_name="papers",
    data=[
        {"id": 1, "vector": embedding, "title": "ReAct Paper", "abstract": "..."},
        # ...
    ]
)

# 创建索引
client.create_index(
    collection_name="papers",
    field_name="vector",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 128}
)

# 搜索
results = client.search(
    collection_name="papers",
    data=[query_embedding],
    limit=10,
    output_fields=["title", "abstract"]
)

from pymilvus import MilvusClient, DataType, FieldSchema, CollectionSchema

# 使用 Milvus Lite（本地开发）或连接远程服务
client = MilvusClient("./agentic_ai.db")  # Lite 模式

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

# 创建 Collection
client.create_collection(
    collection_name="source_embeddings",
    schema=CollectionSchema(fields, description="学术资源向量嵌入"),
    index_params={
        "index_type": "HNSW",
        "metric_type": "COSINE",
        "params": {"M": 16, "efConstruction": 128}
    }

from pymilvus import MilvusClient
from openai import OpenAI

client = MilvusClient("./agentic_ai.db")
openai_client = OpenAI()

def get_embedding(text: str) -> list:
    """生成文本嵌入向量"""
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def semantic_search(query: str, source_type: str = None, top_k: int = 10):
    """语义相似度搜索"""
    query_embedding = get_embedding(query)

    # 构建过滤条件
    filter_expr = f'source_type == "{source_type}"' if source_type else ""

    results = client.search(
        collection_name="source_embeddings",
        data=[query_embedding],
        limit=top_k,
        filter=filter_expr,
        output_fields=["title", "chunk_text", "source_type"]
    )
    return results

def hybrid_search(query: str, top_k: int = 10):
    """混合搜索（向量 + BM25 全文）"""
    # Milvus 2.4+ 支持 BM25 全文搜索（需在 Collection 中启用全文索引）
    from pymilvus import AnnSearchRequest, RRFRanker

    query_embedding = get_embedding(query)

    # 向量搜索请求
    vector_req = AnnSearchRequest(
        data=[query_embedding],
        anns_field="embedding",
        param={"metric_type": "COSINE", "params": {"ef": 100}},
        limit=top_k * 2
    )

    # BM25 全文搜索请求（需要在 Collection 中启用 BM25）
    bm25_req = AnnSearchRequest(
        data=[query],
        anns_field="chunk_text",
        param={"metric_type": "BM25"},
        limit=top_k * 2
    )

    # 使用 RRF 融合结果
    results = client.hybrid_search(
        collection_name="source_embeddings",
        reqs=[vector_req, bm25_req],
        ranker=RRFRanker(k=60),
        limit=top_k,
        output_fields=["title", "chunk_text"]
    )
    return results

from llama_index.core import VectorStoreIndex, Settings
from llama_index.vector_stores.milvus import MilvusVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding

# 配置嵌入模型
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 连接 Milvus（支持 Lite / Standalone / Distributed）
vector_store = MilvusVectorStore(
    uri="./agentic_ai.db",  # Milvus Lite
    # uri="http://localhost:19530",  # Milvus Standalone
    collection_name="source_embeddings",
    dim=1536,
    overwrite=False
)

# 创建索引
index = VectorStoreIndex.from_vector_store(vector_store)

# RAG 查询
query_engine = index.as_query_engine(
    similarity_top_k=10,
    response_mode="tree_summarize"
)

response = query_engine.query(
    "ReAct 和 Chain-of-Thought 有什么区别？"
)
print(response)

from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 连接 Milvus 向量存储
vector_store = Milvus(
    embedding_function=embeddings,
    collection_name="source_embeddings",
    connection_args={
        "uri": "./agentic_ai.db"  # Milvus Lite
        # "uri": "http://localhost:19530"  # Milvus Standalone
    }
)

# 创建检索器
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 10}
)

# 构建 RAG 链
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 执行查询
result = qa_chain.invoke({"query": "什么是 Agentic RAG？"})
print(result["result"])

import weaviate

# ... (连接与集合创建代码略) ...
client = weaviate.connect_to_wcs(
    cluster_url="YOUR_WCS_URL",
    auth_credentials=weaviate.auth.AuthApiKey("YOUR_API_KEY")
)

# 创建 Collection（自动向量化）
collection = client.collections.create(
    name="Article",
    vectorizer_config=weaviate.Configure.Vectorizer.text2vec_openai()
)

# 插入数据（自动生成向量）
collection.data.insert({
    "title": "AI 技术发展",
    "content": "人工智能正在改变世界..."
})

# 1. 语义搜索 (The Radar)
# "这就好比用雷达扫描，寻找‘意思’上接近的目标，不在乎用词是否完全一致"
results = collection.query.near_text(
    query="机器学习的未来",
    limit=5
)

# 2. 混合搜索 (The Fusion)
# "结合雷达与字典。alpha 参数就像‘混音台推杆’："
#  - alpha = 0.0 (纯理性): 完全依赖关键词匹配 (BM25)，就像传统的 SQL 搜索
#  - alpha = 1.0 (纯感性): 完全依赖语义理解 (Vector)，就像人类的直觉
#  - alpha = 0.5 (均衡): 理性与感性的完美融合，通常能获得最佳效果
results = collection.query.hybrid(
    query="AI applications",
    alpha=0.5,
    limit=5
)

# 3. 生成式搜索 (The Reader)
# "不仅帮你把相关书籍找出来(检索)，还当场读一遍并回答你的问题(生成)"
# 这就是内置的 RAG 能力，极大简化了应用开发流程
results = collection.generate.near_text(
    query="人工智能",
    grouped_task="请基于以下检索结果，总结 AI 的核心发展趋势",
    limit=3
)

# 启用 SQ 压缩（兼顾速度与精度的最佳平衡点）
collection = client.collections.create(
    name="Article",
    vectorizer_config=weaviate.Configure.Vectorizer.text2vec_openai(),
    vector_index_config=weaviate.Configure.VectorIndex.hnsw(
        quantizer=weaviate.Configure.VectorIndex.Quantizer.sq()
    )
)

from pinecone import Pinecone

# 初始化客户端
pc = Pinecone(api_key="YOUR_API_KEY")

# 创建 Dense Index（带集成嵌入）
pc.create_index_for_model(
    name="my-index",
    cloud="aws",
    region="us-east-1",
    embed={
        "model": "llama-text-embed-v2",
        "field_map": {"text": "chunk_text"}
    }
)

# 创建 Sparse Index
pc.create_index(
    name="sparse-index",
    dimension=None,  # Sparse 无需指定
    metric="dotproduct",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

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

# 文本搜索（集成嵌入）
results = index.query(
    data={"inputs": {"text": "What is machine learning?"}},
    top_k=10,
    include_metadata=True
)

# 带元数据过滤的搜索
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=10,
    filter={"genre": {"$eq": "technology"}}
)

# 混合搜索（需要同时使用 Dense + Sparse 索引）

# 双索引混合搜索
# 1. 创建 Dense + Sparse 索引
pc.create_index_for_model(
    name="dense-index",
    cloud="aws", region="us-east-1",
    embed={"model": "llama-text-embed-v2", "field_map": {"text": "chunk_text"}}
)
pc.create_index_for_model(
    name="sparse-index",
    cloud="aws", region="us-east-1",
    embed={"model": "pinecone-sparse-english-v0", "field_map": {"text": "chunk_text"}}
)

# 2. 分别查询后使用 RRF 融合结果

# 集成重排序 - 在 search 中直接使用
ranked_results = index.search(
    namespace="example-namespace",
    query={"inputs": {"text": "Disease prevention"}, "top_k": 4},
    rerank={
        "model": "bge-reranker-v2-m3",
        "top_n": 2,
        "rank_fields": ["chunk_text"]
    },
    fields=["category", "chunk_text"]
)

-- 创建向量列（必须指定 vector_length）
CREATE TABLE Documents (
    DocId INT64 NOT NULL,
    DocContents BYTES(MAX),
    DocEmbedding ARRAY<FLOAT32>(vector_length=>768) NOT NULL,
) PRIMARY KEY (DocId);

-- 创建 2 层向量索引（适用于 < 1000万行）
CREATE VECTOR INDEX DocEmbeddingIndex
ON Documents(DocEmbedding)
OPTIONS (
    distance_type = 'COSINE',
    tree_depth = 2,
    num_leaves = 1000
)
STORING (DocContents);

-- 创建 3 层向量索引（适用于大规模数据）
CREATE VECTOR INDEX DocEmbeddingIndexLarge
ON Documents(DocEmbedding)
OPTIONS (
    distance_type = 'COSINE',
    tree_depth = 3,
    num_branches = 1000,
    num_leaves = 1000000
);

-- ANN 搜索（使用向量索引加速）
SELECT DocId, DocContents
FROM Documents
WHERE DocEmbedding IS NOT NULL
ORDER BY APPROXIMATE_COSINE_DISTANCE(
    ARRAY<FLOAT32>[0.1, 0.2, ...],  -- 查询向量
    DocEmbedding,
    options => JSON '{"num_leaves_to_search": 10}'  -- 搜索参数
)
LIMIT 10;

-- KNN 搜索（精确搜索，适用于小数据集）
SELECT DocId, DocContents
FROM Documents
ORDER BY COSINE_DISTANCE(
    ARRAY<FLOAT32>[0.1, 0.2, ...],
    DocEmbedding
)
LIMIT 10;

-- 带过滤条件的 ANN 搜索
SELECT DocId, DocContents
FROM Documents
WHERE DocEmbedding IS NOT NULL
  AND Category = 'Technology'  -- 标量过滤
ORDER BY APPROXIMATE_EUCLIDEAN_DISTANCE(
    ARRAY<FLOAT32>[0.1, 0.2, ...],
    DocEmbedding,
    options => JSON '{"num_leaves_to_search": 20}'  -- 带过滤时增大搜索范围
)
LIMIT 10;

-- 1. 注册 Vertex AI 嵌入模型
CREATE MODEL TextEmbeddingModel
INPUT(content STRING(MAX))
OUTPUT(
    embeddings STRUCT<
        statistics STRUCT<truncated BOOL, token_count FLOAT64>,
        values ARRAY<FLOAT64>
    >
)
REMOTE OPTIONS (
    endpoint = '//aiplatform.googleapis.com/projects/my-project/locations/us-central1/publishers/google/models/text-embedding-004'
);

-- 2. 使用 ML.PREDICT 生成嵌入并存储
INSERT INTO Documents (DocId, Content, Embedding)
SELECT
    @DocId,
    @Content,
    ARRAY<FLOAT32>(embeddings.values)  -- 类型转换
FROM ML.PREDICT(
    MODEL TextEmbeddingModel,
    (SELECT @Content AS content)
);

-- 3. 查询时动态生成查询向量
SELECT DocId, Content
FROM Documents
WHERE Embedding IS NOT NULL
ORDER BY APPROXIMATE_COSINE_DISTANCE(
    (SELECT ARRAY<FLOAT32>(embeddings.values)
     FROM ML.PREDICT(MODEL TextEmbeddingModel, (SELECT @Query AS content))),
    Embedding
)
LIMIT 10;

from langchain_google_spanner import SpannerVectorStore
from langchain_google_vertexai import VertexAIEmbeddings

# 初始化嵌入模型
embeddings = VertexAIEmbeddings(model_name="text-embedding-004")

# 连接 Spanner Vector Store
vector_store = SpannerVectorStore(
    instance_id="my-instance",
    database_id="my-database",
    table_name="documents",
    embedding_service=embeddings,
)

# 语义搜索
results = vector_store.similarity_search(
    query="什么是 Agentic AI？",
    k=5
)

# 与 LLM 链集成
from langchain.chains import RetrievalQA
from langchain_google_vertexai import ChatVertexAI

llm = ChatVertexAI(model_name="gemini-1.5-pro")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vector_store.as_retriever(search_kwargs={"k": 10}),
    chain_type="stuff"
)

answer = qa_chain.invoke({"query": "解释 GraphRAG 的工作原理"})