多Agent 系统里，经常会出现一个单 Agent 里从来不会出现的问题：一个子 Agent 刚写完数据，另一个子 Agent 立刻去读，结果是空的。
根本问题出在 Agent 的写-读模式撞上了很多数据库为单 Agent 场景设计的默认一致性配置。
接下来，这篇文章将说清楚这个矛盾从哪来，以及怎么用一行参数解决它。
01
单 Agent 与多 Agent 的读写设计有何异同？
单 Agent RAG 的工作方式是这样的：用户提出一个问题，Agent 把问题向量化，去 Milvus 检索 Top-K 文档片段，拼成 prompt 喂给模型，模型输出答案。整条链路里，向量数据库是默认只读的——数据在应用启动时、文档更新时已经写好了，推理过程中没有人再继续往里写东西。
但多 Agent 系统里有两类角色：Writer Agent负责执行任务、调用外部工具、发现新信息，把结果 embedding 后写入 Milvus 作为共享记忆；Reader Agent收到协调信号后，从 Milvus 检索最新记忆，基于这些上下文生成下一步行动。
两者是独立的进程或线程，通过消息、回调或事件协调。Writer 写完，立刻通知 Reader，这个间隔是毫秒级的。
在这种情况下，Writer 写完、信号发出、Reader 立刻查，这种模式会导致Reader的查询动作，恰好落在“数据已写入但未对Query Node可见”的时间窗口内，最终返回空结果。
那么，这个时间窗口是怎么产生的，又要如何解决？
02
Milvus如何用四档一致性控制数据对外可见的时机
出现“写后读空”的关键，在于我们对Milvus的insert()操作存在一个认知误区——insert()返回成功，不代表数据已经可以被查询。
具体来说，Milvus 的写入流程分两段，insert()操作在第一阶段完成后就会返回“成功”，但数据此时只是被写入了消息队列（类似Kafka producer ack的语义）安全落盘，但消费者（Query Node）尚未处理，此时读取自然无法看到新数据。
如图所示，这个“写入成功到数据进入Growing Segment、查询可见”的几十毫秒到几秒的时间差，就是多Agent场景下读空问题的核心诱因。
要想解决这个问题，在Milvus中，我们可以通过guarantee_timestamp来控制数据的可见性：每次search()调用都携带上这个时间戳，Query Node执行查询前会先检查自己使用的数据版本是否追上了这个时间戳？没追上就等待，追上了再执行查询。
而我们在代码中设置的consistency_level（一致性级别），本质上就是在控制guarantee_timestamp的设定逻辑。
Milvus提供四档一致性选项，可在创建Collection时设置默认值，也可在每次search()调用时单独覆盖，不同级别对应不同的可见性、性能代价，具体如下：
这里需要重点说明：Milvus创建Collection的默认一致性级别是Bounded，这对单Agent RAG场景是完全合理的——因为单Agent场景没有推理过程中的写入操作，Bounded的5秒窗口不会被触发，既能保证检索性能，又能满足需求，是性能与体验的双赢选择。
但对于Writer写完数据后Reader立即查询的多Agent事件驱动场景，此时查询的guarantee_timestamp如果仍落在Bounded的5秒窗口内，新写入的数据就会不可见，返回空结果。
而解决这个问题的关键，就是将consistency_level从默认的Bounded，切换到适配多Agent场景的strong级别。
03
实验：Bounded 查不到，Strong 查得到
为了直观验证上述结论，我们设计了一组实验：通过模拟生产环境的高写压，让Query Node始终处于数据追赶状态，再执行“写入一条数据后立即查询”的操作，对比Bounded和Strong两种一致性级别的查询结果。
实验设计思路
通过两个机制模拟生产环境的写压，确保Query Node始终处于忙碌的追赶状态：
preload预写：提前写入大批量数据，制造WAL（Write-Ahead Log）历史积压；
storm writers后台写入：用多个后台线程持续高速写入数据，维持Query Node的追赶压力。
每轮实验中，先写入一条带唯一标记（marker）的记录，然后立即分别用Bounded和Strong级别查询该记录——一旦出现Bounded=0、Strong=1，即判定问题复现成功。
运行前提：pymilvus >= 2.6.0 已安装，Milvus 服务可访问。

!/usr/bin/env python3

import argparse
import itertools
import random
import threading
import time
import uuid
from contextlib import suppress
from pymilvus import DataType, MilvusClient
defmake_vector(seed, dim):
rng = random.Random(seed)
vec = [rng.uniform(-1.0, 1.0) for _ inrange(dim)]
norm = sum(x * x for x in vec) ** 0.5or1.0
return [x / norm for x in vec]
defmake_records(start_id, count, dim, marker, round_no):
return [
{
"id": start_id + i,
"vector": make_vector(start_id + i, dim),
"marker": marker,
"round": round_no,
}
for i inrange(count)
]
defcreate_collection(client, name, dim):
if client.has_collection(name):
client.drop_collection(name)
schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=dim)
schema.add_field("marker", DataType.VARCHAR, max_length=128)
schema.add_field("round", DataType.INT64)
index_params = client.prepare_index_params()
index_params.add_index(
field_name="vector",
index_type