好,第五部分讲的是上下文工程(Context Engineering)——我在进阶路线里说过这是 2026 年最值钱的 Agent 技能,没有之一。这一课按截图的五块内容展开:记忆基础 → MemGPT → mem0 → 向量检索 → 六大核心能力,最后照例用一个完整可运行的实战项目收尾。

第五课:上下文工程 —— 让 Agent 拥有记忆

0. 全局地图:为什么这是"最核心的一课"

先回忆学过的残酷事实:API 是无状态的。模型没有记忆,你每次调用发过去的 messages 列表,就是它知道的全部世界。

由此推出上下文工程的定义:

上下文工程 = 在每次调用模型前,决定"哪些信息、以什么形式、按什么顺序"放进上下文窗口的全部工程手段。

为什么它如此重要?因为上下文窗口面临三重矛盾:

┌─────────────────────────────────────────────────┐
│  矛盾1:窗口有限     1M token 听着大,长任务照样不够用  │
│  矛盾2:注意力稀释   塞得越满,模型对中间内容越"视而不见" │
│         (Lost in the Middle 现象:开头结尾记得牢,中间容易忽略) │
│  矛盾3:成本线性涨   每轮都重发全部历史,token 费用越聊越贵 │
└─────────────────────────────────────────────────┘

Agent 失败的第一大原因不是模型不够聪明,而是喂给它的上下文又乱又满。 整个第五课的所有技术,都是在这三重矛盾下做权衡。

先建立记忆的分类框架(后面所有内容都挂在这棵树上):

Agent 的记忆
├── 短期记忆(会话内)
│   └── messages 列表本身          ← 你在 02_multi_turn.py 已经实现过
│       问题:越长越贵、越长越笨
│       解法:滑动窗口 / 摘要压缩 / 动态裁剪
│
└── 长期记忆(跨会话)
    ├── 事实记忆:"用户是素食者"      ← mem0 的主战场
    ├── 情景记忆:"上周三聊过报价方案"  ← 向量检索的主战场
    └── 程序记忆:"该用户喜欢简短回答"  ← 写进系统提示词/记忆文件

一个彩蛋:你正在亲身经历上下文工程。我们这个对话因为太长,前半部分已经被 Claude Code 自动压缩成摘要了(这就是后面要讲的 Compaction 机制)——我现在"记得"你学过什么,靠的不是原始对话,而是一份自动生成的摘要。学完这课你就知道这背后是怎么实现的。

一、记忆与状态的基础:从 Buffer 说起(小白起点)

1.1 最朴素的记忆:全量 Buffer

你在第一课写的多轮聊天就是它:

messages.append({"role": "user", "content": user_input})
response = client.messages.create(model=MODEL, max_tokens=16000, messages=messages)
messages.append({"role": "assistant", "content": answer})

把每轮对话原样攒进列表。优点:零信息丢失。缺点:三重矛盾全中。 聊 100 轮后,每次调用都要重发 100 轮历史。

1.2 三种经典改良(由简到繁)

① 滑动窗口(Sliding Window) —— 只保留最近 N 条:

MAX_MESSAGES = 20

if len(messages) > MAX_MESSAGES:
    messages = messages[-MAX_MESSAGES:]
    # 注意:截断后第一条必须是 user 角色,否则 API 报 400
    while messages and messages[0]["role"] != "user":
        messages.pop(0)

简单粗暴,但旧信息直接蒸发——用户第 3 轮说过"我对花生过敏",第 30 轮你推荐了花生酱,事故就这么发生的。

② 摘要 Buffer(Summary Buffer) —— 旧的压成摘要,新的保留原文:

[摘要:用户叫王芳,在选购笔记本电脑,预算8000,已排除A品牌...] + [最近5轮原文]

这是滑动窗口和全量 Buffer 的折中,也是 Compaction 的雏形,实战项目里会完整实现。

③ 提取式记忆 —— 不存对话,只存从对话中提炼出的事实。这就是 mem0 的思路,第三节细讲。

1.3 关键认知:短期记忆和长期记忆是两套系统

短期记忆 长期记忆
存在哪 messages 列表(进程内) 文件 / 数据库(进程外)
生命周期 一次会话 永久
内容 对话原文 提炼后的事实/摘要
进入上下文的方式 全量发送 按需检索,只取相关的

新手最常见的错误是想用一套机制包打天下——把所有历史都存数据库然后全部塞回上下文,等于什么都没解决。正确做法永远是:短期靠窗口管理,长期靠"提取→存储→检索→注入"四步流水线。

