DEV Community: BAOFUFAN

How We Cut AI Memory Dedup Testing from 20 Minutes to 40 Seconds and Duplication from 18% to 1.5%

BAOFUFAN — Thu, 20 Aug 2026 12:06:53 +0000

At 1 a.m., I was jolted awake by an alert: the production AI assistant's prompt tokens had blown up again. I opened the logs and, lo and behold, three long-term memories for the same user had been stuffed into the context at once — "User likes American-style coffee", "User drinks American-style coffee without sugar every day", "User prefers American-style coffee, no sugar".

This wasn't a bug — it was our memory deduplication strategy running wild.

Breaking Down the Problem

We're building a long-term memory system for an AI personal assistant. After each conversation, the model extracts important information into memories and stores them in PostgreSQL. Before the next conversation, it retrieves relevant memories and assembles them into the prompt. The problem was in two places:

Deduplication only looked at text similarity. We used Jaccard similarity + keyword hashing. Two memories would slip through as long as they were phrased differently. A user saying "I run three times a week" and "I'm used to jogging three times a week" have low text overlap, but they're semantically identical.
The decay strategy had no automated verification. The memory weight decay formula was documented, but in the actual code decay_factor was sometimes 0.9, sometimes a datetime subtraction error, and it had never been covered by tests.

The result: duplicate memories piled up, the context window got stuffed, and search results were full of semantically identical gibberish. The usual approach was to manually run a few SQL queries before each deploy to check whether nearest-neighbor queries returned the right results. But honestly, who has time to manually test every day? By the time we noticed, the duplication rate was already 18%.

Design Choices

The technology choice was a no-brainer: PostgreSQL + pgvector. The reason is practical — we already use Postgres for structured data in production. Adding Qdrant or Milvus would mean another piece of infrastructure and more operational overhead. pgvector extends the existing Postgres instance, so transactions, backups, and permissions all carry over.

For testing, we chose pytest + testcontainers. Why not mock pgvector? Because the biggest risk in a memory system is getting the vector functions wrong in SQL. Mocking would defeat the purpose. We use testcontainers to spin up a real pgvector/pgvector:pg16 container in CI, run the tests, and tear it down. Environment consistency is guaranteed.

The architecture is simple: an upsert_memory function handles deduplicated insertion, and apply_decay handles time-based weight decay. Tests cover these two core functions and run automatically on every PR.

Core Implementation

The first code block solves the test environment problem: use testcontainers to launch a pgvector container, create the table, and register the vector type. Without a clean, reproducible Postgres environment, all subsequent tests would be castles in the air.

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
import psycopg
from pgvector.psycopg import register_vector

@pytest.fixture(scope="session")
def postgres_url():
    # 使用 pgvector 官方镜像，pg16 版本
    with PostgresContainer("pgvector/pgvector:pg16") as postgres:
        postgres.with_env("POSTGRES_PASSWORD", "testpass")
        yield postgres.get_connection_url()

@pytest.fixture(scope="session")
def conn(postgres_url):
    # psycopg 3 连接
    with psycopg.connect(postgres_url) as conn:
        # 关键：注册 vector 类型，否则无法返回 embedding
        register_vector(conn)
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE memories (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    user_id TEXT NOT NULL,
                    content TEXT NOT NULL,
                    embedding vector(1536) NOT NULL,
                    importance FLOAT DEFAULT 1.0,
                    decay_rate FLOAT DEFAULT 0.05,
                    last_access_at TIMESTAMPTZ DEFAULT now(),
                    created_at TIMESTAMPTZ DEFAULT now()
                )
            """)
            # 用 IVFFlat 索引，适合大数据量近似搜索
            cur.execute("""
                CREATE INDEX ON memories 
                USING ivfflat (embedding vector_cosine_ops) 
                WITH (lists = 100)
            """)
        conn.commit()
        yield conn

Note that the register_vector step isn't emphasized in the official docs, but without it, SELECT embedding will throw unsupported type.

The second code block solves the deduplicated insertion problem: when a new memory comes in, first query the nearest vector for the same user. If the distance is below the threshold, update the old memory; otherwise, insert a new row.

# memory_dedup.py
import uuid
import numpy as np
import psycopg
from pgvector.psycopg import register_vector

def upsert_memory(conn, user_id: str, content: str, embedding: list[float],
                  threshold: float = 0.3) -> str:
    """
    去重插入记忆。pgvector 的 cosine distance = 1 - cosine similarity，
    所以距离越小越相似。threshold 0.3 等价于相似度 0.7。
    返回记忆 ID：重复返回旧 ID，否则返

