DEV Community: BAOFUFAN

How I Slashed Context Loss from 30% to 0% with Automated LangChain Memory Tests

BAOFUFAN — Thu, 11 Jun 2026 12:08:56 +0000

I was jolted awake at 1 AM by an alert call. The ops team told me that our chatbot—supposedly “the most empathetic” one—had started babbling nonsense: one moment a user was discussing after-sales support, the next the bot blurted out “Which type of pizza would you like?”—like a completely drunk customer service rep. Digging through logs, I discovered LangChain’s memory module had scrambled the conversation context, messing up the entire history after just a few turns. The real kicker? Manual testing before deployment never caught it. Testers merely clicked around a few times and never hit multi-turn edge cases. Right then, I knew: the practice of “eyeballing” context consistency had to end.

Breaking Down the Problem: Why Manual Testing Can’t Safeguard Context

In a multi-turn dialogue system, LangChain manages memory behind the scenes—storing each user input and system output into some store and injecting it back into prompts on subsequent turns. We used all sorts of memory types: ConversationBufferMemory, ConversationSummaryMemory, ConversationBufferWindowMemory… On the surface, calling save_context and load_memory_variables seems trivial, and the in-memory dictionary should remain stable. In reality:

When the window slides, BufferWindowMemory silently drops messages, and you never know which turn disappeared.
Summary memory (ConversationSummaryMemory) relies on an LLM to re-summarize; after repeatedly saving the same conversation, the summary can produce “hallucinated information” or lose critical entities.
When memory_key conflicts or overlaps with a prompt variable name, the output of load_memory_variables quietly overwrites other variables, warping the prompt.

The biggest flaw of manual testing is that a human cannot simulate dozens of dialogue turns in a short time and meticulously compare the memory output to the original conversation word by word. Typical unit tests only check “save once, load once” and never touch the chaos that accumulates with context. So the root cause wasn’t LangChain—it was that we never wrote true deterministic validations for the memory module.

Designing the Solution: Turning Memory Validation into a Repeatable Machine Task

My approach was straightforward: build an automated test framework that acts like a robot user, runs conversations for N turns, and after each turn asserts that the stored context matches the complete conversation history exactly.

In terms of implementation, the naive way is to sprinkle assert statements throughout business unit tests—but that would mean repeating the same boilerplate for every memory type and inevitably missing edge cases. Instead, I abstracted a MemoryValidator class that:

Drives the memory by calling save_context turn by turn according to a script, while maintaining an independent golden history.
After each save, immediately calls load_memory_variables, retrieves the memory contents, and compares them deterministically against the golden history.
For summary-based memories, it doesn’t require exact string matching, but uses rules (such as entity presence) to ensure no critical information is lost.

Why not use LangChain’s built-in tests? I dug through their source code. Their tests lean toward module-level unit tests, lacking integration checks over accumulated multi-turn history. Even worse, they don’t cover the correct ordering after window truncation and re-injection. So I had to roll my own.

With this setup, whenever we introduce a new memory type or tweak a configuration, a single test run instantly tells us whether the context has “betrayed” us.

Core Implementation: Three Code Chunks to Automate Memory Testing

1. The Universal Validator: No Matter the Memory Type, Behavior Must Be Auditable

The code below implements MemoryValidator, which takes over memory read/write operations while precisely recording each turn in _golden_dialogue. It exposes step_and_validate for uniform usage. The key design choice is normalization of different memory outputs: all outputs are converted into lists of plain text before comparison, avoiding false mismatches caused by varying Message objects or dictionary structures.

from typing import List, Tuple, Any
from langchain.memory import BaseMemory

class MemoryValidator:
    """自动化校验多轮对话记忆的一致性"""

    def __init__(self, memory: BaseMemory, input_key: str = "input",
                 output_key: str = "output", memory_key: str = "history"):
        self.memory = memory
        self.input_key = input_key
        self.output_key = output_key
        self.memory_key = memory_key
        self._golden_dialogue: List[Tuple[str, str]] = []  # 存储原始对话对

    def _normalize_output(self, output: Any) -> List[str]:
        """归一化为字符串列表，屏蔽LangChain不同记忆的输出差异"""
        if isinstance(output, dict):
            # 从返回的字典里取出记忆key对应的内容
            value = output.get(self.memory_key, "")
        else:
            value = output

        if isinstance(value, list):
            # 可能是Message对象列表，转成纯文本
            return [msg.content if hasattr(msg, "content") else str(msg) for msg in value]
        if isinstance(value, str):
            # 某些memory返回长文本，按换行拆开方便逐条比对
            return [line for line in value.split("\n") if line.strip()]
        return [str(value)]

    def step_and_validate(self, user_input: str, ai_output: str):
        """执行一轮对话并立即校验记忆是否完好"""
        # 1. 保存上下文
        self.memory.save_context(
            {self.input_key: user_input},
            {self.output_key: ai_output}
        )
        self._golden_dialogue.append((user_input, ai_output))

        # 2. 加载当前记忆
        loaded = se

From 100 Logins to 1: Cut E2E Test Time by 78%

BAOFUFAN — Thu, 11 Jun 2026 01:05:40 +0000

It’s 2 AM. The CI pipeline has just failed — again — because of E2E test timeouts. You open the Allure report and see that 47 out of 50 test cases were stuck on the login page for over 8 seconds. Your team is writing automated tests with Playwright, but every single test goes through the entire login flow from scratch: type username, type password, click the button, wait for the redirect. You think, “Isn’t this just burning time and money for nothing?” What’s even more absurd is that no one thought about saving and reusing the login state.

That’s what we’re covering today: using Playwright’s storageState together with pytest’s fixture mechanism to compress repeated logins into a single one, while extracting the assertion logic into reusable “memory modules” to make your test code shorter and more stable.

Breaking Down the Problem

The scenario is all too common: you have an admin system (like an operations panel or SaaS console) that requires login to access. You’ve written 50 P0-level functional tests. The simplest approach puts page.goto('/login') in every case, fills the form, logs in, and then tests the actual business logic. The results:

Execution time explodes: The average login takes 2.3 seconds (including page load, typing, clicking, waiting for the dashboard). 50 test cases devour 115 seconds just logging in, and with CI resource queuing, one run easily exceeds 8 minutes.
Maintenance hell: The moment a login page selector changes, every single case must be updated. For example, if the button changes from #login-btn to [data-testid="submit"], you need a global search-and-replace. Miss one and a whole bunch of tests fail.
Assertion duplication: Almost every case contains similar expect(page.locator('.toast')).to_have_text('Success') checks. The same validation logic is scattered everywhere. Adding a new case means copy-pasting assertions, so a slight toast text change can break a dozen tests.

The root cause is clear: the tests don’t separate “state” from “behavior.” Login is a prerequisite state, not a business behavior — it should be shared. Assertions are checks against page state and should be encapsulated as domain language, not littered with locators and raw strings.

What about the usual workarounds? Some people use setUp to log in before each test and tearDown to log out — but that still performs the login every time, only deduplicating the code, which treats the symptom but not the cause. Others manually paste cookies into the code, but the moment a token expires everything breaks. We need an automated, refreshable, cross-case reusable login state solution, while also making assertions as callable as a library.

Solution Design

The technical choice is straightforward: Playwright natively offers context.storage_state(), which serializes cookies, localStorage, and IndexedDB of the current browser context into a JSON file. Next time you load it with browser.new_context(storage_state=path), it restores the login state directly, completely bypassing the login page.

Architecturally, we use pytest fixture scopes to achieve different levels of sharing:

session-scoped fixture: responsible for generating storage_state.json. If the file already exists and hasn’t expired, it reuses it directly; otherwise it performs a single login and saves the state.
function-scoped fixture: each test case derives a fresh page from the already-logged-in context, so tests remain isolated from each other and won’t pollute one another.
Reusable assertion module: common checks like toast verification, table row count, modal text are wrapped as independent functions living in assertions.py, called uniformly by all tests.

Why not other approaches?

Not storing cookies in a global variable: cookies can be large, and localStorage / IndexedDB data can’t be reliably captured that way; storageState fully serializes everything.
Not using pytest-xdist’s group_scope: although it can share fixtures across workers, it requires an extra plugin and has concurrency issues when reading/writing the storageState file. A single session-scoped fixture with a locking mechanism is more stable.
Not calling an API to get a token and then setting localStorage in every case: some systems bind tokens to browser fingerprints, so simply setting them may not work. Following the full login flow is more reliable.