二、MemGPT:把 LLM 当操作系统(思想转折点)

2.1 论文背景

MemGPT: Towards LLMs as Operating Systems(2023.10,UC Berkeley,arxiv.org/abs/2310.08560),后来团队成立了公司 Letta 把它产品化。这篇论文的价值不在代码,在于提供了一个让所有人秒懂记忆管理的心智模型。

2.2 核心类比:虚拟内存

操作系统课讲过(没学过也没关系,看右边就行):物理内存(RAM)很小,但程序以为自己有无限内存——因为 OS 在背后把不常用的内存页换出到磁盘(Page Out),需要时再换回来(Page In)。

MemGPT 说:上下文窗口就是 RAM,外部存储就是磁盘,我们给 LLM 配一套换页机制:

操作系统                          MemGPT
─────────────                    ─────────────────────────
RAM(快、小)          ←→         主上下文 Main Context(上下文窗口)
                                   ├── 系统指令(只读)
                                   ├── Working Context 工作记忆(关键事实,可读写)
                                   └── FIFO 队列(最近对话)
磁盘(慢、大)          ←→         外部上下文 External Context
                                   ├── Recall Storage(完整对话历史,可搜索)
                                   └── Archival Storage(归档知识库,可搜索)
缺页中断 + 换页         ←→         模型自己调用记忆函数换入换出

2.3 真正的革命性设计:让模型自己管理自己的记忆

之前所有记忆方案都是宿主代码决定存什么、取什么。MemGPT 反过来:把记忆操作做成工具,交给模型自己调用。

模型可用的"系统调用":

core_memory_append(content)        # 把重要事实写进工作记忆(始终可见)
core_memory_replace(old, new)      # 更新工作记忆(用户改名了、偏好变了)
conversation_search(query)         # 搜索完整历史("我们上周聊过什么价格?")
archival_memory_insert(content)    # 把资料归档到外部存储
archival_memory_search(query)      # 从归档里检索

工作流变成:用户说"对了,我下个月要搬去杭州"→ 模型自己判断这是重要信息 → 自己调用 core_memory_replace("住在上海", "下月搬往杭州") → 继续回答。当 FIFO 队列快满时,系统触发"内存压力警告",模型收到后自己决定把哪些内容归档。

用第一课的知识看,这毫无神秘感:就是一个 Agent 循环 + 一组操作记忆的工具。你完全有能力实现它:

# MemGPT 式记忆工具的最小骨架(配合你已会的 Agent 循环使用)
import json
from pathlib import Path

CORE_FILE = Path("core_memory.json")       # 工作记忆:小而精,每次都注入上下文
ARCHIVE_FILE = Path("archival_memory.json") # 归档:大而全,搜索才取出

