Agno 记忆优化实战：用 Summarize 与自定义策略压缩 Agent 长期记忆

记忆是 Agent 长期会话能力的核心，但无限制累积的用户记忆会迅速膨胀上下文，推高 Token 成本并稀释关键信息。本文基于 agno 仓库 cookbook 中 optimize_memories 目录的两个完整示例，系统讲解如何通过记忆优化策略（Memory Optimization Strategy）对已存储的用户记忆进行压缩与治理：内置的 Summarize 策略可将多条记忆合并为单条综合摘要，实现最大限度的 Token 缩减；自定义策略则允许按业务规则（如仅保留最近 N 条记忆）精确控制记忆生命周期。读完本文，你将掌握 MemoryManager.optimize_memories() 的完整调用链、策略抽象基类设计，以及如何测量优化前后 Token 收益并安全落地到数据库。

一、为什么要优化记忆：长期记忆的膨胀问题

在 agno 中，Agent 可以通过 update_memory_on_run=True 与 MemoryManager 在每次运行后自动从对话中抽取用户记忆并持久化到数据库（如本目录示例使用的 SqliteDb）。UserMemory 记录包含 memory（记忆正文）、topics（话题标签）、user_id、created_at/updated_at 以及可选的 agent_id/team_id 等字段，具体定义见 UserMemory 模型。

当用户与 Agent 交互足够频繁后，积累的记忆条目会越来越多，带来两个现实问题：

• Token 成本上升：每次会话都需要把相关记忆注入上下文，记忆越多，单次推理的输入 Token 越大；
• 信息信噪比下降：冗余、重复、过时的记忆会稀释真正关键的用户事实，降低记忆检索的准确性。

记忆优化（Memory Optimization）正是为了解决这一矛盾而设计：它把已存储的用户记忆作为输入，通过特定策略将其压缩为更精简的集合，再回写数据库。本目录演示了两类策略——内置的 SummarizeStrategy（LLM 摘要压缩）与用户自定义策略（规则化裁剪），二者分别代表"语义压缩"与"结构裁剪"两种优化路线。

二、核心 API 概览：从抽象基类到管理入口

先梳理记忆优化的完整代码骨架，所有实现均位于 libs/agno/agno/memory 模块：

组件	位置	职责
MemoryOptimizationStrategy	strategies/base.py	策略抽象基类，规定 optimize()/aoptimize() 接口
SummarizeStrategy	strategies/summarize.py	内置策略，调用 LLM 将所有记忆合并为一条摘要
MemoryOptimizationStrategyType	strategies/types.py	策略枚举，当前提供 SUMMARIZE = "summarize"
MemoryOptimizationStrategyFactory	strategies/types.py	根据枚举创建策略实例的工厂
MemoryManager.optimize_memories()	manager.py	同步优化入口，支持枚举或策略实例
MemoryManager.aoptimize_memories()	manager.py	异步优化入口，兼容异步数据库

2.1 策略基类：MemoryOptimizationStrategy

MemoryOptimizationStrategy 是一个 ABC 抽象基类，约定如下（见 strategies/base.py）：

• optimize(memories, model) -> List[UserMemory]：同步优化方法，接收 List[UserMemory] 和用于优化的 Model，返回优化后的记忆列表；
• aoptimize(memories, model) -> List[UserMemory]：异步版本；
• get_system_prompt()：可选，仅 LLM 类策略需要提供系统提示词；
• count_tokens(memories) -> int：基类内置的 Token 统计工具，基于 count_text_tokens 逐条累计，用于量化优化收益（base.py#L58-L66）。

2.2 管理入口：MemoryManager.optimize_memories()

optimize_memories() 是策略执行的调度中枢，其签名与行为（manager.py#L797-L804）：

def optimize_memories(
    self,
    user_id: Optional[str] = None,          # 目标用户，默认 "default"
    strategy: Union[MemoryOptimizationStrategyType, MemoryOptimizationStrategy]
        = MemoryOptimizationStrategyType.SUMMARIZE,  # 策略枚举或自定义实例
    apply: bool = True,                     # 是否将优化结果写回数据库
) -> List[UserMemory]:

内部执行流程（对应 manager.py#L825-L866）：

1. 通过 get_user_memories(user_id=user_id) 取出该用户的全部记忆；若无记忆则直接返回空列表；
2. 若传入的是枚举值，则经由 MemoryOptimizationStrategyFactory.create_strategy() 实例化对应策略；若传入的是自定义策略实例则直接使用；
3. 使用 self.get_model()（MemoryManager 初始化时传入的模型）调用 strategy_instance.optimize(memories=..., model=...) 得到优化后的记忆；
4. 若 apply=True 且配置了数据库：先 clear_user_memories() 清空该用户旧记忆，再为每条新记忆补 memory_id（缺失时用 uuid4 生成）并 upsert_user_memory() 写入，实现原子替换式更新。

