Milvus Dashboard 查询结果不一致问题排查与解决

2025-03-20 13:24:05

最近在使用 Milvus Dashboard 时遇到了一个怪事：查询结果居然会因为选择不同的输出字段而改变！简单来说，当我勾选输出embeddings_vector字段时，得到的结果不对(bad)；不勾选这个字段时，结果又是正常的(good)。这让整个RAG项目效果也变差了。

问题根源剖析

Milvus 作为向量数据库，核心功能是存储和检索向量。查询结果的改变通常与以下几个方面有关：

数据一致性问题: 写入Milvus的数据可能存在不一致，比如写入了脏数据，或者更新了数据却没有重建索引。
索引问题: 索引可能损坏或未及时更新，导致查询结果偏差。
查询参数问题: 虽然问题里说了主要区别是输出字段不同，但也不排除其他查询参数（例如 similarity_top_k）或者搜索请求本身（prompt）悄悄改变了的可能性，要仔细检查下。
Bug: 不排除 Milvus Dashboard 或 pymilvus 客户端存在 Bug 的可能性。
版本兼容性: pymilvus、Milvus Server 以及可能存在的第三方库（如 llama_index）之间版本不兼容。
LlamaIndex 内部机制: llama_index 在处理查询结果时可能有特别的处理逻辑，输出字段的变化可能影响其处理方式。
Milvus 自身的行为改变： 在某个版本迭代中，Milvus 对包含/不包含向量字段的查询处理方式，可能发生了改变。

解决方法

针对以上可能的原因，逐一排查，下面给出相应的解决方案：

1. 检查数据一致性

首先，我们要排除掉脏数据，或者重复数据的影响。

查看原始数据: 用脚本直接从Milvus里取出几条数据，肉眼或者编写脚本检查下数据是否有重复、缺失或者明显错误的情况。
代码示例 (Python):

from pymilvus import connections, Collection

connections.connect(uri="your_milvus_uri", token="your_milvus_api_key")
collection = Collection("your_collection_name")
collection.load()  # 先加载集合

# 获取前 10 条数据
entities = collection.query(expr="chunk_id > 0", limit=10, output_fields=["chunk_id", "chunk_text", "embeddings_vector"])

for entity in entities:
    print(entity)

collection.release()
connections.disconnect("default") #断开默认连接

重复数据处理： 假设 chunk_id 应该是唯一的。我们可以用 pymilvus 查询是否有重复的 chunk_id。

from pymilvus import connections, Collection, utility

connections.connect(uri="your_milvus_uri", token="your_milvus_api_key")
collection = Collection("your_collection_name")
collection.load()

# 查找重复的 chunk_id
has_duplicates = utility.has_duplicate_field_values(collection.name,"chunk_id")

if has_duplicates:
   print("存在重复的 chunk_id! 需要删除重复数据并重建索引.")
   # 根据实际情况删除重复数据, 以下只作为示意，不可直接用于生产环境！
   # ！！！！！在运行任何删除操作前先做备份！！！！！！
   duplicate_ids = [] # 此处编写代码找出所有重复的ID
   for dup_id in duplicate_ids:
        collection.delete(f"chunk_id == '{dup_id}'") # 按ID 删除
   collection.flush() # 重要: 执行删除后一定要 flush！
else:
   print("未发现重复的 chunk_id.")

collection.release()
connections.disconnect("default")

2. 检查和重建索引

确认数据没问题之后, 看看索引。如果数据没问题，那么尝试重建索引。

查看索引状态: 用 Milvus Dashboard 查看, 或者用 Python 脚本。

from pymilvus import connections, Collection

connections.connect(uri="your_milvus_uri", token="your_milvus_api_key")
collection = Collection("your_collection_name")

print(collection.indexes)

connections.disconnect("default")

重建索引 (务必谨慎):
- 先删除原有索引，再重新创建。这通常能解决很多奇怪的索引问题。
- 强烈建议：在执行删除索引和创建新索引操作前，务必对Milvus数据进行备份！

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

connections.connect(uri="your_milvus_uri", token="your_milvus_api_key")
collection = Collection("your_collection_name")

# 1. 删除现有索引 (谨慎操作!)
for index in collection.indexes:
     utility.drop_index(collection.name, index.index_name) #按索引名称删除
print("已删除所有旧索引。")

# 2. (可选) 重新定义 collection schema，如果你的 schema 没有变，这步可以跳过。
# 以下代码仅作为示例, 你需要根据你的实际情况定义 FieldSchema
fields = [
    FieldSchema(name="chunk_id", dtype=DataType.VARCHAR, is_primary=True, max_length=200),
    FieldSchema(name="chunk_text", dtype=DataType.VARCHAR, max_length=65535), #根据文本长度进行调整
    FieldSchema(name="embeddings_vector", dtype=DataType.FLOAT_VECTOR, dim=1536) # 维度需要根据你使用的 embedding 模型进行调整。
]
schema = CollectionSchema(fields, description="Your collection description")
collection = Collection("your_collection_name", schema) # 用新的schema 创建
print(f"集合 '{collection.name}' (可能)已使用新的 Schema 创建.")

