Milvus 分布式向量数据库¶

# 最简单的安装，就是一个 pip 包
pip install pymilvus  # 安装后自带 Milvus Lite

from pymilvus import MilvusClient  # 导入客户端

# Milvus Lite：数据存在本地文件，适合学习和原型
client = MilvusClient("./milvus_demo.db")  # 指定数据库文件路径
# 就这一行，Milvus 就在你的 Python 进程里跑起来了！

# 下载 Docker Compose 文件
wget https://github.com/milvus-io/milvus/releases/download/v2.6.15/milvus-standalone-docker-compose.yml \
  -O docker-compose.yml

# 启动 Milvus（包含 etcd + MinIO + Milvus）
docker compose up -d  # 后台启动

# 验证服务
curl http://localhost:19530/v1/vector/collections  # 列出集合

pip install pymilvus  # 安装 Milvus Python SDK

# 如果需要内置模型支持（自动嵌入）
pip install "pymilvus[model]"  # 包含嵌入模型

from pymilvus import MilvusClient  # 导入客户端

# 连接到 Milvus（Lite 模式或远程服务器）
client = MilvusClient("./milvus_demo.db")  # Lite 模式
# client = MilvusClient(uri="http://localhost:19530")  # 远程服务器

# 创建集合（快速模式，自动配置）
client.create_collection(
    collection_name="papers",  # 集合名
    dimension=384              # 向量维度（取决于嵌入模型）
)

# 准备数据
data = [
    {"id": 1, "vector": [0.1, 0.2, ...], "text": "宏基因组分析流程", "year": 2024},
    {"id": 2, "vector": [0.3, 0.1, ...], "text": "16S rRNA 测序方法", "year": 2023},
    {"id": 3, "vector": [0.2, 0.4, ...], "text": "肠道菌群与糖尿病", "year": 2025},
]

# 插入数据
result = client.insert(
    collection_name="papers",
    data=data
)
print(f"插入了 {result['insert_count']} 条数据")

# 基本搜索
results = client.search(
    collection_name="papers",
    data=[[0.15, 0.25, ...]],  # 查询向量（列表的列表）
    limit=3,                    # 返回前 3 个结果
    output_fields=["text", "year"]  # 返回哪些字段
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit['id']}")
        print(f"距离: {hit['distance']:.4f}")
        print(f"文本: {hit['entity']['text']}")
        print(f"年份: {hit['entity']['year']}")
        print("---")

# 组合向量搜索 + 标量过滤
results = client.search(
    collection_name="papers",
    data=[[0.15, 0.25, ...]],
    limit=5,
    filter="year >= 2024",            # 只搜索 2024 年及以后的
    output_fields=["text", "year"]
)

# 更复杂的过滤条件
results = client.search(
    collection_name="papers",
    data=[[0.15, 0.25, ...]],
    limit=5,
    filter='year >= 2023 and text like "%基因组%"',  # 年份 + 文本匹配
    output_fields=["text", "year"]
)

from pymilvus import MilvusClient
from sentence_transformers import SentenceTransformer  # 嵌入模型

# 加载嵌入模型
model = SentenceTransformer("all-MiniLM-L6-v2")  # 384 维

# 连接 Milvus
client = MilvusClient("./bioinfo_search.db")

# 创建集合
client.create_collection(
    collection_name="bioinfo_tools",
    dimension=384
)

# 准备工具文档
tools = [
    {"name": "Kraken2", "desc": "快速物种分类工具，基于k-mer匹配"},
    {"name": "MEGAHIT", "desc": "内存高效的宏基因组组装工具"},
    {"name": "MetaBAT2", "desc": "自动化宏基因组binning工具"},
    {"name": "CheckM2", "desc": "基因组完整性和污染度评估工具"},
    {"name": "GTDB-Tk", "desc": "基于GTDB数据库的物种注释工具"},
]

# 生成嵌入并插入
vectors = model.encode([t["desc"] for t in tools])
data = [
    {"id": i, "vector": v.tolist(), "name": t["name"], "desc": t["desc"]}
    for i, (v, t) in enumerate(zip(vectors, tools))
]
client.insert(collection_name="bioinfo_tools", data=data)

# 语义搜索
query = "有什么工具可以组装宏基因组数据？"
query_vector = model.encode(query).tolist()

results = client.search(
    collection_name="bioinfo_tools",
    data=[query_vector],
    limit=3,
    output_fields=["name", "desc"]
)

for hits in results:
    for hit in hits:
        print(f"{hit['entity']['name']}: {hit['entity']['desc']}")

from pymilvus import MilvusClient, DataType, Function, FunctionType

client = MilvusClient("./fulltext_demo.db")

# 创建支持全文搜索的集合（Schema 模式）
from pymilvus import CollectionSchema, FieldSchema

schema = CollectionSchema([
    FieldSchema("id", DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema("text", DataType.VARCHAR, max_length=1000,
                enable_analyzer=True),  # 启用分词器
    FieldSchema("sparse", DataType.SPARSE_FLOAT_VECTOR)  # 稀疏向量字段
])

# 添加 BM25 函数
bm25_function = Function(
    name="text_bm25",
    input_field_names=["text"],      # 输入：文本字段
    output_field_names=["sparse"],   # 输出：稀疏向量
    function_type=FunctionType.BM25  # BM25 全文搜索
)
schema.add_function(bm25_function)

# 创建集合
client.create_collection(
    collection_name="fulltext_search",
    schema=schema
)

from pymilvus import AnnSearchRequest, RRFRanker

# 密集向量搜索请求
dense_req = AnnSearchRequest(
    data=[dense_query_vector],   # 密集向量
    anns_field="dense_vector",   # 密集向量字段
    param={"metric_type": "COSINE", "params": {"nprobe": 10}},
    limit=10
)

# 稀疏向量搜索请求（BM25）
sparse_req = AnnSearchRequest(
    data=[sparse_query_vector],  # 稀疏向量
    anns_field="sparse_vector",  # 稀疏向量字段
    param={"metric_type": "IP"},
    limit=10
)

# 混合搜索（使用 RRF 融合排名）
results = client.hybrid_search(
    collection_name="hybrid_collection",
    reqs=[dense_req, sparse_req],     # 两个搜索请求
    ranker=RRFRanker(),               # Reciprocal Rank Fusion
    limit=5
)

# === Milvus Python SDK 速查 ===
from pymilvus import MilvusClient

# 连接
client = MilvusClient("./demo.db")                    # Lite 模式
client = MilvusClient(uri="http://localhost:19530")    # 远程服务器

# 集合操作
client.create_collection("name", dimension=384)
client.list_collections()
client.describe_collection("name")
client.drop_collection("name")

# 数据操作
client.insert("name", data=[{...}])
client.upsert("name", data=[{...}])
client.delete("name", ids=[1, 2, 3])
client.get("name", ids=[1])

# 搜索
client.search("name", data=[[...]], limit=5)
client.search("name", data=[[...]], filter="year > 2023", limit=5)

# 查询（标量过滤，不用向量）
client.query("name", filter="year > 2023", output_fields=["text"])