DEV Community: BAOFUFAN

From 2 Hours to 3 Minutes: Automating LLM Memory Testing with Playwright + LangChain

BAOFUFAN — Wed, 12 Aug 2026 01:04:06 +0000

At 2 AM, my testing colleague’s WeChat voice call jolted me awake: “The conversation memory bug that drops context is back. I ran a full regression and still couldn’t capture a stable reproduction path.” Rubbing my eyes, I opened my laptop, manually launched five browser windows, typed “My name is Ming, remember that,” refreshed the page, asked “What’s my name?” and ticked off results against screenshots. That night I spent nearly two hours just to check whether a memory persistence feature had been broken again by a backend change. The thought that burned in my mind: Manually verifying memory in a browser is like mining with a hand shovel – painfully inefficient and borderline absurd.

Then I slammed the desk and wired up Playwright with LangChain, making the browser automatically chat, refresh, and ask questions, then letting an LLM decide “did it remember or not.” Regression time dropped from 2 hours to 3 minutes, while we managed to cover 30+ edge cases in the same run. This post shares the idea and complete code behind this “automated memory bodyguard.”

Breaking It Down: Why Manual Memory Persistence Testing Is a Nightmare

The essence of “conversation memory persistence” in LLM applications is this: after the frontend is closed or refreshed, the session history and knowledge base remain alive. The standard verification path is:

Open the web page, chat with the bot for a few rounds, inject personal info (name, preferences, to-dos).
Close or refresh the page.
Send another message and visually check whether the response references the earlier information.

Running this manually takes at least 2–3 minutes per cycle. If the system under test uses a deeply customized LangChain Memory, a custom Redis store, or a conversation flow with complex routing, regression needs to cover 20+ conversation shapes (multi-turn, single-turn, mid-disconnect, memory after token trimming, etc.). Humans simply can’t keep up, and “eyeballing” whether the bot remembered is terribly unreliable – sometimes the bot just politely says “Hello,” and you have no idea if that reply came from memory or was randomly generated.

The usual automation approaches are threefold:

Direct API calls: bypasses the browser and can’t verify whether frontend cookies / localStorage / session data synced correctly to the backend.
Selenium + fixed assertions: asserting “response contains ‘Ming’” misses tons of natural variations – the model might say “Your name is Ming, right?” Fixed pattern matching is practically blind.
Fully manual: slow, unrepeatable, error-prone.

This job requires a semantic judge that truly understands meaning, plus a robot that can drive a browser.

Solution Design: Playwright Drives the Browser, LangChain Judges

I raided my toolbox:

Why Playwright instead of Selenium?

Playwright’s auto-wait is far friendlier to modern SPAs – no more false timeouts when an element is in the DOM but not yet rendered. It also effortlessly manages multiple browser contexts, perfect for simulating “refresh and re-enter.” And its Python API mixes seamlessly with LangChain.

Why LangChain for assertions?

Asserting “memory retained or not” is fundamentally a semantic judgment task. Instead of writing a hundred regular expressions, just hand it to ChatOpenAI (or any LLM) and let it respond “yes/no.” LangChain’s ChatModels and Prompt templates let you quickly build a reusable evaluator, and you can easily swap models for cost control.

Alternatives I deliberately skipped:

Pure LLM API to compare texts: no coverage of the frontend link.
Browser extension recording & playback: no intelligent assertions; any minor DOM change breaks everything.
Robot Framework + custom keywords: high extension overhead; just using Python directly is more flexible.

The final architecture is dead simple: Playwright script → browser simulates conversation & refresh → grabs the last reply → LangChain judge => pass/fail, then collect results into a CLI report.

Core Implementation: Three Building Blocks That Form the Memory Bodyguard

The runnable Python code below is split into three functions, tackling simulated conversation, triggering the memory check flow, and intelligent assertions respectively.

1. Playwright Simulated Conversation – Chat the Bot into Position

This code automatically inputs messages, clicks send, and waits for the assistant’s reply to appear. All selectors use common semantic placeholders; tweak them for your project and you’re good to go.

from playwright.sync_api import sync_playwright

def start_session_and_chat(page, initial_messages: list[str]):
    """
    page is a Playwright Page object already on the chat page,
    initial_messages is a list of strings to send in order.
    """
    for msg in initial_messages:
        # Wait for input field to be interactive (SPA loading friendly)
        page.wait_for_selector("textarea[placeholder='输入消息...']", state="visible")
        page.fill("textarea[placeholder='输入消息...']", msg)
        page.click("button:has-text('发送')")
        # Wait for at least one assistant message to appear before continuing, to avoid overlap
        page.wait_for_selector(".assistant-message:last-child", timeout=10000)

# Usage example: implant memory
with sync_playwright() as pw:
    browser = pw.chromium.launch()
    page = browser.new_page()
    page.goto("http://localhost:3000/chat")
    start_session_and_chat(page, ["你好", "我叫小明，我最喜欢蓝色的跑鞋"])

The wait_for_selector and .assistant-message:last-child are key for handling asynchronous streaming replies. If your UI uses SSE, you might need additional DOM change listeners, but usually waiting for the latest message node is enough.

2. Memory Persistence Verification Flow – “Quiz” It after Refresh

We deliberately inject a set of personal information, then refresh the page to simulate a disconnect, and ask a targeted question.