With this design, 50 test cases need only a single real login, and all assertion logic converges into 3–5 functions. Maintenance effort is slashed by half.

Core Implementation

1. Session-Scoped Fixture: “Login Once, Use Everywhere”

What this code does: it runs the login only once during the entire pytest lifecycle and persists the browser state to a temporary file. On subsequent test runs, if the state file exists and the token has not expired, it loads directly and skips login.

# conftest.py
import json
import os
import time
import pytest
from playwright.sync_api import sync_playwright, Browser, BrowserContext

STATE_FILE = "storage_state.json"
TOKEN_EXPIRE_SECONDS = 7200  # 假设 token 有效期 2 小时

@pytest.fixture(scope="session")
def playwright_instance():
    with sync_playwright() as p:
        yield p

@pytest.fixture(scope="session")
def browser(playwright_instance):
    # 这里用 headless 模式，CI 环境友好
    browser = playwright_instance.chromium.launch(headless=True)
    yield browser
    browser.close()

@pytest.fixture(scope="session")
def logged_in_context(browser: Browser) -> BrowserContext:
    # 状态文件如果存在且未过期，直接复用
    if os.path.exists(STATE_FILE):
        mtime = os.path.getmtime(STATE_FILE)
        if time.time() - mtime < TOKEN_EXPIRE_SECONDS:
            context = browser.new_context(storage_state=STATE_FILE)
            yield context
            context.close()
            return

    # 否则走完整登录流程
    conte

Pitfalls of Testing LLM Long-Term Memory: A 3‑Day Debugging Saga

BAOFUFAN — Wed, 10 Jun 2026 12:06:03 +0000

I was jolted awake at 2 a.m. by a PagerDuty alert — users were complaining that the AI “called me Mr. Wang yesterday, but today it doesn’t recognise me at all.” Groggily I pulled up the monitoring dashboards and saw that the vector database’s retrieval latency had spiked, and the last two conversation turns had simply vanished from the memory recall. What made my blood run cold was the realisation: our existing manual test suite couldn’t even reach this cross‑session scenario. Long‑term memory was silently “forgetting”, and we didn’t even have a way to prove it.

This article is the story of how I turned LLM memory verification from “winging it” into a reliable automated test using Playwright — and all the traps I stepped into along the way. If you’re building the memory module for a RAG pipeline or an Agent, this might save you a few days.

Breaking down the problem: why manual testing is “pretend testing”

Our product’s long‑term memory follows a typical RAG architecture: after a conversation ends, key facts or summaries are stored in a vector database. The next time the user asks a question, the most relevant memory snippets are retrieved and stitched into the System Prompt. Sounds simple, but memory accuracy actually depends on three linked steps:

Memory writing: did the summarisation model extract the correct facts? Is the embedding properly aligned?
Retrieval recall: does the vector similarity search rank the older memory first? Is a threshold cutting out relevant fragments?
Fusion & reasoning: once the LLM sees the prompt, is it truly “recalling” or is it just improvising?

The conventional approach is to test the API — create a session, push a few messages, then query the vector database or check the model’s output. But real users switch devices, clear caches, hop between networks. The behaviour of the frontend and the API gateway is completely missed. The streaming response parsing logic, the session‑ID binding — all of that is a blind spot in API‑only testing. We had to go end‑to‑end, through a real browser, the whole journey.

Why not Cypress or Selenium? Playwright has first‑class support for isolated browser contexts, which lets you cleanly simulate “two independent sessions”. It also allows you to intercept network requests and observe exactly when the memory API is called — something that’s quite painful to achieve with Selenium. So the choice fell on Playwright + pytest.

Designing the test: a “dual‑session” pattern to verify memory persistence

The architectural idea is bluntly simple:

Open two completely isolated browser contexts with Playwright (simulating two independent visits — cookies and storage are wiped).
In the first context, run a “teaching conversation” and plant a unique fact (e.g., the user’s name is 张三-2027).
Wait for the backend’s async processing to finish — memory writing and vector index refresh.
In the second context, ask: “What name did I use when we first met?” and assert that the reply contains 张三-2027.

A failed test means the memory is inaccurate — either it was dropped or the model hallucinated. This is not a trivial keyword‑match either; the model might answer “You told me your name is 张三-2027” or “I remember you said 张三-2027”, so the assertion needs some tolerance.

The test wraps the chat UI in a Page Object. The three crucial aspects to verify:

Wait for the AI’s response to finish completely — never take a half‑streamed sentence.
Cross‑session isolation is bullet‑proof — no accidental reuse of the session ID.
Memory retrieval timeliness — if you query right after writing, the record might not be indexed yet, so you must build in a retry window.

Here’s how the implementation actually looks.

Core implementation: copy‑pastable Playwright test code

1. Chat Page Object

This code deals with two headaches: ① how to tell that “the AI has finished streaming”; and ② providing a high‑level send_and_wait_reply method so the test cases stay clean.

# chat_page.py
from playwright.sync_api import Page, Locator
import time

class ChatPage:
    def __init__(self, page: Page):
        self.page = page
        self.input_box = page.locator('textarea[placeholder="输入消息"]')
        self.send_btn = page.locator('button:has-text("发送")')
        # The stop button is only visible while streaming
        self.stop_btn = page.locator('button:has-text("停止生成")')
        # The last AI reply
        self.last_ai_msg = page.locator('[data-role="assistant-message"]').last

    def goto(self, url="/chat"):
        self.page.goto(url)
        self.page.wait_for_selector('textarea', timeout=10000)

    def send_message(self, text: str):
        self.input_box.fill(text)
        self.send_btn.click()

    def wait_for_reply_complete(self, timeout=30000):
        """Wait until streaming finishes: the stop button disappears and the last message text stabilises."""
        start = time.time()
        last_text = ""
        stable_count = 0
        while time.time() - start < timeout:
            # If the stop button is still visible, generation is ongoing
            if self.stop_btn.is_visible():
                stable_count = 0
                time.sleep(0.3)
                continue
            current = self.last_ai_msg.text_content() or ""
            if current == last_text and len(current) > 0:
                stable_count += 1
                if stable_count >= 3:   # 3 consecutive identical samples -> done
                    return current
            else:
                stable_count = 0
                last_text = current
            time.sleep(0.3)
        raise TimeoutError("AI reply did not finish within the timeout")

    def send_and_wait_reply(self, text: str) -> str:
        self.send_message(text)
        return self.wait_for_reply_complete()

From JSON to Pinecone: 90% Accuracy Boost for AI Long-Conversation Memory

BAOFUFAN — Wed, 10 Jun 2026 01:05:08 +0000

At 2 AM, my boss sent three frantic messages in a row: “Users say the AI doesn’t remember what they discussed half an hour ago—is this a bug?” Still half‑asleep, I opened the monitoring dashboard and saw the service had been restarted for a rolling update three hours earlier. The conversation history that had been living in memory was gone—total amnesia. Even worse, right before the restart a user had spent 20 minutes explaining their business rules. Now the AI was acting like a brand‑new intern, asking “How can I help you?” That’s the ugly side of relying on in‑memory conversation memory alone.

Breaking down the problem

When you build an AI chat service with memory, the most common approach is to use LangChain’s ConversationBufferMemory and stuff the whole conversation history into the prompt. It works fine in dev, but the moment you enter production, cracks appear:

On restarts or scaling, the in‑memory history simply evaporates. Your users have to re‑explain everything seconds later.
Long conversations — customer support, legal consultations, code reviews — easily accumulate thousands of tokens. ConversationBufferWindowMemory only keeps the last N turns, so critical context often gets lost.
High concurrency — even if you maintain one memory object per session, memory usage balloons, and persistence is still MIA.

The root cause is one thing: storing and retrieving conversation memory should never be the job of process memory. You need a place that persists history and lets you search it by meaning — that’s where a vector database comes in. Pinecone’s serverless option perfectly solves the ops nightmare of “I don’t want to spin up a Milvus cluster.”

Solution design

The core idea: Generate an embedding for every conversation turn, store it in Pinecone, and when a new question comes in, do a semantic search for the most relevant history. Take those snippets, assemble them into memory, and inject them into the LLM. This way, even after a restart the conversation history is safe in the cloud. And if the conversation runs for hundreds of turns, we only pull the few that actually matter for the current topic — giving the LLM a kind of “locally omniscient” superpower.