Verifying AI Agent Memory Storage Consistency with Pytest and SQLite to Prevent Context Loss

BAOFUFAN — Thu, 20 Aug 2026 01:06:28 +0000

The original article body was not included in your message. Please paste the full Chinese article content so I can translate it into a natural, Dev.to-style English version while preserving code blocks.

向量数据库记忆存储踩坑实录：召回漂移让我排查了8小时

BAOFUFAN — Wed, 19 Aug 2026 12:07:10 +0000

凌晨三点，报警电话把我从床上薅起来。用户投诉“AI 怎么突然记不住我昨天说的我猫叫咪咪”。我打开日志一看，向量数据库的 top1 召回从“用户养了一只猫叫咪咪”变成了“用户喜欢喝拿铁，不加糖”。那个晚上我排查了整整 8 小时，最后发现是 embedding 模型升级导致向量空间整体漂了。那天我下定决心：记忆存储的回归测试必须自动化，不能再靠人肉兜底。

问题拆解

给 LLM 做长期记忆，常见的做法是把对话片段写入向量数据库，查询时用 embedding 召回最相关的 K 条拼进 prompt。这个链路里有两个非常阴的坑：

写入丢失：异步写入失败、集合 schema 变更、持久化 flush 失效，都会导致“你以为写进去了，其实没有”。用户明明说过猫叫咪咪，第二天你查不到。
召回漂移：embedding 模型版本升级、距离度量从 cosine 改成 L2、HNSW 索引参数调整，都会让同一个 query 的 topK 结果悄悄变化。注意，这不一定报错，只是“记错了”——比写入丢失更难排查。

常规方案为啥不行？手工抽样对比太慢，每次模型/索引改动都人肉测，根本不可持续；单元测试只测单个函数，不覆盖真实数据库行为；监控只能等用户投诉后告警，永远是事后补救。我们需要的是：任何代码改动都能在 CI 里跑一套回归测试，把召回结果和写入数量锁死。

方案设计

技术选型：pytest + Chroma。Chroma 是一个可嵌入式向量数据库，本地持久化，pip 装完就能跑，天然适合写进 CI。

为什么不选其他方案？

FAISS：太底层，没有 collection 管理和持久化，要自己封装一堆东西。
Milvus：重，要起服务，CI 里维护成本高。
pgvector：依赖 Postgres，得先在 CI 里准备数据库，配置麻烦。

Chroma 的 PersistentClient 直接落盘到临时目录，测试隔离性也好。

架构思路：维护一个“黄金基准集”（golden set），包含固定文档、固定 query 以及期望的召回 id 列表。每次测试时重建 collection，写入文档后执行召回，断言 TopK 结果与基准完全一致。写入数量、按 id 查询存在性也一并断言。这样 embedding 或索引参数有任何变化，测试立刻红。

核心实现

直接上代码，保存为 test_vector_memory_regression.py。

这段代码解决什么问题：提供一个跨环境稳定、无需安装大模型的 embedding，用来做回归测试基准。

import hashlib
import numpy as np
import pytest
import chromadb

