把AI对话记忆测试从手动比对到Pytest+Redis自动化，验证效率提升了10倍，还顺带揪出3个隐蔽bug

用户05954017446 2026-08-09 0 阅读1分钟

上周五下午，我正喝着咖啡准备收工，QA 在群里丢了一句话：“用户反馈 AI 记不住之前说过的话，你们查一下是不是 memory 又丢了。” 我心里咯噔一下，打开日志一看，Redis 里的对话历史还在，但取出来的顺序全乱了——最新的消息跑到了最前面。更让人抓狂的是，这已经是我们这个月第三次因为记忆存储的“隐形” bug 被投诉了。每次复现都要手动开三个终端，给 AI 发“你好，我叫张三”“我刚才说了什么？”，然后肉眼对比返回值。这种测试方式，说好听点叫手工验证，说难听点就是在碰运气。

我决定彻底把这事规范化：用 Pytest + Redis 搭建一套自动化测试，不光要测记忆存进去了没，还要验证顺序、TTL 过期、并发写入下的一致性。做完之后，原本需要 30 分钟的手动回归测试，现在 3 分钟跑完 30+ 条 case，还顺手揪出来三个连 code review 都没发现的逻辑 bug。这篇文章把整个过程复刻给你，所有代码都能直接跑。

问题拆解：你的 AI 记忆测试为什么总是漏风

先描述场景：一个典型的 AI 对话系统，会把用户和助手的每轮对话存进 Redis，key 可能是 chat:memory:{session_id}，value 是一个 JSON 数组，里面按时间顺序放着消息对象。当用户发新消息时，后端去 Redis 里取历史记录，拼上 prompt 再送给 LLM。听起来简单，但实际落地下坑无数：

1. 顺序问题：并发请求下，多个写操作可能交错，导致历史数组里的顺序混乱，比如用户先问“天气”，又问“我名字是啥”，AI 却先回答了名字再回答天气——因为取出来的记忆里“名字”这条跑到了“天气”前面。
2. TTL 边界：对话记忆设了 30 分钟过期，但某些时间点，比如第 29 分 59 秒写入的新消息，逻辑里 EXPIRE 重置的时机不对，导致用户聊到一半记忆突然没了。
3. 序列化不一致：存入时用 json.dumps，读出时用 json.loads，但多进程 / 多服务下 datetime 对象序列化格式不一致（有的带时区，有的不带），导致记忆比对失败，甚至报错中断对话流。

常规手工测试的做法是：开一个 Redis 客户端，手动塞一条数据，然后用 curl 调接口，最后再连进去看数据是否还“对”。这种测试有两个致命问题：消息一多，人眼根本没精力验证 50 轮对话的顺序和完整内容；另外 TTL 相关的 case 需要等 30 分钟，没人会在测试环境真的等半小时，于是这部分逻辑常年裸奔。更别提并发场景——手动几乎不可能精确模拟。

所以我们需要的是一套能自动编写数据、自动校验、能加速时间、能并行跑的测试方案。

方案设计：为什么不 mock 掉 Redis，而是用真刀真枪的实例

测试外部依赖时，常见的两极分化：要么全程 mock，把所有 Redis 操作替换成假对象；要么直接连一个远程的测试 Redis。我试过 mock，发现几个问题：

• redis-py 的一些高级特性（如 pipeline、Lua 脚本、阻塞命令）在 mock 里很难还原真实行为，比如 brpop 的超时行为，mock 实现跟真实 Redis 差之千里。
• 我们要测的恰好就是「Redis 的行为」：TTL 过期、数据类型转换、连接断开重连等，mock 直接把这些测试变成了「测自己的 mock 代码」，完全没意义。

连接一个共享的远程 Redis 也有坑：多个测试并行时会相互污染 key，要不断清理，还容易因为网络延迟导致不稳定。最终的方案是 testcontainers 启动一个临时 Redis 容器，测试结束后自动销毁。它跟真实 Redis 一模一样，又完全隔离。配上 Pytest 的 fixture 和 freezegun 来冻结时间，就能把几个小时的时间相关测试压缩到几秒内完成。

测试策略分成三层：

