Appearance
七、检索
7.1 检索方式概览
Milvus支持多种检索方式：对向量字段支持精确检索（KNN）和近似近邻检索（ANN）；对标量字段提供条件过滤，主要配合向量检索进行结果筛选。
向量检索的关键参数：
| 参数 | 说明 |
|---|---|
| 查询向量 (data) | 待检索对象的向量表示，维度需与集合中向量字段一致 |
| 向量字段 (anns_field) | 指定参与检索的向量字段名称 |
| TopK (limit) | 返回相似度最高的结果数量 |
| metric_type | 相似度计算方式，如COSINE、L2或IP |
7.2 稠密向量检索
python
def dense_vector_search_example(client, query: str, limit: int = 5):
model = get_bge_m3_model()
dense_vec, _ = encode_query(model, query)
results = client.search(
collection_name=COLLECTION_NAME,
data=[dense_vec],
anns_field="vector",
limit=limit,
search_params={"metric_type": "COSINE"},
output_fields=["id", "text", "metadata"]
)
return results[0]7.3 稀疏向量检索
python
def sparse_vector_search_example(client, query: str, limit: int = 5):
model = get_bge_m3_model()
_, sparse_vec = encode_query(model, query)
results = client.search(
collection_name=COLLECTION_NAME,
data=[sparse_vec],
anns_field="sparse_vector",
limit=limit,
search_params={"metric_type": "IP"},
output_fields=["id", "text", "metadata"],
)
return results[0]7.4 混合检索与重排序
混合检索将稠密向量（语义匹配）和稀疏向量（关键词匹配）结合，并通过重排序器（Reranker） 对两路召回结果进行融合排序。
重排序（Reranker） 是指在初步检索完成后，对候选结果进行二次排序优化。其目标不是扩大召回范围，而是在已有候选集内提升排序质量和结果相关性。
RRFRanker 的工作流程：
python
def hybrid_vector_search_example_rrf(client, query: str, limit: int = 5):
from pymilvus import AnnSearchRequest, RRFRanker
model = get_bge_m3_model()
dense_vec, sparse_vec = encode_query(model, query)
dense_req = AnnSearchRequest(
data=[dense_vec], anns_field="vector",
param={"metric_type": "COSINE"}, limit=limit,
)
sparse_req = AnnSearchRequest(
data=[sparse_vec], anns_field="sparse_vector",
param={"metric_type": "IP"}, limit=limit,
)
results = client.hybrid_search(
collection_name=COLLECTION_NAME,
reqs=[dense_req, sparse_req],
ranker=RRFRanker(k=60),
limit=limit,
output_fields=["id", "text", "metadata"],
)
return results[0]7.5 标量检索
标量检索不涉及向量相似度计算，仅基于标量字段条件进行筛选：
python
def scalar_query_examples(client, keyword: str = "大模型"):
# 对text字段进行模糊检索
like_res = client.query(
collection_name=COLLECTION_NAME,
filter=f'text like "%{keyword}%"',
output_fields=["id", "text"],
limit=5,
)
# 对metadata的JSON字段进行检索
json_res = client.query(
collection_name=COLLECTION_NAME,
filter='metadata["source"] like "%sample%"',
output_fields=["id", "metadata"],
limit=5,
)7.6 完整案例
python
import os
from pymilvus import MilvusClient, AnnSearchRequest, RRFRanker
from FlagEmbedding import BGEM3FlagModel
# ==========================================
# 1. 全局配置与客户端初始化
# ==========================================
COLLECTION_NAME = "demo_collection"
# 确保这里和上一节插入数据时使用的模型路径一致
MODEL_PATH = r"D:\ai_models\huggingface_cache\bge-m3"
MILVUS_URI = "http://localhost:19530"
print("-> 正在连接 Milvus 客户端...")
client = MilvusClient(uri=MILVUS_URI, token="")
print(f"-> 正在加载 BGE-M3 模型 (用于处理查询问题)...")
model = BGEM3FlagModel(MODEL_PATH)
# ==========================================
# 2. 辅助函数：将用户提问转换为向量
# ==========================================
def encode_query(query_text: str):
"""使用 BGE-M3 模型同时生成查询的稠密向量和稀疏向量"""
vectors = model.encode(
[query_text],
return_dense=True,
return_sparse=True
)
# 提取列表中的第一个（也是唯一一个）元素的向量
dense_vec = vectors["dense_vecs"][0]
sparse_vec = vectors["lexical_weights"][0]
return dense_vec, sparse_vec
# ==========================================
# 3. 混合检索与重排序 (Hybrid Search + RRF)
# ==========================================
def hybrid_vector_search_example_rrf(query: str, limit: int = 3):
print(f"\n[{' 混合检索测试 ':=^40}]")
print(f"用户提问: {query}")
# 1. 对查询语句进行向量化
dense_vec, sparse_vec = encode_query(query)
# 2. 构建稠密向量检索请求 (语义检索)
# 注意：这里的 metric_type 必须与你创建索引时保持一致 (你之前代码里建的是 COSINE)
dense_req = AnnSearchRequest(
data=[dense_vec],
anns_field="vector",
param={"metric_type": "COSINE"},
limit=limit,
)
# 3. 构建稀疏向量检索请求 (关键词检索)
# 稀疏向量的度量方式通常使用 IP (内积)
sparse_req = AnnSearchRequest(
data=[sparse_vec],
anns_field="sparse_vector",
param={"metric_type": "IP"},
limit=limit,
)
# 4. 执行混合检索，并使用 RRFRanker 重新排序
results = client.hybrid_search(
collection_name=COLLECTION_NAME,
reqs=[dense_req, sparse_req],
ranker=RRFRanker(k=60), # k 决定了排名权重的衰减速度，一般设为 60
limit=limit,
output_fields=["id", "text", "metadata"], # 告诉 Milvus 返回哪些原始字段
)
# 5. 打印结果
print(f"\n--- 混合检索找到了 {len(results[0])} 条相关片段 ---")
for i, hit in enumerate(results[0]):
print(f"\nTop {i+1} (匹配分数/RRF Score: {hit['distance']:.4f})")
print(f"ID: {hit['id']}")
print(f"来源文件: {hit['entity'].get('metadata', {}).get('source', '未知')}")
print(f"文本内容: {hit['entity'].get('text', '')}")
# ==========================================
# 4. 标量检索 (精准过滤)
# ==========================================
def scalar_query_examples(keyword: str):
print(f"\n[{' 标量检索测试 ':=^40}]")
print(f"正在全文检索包含关键词 '{keyword}' 的片段...")
# 对 text 字段进行模糊匹配 (类似 SQL 的 LIKE)
like_res = client.query(
collection_name=COLLECTION_NAME,
filter=f'text like "%{keyword}%"',
output_fields=["id", "text", "metadata"],
limit=3,
)
print(f"\n--- 标量检索找到了 {len(like_res)} 条结果 ---")
for i, res in enumerate(like_res):
print(f"\n结果 {i+1} (ID: {res['id']})")
print(f"文本内容: {res['text']}")
# ==========================================
# 主运行入口
# ==========================================
if __name__ == "__main__":
try:
# 你可以根据你导入的 sample.docx 的实际内容，修改下面的提问词
test_query = "什么是大语言模型？"
# 执行混合检索测试
hybrid_vector_search_example_rrf(query=test_query, limit=3)
# 执行标量检索测试
test_keyword = "模型"
scalar_query_examples(keyword=test_keyword)
except Exception as e:
print(f"\n 检索过程中发生错误: {e}")检索到了相关文档片段，最后一步就是把它们交给大模型，生成最终答案。