class DeterministicEmbedding:
    """确定性 embedding：同样的文本永远生成同样的向量，避免真实模型版本差异干扰测试"""
    def __init__(self, dim=128):
        self.dim = dim

    def __call__(self, input):
        if isinstance(input, str):
            input = [input]
        vectors = []
        for text in input:
            # 用 md5 做随机种子，同一个 text 每次生成的向量完全一致
            seed = int(hashlib.md5(text.encode("utf-8")).hexdigest(), 16)
            rng = np.random.default_rng(seed)
            vec = rng.normal(0, 1, self.dim).astype(np.float32)
            # 归一化，保证 cosine 距离有效
            vec = vec / np.linalg.norm(vec)
            vectors.append(vec.tolist())
        return vectors

这段代码解决什么问题：创建独立的测试 collection，并验证写入数量正确、每一条都能按 id 取回，防止写入丢失。

@pytest.fixture
def memory_store(tmp_path):
    # tmp_path 是 pytest 提供的隔离临时目录，每个测试独立，避免 Chroma 文件锁冲突
    client = chromadb.PersistentClient(path=str(tmp_path / "chroma"))
    collection = client.create_collection(
        name="test_memory",
        embedding_function=DeterministicEmbedding(dim=128),
        metadata={"hnsw:space": "cosine"}  # 显式指定距离度量，避免默认值变更
    )
    return collection

GOLDEN_DOCS = [
    {"id": "doc_1", "document": "用户喜欢喝拿铁，不加糖"},
    {"id": "doc_2", "document": "用户在杭州工作，后端工程师"},
    {"id": "doc_3", "document": "用户最近在学 Rust"},
    {"id": "doc_4", "document": "用户养了一只猫叫咪咪"},
]

GOLDEN_QUERIES = {
    "喝咖啡习惯": {"query": "用户喝什么咖啡", "expected_ids": ["doc_1"]},
    "职业信息": {"query": "用户做什么工作", "expected_ids": ["doc_2"]},
    "宠物信息": {"query": "用户养的猫叫什么", "expected_ids": ["doc_4"]},
}

def test_write_no_loss(memory_store):
    memory_store.add(
        ids=[d["id"] for d in GOLDEN_DOCS],
        documents=[d["document"] for d in GOLDEN_DOCS]
    )
    # 第一步：数量必须等于基准文档数
    assert memory_store.count() == len(GOLDEN_DOCS)

    # 第二步：按 id 查询，逐条确认存在
    for d in GOLDEN_DOCS:
        res = memory_store.get(ids=[d["id"]])
        assert res["ids"] == [d["id"]], f"写入后按 id 查不到 {d['id']}"

这段代码解决什么问题：锁定每个 query 的 TopK 召回 id，任何 embedding 或索引参数变化导致召回顺序改变时，测试直接失败。

def test_recall_no_drift(memory_store):
    memory_store.add(
        ids=[d["id"] for d in GOLDEN_DOCS],
        documents=[d["document"] for d in GOLDEN_DOCS]
    )

    for name, case in GOLDEN_QUERIES.items():
        result = memory_store.query(
            query_texts=[case["query"]],
            n_results=1
        )
        got_ids = result["ids"][0]
        # 注意：这里比较的是 id 列表，不是距离分数。
        # 分数可以因为浮点计算有微小波动，但召回顺序绝不能变。
        assert got_ids == case["expected_ids"], (
            f"召回漂移: {name} 期望 {case['expected_ids']}，实际 {got_ids}"
        )

踩坑记录

坑1：Chroma 的 add 同时传 embeddings 和 documents 导致召回全乱

现象：我一开始为了“省事”，手动生成了 embeddings 传给 add，结果 query 出来的 id 跟预期完全对不上，debug 了半天。

原因：Chroma 会优先使用你传入的 embeddings，但我手动生成的向量和 query 时的 embedding 函数不是同一套逻辑，导致向量空间不一致。

解决：add 时只传 documents，让 collection 的 embedding_function 自动生成向量。官方文档没把这事说得很直白，新手很容易踩。

坑2：PersistentClient 在并发测试时偶发 PermissionError

现象：CI 里多个测试并行跑，偶尔报 PermissionError: [Errno 13] Permission denied。