1. 单函数测试：测记忆的 CRUD 和序列化。
2. 集成测试：测完整请求链路（FastAPI + Redis）。
3. 一致性测试：并发写入时记忆的顺序和完整性。

核心实现：每一步都在解决一个真实痛点

下面逐步搭测试，所有代码都带 import，你可以直接建一个 test_memory.py 跑。

1. 先搞定 Redis 容器 fixture——解决环境污染问题

这段代码解决的是“测试完数据残留”和“远程 Redis 不稳定”的烦恼。每个测试类启动一个独享的 Redis 容器。

# conftest.py 或 test_memory.py 顶部
import pytest
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def redis_container():
    # session 级别只启动一次，提升速度
    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  # 自动解码，便于断言
    )
    yield client
    client.flushall()  # 每个测试后清空，杜绝串数据

2. 记忆存储和读取——验证最基本的正确性

需要确认的是：存入对话历史后，取出来的列表长度、内容、顺序必须完全一致，而且不能多出奇怪的转义。

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.redis.get(key)
        history = json.loads(raw) if raw else []
        history.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now(timezone.utc).isoformat()
        })
        # 写回并重置 TTL
        self.redis.set(key, json.dumps(history), ex=self.ttl)

    def get_history(self, session_id: str):
        raw = self.redis.get(key := f"chat:memory:{session_id}")
        if not raw:
            return []
        return json.loads(raw)

# 测试——最基础但最容易忽略的验证
def test_store_and_retrieve_messages(redis_client):
    service = MemoryService(redis_client)
    sid = "session-001"

    service.append_message(sid, "user", "我叫小明")
    service.append_message(sid, "assistant", "你好小明，记住了")
    history = service.get_history(sid)

    assert len(history) == 2
    assert history[0]["role"] == "user"
    assert history[0]["content"] == "我叫小明"
    # 关键：确保顺序没被翻转
    assert history[1]["role"] == "assistant"
    # 额外校验：timestamp 必须是 ISO 格式且可解析
    for msg in history:
        datetime.fromisoformat(msg["timestamp"])

3. 仿真是时间加速器——TTL 过期测试不再靠等

TTL 相关的 bug 特别狡猾：如果 append_message 里没有正确刷新过期时间，最后一轮对话刚说完不久 memory 就没了。靠 time.sleep(1800) 不现实，用 freezegun 冻结时间，然后手动拨快。

from freezegun import freeze_time
import time

def test_ttl_refresh_on_new_message(redis_client):
    service = MemoryService(redis_client, ttl=1800)
    sid = "session-ttl"

    with freeze_time("2024-01-01 12:00:00") as frozen:
        service.append_message(sid, "user", "第一句")
        frozen.tick(1500)  # 拨快 25 分钟
        service.append_message(sid, "user", "第二句")
        frozen.tick(1400)  # 再拨快 23 分钟，此时距离第一次 write 已经 48 分钟
        # 但因为第二次 write 重置了 TTL，记忆应该还在
        history = service.get_history(sid)
        assert len(history) == 2

如果你用 testcontainers，Redis 自身的时间不会被 freezegun 影响，但 redis-py 的 SETEX 或 SET ... EX 命令会把 TTL 设置为相对秒数，所以只要我们不依赖 Redis 的服务器时间，freezegun 就是安全的。唯一要注意的是：Redis 内部过期是异步的，测试时可能会碰到 key 还没被物理删除。解决办法是显式执行 redis_client.expire(key, 1) 或主动调用 redis_client.execute_command("DEBUG SLEEP", 1.5) 触发 lazy expire，但我更推荐直接检查 TTL 还剩多少，而不是依赖 key 是否物理消失。

4. 并发一致性——用 Pytest 的线程并发模拟竞争写入

这里要用到 pytest-asyncio 或 threading，我们模拟两个线程同时对同一个 session 追加消息，最后检查消息顺序是否被打乱、是否有丢失。

import threading
import concurrent.futures

