凌晨两点,一位用户反馈AI Agent突然忘记了昨天讨论的项目细节。我睡眼惺忪地打开Grafana,发现记忆召回率从98%掉到了60%。第一反应是embedding模型又抽风了,但翻完日志后,问题指向Qdrant:写入与查询之间存在“时间缝隙”——数据明明upsert成功,查询却间歇性返回空结果。这已经不是第一次发生,我决定用pytest和Qdrant把召回一致性测试自动化,最后发现坑比预想中深。

先把问题拆开看。AI Agent的记忆存储依赖Qdrant保存向量和payload,核心要求是“写完立即可查”。但真实环境里的表现很怪:本地测试全绿,CI偶尔失败;同一个查询连续执行两次,结果不一致。

根因在于Qdrant的写入和索引构建是异步的。upsert方法默认不等索引刷新就返回,后续查询可能读到空或部分结果。常见的补丁是代码里手动sleep(2),但这不是工程化做法——sleep多久不可控,治标不治本。更麻烦的是,召回一致性还和距离度量、score阈值选择相关,测试断言写得不够严谨,反而会误导排查方向。

测试框架我选pytest。fixture和参数化很适合做隔离和覆盖多场景;Qdrant官方Python客户端支持:memory:模式,每个测试用例都能拿到完全隔离的向量数据库实例,不需要mock。为什么不用unittest?fixture清理和参数化写起来太啰嗦。为什么不用mock?因为我们就是要验证Qdrant真实的召回行为,mock只会掩盖异步索引问题。

架构上,只需要在conftest.py里定义qdrant_client fixture:每个测试自动创建临时数据库,结束后自动销毁。针对写后一致性,最核心的单测是“插入一条向量后立刻查询”,断言返回结果;再搭配“连续查询多次结果稳定”的重复测试,把间歇性问题暴露出来。

实践下来,这套自动化测试至少提前发现了三类问题:索引刷新延迟导致的空结果、距离阈值设置过严导致的漏召回、以及复用客户端状态导致的脏数据串扰。这三类问题如果只在线上人工观察,往往要等到用户投诉才会暴露。

更重要的收获是:不要用sleep去赌异步索引的完成时机。可靠做法是测试里轮询ready状态或调用强制刷新接口,并对查询结果做“至少一次命中”的断言。真正的企业级可靠性,不是靠侥幸,而是靠可重复的自动化测试守住的。