原因：Chroma 的 PersistentClient 对同一个 path 会加文件锁，多个进程同时访问同一目录会冲突。我之前为了省事用了固定的 /tmp/chroma_test 路径。

解决：用 pytest 的 tmp_path fixture，每个测试进程分配到独立临时目录。这个坑官方文档里没强调，但在 CI 并行环境下非常常见。

效果验证

场景	加入回归测试前	加入回归测试后
召回漂移漏报	每月 2 次	0 次
写入丢失	每周 1 次	0 次
模型升级回归成本	人工抽查 2 小时	CI 自动跑 15 秒

最近一次 embedding 模型升级，测试直接红了，当场拦截，避免了一次线上事故。这套测试现在是我们 CI 的必跑项。

可直接用的代码/工具

装依赖直接跑：

pip install chromadb pytest numpy
pytest test_vector_memory_regression.py -v

把 GOLDEN_DOCS 和 GOLDEN_QUERIES 换成你真实的记忆数据，就能立刻在项目里用起来。

#Python #向量数据库 #pytest #AI工程 #后端

关于作者

我是宝福，一个实战派后端/架构方向的开发者，喜欢把踩过的坑写成能直接用的代码。

GitHub: https://github.com/baofugege

Sponsor: https://github.com/sponsors/baofugege — 如果这篇文章帮到你，请我喝杯咖啡

提供服务：Python 后端性能优化 / 工具定制 / 技术咨询，联系 Telegram @baofugege

Preventing Agent Context Loss with pytest & Redis Regression Tests

BAOFUFAN — Wed, 19 Aug 2026 01:06:30 +0000

The original article body appears to be missing from your message. Please paste the full Chinese content, including any code blocks, so I can translate it into a natural, Dev.to-style English post while keeping the code unchanged.

After 3 Years of Redis, We Were Testing AI Agent Memory Expiry Wrong

BAOFUFAN — Tue, 18 Aug 2026 12:07:11 +0000

I was woken up at 2 AM by an on-call alert. A user complained that the AI Agent carried over shopping preferences from the previous session into a new one. My first instinct: did the Redis cache fail to isolate? I opened redis-cli and found memory:agent:123:session:456 had existed for 12 hours, with TTL of -1 — this key should have expired after 10 minutes. That's when I realized: after years of using Redis for AI Agent memory storage, our manual tests never truly covered the expiration boundary.

Problem breakdown

AI Agent memory storage has two hard requirements: isolation and expiration policy. Isolation means memories from different sessions of the same agent must be separated, never leaking across. Expiration means session memories should be automatically cleaned up by TTL to avoid contaminating later inference. How did we test before? After development, we manually ran redis-cli set/get to check a few entries, then waited a few minutes to see if they disappeared. The root cause: manual tests only verified that reads/writes “looked like they worked”, but didn't verify concurrent overwrites, TTL boundaries, or key naming rules. On top of that, Redis expiration has two mechanisms — lazy deletion and active deletion — so a manual spot check might see the key still present right after expiry, producing a "false green". Conventional unit tests that mock Redis can't simulate real expiration behavior at all.

Solution design

The choice was straightforward: use Pytest for automated regression, a real Redis 7 instance for testing, switching databases via environment variables. We rejected fakeredis because its expiration policy doesn't fully match real Redis, especially active expiration and TTL-overwriting behavior with SET. Testing against it would be pointless. We also rejected unittest because Pytest's fixtures and scope management give each test case a clean Redis and allow cleanup via yield. Architecturally, we wrapped the storage layer in AgentMemoryStore, and all write operations must go through save_memory. Tests target only this entry point, keeping them aligned with the production path.

Core implementation

This code defines the storage layer, unifies key naming, and uses SET ... EX to ensure every write carries a TTL, avoiding clearing expiration on overwrite.

# memory_store.py
import json
from typing import Any, Optional
import redis

