麻瓜教程

Appearance

向量搜索与查询 ​

基本向量搜索 ​

单向量搜索 ​

python
from pymilvus import Collection

# 获取集合并加载
collection = Collection("article_search")
collection.load()

# 准备查询向量（必须与集合维度一致）
query_vector = [0.1, 0.2, 0.3, ..., 0.128]  # 128维向量

# 执行搜索
results = collection.search(
    data=[query_vector],           # 查询向量列表
    anns_field="article_vector",   # 向量字段名
    param={"metric_type": "L2", "params": {"nprobe": 10}},
    limit=10                       # 返回最相似的10个结果
)

# 处理结果
for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 距离: {hit.distance}")

批量向量搜索 ​

python
# 准备多个查询向量
query_vectors = [
    [0.1, 0.2, 0.3, ..., 0.128],
    [0.5, 0.6, 0.7, ..., 0.128],
    [0.9, 0.8, 0.7, ..., 0.128]
]

# 批量搜索
results = collection.search(
    data=query_vectors,
    anns_field="article_vector",
    param={"metric_type": "L2", "params": {"nprobe": 10}},
    limit=5
)

# 处理批量结果
for i, hits in enumerate(results):
    print(f"\n查询 {i+1} 的结果:")
    for hit in hits:
        print(f"  ID: {hit.id}, 距离: {hit.distance:.4f}")

搜索参数详解 ​

基本搜索参数 ​

python
search_params = {
    "metric_type": "L2",          # 距离度量类型: L2, IP, COSINE
    "params": {
        "nprobe": 10              # IVF 索引的搜索参数
    }
}

不同索引的搜索参数 ​

IVF 系列索引 ​

python
search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 16              # 搜索的聚类数，范围: [1, nlist]
    }
}

HNSW 索引 ​

python
search_params = {
    "metric_type": "L2",
    "params": {
        "ef": 64                  # 搜索范围，必须 >= limit
    }
}

ANNOY 索引 ​

python
search_params = {
    "metric_type": "L2",
    "params": {
        "search_k": -1            # 搜索节点数，-1 表示 n_trees × limit
    }
}

过滤搜索 ​

标量过滤 ​

python
# 按分类过滤
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    expr='category == "技术"'       # 过滤表达式
)

# 多条件过滤
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    expr='category == "技术" and read_count > 1000'
)

# 范围过滤
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    expr='publish_time >= 1704067200 and publish_time <= 1706745600'
)

复杂过滤条件 ​

python
# IN 操作符
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    expr='category in ["技术", "科学", "互联网"]'
)

# 字符串匹配
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    expr='title like "%Milvus%"'   # 模糊匹配
)

# JSON 字段过滤
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    expr='metadata["author"] == "张三"'
)

返回指定字段 ​

指定输出字段 ​

python
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    output_fields=["title", "category", "read_count"]  # 返回的字段
)

# 处理带字段的结果
for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}")
        print(f"距离: {hit.distance}")
        print(f"标题: {hit.entity.get('title')}")
        print(f"分类: {hit.entity.get('category')}")
        print(f"阅读量: {hit.entity.get('read_count')}")
        print("---")

分区搜索 ​

指定分区搜索 ​

python
# 搜索单个分区
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    partition_names=["tech_articles"]  # 指定分区
)

# 搜索多个分区
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    partition_names=["tech_articles", "science_articles"]
)

搜索结果处理 ​

结果结构 ​

python
results = collection.search(...)

for hits in results:              # 每个查询向量的结果
    print(f"查询结果数: {len(hits)}")
    
    for hit in hits:              # 每个相似向量
        print(f"  ID: {hit.id}")           # 实体 ID
        print(f"  距离: {hit.distance}")   # 相似度距离
        print(f"  分数: {hit.score}")      # 相似度分数

结果排序 ​

python
# 按距离排序（默认已排序）
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10
)

# 结果已经是按距离升序排列（L2 距离越小越相似）
for hit in results[0]:
    print(f"ID: {hit.id}, 距离: {hit.distance}")

结果过滤 ​

python
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=100,                     # 先获取更多结果
    output_fields=["score"]
)

# 后处理过滤
filtered_results = []
for hit in results[0]:
    if hit.distance < 0.5:         # 只保留距离小于 0.5 的结果
        filtered_results.append(hit)

print(f"过滤后结果数: {len(filtered_results)}")

高级搜索技巧 ​

多向量字段搜索 ​

python
# 创建包含多个向量字段的集合
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text_vector", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="image_vector", dtype=DataType.FLOAT_VECTOR, dim=512)
]

# 分别在不同向量字段上搜索
text_results = collection.search(
    data=[text_query_vector],
    anns_field="text_vector",
    param=search_params,
    limit=10
)

image_results = collection.search(
    data=[image_query_vector],
    anns_field="image_vector",
    param=search_params,
    limit=10
)

混合搜索（RRF 融合） ​

