凌晨两点十七分。Slack 里炸出一条使用者反馈：“你们的 AI 助理像得了健忘症，刚说完的会议纪要，再问就说没存过。老实说，”日志显示写入 Redis 的消息 ACK 了但向量库里那一条竟然是空的。怎么说呢，这种“间歇性失忆”已经第三次把使用者惹毛了。之前只靠手动 curl 几条对话就算测试通过显然挂了。

问题拆解

大模型的“记忆存储”通常不止一个后端：短期记忆放在 Redis List 或内存环形缓冲区。长期记忆被 Embedding 后存入向量数据库，同时可能还有一份原始文本落在 PostgreSQL。一次 save_memory 调用，如果只是顺序写了 Redis 和 PG。接下来异步推向量库，那么在 Redis 已返回但向量库未落盘的窗口期，读请求可能拿到不完整的历史。

更恶心的是“记忆压缩”——当对话轮数超过阈值。触发摘要生成，同时要清理原始消息、更新摘要向量。这个过程中任何一个步骤失败。都会造成“已压缩”和“原消息残留”并存，导致读到重复或矛盾的内容。

• 异步窗口期的 bug 几乎不可能手复现，靠人眼检查日志就是在大海捞针。
• 对账脚本是事后擦屁股，不是开发阶段的防护。怎么说呢，
• 新来的同事根本不知道哪些场景会翻车。“可工作”的假象一戳就破。

我决定直接把状态一致性的验证写进 Pytest 测试套件，让 CI 替我们跑。

方案设计

选型上，测试框架没得商量就是 Pytest。fixture 能优雅地管理多个存储后端的生命周期；parametrize 能把几十种异常组合一网打尽。

• 单元测试只 mock 单层验证不了跨存储程序的并发竞态。mock 出来的返回值永远是按你预设的，测不出真实窗口。
• 集成测试用 Docker‑compose 拉起真 Redis / 向量库但 Pytest 里直接管理它们要写一堆 boilerplate。按理说，我们用 pytest-docker` 或者干脆用 testcontainers-python`。能让每个测试用例拥有独立且隔离的存储实例，跑完自动销毁。

架构思路很简单：每个测试用例是一个并发写 + 多次读 + 最终断言的模式。我们模拟两个协程同时写入相同使用者的消息，再施加“压缩”触发信号。最终用 alert_eventually-style 的轮询去验证所有后端最终一致。没有用 /sleep/-这种拍脑袋等待，而是用 Tenacity 的 retry 或 asyncio.wait_for 做带超时的条件等待。

. 定义可测试的记忆存储抽象

This段代码解决“究竟要测什么接口”的问题——把产品级的记忆存储抽象成一个MemoryStoreProtocol,后面所有测试只依赖这个协议。

from typing import Protocol,List,Dict,Any,runtime_checkable
@runtime_checkable
class MemoryStoreProtocol:
"""大模型记忆存储必须实现的接口"""
async def append_message -> None:
"""追加一条对话消息到短期记忆"""
...
async def get_history(self,user_id: str。limit: int = 20) -> List]:
"""按时间顺序返回最近的消息列表"""
...
async def compact -> None:
"""触发记忆压缩：生成摘要并清理原始短期消息"""
...
async def consistent_state(self,user_id: str) -> bool:
"""检查所有后端数据是否一致"""
...
}

. 一个带竞态条件的简化实现

FragileMemoryStore—它用 Python list 代替 Redis，用 dict 代替向量库，并在写入时随机 await asyncio.sleep 模拟异步延迟。这段代码是测试靶子，

import asyncio
import random
from dataclasses import dataclass,field
@dataclass
class FragileMemoryStore:
# 模拟三层存储
short_term : Dict] = field # Redis List
vector : Dict] = field # 向量库
raw_pg : Dict] = field # PG 原始表
async def append_message(self,user_id : str,role : str,content : str) -> None:
msg = { "role": role,"content": content }
# . 写短期缓存
self.short_term.setdefault.append
# . 写 PG 原始表
self.raw_pg.setdefault.append
# . 推向量库是异步的，模拟延迟
await asyncio.sleep) # 随机窗口
self.vector.setdefault.append
async def get_history(self,user_id : str,limit:int=20 ) -> List:
if user_id in self.short_term:
return self.short_term
return self.raw_pg.get
async def compact->None:
if user_id not in self.short_term:
return
summary = f"Summary of {len} msgs"
self.vector =
await asyncio.sleep)
self.short_term.pop
# 故意不删 raw_pg。制造不一致
async def consistent_state->bool:
short=len)
raw=len)
vec=len)
if short==0 and raw>0 and vec==1:
return False # 压缩后 raw 还有数据，不一致
return True
}

. Pytest 并发测试用例

import pytest
import asyncio
from tenacity import retry,stop_after_delay,wait_fixed
# 待测实现和协议
from store import FragileMemoryStore,MemoryStoreProtocol
@pytest.fixture
async def store:
s=FragileMemoryStore
yield s
# 清理
@pytest.mark.asyncio
async def test_concurrent_write_and_compact_consistency:
user="user_42"
# 并发写入两条消息
async def writer:
> for i in range:
> await store.append_message
await asyncio.gar,writer)
&nbsbp;# 在写入未完全落盘时触发压缩
await asyncio.sleep # 模拟真实调度间隙
await store.compact
@retry,wait=wait_fixed。reraise=True)
async def assert_eventually_consistent:
ok=await store.consistent_state
if not ok:
raise AssertionError("State inconsistent:"+\
"
compact left raw data behind")
await assert_eventually_consistent
}

You guessed what happened?This test必然失败 ,因为 compact

. 用参数化覆盖多种异常时序

,配合 pytest‑asyncio 和 tenacity。就能跑起上面的完整套件：

# conftest.py
import pytest
from store import FragileMemoryStore 

@pytest.fixture async def store: s = FragileMemoryStore yield s # 极端情况下手动清理，以保证隔离 s.shortterm.clear;其实，s.vector.clear;s.rawpg.clear }​​​​​​​​​​​

#Python #Pytest #大模型 #状态一致性 #后端测试 \u5177\u4E8E\u7528\u8005\u83B7\u53D7\u7684\u5934\u72C0: * \u7528\u6237 \u60A8 \u7684 AI \u529F\ud60d \u5217\ud604。* \u89c7\ud56fd\ud60d \u52a8\ud603,* \u6570\ud71c\ud600. --- \u5177\u5907: * \u521a\u96c9- devtools,* pydevx,* CI/CD. --- 关于作者
一位习惯深夜编排自动化而非修补bug 的 Backend / 架构工程师。怎么说呢，关注 GitHub:\. "}

---