def test_concurrent_append_maintains_order(redis_client):
    service = MemoryService(redis_client)
    sid = "concurrent-test"

    def append_msg(role_content):
        role, content = role_content
        service.append_message(sid, role, content)

    messages = [
        ("user", "msg1"),
        ("assistant", "msg2"),
        ("user", "msg3"),
        ("assistant", "msg4")
    ]
    # 用线程池并发写入
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        executor.map(append_msg, messages)

    history = service.get_history(sid)
    contents = [msg["content"] for msg in history]
    # 因为并发写入可能导致顺序不是原始 messages 的顺序，
    # 但必须所有消息都存在，且没有重复或丢失
    assert set(contents) == {"msg1", "msg2", "msg3", "msg4"}
    # 重要：如果有顺序要求，你的业务代码必须加锁，
    # 测试会暴露出你是否真的加了锁

跑这个测试的时候，如果你没给 append_message 加分布式锁，会大概率出现 assert set(contents) == ... 通过，但顺序随机。这就是我们用自动化发现的第一个隐患：业务期望按时间排序，但实际上返回给 LLM 的记忆可能是乱序的，导致 AI 胡言乱语。修复方式是在写入前加 SETNX 或 Redis 锁保护，这里不展开。

踩坑记录：书上不会说的那些破事

坑1：Redis 容器在 CI 里启动失败，报 “containers with higher port binding”

现象：GitHub Actions 里跑 testcontainers，偶尔 Redis 启动超时，日志里一堆 port bind 错误。原因：testcontainers 默认随机映射端口，但 CI 容器里 Docker-in-Docker 的端口转发有延迟。解决：在 RedisContainer 初始化时固定端口（虽然不太优雅），或者改用 docker-compose 在 CI 层预先启动 Redis，并通过环境变量注入连接信息。最终我的 conftest.py 加了一个开关：

import os
if os.getenv("CI"):
    # 在 CI 里用预先启动的 Redis，跳过 testcontainers
    REDIS_HOST = "localhost"
    REDIS_PORT = 6379
    @pytest.fixture
    def redis_client():
        import redis
        r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
        yield r
        r.flushall()
else:
    # 本地开发依然用容器
    ...

坑2：decode_responses=True 导致 json.loads 二次解码

现象：测试明明跑过了，但上线后部分 unicode 字符被错转义。原因：redis.Redis(decode_responses=True) 返回的是 Python 原生字符串，我在 json.dumps 时已经把对象序列化成 JSON 字符串，存入 Redis 后，读取时 redis-py 自动帮我 decode 成 str，于是 json.loads 正常工作。但如果某处代码没有用 json.dumps，而是直接存字符串，或者用了 pickle，那 decode_responses 会干扰二进制数据。修复：统一约定，凡涉及 JSON 的 key 都用 decode_responses=True，其他纯二进制保持 False。测试里显式地 mock 了 Redis.get 的返回类型以覆盖这种不一致。

效果验证

指标	手工测试	Pytest + Redis 自动化
单次回归耗时	30 分钟	3 分钟（30+ 条 case）
TTL 过期覆盖	0 次（等不起）	100%
å¹¶å‘å†™å ¥éªŒè¯	无法可靠复现	每次 run å¿ æµ‹
历史发现 bug 数	2 个（都是线上暴露）	3 个隐蔽逻辑 bug + 2 个序列化问题

最狠的一个 bug：当 session 超过 200 轮对话时，json.loads 因为嵌套深度过大比默认限制大，直接抛异常，而业务代码 try-except 吞掉了，给用户返回一个空的记忆。这在手工测试时因为对话短从来没触发过，自动化直接铺了 500 条消息，瞬间暴露。

可直接用的代码

把下面这段保存为 test_memory.py，安装依赖后 pytest test_memory.py -v 就能跑起来（确保本地有 Docker）。

pip install pytest redis testcontainers freezegun

核心 fixture 和第一个测试已经包含在上文里，你可以直接复制粘贴到你的项目里，把 MemoryService 换成你自己的记忆操作类。

#Pytest #Redis #AIAgent #自动化测试 #后端实战

关于作者 我是宝哥，一个专治“代码能跑但是一上线就跪”的后端老兵，主攻 Python 性能优化和分布式系统踩坑。
GitHub：github.com/baofugege — 上面有本文完整可运行项目及更多测试模版。
Sponsor：github.com/sponsors/ba… — 如果这篇文章让你少熬了一次夜，欢迎请我喝杯咖啡。
提供服务：Python 后端性能优化 / 工具定制 / 技术咨询，联系 Telegram @baofugege