Agno 多用户记忆持久化实战:从 update_memory_on_run 到 Agentic Memory 与记忆优化
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
本篇技术指南围绕 Agno 官方 Cookbook 中cookbook/11_memory/目录展开,系统讲解 Agno Agent 如何跨运行(run)、跨会话(session)、跨 Agent 持久化并使用用户记忆(user memory)。读完本文,你将掌握四种核心记忆接入方式(自动记忆、Agentic 记忆、MemoryTools、MemoryManager 直接 API)、多用户多会话隔离下的记忆共享模式,以及 Summarize / 自定义策略对记忆体量的优化手段,并了解其背后的源码级实现(MemoryManager 实现、UserMemory 数据模型)。
一、Agno 记忆体系概览:记忆存在哪、由谁管理
Agno 的用户记忆(User Memory)是结构化、跨会话持久化的:每当 Agent 完成一次运行,模型可以从对话中提炼出关于用户的持久事实(如姓名、爱好、居住地、职业),以UserMemory记录的形式写入数据库,并在后续运行中自动加载回上下文,让 Agent 表现出"记得你"的能力。
从源码看,记忆的数据模型定义在 libs/agno/agno/db/schemas/memory.py:
memory:记忆正文(如 "The user's name is John Doe");memory_id:记忆唯一 ID(用于删除、替换);topics:话题标签列表(如["name"]、["hobbies"]),便于检索归类;user_id:记忆归属用户,是记忆共享/隔离的核心键;created_at/updated_at:创建与更新时间(epoch 秒,__post_init__自动归一化);agent_id/team_id/input/feedback等扩展字段。
记忆的"写手"是MemoryManager(libs/agno/agno/memory/manager.py),它内部封装了对数据库的增删改查(add / get / delete / replace / clear),并可通过一个 LLM 模型来自动生成记忆内容。用户记忆与"会话历史"(chat history)是两个独立概念:历史记录属于某一具体 session,而记忆是跨 session 归属于 user 的长期事实。
二、环境搭建:数据库 + 依赖 + 运行
cookbook/11_memory/下的示例绝大多数依赖 PostgreSQL(部分示例也演示了 SQLite),全部命令以仓库根目录GitHub_Trending/ag/agno为基准。
1. 创建虚拟环境
python3 -m venv ~/.venvs/aienv source ~/.venvs/aienv/bin/activate2. 安装依赖
pip install -U psycopg sqlalchemy openai agnopsycopg与sqlalchemy:PostgreSQL 驱动与 ORM,供agno.db.postgres.PostgresDb使用;openai:默认模型提供方(OpenAIChat),也是MemoryManager.get_model()的默认回退模型(见 manager.py);agno:Agno 核心库。
3. 准备 PostgreSQL
示例默认连接串为:
db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" db = PostgresDb(db_url=db_url)即本机5532端口、用户/密码均为ai、数据库名为ai。仓库 scripts 目录下提供了run_postgres.sh/run_postgres.bat等一键启动脚本(如 run_pgvector.sh、run_postgres.sh),可按需选择;不想启动数据库的读者,可以直接改用示例中的 SQLite 写法:
db = SqliteDb(db_file="tmp/agent_sessions.db")4. 运行一个示例
python cookbook/11_memory/01_agent_with_memory.py注意:模型 ID(如
gpt-5.6-luna、gpt-5-mini)以当前仓库示例代码为准,运行时请替换为你实际可用的 OpenAI 模型并配置好OPENAI_API_KEY。
三、目录结构:三大模块、八个核心示例
cookbook/11_memory/的目录布局如下(见 README.md):
| 位置 | 内容 |
|---|---|
01_agent_with_memory.py~08_memory_tools.py | 核心记忆模式:自动记忆、Agentic 记忆、多 Agent 共享、自定义管理器、多用户多会话、记忆工具等 |
memory_manager/ | 直接使用MemoryManagerAPI 的示例(基于 PostgreSQL):独立 CRUD、记忆创建、自定义指令、记忆检索、DB 工具控制 |
optimize_memories/ | 记忆优化策略示例:Summarize 汇总策略与自定义策略 |
integrations/ | 第三方记忆后端集成(mem0、zep、memori、dakera 等) |
说明:基于 SurrealDB 的 MemoryManager 示例已迁移至 cookbook/integrations/surrealdb/,这也是 README 中特别标注的注意事项。
四、模式一:自动持久化记忆(update_memory_on_run)
核心示例 01_agent_with_memory.py 演示了最简接入方式:只要在 Agent 上开启update_memory_on_run=True,每次运行结束后,Agent 便会自动从对话中创建或更新用户记忆。
from agno.agent.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIChat db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" db = PostgresDb(db_url=db_url) agent = Agent( model=OpenAIChat(id="gpt-5.6-luna"), db=db, update_memory_on_run=True, )运行流程(示例代码核心逻辑):
if __name__ == "__main__": db.clear_memories() # 清理历史记忆,保证可复现 session_id = str(uuid4()) john_doe_id = "john_doe@example.com" agent.aprint_response( "My name is John Doe and I like to hike in the mountains on weekends.", stream=True, user_id=john_doe_id, session_id=session_id, ) agent.print_response("What are my hobbies?", stream=True, user_id=john_doe_id, session_id=session_id) memories = agent.get_user_memories(user_id=john_doe_id) pprint(memories) # 已包含 "hiking" 相关记忆关键点:
user_id决定记忆归属:同一user_id下多次运行共享记忆,因此第二次提问 "What are my hobbies?" 时,Agent 能基于记忆作答,而不是依赖本次会话历史;- 记忆会自动更新:当用户说 "Ok i dont like hiking anymore, i like to play soccer instead." 后,再次
get_user_memories()会发现原有爱好记忆被更新为踢足球——即"创建或更新"由MemoryManager在幕后完成; session_id与记忆解耦:即使换一个新session_id,同一user_id的记忆依然生效。
在 Agent 的源码中,对应的开关参数定义于 libs/agno/agno/agent/agent.py:memory_manager、enable_agentic_memory(默认 False)、update_memory_on_run(默认 False),三者共同决定 Agent 采用哪种记忆写入方式。
五、模式二:Agentic Memory(enable_agentic_memory)
02_agentic_memory.py 演示了"Agent 自主管理记忆"的模式:开启enable_agentic_memory=True后,Agent 在运行过程中可以主动创建、更新、删除自己的记忆,无需固定地在每次运行后批量提炼。
agent = Agent( model=OpenAIChat(id="gpt-5.6-luna"), db=db, enable_agentic_memory=True, )示例运行序列及其效果:
"My name is John Doe and I like to hike in the mountains on weekends."→ 生成记忆;"What are my hobbies?"→ 基于记忆回答;"Remove all existing memories of me."→ 再次get_user_memories()得到空列表,说明删除指令被执行;"My name is John Doe and I like to paint."→ 重新生成记忆;"I don't paint anymore, i draw instead."→ 记忆从 paint 更新为 draw。
与模式一的区别:update_memory_on_run是"运行后统一提炼",enable_agentic_memory是"运行中按需管理(增/改/删)"。从 agent.py 的构造逻辑可以推断,两种开关可以独立或组合使用,业务上"允许用户删除自己的记忆"这类需求尤其适合 Agentic 模式。
六、模式三:多 Agent 共享同一份用户记忆
1. 通过同一数据库 + 同一 user_id 共享
03_agents_share_memory.py 创建了两个行为不同的 Agent(chat_agent负责闲聊、research_agent带WebSearchTools负责调研),二者共用同一个db,并都开启update_memory_on_run=True:
chat_agent = Agent(model=OpenAIChat(id="gpt-5.6-luna"), description="You are a helpful assistant that can chat with users", db=db, update_memory_on_run=True) research_agent = Agent(model=OpenAIChat(id="gpt-5.6-luna"), description="You are a research assistant...", tools=[WebSearchTools()], db=db, update_memory_on_run=True)随后chat_agent记录了用户喜欢爬山,research_agent在回答量子计算问题时写入新的记忆;最终用research_agent.get_user_memories(user_id=john_doe_id)能看到两个 Agent 写入的全部记忆。结论:只要db相同、user_id相同,记忆天然共享——这正是多智能体系统中"公共用户画像"的基础形态。
2. 同时共享记忆与会话历史
07_share_memory_and_history_between_agents.py 更进一步:两个风格截然相反的 Agent(一个友好、一个毒舌)不仅共享记忆,还通过相同的session_id共享完整对话历史。
db = SqliteDb(db_file="tmp/agent_sessions.db") agent_1 = Agent(model=OpenAIChat(id="gpt-5.6-luna"), instructions="You are really friendly and helpful.", db=db, add_history_to_context=True, update_memory_on_run=True) agent_2 = Agent(model=OpenAIChat(id="gpt-5.6-luna"), instructions="You are really grumpy and mean.", db=db, add_history_to_context=True, update_memory_on_run=True)运行序列:
session_id = str(uuid4()) user_id = "john_doe@example.com" agent_1.print_response("Hi! My name is John Doe.", session_id=session_id, user_id=user_id) agent_2.print_response("What is my name?", session_id=session_id, user_id=user_id) agent_2.print_response("I like to hike in the mountains on weekends.", session_id=session_id, user_id=user_id) agent_1.print_response("What are my hobbies?", session_id=session_id, user_id=user_id) agent_1.print_response("What have we been discussing? Give me bullet points.", session_id=session_id, user_id=user_id)这里用到了三个关键参数:
db=db:共享持久化层;add_history_to_context=True:把存储的会话历史注入上下文,使第二个 Agent 能"接续"第一个 Agent 的对话(回答 "What is my name?" 时知道用户叫 John Doe);update_memory_on_run=True:记忆也随对话不断累积,两个 Agent 看到的用户画像一致。
注意该示例使用 SQLite(SqliteDb(db_file="tmp/agent_sessions.db")),无需启动 PostgreSQL,是快速验证记忆共享的最佳起点。
七、模式四:自定义 MemoryManager
默认的MemoryManager会随 Agent 自动创建;当你需要精细控制记忆生成的指令、模型或读写权限时,可以像 04_custom_memory_manager.py 那样独立构造并注入 Agent:
from agno.memory import MemoryManager memory_manager = MemoryManager( model=OpenAIChat(id="gpt-5.6-luna"), additional_instructions=""" IMPORTANT: Don't store any memories about the user's name. Just say "The User" instead of referencing the user's name. """, db=db, ) agent = Agent( model=OpenAIChat(id="gpt-5.6-luna"), db=db, memory_manager=memory_manager, update_memory_on_run=True, user_id="john_doe@example.com", )运行后,即使用户自报姓名 "John Doe" 且爱好游泳与足球,get_user_memories()返回的记忆中也不会出现姓名,而是以 "The User" 代指——自定义指令生效。additional_instructions会被追加到MemoryManager的默认系统提示之后(源码见 manager.py),用于约束"记什么、不记什么、怎么措辞"。
从MemoryManager.__init__的签名(manager.py)可以梳理出完整可调参数:
| 参数 | 默认值 | 说明 |
|---|---|---|
model | None(回退 OpenAI gpt-4o) | 负责生成/管理记忆的 LLM,可与 Agent 主模型不同 |
system_message | None | 覆盖默认系统提示 |
memory_capture_instructions | None | 覆盖默认记忆捕获指令 |
additional_instructions | None | 追加到默认系统提示的补充指令 |
db | None | 记忆存储数据库(BaseDb/AsyncBaseDb) |
delete_memories | False | 是否允许删除记忆 |
update_memories | True | 是否允许更新记忆 |
add_memories | True | 是否允许新增记忆 |
clear_memories | False | 是否允许清空记忆 |
debug_mode | False | 调试日志开关 |
四个布尔开关(delete / update / add / clear)组成了 MemoryManager 对外的"DB 工具权限面",直接决定模型在记忆写操作上能调用哪些能力。
八、多用户多会话实战:隔离与并发
1. 串行多用户多会话
05_multi_user_multi_session_chat.py 展示了最贴近生产的多租户场景:一个 Agent 实例服务多个用户,每个用户拥有多个会话。
user_1_session_1_id = "user_1_session_1" user_1_session_2_id = "user_1_session_2" user_2_session_1_id = "user_2_session_1" user_3_session_1_id = "user_3_session_1" chat_agent = Agent(model=OpenAIChat(id="gpt-5.6-luna"), db=db, update_memory_on_run=True)核心规律:
- 同一 user 的不同 session 共享记忆:user_1 在
user_1_session_1中介绍自己"喜欢动漫和游戏",随后在另一个会话user_1_session_2说 "I'm going to the movies tonight.";当回到user_1_session_1询问 "What do you suggest I do this weekend?" 时,Agent 可以综合两个会话积累的记忆给出个性化建议; - 不同 user 的记忆严格隔离:user_2(John Doe,计划徒步)、user_3(Jane Smith,去健身房)各自的记忆互不可见,验证时分别
get_user_memories(user_id=...)并断言非空。
2. 并发多用户多会话
06_multi_user_multi_session_chat_concurrent.py 将上述流程异步化:每个用户一个协程,通过asyncio.gather并发执行arun,同时仍保证"记忆按用户、按会话正确落库"。
async def run_concurrent_chat_agent() -> None: await asyncio.gather( user_1_conversation(), user_2_conversation(), user_3_conversation() )每个会话函数内部使用await chat_agent.arun(...)(异步运行接口,而非同步的print_response)。这一模式验证了 Agno 的AsyncBaseDb记忆写入链路在并发下的隔离正确性,是搭建高并发聊天服务(如 WebSocket 多房间)时的直接参考模板。
九、把记忆交还给模型:MemoryTools
08_memory_tools.py 展示了一种更灵活的接入方式:把记忆能力封装成**工具(Tool)**交给模型自主调用,与WebSearchTools组合实现"边规划旅行边记用户偏好":
from agno.tools.memory import MemoryTools from agno.tools.websearch import WebSearchTools memory_tools = MemoryTools(db=db) agent = Agent( model=OpenAIChat(id="gpt-5-mini"), tools=[memory_tools, WebSearchTools()], instructions=[ "You are a trip planner bot and you are helping the user plan their trip.", "You should use the WebSearchTools to get information about the destination and activities.", "You should use the MemoryTools to store information about the user for future reference.", "Don't ask the user for more information, make up what you don't know.", ], markdown=True, )MemoryTools的实现位于 libs/agno/agno/tools/memory.py,为模型暴露了get_memories、add_memory、update_memory、delete_memory四个工具函数。运行时,模型在回答"给我规划非洲旅行"前,可以先用get_memories读取用户画像,再结合搜索结果生成个性化 itinerary;当用户在对话中透露新偏好时,模型自主调用add_memory/update_memory落库。相比update_memory_on_run的"每次运行统一提炼",这是由模型按需读写的声明式用法,与 Agentic Memory 理念一脉相承。
十、直接使用 MemoryManager API(memory_manager/)
不想让 LLM 参与记忆生成、只想用代码手动管理记忆时,可以绕过 Agent 直接操作MemoryManager。该子目录包含五个示例(见 memory_manager/README.md):
- 01_standalone_memory.py:独立 CRUD;
- 02_memory_creation.py:记忆创建;
- 03_custom_memory_instructions.py:自定义记忆指令;
- 04_memory_search.py:记忆检索;
- 05_db_tools_control.py:DB 工具开关控制。
以独立 CRUD 为例,关键 API 一目了然:
from agno.db.postgres import PostgresDb from agno.memory import MemoryManager, UserMemory memory = MemoryManager(db=PostgresDb(db_url=db_url)) # 新增(返回 memory_id) 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) # 删除 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, )这里直接实例化UserMemory(memory=..., topics=[...])并调用add_user_memory/get_user_memories/delete_user_memory/replace_user_memory,适合批量导入、数据清洗、后端管理脚本等不需要 LLM 参与的确定性场景。MemoryManager还提供基于语义相似度的记忆检索(MemorySearchResponse,见 manager.py),对应04_memory_search.py示例。若使用 SurrealDB 作为存储后端,示例位于 cookbook/integrations/surrealdb/。
十一、记忆优化策略(optimize_memories/)
随着对话积累,用户记忆会越来越多,占用上下文 token 也水涨船高。optimize_memories/子目录专门演示记忆体量优化(见 optimize_memories/README.md)。
1. Summarize 汇总策略
01_memory_summarize_strategy.py 先让 Agent 在多次运行中写入 4 段长记忆(宠物狗 Max、旧金山生活、周末爱好、学钢琴),再统计优化前后对比:
strategy = SummarizeStrategy() tokens_before = strategy.count_tokens(memories_before) memory_manager.optimize_memories( user_id=user_id, strategy=MemoryOptimizationStrategyType.SUMMARIZE, apply=True, ) memories_after = agent.get_user_memories(user_id=user_id) tokens_after = strategy.count_tokens(memories_after)脚本会打印"优化前记忆条数 / token 数 → 优化后记忆条数 / token 数 / 压缩百分比"。其原理是:把所有记忆合并成一条综述记忆,用 LLM 重写去重去冗,实现 token 削减。策略类型枚举定义于 libs/agno/agno/memory/strategies/types.py,当前内置SUMMARIZE = "summarize",工厂类MemoryOptimizationStrategyFactory负责按类型实例化策略。
2. 自定义策略
02_custom_memory_strategy.py 演示如何通过继承MemoryOptimizationStrategy实现自己的策略——示例RecentOnlyStrategy只保留最近 N 条记忆:
class RecentOnlyStrategy(MemoryOptimizationStrategy): """Keep only the N most recent memories.""" def __init__(self, keep_count: int = 2): self.keep_count = keep_count def optimize(self, memories: List[UserMemory], model: Model) -> List[UserMemory]: sorted_memories = sorted( memories, key=lambda m: m.updated_at or m.created_at or datetime.min, reverse=True, ) return sorted_memories[: self.keep_count] async def aoptimize(self, memories: List[UserMemory], model: Model) -> List[UserMemory]: # 异步版本,逻辑与 optimize 一致 ...策略接口需要同时实现同步optimize与异步aoptimize两个方法,返回裁剪后的记忆列表。自定义策略通过MemoryOptimizationStrategyType枚举之外的工厂通道接入(当前枚举仅内置 summarize,自定义策略可直接实例化后使用),适用于"按时间窗保留""按话题过滤""按记忆置信度裁剪"等业务定制场景。
十二、源码级总结:记忆写入的完整调用链
综合 agent.py、manager.py 与示例代码,可以梳理出 Agno 记忆系统的整体脉络:
- 触发:
Agent上的update_memory_on_run/enable_agentic_memory/memory_manager/tools=[MemoryTools()]决定记忆如何被触发(自动、Agentic、工具调用); - 生成:
MemoryManager调用其model(默认回退 OpenAI gpt-4o,见 manager.py)从对话上下文提炼记忆,并受additional_instructions、memory_capture_instructions约束; - 落库:记忆以
UserMemory(db/schemas/memory.py)形式,通过db(BaseDb/AsyncBaseDb抽象,PostgresDb / SqliteDb 为具体实现)持久化,权限由delete_memories/update_memories/add_memories/clear_memories控制; - 召回:后续运行按
user_id加载记忆注入上下文;多 Agent 场景下共享同一db与user_id即可共享画像,叠加相同session_id还能共享会话历史; - 优化:当记忆膨胀时,通过
MemoryOptimizationStrategy(内置 Summarize 或自定义策略)在保证信息量的前提下压缩 token 开销。
这一链路覆盖了从"单 Agent 记住用户"到"多用户、多会话、多 Agent、可优化"的完整记忆工程能力,是构建个性化 Agent 应用时可复用的标准范式。更进一步,Agno 的 OS 层还提供MemoryManagerMetadata(libs/agno/agno/os/schema.py)等元数据结构,为记忆的托管化、可观测化提供扩展空间,感兴趣的读者可沿此方向继续深入。
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考