给 AI Agent 加上持久记忆:用 Supermemory 构建 Python 私人训练助手
多数 LLM Agent 每次启动都像第一次见到用户。用户姓名、上次对话内容、最近处理过的任务、偏好设置,都不会自动保留下来。只靠模型上下文窗口,记忆只能停留在单次会话里;一旦脚本重启,状态就丢失。
要解决这个问题,需要把“用户相关事实”从临时上下文中抽离出来,交给专门的记忆层保存和检索。这里使用 Supermemory 作为托管式 memory API,再配合 OpenAI Agents SDK,构建一个 Python 私人训练助手:它能记录训练内容、基于历史推荐下一次训练,并且在不同脚本运行之间保留记忆。
整套实现很小:
- Supermemory 负责长期记忆
- OpenAI Agents SDK 负责 Agent 循环
- 业务代码只有两个 Python 文件和一个
pyproject.toml
运行环境要求:
- Python 3.10 及以上
- OpenAI 账号
- Supermemory 账号
- 基本命令行操作能力
Supermemory 是什么
Supermemory 可以理解为面向 Agent 的记忆 API。向它写入关于用户的文本后,它会在后续读取时返回一个压缩后的用户视图:用户是谁、最近在做什么、有哪些稳定偏好。
嵌入、索引、检索都在 Supermemory 内部完成,Agent 代码只需要做两件事:
- 写入记忆
- 按用户读取记忆
在 LongMemEval 基准测试中,Supermemory 的事实召回率为 81.6%。紧随其后的 Zep 为 71.2%,差距约 10 个百分点,等价于每 10 个用户问题多答对约 1 个。其开源仓库也已经获得 22k+ GitHub stars。
记忆系统与 RAG 的区别
记忆系统和 RAG 解决的不是同一类问题。
RAG 面向“文档语料”。开发者通常在部署阶段准备好一批资料,例如:
- 产品手册
- 支持文档
- 内部知识库
这些数据在运行时被检索,变化频率不高。它解决的是“产品本身知道什么”。
记忆系统面向“用户”。随着对话进行,系统持续写入用户特定事实,存储会不断增长。它解决的是“只有这个用户自己知道什么”。
| 机制 | 数据来源 | 适合回答的问题 |
|---|---|---|
| RAG | 预先准备好的文档库 | “退款政策是什么?” |
| Memory | 与用户交互中积累的事实 | “我上周卧推是多少?” |
实际产品里,这两者通常并存:
- RAG 回答产品知识
- Memory 回答用户历史与偏好
同一个 Agent,背后是两套数据源,两种职责。
用户画像:静态事实与动态事实
Supermemory 的关键概念是用户画像。写入的内容会被归类到两个桶中:
- 静态事实:长期稳定、不常变化
- 动态事实:近期活动、当前状态、最近行为
当某些动态模式重复出现,它们会逐渐提升为静态事实。读取用户画像时,一次调用会返回这两部分,以及匹配到的记忆片段。
这个拆分很重要,因为静态和动态回答的是同一个用户的不同侧面。
| 静态事实 | 动态事实 |
|---|---|
| 在家训练,器械只有哑铃和引体向上杆 | 当前重点:上肢力量 |
| 左膝有伤,不做深蹲 | 最近一次卧推:185 lb,4 组,每组 5 次 |
| 年底前想把卧推提高 20 lb | 本周在做 grease-the-groove 引体训练 |
| 只在晚上训练,不在早上训练 | 昨天跑了 5k,用时 28 分钟 |
一个训练推荐器需要同时用到两类信息:
- 静态事实排除不适合的动作,比如只能在健身房完成的训练
- 动态事实决定当前阶段更适合安排什么内容
在底层,Supermemory 实际上把本来需要自己做的四件事都接管了:
- 存储原始记忆
- 对每个片段做 embedding
- 在读取时做相似度检索
- 从日志中提取用户画像事实
这些过程都不会出现在业务代码里。
另外,每条记忆都带有开发者指定的 container_tag。读取时再传入相同标签,就能限定只返回对应用户的数据。演示里固定写死一个标签,真实系统通常会从认证用户标识计算得到,比如 JWT 中的用户 ID。
环境准备
这个训练助手需要两类 API Key:
- Supermemory API Key
- OpenAI API Key
然后准备一个 Python 项目和三个依赖。正式写 Agent 代码之前,先做一次最小验证,确认 API 和 SDK 都正常可用。
获取 API Key
Supermemory 的 API Key 在:
https://console.supermemory.ai
不是:
app.supermemory.ai
app 子域是面向普通用户的记忆产品界面,不提供 API Key 页面。需要直接进入开发者控制台。
在 console.supermemory.ai 中:
- 登录
- 点击侧边栏 API Keys
- 点击 Create API Key
- 为 Key 命名,这里可使用
datacamp-tutorial - 复制生成的 Key,前缀为
sm_
OpenAI Key 可在 https://platform.openai.com/api-keys 获取。
在项目根目录创建 .env 文件,并写入两项密钥。不要提交到版本库。
SUPERMEMORY_API_KEY=sm_your_key_here
OPENAI_API_KEY=sk-your_key_here
Supermemory 的免费额度足够覆盖这个演示,无需绑定支付信息。具体限制以其定价页为准。
安装依赖
项目使用 uv 做初始化和执行。若尚未安装 uv,先按 astral 官方方式安装。
初始化项目:
uv init supermemory-trainer
cd supermemory-trainer
uv init 会自动生成 README.md 和 hello.py:
README.md可以删除hello.py稍后会被覆盖
安装三个依赖:
supermemory==3.37.0:记忆客户端,固定为已验证版本openai-agents:OpenAI Agents SDK,包名带连字符,导入路径是agentspython-dotenv:读取.env
uv add supermemory==3.37.0 openai-agents python-dotenv
最终的 pyproject.toml 如下:
[project]
name = "supermemory-trainer"
version = "0.1.0"
description = "Personal exercise trainer agent built with Supermemory and the OpenAI Agents SDK."
requires-python = ">=3.10"
dependencies = [
"openai-agents>=0.10.2",
"python-dotenv>=1.2.1",
"supermemory==3.37.0",
]
先做一轮最小验证
在真正写 Agent 之前,先用一条句子验证 Supermemory 的写入与读取流程。这个脚本会:
- 写入一条事实
- 等待后台处理
- 读取用户画像
- 打印静态事实、动态事实和检索结果
把项目根目录里的 hello.py 替换成下面的内容。
import time
from dotenv import load_dotenv
from supermemory import Supermemory
load_dotenv()
client = Supermemory()
USER_ID = "demo_warmup"
response = client.add(
content="The user is learning Supermemory by building a personal trainer agent.",
container_tag=USER_ID,
)
print(f"client.add() -> id={response.id} status={response.status}")
print("Waiting 20 seconds for processing...")
time.sleep(20)
prof = client.profile(container_tag=USER_ID, q="learning")
print(f"profile.static ({len(prof.profile.static)}): {prof.profile.static}")
print(f"profile.dynamic ({len(prof.profile.dynamic)}): {prof.profile.dynamic}")
print(f"search_results.results ({len(prof.search_results.results)}):")
for r in prof.search_results.results[:3]:
print(f" - {r['memory']} (similarity={r['similarity']:.3f})")
这里有几个关键点:
load_dotenv()会在构造Supermemory()之前把SUPERMEMORY_API_KEY加载进环境变量- 客户端会自动读取这个环境变量
container_tag="demo_warmup"把这条记忆限定在一个临时用户范围内
执行脚本:
uv run python hello.py
预期输出:
client.add() -> id=zNLsJBrY1PZupAeZ3Qn6EL status=queued
Waiting 20 seconds for processing...
profile.static (0): []
profile.dynamic (1): ['Building a personal trainer agent to learn Supermemory.']
search_results.results (1):
- Building a personal trainer agent to learn Supermemory. (similarity=0.650)
有三点需要关注。
第一,client.add() 会立刻返回,状态是 queued。Supermemory 的处理是异步的,所以写入成功不等于立刻可检索。
第二,20 秒等待是必须的。它给 embedding 和事实抽取流水线留出处理时间。省略这一段,读取结果通常为空,看起来像是脚本失败,实际不是。
第三,最有价值的是 profile.dynamic。输入文本是:
The user is learning Supermemory by building a personal trainer agent.
输出却变成了:
'Building a personal trainer agent to learn Supermemory.'
也就是说,Supermemory 已经把第三人称叙述转成了关于用户的动态事实。这就是画像提取器的作用。
profile.static 此时为空是正常现象。静态事实通常要在多条相关日志积累后才会逐渐形成,因此读取逻辑不能假设它一定存在。
用 Supermemory 构建私人训练助手
接下来把 client.add() 和 client.profile() 封装进两个 Agent 工具中,让读写记忆随着用户对话自动发生。
训练历史是很适合放进记忆层的场景:
- 器械条件不在模型训练数据里
- 伤病信息不在模型训练数据里
- 最近训练内容需要按用户逐步积累
项目结构
完整项目只需要下面这些文件:
supermemory-trainer/
├── .env # 真实密钥,加入 .gitignore
├── .env.example # 占位示例,可提交
├── .gitignore
├── .python-version
├── main.py # agent 定义、system prompt、REPL 循环
├── pyproject.toml
└── tools.py # log_workout 和 suggest_next_session
职责划分如下:
tools.pylog_workout:写训练记录到 Supermemorysuggest_next_session:从 Supermemory 读取用户画像
main.py- 定义 Agent
- 注册工具
- 启动交互循环
编写 main.py
训练助手的关键不在于 SDK 样板代码,而在于 system prompt 里的一条约束:关于用户的事实必须通过工具调用获得,模型自身不能假设自己有长期记忆。
先写 main.py 的导入和系统提示词:
import asyncio
from agents import Agent, Runner, SQLiteSession
from tools import log_workout, suggest_next_session
SYSTEM_PROMPT = """You are a personal exercise trainer who logs the user's
workouts and recommends what to do next.
You have no memory of the user's history on your own. Every fact about the
user lives in Supermemory and reaches you only through tool calls.
Two rules, no exceptions:
1. Whenever the user reports completing a workout, call log_workout immediately, before responding. Extract the exercise, sets, reps, weight, and any notes from what they said. If a value is missing, ask one short follow-up question instead of guessing. After logging, confirm in one short sentence and stop. Do NOT recommend the next session unless the user asks for one.
2. When the user explicitly asks what to do next (or asks for a recommendation, suggestion, or plan), call suggest_next_session first. Never recommend from your own training data. The tool returns the user's
recent activity, stable preferences, and matching past sessions. Reference those facts directly in your reply.
Keep replies concise (2-4 sentences). Be specific: name the exercise, sets, reps, and weight. Honor any injuries or equipment constraints the tool surfaces.
"""
这段 prompt 的作用非常直接:
- 规则 1 强制每次用户报告训练完成后,必须先调用
log_workout - 规则 2 强制每次用户询问接下来练什么时,必须先调用
suggest_next_session
如果没有这两条规则,Agent 会直接依据通用训练知识回答,记忆层就失去意义。
继续定义 Agent 和命令行聊天循环:
def build_agent() -> Agent:
return Agent(
name="Trainer",
instructions=SYSTEM_PROMPT,
tools=[log_workout, suggest_next_session],
model="gpt-5",
)
async def chat() -> None:
agent = build_agent()
session = SQLiteSession(session_id="trainer-cli")
print("Trainer ready. Type a message, or 'exit' to quit.\n")
while True:
try:
message = input("You: ").strip()
except (EOFError, KeyboardInterrupt):
print()
break
if not message:
continue
if message.lower() in {"exit", "quit"}:
break
result = await Runner.run(agent, message, session=session)
print(f"\nTrainer: {result.final_output}\n")
if __name__ == "__main__":
asyncio.run(chat())
这里有两行需要单独说明。
第一,tools=[log_workout, suggest_next_session] 注册了两个记忆相关工具。它们必须在 tools.py 中使用 @function_tool 装饰器,否则运行时 Agent 实际拿不到这些工具,即使构造过程本身不报错。
第二,SQLiteSession(session_id="trainer-cli") 只保存运行中这一个 Python 进程的短期上下文。脚本退出后,这部分会失效;而 Supermemory 中的长期记忆仍然保留。
还有一个运行边界需要注意:
main.py应作为脚本运行,不要放在 Jupyter 单元格中执行- Jupyter 的事件循环通常会与
asyncio.run()冲突 Supermemory()客户端本身是同步的,但放在异步工具函数中仍能工作,因为 Agents SDK 会在线程池中执行工具调用
编写记忆写入工具 log_workout
log_workout 负责把一次训练写入长期记忆。它接收结构化参数:
- 动作名
- 组数
- 次数
- 重量
- 可选备注
然后把这些参数拼成一条自然语言句子,再交给 client.add()。
打开 tools.py,先写导入和共享客户端:
from agents import function_tool
from dotenv import load_dotenv
from supermemory import Supermemory
load_dotenv()
USER_ID = "demo_user"
client = Supermemory()
这里的初始化顺序很重要:
- 必须先
load_dotenv() - 再构造
Supermemory()
如果顺序反了,客户端会在没有认证信息的情况下创建,第一次调用时才报 401,排查起来会比较绕。
然后加入 log_workout:
@function_tool
def log_workout(
exercise: str,
sets: int,
reps: int,
weight: float,
notes: str = "",
) -> str:
"""Log a completed workout to the user's memory.
Args:
exercise: Name of the exercise.
sets: Number of sets performed.
reps: Number of reps per set.
weight: Weight in pounds. Pass 0 for bodyweight or cardio.
notes: Optional notes about the session.
"""
print(f"[log_workout] {exercise=}{sets=}{reps=}{weight=}{notes=}")
content = f"Performed {exercise}: {sets} sets of {reps} reps at {weight} lbs."
if notes:
content += f" Notes: {notes}"
response = client.add(content=content, container_tag=USER_ID)
print(f"[log_workout] -> id={response.id} status={response.status}")
return f"Logged {exercise} ({sets}x{reps} @ {weight} lb)."
几个实现细节:
1. @function_tool 的 docstring 很重要
函数的 docstring 和 Args: 描述会暴露给 LLM,作为它决定是否调用工具、如何填参数的依据。这里不是普通注释,而是 Agent 与工具之间的契约。
2. 发送自然语言比 JSON 更合适
这里传给 client.add() 的是自然语言句子,而不是 JSON。虽然 JSON 也能存,但事实抽取质量会下降,因为抽取模型缺少可概括的叙述结构。
例如这一句:
Performed bench press: 4 sets of 5 reps at 185.0 lbs.
对于抽取器来说要比结构化 JSON 更容易转成稳定的用户事实。
3. 终端打印用于观察工具调用
两次 print() 会把工具调用过程打印到终端,便于确认 Agent 是否真的按预期走工具路径。
示例输出:
[log_workout] exercise='bench press' sets=4 reps=5 weight=185.0 notes=''
[log_workout] -> id=xY7AK3qLzBPx5Vd2HnRf1M status=queued
看到 status="queued" 说明写入已提交,但还没有完成后台处理。这与前面的 warm-up 脚本一致。
编写记忆读取工具 suggest_next_session
suggest_next_session 是读取侧工具。它的价值在于:一次 client.profile() 调用同时返回三类信息。
profile.static:稳定偏好与长期约束profile.dynamic:近期活动search_results.results:与查询最相近的历史记忆
工具的目标不是直接生成训练建议,而是把这些信息拼成一段上下文,让 Agent 据此作答。
把下面代码追加到 tools.py 中:
@function_tool
def suggest_next_session(focus: str) -> str:
"""Fetch the user's training history and preferences for a given focus.
Returns a context string the agent can use to recommend the next session.
The agent is responsible for the actual recommendation. This tool only
surfaces what Supermemory knows about the user.
Args:
focus: What the user wants to train next (e.g. "upper body", "legs",
"cardio", "today"). Drives semantic search against past logs.
"""
print(f"[suggest_next_session] focus={focus!r}")
profile = client.profile(container_tag=USER_ID, q=focus)
static_facts = profile.profile.static
dynamic_facts = profile.profile.dynamic
matches = profile.search_results.results
print(
f"[suggest_next_session] static={len(static_facts)} "
f"dynamic={len(dynamic_facts)} matches={len(matches)}"
)
sections = []
if static_facts:
sections.append("Stable preferences and constraints:")
sections.extend(f"- {fact}" for fact in static_facts)
if dynamic_facts:
sections.append("Recent activity:")
sections.extend(f"- {fact}" for fact in dynamic_facts)
if matches:
sections.append("Closest matching past entries:")
for r in matches[:5]:
sections.append(f"- {r['memory']}")
if not sections:
return (
"No prior training history found for this user. "
"Ask the user about their goals, equipment, and recent training."
)
return "\n".join(sections)
返回值结构
在积累几条日志后,client.profile(container_tag=USER_ID, q=focus) 返回的大致结构如下:
profile.profile.static # [] (list[str])
profile.profile.dynamic # ["Performed bench press: 4 sets of 5 reps at 185.0 lbs", ...]
profile.search_results.results # [{"memory": "...", "similarity": 0.631, ...}, ...]
其中 search_results.results 的每一项是 Python dict,不是 Pydantic 对象,因此要用:
r["memory"]
r["similarity"]
不能写成:
r.memory
在 supermemory==3.37.0 下,使用属性访问会触发 AttributeError。
完整字段包括:
idmemoryrootMemoryIdmetadataupdatedAtversionsimilarityfilepathdocuments
为什么要做空值分支
前期数据很少时:
static可能为空matches也可能因为相似度阈值过滤而为空
因此读取逻辑必须对每一部分做存在性判断。训练助手最初几次日志主要依赖 dynamic 和 search_results,而不能假设静态画像已经形成。
运行第一轮会话:写入训练记录
现在可以启动 Agent,向 Supermemory 写入一些真实的训练数据。
运行:
uv run python main.py
建议依次输入几条内容:
- 卧推记录
- 一次 5k 跑步
- 一次硬拉
- 一条偏好说明,例如:
I only train at home, no gym.
每次用户报告完成训练时,Agent 都应调用一次 log_workout,并在终端显示工具日志。
对于这类短文本,写入后通常大约 12 秒左右就能通过 client.profile() 检索到。训练助手本身不会等待这个过程结束,Supermemory 会在后台异步完成处理。
输入 exit 结束第一轮会话后:
- Python 进程退出
SQLiteSession失效- 训练记录和偏好仍保存在 Supermemory 中,绑定在
container_tag="demo_user"下
这就是跨进程持久记忆与进程内临时会话的区别。
在新进程中验证记忆召回
进入第二轮会话之前,可以先单独验证第一轮写入的数据已经可检索。
打开一个新的 Python REPL,或者保存为临时脚本执行:
from dotenv import load_dotenv
from supermemory import Supermemory
load_dotenv()
client = Supermemory()
prof = client.profile(container_tag="demo_user", q="training")
print(f"static ({len(prof.profile.static)}): {prof.profile.static}")
print(f"dynamic ({len(prof.profile.dynamic)}):")
for fact in prof.profile.dynamic:
print(f" - {fact}")
print(f"matches ({len(prof.search_results.results)}):")
for r in prof.search_results.results[:5]:
print(f" - {r['memory']} (similarity={r['similarity']:.3f})")
一组实际可见的输出如下:
static (0): []
dynamic (5):
- Trains at home instead of a gym
- Performed deadlift: 3 sets of 5 reps at 225.0 lbs
- Performed 5k run in 26 minutes
- Reports no knee pain during bench press
- Performed bench press: 4 sets of 5 reps at 185.0 lbs
matches (5):
- Trains at home instead of a gym (similarity=0.682)
- Performed deadlift: 3 sets of 5 reps at 225.0 lbs (similarity=0.643)
- Performed bench press: 4 sets of 5 reps at 185.0 lbs (similarity=0.631)
- Performed 5k run in 26 minutes (similarity=0.585)
- Reports no knee pain during bench press (similarity=0.585)
这组结果能看出 Supermemory 做了几件业务代码之外的工作。
第一,用户只说过一次:
I only train at home, no gym.
读取时已经变成了标准化动态事实:
Trains at home instead of a gym
第二,卧推日志里的备注提到膝盖没有疼痛,结果被拆成了两条动态事实:
- 一条关于训练动作
- 一条关于身体状态
第三,四条原始日志最终变成了五条归一化动态事实,并带有相似度分数。这些拆分、规范化和匹配过程都没有写在训练助手里。
如果这时 dynamic 还是空的,通常只需要再等 10 秒重新执行一次。后台处理队列偶尔会有波动。
运行第二轮会话:在新进程中让 Agent 读回记忆
现在重新启动一个全新的 Python 进程:
uv run python main.py
这是完全新的解释器实例:
- 没有第一轮会话的进程内内存
- 没有共享缓存
- 如果 Agent 能回忆出历史,来源只能是 Supermemory
输入一句:
What should I do for my workout today?
此时 Agent 会调用:
suggest_next_session("today")
终端中应看到类似输出:
[suggest_next_session] focus='today'
[suggest_next_session] static=0 dynamic=5 matches=5
之后 Agent 会基于这些记忆生成当天训练建议。一次实际运行中,给出的建议是居家下肢训练,包括深蹲、弓步和台阶训练。
这个推荐之所以合理,是因为记忆层已经把以下事实返回给了 Agent:
- 之前做过卧推、硬拉和 5k,近期训练偏上肢或有氧
- 用户只在家训练,不去健身房
而这些信息来自同一次 client.profile() 调用。
模型回复措辞会有波动,但召回路径不会变:新进程读取旧数据,证明长期记忆生效。
关键实现判断与边界
这个方案能工作,依赖几个明确前提。
1. 记忆层不能靠 prompt 假装出来
如果 Agent 被允许直接从模型参数里回答“你上次练了什么”,就无法判断它是在复述训练数据中的通用模式,还是基于真实用户记忆在回答。必须强制要求用户事实全部走工具调用。
2. SQLiteSession 不是长期记忆
它只负责本地短期对话上下文,进程退出即丢失。真正的长期记忆由 Supermemory 承担。两者职责不同,不应混用概念。
3. status=queued 是正常行为
写入后短时间内查不到结果,不代表失败。Supermemory 的处理链路是异步的,验证脚本和调试步骤都需要留等待时间。
4. 静态事实形成较慢
初期主要依赖动态事实与相似检索结果。工具实现不能强依赖 profile.static 非空,否则前几次交互体验会很差。
5. 检索结果受相似度阈值影响
即使某条事实曾被写入,也不保证在所有查询下都会命中。因此读取工具必须接受“有些字段为空”的现实,而不是把它当作异常。
下一步怎么扩展
这个示例只有一个用户、两个工具和一个 CLI。要走向真实应用,优先扩展的方向主要有三个。
按真实用户隔离记忆
现在的实现里写死了:
USER_ID = "demo_user"
生产环境应改为基于认证用户生成 container_tag,例如:
container_tag = customer_id
或:
container_tag = "user_sarah"
这样每个用户的记忆天然隔离,读取时也用相同标签限定范围。
增加更多基于记忆的工具
同样模式可以继续扩展:
- deload 周安排
- PR 追踪
- 每周灵活性训练提醒
实现方式不变:
- 写入类工具调用
client.add() - 读取类工具调用
client.profile() - 共用相同的
container_tag
变化的只是记录什么、查询什么。
处理 Supermemory 调用失败
真实系统中,需要为这两个调用加异常处理:
client.add()client.profile()
可用:
try:
...
except supermemory.APIError:
...
这样临时网络失败或服务端抖动不会直接让 Agent 崩溃。在资源受限环境中,还应为请求配置超时。
这个模式适合哪些场景
“静态事实 + 动态事实”的拆分并不局限于训练助手。只要用户本身是真实信息来源,这种结构就成立。
-
客服 Agent
- 静态:账号偏好、已知问题
- 动态:当前工单、最近联系记录
-
编码 Agent
- 静态:偏好语言、框架、代码风格
- 动态:当前任务、最近修改文件
这类系统的共同点是:通用知识不够,用户历史才是决定答案质量的关键。
小结
这个训练助手的核心只有两条 API:
client.add()负责写入训练日志client.profile()负责一次性读回静态事实、动态事实和相似记忆
通过 container_tag 绑定用户,Supermemory 提供了跨进程、跨会话的长期记忆;通过工具约束,Agent 的回答被强制建立在用户真实历史之上,而不是建立在模型的泛化常识之上。
如果系统里还需要回答产品知识问题,可以把这套记忆层和 RAG 并列使用:一个回答“用户是谁、刚做过什么”,另一个回答“产品有哪些规则、文档里写了什么”。
关于
关注我获取更多资讯