Agno 多存储自学习系统

Agno 多存储自学习系统

学习目标

本章将分析 Agno 的 LearningMachine 自学习系统：

• 六种学习存储的职责分工与协调
• bool / Config / Store 三层配置协议
• learning=True 一行启用全部学习的背后机制
• Curator 的记忆维护管线

前置知识

建议先阅读：

• Agent 自学习与模式识别 — RETRIEVE → JUDGE → DISTILL → CONSOLIDATE 学习管线

下文假设你已理解上述概念，直接聚焦 Agno 的具体实现。

---

项目实践

六种学习存储

LearningMachine 协调六种学习存储，每种负责不同维度的知识：

存储	捕获内容	范围	典型用例
User Profile	结构化字段（姓名、邮箱、偏好）	每用户	个性化问候、偏好路由
User Memory	非结构化观察（“用户喜欢简洁答案”）	每用户	上下文适配
Session Context	当前会话的目标、计划、进展、摘要	每会话	任务连续性
Entity Memory	外部事物的事实、事件、关系	可配置命名空间	CRM、知识图谱
Learned Knowledge	可复用的模式、最佳实践	可配置命名空间	集体智能
Decision Log	决策原因（“选了方案 A 而非 B，因为…”）	每用户/会话	决策可追溯

三层配置协议

每种存储接受三种配置输入：

# 伪代码：三层配置
agent = Agent(
    learning=LearningMachine(
        user_profile=True,                          # bool: 启用默认配置
        user_memory=UserMemoryConfig(                # Config: 自定义配置
            model=OpenAIResponses(id="gpt-4o-mini"),
            system_message="提取用户偏好...",
        ),
        entity_memory=CustomEntityStore(...),        # Store: 完全自定义实现
    ),
)

配置协议的内部处理：

# 伪代码：配置解析
def _resolve_store(config_input):
    if isinstance(config_input, bool):
        return DefaultStore() if config_input else None
    elif isinstance(config_input, LearningStore):
        return config_input  # 直接使用传入的 Store
    else:
        return DefaultStore.from_config(config_input)  # 从 Config 构建

learning=True 一行启用

最简单的用法：

# 伪代码：最简学习 Agent
agent = Agent(
    model=OpenAIResponses(id="gpt-4o"),
    db=PostgresDb(db_url="..."),
    learning=True,  # 一行启用全部 6 种学习存储
)

Agent.__init__ 内部将 learning=True 转为默认的 LearningMachine：

# 伪代码：默认转换
if learning is True:
    self.learning = LearningMachine(db=db, model=model)

所有 6 种存储使用默认配置（启用，使用主 Agent 的模型和数据库）。

Curator 记忆维护

Curator 负责后台记忆维护：

• 合并：将语义相似的记忆合并为一条
• 淘汰：删除过期或不再相关的记忆
• 去重：检测并移除重复记忆
• 关联：建立记忆之间的关联（如”用户 A 的偏好 X 与实体 Y 相关”）

Curator 在后台线程中运行，不阻塞主执行管线。

自定义存储扩展

通过 custom_stores 参数添加项目特定的学习存储：

# 伪代码：自定义存储
class BugPatternStore(LearningStore):
    """专门记录常见 bug 模式和解决方案的学习存储"""

    def recall(self, query: str) -> str:
        """检索与查询相关的 bug 模式"""
        ...

    def process(self, messages: list) -> None:
        """从对话中提取新的 bug 模式"""
        ...

    def get_tools(self) -> list:
        """暴露给 Agent 的工具"""
        return [self.search_bug_patterns, self.record_bug_pattern]

agent = Agent(
    learning=LearningMachine(
        custom_stores=[BugPatternStore(db=db)],
    ),
)

问题与规避

1. 六种存储全部启用的性能开销

默认 learning=True 启用全部 6 种存储，每种存储可能需要一次 LLM 调用来提取学习。这意味着每次 run 可能额外产生 6 次 LLM 调用。

规避：

• 后台线程模式确保这些调用与主管线并行，不增加响应延迟
• 使用更便宜的模型配置各存储（如 UserMemoryConfig(model=...)）
• 按需启用：只开启需要的存储（如 learning=LearningMachine(user_profile=True, entity_memory=True)）

2. 实体记忆的命名空间冲突

Entity Memory 和 Learned Knowledge 使用可配置的命名空间。如果命名空间选择不当，可能导致不同实体的数据混合。

规避：使用项目/领域级别的命名空间前缀（如 "my-project/users/"），避免跨项目泄漏。

3. 学习提取的质量不稳定

LearningMachine 依赖 LLM 从对话中提取学习。如果 LLM 提取了不准确或不相关的信息，会污染学习存储。

规避：

• 使用自定义 system_message 指导提取标准
• Curator 的合并和去重机制可以缓解低质量学习
• 定期人工审查学习存储

设计取舍

为什么用统一协调器而非独立组件？

Agno 选择用 LearningMachine 统一协调 6 种存储：

• 优势：统一的 recall / process / get_tools 接口，Agent 只需与一个组件交互
• 代价：协调器本身增加了复杂度
• 替代方案：6 个独立组件，Agent 直接配置每个——灵活性更高但使用更繁琐

为什么 learning=True 默认启用全部？

优势：降低使用门槛，用户无需逐个配置 代价：可能启用了不需要的存储，产生额外的 LLM 调用 替代方案：默认不启用任何存储，用户按需开启——更安全但更繁琐

参考来源

• Agno learn/machine.py 源码
• Agno learn/config.py 源码