python
def verify_memory_after_reload(page, user_name: str, memory_keyword: str) -> bool:
    """
    Implant name and preference into the bot, refresh, then ask whether it remembers.
    user_name        : user name, used for questions like "What's my name?"
    memory_keyword   : expected keyword in the response (e.g., the name).
    """
    # Implant phase
    start_session_and_chat(page, [
        f"记住：我

Automating LLM Agent Memory Testing with Playwright: 10x Efficiency Boost

BAOFUFAN — Tue, 11 Aug 2026 01:04:14 +0000

A while ago, our team took on a task: regression testing for an enterprise AI assistant, with a focus on its “memory” and context retention. In round 1 you tell it “My name is Zhang San and I work at ByteDance.” By round 5, when you ask “What’s my name? Which company do I work for?”, it must answer correctly. At first, three of us were doing this manually. We’d run at most 30 test cases a day, frequently miss subtle regressions when our attention drifted, and constantly worry about edge cases. Frustrated, I automated the entire workflow with Playwright — now a single machine easily runs 300+ cases a day, and I no longer dread being woken up in the middle of the night to “put out fires”. This post is a recap of that overhaul, and all the code is ready to run.

Breaking down the problem

Our test target is an in‑house LLM Agent web app. Users access it through a browser, can have multi‑turn conversations, and the agent is expected to have “long‑term memory” — remembering user preferences, facts, and recalling them even after the conversation shifts to different topics.

A typical manual test flow goes like this: open the page → type “My name is Zhang San and I work at ByteDance” → chat about unrelated stuff → suddenly ask “What’s my name?” → manually judge if the answer is correct. A few things drive you crazy:

Long context windows — a single session often spans a dozen turns. By the later rounds, your brain is fried.
Blurred forgetting boundaries — LLM “forgetting” is rarely a hard drop; it might misremember a single character (“Zhang Shan”), which is painfully easy to miss with the naked eye.
Combinatorial explosion of scenarios — name, company, preferences, order history… the permutations quickly become impossible to cover manually.
Interleaved tool calls — the agent supports in‑conversation interruptions like “Book a flight for me”. Crafting that test data by hand is a huge pain.

Classic API‑level testing (calling /chat directly) has two fatal flaws: it skips the front‑end message‑history stitching logic (our app has its own message merging strategy), so you never test the full chain; and the API often returns streaming chunks, making assertions tricky, and you can’t reproduce the final rendered text the user actually sees. End‑to‑end browser automation is the only reliable way.

Design decisions

When picking a tool, I ruled out Selenium immediately. Not because it’s bad — it’s just that our Agent’s front‑end relies heavily on WebSockets pushing streaming tokens. Waiting for dynamic content with Selenium feels clunky; you end up writing piles of WebDriverWait. Playwright natively supports waiting for network idle, text changes in elements, and can even intercept WebSocket frames — plus its Python async API is butter smooth.

The architecture boils down to three simple rules:

Tests as configuration – each memory test becomes a “dialogue sequence + final assertion”, managed in YAML, completely decoupled from code.
One browser context per session – each test case uses an isolated browser context, keeping localStorage/sessionStorage separate, avoiding cross‑contamination and allowing parallel execution.
Smart waiting – no hardcoded sleep(2). Instead, we use Playwright’s expect(page.locator(...)).to_contain_text(...) to wait for the AI’s complete response. More stable than any fixed delay.

Why not use an existing AI testing framework (like DeepEval’s E2E part)? Most of them are still API‑oriented, offer weak support for front‑end rendering and multi‑step interactions, and add extra dependencies that the team has to maintain. Wrapping Playwright ourselves took about 200 lines of code, and we have far more control.

Core implementation

Let’s get to the good stuff. The three code blocks below make up a minimal, runnable memory test script.

First block: page interaction helper. It solves “how to reliably send a message and get back the AI’s complete response”.

import asyncio
from playwright.async_api import async_playwright, expect

async def send_and_get_reply(page, message: str, timeout: int = 15000) -> str:
    """
    向聊天框发送 message，等待 AI 返回完整响应文本
    特别处理流式输出：等待"停止生成"按钮消失，表明回复结束
    """
    # 定位输入框并填入文本，通常 chatbot 都有一个 textarea
    input_box = page.locator('textarea[placeholder*="输入消息"]')
    await input_box.fill(message)

    # 点击发送按钮
    send_btn = page.locator('button:has-text("发送")')
    await send_btn.click()

    # 关键：等待流式输出完成。我们前端在生成时会显示一个"停止"按钮，
    # 生成结束后按钮消失。也可以用其他页面标志，比如光标出现。
    stop_btn = page.locator('button:has-text("停止生成")')
    await expect(stop_btn).to_be_hidden(timeout=timeout)  # 等15秒

    # 获取最后一条 AI 消息的完整文本
    # 假设消息列表最后一条 class="assistant-message"
    last_message = page.locator('.assistant-message').last
    await expect(last_message).to_be_visible()
    return await last_message.inner_text()

Second block: memory test case runner. It solves “how to execute a multi‑turn conversation and run assertions”.

async def run_memory_test(browser, case: dict):
    """
    case 格式:
    {
      "name": "基础记忆-姓名公司",
      "conversations": [
        {"role": "user", "content": "我叫张三，我在字节跳动工作"},
        {"role": "assistant", "ignore": True},  # 不验证
        {"role": "user", "content": "今天天气真好"},
        {"role": "assistant", "ignore": True},
        {"role": "user", "content": "我叫什么？我在哪工作？"},
      ],
      "expected": ["张三", "字节跳动"]  # 最后一条回复必须同时包含这些词
    }
    """
    context = await browser.new_context()
    page =

Third block: parallel execution entry point. Kick off all the test cases at once. (I’ll leave that as an exercise — it’s really just an asyncio.gather loop over your YAML‑defined cases using the two helpers above.)

With these pieces, we turned a tedious, error‑prone process into something reliable and massively scalable. The team now adds new memory scenarios by simply appending YAML, and the entire suite runs multiple times a day in CI. If you’re testing LLM‑powered chat interfaces, give Playwright end‑to‑end automation a try — your sleep schedule will thank you.

How I Spent 6 Hours Debugging LangChain Memory’s Silent Message Drop

BAOFUFAN — Mon, 10 Aug 2026 12:04:49 +0000

At 2 a.m. an alert yanked me out of bed. I pulled up the logs and my blood pressure went through the roof: a user asked “Where is my order?” and the AI agent replied “What’s your favorite color?” I had trained this agent myself, run over 20 rounds of dialogue before deployment, and everything looked flawless. Right then I knew—it had to be the memory system again. Six hours later I finally pinpointed the root cause, and it made me want to throw my keyboard out the window: LangChain’s Memory silently drops messages under certain edge cases, and manual testing never covered them. In this post I’ll share the complete memory-consistency testing setup I built with Pytest—along with the debugging journey—so you can save those six hours.

Problem Breakdown: Why “20 rounds worked” still exploded

Our scenario is an order-inquiry agent. Users provide their phone number and order ID over multiple turns, the agent calls a lookup tool, and the conversation might branch into phone-number changes, coupon inquiries, and the like. We used LangChain’s ConversationBufferMemory to store history and injected it into the prompt via {chat_history}.

During the first week in production we caught occasional reports that “the agent seems to forget who I am,” but reproducing it was tough. That night we finally pulled the full-chain logs and spotted a pattern:

When a tool returned really long content (e.g., a hundreds-of-characters coupon list), by the time the next user message arrived, only the last two messages survived in memory. The phone number and order ID from earlier rounds simply evaporated. Deprived of context, the agent would blurt out a random fallback reply.

The root cause? We had set k=3 on the Memory to control token usage, meaning “keep only the last 3 turns of interaction.” But in ConversationBufferWindowMemory with return_messages=True, when the total token length exceeded the model limit, LangChain would perform a secondary trim—at a higher priority—that completely ignored the k value, chopping history arbitrarily. This behavior wasn’t documented anywhere.

Why did our regular tests miss it? We used to spin up a real LLM, run 20-scripted conversations, and spot-check outputs manually. Slow, expensive, and because of LLM randomness we couldn’t write assertions. Those tests could only catch “completely broken” bugs, never the silent memory corruption.

Solution Design: Make Pytest the Black-Box Judge of Memory

The core idea is dead simple: mock all LLM calls and test only the Memory read/write behavior inside the Agent flow. Tests execute in milliseconds, cost zero, and let you parametrize the heck out of edge cases.

Technical choices:

Pytest — parametrization and fixtures are far more elegant than unittest. @pytest.mark.parametrize lets a single function run dozens of cases.
unittest.mock — patch ChatOpenAI and similar LLM calls to make the agent’s “thinking” fully deterministic.
Why not LangSmith or other platforms? — They’re great, but require internet access, introduce latency, and aren’t suitable for fast local TDD red-green loops. Budget was tight, too.
Why not integration tests? — Integration tests ensure components interact correctly, but memory-logic edge cases need to be violently enumerated with unit tests. They complement each other.

Architecture-wise, we designed a set of fixtures: mock_llm → inject a custom reply sequence → create AgentExecutor or ConversationChain → run multiple turns → assert the contents of memory.chat_memory.messages. This puts the memory component in the interrogation room all by itself.

Core Implementation: Three Steps to a Memory-Consistency Test

Step 1: Build an obedient Mock LLM

This snippet solves the “uncontrollable LLM” problem. The mock returns preset replies one by one, simulating an AI that always follows the script.

from unittest.mock import MagicMock, patch
from langchain.schema import AIMessage, HumanMessage
from langchain.chat_models import ChatOpenAI
import pytest

def create_mock_llm(responses):
    """创建一个顺序返回指定回复的mock LLM"""
    mock = MagicMock(spec=ChatOpenAI)
    # 让mock的invoke方法依次返回AIMessage
    mock.invoke.side_effect = [AIMessage(content=resp) for resp in responses]
    mock.return_value = mock  # 支持链式调用
    return mock

Step 2: Wire up a Chain with Memory, run multiple turns

This code verifies that under normal flow, memory accumulates correctly. A pytest fixture initializes the chain; the test function feeds three user messages back to back, then asserts the correct number of messages in memory.

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

@pytest.fixture
def chain_with_memory():
    # 注意：return_messages=True 让memory存储消息对象，便于断言
    memory = ConversationBufferMemory(return_messages=True)
    mock_llm = create_mock_llm(["你好！", "你叫什么？", "我叫小智"])
    chain = ConversationChain(llm=mock_llm, memory=memory)
    return chain

def test_memory_accumulates_correctly(chain_with_memory):
    chain = chain_with_memory
    chain.predict(input="你好")
    chain.predict(input="我叫小明")
    chain.predict(input="帮我查订单")

    messages = chain.memory.chat_memory.messages
    # 期望3轮对话 = 6条消息（Human/AI交替）
    assert len(messages) == 6
    # 第一条用户消息应始终保留
    assert messages[0].content == "你好"
    # 最后一条AI消息应为我预设的"我叫小智"
    assert messages[-1].content == "我叫小智"

Step 3: Reproduce the bug that cost me 6 hours

This test reveals the boundary vulnerability in k truncation. We deliberately set k=2 (keep only the last 2 rounds) and run three rounds of dialogue. If early information still exists, the configuration failed to take effect; if the third user message is present but the second is missing, it indicates the truncation logic has a bug—exactly the silent corruption we saw in production. By writing this as a fast unit test, we caught the exact scenario that took me all night to find in logs, and we can now run it in seconds on every commit.

I Let an LLM Write My API Tests—Coverage Jumped from 41% to 98% and Caught a Hidden Bug

BAOFUFAN — Mon, 10 Aug 2026 01:04:16 +0000

Liquid syntax error: Unknown tag 'endraw'

Automated AI Memory Testing with Pytest & Redis: 10x Faster Validation and 3 Hidden Bugs Squashed

BAOFUFAN — Sun, 09 Aug 2026 12:05:12 +0000

Last Friday afternoon, I was sipping coffee and getting ready to wrap up the week when a message from QA landed in our group chat: “Users are complaining the AI can’t remember earlier conversations – check if the memory disappeared again.” My heart sank a little. I opened the logs – the Redis key still contained the conversation history, but the order was completely scrambled, with the latest message sitting at the front. What annoyed me even more was that this was already the third time this month we’d been bitten by an “invisible” memory storage bug.

Every time we tried to reproduce the issue, it involved spinning up three terminals, manually sending messages like “Hi, my name is Alice” and “What’s my name?” to the AI, then visually comparing the responses. “Manual validation” sounds generous – it was basically crossing our fingers.

I decided to put an end to the guesswork by building a Pytest + Redis test suite. The goal wasn’t just to check whether memories were saved, but to verify order, TTL expiry, and consistency under concurrent writes. After setting it up, a manual regression cycle that used to take 30 minutes now finishes in 3 minutes for 30+ test cases, and I caught three logical bugs that even code review had missed. This article reproduces the entire journey – all the code is ready to run.

Breaking down the problem: why your AI memory tests have gaps

Here’s the typical setup: an AI chat system stores every user–assistant turn in Redis, probably under a key like chat:memory:{session_id}. The value is a JSON array containing message objects in chronological order. When the user sends a new message, the backend fetches the history from Redis, prepends it to the prompt, and feeds everything to the LLM. Simple on paper, but riddled with traps in practice:

Ordering issues – under concurrent requests, multiple writes can interleave, messing up the array order. A user asks “What’s the weather?” and then “What’s my name?”; the AI answers the name first and the weather later because in the stored memory the “name” message jumped ahead of the “weather” one.
TTL boundaries – the conversation memory is set to expire after 30 minutes, but if a new message is written at 29:59 and the EXPIRE reset happens at the wrong moment, the memory vanishes in the middle of the chat.
Serialization inconsistencies – you save with json.dumps and load with json.loads, but when multiple processes or services are involved, datetime objects can serialize differently (one process includes timezone info, another doesn’t). This causes memory comparisons to fail or even breaks the conversation flow with an error.

Manual testing typically looks like this: open a Redis client, insert a record by hand, call the API with curl, then reconnect to check if the data is still “correct”. This approach has two fatal flaws: when there are 50 turns in a conversation, no human can reliably verify the order and content of all messages. Moreover, TTL-dependent test cases require waiting 30 minutes – nobody actually waits that long in a test environment, so that logic is almost never exercised. And concurrency scenarios? Nearly impossible to simulate by hand.

What we really need is a test setup that can generate data, assert correctness automatically, fast‑forward time, and run in parallel.

Design decisions: why we test against a real Redis instance instead of mocking it

When testing external dependencies, people usually pick one of two extremes: mock everything (replace all Redis calls with fake objects) or connect directly to a remote test Redis. I tried mocking and hit several walls:

Advanced features of redis-py like pipelines, Lua scripts, and blocking commands are extremely hard to fake realistically. For example, the timeout behavior of brpop in a mock is nothing like the real thing.
The very things we need to test are Redis’s real behaviors: TTL expiry, type conversions, and reconnection after a disconnect. Mocking would turn the tests into “testing my own mock code”, which is pointless.

Using a shared remote Redis instance also has downsides: parallel tests pollute each other’s keys, requiring constant cleanup, and network latency can cause flakiness. The sweet spot is testcontainers – we spin up a temporary Redis container that behaves exactly like a production instance but is fully isolated, and it’s destroyed automatically when the tests finish. Combined with Pytest fixtures and freezegun to freeze time, we can compress hours of time‑dependent tests into a few seconds.

The test strategy is layered into three levels:

Unit tests – verify the CRUD operations and serialization logic of the memory handler.
Integration tests – exercise the full request path (FastAPI + Redis).
Consistency tests – validate memory order and completeness under concurrent writes.

Core implementation: solving a real pain point at every step

Let’s build the test suite step by step. All the code includes imports so you can drop it into a test_memory.py and run it.

1. Redis container fixture – solving environmental pollution

This snippet eliminates leftover data and instability issues. Each test class gets its own Redis container.

# conftest.py or top of test_memory.py
import pytest
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def redis_container():
    # session-scoped to boost speed – starts once
    with RedisContainer("redis:7-alpine") as container:
        container.start()
        yield container

@pytest.fixture
def redis_client(redis_container):
    import redis
    client = redis.Redis(
        host=redis_container.get_container_host_ip(),
        port=redis_container.get_exposed_port(6379),
        decode_responses=True  # auto-decode for easy assertions
    )
    yield client
    client.flushall()  # clean after each test to avoid cross-contamination

2. Storing and reading memory – verifying basic correctness

We need to confirm that after saving conversation history, the retrieved list has the exact same length, content, and order, without any weird escaping.

import json
from datetime import datetime, timezone

class MemoryService:
    """实际业务代码（简化版），你项目里可能更复杂"""
    def __init__(self, redis_client, ttl=1800):
        self.redis = redis_client
        self.ttl = ttl

    def append_message(self, session_id: str, role: str, content: str):
        key = f"chat:memory:{session_id}"
        # 取出历史
        raw = self.red

Debugging Playwright LocalStorage Persistence: A 3-Hour Puzzle

BAOFUFAN — Sun, 09 Aug 2026 01:04:13 +0000

It was 2 a.m. when the CI alert fired. A front-end login test case had suddenly started failing — it was perfectly fine yesterday. Bleary-eyed, I opened the Allure report and saw the failure screenshot: login succeeded, the token was saved into LocalStorage, but after a page reload, the token was gone and the user got kicked back to the login page.

That makes no sense, I thought. The front-end colleague swore that localStorage is persistent. How could it disappear after a refresh? Even weirder, the flow worked flawlessly when I tested it manually in the browser, yet the automation script reproduced the issue consistently. At that moment I knew: I wasn't going to bed anytime soon.

Breaking Down the Problem

Our scenario is quite typical: after a user logs in, the front end writes a JWT via localStorage.setItem('token', xxx) and then navigates to the home page. To verify that "the user stays logged in after a page refresh," we wrote the following Playwright test:

Simulate login → assert that a token exists in localStorage
page.reload()
Assert again that the token is still there and that the page doesn’t redirect to /login

Manual testing worked perfectly, yet the automation failed most of the time: after reload, localStorage was cleared. I initially suspected the front end was wiping it during a reload. I combed through the source code — no cleanup logic. Then I wondered if Playwright’s page.reload behaved oddly, so I swapped it with page.goto — still broken.

The root cause gradually surfaced: localStorage persistence is tied to the browser storage directory, and we were launching Playwright with the default launch parameters, without specifying a persistent directory. This means that every browser.new_context() gave us a temporary profile. Data wouldn’t get lost when you call page.reload inside the same context. But our test framework used @pytest.fixture(scope="function") to create a brand-new context for each test case. In cross-context scenarios — such as recreating a context to simulate closing and reopening the browser — localStorage was gone.

To make matters worse, CI container environments naturally create a fresh browser instance per job, and the behavior of new_context can differ from a developer’s local machine (where caching and recycling mechanisms may kick in). This led to tests that occasionally passed locally but always failed in CI. Here lies the reason why "the conventional approach doesn't work": directly asserting on localStorage only verifies the data living inside the current session’s memory, not actual "disk persistence."

Solution Design

I outlined several possible verification paths:

Option A: Read localStorage directly via page.evaluate – Verifies the current session but cannot guarantee that the data has been flushed to disk and can survive a “close browser and reopen” scenario. Pass.
Option B: Selenium’s local_storage interface – Selenium 4 supports it, but it requires extra configuration for the Chrome user data directory. The maintenance cost is high, and since we’ve already fully migrated to Playwright, there’s no reason to go backwards.
Option C: Playwright’s storageState snapshot + reload – Playwright provides context.storage_state() to export the entire storage state (cookies + LocalStorage and other origin storage), and allows you to hydrate it back via the storage_state parameter when creating a new context. This precisely simulates the data recovery process of “disk persistence → next browser session.”

So the final choice: Combine Playwright + pytest with the storageState mechanism, and design a fixture to truly verify localStorage persistence. This not only tests the front-end logic but also incidentally validates the correctness of storage recovery.

Core Implementation

1. A reusable helper: save and restore storage state

This code addresses the problem of “accurately simulating closing and reopening the browser while preserving localStorage.”

import json
import os
from pathlib import Path
from playwright.sync_api import BrowserContext

def save_and_reload_storage(context: BrowserContext, filepath: str = "state.json") -> BrowserContext:
    """
    Serialize the current context's storageState to a file,
    then close the original context and create a brand new one
    using the saved state. This simulates closing the browser
    and opening it again.
    """
    # Save the complete storage state (cookies + localStorage)
    state = context.storage_state()
    Path(filepath).write_text(json.dumps(state), encoding="utf-8")

    # Close the original context, releasing resources
    page = context.pages[0] if context.pages else None
    if page:
        page.close()
    context.close()

    # Rebuild context from the saved state
    browser = context.browser
    # Important: use the same browser instance, otherwise a new variable is introduced
    new_context = browser.new_context(storage_state=filepath)
    return new_context

Key points: The storage_state parameter must point to a file path (Playwright reads it internally); you can also pass a dict, but a file is easier to audit in logs. Also, be sure to close the old context first before creating the new one to guarantee a fully fresh session simulation.

2. pytest fixture: providing independent context and verification tools for tests

This code addresses “managing the Playwright lifecycle uniformly inside pytest and offering persistent verification capabilities.”

import pytest
from playwright.sync_api import sync_playwright, Browser, BrowserContext, Page

@pytest.fixture(scope="function")
def persisted_context() -> BrowserContext:
    """
    Each test case receives a BrowserContext equipped with
    persistent verification capabilities.
    Inside the test, you can simulate reloading state via reload_state()
    to model

Pytest + Chroma: The 6‑Hour Bug That Erased AI’s Memory

BAOFUFAN — Sat, 08 Aug 2026 12:04:37 +0000

At 2:30 AM, I was jolted awake by a stream of PagerDuty alerts. Users were complaining that our “AI memory assistant” kept asking “Hello, who am I?” in a loop—despite having chatted the day before about a cat named Pudding. I groggily opened the Chroma console. All vector records were still there. I reran the Pytest suite. Twenty‑seven tests glowed green like a spring meadow. Data intact, tests passing—yet the AI had amnesia. That night I stared at Chroma’s source code and logs from 2:30 to 8:30 AM, finally smoking out a ghost hiding inside a testing blind spot: a boundary case that only real users triggered, rendering our automated verification completely useless.

When your tests become a placebo

Our architecture was straightforward: LangChain’s ConversationBufferMemory stored conversation summaries in Chroma. At query time, we used semantic search with the current user input to fetch relevant memories and inject them into the LLM. To guarantee storage reliability, we wrote a full Pytest suite: insert a memory, query with the exact same text, assert that the returned document ID matched. It looked bulletproof—until users rephrased their sentences in production and the memories never came back.

The root cause quickly surfaced: a missing distance threshold on query results. Chroma returns results by top_k and will happily give you garbage when similarity is extremely low. Because our test always queried with the identical insertion text, the cosine similarity was always 1.0, the result was always first, and the assertion always passed. In reality, users don’t repeat themselves like a broken record. They reword, add filler words, make typos. Once the vector distance between the rephrased query and the stored memory widened to 0.6 or lower, Chroma still returned something—possibly completely irrelevant memory fragments with a miserable score of 0.4. We never validated the score, so the LLM swallowed that noise as real memory, producing incoherent replies that looked like amnesia.

Standard solutions—like LangChain’s built‑in Memory test utilities—failed here. LangChain’s testing wrappers are too opaque: you can only check whether a final string appears, but you cannot control the vector search distance threshold, the similarity metric, or run parametrized tests with borderline text variations. To truly guard Chroma’s quality, we had to tear into the testing logic ourselves at the Pytest level.

Turning Pytest into the vector‑DB coroner

We ditched all high‑level wrappers and drove the chromadb client directly, building three dimensions of integrity validation inside Pytest:

Content correctness: don’t just check IDs; assert the returned document text is genuinely relevant.
Distance reasonableness: force an assertion that distance (or score) exceeds our business threshold. If it doesn’t, the retrieval is considered a failure even if something was returned.
Semantic robustness: hammer the retrieval with dozens of semantically similar queries that use different wording—misspellings, synonyms, colloquial expressions—to guarantee the memory’s “fault tolerance.”

Why not use Chroma’s own integration tests? They’re too coarse, and the default similarity metric even changed between versions (pitfall #1, for another day). We chose Pytest for brutally simple reasons: great ecosystem, effortless parametrization, and fixture lifecycle management that isolates Chroma collections cleanly. We weren’t writing a one‑off script; we were building an automated gate that must turn green before any deployment can happen.

Core implementation: three building blocks for an amnesia‑proof test suite

Block 1: a reusable Chroma fixture for perfect isolation

This snippet solves the classic “one test pollutes the next” problem. We use tmp_path to give each test its own persistent directory and explicitly delete_collection during teardown—ephemeral mode alone wasn’t enough after a leftover bug we hit.

import pytest
import chromadb
from chromadb.config import Settings

@pytest.fixture
def memory_collection(tmp_path):
    """
    每个测试用例独享一个 Chroma 集合，彻底隔离。
    强制指定余弦相似度，避免默认度量因版本变化。
    """
    client = chromadb.Client(Settings(
        chroma_db_impl="duckdb+parquet",
        persist_directory=str(tmp_path / "chroma_test")
    ))
    collection = client.create_collection(
        name="test_memory",
        metadata={"hnsw:space": "cosine"}  # 强制余弦距离
    )
    yield collection
    # 清理——官方文档没强调，但不删会导致多测试套件相互影响
    client.delete_collection("test_memory")

Block 2: dual assertions on content and score

This is the core validation: insert a known memory, then query with differently worded phrases, and simultaneously verify both the returned text and the distance score. Our previous mistake was checking only IDs and ignoring the score, which let low‑similarity “fake memories” slip through.

def test_memory_retrieval_with_score_threshold(memory_collection):
    """
    验证：用语义相近但措辞不同的查询，仍能召回正确记忆，
    且余弦距离分数不低于 0.75（我们的业务安全阈值）。
    """
    # Arrange：插入一段用户的记忆
    memory_collection.add(
        documents=["用户养了一只名叫布丁的橘猫，今年三岁。"],
        metadatas=[{"user_id": "u1", "session": "s1"}],
        ids=["mem-1"]
    )

    # 用户真实场景可能说的话
    queries = [
        "我家的猫叫什么来着？",
        "我那只橘猫多大了？",
        "布丁几岁了",           # 省略“猫”
        "我的宠物是什么品种"    # 语义相关但不完全重叠
    ]

    for query in queries:
        results = memory_collection.query(
            query_texts=[query],
            n_results=1,
            include=["documents", "distances"]
        )
        # 关键断言 1：必须返回至少一条结果
        assert len(results["documents"][0]) > 0, f"查询 '{query}' 未返回任何结果"

For each query we also assert that the returned document contains a key piece of information (e.g., “布丁”) and that the distance stays below the 0.75 threshold—the full test goes on to catch exactly those low‑score poison pills. Without this, our AI would keep “remembering” things that never happened.

With the fixture and the parametrized barrage of real‑world queries, the test suite finally became the brutal gatekeeper we needed. No green, no deployment—and no more 2 AM amnesia wake‑up calls.

From 3 Hours to 5 Minutes: Automating AI Chatbot Memory Regression with Pytest & Docker

BAOFUFAN — Sat, 08 Aug 2026 01:03:34 +0000

It was 1 AM when a Slack alert jolted me awake: a user complained that “it suddenly lost its memory”—I had just told the AI I'm from Hangzhou, but in the next turn when asked “Where is my hometown?”, the bot replied deadpan, “I don't know.” Checking the latest release, sure enough the memory storage module had changed its serialization logic, quietly breaking the path for loading conversation history.

A manual regression pass took 3 hours: set up Redis and Postgres environments separately, simulate multi-turn conversations, switch users, check memory recall… After that night I decided: this grunt work had to be automated away, once and for all.

Where Exactly Is the Pain?

Memory storage in an AI chatbot isn't as simple as saving a single key-value pair. It involves:

Multiple storage backends: Redis for short‑term, Postgres for long‑term, sometimes a hybrid architecture
Memory lifecycle: reads/writes, expiration, merging, rollback
Context and user isolation: threading memories across multiple sessions of the same user, while strictly preventing cross‑user contamination
Sensitivity to release changes: a single change to serialization format, schema, or TTL logic can silently drop memories

The typical approach is to mock the storage layer in unit tests, but mocks can't expose the real serialization/deserialization pitfalls—many production incidents were cases where mocks passed with flying colors, only to fail miserably in production. A manual regression pass went like: spin up environments → fabricate conversation data → deploy the change → verify memories → fabricate more data → verify again. Half an hour was the bare minimum; a full‑scenario regression could easily take 3 hours and still miss bugs.

Design: Turn Real Middleware into Disposable Test Environments

The core idea: use Docker containers to provide real Redis/Postgres, drive tests with Pytest parametrization, and turn memory logic into repeatable regression test cases.

Why not other approaches?

Why not fakeredis or testcontainers-python with an in-memory fake Redis? Because serialization/deserialization, Lua script behavior under cluster mode, and other nuances can't be replicated by a fake library—that only gives you false confidence.
Why not a shared remote environment integrated with CI? Multiple tests running in parallel would pollute each other, plus it's expensive and slow.
What about Docker + the pytest-docker plugin? The plugin's session scope turned out to be tricky (more on that later). Ultimately I chose pytest fixture + manual lifecycle control with docker-compose, which proved flexible and stable.

Architecture: docker-compose.yml defines the Redis and Postgres services; a module-scoped fixture in conftest.py handles docker-compose up and tears it down after tests; test cases use parametrize to cover different storage backends, conversation turns, and data sizes, verifying memory read/write, isolation, and migration at a click.

Core Implementation: Building the Regression Step by Step

1. Use docker-compose to define disposable real middleware

This config solves the problem of differing databases across environments—dev, CI, and local all use the same image versions.

# docker-compose.yml
version: "3.8"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379"
    # No fixed external port to avoid conflicts
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: test
      POSTGRES_PASSWORD: test
      POSTGRES_DB: memory_test
    ports:
      - "5432"

2. Use a fixture in conftest.py to manage container lifecycle and inject dynamic connection info

This code eliminates the pain of manually starting services every time you run tests, and dynamically assigns ports to avoid conflicts during parallel runs.

# conftest.py
import pytest
import docker
import time
import os

# Use module scope to balance speed and isolation
@pytest.fixture(scope="module")
def docker_services():
    client = docker.from_env()
    # Start with docker-compose, isolated project name prevents clashes
    project_name = f"memtest_{os.getpid()}"
    compose_file = os.path.join(os.path.dirname(__file__), "docker-compose.yml")
    # Use the docker-compose CLI (or Compose API, but CLI is more universal)
    os.system(f"docker-compose -p {project_name} -f {compose_file} up -d")
    # Wait for Redis/Postgres to be ready (health check polling)
    redis_container = None
    pg_container = None
    for c in client.containers.list(filters={"name": project_name}):
        if "redis" in c.name:
            redis_container = c
        elif "postgres" in c.name:
            pg_container = c

    # Poll until Redis is ready
    for _ in range(30):
        exit_code, _ = redis_container.exec_run("redis-cli ping")
        if b"PONG" in exit_code:
            break
        time.sleep(0.5)
    else:
        raise RuntimeError("Redis did not start")

    # Wait for Postgres to accept connections
    for _ in range(30):
        exit_code, _ = pg_container.exec_run("pg_isready -U test")
        if b"accepting" in exit_code:
            break
        time.sleep(0.5)
    else:
        raise RuntimeError("Postgres did not start")

    # 拿到动态端口
    redis_port = int(redis_container.attrs["NetworkSettings"]["Ports"]["6379/tcp"][0]["Ho

Redis Cache Inconsistency: My 48-Hour Debugging Nightmare & The Automated Test That Saved Me

BAOFUFAN — Fri, 07 Aug 2026 12:04:02 +0000

At 1 a.m., an alert call yanked me out of sleep: users were seeing their old nicknames on the page, but after a few refreshes it would fix itself. First thought: the cache wasn’t invalidated. I logged into Redis — the key was still there, but the database already had the updated value. Classic cache-database inconsistency. I assumed it was a small bug. Instead, I spent the next two days hunting this ghost. It was intermittent and only appeared under concurrent updates. Frustrated with debugging by luck, I spent half a weekend building an automated consistency test suite with pytest + Docker. Now it runs before every release and I haven’t been bitten by stale cache data since.

Let me break down the entire approach. If you’ve dealt with similar headaches, you might be able to adopt it directly.

The Problem

The scenario isn’t complicated. A user updates an order status. The code writes to MySQL, then deletes the Redis cache entry. On the next read, the cache is rebuilt. Simple logic, but under concurrency something ugly happens: process A updates the DB and deletes the cache; process B, after A deletes the cache but before A’s transaction commits, reads the old value from the DB and populates it back into the cache. The result? The DB has the new value, the cache holds the old one — only to be fixed when the cache expires. For low-frequency data, this dirty window can stretch for hours.

Standard local tests totally miss this: single-threaded serial execution is always correct. Load-testing tools can trigger concurrency, but they struggle to assert “is the cache consistent with the DB right now?”. You usually only see the final state, and the fleeting intermediate inconsistency slips through. Even worse, these bugs are often introduced after tweaking a SQL query or adding some caching logic — by the time you notice in production, it’s already too late.

The Plan

My core requirement: in a local/CI environment, precisely create cache–database inconsistency with controlled concurrent operations, and assert immediately.

Why I chose what I chose:

pytest? Flexible enough to write test cases, fixture management for resources is a breeze, and you can parameterize concurrency combos. Way better than shell scripts and duct-tape tools.
Docker? Every test run needs pristine Redis + MySQL instances. Docker Compose spins them up with a single command, locks the versions, and avoids the dreaded “works on my machine” dance.
Why not Testcontainers? Not that it’s bad; I just needed finer-grained control over container lifecycles. For example, I wanted to restart Redis mid-test or simulate network glitches. Invoking the Docker CLI directly from a pytest fixture gave me that control.
Why not rely on production monitoring? Monitoring only tells you when something broke. I wanted to stop defects at the CI stage.

The overall idea: define the dependent services in docker-compose, use pytest fixtures to start/stop containers, create tables, and seed data. Each test case simulates a specific concurrency model (write-then-delete, delayed double-delete, binlog-subscription refresh, etc.). Concurrency is orchestrated with threading or asyncio. Finally, read from both Redis and MySQL and assert they match.

Core Implementation

The project layout looks like this:

.
├── docker-compose.yml
├── conftest.py
├── test_cache_consistency.py
└── requirements.txt

1. One-Command Environment with Docker Compose

This configuration solves the pain of manually setting up DB and Redis for every test run.

# docker-compose.yml
version: '3.8'
services:
  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: testpwd
      MYSQL_DATABASE: testdb
    ports:
      - "3306:3306"
    command: --default-authentication-plugin=mysql_native_password
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
      interval: 2s
      retries: 10

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 2s
      retries: 10

Adding healthcheck was the first trap — more on that later.

2. pytest Fixtures: Initialize Connections, Create Tables, Tear Down Data

# conftest.py
import pytest
import subprocess
import time
import redis
import pymysql

@pytest.fixture(scope="session")
def docker_services():
    # 启动 docker-compose
    subprocess.run(
        ["docker-compose", "-f", "docker-compose.yml", "down", "-v"],
        check=True, stdout=subprocess.DEVNULL
    )
    subprocess.run(
        ["docker-compose", "-f", "docker-compose.yml", "up", "-d"],
        check=True
    )
    # 等待健康检查全部通过，避免服务未就绪就开始测试
    time.sleep(8)
    yield
    subprocess.run(
        ["docker-compose", "-f", "docker-compose.yml", "down", "-v"],
        check=True, stdout=subprocess.DEVNULL
    )

@pytest.fixture(scope="function")
def db_conn(docker_services):
    conn = pymysql.connect(
        host="127.0.0.1",
        port=3306,
        user="root",
        password="testpwd",
        database="testdb",
        autocommit=True  # 后面踩坑会解释
    )
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INT PRIMARY KEY,
                name VARCHAR(50)
            )
    # ... (the fixture continues to return the connection and clean up after the test)

With this foundation in place, the actual concurrency tests become surprisingly straightforward — but that’s a story for the next section. The key takeaway: if you can reproduce a cache bug deterministically in CI, you’ve already won half the battle.

From 200s to 15s: 13x Faster Tests with Pytest + Redis Memory Store

BAOFUFAN — Fri, 07 Aug 2026 01:05:04 +0000

At 2 a.m., PagerDuty woke me up not for a production outage, but because the CI pipeline had turned red again—the test stage hit a hard timeout of 600 seconds. That moment felt even more frustrating than fixing a midnight bug: the code itself was fine, but the tests were just too slow. Every single run recreated data, re-called external APIs, and recomputed expensive fixtures from scratch.

The root cause was blunt: our tests had no memory. The same fixture that ran yesterday would be recomputed from scratch today on a different machine or branch. Some team members used .pytest_cache for local caching, but that can’t be shared across distributed CI nodes—a fresh machine still suffered the same painful first run. Even worse, we occasionally hit consistency failures where stale cache results didn’t reflect fresh data, and when a popular key expired, dozens of test workers simultaneously hammered the data source—classic cache stampede and penetration issues, all present.

Finally, I spent half a day embedding Redis directly into the Pytest fixture machinery, building a caching layer with memory storage, anti-penetration, and anti-stampede safeguards. Our test pipeline dropped from 200s to 15s—a 13x improvement—and never timed out in CI again. Here’s the full breakdown of the approach.

Problem Dissection: It’s Not Slow Code, It’s Wheel Reinvention

Our test suite had two typical characteristics:

Many fixtures depend on expensive operations: complex database queries, model inference, external API calls (e.g., fetching tokens).
The test environment was dynamically scaling: 5 concurrent nodes today, maybe 10 tomorrow. Each new runner started as a blank slate.

The usual approach was @pytest.fixture(scope="session") for in-process reuse, but that only works within a single node’s session. What truly slowed us down was cross-node, cross-session repeated computation.

Why not rely on .pytest_cache or local files? Because distributed CI doesn’t share file systems, and local caches can’t control expiration properly, leading to consistency pitfalls like “the code changed but the cache still returns yesterday’s result”. More critically:

Cache stampede: When a hot key (e.g., a large query result) just expired, dozens of test workers simultaneously hit the source, overwhelming the database or external API.
Cache penetration: Test code passes nonexistent parameters (e.g., a wrong user ID) that constantly bypass the cache and hit the backend, causing a flood of empty queries.
Expiry consistency: After the source data updates, the cache still holds the old value, making assertions falsely pass while production breaks.

To fix this fundamentally, we needed a distributed, penetration-proof, and actively expirable centralized memory store. Redis fit that role perfectly.

Solution Design: Turning Redis into the Tests’ “Shared Memory”

My reasoning wasn’t “I want Redis.” It was “I need a cache that can be shared quickly across distributed test nodes and natively supports atomic operations.”

Candidate approaches:

Local files + NFS: passable sharing, but no atomic ops—building custom locks and expiration is too painful.
Memcached: pure cache with good expiry strategies, but weak data structures; implementing Bloom filters or null markers is awkward.
Redis: strings, hashes, SETNX locks, Lua scripts, expiration—everything we needed from a mature ecosystem. No reason not to pick it.

Architecturally, I designed three layers of protection:

Use SETNX-based mutual exclusion so that only the first process fetches the data for a given key while others wait—solving stampedes.
For non-existent data, store a null marker with a short TTL to prevent penetration.
When generating cache entries, embed a version or fixed TTL and provide explicit invalidation commands—developers or CI scripts can purge entries when the source data changes, solving consistency.

The final form: a lightweight Pytest-plugin-style conftest.py where developers simply wrap a fixture’s return value with a function called cached_fixture to gain memory storage.

Core Implementation: A Memory Layer from Zero to Usable

Here’s the runnable code step by step. You can drop it directly into your conftest.py.

Step 1: Redis Connection and Lock Utilities

This block handles safe Redis communication and provides basic locking. We create one Redis connection per Pytest session and release it automatically when the session ends.

# conftest.py
import pytest
import redis
import time
import hashlib
import json
from contextlib import contextmanager

REDIS_URL = "redis://localhost:6379/0"

@pytest.fixture(scope="session")
def redis_client():
    """Session‑level Redis connection, reused for the entire test lifecycle"""
    client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
    yield client
    client.close()

@contextmanager
def redis_lock(client: redis.Redis, key: str, timeout: int = 10):
    """
    Simple mutual exclusion lock based on SETNX.
    Serializes "fetch from source" operations on the same key to prevent stampedes.
    """
    lock_key = f"lock:{key}"
    lock_acquired = client.setnx(lock_key, "1")
    if lock_acquired:
        client.expire(lock_key, timeout)
    try:
        yield lock_acquired
    finally:
        if lock_acquired:
            client.delete(lock_key)

Step 2: Core Memory‑Store Function

This section solves “how to store and retrieve fixture return values while handling penetration and expiration.” The function first checks the cache; on a miss, it applies null-marker protection, then uses the lock to prevent a stampede when fetching the real data.


python
NULL_MARKER = "__NULL__"          # Marker for null values, distinct from None
DEFAULT_TTL = 3600 * 24 * 7       # Normal cache lives 7 days
NULL_TTL = 60                     # Null marker only lives 60 seconds to allow quick correction

def _build_cache_key(func_name: str, args, kwargs) -> str:
    """Generate a unique cache key from function name and parameters, ensuring same inputs hit the same cache."""
    params = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True)
    raw = f"{func_name}:{params}"
    return hashlib.md5(ra

Debugging an LLM Long-Term Memory Disaster: 2000 Misplaced Memories in 3 Hours

BAOFUFAN — Thu, 06 Aug 2026 12:04:31 +0000

At 2 AM, my QA colleague frantically @-ed me in the group chat: “The chatbot’s long-term memory is completely broken. A user mentioned ‘my cat is named Doubao’, but days later when asked about it, the bot recalled ‘your dog is called Xuebing’. Even creepier, when you flip to page 3 you see the same content as page 1. Right now, over 2000 users’ memories are misaligned — take a look.”

Bleary‑eyed, I opened the monitoring dashboards. The tables housing vectors and metadata showed normal write QPS, APIs were returning 200, and there were zero errors. In that moment I knew — this wasn’t a simple code bug. We had shipped a long‑term memory system without any consistency regression testing. Our trust in that “memory storage” black box was running naked from day one.

Breaking down the problem

LLM long‑term memory stores (like Mem0, LangChain’s memory modules, or a home‑grown vector index) typically rely on a “memory pipeline”:

The user converses with the model and facts (memory facts) are created.
Each fact is embedded and stored in a vector database (pgvector, Milvus, etc.), together with structured fields such as timestamp, user ID, and session ID.
During future conversations, the top‑N memories are recalled by similarity plus time‑decay, then stitched together into pages for the model.

The third step is where things went wrong. To make every retrieval reproducible and pagination coherent, you must simultaneously satisfy sort determinism and no‑duplicate‑no‑missing pagination. In practice, many teams just throw an ORDER BY created_at with LIMIT/OFFSET into production. Under concurrent writes, it falls apart immediately — insert order within the same timestamp is non‑deterministic, so OFFSET can skip or repeat records. Additionally, some services cache metadata for performance; if the cache isn’t invalidated promptly, you get phantom reads where a just‑inserted memory “disappears” on query.

Traditional CRUD backend testing falls flat here. You can’t simulate 2000 concurrent writes, page drifting, and cache anomalies with a few hand‑crafted timestamps. You need an automated consistency test suite — one that kneads your memory system like dough until it submits.

Designing the solution

The goal was to build a test suite that could run against any long‑term memory backend — Postgres, Milvus, even an in‑memory KV store. The core idea: generate a large volume of memories that mimic real distributions → batch‑insert them concurrently → fetch via different pagination/sort strategies → verify mathematical invariants.

Tech stack: Python + pytest, memory generation with Faker, storage abstracted behind an interface. Why not just validate with the storage’s own SQL scripts? Because we need to test the final “memory view” that is handed to the LLM — the API’s paginated results — not the raw physical rows in the database. Querying the DB directly would test the wrong thing. Why not Go/JMeter for stress testing? We need logical assertions (e.g., the union of all pages yields the same set), not throughput benchmarks. pytest’s parametrized fixtures and Allure reporting offer at least an order of magnitude higher development speed for this kind of scenario.

The architecture is dead simple: define a MemoryStore protocol with two methods — insert(memories) and recall(user_id, page, size, sort_order). Write a set of pytest test cases around it, then inject the concrete database implementation via dependency injection. Even if you swap Postgres for a custom vector engine, the test suite stays untouched.

Core implementation

Why does this code exist?

First, define the memory data model and the storage abstraction interface. This decouples the test logic from any specific backend. When you change databases, you only need to implement two methods and the entire suite runs immediately.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Protocol, List, Optional
from uuid import uuid4

@dataclass
class Memory:
    """A single memory entity"""
    memory_id: str = field(default_factory=lambda: uuid4().hex)
    user_id: str = ""
    content: str = ""
    created_at: datetime = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )  # Crucial: must carry UTC timezone, otherwise sorting will bite you

class MemoryStore(Protocol):
    """Memory storage protocol – every backend implements this"""
    async def insert_batch(self, memories: List[Memory]) -> None:
        ...

    async def recall_page(
        self,
        user_id: str,
        page: int,
        page_size: int,
        sort_order: str = "desc"
    ) -> List[Memory]:
        ...

Next comes the core pagination‑consistency test function — it generates a flood of memories, pumps them into the store, then reads page by page and verifies mathematical invariants:

Why does this code exist?

It simulates a real user with 2000 inserted memories, then traverses all pages with different page sizes to check for no duplicates, no missing entries, and strict sort consistency. Any breach stops immediately with reproducible failure information.

import pytest
from faker import Faker
from collections import OrderedDict

fake = Faker()

async def run_pagination_consistency(
    store: MemoryStore,
    user_id: str,
    total: int,
    page_size: int
):
    # 1. Insert `total` memories, deliberately jittered timestamps to mimic concurrency
    memories = [
        Memory(
            user_id=user_id,
            content=fake.sentence(),
            # Make some timestamps extremely close to trigger sort instability
            created_at=datetime(
                2025, 5, 1, 12, 0, tzinfo=timezone.utc
            ) + timedelta(seconds=i * 0.01)
        )
        for i in range(total)
    ]
    # Randomly shuffle the insertion order to simulate out‑of‑order concurrent writes
    random.shuffle(memories)
    await store.insert_batch(memories)

    # 2. Expected full set: sorted by created_at descending (newest first), with memory_id tie‑breaker
    ex

LangChain Memory in Production: 10 Edge Cases We Hit and the Automated Tests That Saved Us

BAOFUFAN — Thu, 06 Aug 2026 01:04:00 +0000

At 2:17 AM, my phone vibrated harder than the coffee machine. I opened the monitoring dashboard to find users complaining that “the bot was spouting nonsense.” The real cause: LangChain’s ConversationBufferMemory had silently swallowed an exception after a Redis timeout, returned an empty history, and the downstream Chain kept generating responses with the wrong context. After debugging until 4 AM, the root cause came down to one sentence: the memory component never accounted for the storage layer failing. The next day, we decided not to fix the bug first – we fixed the testing. We designed an automated test suite that cages all 10 conceivable failure scenarios.

Why Memory Breakdowns Break Everything

LangChain’s Memory looks like “just storing the conversation history,” but in production, the backend can be Redis, Postgres, or even a custom vector store. Failures in the storage layer are far more complex than what you see in a demo:

Network jitter causing connect or read timeouts
Connection pool exhaustion – new requests throw instantly
Serialization / deserialization failures (e.g., a message contains an unserializable object)
Concurrent writes to the same session_id creating a race-condition overwrite
Oversized message bodies that blow up Redis memory or hit transfer timeouts
Data corruption (e.g., someone manually tweaked a value in Redis and the checksum no longer matches)
The storage backend becomes completely unavailable (Redis goes down)

The official docs only teach you memory.chat_memory.add_user_message(...), and for exceptions they throw in a single line: “we recommend wrapping with try/except.” When you test manually, you can’t just unplug the Redis network cable and plug it back in. Even if you could, a full regression takes at least two hours and at best covers one or two happy paths. We needed an automated test suite that can precisely inject faults, validate Memory degradation, retries, and data consistency, and run the entire suite in under 3 minutes.

Test Strategy: Mock All the Things? Both Approaches Have Pitfalls

First, let’s be clear: we want to test the anomalous behavior of “Memory + a real storage backend,” not just the in-memory logic of Memory itself. The storage layer must be forced to fail.

Candidate approaches:

Integration tests + manually stopping services: too slow, not repeatable, can’t run in CI.
Locust / chaos engineering: too heavy; the goal is unit-level reliability, not load testing.
pytest + fakeredis: can simulate most Redis commands, and with monkeypatch you can inject timeouts, connection refusals, etc. – fast and CI-friendly. But fakeredis and real Redis have behavioral differences (more on that later).
testcontainers + real Redis: fully realistic, but slow to start; better as a safety-net integration test than the main workhorse.

Final decision: pytest + fakeredis as the daily test workhorse, with testcontainers as a gating integration check. We also wrapped our own RobustMemoryWrapper that encapsulates all exception handling, retries, and degradation logic – instead of scattering it across business code. The test layout:

tests/
  unit/          # fast fakeredis tests covering 10 failure scenarios
  integration/   # testcontainers with real Redis for final verification

The 10 failure scenarios:

Connection timeout (connect timeout)
Read timeout (read timeout)
Connection pool exhaustion
Redis OOM on write
Serialization failure (unserializable object)
Deserialization failure (corrupted data)
Concurrent write race condition
Oversized messages (exceeding max_chunk_size)
Storage backend completely unavailable
Multi-key conflicts (different Memory types sharing the same key)

Implementation: Build a Robust Wrapper First, Then Write the Tests

1. Wrap Memory into a "drop-proof" component

The following code addresses the problem of “the official Memory component throws exceptions straight up with no degradation strategy.” We add timeouts, retries, and fallback behavior to all read/write operations – on timeout, we return an empty history so the Chain won’t crash, but we log an alert.

import time
import logging
from typing import List, Optional
from langchain.schema import BaseMessage
from langchain.memory.chat_memory import BaseChatMemory
from redis.exceptions import TimeoutError, ConnectionError

logger = logging.getLogger(__name__)

class RobustMemoryWrapper:
    """带重试与降级的 Memory 包装器，适用于 Redis 等不可靠后端"""

    def __init__(self, memory: BaseChatMemory, max_retries: int = 2, timeout: float = 0.5):
        self._memory = memory
        self.max_retries = max_retries
        self.timeout = timeout

    # ---------- 对外保持与 BaseChatMemory 兼容 ----------
    @property
    def chat_memory(self):
        return self._memory.chat_memory

    @property
    def memory_variables(self):
        return self._memory.memory_variables

    def load_memory_variables(self, inputs: dict) -> dict:
        for attempt in range(self.max_retries + 1):
            try:
                # 假设底层存储操作会触发 Redis 访问
                return self._memory.load_memory_variables(inputs)
            except (TimeoutError, ConnectionError) as e:
                logger.warning(f"Memory load failed (attempt {attempt+1}): {e}")
                if attempt == self.max_retries:
                    # 最终降级：返回空历史，避免 Chain 中断
                    logger.erro