DEV Community

BAOFUFAN
BAOFUFAN

Posted on

From Manual to Automated: 10x Faster LLM Memory Bug Detection (with 3 Pitfalls)

At 1 a.m., a colleague dropped a screenshot into our group chat: the production chatbot had completely forgotten a user’s context from three days earlier and was responding with total nonsense. My first reaction was “that’s impossible, Redis still has the data.” I logged in — the keys were almost all gone, with only a few newly written records from today. That was the moment I knew our old way of manually testing memory (“send a message, scroll up, see if context is still there”) belonged in the trash.

Breaking down the problem

Storing conversation memory for large language models seems simple at first: one session ID maps to one list of messages, stuffed into a Redis List or ZSET. Read, write, done. But real-world usage introduces a lot of ugly edge cases:

  • Context window trimming: Models have token limits, so the oldest messages must be dropped automatically. If that logic is slightly off, the bot effectively develops amnesia in production.
  • Expiration and eviction policies: Memory gets a TTL, but what happens when Redis hits max memory? How do you guarantee critical sessions aren’t silently kicked out?
  • Concurrent writes: Multiple messages pouring into the same session simultaneously must preserve order without overwrites.
  • Environment differences: Locally everything hums along using fakeredis, but live Redis behaves differently — sometimes radically so.

Manual testing only ever covers the happy path of “write one, read one.” The edge cases above barely get touched. What we really needed was an automated test suite that validates memory storage correctness, runs in CI, and exposes environment discrepancies.

Designing the approach

The technology choices were straightforward: Pytest + real Redis (not fakeredis).

Why not fakeredis?

Fakeredis is a pure-Python Redis mock. It’s great because you don’t need to install Redis, but its handling of expiration, eviction policies, and Lua scripts subtly differs from the real thing. Once your memory logic relies on TTL or LRANGE + LTRIM, a passing test means very little. The decision was clear: tests must use a real Redis instance, spun up via Docker in CI.

Why not unittest?

Pytest fixtures are a natural fit for managing Redis connections and session isolation, and parametrized tests let you cover multiple model configurations (GPT‑4, Claude, etc.) in one sweep.

Overall, we define a MemoryStore abstraction that encapsulates appending, querying, trimming, and expiration. Pytest fixtures inject a Redis client and generate a unique namespace (key prefix) per test function to avoid parallel collision. Then pytest.mark.parametrize feeds in different window sizes and message lengths to validate overall correctness in batch.

Core implementation

The class below encapsulates Redis List operations and implements context window trimming.

import time
import json
from typing import List, Dict, Optional
import redis

class RedisMemoryStore:
    """Redis-List-based conversation memory with automatic window trimming"""

    def __init__(self, client: redis.Redis, max_tokens: int = 2000):
        self.client = client
        self.max_tokens = max_tokens   # using char count as a rough token limit

    def _session_key(self, session_id: str) -> str:
        return f"mem:{session_id}"

    def append(self, session_id: str, role: str, content: str, ttl: int = 3600):
        """Append a message, automatically trim old ones that exceed the window"""
        key = self._session_key(session_id)
        msg = json.dumps({"role": role, "content": content, "ts": time.time()})
        pipe = self.client.pipeline()
        pipe.rpush(key, msg)
        # Trimming logic: delete from the oldest until total chars are within limit
        pipe.lrange(key, 0, -1)
        pipe.expire(key, ttl)
        results = pipe.execute()
        all_msgs = [json.loads(m) for m in results[1]]
        total = sum(len(m["content"]) for m in all_msgs)
        # Note: ltrim removes from the head (oldest first)
        while total > self.max_tokens and all_msgs:
            total -= len(all_msgs[0]["content"])
            all_msgs.pop(0)
        # Keep only the trimmed portion
        if len(all_msgs) < len(results[1]):
            self.client.ltrim(key, -len(all_msgs), -1)

    def get(self, session_id: str, n: Optional[int] = None) -> List[Dict]:
        """Fetch the last n messages, or all if n is None"""
        key = self._session_key(session_id)
        end = -1 if n is None else -(n+1)
        msgs = self.client.lrange(key, -n, -1) if n else self.client.lrange(key, 0, -1)
        return [json.loads(m) for m in msgs]
Enter fullscreen mode Exit fullscreen mode

The following Pytest fixture provides each test function with an isolated namespace to avoid parallel-test interference.

# conftest.py
import pytest
import redis
import uuid

@pytest.fixture(scope="function")
def redis_client():
    """Each test function gets a dedicated Redis DB and key prefix"""
    r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
    # Isolated namespace
    namespace
Enter fullscreen mode Exit fullscreen mode

Top comments (0)