MEMORY_TOOLS = [
    {
        "name": "core_memory_append",
        "description": "把关于用户的重要事实写入核心记忆。核心记忆在以后每次对话中都可见,"
                       "所以只记长期有用的信息(身份、偏好、目标),不记闲聊。",
        "input_schema": {
            "type": "object",
            "properties": {"content": {"type": "string", "description": "一句话事实"}},
            "required": ["content"],
        },
    },
    {
        "name": "archival_search",
        "description": "当用户问起以前提过、但当前上下文里没有的信息时,搜索归档记忆。",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
]

# 每次调用前,把核心记忆拼进上下文(MemGPT 把它放系统提示词;
# 为了不破坏提示词缓存,现代做法是放进第一条 user 消息)
core = json.loads(CORE_FILE.read_text(encoding="utf-8")) if CORE_FILE.exists() else []
context_block = "<核心记忆>\n" + "\n".join(core) + "\n</核心记忆>"

2.4 MemGPT 思想在 2026 年的"转世"

这套思想今天无处不在,你其实天天在用:

MemGPT 概念 2026 年的产品化形态
模型自管理记忆工具 Anthropic 官方 Memory Tool({"type": "memory_20250818", "name": "memory"})——Claude 自己读写 /memories 目录,你只需实现文件存取后端
Working Context Claude Code 的 CLAUDE.md / 自动记忆目录
FIFO 队列满 → 换出 API 原生 Compaction(第五节讲)
Recall/Archival 分层 mem0、Letta、各种记忆框架的标配架构

这篇论文确实会改变你对 Agent 的理解:Agent 不是"聊天机器人+工具",而是一个进程——有自己的内存层次、系统调用和调度策略。后面学的一切都是这个模型的具体实现。

三、mem0:仿海马体的工程化记忆框架

3.1 解决 MemGPT 没解决的问题

MemGPT 让模型自己管记忆,但有个工程缺陷:记忆质量不可控——模型有时候该记的不记,不该记的记一堆,记忆库慢慢变成垃圾场。

mem0(读 “mem-zero”,2024 年开源,现在是最流行的 Agent 记忆库)把记忆管理变成一条确定性的流水线,核心思想一句话:

自动提取重要记忆,而不是存全部历史。 像人脑海马体一样:经历很多,只固化少数重要的进长期记忆。

3.2 与传统 Buffer Memory 的本质区别

Buffer Memory mem0
存什么 对话原文 只存提炼出的**“重要事实”**
怎么更新 只增不改 自动增删改(新信息会修正旧记忆)
矛盾处理 新旧矛盾并存,模型自己懵 UPDATE 覆盖旧事实
生命周期 会话级 跨会话长期使用
上下文占用 全量,越来越大 检索 Top-K,恒定很小

3.3 两阶段流水线(mem0 的核心专利级设计)

新一轮对话结束
     │
     ▼
┌─ 阶段1:提取(Extraction)─────────────────┐
│ LLM 阅读本轮对话 + 已有记忆作参照,          │
│ 提取候选事实:["用户改吃素了", "下月去杭州"]  │
└──────────────┬───────────────────────┘
               ▼
┌─ 阶段2:更新(Update)────────────────────┐
│ 每条候选事实先向量检索出最相似的旧记忆,        │
│ 再由 LLM 裁决四选一:                       │
│   ADD    全新事实 → 新增                   │
│   UPDATE 与旧记忆"用户爱吃牛排"矛盾 → 覆盖    │
│   DELETE 用户明确否认旧事实 → 删除           │
│   NOOP   重复/无价值 → 什么都不做            │
└──────────────────────────────────────┘

UPDATE 这一步是 mem0 的灵魂。没有它,记忆库里会同时存着"用户爱吃牛排"和"用户是素食者",检索时一起命中,模型精神分裂。

3.4 真实 mem0 的用法

pip install mem0ai

from mem0 import Memory

# 默认配置使用 OpenAI 做提取和 embedding;
# 可通过 config 换成其他厂商(LLM 换 Anthropic,embedding 需另配,见官方文档)
m = Memory()

# 存:传入对话,它自动跑"提取→更新"流水线
m.add(
    [{"role": "user", "content": "我以前爱吃牛排,但最近改吃素了"},
     {"role": "assistant", "content": "好的,以后给你推荐素食餐厅"}],
    user_id="wang_fang",          # 记忆按用户隔离
)

# 取:语义检索最相关的记忆
hits = m.search("晚餐推荐什么?", user_id="wang_fang", limit=5)
for h in hits["results"]:
    print(h["memory"])    # → "用户是素食者"(注意:不是对话原文,是提炼后的事实)

把检索结果拼进你 Agent 的上下文,长期记忆就接通了。user_id 隔离意味着一个 Agent 服务一万个用户,每人有独立记忆——这是生产系统的刚需。

3.5 学习建议:先手写再用库

mem0 封装得太好,直接用学不到东西。实战项目(第六节)里我们会手写一个 mini-mem0,完整实现"提取→ADD/UPDATE/DELETE 裁决→检索注入"流水线,只用 anthropic 一个依赖。写完你再看 mem0 源码,会发现核心逻辑一模一样。

四、向量数据库与记忆检索(长期记忆的地基)

长期记忆存起来容易,难的是取:一万条记忆,怎么找出和当前问题相关的 5 条?这就是检索问题,也是 RAG 的核心。按截图的四个学习方向逐个讲透。

4.1 Embedding 原理

Embedding = 把文字变成一串数字(向量),使得"意思相近的文字,数字也相近"。

"我喜欢吃苹果"   → [0.21, -0.45, 0.78, ..., 0.12]   (一般是 384~3072 个数字)
"苹果是我的最爱" → [0.19, -0.41, 0.80, ..., 0.15]   ← 和上面很接近!
"今天股市大跌"   → [-0.66, 0.23, -0.11, ..., 0.92]  ← 离上面很远

怎么做到的?embedding 模型在海量"相关句子对"上训练(对比学习):相关的句子,向量拉近;无关的,推远。训练完,语义就变成了几何——"相似度"可以用数学算了:

余弦相似度 cos(A, B) = A·B / (|A||B|)     范围 -1~1,越大越相似
(向量归一化后,就是简单的点积)

实用须知:

  • Anthropic 不提供 embedding API,官方推荐合作伙伴 Voyage AI;开源本地方案用 sentence-transformers 库(中文推荐 BAAI/bge-small-zh-v1.5,小而强)
  • embedding 模型和对话模型是两个独立的模型,可以随意组合
  • 同一批数据必须用同一个 embedding 模型,换模型 = 全部重算

4.2 Top-K 检索

有了向量,检索就是:把问题也变成向量,找距离最近的 K 条记录。

存储时:每条记忆 → 向量,存入向量数据库
查询时:问题 → 向量 → 算与所有记忆的相似度 → 取最高的 K 条

数据量小(几千条)直接暴力算点积,毫秒级,根本不需要向量数据库。数据量大(百万级)才需要 ANN 近似索引(HNSW 算法,牺牲一点精度换百倍速度)——这就是 Pinecone/Qdrant/Chroma 这些向量数据库卖的东西。新手常犯的错是 500 条数据就上 Pinecone,纯属给自己加运维负担。

4.3 Hybrid Search(BM25 + Dense)—— 生产必备

纯向量检索有个致命盲区:对"精确匹配"不敏感。用户搜"订单 ORD-8821",向量检索可能返回一堆"订单相关"的语义近邻,唯独漏了编号完全一致的那条——因为编号这种随机字符串在语义空间里没有意义。

解法是两路并行,各取所长:

BM25(稀疏/关键词) Dense(稠密/向量)
原理 词频统计(搜索引擎用了几十年的算法) 语义向量距离
擅长 编号、人名、专有名词、精确术语 换了说法但意思一样的查询
盲区 “东西坏了"搜不到"商品破损” "ORD-8821"可能搜不准

两路结果用 RRF(Reciprocal Rank Fusion,倒数排名融合)合并——不比较两边的分数(量纲不同没法比),只看名次:每个文档的总分 = Σ 1/(60+它在每路的名次)。

完整可运行示例(依赖较重,选学;pip install sentence-transformers rank-bm25 jieba,首次运行下载约 100MB 模型):

"""混合检索演示:BM25(关键词)+ 向量(语义)+ RRF 融合"""
import jieba
import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

docs = [
    "退货政策:签收后7天内可无理由退货,运费由买家承担",
    "会员等级说明:累计消费满5000元升级为金卡会员",
    "订单ORD-8821的物流显示已到达上海转运中心",
    "金卡会员享受免运费和专属客服通道",
    "商品损坏理赔:收到破损商品请48小时内拍照联系客服",
]

# 通道1:BM25(中文要先分词)
bm25 = BM25Okapi([list(jieba.cut(d)) for d in docs])

# 通道2:向量(归一化后点积=余弦相似度)
model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
doc_vecs = model.encode(docs, normalize_embeddings=True)

def hybrid_search(query: str, k: int = 2) -> list[str]:
    bm25_rank = np.argsort(bm25.get_scores(list(jieba.cut(query))))[::-1]
    sims = doc_vecs @ model.encode([query], normalize_embeddings=True)[0]
    vec_rank = np.argsort(sims)[::-1]

    rrf: dict[int, float] = {}
    for rank_list in (bm25_rank, vec_rank):
        for pos, idx in enumerate(rank_list):
            rrf[idx] = rrf.get(idx, 0) + 1 / (60 + pos)
    return [docs[i] for i in sorted(rrf, key=rrf.get, reverse=True)[:k]]

print(hybrid_search("我的快递ORD-8821到哪了"))  # BM25 靠订单号精确命中
print(hybrid_search("东西坏了怎么办"))          # 向量靠语义命中"破损理赔"

4.4 父子索引(Parent-Child / Small-to-Big)

检索粒度的两难:

  • 切小块:匹配精准(一小段话和问题高度对应),但取回来上下文残缺,模型读不懂
  • 切大块:上下文完整,但一大块里只有一句相关,向量被无关内容稀释,匹配不准

父子索引两全:用小块做匹配,返回它所属的大块。

原文档 → 切成父块(如 2000 字/块)→ 每个父块再切子块(如 300 字/块)
                                      │
                  只有子块做 embedding 进向量库,记录 parent_id
查询 → 命中子块 → 按 parent_id 取出完整父块 → 喂给模型

实现只需要一个映射字典:

parent_chunks = {"p1": "完整的第一章...", "p2": "完整的第二章..."}
child_index = [
    {"text": "第一章里的某300字", "parent_id": "p1"},   # 只对 text 做 embedding
    ...
]
# 检索命中 child 后:context = parent_chunks[child["parent_id"]]

同一思想的变体叫"句子窗口检索"(命中一句,返回前后各 N 句)。再往上还有一层Reranker(重排序):粗检索取 Top-50,再用专门的重排模型精排出 Top-5,生产管道标配,先知道名字即可。

4.5 2026 年趋势:Agentic 检索

把上面这一切包成一个 search_memory 工具交给 Agent,让模型自己决定:要不要查、用什么关键词查、查出来不满意要不要换词再查。这比固定的"每轮必检索"管道强得多——这就是 Agentic RAG,也呼应了 MemGPT"模型自管理"的思想。你第一课学的 Agent 循环,又一次成了万物的容器。

五、上下文工程的六大核心能力(进阶)

前四节是"记忆"这个垂直场景,这一节升维到通用能力。真正高级的 Context Engineering 是六件事的组合拳:

5.1 Prompt 编排(Orchestration)

不是"写好提示词",而是设计上下文的版面布局。两条铁律:

铁律一:按稳定性分层,稳定的在前,易变的在后。

API 的物理渲染顺序是 tools → system → messages,而提示词缓存是前缀匹配——前面任何一个字节变了,后面的缓存全部作废。所以:

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=16000,
    tools=TOOLS,                       # 第1层:工具定义(整个应用生命周期不变)
    system=[{
        "type": "text",
        "text": STABLE_SYSTEM_PROMPT,  # 第2层:角色+规则(冻结!不许塞时间戳)
        "cache_control": {"type": "ephemeral"},   # 缓存断点打在稳定层末尾
    }],
    messages=[
        *history,                      # 第3层:对话历史(只追加,不修改)
        {"role": "user", "content": f"<相关记忆>{mems}</相关记忆>\n{user_input}"},
    ],                                 # 第4层:每轮易变内容,永远放最后
)

