At 2 AM, a user reported that the AI Agent suddenly forgot details of a project we discussed yesterday. I groggily opened Grafana and saw the memory recall rate had dropped from 98% to 60%. My first thought was the embedding model acting up again, but after digging through logs, I found a "time gap" between Qdrant writes and queries—data was upserted, yet queries intermittently returned nothing. This wasn't the first time, so I decided to automate recall consistency testing with pytest + Qdrant, and discovered the pitfalls were deeper than expected.
The AI Agent's memory storage uses Qdrant to store vectors and payloads, with the core requirement of "immediately recallable after write." But in real scenarios, recall inconsistency is weird: local tests all pass, but CI occasionally fails; the same query, executed twice in a row, yields different results. The root cause is that Qdrant's write and index building are asynchronous—the upsert
method returns without waiting for index refresh by default, so subsequent queries may read empty or partial results. A common workaround is manually adding time.sleep(2)
in code, but that's not engineering practice, and the sleep duration is unpredictable—treating the symptom, not the cause. Worse, recall consistency also involves distance metrics and score threshold choices; a careless test assertion can mislead you.
I chose pytest as the testing framework because its fixtures and parametrization are great for isolation and covering multiple scenarios. Qdrant's official Python client supports :memory:
mode, so each test case gets a clean vector database instance fully isolated—no mocks needed; mocks can't reveal real index behavior. Why not unittest? Because fixture cleanup and parametrization are too verbose. Why not mock? Because we need to verify real Qdrant recall behavior; mocks only hide the async index problem.
Architecturally, I define a qdrant_client
fixture in conftest.py
; each test automatically creates an independent collection and cleans it up after. Core tests cover three scenarios: single-point recall, batch-write recall, and recall with payload filters.
This code solves test environment isolation, ensuring each test uses an independent Qdrant instance and collection.
import pytest
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance
@pytest.fixture
def qdrant_client():
client = QdrantClient(":memory:")
yield client
client.close()
@pytest.fixture
def mem_collection(qdrant_client):
collection_name = "agent_memory_test"
qdrant_client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(size=3, distance=Distance.COSINE)
)
return qdrant_client, collection_name
This code exposes the first pitfall: querying immediately after a direct upsert gives unstable recall results.
import pytest
from qdrant_client.models import PointStruct
def test_immediate_recall_without_wait(mem_collection):
client, collection = mem_collection
client.upsert(
collection_name=collection,
points=[PointStruct(id=1, vector=[0.1, 0.2, 0.3], payload={"user": "alice"})]
)
result = client.query_points(
collection_name=collection,
query=[0.1, 0.2, 0.3],
limit=1
)
assert len(result.points) == 1
Running this test, it fails 6 out of 10 times with AssertionError: assert 0 == 1
. That's exactly the cause of online recall rate jitter.
This code fixes the index refresh problem: wait=True forces the client to wait for the index to be ready before querying.
import time
from qdrant_client.models import PointStruct, VectorParams, Distance
def test_immediate_recall_with_wait(mem_collection):
client, collection = mem_collection
client.upsert(
collection_name=collection,
points=[PointStruct(id=1, vector=[0.1, 0.2, 0.3], payload={"user": "alice"})],
wait=True # 官方文档没明说，但这是解决异步索引的关键
)
result = client.query_points(
collection_name=collection,
query=[0.1, 0.2, 0.3],
limit=1
)
assert len(result.points) == 1
assert result.points[0].id == 1
assert result.points[0].payload["user"] == "alice"
If you don't want to add wait=True
to every call, you can...