需要注意：同步 optimize_memories() 不支持异步数据库（AsyncBaseDb），会直接抛出 ValueError，此时必须改用 aoptimize_memories()（manager.py#L820-L823）。

三、策略一：内置 Summarize 策略（LLM 摘要压缩）

示例脚本 01_memory_summarize_strategy.py 演示了最常用、压缩率最高的 Summarize 策略。

3.1 准备工作：Agent 与 MemoryManager 的搭建

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.memory import MemoryManager, SummarizeStrategy
from agno.memory.strategies.types import MemoryOptimizationStrategyType
from agno.models.openai import OpenAIChat

db_file = "tmp/memory_summarize_strategy.db"
db = SqliteDb(db_file=db_file)
user_id = "user2"

agent = Agent(
    model=OpenAIChat(id="gpt-5.6-luna"),
    db=db,
    update_memory_on_run=True,   # 每次运行后自动抽取并更新用户记忆
)

memory_manager = MemoryManager(
    model=OpenAIChat(id="gpt-5.6-luna"),
    db=db,
)

要点说明：

• update_memory_on_run=True 让 Agent 在 print_response() 后自动把对话内容提炼为记忆并持久化；脚本依次输入四段关于宠物狗 Max、旧金山生活、周末徒步与美食、钢琴学习的长文本，从而生成 4 条记忆；
• MemoryManager 与 Agent 共享同一个 SqliteDb 实例，这样优化工具才能读到 Agent 写入的记忆；
• MemoryManager 需要独立的 model 参数，它将作为后续 LLM 摘要生成所用的模型。

3.2 优化前后测量与执行

脚本先用 agent.get_user_memories(user_id=user_id) 取回优化前的记忆，打印记忆条数与 SummarizeStrategy().count_tokens() 统计的 Token 数，随后执行优化：

strategy = SummarizeStrategy()
tokens_before = strategy.count_tokens(memories_before)

memory_manager.optimize_memories(
    user_id=user_id,
    strategy=MemoryOptimizationStrategyType.SUMMARIZE,  # 也可直接传 SummarizeStrategy() 实例
    apply=True,
)

优化完成后再次读取记忆并统计，脚本会打印 Token 缩减百分比：

reduction_pct = ((tokens_before - tokens_after) / tokens_before) * 100
tokens_saved = tokens_before - tokens_after
print(f"  Reduction: {reduction_pct:.1f}% ({tokens_saved} tokens saved)")

3.3 底层实现：SummarizeStrategy 如何工作

SummarizeStrategy 的实现位于 strategies/summarize.py，其 optimize() 核心逻辑如下：

1. 校验输入：记忆列表为空或首条记忆缺少 user_id 时抛出 ValueError；
2. 合并话题标签：收集所有记忆的 topics 并去重，作为摘要记忆的 topics 保留；
3. 一致性推断：若所有记忆的 agent_id（或 team_id）一致，则将该 ID 保留到摘要记忆中，否则置空；
4. 构造 LLM 调用：把所有记忆拼接为 Memory 1: ...、Memory 2: ... 的文本块，配合系统提示词调用 model.response()。系统提示词要求模型"保留所有事实信息、消除冗余、合并重复事实、以第三人称叙述、不添加原文不存在的信息"（summarize.py#L29-L42）；
5. 生成新记忆：为摘要分配新的 memory_id（uuid4），构建一条带 updated_at 的 UserMemory 返回。

同步与异步版本分别使用 model.response() 与 await model.aresponse()，异步实现见 summarize.py#L121-L196。因此 Summarize 策略的本质是"多进一出"：N 条记忆 → 1 条综合摘要，压缩率取决于原始记忆的冗余程度，在典型场景下可节省大量 Token。策略工厂 MemoryOptimizationStrategyFactory.create_strategy(MemoryOptimizationStrategyType.SUMMARIZE) 即返回该策略实例（types.py#L28-L38）。

四、策略二：自定义策略（规则化裁剪）

当你不希望依赖 LLM 摘要（例如追求确定性的输出、零延迟或低成本），可以继承 MemoryOptimizationStrategy 编写纯规则策略。示例 02_custom_memory_strategy.py 展示了这一模式。

4.1 实现 RecentOnlyStrategy：仅保留最近 N 条记忆

from datetime import datetime
from typing import List

from agno.agent import Agent
from agno.db.schemas import UserMemory
from agno.db.sqlite import SqliteDb
from agno.memory import MemoryManager, MemoryOptimizationStrategy
from agno.models.base import Model
from agno.models.openai import OpenAIChat

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]:
        """Keep only the most recent N memories."""
        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]:
        """Async version of optimize."""
        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]

这段自定义策略的要点：