I considered a few options:

Redis + vector plugin: you still manage Redis, vector filtering capabilities are weak, and writing Lua scripts for metadata filtering is painful.
Storing full conversations in Pinecone and pulling them all back: wastes bandwidth, misses the point of vector search — basically using Pinecone like MongoDB.
LangChain’s VectorStoreRetrieverMemory + Pinecone: a perfect match. LangChain already wraps the retrieval logic and memory interface; Pinecone provides highly available, elastic vector storage with built‑in metadata filtering (so you can isolate users by session_id).

Why not the others? Because with this combo, developers get persistent semantic memory in less than 50 lines of code — no sharding, no replicas, no scaling headaches.

Core implementation

We’ll use langchain + pinecone-client, built for production readiness. Three steps.

1. Initialize the Pinecone index and embeddings

This snippet solves the “bridge between conversation history and vector storage” problem. You have to create the index first, and its dimension must match your embedding model — OpenAI’s text-embedding-ada-002 outputs 1536‑dimensional vectors. (I’ll explain a nasty dimension‑mismatch pitfall later.)

import os
from pinecone import Pinecone, ServerlessSpec
from langchain_openai import OpenAIEmbeddings

# 初始化 Pinecone 客户端
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index_name = "chat-memory"

# 如果索引不存在，创建它（注意：Serverless 需要指定 cloud 和 region）
if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,  # 必须与 embeddings 模型输出维度一致
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

# 拿到索引对象
index = pc.Index(index_name)

# 初始化 embeddings（可以替换成其他兼容接口的模型）
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")

2. Build the memory‑augmented LLM chain

This part solves “how to give the LLM relevant history on every turn.” The heart is VectorStoreRetrieverMemory, which persists conversations to Pinecone and retrieves the top‑k most similar records on demand.

from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.schema import Document

# 创建 vector store（基于 Pinecone 索引）
from langchain_pinecone import PineconeVectorStore  # 需安装 langchain-pinecone

vectorstore = PineconeVectorStore(index, embeddings, text_key="text")

# 先存入一条初始记忆，避免空检索报错（真实场景可以跳过）
if not index.describe_index_stats()['total_vector_count']:
    vectorstore.add_documents([
        Document(page_content="对话开始，用户和助手开始交流。", metadata={"session_id": "default"})
    ])

# 定义 retriever，返回相似度最高的 4 条记忆
retriever = vectorstore.as_retriever(search_kwargs=dict(k=4))

# 包装成 LangChain Memory，注意 memory_key 要与 prompt 模板里的占位符一致
memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="chat_history",       # prompt 里用 {chat_history} 引用
    input_key="input",

Bringing LLM Memory Regression Tests from 30 Minutes Down to 90 Seconds with pytest + Redis

BAOFUFAN — Tue, 09 Jun 2026 12:05:24 +0000

At 2:17 a.m., a chain of alarms yanked me out of sleep — the LLM in production had suddenly "lost its memory." One moment users were discussing project timelines, the next it was naïvely asking "How can I help you?" After digging through logs, I found the culprit: a single line change to Redis’s expiry policy during a memory-store release. Nobody had tested the memory persistence flow before deployment, causing every session’s context to expire within 5 minutes. While fixing the bug, I cursed to myself: if only we had automated, repeatable regression tests that never touch production data, this would never have happened.

Breaking down the problem

At its core, an LLM memory store is a key–value system with TTL: session IDs serve as keys, and dialogue history, summaries, vector indexes are serialized and dumped into Redis. The business requirement is clear — "7×24 multi-turn conversations must never lose memory." Yet our testing process was stuck here:

Fire a few messages manually via Postman, then eyeball Redis keys to guess if it’s working.
Test data shares the same Redis instance as production; one slip and you’ve deleted real sessions.
Redis specifics like lazy expiration, asynchronous deletion are ignored, making timeout tests nothing more than superstition.

The root cause isn’t that Redis is unreliable — it’s that we never integrated Redis’s real behavior into regression tests. Mocking Redis with a plain dict means you’re not testing timeouts, serialization, or connection-pool exhaustion. Deploying without those checks is like flying blind.

Solution design

What we need is a test setup that’s “real Redis, fully isolated, auto-cleaned, and repeatable.”

Technology choices:

pytest: Its fixture system is perfect for managing test resources. Tagging tests with @pytest.mark.redis lets you flexibly skip real-backend tests in CI when needed.
Real Redis: No fakeredis — it doesn’t support time travel (you have to actually wait for expiration), and its lagging support for Lua scripts, Streams, and modules leads to tests passing while production explodes. For local development, use a dedicated db=15; in CI, spin up a dedicated instance with docker-compose.
Why not testcontainers-python: For small teams with flaky networks, pulling an image on every test run is painfully slow. Running Docker-in-Docker inside CI containers is also a configuration mess. Directly connecting to a “prepared, dedicated Redis” is far more practical.

Architecture idea: Each test gets a client scoped to a unique namespace via the redis_memory_store fixture defined in conftest.py. After the test, all keys under that namespace are atomically deleted, preventing interference between parallel tests.

Core implementation

1. Fixture providing a fully isolated Redis client

The code below solves the problem of “dirty data leaking between tests.” By using a random prefix and an autouse cleanup hook, each test lives in its own sandbox.

# conftest.py
import pytest
import redis
import uuid

# 命令行参数：允许跳过需要真实Redis的测试
def pytest_addoption(parser):
    parser.addoption("--real-redis", action="store_true", default=False,
                     help="run tests that require a real Redis instance")

@pytest.fixture(scope="session")
def redis_url():
    # 默认连本地 test 数据库，生产请通过环境变量覆盖
    return "redis://localhost:6379/15"

@pytest.fixture
def redis_memory_store(redis_url):
    """
    为每个测试函数创建一个前缀隔离的 Redis 客户端，
    测试结束后自动删除该前缀下所有 key。
    """
    prefix = f"test:{uuid.uuid4().hex[:8]}:"  # 避免并行测试 key 冲突
    client = redis.Redis.from_url(redis_url, decode_responses=True)
    client._test_prefix = prefix              # 挂个属性方便业务代码使用

    yield client

    # 清理：用 scan 分批删除，防止 keys * 阻塞
    cursor = 0
    while True:
        cursor, keys = client.scan(cursor, match=f"{prefix}*", count=100)
        if keys:
            client.delete(*keys)
        if cursor == 0:
            break
    client.close()

2. Business tests for the memory store

These tests cover the core scenarios: write a dialogue memory → read → update → automatic expiration. The design injects a MemoryStore dependency so that during tests you pass in redis_memory_store, while in production you inject a connection pool.

# test_memory_store.py
import time
import pytest
from myapp import MemoryStore  # 你的实际记忆存储模块

@pytest.mark.redis
class TestMemoryPersistence:
    """记忆持久化回归测试"""

    def test_write_and_read_conversation(self, redis_memory_store):
        """基本读写：存一段对话，拉回来必须完全一致"""
        store = MemoryStore(redis_memory_store, prefix=redis_memory_store._test_prefix)
        session_id = "user_abc"
        messages = [{"role": "user", "content": "我叫小明"}, {"role": "assistant", "content": "你好小明"}]

        store.save(session_id, messages)
        loaded = store.load(session_id)

        assert loaded == messages, f"预期 {messages}，实际 {loaded}"

    def test_memory_ttl(self, redis_memory_store):
        """TTL过期：写入短暂TTL，等待后数据应该消失"""
        store = MemoryStore(redis_memory_store, prefix=redis_memory_store._test_prefix, ttl=2)
        session_id = "user_ttl"
        store.save(session_id, [{"role": "user", "content": "test"}])

        time.sleep(3)  # 等待过期
        loaded = store.load(session_id)

        assert loaded is None or loaded == [], f"过期后应返回空，实际 {loaded}"

    def test_parallel_isolation(self, redis_memory_store):
        """并行隔离：两个测试使用不同前缀，互不影响"""
        store_a = MemoryStore(redis_memory_store, prefix="a:")
        store_b = MemoryStore(redis_memory_store, prefix="b:")
        store_a.save("1", [{"role": "x"}])
        store_b.save("1", [{"role": "y"}])

        assert store_a.load("1") == [{"role": "x"}]
        assert store_b.load("1") == [{"role": "y"}]

What we gained

