Agent 记忆分层实践

目标：把 Agent 核心概念与架构 里的 Memory 拆成可跑的三层——工作记忆、会话摘要、长期画像——并写清写入策略与评测。本轮实际塞进模型的字节预算见 上下文工程。面试题型见 Agent 系统高频题；会话 state 在图上怎么续见 [LangGraph 编排实践](/docs/ai/Agent/LangGraph 编排实践)。

记忆 ≠ RAG：RAG 是共享知识库（手册、政策）；记忆是「这个用户 / 这次任务」的状态。公司退货规则走 RAG，用户「喜欢简短回答」走画像。

目标与验收

验收项	标准
窗口	超长对话只把最近 N 条塞进 Prompt
摘要	被裁掉的旧轮次压缩成一段 summary，下一轮仍能引用
跨会话	新开会话（空 messages）仍读到用户偏好
写入	只落稳定偏好，不把工具轨迹整段当长期记忆
评测	jsonl 用例能测「第二场对话是否用上偏好」

三层模型

本轮要塞进模型的上下文
  ├─ 工作记忆     最近消息 + 当前任务 state / 工具观察
  ├─ 会话摘要     本 thread 里被窗口裁掉的旧对话压缩
  └─ 长期画像     跨 thread：偏好、已确认事实（按 user_id）
业务真源（订单、库存、政策）→ 查库 / RAG，不当「模型记得」

层	生命周期	典型存放	丢了会怎样
工作记忆	本轮 / 本 thread	消息列表、scratchpad、checkpointer	当前任务失忆、重复调工具
会话摘要	本 thread 变长时	一段 summary 字符串	窗口外的约定被忘
长期画像	跨会话、跨天	业务库 / JSON / LangGraph Store	每次都要重新自我介绍
业务真源	以库为准	DB / RAG	用过期记忆答事实题

记忆口诀：窗口管现在，摘要管今晚，画像管以后；事实以库为准。

建议目录

mini_memory/
  memory.py        # 窗口、摘要、画像读写
  agent.py         # 假 LLM 把三层拼进上下文
  eval_set.jsonl
  profiles/        # 长期画像 PoC（按 user_id 一个 json）

无框架即可跑；LangGraph 对照见文末映射表。不依赖付费 API。

工作记忆（窗口）

只保留最近 N 条。工具观察也算工作记忆，但不要无限追加。

# memory.py
WINDOW = 6  # 条消息，PoC 用条数；上线按 token 预算


def working_context(messages: list[dict], summary: str | None) -> str:
    recent = messages[-WINDOW:]
    parts = []
    if summary:
        parts.append(f"[较早对话摘要] {summary}")
    for m in recent:
        parts.append(f"{m['role']}: {m['content']}")
    return "\n".join(parts)

面试说法：工作记忆 = 现在必须看见的东西；不是「把历史全塞进去」。

会话摘要

消息超过窗口时，把被裁掉的前半压成一段话，与窗口并存。PoC 用拼接；上线换成 LLM 摘要即可。

def maybe_summarize(
    messages: list[dict], summary: str | None
) -> tuple[str | None, list[dict]]:
    if len(messages) <= WINDOW:
        return summary, messages
    older = messages[:-WINDOW]
    blob = "；".join(m["content"] for m in older if m["role"] == "user")
    piece = f"用户曾提到：{blob[:200]}"
    new_summary = f"{summary}；{piece}" if summary else piece
    return new_summary, messages[-WINDOW:]

摘要是有损压缩：细节会丢。关键数字、订单号应进业务库或画像，不要只活在 summary 里。

长期画像

跨会话：新 thread 的 messages 为空，仍要能读到偏好。PoC 用文件；生产用用户表 / KV，按 user_id 隔离。

from pathlib import Path
import json

PROFILE_DIR = Path("profiles")


def load_profile(user_id: str) -> dict:
    path = PROFILE_DIR / f"{user_id}.json"
    if not path.exists():
        return {"prefs": {}, "facts": []}
    return json.loads(path.read_text(encoding="utf-8"))


def save_profile(user_id: str, profile: dict) -> None:
    PROFILE_DIR.mkdir(parents=True, exist_ok=True)
    path = PROFILE_DIR / f"{user_id}.json"
    path.write_text(json.dumps(profile, ensure_ascii=False, indent=2), encoding="utf-8")

写入策略

长期记忆默认不写。整段对话、工具 JSON、一次性任务结果都不该进画像。

import re

ALLOWED_PREFS = {"style", "name", "lang"}


def extract_pref(user_text: str) -> dict[str, str]:
    """假抽取：规则够演示；上线换成带 schema 的抽取再人工/自动确认。"""
    found: dict[str, str] = {}
    if re.search(r"简短|少说两句", user_text):
        found["style"] = "brief"
    if re.search(r"详细|展开讲", user_text):
        found["style"] = "detailed"
    m = re.search(r"叫我(.{1,8})", user_text)
    if m:
        found["name"] = m.group(1).strip(" ，。,.")
    return {k: v for k, v in found.items() if k in ALLOWED_PREFS}


def apply_writes(profile: dict, prefs: dict[str, str]) -> dict:
    if not prefs:
        return profile
    merged = {**profile, "prefs": {**profile.get("prefs", {}), **prefs}}
    return merged

规则	原因
白名单字段	防止把密钥、身份证、工具原始输出写进去
稳定偏好才写	「这次帮我查一下」不是偏好
可覆盖、可删	用户说「以后别叫小名」必须能改 / 遗忘
租户隔离	user_id 做 key，禁止跨用户检索画像
事实走真源	「订单已发货」查订单库，不写进 prefs