# 3. 重新创建索引
#  索引参数 (index_params) 需要根据你实际使用的索引类型和参数调整
index_params = {
    "index_type": "IVF_FLAT",  # 示例, 例如: IVF_FLAT, HNSW, etc.
    "metric_type": "IP",  # 示例, 例如: IP, L2, etc.
    "params": {"nlist": 1024}  # 示例, 根据 index_type 调整.
}
collection.create_index("embeddings_vector", index_params)
print("已创建新索引.")
collection.load()
print("collection 已加载")

connections.disconnect("default")

3. 核对查询参数及Prompt

虽然问题表面上和输出字段有关，但也要细心排查下是不是别的地方也动了。
把之前的代码和现在的仔细对比下, 特别注意：

similarity_top_k 的值是不是一致。
查询时用的 prompt 有没有变化。

4. Milvus Dashboard 和 pymilvus 版本

升级 pymilvus: 试试升级到最新版本的 pymilvus，看看问题是否解决。
```
pip install -U pymilvus
```
考虑Milvus版本： 如果方便，看看Milvus server 版本是不是有更新，可以尝试更新Milvus。更新Milvus之前做好备份。

5. llama_index 版本和配置

llama_index 在这中间也起到作用，检查下版本。

升级 llama_index:

```bash
pip install -U llama-index
```

llama_index 配置: 仔细阅读llama_index 关于 MilvusVectorStore 的文档, 看有没有什么特别的配置项会影响到结果. 排除 llama_index 这一层是否会有影响。尤其是注意from_vector_store这个函数的使用。
直接使用 pymilvus 查询: 为了排除 llama_index 的影响, 我们可以直接用 pymilvus 进行查询，看看结果是否一致.

from pymilvus import connections, Collection

connections.connect(uri="your_milvus_uri", token="your_milvus_api_key")
collection = Collection("your_collection_name")
collection.load()

# 使用相同的搜索向量和参数
search_vector = [...]  #  这里替换成你的搜索向量，跟你传给 llama_index 的一样.
search_params = {
    "metric_type": "IP",
    "params": {"nprobe": 16},  # 根据实际情况调整
}

# 1. 查询时不包含 embeddings_vector
results_without_vector = collection.search([search_vector], "embeddings_vector", search_params, limit=5, output_fields=["chunk_id", "chunk_text"])

# 2. 查询时包含 embeddings_vector
results_with_vector = collection.search([search_vector], "embeddings_vector", search_params, limit=5, output_fields=["chunk_id", "chunk_text", "embeddings_vector"])

print("不包含 embeddings_vector 的结果:", results_without_vector)
print("包含 embeddings_vector 的结果:", results_with_vector)

collection.release()
connections.disconnect("default")

对比这两种查询的结果，如果 pymilvus 的结果也是不一致的，那就说明问题出在 Milvus 层面，跟 llama_index 无关。

6. 深入排查 (Milvus行为改变)

如果排除了上面所有的可能原因，仍未找到答案，有一个可能性，就是 Milvus Server 自身可能在处理是否返回向量数据时逻辑上做了改变。这通常是 Milvus 本身的内部机制优化导致的。

查阅文档和Release Note： 回顾你用的 Milvus 版本的更新说明，仔细寻找与查询相关的更新。看是否有提到相关的改动。
尝试其他搜索方法： Milvus 有时会有不同的 Search API (比如 hybrid search, 或者不同的 search interface)，尝试不同的搜索方式看看能否绕过这个问题。

7.临时规避方案 (Workaround)

如果在debug过程中实在找不到问题，或者需要立即恢复功能。可以暂时使用如下方案：

既然已知不返回 embeddings_vector时，结果是正确的。那就可以先不返回这个字段，然后在得到 chunk_id后，根据chunk_id 再单独去查一次，把 embeddings_vector 取回来。虽然多了查询次数,但起码能拿到正确结果。

# 使用 query_engine 获取初始结果, 不包含 embeddings_vector
# ... (你的原始代码, 但 output_fields 不包含 'embeddings_vector') ...

# 获取到的 chunk_ids (假设结果存储在 results 变量里)
retrieved_chunk_ids = [result.node.node_id for result in rag_result.source_nodes] # 取决于llama_index 返回结果的数据结构

from pymilvus import connections, Collection

connections.connect(uri="your_milvus_uri", token="your_milvus_api_key")
collection = Collection("your_collection_name")
collection.load()

# 根据 chunk_ids 再次查询, 获取 embeddings_vector
if retrieved_chunk_ids: # 确保非空列表

    # 构建表达式字符串。因为ID可能有多个，使用'in' 操作符进行查询。
    expr = f"chunk_id in [{','.join([f'\'{id}\'' for id in retrieved_chunk_ids])}]"

    final_results = collection.query(expr= expr , output_fields=["chunk_id", "chunk_text", "embeddings_vector"])
    print(final_results)
    #然后根据实际的数据结构, 把final_result 里对应的向量信息补充回你的 rag_result
else:
    print("没有检索到任何 chunk_id，请检查之前的查询。")

collection.release()
connections.disconnect("default")