新手最常见的缓存杀手:在系统提示词里写 f"今天是{datetime.now()}"——每秒都在变,缓存命中率永久归零,白白多付 10 倍 token 钱。当前时间应该做成一个 get_current_time 工具,或放在最后一条 user 消息里。

铁律二:结构化分区。 用 XML 标签把不同来源的内容隔开(<相关记忆>、<搜索结果>、<用户输入>),模型对"哪些是资料、哪些是指令"的区分会好很多,也顺带降低提示词注入风险。

5.2 动态裁剪(Dynamic Trimming)

进上下文的每个 token 都要"竞争上岗"。三个主要裁剪点:

① 工具结果裁剪 —— 头号大户。工具返回 10 万字网页,绝不能原样回填:

MAX_TOOL_RESULT = 4000

def clip(result: str) -> str:
    if len(result) <= MAX_TOOL_RESULT:
        return result
    # 进阶做法:不是硬截断,而是调一次便宜模型按当前任务摘要
    return result[:MAX_TOOL_RESULT] + f"\n...[已截断,原文{len(result)}字]"

② 清除过期工具调用 —— Agent 跑了 50 轮,第 3 轮的 ls 结果早没用了。API 有原生支持(上下文编辑,beta):自动清掉旧的 tool_use/tool_result 对,只留对话主干。