From 30‑minute manual verification to a 90‑second fully automated regression suite that catches real-world Redis edge cases before they hit production. Now every PR that touches memory‑related code must pass these tests, and the “LLM amnesia” pager has stayed silent ever since. The real Redis, clever namespace isolation, and pytest fixtures are the three pillars that made it possible. If your team is still testing LLM memory with mocks or shared instances, it’s time to give this approach a shot.

Automating Agent Memory Regression with pytest & Vector DB: 5x Defect Discovery Speedup

BAOFUFAN — Tue, 09 Jun 2026 05:05:05 +0000

It was 1 AM when my phone rang. "You updated the embedding model this afternoon, and now the agent’s memory retrieval is completely broken — every user session is spouting nonsense." I scrolled through the chat history and realized the issue: after the model change, a memory that used to get perfect hits was now ranked eighth. I had manually tested only two examples and thought everything was fine. While rolling back the change that night, I made a decision: regression testing for agent memory storage must be automated.

Breaking down the problem

Long-term memory in an AI agent boils down to “store vectors, query by similarity.” Facts, preferences, and context generated during user interactions are embedded into high-dimensional vectors and inserted into a vector database. Later, when the agent needs to make a decision, it converts the current question into a vector, retrieves the top‑N most similar memories, and feeds them as part of the prompt.

The crux: semantic similarity is not an exact match. Every change to the embedding model, chunking strategy, or even a version bump of the vector database can silently alter retrieval results. Manual verification is a disaster — you throw a few keywords into the UI and eyeball whether the recalled results “look right.” This approach has three fatal flaws:

No quantification: “Looks about right” can’t be turned into an assertion; ranking shifts often go unnoticed.
Not repeatable: Each manual test uses different query terms, making it normal to miss regressions.
Maintenance cost explosion: Once the number of stored memories grows large, manually running a full regression is impossible — and therefore simply doesn’t happen.

Ordinary unit tests can check that a function “returns a list,” but they can’t verify that “the first item in the list is the sentence the user said last week.” We need a regression test suite that can make precise assertions about vector search results while being fast and repeatable.

Designing the solution

My goal was clear: drive real vector database queries through pytest, in a controlled environment, and validate retrieval accuracy and stability.

For the tech stack, I chose ChromaDB’s in-memory mode as the test vector store, with all-MiniLM-L6-v2 as a lightweight embedding model. Why not just compute cosine similarity with numpy and assert on that? Because we want to test real production behavior — the vector store’s indexing algorithm (e.g., HNSW), distance metric, and result ordering. None of these can be simulated by numpy. If we recalculate similarity ourselves, we’re testing a toy that never runs in production.

ChromaDB’s EphemeralClient fits pytest perfectly: a function-scoped fixture creates a temporary database that is destroyed after the test, leaving zero pollution. Each test case inserts its own memories, performs queries, and makes assertions in complete isolation.

I structured the solution in three layers:

Fixture layer: responsible for creating the Chroma client and collection, and inserting a standard set of test memories.
Case layer: uses @pytest.mark.parametrize to define different query scenarios and calls the agent’s memory retrieval function.
Assertion layer: validates the ID order of top‑k results, text prefixes, score thresholds, and fallback behavior for empty results.

With this setup, after any change — model, chunking, or library version — a single pytest -v command runs 30+ regression scenarios in under five seconds.

Core implementation

First code block: conftest.py. Solves the problem of giving each test case an isolated, pre-populated vector database.

# conftest.py
import pytest
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer

# Use a small model for test speed, fix model version to avoid result drift
EMBEDDING_MODEL = SentenceTransformer("all-MiniLM-L6-v2")

@pytest.fixture(scope="function")
def memory_collection():
    """Create a temporary collection per test function, insert standard memories, and return it"""
    client = chromadb.Client(Settings(anonymized_telemetry=False))
    collection = client.create_collection(
        name=f"test_memory_{id(client)}",  # Unique name to prevent leftovers
        metadata={"hnsw:space": "cosine"}  # Critical: use cosine similarity, matching production
    )

    # Standard memory set: 7 facts covering varying lengths and semantic distances
    memories = [
        {"id": "mem_1", "text": "用户名叫张明，住在北京朝阳区"},
        {"id": "mem_2", "text": "张明喜欢吃川菜，尤其是水煮鱼"},
        {"id": "mem_3", "text": "张明的车是白色特斯拉 Model 3"},
        {"id": "mem_4", "text": "用户计划下个月去杭州旅游"},
        {"id": "mem_5", "text": "张明的手机号是 138xxxx1234"},
        {"id": "mem_6", "text": "张明说他不喜欢吃香菜"},
        {"id": "mem_7", "text": "用户是一名后端工程师"},
    ]

    texts = [m["text"] for m in memories]
    ids = [m["id"] for m in memories]
    embeddings = EMBEDDING_MODEL.encode(texts).tolist()

    collection.add(
        ids=ids,
        embeddings=embeddings,
        metadatas=[{"text": t} for t in texts]  # Redundant storage for easy assertion
    )

    yield collection  # The test case receives the collection for direct querying

    # teardown: delete collection to prevent chroma in-memory leftovers
    client.delete_collection(collection.name)

Second code block: memory_retrieval.py. This is the actual retrieval function called within the agent (the system under test).