class AgentMemoryStore:
    """AI Agent 会话记忆存储，基于 Redis String 实现。

    key 规则：memory:{agent_id}:{session_id}
    为什么这样设计：Redis 的 key 自带命名空间隔离，能防止不同 agent/session 串数据。
    """
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def save_memory(self, agent_id: str, session_id: str, memory: dict[str, Any], ttl_seconds: int = 600) -> None:
        key = f"memory:{agent_id}:{session_id}"
        payload = json.dumps(memory, ensure_ascii=False)
        # 关键：必须用 SET key value EX ttl，而不是先 SET 再 EXPIRE。
        # 否则如果两条命令之间 Redis 重启/故障，可能留下没有 TTL 的 key。
        self.redis.set(key, payload, ex=ttl_seconds)

    def get_memory(self, agent_id: str, session_id: str) -> Optional[dict[str, Any]]:
        key = f"memory:{agent_id}:{session_id}"
        raw = self.redis.get(key)
        if raw is None:
            return None
        return json.loads(raw)

    def get_ttl(self, agent_id: str, session_id: str) -> int:
        key = f"memory:{agent_id}:{session_id}"
        return self.redis.ttl(key)

This code provides pytest fixtures: each test case gets an isolated database, with cleanup after to avoid cross-test contamination.

# tests/conftest.py
import os
import pytest
import redis
from memory_store import AgentMemoryStore

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/15")

@pytest.fixture()
def redis_client():
    # 用 db=15 作为测试专用，避免误删本地开发缓存
    client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
    client.flushdb()
    yield client
    client.flushdb()
    client.close()

@pytest.fixture()
def store(redis_client):
    return AgentMemoryStore(redis_client)

This code contains the core test cases, directly verifying isolation and expiration policy.

# tests/test_memory_store.py
import time
import pytest

def test_memory_isolated_between_sessions(store):
    agent_id = "agent-1"
    store.save_memory(agent_id, "session-a", {"intent": "buy_shoes"}, ttl_seconds=600)
    store.save_memory(agent_id, "session-b", {"intent": "compare_prices"}, ttl_seconds=600)

    # 同一个 agent 的两个 session 应该读到各自记忆，不能串
    assert store.get_memory(agent_id, "session-a")["intent"] == "buy_shoes"
    assert stor

From 30 Minutes to 3 Seconds: Automated LLM Memory Recall Testing with pytest + FAISS

BAOFUFAN — Tue, 18 Aug 2026 01:05:47 +0000

At 1:30 a.m., the user group blew up: "Why did the AI forget my dietary restriction again?" I dragged myself out of bed and checked. The memory "no cilantro" was still in the memory store, but top-k recall just didn't include it. Manually comparing 20 memories took 30 minutes. As my eyelids grew heavy, it hit me: this kind of regression testing should have been automated long ago.

Breaking Down the Problem

A typical LLM memory pipeline looks like this: conversation -> extract memory -> vectorize -> write to FAISS/vector store -> embed query -> top-k recall. The root causes of recall inconsistency are often subtle: unnormalized vectors, embedding model version drift, order changes during index rebuild, top-k truncation, or incorrect similarity threshold settings. Occasional production hiccups are only investigated after users complain, and running a few manual queries simply cannot prevent regressions. We need an automated verification suite that runs in seconds locally and on every commit.

The Design

I chose pytest as the test framework. Fixtures manage index lifecycle, parametrize covers query cases in bulk, and assertions are clear. The vector index is built in memory with FAISS and paired with deterministic embeddings—simulating real embeddings, replaceable with the OpenAI API—so every run is reproducible. Why not a real vector database? External dependencies are slow and state is hard to clean. Why not LangChain memory tests? Too black-box; you can only see the final answer and cannot locate problems at the vector layer. Plain unittest is not flexible enough. pytest + FAISS is the minimal reproducible unit that can pin recall problems exactly at the vector layer.

Core Implementation

