At 2 AM, my QA colleague frantically @-ed me in the group chat: “The chatbot’s long-term memory is completely broken. A user mentioned ‘my cat is named Doubao’, but days later when asked about it, the bot recalled ‘your dog is called Xuebing’. Even creepier, when you flip to page 3 you see the same content as page 1. Right now, over 2000 users’ memories are misaligned — take a look.”
Bleary‑eyed, I opened the monitoring dashboards. The tables housing vectors and metadata showed normal write QPS, APIs were returning 200, and there were zero errors. In that moment I knew — this wasn’t a simple code bug. We had shipped a long‑term memory system without any consistency regression testing. Our trust in that “memory storage” black box was running naked from day one.
Breaking down the problem
LLM long‑term memory stores (like Mem0, LangChain’s memory modules, or a home‑grown vector index) typically rely on a “memory pipeline”:
- The user converses with the model and facts (memory facts) are created.
- Each fact is embedded and stored in a vector database (pgvector, Milvus, etc.), together with structured fields such as timestamp, user ID, and session ID.
- During future conversations, the top‑N memories are recalled by similarity plus time‑decay, then stitched together into pages for the model.
The third step is where things went wrong. To make every retrieval reproducible and pagination coherent, you must simultaneously satisfy sort determinism and no‑duplicate‑no‑missing pagination. In practice, many teams just throw an ORDER BY created_at with LIMIT/OFFSET into production. Under concurrent writes, it falls apart immediately — insert order within the same timestamp is non‑deterministic, so OFFSET can skip or repeat records. Additionally, some services cache metadata for performance; if the cache isn’t invalidated promptly, you get phantom reads where a just‑inserted memory “disappears” on query.
Traditional CRUD backend testing falls flat here. You can’t simulate 2000 concurrent writes, page drifting, and cache anomalies with a few hand‑crafted timestamps. You need an automated consistency test suite — one that kneads your memory system like dough until it submits.
Designing the solution
The goal was to build a test suite that could run against any long‑term memory backend — Postgres, Milvus, even an in‑memory KV store. The core idea: generate a large volume of memories that mimic real distributions → batch‑insert them concurrently → fetch via different pagination/sort strategies → verify mathematical invariants.
Tech stack: Python + pytest, memory generation with Faker, storage abstracted behind an interface. Why not just validate with the storage’s own SQL scripts? Because we need to test the final “memory view” that is handed to the LLM — the API’s paginated results — not the raw physical rows in the database. Querying the DB directly would test the wrong thing. Why not Go/JMeter for stress testing? We need logical assertions (e.g., the union of all pages yields the same set), not throughput benchmarks. pytest’s parametrized fixtures and Allure reporting offer at least an order of magnitude higher development speed for this kind of scenario.
The architecture is dead simple: define a MemoryStore protocol with two methods — insert(memories) and recall(user_id, page, size, sort_order). Write a set of pytest test cases around it, then inject the concrete database implementation via dependency injection. Even if you swap Postgres for a custom vector engine, the test suite stays untouched.
Core implementation
Why does this code exist?
First, define the memory data model and the storage abstraction interface. This decouples the test logic from any specific backend. When you change databases, you only need to implement two methods and the entire suite runs immediately.
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Protocol, List, Optional
from uuid import uuid4
@dataclass
class Memory:
"""A single memory entity"""
memory_id: str = field(default_factory=lambda: uuid4().hex)
user_id: str = ""
content: str = ""
created_at: datetime = field(
default_factory=lambda: datetime.now(timezone.utc)
) # Crucial: must carry UTC timezone, otherwise sorting will bite you
class MemoryStore(Protocol):
"""Memory storage protocol – every backend implements this"""
async def insert_batch(self, memories: List[Memory]) -> None:
...
async def recall_page(
self,
user_id: str,
page: int,
page_size: int,
sort_order: str = "desc"
) -> List[Memory]:
...
Next comes the core pagination‑consistency test function — it generates a flood of memories, pumps them into the store, then reads page by page and verifies mathematical invariants:
Why does this code exist?
It simulates a real user with 2000 inserted memories, then traverses all pages with different page sizes to check for no duplicates, no missing entries, and strict sort consistency. Any breach stops immediately with reproducible failure information.
import pytest
from faker import Faker
from collections import OrderedDict
fake = Faker()
async def run_pagination_consistency(
store: MemoryStore,
user_id: str,
total: int,
page_size: int
):
# 1. Insert `total` memories, deliberately jittered timestamps to mimic concurrency
memories = [
Memory(
user_id=user_id,
content=fake.sentence(),
# Make some timestamps extremely close to trigger sort instability
created_at=datetime(
2025, 5, 1, 12, 0, tzinfo=timezone.utc
) + timedelta(seconds=i * 0.01)
)
for i in range(total)
]
# Randomly shuffle the insertion order to simulate out‑of‑order concurrent writes
random.shuffle(memories)
await store.insert_batch(memories)
# 2. Expected full set: sorted by created_at descending (newest first), with memory_id tie‑breaker
ex
Top comments (0)