# memory_retrieval.py
class AgentMemory:
    def __init__(self, collection, embedding_model):
        self.collection = collection
        self.embedding_model = embedding_model

    def retrieve(self, query:

LangChain Memory Pitfall: A Concurrency Race Condition That Cost Me 6 Hours

BAOFUFAN — Mon, 08 Jun 2026 12:07:08 +0000

At 2:17 AM, I was jolted awake by a phone call. The ops team told me the production chatbot had suddenly developed a “split personality”—User A was watching the bot chat with User B about a loan. They took screenshots and posted them on social media. I snapped wide awake, grabbed my laptop, and started digging through the logs.

Just a few hours earlier we had shipped a “session memory” feature, built on LangChain’s ConversationBufferMemory paired with our own session_id caching strategy. Why were memories bleeding across users? I couldn’t reproduce it manually no matter how hard I tried. In the end I wrote an automated Pytest concurrency test, and only then did I flush out the deeply hidden race condition. That experience drilled one thing into my skull: If you don’t write automated concurrency tests for AI agent memory, you’re burying a landmine.

The Illusion of Static Isolation for Session Memory

Our setup was typical: a customer-service bot that keeps independent context for each visitor. Session memory lived in a server-side dict—keys were session_id, values were ConversationBufferMemory instances. The happy path looked like: request arrives → fetch the session’s memory → inject it into a ConversationChain → generate a response → update the memory. In dev and QA, clicking buttons by hand, memory isolation was flawless.

But once we hit production, when the same user sent messages in extremely rapid succession or when multiple users made concurrent requests, cross-talk appeared randomly. I narrowed down the root cause quickly: in an attempt to save memory, we had mistakenly reused the same ConversationBufferMemory instance from our cache. Every fetch from the dict returned the exact same object reference. Different coroutines were calling append on it at the same time—naturally, their messages got tangled.

Why couldn’t a manual test catch this? Manual testing thinks in pure synchronous terms; it can never manufacture the microsecond-level coroutine interleaving that happens under real load. You might do what I did at first: write a simple script that loops for i in range(100), see no problem, and declare it safe. But only by using asyncio.gather to fire two coroutines simultaneously can you stretch the race window wide enough to reproduce the bug reliably.

Designing the Test: Creating a Miniature Concurrency Crash with Pytest

To build an automated guard, you need a way to reliably reproduce the cross-talk. The toolbox was clear:

Pytest + pytest-asyncio: the foundation for async tests, giving precise control over the event loop and the ability to simulate real concurrency.
LangChain native components: ConversationChain + ConversationBufferMemory, tested directly from the production code path.
Fake LLM: instead of hitting the OpenAI API, a simple async fake that returns predetermined responses—stable and fast.
asyncio.gather: to fire two coroutines at nearly the same instant via chain.apredict, intentionally creating the interleaving.

Why not Locust or JMeter? Those load-testing tools excel at traffic simulation but are poor at making precise in-memory assertions. I needed to inspect the memory after a conversation and verify that it didn’t contain any user input from the wrong session. That requires unit-test-level freedom. Pytest fits perfectly: it can manufacture concurrency and, with assert, give you a precise postmortem.

Even more important: with automated tests I could solidify an “anti-pattern” (the shared memory instance) and let CI run that anti-pattern case on every commit, guaranteeing that no future refactor would accidentally break isolation again.

Core Implementation: From “Reproduce the Bug with Tests” to “Guard the Fix with Tests”

1. A Fake LLM Built Solely to Reproduce the Problem

This snippet solves the “no OpenAI keys, but we still need async LangChain tests” pain point. I made the LLM return fixed responses from a preset list, and it supports async.

from typing import Any, List, Optional
from langchain_core.language_models.llms import LLM
from langchain_core.callbacks import CallbackManagerForLLMRun

class FakeAsyncLLM(LLM):
    """按顺序返回预设回复的异步 LLM，专门给测试用"""
    responses: List[str]
    i: int = 0

    def _call(self, prompt: str, stop: Optional[List[str]] = None,
              run_manager: Optional[CallbackManagerForLLMRun] = None) -> str:
        resp = self.responses[self.i % len(self.responses)]
        self.i += 1
        return resp

    async def _acall(self, prompt: str, stop: Optional[List[str]] = None,
                     run_manager: Optional[CallbackManagerForLLMRun] = None) -> str:
        # 模拟异步调用，实际同步返回以保证结果可预测
        return self._call(prompt, stop, run_manager)

    @property
    def _llm_type(self) -> str:
        return "fake_async"

2. The Test That Deliberately Shares Memory—Exposing the Race Condition

This test proves that when two ConversationChain instances share a single ConversationBufferMemory, concurrent calls will inevitably cross-contaminate the memory. Run it, and you’ll see red.

import pytest
import asyncio
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

@pytest.fixture
def shared_memory():
    """共享 Memory 实例——这是我们要测试的反模式"""
    return ConversationBufferMemory()

@pytest.mark.asyncio
async def test_concurrent_corruption_with_shared_memory(shared_memory):
    llm = FakeAsyncLLM(responses=["Hello", "How can I help?", "Sure", "Goodbye"])
    # 错误：两个 chain 共用了同一个 memory 对象
    chain_a = ConversationChain(llm=llm, memory=shared_memory)
    chain_b = Conversat

We Automated LLM Memory Tests and Got 8x Efficiency

BAOFUFAN — Mon, 08 Jun 2026 01:07:09 +0000

At 2 a.m., I was jolted awake by a DingTalk message from my QA colleague: “ChatGPT’s Memory is broken again. It forgot all the preferences I taught it yesterday.” I got up, opened my laptop, logged in, switched sessions, typed prompts, compared results — half an hour gone. And this was already the third time this month. Even worse, nobody wanted to do this tedious manual work every day. We were just clicking around, and missed regressions were almost inevitable. I started thinking: could I write a Playwright script, hook it up to GitHub Actions to run automatically every day, and get an alert when something breaks? This article walks through the entire implementation.

Breaking Down the Problem: Why Memory Is Hard to Test

The “Memory” of a large language model isn’t the same as traditional conversation context. Conversation context only lives within a single session window, while Memory persists across sessions. For example, if you tell the model “I prefer replies in Chinese,” it should remember that preference even when you start a brand new chat.

The challenges in testing this feature are:

It depends on real user interaction paths. Many commercial LLMs (like ChatGPT, Claude) only allow memory configuration through the web UI. The API either lacks memory management or goes through a different backend path, so API tests cannot truly replicate the end-user experience.
Cross-session state isolation. You must completely end one session and start a new one between “setting memory” and “verifying memory.” You cannot reuse the same conversation ID. Pure API scripts struggle to precisely simulate the front-end behavior of starting a new chat.
Manual regression is expensive and unreliable. Running through a dozen test cases takes half an hour. Nobody wants to do that daily, so testing often only happens right before a release — by then the bugs have been hiding for several versions.

In one sentence: we needed a test solution that can genuinely drive a browser, switch sessions automatically, and run on a schedule every day.

Solution Design: Playwright + GitHub Actions

I compared a few technical options:

Selenium: The old guard. Its waiting mechanisms for modern web apps aren’t as elegant as Playwright’s, and you end up writing a lot of WebDriverWait. Cross-browser context isolation also feels clunky.
Direct API calls: As mentioned, the memory functionality isn’t fully exposed via APIs, and this wouldn’t test the real user path anyway.
Playwright: Natively supports multiple Browser Contexts, has built-in auto-waiting and network monitoring, and writing a script feels just like describing user actions. Plus, its storageState capability makes it easy to persist the login state. You can keep the user logged in while still starting a completely fresh conversation — perfect for simulating cross-session scenarios.

The architecture is dead simple: a Python + pytest test script using Playwright’s synchronous API to define the business flow, combined with a GitHub Actions workflow scheduled to run daily at 2:00 AM UTC (10:00 AM Beijing Time). Once the run finishes, results get pushed to Slack or the Action itself fails and triggers an alert.

The biggest win here is zero extra operational cost. No need to host your own runner — GitHub Actions’ 2,000 free minutes per month is plenty for most teams.

Core Implementation: Building a Working Test Step by Step

1. Handling Login and Memory Setup Flow

This snippet logs into the platform, sends an instruction to the LLM to set a memory, and confirms the setup was successful. Using sync_playwright is straightforward; every line reads like “click here, type that.”

# test_memory.py
import os
import time
from playwright.sync_api import sync_playwright, expect

BASE_URL = "https://chat.example.com"          # 替换为你的大模型Web地址
EMAIL = os.getenv("MODEL_EMAIL")                # 从环境变量读取，不硬编码
PASSWORD = os.getenv("MODEL_PASSWORD")

def test_memory_persists_across_sessions():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)   # CI环境下必须无头模式
        context = browser.new_context()
        page = context.new_page()

        # ---- 登录 ----
        page.goto(f"{BASE_URL}/login")
        page.fill('input[name="email"]', EMAIL)
        page.fill('input[name="password"]', PASSWORD)
        page.click('button[type="submit"]')
        # 等待登录成功后的页面元素，比如侧边栏出现
        page.wait_for_selector('nav.sidebar', timeout=10000)

        # ---- 设置记忆：告诉模型用户偏好 ----
        # 确保在一个干净的对话里设置，先点“新建聊天”
        page.click('button:has-text("New chat")')
        page.wait_for_selector('textarea[placeholder*="Send a message"]')
        msg_input = page.locator('textarea[placeholder*="Send a message"]')
        msg_input.fill("记住，我接下来所有对话都请用中文回复，且称呼我为老张。")
        page.click('button[type="submit"]')     # 发送按钮
        # 等待模型回复，且回复中需包含确认信息，避免记忆还没保存就继续
        page.wait_for_selector('text=记住了', timeout=15000)

        # ---- 至关重要的一步：等待记忆后台写入 ----
        # 官方文档不会告诉你这个延迟，但实测记忆同步需要几秒
        time.sleep(5)

        # 保存当前浏览器状态，以备后续新会话复用登录态（但我们不用它来“新建会话”）
        context.storage_state(path="/tmp/state.json")
        context.close()
        browser.close()

2. Verifying That Memory Persists Across Sessions

This step creates a completely independent Browser Context, loads the previous login state to maintain the same user identity, but starts a brand new conversation session. It asks the model “What’s my name?” and expects a reply containing “老张” or a Chinese response.

def test_memory_verif

Pytest + Redis: My 3‑Hour Battle with a Data Inconsistency Bug

BAOFUFAN — Sun, 07 Jun 2026 12:06:42 +0000

At 2 AM, my phone exploded – the in‑production user memory storage service was returning dirty data, and the customer support group chat was in chaos. I groggily pulled up the monitoring: Redis clearly had the key, but PostgreSQL held a completely different version. I tracked it down to a classic “cache update strategy” race condition. I rolled a fix and pushed it live just before dawn. Staring at the ceiling afterward, I kept thinking: if only an automated test had caught this bug before deployment, would I have had to crawl out of bed at all?

That’s when I decided to build an automated consistency‑verification framework for the memory store module using Pytest + Redis. Here’s the whole hands‑on journey, complete with the pitfalls the official docs don’t mention.

Problem Analysis: Why Memory Stores Are Prone to “Misremembering”

A memory store in many projects is essentially a wrapper around multi‑level caching and persistence: hot data lives in Redis, the full dataset lives in a database, and we juggle cache invalidation, cache‑as‑side population, and dual‑write logic. Typical use cases: user preferences, session state, feature‑vector caches. The pain points:

Dual‑write ordering: Write DB then delete cache, or delete cache then write DB? Both have windows where a reader can pick up stale values.
Expiry‑driven refresh: When a cache entry expires, a flood of concurrent requests may race to rebuild; we need to guarantee a single consistent rebuild.
Serialization asymmetry: The same data might be stored as JSON in Redis and as a binary blob in the DB. If serialize/deserialize behaviour isn’t perfectly symmetric, you get those creepy “looks the same but isn’t” bugs.

Manual testing can’t cover these concurrency windows and edge behaviours. The usual approach is to mock Redis in unit tests, but mocks fail to expose real differences in serialization, expiration policies, and Lua script execution. That’s why we need an automated testing framework that runs against a real Redis instance.

Design Rationale: Why Pytest + Real Redis

The tech choices weren’t hard:

Pytest over unittest: fixture‑driven dependency injection, parametrization, and conftest plugins dramatically lower mental overhead when writing test cases.
Real Redis over fakeredis/mock: Memory stores lean heavily on TTLs, pipelines, Lua atomic operations, and even modules like RedisJSON/RedisSearch. fakeredis doesn’t emulate them completely – and discovering that only when your green test suite leads to a production blow‑up is too expensive.
No need for a complicated docker‑compose setup: I already have Redis on my dev machine; in CI, a one‑line redis:alpine image suffices.

The architectural idea is simple: each test case receives, via fixture, an isolated Redis connection and a real in‑memory DB (SQLite). We then verify the memory store’s public interface (set, get, delete, refresh) through black‑box and grey‑box checks, focusing on three dimensions:

Single‑operation consistency: what you write is exactly what you read back.
Concurrent‑window consistency: after simulated concurrent updates, cache and DB eventually agree.
Error‑path consistency: behaviour on cache misses, DB outages, and serialization anomalies.

Core Implementation: Three Test Layers, Step by Step

1. Fixture Design for Isolation

The core problem this code solves: how to ensure each test case’s Redis and DB don’t interfere with each other, while still being easy to clean up.

# conftest.py
import pytest
import redis
import sqlite3
from memory_store import MemoryStore

@pytest.fixture
def redis_client():
    """使用 db=15 隔离测试数据，并确保每个用例启动时是干净的"""
    r = redis.Redis(host='localhost', port=6379, db=15, decode_responses=True)
    r.flushdb()  # 测试前清空，避免上次失败残留
    yield r
    r.flushdb()  # 测试后清理，不给下个用例留垃圾
    r.close()

@pytest.fixture
def db_conn():
    """SQLite 内存库，天然隔离且飞快"""
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE IF NOT EXISTS memory (key TEXT PRIMARY KEY, value TEXT)")
    yield conn
    conn.close()

@pytest.fixture
def store(redis_client, db_conn):
    """记忆存储实例，依赖注入"""
    return MemoryStore(redis=redis_client, db=db_conn)

I deliberately avoided scope="session", because we want every test to start from exactly the same state. If you want to speed things up by reusing connections, you can use scope="module" but must be extremely careful about key‑naming conflicts – a pitfall I’ll discuss later.

2. Basic Read/Write Consistency Validation

This code verifies the most fundamental contract: what you store is what you get back – data type and structure must not be silently altered.

import pytest
import json

@pytest.mark.parametrize("payload", [
    {"name": "Alice", "age": 30},
    {"nested": {"deep": [1,2,3]}},
    "simple_string",
    42,
    None
])
def test_set_get_consistency(store, payload):
    store.set("user:1", payload)
    result = store.get("user:1")
    # 确保值和类型完全一致，这里用 json.dumps 做标准化比较，避免 dict 顺序陷阱
    assert json.dumps(result, sort_keys=True) == json.dumps(payload, sort_keys=True)

def test_delete_then_get_returns_none(store):
    store.set("temp", "data")
    store.delete("temp")
    assert store.get("temp") is None

Note that I used json.dumps for a deep comparison rather than a direct ==. Dictionaries fetched from Redis can have different key ordering, and direct == would fail – but that’s not a business logic error, merely a serialization‑order difference. This foreshadows the first real pitfall.

3. Consistency Validation Under Concurrent Writes

This code simulates a real‑world scenario: two requests update the same key almost simultaneously; in the end, DB and cache must converge on the same value, not hold conflicting versions.

imp

LangChain + Chroma: Multi-turn RAG Memory and Automated Testing That Turned 2-Hour Bugs Into 5-Minute Fixes

BAOFUFAN — Sun, 07 Jun 2026 01:06:14 +0000

At 1 a.m., the customer group chat exploded: “Does your customer service bot have only a 7-second memory? I just gave it the order number, and the next turn it asks me again ‘Please provide the order number.’ I feel like I’m talking to a goldfish!”

I crawled out of bed and checked the logs. The RAG conversation memory module had lost all history after a service restart. When a user asked, “Can I refund that order I mentioned?”, the retriever couldn’t pull up the order number from earlier turns, so the answer was completely off. After fixing that bug, I realized: If every change to the memory logic relies on users to “test” it for me, the system is doomed. I had to make memory persistent and cover multi-turn scenarios with automated tests. This article is the hard-won summary of my post-mortem.

Problem Breakdown: Why Multi-turn RAG Memory Breaks So Easily

In multi-turn RAG, a user’s questions often depend on information from the previous turn, for example: “Look up order 12345” → “Can I get a refund?” The second question doesn’t contain the order number, but the LLM needs to know that “it” refers to order 12345. The traditional approach is to stuff the full conversation history into the prompt, but two pain points are obvious:

Unreliable memory: ConversationBufferMemory stores history in process memory and loses everything on restart. The user is halfway through a conversation, you deploy a new version, and all context is gone.
Fixed window loses context: ConversationBufferWindowMemory only keeps the last K turns. If the user mentioned an order number 5 turns ago and asks “Can I refund that order?” on turn 6, information outside the window is lost and cannot be retrieved.

What’s worse, in multi-turn RAG, both the history and the new question must be vectorized to search the knowledge base. If the history is stored only as raw text without semantic indexing, you simply cannot quickly recall “that order number” among hundreds of chat records. A conventional Redis cache solves persistence, but it cannot recall relevant historical snippets by semantic meaning. That’s the root cause: We need a memory storage solution that is persistent, supports vector semantic retrieval, and integrates easily into the LangChain pipeline.

Solution Design: Why Chroma Instead of Rolling Vectors on Redis

Redis + vector plugin: You’d have to maintain your own inverted indexes, expiration policies, and manual serialization—high development cost.
FAISS + manual persistence: FAISS has no built-in persistence; you must save and load index files every time, and version mismatches can cause headaches.
Qdrant / Weaviate: Very powerful, but they require running separate services, a maintenance burden that’s hard to justify for small teams.
Chroma: An embedded vector database. A single pip install chromadb gets you metadata filtering, persistence, and native LangChain integration as both a vectorstore and retriever. It ships with VectorStoreRetrieverMemory, a ready-made memory wrapper. Out of the box, you can store historical conversations in Chroma, recall the top-K most relevant history items by semantic similarity, and filter by metadata such as timestamp and user ID.

The architecture is simple: at the end of each conversation turn, concatenate the user’s question and the AI’s answer into a single document, compute its embedding, and store it in a Chroma collection with metadata like timestamp and session ID. When the next turn arrives, use the current question’s vector to retrieve the most similar N history items from Chroma, combine them with the last 3 raw messages as short-term memory, and feed the merged context to the LLM. This satisfies both “semantically similar history” and “recent time window” requirements.

For automated testing, we use pytest to write a fixed test suite: simulate 5 consecutive turns, insert them into Chroma, then verify that on turn 6 the system can recall the specific information mentioned in turn 2. Every time we modify the memory strategy, running the tests instantly tells us whether we’ve introduced a regression.

Core Implementation: Three Code Blocks to Nail Memory Storage and Testing

1. Wrapping Chroma into a Memory Class with Time Filtering

This code solves the problem of blending history storage, semantic recall, and a time window. By building on VectorStoreRetrieverMemory, we can filter by metadata after retrieval and then prepend the most recent raw messages.

import uuid
from datetime import datetime
from typing import List, Dict, Any
from langchain.memory import VectorStoreRetrieverMemory
from langchain.schema import Document
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import Chroma