python
def reciprocal_rank_fusion(text_results, image_results, k=60):
    """使用 RRF 融合多路搜索结果"""
    scores = {}
    
    # 处理文本搜索结果
    for rank, hit in enumerate(text_results[0]):
        scores[hit.id] = scores.get(hit.id, 0) + 1 / (k + rank + 1)
    
    # 处理图像搜索结果
    for rank, hit in enumerate(image_results[0]):
        scores[hit.id] = scores.get(hit.id, 0) + 1 / (k + rank + 1)
    
    # 按分数排序
    sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

# 融合结果
fused_results = reciprocal_rank_fusion(text_results, image_results)

分页搜索 ​

python
def paginated_search(collection, query_vector, offset=0, page_size=10):
    """分页搜索"""
    results = collection.search(
        data=[query_vector],
        anns_field="article_vector",
        param=search_params,
        limit=offset + page_size,      # 获取足够多的结果
        output_fields=["title"]
    )
    
    # 返回指定范围的结果
    hits = results[0]
    if offset < len(hits):
        return hits[offset:offset + page_size]
    return []

# 使用示例
page_1 = paginated_search(collection, query_vector, offset=0, page_size=10)
page_2 = paginated_search(collection, query_vector, offset=10, page_size=10)

搜索性能优化 ​

使用合适的索引参数 ​

python
# 平衡速度和精度
search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 32              # 适当增加 nprobe 提高召回率
    }
}

批量搜索 ​

python
# 一次性搜索多个向量，减少网络开销
batch_size = 100
all_results = []

for i in range(0, len(query_vectors), batch_size):
    batch = query_vectors[i:i + batch_size]
    results = collection.search(
        data=batch,
        anns_field="article_vector",
        param=search_params,
        limit=10
    )
    all_results.extend(results)

使用分区减少搜索范围 ​

python
# 只在相关分区中搜索
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param=search_params,
    limit=10,
    partition_names=["recent_articles"]  # 只搜索最近的分区
)

完整搜索示例 ​

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

def search_demo():
    """搜索功能完整示例"""
    
    # 连接 Milvus
    connections.connect(host="localhost", port="19530")
    
    # 创建测试集合
    fields = [
        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128),
        FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=256),
        FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64),
        FieldSchema(name="score", dtype=DataType.FLOAT)
    ]
    
    from pymilvus import utility
    
    # 清理旧集合
    if utility.has_collection("search_demo"):
        utility.drop_collection("search_demo")
    
    schema = CollectionSchema(fields, "搜索演示集合")
    collection = Collection("search_demo", schema)
    
    # 插入测试数据
    print("插入测试数据...")
    categories = ["科技", "生活", "娱乐", "教育"]
    data = []
    for i in range(1000):
        data.append({
            "vector": [random.random() for _ in range(128)],
            "title": f"文章标题_{i}",
            "category": random.choice(categories),
            "score": random.uniform(0, 100)
        })
    
    collection.insert(data)
    
    # 创建索引
    print("创建索引...")
    index_params = {
        "index_type": "IVF_FLAT",
        "metric_type": "L2",
        "params": {"nlist": 128}
    }
    collection.create_index("vector", index_params)
    utility.wait_for_index_building_complete("search_demo")
    
    # 加载集合并搜索
    collection.load()
    
    # 生成查询向量
    query_vector = [random.random() for _ in range(128)]
    
    print("\n=== 基础搜索 ===")
    results = collection.search(
        data=[query_vector],
        anns_field="vector",
        param={"metric_type": "L2", "params": {"nprobe": 16}},
        limit=5,
        output_fields=["title", "category"]
    )
    
    for hit in results[0]:
        print(f"ID: {hit.id}, 距离: {hit.distance:.4f}, "
              f"标题: {hit.entity.get('title')}, "
              f"分类: {hit.entity.get('category')}")
    
    print("\n=== 过滤搜索 ===")
    results = collection.search(
        data=[query_vector],
        anns_field="vector",
        param={"metric_type": "L2", "params": {"nprobe": 16}},
        limit=5,
        expr='category == "科技" and score > 50',
        output_fields=["title", "category", "score"]
    )
    
    for hit in results[0]:
        print(f"ID: {hit.id}, 距离: {hit.distance:.4f}, "
              f"分类: {hit.entity.get('category')}, "
              f"评分: {hit.entity.get('score'):.2f}")
    
    # 清理
    utility.drop_collection("search_demo")
    print("\n演示完成！")

if __name__ == "__main__":
    search_demo()

常见问题 ​

搜索结果为空 ​

python
# 检查集合是否已加载
if not collection.is_loaded:
    collection.load()

# 检查是否有数据
print(f"集合数据量: {collection.num_entities}")

# 检查索引是否存在
print(f"索引: {collection.indexes}")

召回率过低 ​

python
# 增加搜索参数
search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 128,            # 增加搜索范围
        "ef": 256                 # HNSW 索引参数
    }
}

搜索速度慢 ​

python
# 减少搜索范围
results = collection.search(
    data=[query_vector],
    anns_field="article_vector",
    param={"metric_type": "L2", "params": {"nprobe": 8}},  # 减小 nprobe
    limit=10,                                                  # 减少返回数量
    partition_names=["recent"]                                 # 限定分区
)

下一步 ​

掌握向量搜索后，你可以：

  1. 学习分区与别名管理
  2. 了解高级搜索技巧
  3. 探索实际应用案例

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1