docker run -d --name kb-pg -p 5432:5432 \
  -e POSTGRES_PASSWORD=pw pgvector/pgvector:pg17

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
  id          bigserial PRIMARY KEY,
  tenant_id   text          NOT NULL,
  doc_id      text          NOT NULL,
  version     int           NOT NULL DEFAULT 1,
  content     text          NOT NULL,           -- 原文，给模型和用户看
  content_seg text          NOT NULL,           -- 应用层分词后用空格连接（中文必需，见下方说明）
  embedding   vector(1024)  NOT NULL,
  tsv         tsvector GENERATED ALWAYS AS (to_tsvector('simple', content_seg)) STORED
);

-- 向量索引：m / ef_construction 为 pgvector 默认值，建库后不再改
CREATE INDEX chunks_hnsw ON chunks USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);
CREATE INDEX chunks_tsv    ON chunks USING gin (tsv);
CREATE INDEX chunks_tenant ON chunks (tenant_id);

import jieba, numpy as np, psycopg
from pgvector.psycopg import register_vector

conn = psycopg.connect("postgresql://postgres:pw@localhost:5432/postgres", autocommit=True)
register_vector(conn)                                  # 让 numpy 数组可直接作为 vector 参数

seg = lambda s: " ".join(jieba.lcut(s))

def upsert_doc(tenant, doc_id, version, chunks, embed):
    # 更新 = 先删旧版本的块，再写新块（不要只追加）
    conn.execute("DELETE FROM chunks WHERE tenant_id=%s AND doc_id=%s", (tenant, doc_id))
    for text in chunks:
        conn.execute(
            "INSERT INTO chunks (tenant_id, doc_id, version, content, content_seg, embedding) VALUES (%s,%s,%s,%s,%s,%s)",
            (tenant, doc_id, version, text, seg(text), np.array(embed(text))))

HYBRID_SQL = """
WITH vec AS (
  SELECT id, row_number() OVER (ORDER BY embedding <=> %(qv)s) AS rnk
  FROM chunks WHERE tenant_id = %(tenant)s
  ORDER BY embedding <=> %(qv)s LIMIT 20
), lex AS (
  SELECT id, row_number() OVER (ORDER BY ts_rank_cd(tsv, q) DESC) AS rnk
  FROM chunks, to_tsquery('simple', %(qs)s) q
  WHERE tenant_id = %(tenant)s AND tsv @@ q
  ORDER BY ts_rank_cd(tsv, q) DESC LIMIT 20
)
SELECT c.id, c.doc_id, c.content,
       COALESCE(1.0 / (60 + vec.rnk), 0) + COALESCE(1.0 / (60 + lex.rnk), 0) AS rrf
FROM vec FULL OUTER JOIN lex USING (id)
JOIN chunks c USING (id)
ORDER BY rrf DESC LIMIT 5
"""

def search(tenant, question, embed):
    terms = [t for t in jieba.lcut(question) if t.strip() and t.isalnum()]
    qs = " | ".join(terms) or "''"              # 关键词路用 OR 连接，提升召回
    conn.execute("SET hnsw.ef_search = 100")                 # 查询期调召回/延迟
    conn.execute("SET hnsw.iterative_scan = strict_order")   # 带过滤时自动继续扫索引（pgvector 0.8+）
    return conn.execute(HYBRID_SQL, {"qv": np.array(embed(question)), "qs": qs, "tenant": tenant}).fetchall()

EXPLAIN (ANALYZE, BUFFERS)
SELECT id FROM chunks WHERE tenant_id = 'acme' ORDER BY embedding <=> '[0.1, 0.2, ...]' LIMIT 20;
-- 期望看到：Index Scan using chunks_hnsw ...
-- 如果是 Seq Scan：检查向量维度、距离算子是否与索引的 vector_cosine_ops 一致（<=>）

curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh
bash standalone_embed.sh start        # 默认监听 19530
pip install -U pymilvus

from pymilvus import (MilvusClient, DataType, Function, FunctionType,
                     AnnSearchRequest, RRFRanker)

client = MilvusClient(uri="http://localhost:19530")

schema = client.create_schema(auto_id=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("tenant_id", DataType.VARCHAR, max_length=64, is_partition_key=True)  # 多租户：分区键
schema.add_field("doc_id", DataType.VARCHAR, max_length=128)
schema.add_field("text", DataType.VARCHAR, max_length=8192,
                 enable_analyzer=True, analyzer_params={"type": "chinese"})      # 中文分析器；英文用 "standard"，以官方文档为准
schema.add_field("dense", DataType.FLOAT_VECTOR, dim=1024)
schema.add_field("sparse", DataType.SPARSE_FLOAT_VECTOR)                          # 由 BM25 函数自动生成

schema.add_function(Function(name="text_bm25", function_type=FunctionType.BM25,
                             input_field_names=["text"], output_field_names=["sparse"]))

index = client.prepare_index_params()
index.add_index(field_name="dense", index_type="HNSW", metric_type="COSINE",
                params={"M": 16, "efConstruction": 200})
index.add_index(field_name="sparse", index_type="SPARSE_INVERTED_INDEX", metric_type="BM25")

client.create_collection("kb", schema=schema, index_params=index)

def upsert_doc(tenant, doc_id, chunks, embed):
    client.delete("kb", filter=f'tenant_id == "{tenant}" and doc_id == "{doc_id}"')   # 先删旧块
    client.insert("kb", [{"tenant_id": tenant, "doc_id": doc_id, "text": t, "dense": embed(t)} for t in chunks])
    # 注意：不用自己算 sparse，BM25 函数会根据 text 自动生成

def search(tenant, question, embed):
    flt = f'tenant_id == "{tenant}"'
    dense_req  = AnnSearchRequest(data=[embed(question)], anns_field="dense",
                                  param={"metric_type": "COSINE", "params": {"ef": 100}}, limit=20, expr=flt)
    sparse_req = AnnSearchRequest(data=[question], anns_field="sparse",           # 直接传查询文本
                                  param={"metric_type": "BM25"}, limit=20, expr=flt)
    res = client.hybrid_search("kb", reqs=[dense_req, sparse_req],
                               ranker=RRFRanker(60), limit=5, output_fields=["doc_id", "text"])
    return [(h["entity"]["doc_id"], h["entity"]["text"], h["distance"]) for h in res[0]]