先分清记忆写入与检索两条链路 / MemT 1.0 该在项目哪一层落地

文章导读
MemT 1.0 放在项目哪一层,取决于你现有会话数据的流向,而不是取决于记忆组件本身的接口长什么样。比较稳妥的落位是:记忆写入放在结构化状态生成之后,记忆检索放在上下文拼装之前;中间这两步之间不要插入模型调用,否则本轮检索就拿不到本轮刚生成的结论。要下这个判断,先别急着改代码,把当前项目里从用户输入到模型调用这条链路画出来,标出已经存在的持久化位置,再决定 MemT 1.0 的写入端和检索端各自
📋 目录
  1. 壹 画出当前项目里会话数据的流向
  2. 贰 把记忆写入放在结构化状态生成之后
  3. 叁 把记忆检索放在上下文拼装之前
  4. 肆 用一次两轮任务验证读写顺序
A A

MemT 1.0 放在项目哪一层,取决于你现有会话数据的流向,而不是取决于记忆组件本身的接口长什么样。比较稳妥的落位是:记忆写入放在结构化状态生成之后,记忆检索放在上下文拼装之前;中间这两步之间不要插入模型调用,否则本轮检索就拿不到本轮刚生成的结论。要下这个判断,先别急着改代码,把当前项目里从用户输入到模型调用这条链路画出来,标出已经存在的持久化位置,再决定 MemT 1.0 的写入端和检索端各自挂在哪个环节。

适用场景:已有会话状态管理、准备接入记忆层的项目。操作动作:先输出现有会话数据流向图并标出已有持久化位置,再把写入点固定在结构化状态生成之后、检索点固定在上下文拼装之前。验证方式:用带中间结论的两轮任务跑一遍,看第二轮拼装出的上下文里是否带上第一轮写入的记录。风险边界:状态结构不稳定、结论字段定义为空时,写入会变成噪声;检索为空时必须有回退分支,不能直接把空结果注入上下文。

画出当前项目里会话数据的流向

这一步的用途是防止重复建存储。很多项目已经有消息表、会话状态快照或 Redis 里的临时状态,再接一个记忆层时如果不先画图,很容易出现同一份会话结论在三个地方各存一份、彼此不一致。建议按四个环节画:用户输入、状态更新、上下文拼装、模型调用,并在每个环节旁标注已有的持久化位置。

用户输入
  │
  ├─[已有持久化?] 原始消息落库(messages 表 / 请求日志)
  ▼
状态更新(抽取或合并会话状态:意图、槽位、中间结论)
  │
  ├─[已有持久化?] 会话状态快照(session_state 表 / 缓存)
  ▼
上下文拼装(system prompt + 历史摘要 + 本轮状态)
  │
  ▼
模型调用(LLM 接口)
  │
  ▼
回复返回 → 写回消息表

画完之后,写入点与检索点的位置就比较清楚了:

  • 写入点:状态更新完成、状态对象拿到本轮稳定结论之后,进入上下文拼装之前。
  • 检索点:上下文拼装函数的内部第一步,拼装主体之前。

如果图里显示「状态更新」这一环目前并不存在,会话状态是靠临时拼接历史消息得到的,那接入顺序要反过来:先把状态对象抽出来,再谈 MemT 1.0 的写入。缺这一层时直接写记忆,写进去的多半是原始对话片段。

先分清记忆写入与检索两条链路 / MemT 1.0 该在项目哪一层落地

把记忆写入放在结构化状态生成之后

不放在用户输入处,原因是用户输入里的信息大多是半成品:口头补充、否定前一句、临时试探都会被原样写进去。等到下一轮检索命中这些片段,模型看到的是噪声而不是结论。放在状态更新之后,写入对象是一个已经归一化过的状态对象,字段含义固定,后续检索、去重、失效判断都好做。

写入触发时机建议加几个判断条件,宁可少写也不要乱写。下面是通用骨架,字段名按项目实际状态结构替换:

def should_write(state, turn):
    if not state.get("conclusion"):            # 没有稳定结论,跳过
        return False
    if state.get("updated_at") == turn.prev_state_updated_at:
        return False                            # 本轮状态没变化,不重复写
    if not is_durable(state["conclusion"]):     # 结论是否可跨轮复用
        return False
    return True

if should_write(state, turn):
    memory.write(key=session.id, payload=to_record(state))

这段逻辑放在状态更新函数返回之后、上下文拼装函数调用之前,用一行日志打印 key 和 payload 摘要,方便后面验证。写入开关建议做成配置项(例如 MEMT_WRITE=on/off),先用关闭状态跑通链路,再打开观察行为。

先分清记忆写入与检索两条链路 / MemT 1.0 该在项目哪一层落地

把记忆检索放在上下文拼装之前

检索要参与本轮上下文组装,所以必须在拼装开始之前拿到结果,而不是拼装完再补一段。放在模型调用之后就更没有意义了,那是给下一轮用的。推荐的调用顺序是:状态更新 → 写入 → 检索 → 拼装 → 模型调用。

def handle_turn(user_input, session):
    turn  = new_turn(user_input)
    state = update_state(session.state, turn)        # 1 状态更新

    if should_write(state, turn):                     # 2 写入点
        memory.write(key=session.id, payload=to_record(state))

    hits = memory.search(query=to_query(state), top_k=cfg.top_k)  # 3 检索点

    ctx = build_context(state, session.history)       # 4 拼装
    if hits:
        ctx = inject(ctx, hits)                       # 命中:放在历史摘要之后
    else:
        ctx = inject(ctx, fallback_note(cfg))         # 回退分支
    return llm.call(ctx)                              # 5 模型调用

检索为空是常态,回退分支要事先定好,常见做法有两种:一是不注入任何空块,直接用状态对象里的本轮信息拼装,避免模型误以为存在历史记忆;二是注入一条极短的固定提示,说明本轮没有可复用记忆。无论选哪种,都要在拼装日志里记录 hits=0,否则排查时无法区分「没检索」和「检索了但没命中」。

先分清记忆写入与检索两条链路 / MemT 1.0 该在项目哪一层落地

用一次两轮任务验证读写顺序

验证目标是顺序而不是效果,所以任务要带一个明确的中间结论。可以这样设计第一轮:给出两三条约束,要求模型先确认方案再回答;第二轮只发一句「按上次确定的方案继续」,刻意不在第二轮重复约束。如果顺序正确,第二轮拼装出的上下文里应当出现第一轮写入的那条记录;如果第二轮看不到约束,说明检索没有命中或拼装时被丢掉。

验证时把写入日志和检索日志都打出来,重点看两件事:第一轮的 memory.write 是否发生在第一轮拼装之前,第二轮的 memory.search 是否发生在第二轮拼装之前且命中条数大于零。

[验证记录]
日期:
环境:本地 / 测试环境
配置:MEMT_WRITE=on,top_k=
第一轮输入:
第一轮状态快照(日志字段):
写入日志(key / payload 摘要 / 时间戳):
第二轮输入:
检索日志(query / top_k / 命中条数 / 时间戳):
第二轮拼装结果:是否包含第一轮结论(字段或行号)
回退分支是否触发:
结论:顺序正确 / 需调整(写明调整点)

同一组输入建议重复跑两到三次,确认写入内容和命中结果可复现;如果两次结果不一致,通常先查状态对象里是否有未固定下来的字段(例如时间戳、随机 id)被一起写进了记忆,而不是先怀疑检索逻辑。环境差异、缓存状态和并发会话都可能影响结果,需要结合具体部署确认。