Agno MemoryManager 完全指南:基于 PostgreSQL 的用户记忆 CRUD、智能检索与 Agent 集成
Agno 的 MemoryManager 是负责用户记忆持久化与智能管理的核心组件,它把“用户说了什么、偏好是什么”这类信息从一次性对话中抽离出来,落库为可增删改查、可按需检索的长期记忆。本文以 cookbook/11_memory/memory_manager/ 目录下的五个实战示例为骨架,结合 MemoryManager 源码 与其单元测试,系统讲解记忆的增删改查(CRUD)、从文本与对话历史自动生成记忆、自定义记忆捕获指令、三种检索策略(last_n/first_n/agentic)以及通过 DB 工具开关控制 Agent 记忆行为,读完后你可以直接在 Agno 中搭建一套完整、可落地的用户记忆管理方案。
目录结构与学习路径
该模块位于 cookbook/11_memory/memory_manager/,包含一个概览 README、五个独立可运行的示例脚本和一个测试日志(TEST_LOG)。五个示例按能力递进编排:
| 示例文件 | 核心主题 |
|---|---|
| 01_standalone_memory.py | 直接使用 MemoryManager 手工执行记忆的添加、查询、删除与替换 |
| 02_memory_creation.py | 从一段文本和一段对话历史(Message 列表)自动创建记忆 |
| 03_custom_memory_instructions.py | 自定义记忆捕获指令,对比默认管理器的捕获效果 |
| 04_memory_search.py | 使用 last_n、first_n、agentic 三种方式检索记忆 |
| 05_db_tools_control.py | 通过 DB 工具开关控制 Agent 可用的记忆数据库操作 |
提示:本目录的示例聚焦 PostgreSQL(以及最后一个示例中的 SQLite)上的
MemoryManager直接 API;基于 SurrealDB 的记忆管理器示例位于 cookbook/integrations/surrealdb/,配套的记忆优化策略示例则位于 cookbook/11_memory/optimize_memories/。
环境准备与运行前提
安装依赖
在运行示例前,需要准备虚拟环境并安装依赖(参考 cookbook/11_memory/README.md):
python3 -m venv ~/.venvs/aienv
source ~/.venvs/aienv/bin/activate
pip install -U psycopg sqlalchemy openai agno
其中:
agno:框架本体,提供MemoryManager、Agent、UserMemory等类型;psycopg+sqlalchemy:PostgreSQL 驱动与 ORM,PostgresDb基于 SQLAlchemy 引擎工作;openai:MemoryManager 的默认模型供应商(不指定model时会回退到OpenAIChat(id="gpt-4o"),见 manager.py 的get_model)。
数据库连接串
所有 PostgreSQL 示例使用统一连接串:
postgresql+psycopg://ai:ai@localhost:5532/ai
其含义为:ai 用户、密码 ai、主机 localhost、端口 5532、数据库 ai。从 PostgresDb 构造函数 可以看出,它优先使用 db_engine,其次使用 db_url,并且默认 create_schema=True——即首次连接时若 schema 不存在会自动建表,无需手工初始化。记忆表结构定义在 libs/agno/agno/db/postgres/schemas.py 的 MEMORY_TABLE_SCHEMA:主键 memory_id(String)、记忆内容 memory(JSONB)、主题标签 topics(JSONB)、归属字段 user_id/agent_id/team_id、时间戳 created_at/updated_at(BigInteger,Unix 秒级)以及 input/feedback。
核心类型:UserMemory
记忆的最小单位是 UserMemory 数据类,定义在 libs/agno/agno/db/schemas/memory.py,关键字段:
memory: str:记忆正文(必填),通常为第三人称陈述句,例如"The user's name is John Doe";memory_id: Optional[str]:记忆唯一 ID,未指定时由 MemoryManager 自动生成(UUID4);topics: Optional[List[str]]:主题标签列表,如["name"]、["hobbies"],便于分类与检索;user_id / agent_id / team_id:归属信息,用于多用户、多 Agent 场景下的隔离;created_at / updated_at:Unix 秒级时间戳,__post_init__会自动填充created_at,也是last_n/first_n排序的依据;input / feedback:来源输入与人工反馈。
示例一:独立 MemoryManager 的 CRUD 操作
01_standalone_memory.py 展示了不经过 Agent、直接调用 MemoryManager 的四种核心操作。
创建管理器
from agno.db.postgres import PostgresDb
from agno.memory import MemoryManager, UserMemory
db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
memory = MemoryManager(db=PostgresDb(db_url=db_url))
这里不传 model 也可以执行 CRUD——因为增删改查是纯数据库操作,不依赖 LLM;只有自动生成/搜索记忆这类需要模型能力的操作才必须提供 model。
添加记忆
memory.add_user_memory(
memory=UserMemory(memory="The user's name is John Doe", topics=["name"]),
)
不传 user_id 时,记忆会挂到默认用户 "default" 名下。源码 add_user_memory 会补齐 memory_id(若缺失)、user_id 与 updated_at,然后调用 db.upsert_user_memory 落库并返回记忆 ID。
按用户隔离操作
jane_doe_id = "jane_doe@example.com"
memory_id_1 = memory.add_user_memory(
memory=UserMemory(memory="The user's name is Jane Doe", topics=["name"]),
user_id=jane_doe_id,
)
memory_id_2 = memory.add_user_memory(
memory=UserMemory(memory="She likes to play tennis", topics=["hobbies"]),
user_id=jane_doe_id,
)
memories = memory.get_user_memories(user_id=jane_doe_id)
user_id 是记忆隔离的键:不同用户拥有完全独立的记忆空间。get_user_memories 每次都会从数据库重新读取(read_from_db 按 user_id 分组),保证拿到的是最新数据。
删除与替换
memory.delete_user_memory(user_id=jane_doe_id, memory_id=memory_id_2)
memory.replace_user_memory(
memory_id=memory_id_1,
memory=UserMemory(memory="The user's name is Jane Mary Doe", topics=["name"]),
user_id=jane_doe_id,
)
delete_user_memory按memory_id+user_id精确定位删除(源码);replace_user_memory则以新UserMemory整体覆盖指定memory_id的旧内容(源码),返回该记忆 ID。
测试印证
单元测试 test_memory_manager_crud.py 验证了这些行为:默认初始化时 add_memories=True、update_memories=True、delete_memories=False、clear_memories=False(TestMemoryManagerInit);add_user_memory 在未传 ID 时自动生成 UUID、未传 updated_at 时自动补时间戳(TestAddUserMemory);get_user_memories 缺省 user_id 时回落到 "default"(TestGetUserMemories)。
示例二:从文本与消息历史创建记忆
02_memory_creation.py 展示了由 LLM 驱动的记忆生成,需要显式指定模型:
memory = MemoryManager(model=OpenAIChat(id="gpt-5.6-luna"), db=memory_db)
从一段文本创建
memory.add_user_memory(
memory=UserMemory(memory="""I enjoy hiking in the mountains on weekends,
reading science fiction novels before bed, ...""" ),
user_id=john_doe_id,
)
注意这里的 add_user_memory 接收的是已经整理好的整段记忆文本——MemoryManager 不会加工这段内容,原样入库。想要“让模型从一段用户陈述中提炼多条结构化记忆”,应使用 create_user_memories。
从消息历史创建
memory.create_user_memories(
messages=[
Message(role="user", content="My name is Jane Doe"),
Message(role="assistant", content="That is great!"),
Message(role="user", content="I like to play chess"),
Message(role="assistant", content="That is great!"),
],
user_id=jane_doe_id,
)
create_user_memories 的内部流程(见 源码)可以拆解为:
- 参数归一:接受单个
message字符串或messages消息列表;若传message,内部包装成Message(role="user", content=message); - 读取已有记忆:从数据库读出该用户当前全部记忆,作为
<existing_memories>上下文传给模型,避免重复捕获; - 装配系统提示与工具:调用
get_system_message生成默认“记忆管理器”系统提示,并通过_get_db_tools把add_memory、update_memory、delete_memory注册为可调用的函数工具; - 模型裁决:模型阅读用户消息与已有记忆,决定“无需改动 / 新增 / 更新 / 删除”,并通过工具调用直接写库;
- 刷新缓存:执行完再调用
read_from_db刷新内存中的记忆视图。
运行前提:
create_user_memories需要模型具备函数调用能力,并且MemoryManager必须配置model与db;若未配置db,方法会直接返回提示字符串"Please provide a db to store memories"。
示例三:自定义记忆捕获指令
默认的记忆管理器会捕获“与当前对话相关的个人信息”,其默认指令(manager.py 的 get_system_message)关注个人事实(姓名、年龄、职业、所在地、兴趣、偏好)、观点与偏好、重要生活事件、当前处境与目标等。
03_custom_memory_instructions.py 通过 memory_capture_instructions 覆盖这一默认行为:
memory = MemoryManager(
model=OpenAIChat(id="gpt-5.6-luna"),
memory_capture_instructions="""\
Memories should only include details about the user's academic interests.
Only include which subjects they are interested in.
Ignore names, hobbies, and personal interests.
""",
db=memory_db,
)
这段指令被嵌入系统提示的 <memories_to_capture> 区块中(见 源码 L1006-L1011),成为模型“何时该记、记什么”的唯一准则。示例中用 John Doe 的自我介绍测试:尽管他提到姓名、徒步、科幻小说、烹饪、国际象棋,但因为指令明确“忽略姓名、爱好和个人兴趣”,模型只会捕获“对宇宙历史与天文话题感兴趣”这条学术兴趣。
偏好变更的自愈更新
示例的第二部分改用 Claude(Claude(id="claude-sonnet-4-5-20250929"))配合默认指令,并传入一段包含偏好修正的对话:
"Actually, forget that I like to play chess. I more enjoy playing table top games like dungeons and dragons"
这里体现了默认系统提示中的更新准则(源码 L1002-L1004):
- 用户要求“忘记/修改”某条信息时,应删除相关旧表述,而不是写“用户以前喜欢……”;
- 偏好变化时,既反映新偏好,也记录“曾经是什么、发生了什么变化”;
- 更新已有记忆时采用追加而非整体覆盖的策略,避免丢失上下文。
附加指令与自定义系统消息
MemoryManager 还支持另外两个提示定制入口(源码 L52-L57):
system_message:完全替换默认系统消息;additional_instructions:追加在默认系统提示末尾(源码 L1039-L1040),适合在不重写整个提示的前提下补充少量约束。
示例四:记忆检索的三种策略
04_memory_search.py 先为 John Doe 写入两条记忆(喜欢周末爬山、睡前读科幻小说),再演示 search_user_memories 的三种检索方式(源码):
memories = memory.search_user_memories(
user_id=john_doe_id, limit=1, retrieval_method="last_n"
)
memories = memory.search_user_memories(
user_id=john_doe_id, limit=1, retrieval_method="first_n"
)
memories = memory.search_user_memories(
user_id=john_doe_id,
query="What does the user like to do on weekends?",
retrieval_method="agentic",
)
| 检索方法 | 行为 | 实现位置 |
|---|---|---|
last_n(默认) |
按 updated_at 升序排序后取最后 N 条,即最近更新的记忆;updated_at 缺失的记录排在最前 |
_get_last_n_memories |
first_n |
按 updated_at 升序排序后取前 N 条,即最早创建的记忆;缺失时间戳的记录排到队尾(用 MAX_UNIX_TS = 2**63 - 1 兜底) |
_get_first_n_memories |
agentic |
必须传 query;由 LLM 基于语义判断哪些记忆与该查询相关,返回对应的 memory_ids |
_search_user_memories_agentic |
agentic 检索的原理
agentic 检索不是向量相似度匹配,而是“让模型读全部记忆,再挑出与查询相关的 ID”:
- 把该用户的全部记忆(含
memory_id、memory、topics)拼装成<user_memories>系统上下文; - 要求模型以结构化输出(
MemorySearchResponse,即memory_ids列表,定义于 manager.py L36-L42)返回相关记忆 ID; - 根据模型能力自动选择输出格式:原生结构化输出、JSON Schema 或
json_object(见 _get_response_format); - 将模型返回的 ID 映射回
UserMemory列表,并按limit截断。
limit 参数对三种方式通用:不传则返回全部;limit=0 或负数视为不限制(源码 L760-L761)。检索默认排序行为同样有测试覆盖,参见 test_search_user_memories_defaults.py。
示例五:通过 DB 工具开关控制 Agent 的记忆行为
前四个示例都在操作“裸”的 MemoryManager;05_db_tools_control.py 则演示了把 MemoryManager 接入 Agent,并用开关精细控制模型可调用的记忆工具:
from agno.agent.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.memory.manager import MemoryManager
memory_db = SqliteDb(db_file="tmp/memory_control_demo.db")
memory_manager_full = MemoryManager(
model=OpenAIChat(id="gpt-5.6-luna"),
db=memory_db,
add_memories=True,
update_memories=True,
)
agent_full = Agent(
model=OpenAIChat(id="gpt-5.6-luna"),
memory_manager=memory_manager_full,
enable_agentic_memory=True,
db=memory_db,
)
要点解析:
- 存储后端的灵活性:这里使用 SqliteDb(
db_file="tmp/memory_control_demo.db"),证明 MemoryManager 不绑定 PostgreSQL——凡实现BaseDb/AsyncBaseDb接口的存储均可接入(包括 Redis、Mongo、MySQL 等,参见 cookbook/06_storage/); - 工具开关语义:
MemoryManager的四个布尔标志(源码 L63-L70)决定模型能看到并调用哪些数据库工具——add_memories对应add_memory、update_memories对应update_memory、delete_memories对应delete_memory、clear_memories对应clear_memory(工具注册逻辑见 _get_db_tools);同时,只有开启的工具才会被写进系统提示的可选动作列表(源码 L1017-L1024); - 注意构造器默认值:
MemoryManager.__init__中delete_memories与clear_memories默认是False(源码 L84-L87),即默认不授予删除类工具,这是出于安全考虑;类属性上的同名默认值(True)仅用于Agent内部以关键字方式配置记忆行为时的兜底; - Agent 侧开关:
enable_agentic_memory=True让 Agent 在每轮对话后调用 MemoryManager 自动沉淀记忆(agent.py L130)。运行中可通过对话自然触发记忆生命周期:"My name is John Doe and I like to hike in the mountains on weekends. I also enjoy photography."→ 新增记忆;"What are my hobbies?"→ 查询时由 Agent 用记忆回答;"I no longer enjoy photography. Instead, I've taken up rock climbing."→ 模型调用update_memory工具修正既有记忆,而不是新增一条互相矛盾的内容。
跑完后打印该用户的记忆列表,可直观看到“摄影→攀岩”的偏好已被更新覆盖。
记忆管理器的完整能力矩阵
除示例展示的 API 外,从 manager.py 还能看到更多生产可用的能力,一并汇总如下:
| 能力 | 方法(同步 / 异步) | 说明 |
|---|---|---|
| 添加记忆 | add_user_memory |
直接落一条 UserMemory,返回记忆 ID |
| 获取记忆列表 | get_user_memories / aget_user_memories |
按用户读取,缺省 user_id="default" |
| 获取单条记忆 | get_user_memory |
按 memory_id 精确查找 |
| 替换记忆 | replace_user_memory |
整体覆盖指定 ID 的记忆 |
| 删除记忆 | delete_user_memory |
按 memory_id + user_id 删除 |
| 清空某用户记忆 | clear_user_memories / aclear_user_memories |
先取全部 ID 再批量删除;异步数据库需用异步版本 |
| 全库清空 | clear |
调用 db.clear_memories() |
| 自动创建记忆 | create_user_memories / acreate_user_memories |
从文本或消息列表,经 LLM 裁决后增/改/删 |
| 任务式更新 | update_memory_task / aupdate_memory_task |
以自由文本任务驱动记忆变更(如“记住我住在上海”) |
| 记忆检索 | search_user_memories |
last_n / first_n / agentic 三种策略 |
| 记忆优化 | optimize_memories / aoptimize_memories |
支持 MemoryOptimizationStrategy(默认 SUMMARIZE 摘要策略),apply=True 时用优化结果替换库中记忆 |
异步方法(a* 前缀)专门适配 AsyncBaseDb;使用同步方法操作异步数据库会抛出 ValueError 提示改用异步版本(源码 L322-L325)。上述类型与策略统一从 agno.memory 包 导出(MemoryManager、UserMemory、MemoryOptimizationStrategy 等),异步行为测试可参考 test_memory_manager_async.py。
推荐实践与注意事项
- 多用户场景务必显式传
user_id:MemoryManager 对缺省用户统一使用"default",多人共用会导致记忆互相污染;生产环境应以邮箱、会话或平台用户 ID 作为记忆隔离键。 - 删除类工具按需开放:默认构造器不开放
delete_memories/clear_memories;只有当产品确实允许模型“遗忘”时再显式开启,避免误删长期积累的用户画像。 - 生成型操作必须配模型:
create_user_memories、agentic检索、update_memory_task、optimize_memories依赖 LLM 裁决;纯 CRUD 则不需要。未指定模型时默认回退OpenAIChat(id="gpt-4o")。 - 自定义捕获指令是控制记忆质量的最直接手段:通过
memory_capture_instructions收敛“记什么”,通过additional_instructions补充约束,可以在不引入额外系统的情况下显著减少记忆噪音。 - 偏好更新遵循“追加 + 记录变化”:默认提示要求更新时保留“过去偏好与变化过程”,这让记忆在用户口味漂移时依然可追溯。
小结
从手工 CRUD 到 LLM 驱动的自动记忆生成、从三种检索策略到 Agent 内的工具开关控制,cookbook/11_memory/memory_manager/ 的五个示例完整覆盖了 Agno MemoryManager 的核心使用路径。其底层实现(libs/agno/agno/memory/manager.py)把“记忆即数据 + 模型即裁判”的思想落地为统一的系统提示、数据库工具与结构化输出机制,配合 test_memory_manager_crud.py 等测试验证,无论是直接编程调用还是嵌入 Agent 自动沉淀,都能获得稳定可控的用户长期记忆能力。若需探索记忆优化(摘要/去重)与 SurrealDB 等替代存储,可继续阅读 cookbook/11_memory/optimize_memories/ 与 cookbook/integrations/surrealdb/。
atomcodeClaude Code 的开源替代方案。连接任意大模型,编辑代码,运行命令,自动验证 — 全自动执行。用 Rust 构建,极致性能。 | An open-source alternative to Claude Code. Connect any LLM, edit code, run commands, and verify changes — autonomously. Built in Rust for speed. Get StartedRust0631
MiniCPM5-2BMiniCPM5-2B 是一款面向端侧、本地部署和资源受限场景的 2B 稠密 Transformer,能够达到同尺寸开源模型 SOTA 水平。Markdown00
video-shotcraftAI宣传片skill,使用 Remotion 制作电影级产品视频:提供106 张镜头配方卡和可复用的视频魔板。适用于 Claude Code 与 Codex以及所有其他智能体Markdown00
HivisionIDPhotos⚡️HivisionIDPhotos: a lightweight and efficient AI ID photos tools. 一个轻量级的AI证件照制作算法。Python09
DragonOSDragonOS is an operating system developed from scratch using Rust, with Linux compatibility. It is designed for **Serverless** scenarios. 使用Rust从0自研内核,具有Linux兼容性的操作系统,面向云计算Serverless场景而设计。Rust00
Spark-X2.5-1.7BSpark-X2.5-1.7B 旨在让强大的 AI 更加实用、高效且易于获取。这些模型在广泛的日常任务中表现出色,涵盖对话、写作、翻译、推理、编程、工具调用和智能体工作流,并在同等规模的开源模型中取得领先结果。Spark-X2.5 将面向效率的架构与最高 1M tokens 的原生上下文窗口相结合,并支持 200 多种语言。Python00