The first code block solves deterministic vectorization. Real embedding models are often uncontrollable and slow, so we generate fixed-dimension vectors from a hash function and apply L2 normalization, semantically approximating "same text, same vector". Production can seamlessly replace it with OpenAI embeddings.

import hashlib
import numpy as np
import faiss

DIM = 128

def text2vec(text: str, dim: int = DIM) -> np.ndarray:
    """确定性文本转向量，模拟 embedding。生产可替换为真实模型输出。"""
    digest = hashlib.sha256(text.encode("utf-8")).digest()
    arr = np.frombuffer(digest, dtype=np.uint8).astype(np.float32)
    vec = np.resize(arr, (dim,))
    # L2 归一化，保证内积等于余弦相似度（IndexFlatIP 依赖这个前提）
    norm = np.linalg.norm(vec)
    if norm > 0:
        vec = vec / norm
    return vec.astype(np.float32)

The second code block solves test isolation and index construction. Each test gets an independent FAISS index to avoid cross-test contamination. A dataclass manages memory records, and the fixture returns both the index and original data for convenient assertions.

import pytest
from dataclasses import dataclass

@dataclass
class MemoryRecord:
    id: str
    content: str
    metadata: dict = None

# 测试用记忆库
MEMORIES = [
    MemoryRecord("m1", "用户不吃香菜", {"user_id": "u1"}),
    MemoryRecord("m2", "用户对花生过敏", {"user_id": "u1"}),
    MemoryRecord("m3", "用户喜欢喝美式咖啡", {"user_id": "u1"}),
    MemoryRecord("m4", "用户是素食主义者", {"user_id": "u2"}),
]

@pytest.fixture(scope="function")
def memory_index():
    """每个测试独立构建索引，避免状态污染"""
    index = faiss.IndexFlatIP(DIM)  # 内积索引，要求向量已归一化
    vectors = np.zeros((len(MEMORIES), DIM), dtype=np.float32)
    for i, mem in enumerate(MEMORIES):
        vectors[i] = text2vec(mem.content)
    index.add(vectors)
    return index, MEMORIES

The third code block solves recall consistency assertions. Multiple queries are parametrized, asserting top-1 must hit the expected memory. It also verifies that recall results remain unchanged after rebuilding the index, covering the "redeploy" scenario.


python
@pytest.mark.parametrize("query,expected_id", [
    ("不吃香菜", "m1"),
    ("花生过敏", "m2"),
    ("美式咖啡", "m3"),
    ("素食", "m4"),
])
def test_recall_top1_consistency(memory_index, query, expected_id):
    index, memories = memory_index
    q_vec = text2vec(query).reshape(1, -1)
    scores, ids = index.search(q_vec, k=3)
    top1_id = memories[ids[0][0]].id
    assert top1_id == expected_id, f"召回不一致: 期望 {expected_id}, 实际 {top1_id}"
    # 额外断言得分范围，防止归一化失效
    assert -1.0 <= scores[0][0] <= 1.0

def test_rebuild_index_recall_consistent(memory_index):
    """模拟索引重建，验证同一批记忆重建后召回结果不变"""
    index, memories = memory_index
    new_index = faiss.IndexFlatIP(DIM)
    vectors = np.zeros((len(memories), DIM), dtype=np.float32)
    for

From 30 Minutes to 5 Seconds: Testing Agent Memory with Pytest + SQLite

BAOFUFAN — Mon, 17 Aug 2026 12:06:23 +0000

At 1 AM, a message popped up in the test group chat: the agent had leaked one user's memory into a new session. I opened DB Browser, hand-wrote 8 SELECT queries, and compared session_id and key row by row. Half an hour later, I tracked it down to a missing session_id in the WHERE condition of the upsert. This wasn't the first time.

Breaking Down the Problem

The agent's memory storage is basically just a SQLite table with three core fields: session_id, key, and value. The consistency requirements are simple too: updating the same key within a session must not create duplicate rows; different sessions must not bleed into each other; concurrent upserts must not overwrite each other.

