How I Spent 48 Hours Debugging a LangChain Agent Memory Race Condition — And How Playwright Solved It

At 2:17 AM, my phone exploded. Someone in the user group posted a screenshot: “I just told it my name is Boss Wang, and the next thing it asked was ‘How should I address you?’ — is this AI a goldfish?” I pulled up the logs. The agent had correctly invoked the memory storage, and Redis held the conversation history for that session, yet the frontend showed total amnesia. I manually tested it seven or eight times without reproducing the issue; I even suspected network hiccups. It wasn’t until I wrote a set of Playwright automated tests — firing 20 messages concurrently into the same session — that I witnessed it firsthand: LangChain’s ConversationBufferMemory, when writing asynchronously to storage, was overwriting the order of two conversation turns. The later message was “swallowing” the earlier context. This article shares the complete debugging journey, the automated verification approach, and a ready-to-use test script you can drop straight into your project.

Problem Breakdown: Why the Agent “Forgets”

The scenario was straightforward: a customer support agent built with LangChain, a Streamlit web chat interface, and Redis as the persistence layer for memory. Users provided personal information over multiple turns, and the agent’s follow-up answers had to be based on that context. For example, if a user said “My name is Zhang San,” the agent needed to remember — not ask for the name again in the very next message.

After going live, nearly 5% of conversations showed a “memory gap”: two consecutive questions arrived within a very short interval (<200 ms), or a user on a slow network rapidly clicked the “Send” button. The backend sometimes fetched a history record that was missing the second-to-last message. Two root causes emerged:

1. LangChain’s ConversationBufferMemory.save_context is not atomic. Internally it calls load_memory_variables to fetch the old history, concatenates the new messages, and then writes the whole thing back to storage. Under asynchronous concurrency, two requests can both read the same version of the history, each append its own message, and then overwrite the previous write — causing the first message to be lost entirely.
2. The frontend had de-duplication logic that blocked a second rapid click, but network latency still allowed two nearly simultaneous requests to reach the backend, triggering that race condition.

Conventional “click-around” manual testing simply couldn’t catch this: you can’t click with 200 ms intervals by hand, nor reliably reproduce concurrent requests. We needed a tool that could precisely control timing, simulate real user behavior, and automatically assert memory consistency.

Solution Design: Why Playwright Instead of a Load Testing Tool

Our goal was to emulate real users sending multi-turn messages in the chat interface — including the extreme “rapid fire” scenario — and verify that the agent’s replies correctly referenced the full conversation history.

We evaluated a few options:

• HTTP load testing tools like locust / wrk can fire POST requests directly at the /chat endpoint, but they require manually managing sessionId and simulating a complete message flow. They can’t sense UI-level de-duplication logic. Load testers excel at measuring throughput, not at asserting stateful “conversation context consistency.”
• Selenium can drive a browser, but its async waiting mechanisms are outdated, and handling WebSocket real-time updates is painful.
• Playwright offers native auto-waiting, network interception, and WebSocket listening, plus a smooth pytest-playwright integration — writing assertions feels like writing unit tests. Most importantly, it can spin up multiple browser contexts simultaneously to simulate concurrent users, and within each context you can control message timing precisely — even sending two messages within 50 ms.

Architecturally, the tests were split into three layers:

1. Fixture layer – launch a Dockerized agent backend (FastAPI + LangChain + Redis) for a clean, reproducible environment.
2. Page Object layer – encapsulate chat page operations (send message, read the latest reply).
3. Test case layer – call Page Objects, execute multi-turn conversations and concurrency scenarios, and assert memory consistency.

Core Implementation: Turning “Goldfish Memory” into a Reproducible Failure

Below is the intentionally flawed backend endpoint used as our test target. It uses a manual “read-modify-write” pattern on Redis that perfectly exposes the race condition (a production fix would use Redis list operations or locking — here the bug is the point of the test).

# agent_backend.py
import os
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import redis.asyncio as redis
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.chat_models import ChatOpenAI

app = FastAPI()
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

class ChatRequest(BaseModel):
    session_id: str
    message: str

class ChatResponse(BaseModel):
    reply: str

# 创建带记忆的链（注意：记忆的存储由我们手动操作 Redis，而不是用内置的 RedisChatMessageHistory，
# 目的是模拟那些自己实现存储层的场景——很多团队就是这么干的，而且容易出 bug）
llm = ChatOpenAI(temperature=0)
memory = ConversationBufferMemory()

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    # 1. 从 Redis 加载该 session 的历史
    history_json = await r.get(f"history:{req.session_id}")
    if history_json:
        # 反序列化历史消息并注入 memory（这里是一个简化的模拟；真实场景需要格式化）
        # 为了演示，我们直接拼接一个 context 字符串
        memory.clear()
        memory.save_context({"input": "历史"}, {"output": history_json})

    # 2. 创建链并执行
    chain = ConversationChain(llm=llm, memory=memory)
    response = await chain.arun(req.message)

    # 3. 更新历史并写回 Redis（**这里就是竞态的温床**）
    new_hi