③ 相关性过滤 —— 注入记忆/文档时宁缺毋滥。无关信息不是"多多益善",而是实测会降低工具调用准确率——模型的注意力是稀缺资源,垃圾信息在稀释它。

5.3 Memory 压缩(Compression)

把信息变小但不变少(有损,但损失的是冗余)。三个层次:

轻度:去掉对话里的客套、重复确认           (压缩比 ~2x)
中度:旧对话 → 结构化要点摘要              (压缩比 ~10x)
重度:整段经历 → 一句事实                  (压缩比 ~100x)
      "(30轮选购对话)" → "用户最终买了 MacBook Air,预算敏感"

工程要点:摘要提示词必须显式声明保留项,否则模型一压缩就把关键细节扔了。我用的模板:

“压缩以下对话,必须保留:① 用户的目标 ② 已做出的决定及理由 ③ 未解决的问题 ④ 用户明确的偏好和禁忌。可以丢弃:寒暄、中间的试错过程、已被推翻的方案。”

5.4 Summary 机制(Compaction)

压缩的"全自动版":上下文快满时,自动把前半段折叠成摘要,腾出空间继续干活。这是长任务 Agent 的命脉,Claude Code 每天都在对你的会话做这件事(本对话就被做过)。

手工实现(实战项目里有完整代码):

