深入理解 agno 自定义学习存储Custom Learning Store从内存实现到数据库持久化的完整实战指南【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno导读在 agno 的 Agent 2.0 学习体系LearningMachine中内置的六类学习存储用户画像、用户记忆、会话上下文、实体记忆、习得知识、决策日志覆盖了大多数场景但真实业务总有仓库之外的知识形态——例如某个项目的专属上下文、某客户的私有笔记。此时就需要自定义学习存储Custom Learning Store按LearningStore协议实现自己的 recall / process / build_context 等方法再通过LearningMachine(custom_stores{...})挂接进 Agent即可让 Agent 围绕你自己的数据形态自动记忆、自动回忆、自动注入上下文。本文以 cookbook/08_learning/08_custom_stores 下的两个示例为主线结合 学习存储协议 与 LearningMachine 实现 的源码带你从零实现一个内存版自定义存储、一个基于 PostgreSQL 持久化的可调用工具版存储并理解协议背后的调度机制。一、先理解自定义存储在整个学习体系中的位置agno 的 LearningMachine 是一个统一学习系统它协调多个学习存储learning store每个存储负责一类知识的记忆、回忆与上下文注入。完整的存储全景见 cookbook/08_learning/README.md 中的表格存储捕获内容范围适用场景User Profile结构化字段姓名、偏好按用户个性化User Memory关于用户的非结构化观察按用户上下文、偏好Session Context目标、计划、进度、总结按会话任务连续性Entity Memory事实、事件、关系可配置CRM、知识图谱Learned Knowledge洞察、模式、最佳实践可配置集体智能Decision Log含推理与备选方案的决策按 Agent审计、反馈闭环当你需要的知识不属于以上任何一类时custom_stores参数就是官方预留的扩展位。它接收一个字典key 是存储的名字value 是实现了LearningStore协议的实例from agno.learn import LearningMachine agent Agent( modelOpenAIResponses(idgpt-5.5), learningLearningMachine( custom_stores{ project: project_store, # 自定义存储按名字注册 }, ), markdownTrue, )从源码看注册的逻辑在 machine.py 的__init__中内置存储与custom_stores会被合并进同一个stores字典self.stores因此自定义存储在后续的 recall、process、build_context 调度中和内置存储地位完全对等。二、LearningStore 协议需要实现哪些方法协议定义位于 libs/agno/agno/learn/stores/protocol.py。一个学习存储的核心职责有四条对应四组方法协议成员类型职责learning_typeproperty - str该学习类型的唯一标识如project_context用于存储键与日志schemaproperty - Any该学习类型使用的数据类简单存储可直接返回dictrecall(**kwargs)/arecall(**kwargs)方法根据user_id、session_id等上下文取回已保存的数据无数据时返回Noneprocess(messages, **kwargs)/aprocess(...)方法对话结束后被调用从消息中抽取并保存学习内容build_context(data) - str方法把 recall 到的数据格式化为可注入系统提示词的字符串仅数据不含使用指引instructions() - str方法Agent 可见的使用指引何时、如何调用该存储的工具无工具时返回空串get_tools(**kwargs)/aget_tools(...)方法暴露给 Agent 的可调用工具列表无工具返回[]was_updatedproperty - bool最近一次操作中该存储是否发生了更新值得注意的是虽然协议声明了instructions()但源码对第三方存储是鸭子类型兼容的。在 machine.py 中存储实例只要通过isinstance(value, LearningStore)校验或同时具备可调用的recall、process、build_context即可被接受——协议注释明确写到instructions()在 2.8.4 加入协议此前的第三方存储没有该方法也能正常工作只是不会贡献引导文本。因此下面两个示例都没有实现instructions()仍然成立。另一个关键的工程细节是签名感知的参数分发。machine.py 中的_filter_store_kwargs使用inspect.signature检查被调用方法实际接受的参数再裁剪上下文 kwargs 后分发第三方 custom store 只是满足一个 Protocol如果某个存储写成def recall(self, user_idNone)没有**kwargs一旦框架新增上下文参数message、run_context……就会抛 TypeError。内置存储带**kwargs能看到全部参数窄签名的自定义存储则原样继续工作。这意味着你的自定义方法不需要列出框架可能传入的所有参数按需声明或统一用**kwargs接收即可LearningMachine 会替你安全地分发调用点在 machine.py 的recall()/process()/abuild_context()中均以store.recall(**_filter_store_kwargs(store.recall, context))形式出现。三、示例一最小内存型自定义存储完整代码见 01_minimal_custom_store.py。它的目标是演示协议实现的最小骨架用一个进程级字典当存储围绕project_id保存某项目的工作上下文。3.1 定义存储类dataclass class ProjectContextStore(LearningStore): Custom store for project-specific context. # 构造时传入的自定义上下文示例的模式选择并非协议要求 context: Dict[str, Any] field(default_factorydict) # 内部状态 _updated: bool field(defaultFalse, initFalse) property def learning_type(self) - str: return project_context property def schema(self) - Any: return dict # 简单存储可直接用 dict 作为 schema示例注释特别强调了一个易混淆点context字段承载project_id是示例自己的模式选择不是协议要求。协议层并不规定项目维度如何传递你可以像内置存储那样用类型化配置类、也可以直接用字段保存具体参数。这里选择构造时注入context字典是为了展示如何把外部参数如project_id、team带进存储。3.2 实现回忆recall / arecallrecall按构造时给定的project_id从全局内存字典取数据arecall直接委托同步版本。def recall(self, **kwargs) - Optional[Dict[str, Any]]: project_id self.context.get(project_id) if not project_id: return None return _project_data.get(project_id) async def arecall(self, **kwargs) - Optional[Dict[str, Any]]: return self.recall(**kwargs)回忆逻辑的通用范式是先确定命名空间/维度这里是 project_id再从存储介质取出对应当前上下文的记录。在第二个数据库示例中会看到同样的范式只是把取数换成db.get_learning(learning_type..., namespaceproject_id)。3.3 实现学习process / aprocessprocess在每次对话后被自动调用ALWAYS 型提取流程接收整段会话消息messages。示例使用最简单的关键词启发式抽取——命中 goal/objective 记为话题 goals命中 blocker/stuck 记为 blockersdef process(self, messages: List[Any], **kwargs) - None: project_id self.context.get(project_id) if not project_id or not messages: return current _project_data.get(project_id, {}) for msg in messages: content getattr(msg, content, str(msg)) if isinstance(content, str): if goal in content.lower() or objective in content.lower(): current[last_discussed_topic] goals self._updated True elif blocker in content.lower() or stuck in content.lower(): current[last_discussed_topic] blockers self._updated True if current: _project_data[project_id] current代码注释也明确提示了生产环境的关键差别关键词匹配只是 demo在生产中应当用一个模型做智能抽取。这也是自动学习 vs 工具学习两种路线的分水岭——如果你想做真正的语义抽取可参考后面数据库示例里的工具路线或给 process 内部接一个 LLM 调用。3.4 构建上下文与工具build_context决定Agent 能在系统提示词里看到什么。这里把 recall 到的数据渲染成 XML 风格的project_context块且保证还没有任何上下文时也输出一个占位块避免提示词里出现空洞def build_context(self, data: Any) - str: if not data: project_id self.context.get(project_id, unknown) return fproject_context\nProject: {project_id}\nNo context saved yet.\n/project_context project_id self.context.get(project_id, unknown) lines [project_context, fProject: {project_id}] for key, value in data.items(): lines.append(f{key}: {value}) lines.append(/project_context) return \n.join(lines) def get_tools(self, **kwargs) - List[Callable]: return [] # 本示例不需要工具 property def was_updated(self) - bool: return self._updated从 machine.py 的_format_results/build_context实现可以看出LearningMachine 会把每个存储build_context的产物用空行拼接、统一注入——所以你的输出块格式是否成对闭合标签、是否自包含直接决定了 Agent 读到的上下文质量。协议注释强调build_context只放数据怎么用的指引放在instructions()本示例没有工具因此也无指引可给。3.5 挂接并运行project_store ProjectContextStore( context{project_id: learning-machine, team: platform}, ) agent Agent( modelOpenAIResponses(idgpt-5.5), learningLearningMachine( custom_stores{project: project_store}, ), markdownTrue, )运行 demo 时示例还会演示两个自定义的便捷方法它们不属于协议set_context(key, value)手动写入某条项目上下文print()打印当前项目上下文。完整流程为手动set_context(current_sprint, Sprint 23)、set_context(tech_stack, Python, PostgreSQL)提问 What project am I working on?——该项目上下文此时已通过 system prompt 注入Agent 可以直接回答再说 Im stuck on the database migration. Its a blocker for the release.——process命中 blocker/stuck 关键词把last_discussed_topic更新为blockers。if __name__ __main__: user_id developerexample.com project_store.set_context(current_sprint, Sprint 23) project_store.set_context(tech_stack, Python, PostgreSQL) agent.print_response(What project am I working on?, user_iduser_id, streamTrue) project_store.print() agent.print_response( Im stuck on the database migration. Its a blocker for the release., user_iduser_id, streamTrue, ) project_store.print()注意该示例刻意不依赖数据库数据只存在于进程内的_project_data字典进程退出即丢失。若要跨会话、跨进程、跨重启地记忆请进入第二个示例。四、示例二数据库持久化 工具驱动的自定义存储完整代码见 02_custom_store_with_db.py。它的核心差异是三点数据落到 PostgreSQL、用类型化 dataclass 做 schema、让 Agent 通过工具主动记录项目笔记。4.1 类型化 Schema不再返回dict而是定义结构化笔记dataclass class ProjectNotes: summary: Optional[str] None goals: Optional[List[str]] None blockers: Optional[List[str]] None decisions: Optional[List[str]] Noneschema属性返回ProjectNotes。类型化 schema 的好处是recall 回来的是一个强类型对象build_context可以按字段精细排版有 goals 才输出 Goals 列表、有 blockers 才输出 Blockers 列表工具写入时也能按结构校验。4.2 借助数据库的 learning 方法做持久化存储类通过db字段接收一个PostgresDbrecall 时调用数据库的get_learning方法保存时调用upsert_learning方法并且用project_id充当 namespace实现数据隔离def recall(self, **kwargs) - Optional[ProjectNotes]: if not self.db: return None project_id self.context.get(project_id) if not project_id: return None try: result self.db.get_learning( learning_typeself.learning_type, # project_notes namespaceproject_id, # 以 project_id 作为 namespace ) if result and result.get(content): content result[content] return ProjectNotes( summarycontent.get(summary), goalscontent.get(goals), blockerscontent.get(blockers), decisionscontent.get(decisions), ) return None except Exception as e: print(fError retrieving project notes: {e}) return None对应的写路径在私有方法_save中upsert_learning的 id 也按项目维度构造为fproject_notes:{project_id}self.db.upsert_learning( idfproject_notes:{project_id}, learning_typeself.learning_type, namespaceproject_id, content{...}, # summary/goals/blockers/decisions )这是数据库 学习存储的关键集成模式你不必自己建表、自己写 SQL直接复用 agno 数据库层为学习系统提供的get_learning/upsert_learning读写接口即可同步方法之外还提供await版本的arecall两者实现几乎一一对应。4.3 跳过自动抽取改用 Agent 工具与前例不同这里的process直接pass注释写明Skip automatic extraction - use tools instead学习行为完全由 Agent 主动触发。这是 AGENTIC 风格的设计自动抽取适合高频、无干扰的场景需要精准控制的场景则应把写入口交给 Agent 的决策。工具在get_tools中返回enable_toolsFalse时可整体关闭def get_tools(self, **kwargs) - List[Callable]: if not self.enable_tools: return [] def add_project_note(note_type: str, content: str) - str: note_type 限定 goal / blocker / decision 之一 ... current self.recall() or ProjectNotes() # 追加到对应列表后 _save(current) def update_project_summary(summary: str) - str: ... current.summary summary self._save(current) tools.append(add_project_note) tools.append(update_project_summary) return tools从 machine.py 的get_tools调度看框架遍历每个存储、调用store.get_tools(...)后把返回的可调用对象统一收进 Agent 的工具集逐个 try/except 并打 debug 日志。因此你在get_tools里返回的普通函数——注意它们闭包捕获了self——会成为真正的 Agent 工具由模型自主决定何时调用调用时经recall - 修改 - _save完整落库。build_context会根据enable_tools在 XML 块尾部追加工具使用指引Use add_project_note to save new goals, blockers, or decisions.让 Agent 知道什么时候该动笔。4.4 挂接数据库与 Agent验证跨会话持久化db PostgresDb(db_urlpostgresqlpsycopg://ai:ailocalhost:5532/ai) project_notes_store ProjectNotesStore( dbdb, context{project_id: learning-machine}, enable_toolsTrue, ) agent Agent( modelOpenAIResponses(idgpt-5.5), dbdb, # Agent 自身也要挂 db 以持久化会话 learningLearningMachine( custom_stores{project_notes: project_notes_store}, ), markdownTrue, )demo 精心设计了三轮对话来证明持久化第一轮让 Agent 把 main goal 记下来——它应当调用add_project_note(goal, ...)第二轮抛出 blocker——Agent 再次调用工具写入blockers第三轮用全新的session_idnew_session提问 What are our current project notes?——因为数据不在会话里、而在数据库的 learning 表中按 project_id 命名空间隔离换会话后依然能 recall 到完整笔记。这正是自定义存储相对会话记忆的核心价值学习是跨会话、按领域维度沉淀的而不是绑定在一次对话里。五、接入 Agent 学习流程的完整时序综合两个示例与 machine.py 源码一个自定义存储注册进LearningMachine后其方法在 Agent 运行时被调用的时机是上下文构建期Agent 准备系统提示词时LearningMachine 对每个存储调用recall(**context)→ 对召回结果调用build_context(data)→ 拼接所有存储的上下文块注入提示词工具注册期对每个存储调用get_tools(**context)返回值并入 Agent 工具集学习沉淀期一次对话结束后或在 ALWAYS 提取流程中调用process(messages, **context)让存储自行抽取抽取前后框架读取was_updated判断是否真的发生了更新每一步都有异常兜底machine.py 对每个存储的 recall / process / build_context / get_tools 都包了 try/except 并打 warning单个存储出错不会拖垮整次运行。这些阶段对应的框架层实现可分别对照 machine.py 中的build_context/abuild_context、get_tools、process/aprocess、recall方法查看。六、从源码可见的自定义存储注意事项命名空间即隔离手段LearningMachine 内置存储对entity_memory、learned_knowledge有默认namespace默认 global自定义存储没有自动命名空间——示例用context[project_id]手动承担了这一角色。多项目、多租户部署时务必自己落实命名空间。max_updates_per_run不直接作用于自定义存储全局的max_updates_per_runmachine.py 中默认 10用来限制每轮提取的更新次数、防止工具调用死循环它主要约束框架代管的 ALWAYS 提取流程自定义存储在process里自行控制写频次更稳妥。序列化限制LearningMachine.from_dict无法从序列化配置重建自定义存储实例——它只能把自定义存储记成import 路径引用仅作信息展示反序列化时你会收到警告custom stores ... cannot be rebuilt from a serialized config; re-attach them programmatically。也就是说Agent 配置走 JSON/字典持久化时自定义存储必须在程序里重新 attach。隔离环境的限制在 rollout/环境隔离场景见 environments/runner.py中框架无法让不透明的自定义存储只读化会直接告警并丢弃custom_stores——这与它对内置存储剥离写标志的处理不同做强化学习回滚类实验时需留意。运行环境前提示例基于OpenAIResponses(idgpt-5.5)需要OPENAI_API_KEY第二个示例需要本地 PostgreSQLpgvector 容器例如./cookbook/scripts/run_pgvector.sh提供的agnohq/pgvector连接串postgresqlpsycopg://ai:ailocalhost:5532/ai。依赖清单与虚拟环境搭建可参考 08_learning 目录的 setup 脚本与 requirements.txt运行命令形如python cookbook/08_learning/08_custom_stores/01_minimal_custom_store.py python cookbook/08_learning/08_custom_stores/02_custom_store_with_db.py本目录的 TEST_LOG.md 显示测试尚处于 PENDING 状态运行后可按其格式回填验证记录。七、如何选择你的实现路线结合两个示例可以沉淀出一张决策表维度内存型示例一数据库型示例二存储介质进程内字典PostgreSQLget_learning/upsert_learningSchemadict类型化 dataclass命名空间context[project_id]namespaceproject_id学习触发process自动抽取可换成模型抽取Agent 通过工具add_project_note等主动记录生命周期进程内重启即失跨会话、跨进程持久适用场景原型、单进程、演示生产、审计、需要沉淀长期项目知识如果你的自定义存储只在进程内充当临时上下文如示例一跳过数据库即可一旦知识需要换个 session、换个进程还能想起来就必须走数据库路线并正确设计learning_typenamespaceid三个键。至于抽取质量关键词启发式只适合演示生产实现要么在process里用模型精炼抽取ALWAYS 自动模式要么把写权限交给带参工具、让 Agent 自己判断何时记录AGENTIC 工具模式即示例二的设计。结语自定义学习存储是 agno LearningMachine 的可扩展出口你只需要守住learning_type、schema、recall/arecall、process/aprocess、build_context、get_tools/aget_tools、was_updated这套接口且框架对窄签名、缺失instructions()的旧式实现保持鸭子类型兼容就能让 Agent 记住任何形态的知识——项目上下文也好、客户笔记也罢。从 protocol.py 定义接口到 machine.py 统一调度再到两个 cookbook 示例给出内存与数据库两套完整落地模板这条从接口契约到生产可用的路径在仓库中是连贯、可验证的。动手运行示例、替换成你自己的业务 schema 与存储介质就是接入 agno 自定义学习体系最快的方式。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