But manual testing only covers whichever SQL query you happen to have in mind at the time. Each regression run requires preparing data, cleaning up old tables, and eyeballing the results—those three scenarios alone eat up 30 minutes. Even more frustrating, the occasional concurrent write issue is nearly impossible to reproduce by hand. The root cause is the lack of a fast, repeatable, automated consistency test—not that the business code itself is particularly complex.

Design

I chose pytest with an in-memory SQLite database. The core idea is that each test function gets a fresh :memory: store, and the connection closes automatically after the test.

I didn't use unittest because fixtures give you more flexible setup/teardown, and pytest's rewritten assertions produce clearer failure messages. I also skipped the production MySQL: it's slower and pulls in more environment dependencies. Since most of the consistency logic lives at the SQL layer, SQLite is enough to simulate it. The only extra thing needed is a file-backed fixture to verify that data survives a restart after disk persistence.

Core Implementation

This code implements a minimal but complete agent memory store. It uses a composite primary key (session_id, key) and an upsert with ON CONFLICT for updates. All the tests below build on it.

import sqlite3
import time
from contextlib import closing
from typing import Optional


class AgentMemoryStore:
    """Agent 记忆存储的最小实现，核心是复合主键和 upsert 一致性。"""

    def __init__(self, db_path: str = ":memory:", check_same_thread: bool = False):
        # check_same_thread=False 是为了后续并发测试不被线程检查打断
        self.conn = sqlite3.connect(db_path, check_same_thread=check_same_thread)
        self.conn.row_factory = sqlite3.Row
        self._init_schema()

    def _init_schema(self) -> None:
        with closing(self.conn.cursor()) as cur:
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS memories (
                    session_id TEXT NOT NULL,
                    key TEXT NOT NULL,
                    value TEXT NOT NULL,
                    updated_at REAL NOT NULL,
                    PRIMARY KEY (session_id, key)
                )
                """
            )
            self.conn.commit()

    def upsert(self, session_id: str, key: str, value: str) -> float:
        ts = time.time()
        with closing(self.conn.cursor()) as cur:
            cur.execute(
                """
                INSERT INTO memories (session_id, key, value, updated_at)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(session_id, key)
                DO UPDATE SET value = excluded.value,
                              updated_at = excluded.updated_at
                """,
                (session_id, key, value, ts),
            )
            self.conn.commit()
        return ts

    def get(self, session_id: str, key: str) -> Optional[str]:
        with closing(self.conn.cursor()) as cur:
            cur.execute(
                "SELECT value FROM memories WHERE session_id = ? AND key = ?",
                (session_id, key),
            )
            row = cur.fetchone()
        return row["value"] if row else None

    def close(self) -> None:
        self.conn.close()

The test file below uses a function-scope fixture to give each test case its own database. It covers the three consistency points most likely to break: updating the same key, session isolation, and duplicate composite primary keys.

import pytest

from memory_store import AgentMemoryStore  # 假设上面的类放在 memory_store.py


@pytest.fixture()
def store():
    store = AgentMemoryStore(":memory:")
    yield store
    store.close()


def test_upsert_

2026-08-17-用-pytest--redis-为-ai-agent-记忆存储实现自动化回归测试

BAOFUFAN — Mon, 17 Aug 2026 01:05:22 +0000

Automating AI Agent Memory Write, Expiry, and Recall Consistency Verification with Pytest and Redis

BAOFUFAN — Sun, 16 Aug 2026 12:05:48 +0000

It looks like the article body was not included in your request. Please paste the full Chinese article content and I’ll translate it into natural, Dev.to-style English, keeping all code blocks unchanged.

Qdrant Recall Inconsistency: It Took 300 Test Runs to Discover the Index Wasn't Refreshed

BAOFUFAN — Sun, 16 Aug 2026 01:05:06 +0000

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.

Problem Breakdown

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.

Design

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.

Core Implementation

This code solves test environment isolation, ensuring each test uses an independent Qdrant instance and collection.

# conftest.py
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"
    # 向量维度 3，余弦距离
    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.

# test_recall_failure.py
import pytest
from qdrant_client.models import PointStruct

def test_immediate_recall_without_wait(mem_collection):
    client, collection = mem_collection
    # 写入一条向量，注意：没有 wait=True
    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
    )
    # 这一行经常失败：result.points 为空
    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.

# test_recall_fix.py
import time
from qdrant_client.models import PointStruct, VectorParams, Distance

def test_immediate_recall_with_wait(mem_collection):
    client, collection = mem_collection
    # 关键参数：wait=True，确保写入后索引刷新完成
    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
    # 还可以进一步验证 payload 一致性
    assert result.points[0].payload["user"] == "alice"

If you don't want to add wait=True to every call, you can...

Debugging AI Agent Context Pollution with pytest + Redis: 3 Cross-Talk Bugs in 6 Hours

BAOFUFAN — Sat, 15 Aug 2026 12:05:49 +0000

At 2 a.m., a screenshot landed in the user group: the AI support agent replied to a ticket and leaked another user's phone number and shipping address. My first instinct was model hallucination, but digging into Redis revealed that two sessions with different user_id values shared the same session_id as the key — the context was cross-wired. This wasn't a model problem; the memory layer was running loose. I later built a consistency test suite with pytest + Redis to finally kill this class of bugs.

Breaking Down the Problem

Memory storage for AI agents is different from a normal cache: losing a cache only makes things slower, but mixing up memories directly leaks privacy and derails conversations. Our team stores the agent's short-term memory in Redis, relying on it to keep multi-turn conversations coherent. After launch, we occasionally saw user A seeing user B's history. The root cause wasn't the model — it was three issues in the memory layer: key design lacking a user dimension, messy TTL refresh logic, and inconsistent serialization protocols. Why weren't regular approaches enough? Code review and manual testing can't catch this because cross-talk only appears under concurrency plus edge-case combinations; mocking Redis can't verify real TTL behavior or atomicity. In the end, I decided to treat the memory layer as infrastructure and write consistency tests with pytest + Redis.

Solution Design

The reason to choose pytest + Redis instead of unit-test mocks is simple: Redis's real behavior — TTL calculation, WATCH optimistic locking, and binary differences after serialization — cannot be simulated by mocks. Pytest fixtures can give each test a clean database, and a real Redis can be spun up with Docker. In the design, we wrap memory reads and writes into MemoryStore, then write tests against three invariants: key isolation, TTL/version compatibility, and concurrent append consistency. Why not test directly against production? Too dangerous. Why not only write unit tests? Because they treat Redis like a plain dict and hide TTL and concurrency problems.

Core Implementation

This code solves the test isolation problem: each test case gets an independent Redis db so leftover data doesn't pollute assertions.

# conftest.py
import pytest
import redis

@pytest.fixture
def redis_client():
    """连接本地 Redis，测试前清空当前 db，保证隔离"""
    client = redis.Redis(host="localhost", port=6379, db=15, decode_responses=True)
    client.flushdb()
    yield client
    client.flushdb()
    client.close()

This class encapsulates production-grade Redis memory reads and writes, including key namespacing, JSON versioning, SETEX TTL, and concurrency-safe WATCH appends. Key lines are commented.


python
# memory_store.py
import json
from typing import Any, Dict, List, Optional

import redis
from redis.exceptions import WatchError

class MemoryStore:
    """AI Agent 短期记忆存储，负责 key 隔离、TTL 和版本化序列化"""

    def __init__(self, client: redis.Redis, ttl: int = 3600):
        self.client = client
        self.ttl = ttl
        self.version = 1  # 序列化

Regression Testing LLM Memory Storage with pytest & Redis to Prevent Multi-turn Context Loss

BAOFUFAN — Sat, 15 Aug 2026 01:05:51 +0000

The original article body was not provided. Please paste the full Chinese article content so I can translate it into natural English for the Dev.to community.