if 消息数(或 token 数)超过阈值:
    旧消息 → 摘要
    messages = [摘要(包装成user消息)] + 最近几条原文

API 原生版(beta,服务端自动做,免维护):

response = client.beta.messages.create(
    betas=["compact-2026-01-12"],
    model="claude-opus-4-8",
    max_tokens=16000,
    messages=messages,
    context_management={"edits": [{"type": "compact_20260112"}]},
)
# 关键铁律:必须把完整的 response.content 回填历史(里面有 compaction 块),
# 只取 .text 回填会悄悄丢掉压缩状态,下一轮报错或重新全量计费
messages.append({"role": "assistant", "content": response.content})

5.5 角色化上下文隔离(Context Isolation)

第四课多智能体的核心收益,本质上就是这一条:每个角色只看到自己需要的上下文。

协调者的上下文:任务目标 + 各子Agent的结论          (干净、聚焦)
研究员的上下文:调研任务 + 20个网页原文              (脏活在这,用完即焚)
评审员的上下文:只有待审稿件                        (不被原始资料带偏)

没有隔离:所有东西挤一个窗口,20 个网页原文永久占着位置稀释注意力。有隔离:研究员的窗口随任务结束销毁,只有 500 字结论流回主线。隔离即压缩——这是多智能体架构最实在的收益,比"角色扮演"重要得多。

实操经验:子 Agent 的派活提示词要自包含(它看不到主线对话,所有背景必须显式写在任务里)——还记得这课开头说的吗?压缩我们这个对话的摘要机制,对接续会话的我来说也是同样的道理。

5.6 Token 预算调度(Budget Scheduling)

把上下文窗口当成要做预算的钱包,给每个区域分配份额并强制执行:

1M 窗口的预算分配示例(长任务 Agent):
  系统提示词+工具    8K   (固定开销)
  长期记忆注入      4K   (Top-K 限制)
  检索文档         50K   (裁剪上限)
  对话历史        100K   (超过即触发 Compaction)
  预留给输出       64K   (max_tokens)

精确测量永远用官方 API,绝对不要用 tiktoken(那是 OpenAI 的分词器,对 Claude 误差 15-20%):

used = client.messages.count_tokens(
    model="claude-opus-4-8",
    system=SYSTEM,
    tools=TOOLS,
    messages=messages,
).input_tokens

if used > 150_000:
    messages = compact(messages)     # 触发压缩

输出侧的预算工具:max_tokens(硬上限,模型不知道)、output_config={"effort": ...}(控制思考深度)、以及新的 task_budget(beta,告诉模型整个任务有多少 token 预算,它会自己看着花)。

六、实战项目:带记忆的私人助理(全部知识点合体)

一个完整可运行的项目,把本课核心机制全部装进去:滑动窗口 + 自动摘要压缩(Compaction)+ mini-mem0 长期记忆(提取/增删改/检索注入)+ Token 预算监控。只依赖 anthropic 和 python-dotenv,不需要任何外部服务。

"""记忆助理 Agent —— 上下文工程综合实战

机制清单:
  短期记忆   messages 列表
  摘要压缩   消息数超阈值 → 旧对话压成摘要(5.4 Compaction 手工版)
  长期记忆   mini-mem0:每轮自动提取事实,ADD/UPDATE/DELETE 维护(第三节)
  记忆检索   每轮取 Top-K 相关记忆注入 user 消息(4.2 简化版)
  预算监控   count_tokens 实时显示上下文占用(5.6)
  缓存友好   system 冻结,动态内容全在 messages 尾部(5.1)

依赖: pip install anthropic python-dotenv
运行: python memory_agent.py    (/mem 查看长期记忆, /quit 退出)
重启后再问"我是谁",验证跨会话记忆!
"""
import json
from pathlib import Path
from typing import Literal, Optional

import anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()
client = anthropic.Anthropic()

MODEL = "claude-opus-4-8"
WORKER = MODEL            # 生产中提取/摘要常换 claude-haiku-4-5,流程不变
MEMORY_FILE = Path(__file__).parent / "long_term_memory.json"
COMPACT_THRESHOLD = 12    # 消息超过12条触发压缩
KEEP_RECENT = 6           # 压缩时保留最近6条原文