拼进 Agent

假 LLM：看见画像里的 style=brief 就短答；否则正常答。用来证明第二场空会话仍能读到偏好。

# agent.py
from memory import (
    apply_writes,
    extract_pref,
    load_profile,
    maybe_summarize,
    save_profile,
    working_context,
)


def reply(user_id: str, messages: list[dict], summary: str | None, text: str) -> dict:
    profile = load_profile(user_id)
    prefs = extract_pref(text)
    profile = apply_writes(profile, prefs)
    save_profile(user_id, profile)

    messages = messages + [{"role": "user", "content": text}]
    summary, messages = maybe_summarize(messages, summary)
    ctx = working_context(messages, summary)
    style = profile.get("prefs", {}).get("style")
    name = profile.get("prefs", {}).get("name", "")
    greet = f"{name}，" if name else ""
    if style == "brief":
        answer = f"{greet}好。"  # 演示：短答
    else:
        answer = f"{greet}收到。上下文如下：\n{ctx}"

    messages = messages + [{"role": "assistant", "content": answer}]
    return {"messages": messages, "summary": summary, "answer": answer}


if __name__ == "__main__":
    # 会话 1：写入偏好
    s1 = reply("u1", [], None, "请简短回答，叫我小王")
    print("s1", s1["answer"])
    # 会话 2：空历史，只带 user_id
    s2 = reply("u1", [], None, "还在吗")
    print("s2", s2["answer"])  # 应仍是短答且带「小王」

运行：python agent.py。会话 2 不传上一场 messages，只靠 profiles/u1.json。

评测

先评「跨会话是否用上画像」，再评摘要是否丢关键约束。

{"id": "pref-brief", "user_id": "u1", "session1": "请简短回答", "session2": "还在吗", "expect_style": "brief"}
{"id": "pref-name", "user_id": "u2", "session1": "叫我小王", "session2": "你好", "expect_name": "小王"}
{"id": "no-leak", "user_id": "u3", "session1": "请简短回答", "session2_user": "u4", "session2": "还在吗", "expect_style": null}

import json
from agent import reply
from memory import load_profile


def eval_cases(path: str) -> None:
    total = passed = 0
    for line in open(path, encoding="utf-8"):
        case = json.loads(line)
        total += 1
        reply(case["user_id"], [], None, case["session1"])
        uid2 = case.get("session2_user", case["user_id"])
        reply(uid2, [], None, case["session2"])
        prefs = load_profile(uid2).get("prefs", {})
        ok = True
        if "expect_style" in case:
            ok = ok and prefs.get("style") == case["expect_style"]
        if "expect_name" in case:
            ok = ok and prefs.get("name") == case["expect_name"]
        passed += int(ok)
    print(f"pass={passed}/{total}")

no-leak：写在 u3 的偏好不得出现在 u4——租户隔离和「记得」一样要测。轨迹与工具评测见 工具调用与评测实践。

和 RAG / LangGraph 怎么接

东西	放哪
公司手册、政策条文	RAG / [LangChain RAG](/docs/ai/RAG/LangChain RAG 实践)
本会话 messages、中间 state	工作记忆；LangGraph checkpointer + thread_id
跨会话偏好、用户事实	长期画像；LangGraph Store（namespace 带 user_id）或业务表
订单是否发货	业务库工具，不写进记忆

LangGraph 对照（点到为止，API 以官方文档为准）：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()   # 短：同一 thread 续跑
store = InMemoryStore()          # 长：跨 thread
# store.put(("users", "u1"), "profile", {"prefs": {"style": "brief"}})
# app = graph.compile(checkpointer=checkpointer, store=store)

checkpointer 解决「刷新后对话还在」；Store / 业务库解决「换一个会话窗口还认识我」。二者都不是 RAG。

常见坑

1. 记忆 = 加长 context：窗口爆了就贵、就慢、就乱；要裁、要摘要、要分层。
2. 把 RAG 当记忆：共享知识检索解决不了「他叫我小王」。
3. 把记忆当真源：库存、余额、工单状态必须查库。
4. 无写入门禁：模型随口总结就落库，会固化幻觉和隐私。
5. 不能遗忘：合规与产品都要求改偏好、删画像。
6. 跨用户检索：长期库必须按 user_id 隔离，评测里要有 no-leak。

实践要点

1. 先窗口，再摘要，最后才上跨会话画像——多数 FAQ 做到窗口就够。
2. 写入用白名单 + 可覆盖；抽取可先规则后 LLM。
3. 评测至少覆盖：跨会话命中、跨用户不泄漏。
4. LangGraph：短用 checkpointer，长用 Store / 业务库，别把 Store 当成向量知识库去替代 RAG。
5. 面试挂钩：能画出三层，并说出「事实走真源、偏好走画像、当前任务走窗口」。

面试速记

1. 记忆分几层？ 工作记忆（窗口/state）、会话摘要、长期画像；业务真源单独一层。
2. 和 RAG？ RAG 补共享知识；记忆补「这个用户/这次任务」。
3. 为什么不能无限加 context？ 成本、延迟、噪声；要窗口 + 摘要。
4. 长期写什么？ 稳定偏好与已确认事实；工具轨迹和一次性任务不写。
5. LangGraph 怎么对应？ checkpointer = 短（thread）；Store = 长（跨 thread，按 user 隔离）。