• 必须同时实现 optimize() 与 aoptimize()（基类均标为 @abstractmethod），否则实例化会失败；
• model 参数在纯规则策略中可以忽略，但其签名必须保留以满足接口约定；
• 排序键 m.updated_at or m.created_at or datetime.min 充分利用 UserMemory 的时间戳字段：优先按更新时间、其次创建时间排序，updated_at 会在 __post_init__ 中自动规范化（见 memory.py#L24-L28）；
• 返回 sorted_memories[:keep_count]，即确定性裁剪——结果完全可预期，不消耗任何模型调用。

4.2 应用自定义策略

custom_strategy = RecentOnlyStrategy(keep_count=2)
tokens_before = custom_strategy.count_tokens(memories_before)

memory_manager.optimize_memories(
    user_id=user_id,
    strategy=custom_strategy,   # 直接传入策略实例
    apply=True,
)

由于 optimize_memories() 的 strategy 参数同时接受枚举与策略实例（Union[MemoryOptimizationStrategyType, MemoryOptimizationStrategy]），自定义实例可以直接透传，无需修改任何框架代码。脚本会打印"优化前 4 条记忆 → 优化后 2 条记忆"以及相应的 Token 缩减量。

五、同步与异步：optimize_memories vs aoptimize_memories

MemoryManager 同时提供同步与异步两个优化入口，二者行为对齐（均支持 user_id、strategy、apply 三参数），差异在于：

维度	optimize_memories()	aoptimize_memories()
数据库兼容	仅支持同步数据库（BaseDb），异步库直接抛错	同步/异步数据库均兼容，异步库走 aget_user_memories() 与 aclear_user_memories()
策略调用	strategy_instance.optimize(...)	await strategy_instance.aoptimize(...)
回写数据库	clear_user_memories() + upsert_user_memory()	aclear_user_memories() + 对应异步 upsert

从 manager.py#L868-L939 可以看到，异步版本内部会先判断 self.db 是否为 AsyncBaseDb，进而选择同步或异步的记忆读取/清理路径，因此在使用 PostgreSQL、MongoDB 等异步数据库驱动时务必选择 aoptimize_memories()。

六、apply 语义与落地注意点

apply 参数决定优化结果的去向，这是两个示例脚本最容易忽略却最关键的行为差异：

• apply=True（示例默认）：优化结果会替换数据库中的既有记忆——先 clear_user_memories(user_id) 清空该用户全部旧记忆，再逐条 upsert 优化后的记忆（manager.py#L842-L861）。这是一次不可逆的写操作，建议在正式调用前先用 apply=False 预演并人工检查摘要质量；
• apply=False：仅返回优化后的记忆列表，不触碰数据库，适合"先看结果再决定"的预览模式。

另外两个实践建议：

1. 频率控制：记忆优化应在会话间隔、定时任务或达到阈值条数时触发，而非每次运行都执行，避免反复调用 LLM 摘要带来的额外成本；
2. 策略选型：追求最大压缩与语义连贯选 SummarizeStrategy；追求确定性、零模型成本选自定义裁剪策略；也可以组合使用——例如先用规则策略剔除过期记忆，再对剩余记忆做 LLM 摘要。

七、从示例到生产：完整运行与验证

两个示例都遵循"造数据 → 测量 → 优化 → 再测量"的可复现闭环，建议按以下步骤运行验证：

# 在仓库根目录执行（示例内部使用 SQLite，无需额外服务）
cd cookbook/11_memory/optimize_memories
python 01_memory_summarize_strategy.py
python 02_custom_memory_strategy.py

预期输出形态：Before optimization 显示记忆条数与 Token 数；执行优化后，Summarize 示例打印"Memory count: 1"及合并后的单条摘要，Custom 示例打印"Memory count: 2"及保留的两条最近记忆，同时给出 Reduction: xx.x% (xx tokens saved) 的量化收益。数据库文件生成于示例内指定的 tmp/ 目录。

如果希望进一步观察优化过程，可将 MemoryManager(debug_mode=True) 打开调试日志，此时会输出"Summarized N memories into 1"等底层记录（见 summarize.py#L115-L117）。

八、小结

记忆优化是 agno 记忆体系"写入—存储—检索—治理"闭环中的关键一环。本目录通过两个可直接运行的示例，完整覆盖了 MemoryManager + MemoryOptimizationStrategy 两种扩展路径：

• 内置 Summarize 策略：基于 SummarizeStrategy 将 N 条记忆压缩为 1 条综合摘要，最大化 Token 缩减，适合对语义保真度要求高的场景；
• 自定义策略：通过继承 MemoryOptimizationStrategy 并实现 optimize/aoptimize，实现完全确定性的记忆裁剪，适合对可控性与成本敏感的场景；
• 二者的统一调度均经由 MemoryManager.optimize_memories() / aoptimize_memories()，通过 apply 参数控制是否回写数据库。

掌握了策略基类接口、工厂注册机制与 apply 的落地语义后，你可以轻松为项目扩展第三、第四种策略（如按话题分组去重、按时间窗口保留、结合 embedding 相似度合并等），让 Agent 的长期记忆始终保持在"信息完整、体积精简"的理想状态。