class ChromaMemory:
    """把多轮对话历史存入Chroma，并用语义+时间窗口混合检索记忆"""
    def __init__(self, collection_name: str = "chat_history", k: int = 4, window_size: int = 3):
        self.embeddings = OpenAIEmbeddings()  # 统一1536维
        self.vectorstore = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
            persist_directory="./chroma_db"   # 持久化落盘
        )
        self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        self.memory = VectorStoreRetrieverMemory(retriever=self.retriever)
        self.window_size = window_size        # 始终保留最近N条原始消息
        self.recent_history: List[str] = []

    def save_context(self, user_input: str, ai_output: str) -> None:
        """每次交互后存一条文档到Chroma，同时更新最近历史窗口"""
        doc = Document(
            page_content=f"User: {user_input}\nAI: {ai_output}",
            metadata={
                "timestamp": datetime.now().isoformat(),
                "session_id": "default"
            }
        )
        self.vectorstore.add_documents([doc])
        # 维护窗口
        self.recent_history.append(f"User: {user_input}\nAI: {ai_output}")
        if len(self.recent_history) > self.window_size:
            self.recent_history.pop(0)

How Automating RAG Memory Tests with ChromaDB Quadrupled Our Bug Discovery Rate

BAOFUFAN — Sat, 06 Jun 2026 12:08:09 +0000

At 2 a.m., our ops group chat exploded — “The support bot is mixing up historical orders from the same user again.” I crawled out of bed and quickly realized that a boundary condition wasn’t covered when we changed the memory storage last week: conversations with the same session_id but different topics were mistakenly merged. I had manually run over 40 test cases that afternoon, but this exact combination slipped through. That moment, I decided: we can never rely on manual tracing to catch these gaps. Dialogue memory testing must be automated, and it must use semantic evaluation — not naive string matching.

Breaking Down the Problem

If you’ve built RAG applications, you already know conversation memory is not a simple KV cache. When a user asks “How’s that proposal going?”, the system needs to pull “that proposal” from a context that could be several turns old. The common approach is to dump chat history into a vector store like ChromaDB via LangChain’s ConversationBufferMemory, then retrieve the most relevant memory chunks by semantic search when needed.

What makes testing this memory layer so painful?

Hard to exhaust coverage: The same user intent phrased differently (“my order” vs. “what did I buy”) can produce wildly different retrieval results. Manually writing test cases simply can’t keep up.
Sky-high regression cost: Every time you tweak the embedding model or chunking strategy, you have to replay core conversation flows and manually judge whether the recalled memories “make sense” — often an entire afternoon’s work.
Vague evaluation criteria: During self-testing, developers often settle for “it feels about right.” But once users rephrase a sentence in production, things break. No quantitative metrics, just gut feelings.

The naive approach is to use in‑memory storage and assert on memory.chat_memory.messages. But that only tests exact storage, not the quality of semantic retrieval — and the latter is exactly where most RAG memory bugs breed.

Solution Design

The core idea: turn dialogue memory testing from “asserting on objects” into a fully automated semantic evaluation pipeline.

We chose LangChain’s VectorStoreRetrieverMemory backed by ChromaDB as the memory layer, and reused pytest for the test harness. Why ChromaDB? FAISS is aimed at large‑scale production and requires native C++ libraries that often break in mixed Windows/Mac developer environments. ChromaDB can be installed with a single pip install chromadb, ships with a lightweight embedding function (though the default all‑MiniLM needs internet — we mock it), and it’s extremely cheap to run in CI.

The architecture has three layers:

Test dataset generation layer – automatically produces conversation samples from predefined intent templates (greetings, order queries, mixed intents). Each sample carries an expected “memory chunk that should be recalled.”
Memory storage & retrieval layer – inside a pytest fixture, we initialize ChromaDB in ephemeral mode (or a temp directory), inject the sample conversations, and trigger retrieval via LangChain’s load_memory_variables.
Semantic evaluation layer – performs a joint assessment of vector similarity and key‑entity matching on the retrieved results, then outputs precision and recall.

The decision to abandon manual string comparison was critical here — phrases like “that proposal last time” and “the blue one we discussed earlier” may share no common words, yet they must match semantically. Regular expressions can’t cut it.

Core Implementation

1. Isolated, injectable memory fixture

This snippet solves one nuisance: quickly creating a clean, embedding‑injectable ChromaDB memory instance inside a test, without polluting the local disk. Most tutorials assume a production setup; here we enforce a temp directory and a lightweight mock embedding.

# test_memory_fixture.py
import tempfile
import shutil
from pathlib import Path
from chromadb.config import Settings
from langchain_chroma import Chroma
from langchain.embeddings import FakeEmbeddings
from langchain.memory import VectorStoreRetrieverMemory
from langchain.schema import Document
import pytest

class FakeEmbeddingsWithDim(FakeEmbeddings):
    """FakeEmbeddings 默认 size=10，但 Chroma 建索引时可能要求固定维度；
       这里强制返回固定维度向量，避免 embedding 维度 mismatch。"""
    size: int = 384   # 和 all-MiniLM-L6-v2 对齐，方便混合测试
    def embed_documents(self, texts):
        return [[0.0] * self.size for _ in texts]
    def embed_query(self, text):
        return [0.0] * self.size

@pytest.fixture
def chroma_memory_fixture():
    tmp = tempfile.mkdtemp()
    embeddings = FakeEmbeddingsWithDim(size=384)
    client_settings = Settings(
        chroma_db_impl="duckdb+parquet",
        persist_directory=tmp,
        anonymized_telemetry=False
    )
    vectorstore = Chroma(
        embedding_function=embeddings,
        client_settings=client_settings,
        collection_name="test_memory"
    )
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    memory = VectorStoreRetrieverMemory(
        retriever=retriever,
        memory_key="chat_history",
        input_key="input",
        output_key=None
    )
    yield memory
    # 清场
    shutil.rmtree(tmp, ignore_errors=True)

2. Dialogue injection & automated evaluation

This script tackles the second hard part: pumping test dialogues into memory and then evaluating retrieval quality using semantic similarity, instead of human eyeballing. Notice how we compute cosine similarity between embeddings and additionally hard‑match key entities to avoid “right vector, wrong person” mistakes.

# eval_memo

After the switch, every time we changed the embedding model or memory strategy our automated evaluation caught semantic drift immediately. Those 2 a.m. bug‑hunts never happened again.

From Mock to Real Redis: Cutting Agent Memory Test Leakage from 30% to 0

BAOFUFAN — Sat, 06 Jun 2026 01:07:23 +0000

Woken up by PagerDuty at 2 AM. The user group was on fire — our AI customer service agent suddenly lost its memory. One message confirmed the user's phone number, the next asked "How may I address you?" Checking the logs revealed that the Redis connection pool threw a ConnectionError during a network hiccup. Our supposedly bulletproof mock tests had never simulated that exception. The code simply skipped memory persistence, and all context was lost. Even scarier: the regression suite was green.

This is a textbook disaster of having mocks without real-middleware tests. We spent two weeks rebuilding the automated verification of the agent’s memory module with pytest + a real Redis instance. The result: online memory-related bugs went from a 30% leakage rate to zero. Here’s the full blueprint, code, and the sharp edges we found along the way.

1. Why mocks can't catch the real risks of an agent memory module

An agent’s memory isn’t simple key-value storage. It handles three things:

Short-term memory: the last N turns of dialogue, stored in a Redis List or ZSet with TTL-based eviction.
Long-term summaries: compressing long conversations into summaries via an LLM, saved in a Redis String or Hash with longer expiration.
Context assembly: fetching memories from Redis and stitching them into the prompt. A failure at any step makes the agent spout nonsense.

A typical mock test patches redis.Redis with unittest.mock.patch or builds a fake client. It’s easy to reach 80% coverage this way. But the real world doesn’t offer “always-succeeding storage”:

A network blip can exhaust the connection pool, raising redis.exceptions.ConnectionError. The mock version just returns True.
Serialization/deserialization isn’t always flawless — pickle protocol mismatches or complex nested objects with unpicklable lambdas never surface with mocks.
Redis eviction policies (like allkeys-lru) silently drop keys when memory gets tight. Your agent’s notepad vanishes. A mock only returns None when you explicitly set it.