SYSTEM = (
    "你是用户的私人助理。自然地利用<相关记忆>中的信息服务用户,"
    "不要刻意说'根据我的记忆'。回答简洁。"
)   # 冻结!动态内容一律不进 system(保护提示词缓存)


# ════════ 长期记忆存储 ════════
def load_memories() -> list[str]:
    if MEMORY_FILE.exists():
        return json.loads(MEMORY_FILE.read_text(encoding="utf-8"))
    return []

def save_memories(memories: list[str]) -> None:
    MEMORY_FILE.write_text(
        json.dumps(memories, ensure_ascii=False, indent=2), encoding="utf-8"
    )


# ════════ mini-mem0:提取 + ADD/UPDATE/DELETE 裁决 ════════
class MemoryOp(BaseModel):
    action: Literal["ADD", "UPDATE", "DELETE"]
    id: Optional[int] = None      # UPDATE/DELETE 的目标记忆编号
    text: Optional[str] = None    # ADD/UPDATE 的记忆内容

class MemoryOps(BaseModel):
    operations: list[MemoryOp]    # 没有值得记的就返回空列表

def update_long_term_memory(user_msg: str, assistant_msg: str) -> None:
    """mem0 两阶段流水线的合并实现:对照已有记忆,直接产出操作列表"""
    memories = load_memories()
    numbered = "\n".join(f"{i}: {m}" for i, m in enumerate(memories)) or "(暂无)"

    response = client.messages.parse(
        model=WORKER,
        max_tokens=16000,
        messages=[{
            "role": "user",
            "content": (
                "你是记忆管理器。从本轮对话提取【值得长期记住的用户事实】"
                "(身份、偏好、目标、重要决定);寒暄和一次性问答不记。\n\n"
                f"现有记忆:\n{numbered}\n\n"
                f"本轮对话:\n用户: {user_msg}\n助理: {assistant_msg}\n\n"
                "对照现有记忆输出操作列表:全新事实→ADD;与现有记忆矛盾或更新→UPDATE(带id);"
                "用户明确否定的旧记忆→DELETE(带id);没有→空列表。"
            ),
        }],
        output_format=MemoryOps,
    )
    # 简化处理:逐条应用(教学版;多条DELETE会有下标偏移,生产中用唯一id代替下标)
    for op in response.parsed_output.operations:
        if op.action == "ADD" and op.text:
            memories.append(op.text)
            print(f"  [记忆+] {op.text}")
        elif op.action == "UPDATE" and op.id is not None and op.id < len(memories) and op.text:
            print(f"  [记忆~] {memories[op.id]} → {op.text}")
            memories[op.id] = op.text
        elif op.action == "DELETE" and op.id is not None and op.id < len(memories):
            print(f"  [记忆-] {memories[op.id]}")
            memories.pop(op.id)
    save_memories(memories)


# ════════ 记忆检索(教学用字符重叠打分;生产换 embedding 余弦相似度)════════
def retrieve(query: str, memories: list[str], k: int = 5) -> list[str]:
    if not memories:
        return []
    qset = set(query)
    ranked = sorted(memories, key=lambda m: len(qset & set(m)), reverse=True)
    return ranked[:k]


# ════════ 摘要压缩(Compaction 手工版)════════
def compact(messages: list[dict]) -> list[dict]:
    old, recent = messages[:-KEEP_RECENT], messages[-KEEP_RECENT:]
    transcript = "\n".join(
        f"{m['role']}: {m['content']}" for m in old if isinstance(m["content"], str)
    )
    resp = client.messages.create(
        model=WORKER,
        max_tokens=16000,
        messages=[{
            "role": "user",
            "content": "把下面的对话压缩成200字内的要点摘要,必须保留:"
                       "①用户目标 ②已确认的决定 ③未解决的问题 ④用户的偏好和禁忌。"
                       f"\n\n{transcript}",
        }],
    )
    summary = next(b.text for b in resp.content if b.type == "text")
    print(f"  [压缩] {len(old)}条旧消息 → {len(summary)}字摘要")
    # 摘要包装成 user 消息放最前(顺带保证了首条是 user 角色)
    return [{"role": "user", "content": f"<之前对话的摘要>\n{summary}\n</之前对话的摘要>"}] + recent