The root cause: a mock simulates the Redis you imagine, not real Redis. Unit tests verify logic branches but can’t expose process boundaries, network boundaries, or data consistency issues. For a module like agent memory that depends heavily on external state, integration tests must run against real Redis. Otherwise you’ll eventually pay the technical debt in production.

2. Design choices — why we skipped the alternatives

We considered three paths for testing with real Redis:

Manually maintained dev Redis instance: Clean it manually before tests, check results manually after. Cons: easy to forget, environment drift, and nobody starts Redis for you in CI.
Embedded Redis (e.g., embedded-redis): Common in the JVM world; the Python alternatives are scarce, outdated, and incompatible with Redis 7 features we rely on.
Dockerized Redis + pytest for automatic life-cycle management: This is the right path. Spin up a Redis container before tests, tear it down after. Clean, reproducible, and one CI command covers it.

We chose pytest + redis-py + testcontainers-python (with a fallback to docker-compose). The reason: testcontainers lets you declare a Redis container right in conftest, automatically waits for the port to be ready, and destroys it when tests end — no extra scripting. Combined with a scope="session" fixture to share the container across tests, each test function uses an isolated Redis namespace (a prefix or a dedicated db number) to avoid cross-contamination while staying realistic.

Why not just run tests against a real production Redis on a dedicated db number? If you accidentally misconfigure something and flushdb isn’t blocked, you’ll have another horror story to tell.

3. Core implementation — step-by-step migration from mock to real Redis

Here are three key code blocks: the container-management fixture, the memory storage implementation, and the test cases. You should be able to drop them into your project and run.

3.1 The pytest fixture that manages a Redis container automatically

This solves the “how to get a clean Redis instance for tests” problem. We use testcontainers to launch Redis 7 and add a manual wait_for to be absolutely sure the container is ready before handing it over.

# conftest.py
import pytest
import redis
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def redis_container():
    """Session-scoped Redis container – started once per test session"""
    container = RedisContainer("redis:7-alpine")
    container.with_exposed_ports(6379)
    container.start()
    # Ensure Redis is truly ready (the built-in wait strategy sometimes isn't enough)
    client = redis.Redis(
        host=container.get_container_host_ip(),
        port=container.get_exposed_port(6379),
    )
    client.ping()  # Will fail fast here if not ready
    yield container
    container.stop()

@pytest.fixture
def redis_client(redis_container):
    """Per-test Redis client with an isolated DB to avoid interference"""
    client = redis.Redis(
        host=redis_container.get_container_host_ip(),
        port=redis_container.get_exposed_port(6379),
        db=0,
        decode_responses=True,  # avoid manual .decode()
    )
    client.flushdb()  # Start clean
    yield client
    client.close()

Why decode_responses=True? The agent’s memory module mostly stores natural language text. Returning bytes forces a .decode() everywhere, adding noise that buries the actual assertions.

3.2 The agent memory storage module under test

This is a simplified version of production code, living in memory_store.py. It stores session context in a Redis Hash with a TTL. We intentionally kept a serialization path that is easy to break in production (using json instead of pickle, but validating data integrity).

# memory_store.py
import json
import redis
from typing import Dict, Optional
import logging

logger = logging.getLogger(__name__)

class AgentMemoryStore:
    def __init__(self, redis_client: redis.Redis, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl

    def save_context(self, session_id: str, context: Dict) -> bool:
        try:
            key = f"agent:session:{session_id}"
            serialized = json.dumps(context, default=str)
            self.redis.setex(key, self.ttl, serialized)
            return True
        except (redis.exceptions.ConnectionError, TypeError) as e:
            logger.error(f"Failed to save context for {session_id}: {e}")
            return False

    def load_context(self, session_id: str) -> Optional[Dict]:
        try:
            key = f"agent:session:{session_id}"
            data = self.redis.get(key)
            if data is None:
                return None
            return json.loads(data)
        except (redis.exceptions.ConnectionError, json.JSONDecodeError) as e:
            logger.error(f"Failed to load context for {session_id}: {e}")
            return None

3.3 Test cases against real Redis — including failure scenarios

This tests the happy path and the scenarios that mock tests never touch: connection failures, serialization errors, and Redis eviction behavior.

# test_memory_store_real.py
import json
import pytest
from unittest.mock import patch
import redis
from memory_store import AgentMemoryStore

def test_save_and_load_context_success(redis_client):
    store = AgentMemoryStore(redis_client)
    context = {"phone": "13800138000", "intent": "return_order"}
    assert store.save_context("session_1", context)
    loaded = store.load_context("session_1")
    assert loaded == context

def test_context_not_found(redis_client):
    store = AgentMemoryStore(redis_client)
    assert store.load_context("nonexistent_session") is None

def test_connection_error_graceful_failure(redis_client, mocker):
    """Simulate a ConnectionError — the store must handle it gracefully."""
    store = AgentMemoryStore(redis_client)
    # Force connection failure on setex
    mocker.patch.object(redis_client, 'setex', side_effect=redis.exceptions.ConnectionError("boom"))
    result = store.save_context("session_err", {"a": 1})
    assert result is False

def test_serialization_error_handling(redis_client):
    """What happens when we try to save an unserializable object?"""
    store = AgentMemoryStore(redis_client)
    bad_context = {"fn": lambda x: x}  # lambda not serializable by json
    result = store.save_context("session_bad", bad_context)
    # Should gracefully fail, not throw
    assert result is False

def test_eviction_behavior(redis_client):
    """Set a tiny TTL and wait — then check that data disappears."""
    store = AgentMemoryStore(redis_client, ttl=2)
    store.save_context("short_lived", {"value": "ephemeral"})
    import time
    time.sleep(3)
    assert store.load_context("short_lived") is None

These tests run against the exact Redis version and memory constraints that match production. No more “it works on my fake client.”

4. CI integration — one-line command + environment isolation

Running real Redis in CI often raises concerns about Docker/socket access. We added the CI configuration to ensure smooth execution:

# .github/workflows/agent_memory_tests.yml (excerpt)
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      # Optional fallback if testcontainers can't access Docker socket
      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements-test.txt
      - name: Run memory integration tests
        run: pytest tests/ --real-redis

We added the --real-redis custom marker so that unit tests relying on mocks and integration tests requiring the container can coexist. The marker also skips these tests automatically when no Redis is available (e.g., a local dev environment without Docker).

# conftest.py (continued)
def pytest_addoption(parser):
    parser.addoption("--real-redis", action="store_true", default=False,
                     help="run tests against real Redis")

def pytest_configure(config):
    config.addinivalue_line("markers", "real_redis: mark test as requiring real Redis")

def pytest_collection_modifyitems(config, items):
    if not config.getoption("--real-redis"):
        skip_real = pytest.mark.skip(reason="need --real-redis option to run")
        for item in items:
            if "real_redis" in item.keywords:
                item.add_marker(skip_real)

This gives us two layers of safety: CI always runs the real-Redis suite, while local developers can quickly iterate without Docker when they choose.

5. The payoff — measurable metrics

After rolling out real Redis tests, we tracked the agent memory module’s bug escape rate for three months.

Before: 30% of memory-related production incidents originated from behavior that passed the mock test suite. Typical causes: connection pool exhaustion, key eviction, serialization errors.
After: zero production incidents caused by the same class of problems. The team also gained confidence to refactor memory storage, including a migration from JSON to MessagePack, because the tests caught the edge cases immediately.

The stability improvement went beyond metrics: the on-call team stopped being woken up at 2 AM for memory loss bugs.

6. Lessons & watch out

Throughout the migration, we hit several pitfalls worth sharing:

Don’t share the same Redis DB across test functions — even with flushdb(), concurrent test runs (pytest-xdist) will collide. Use unique key prefixes or separate DBs.
Testcontainers startup time can be sneaky in large suites. We pinned the Redis image version and used scope="session" to keep total test time under 10 seconds.
Connection pool exhaustion is real. In one test we simulated 1000 rapid-fire saves, which exhausted the default connection pool. Setting max_connections higher and adding close logic in teardown fixed it — and caught a production leak we didn’t know we had.
Keep your test Redis version aligned with production. We originally used redis:latest and missed behavior changes when production was still on Redis 6. Now we explicitly pin redis:7-alpine.

Swapping mocks for real Redis isn’t just about testing — it’s about trust. For stateful agents, the distance between “tests pass” and “it works in production” is measured by how closely your test environment mirrors reality. In our case, that mirror was a Docker container, and it saved our sleep.