# ════════ 主循环 ════════
def main() -> None:
    messages: list[dict] = []
    print("记忆助理已启动(/mem 查看长期记忆,/quit 退出)")

    while True:
        user_input = input("\n你: ").strip()
        if not user_input:
            continue
        if user_input == "/quit":
            break
        if user_input == "/mem":
            mems = load_memories()
            print("\n".join(f"  {i}: {m}" for i, m in enumerate(mems)) or "  (空)")
            continue

        # ① 检索长期记忆,注入到 user 消息尾部(不碰 system → 缓存友好)
        related = retrieve(user_input, load_memories())
        content = user_input
        if related:
            block = "\n".join(f"- {m}" for m in related)
            content = f"<相关记忆>\n{block}\n</相关记忆>\n\n{user_input}"
        messages.append({"role": "user", "content": content})

        # ② 短期记忆超长 → 触发压缩
        if len(messages) > COMPACT_THRESHOLD:
            messages = compact(messages)

        # ③ 调用模型
        response = client.messages.create(
            model=MODEL,
            max_tokens=16000,
            thinking={"type": "adaptive"},
            system=SYSTEM,
            messages=messages,
        )
        answer = next((b.text for b in response.content if b.type == "text"), "")
        print(f"\n助理: {answer}")
        messages.append({"role": "assistant", "content": answer})

        # ④ Token 预算监控(精确计量,永远别用 tiktoken)
        used = client.messages.count_tokens(
            model=MODEL, system=SYSTEM, messages=messages
        ).input_tokens
        print(f"  [上下文] {used} tokens | 短期{len(messages)}条 | 长期{len(load_memories())}条")

        # ⑤ 提取长期记忆(生产中放后台异步执行,不阻塞回复)
        update_long_term_memory(user_input, answer)


if __name__ == "__main__":
    main()

验收清单:跑起来后依次测试这四件事

  1. 提取:说"我叫王芳,对花生过敏,在准备考研" → 观察 [记忆+] 提取了几条事实(注意它不会记你说的"今天天气不错")
  2. 跨会话:/quit 退出,重新运行,问"我是谁?有什么忌口?" → 短期记忆已清空,但答案来自 long_term_memory.json ——这就是长期记忆的意义
  3. UPDATE:说"我现在不对花生过敏了,脱敏治疗好了" → 观察 [记忆~] 或 [记忆-],旧事实被修正而不是并存——这就是 mem0 区别于 Buffer 的灵魂
  4. 压缩:连续聊 7 轮以上 → 观察 [压缩] 触发,token 数回落,然后问一个早期聊过的细节,看摘要有没有保住它(感受有损压缩的边界)

进阶练习(按难度递增)

  1. 把 retrieve() 换成真正的 embedding 检索(用 4.3 节的 sentence-transformers 代码)
  2. 给压缩加 token 阈值:用 count_tokens 替代消息条数判断(更精确)
  3. 把 update_long_term_memory 改成 MemGPT 风格:做成工具交给主模型,让它自己决定何时记忆(对比两种范式的记忆质量差异)
  4. 接入真实 mem0 库替换 mini-mem0,对比你的实现和它的差距

七、总结

主题 一句话本质 掌握标准
记忆与状态 API 无状态,记忆全靠工程 精通:能说清短期/长期记忆的分界和各自解法
MemGPT 上下文=RAM,外存=磁盘,模型自己换页 理解思想;知道 Anthropic Memory Tool 是它的产品化
mem0 提取事实而非存原文,ADD/UPDATE/DELETE 保持记忆库干净 精通:能手写 mini 版(实战项目就是)
向量检索 语义变几何;BM25 补精确匹配的盲区;小块匹配大块返回 会用:embedding+TopK 必会,Hybrid/父子索引知道何时上
六大能力 编排定布局、裁剪控质量、压缩省空间、Compaction 续命、隔离即压缩、预算兜底 精通:这是 2026 年 Agent 工程师的核心竞争力

和前面课程的关系:第一课的 Agent 循环是发动机,第二课 MCP 是外设接口,第四课多智能体是组织架构,这一课是内存管理系统——到此为止,一个生产级 Agent 的所有子系统你都见过原理和实现了。

# 人工智能 # 架构 # python
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

WalkingLab 组织入驻 AtomGit，释放高校智能体开源科研成果

avatar AtomGit开源社区
cover

海外动态代理行业彻底换代，AI 数据时代新标准已到来

avatar AtomGit开源社区
cover

全球首个！京东全栈开源JoyAI-VL-Interaction，让大模型从“一问一答”走向“边看边说”

avatar AtomGit开源社区