Agno MongoDB 存储集成指南：为 Agent、Team 与 Workflow 构建持久化会话

Agno MongoDB 存储集成指南为 Agent、Team 与 Workflow 构建持久化会话【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南聚焦 Agno 框架开源项目agno如何通过 MongoDB 为 Agent、Team 与 Workflow 提供持久化存储。你将掌握从环境搭建、连接配置到运行 Agent/Team/Workflow 完整示例的实战路径并深入理解MongoDb/AsyncMongoDb底层集合设计与索引机制从而为多轮对话记忆、会话续接等生产场景构建可靠的存储底座。概述为什么用 MongoDB 作为 Agno 的存储层Agno 是一个用于构建、运行和管理 Agent 平台的开源框架。在实际应用中Agent 与用户的对话往往需要跨请求保持状态——例如记住上一轮聊到的内容下次继续这个话题。Agno 通过db参数把 Agent、Team、Workflow 的运行数据会话、每次运行、记忆、指标等写入外部数据库。本指南对应的官方示例位于 cookbook/06_storage/mongo/README.md它演示了Agent with MongoDB storage—— 单个 Agent 使用 MongoDB 持久化会话Team with MongoDB storage—— 由多个成员 Agent 组成的 Team 共享同一个 MongoDB 存储同一目录下的async_mongo/子目录还提供了 Agent、Team、Workflow 的异步版本示例。MongoDB 的优势在于文档型模型与 Agent 运行记录会话、run、memory天然契合无需预定义 Schema内置唯一索引与聚合管道可用于高效查询历史会话与上下文压缩。环境搭建安装依赖与启动本地 MongoDB安装 Python 依赖同步阻塞式接口只需要pymongouv pip install pymongo异步接口AsyncMongoDb需要额外安装motoruv pip install pymongo motor说明在底层实现 libs/agno/agno/db/mongo/mongo.py 中MongoDb会在pymongo缺失时直接抛出ImportError(pymongo not installed. Please install it using pip install pymongo)而AsyncMongoDb见 libs/agno/agno/db/mongo/async_mongo.py要求pymongo且motor与 PyMongo 异步接口pymongo4.9的AsyncMongoClient至少安装其一优先推荐 PyMongo 异步接口。启动本地 MongoDB使用 Docker 快速启动一个带认证的本地实例docker run -d \ --name local-mongo \ -p 27017:27017 \ -e MONGO_INITDB_ROOT_USERNAMEmongoadmin \ -e MONGO_INITDB_ROOT_PASSWORDsecret \ mongo也可以直接使用仓库自带的一键脚本./cookbook/scripts/run_mongodb.sh该脚本内容与上面的docker run命令完全一致见 cookbook/scripts/run_mongodb.sh适合在本地开发时快速拉起服务。连接串与凭据示例中的连接串为mongodb://mongoadmin:secretlocalhost:27017即用户名mongoadmin、密码secret对应 Docker 启动时设置的MONGO_INITDB_ROOT_USERNAME/MONGO_INITDB_ROOT_PASSWORD环境变量。生产环境中请替换为真实的账号密码并建议使用环境变量或密钥管理工具注入而不是硬编码在代码里。配置创建 MongoDb 数据库对象在示例 cookbook/06_storage/mongo/mongodb_for_agent.py 中配置方式非常简洁from agno.agent import Agent from agno.db.mongo import MongoDb db MongoDb(db_urlmongodb://username:passwordlocalhost:27017) agent Agent( dbdb, add_history_to_contextTrue, )MongoDb 构造参数详解从源码 libs/agno/agno/db/mongo/mongo.py 的MongoDb.__init__可以看到除db_url外还支持以下常用参数参数默认值说明db_urlNoneMongoDB 连接串如mongodb://user:passhost:27017与db_client二选一均未提供时抛出ValueErrordb_clientNone直接传入已构造好的pymongo.MongoClient实例适合复用既有连接池db_nameagno使用的数据库名称session_collection继承自BaseDb的默认值存储会话的集合名默认agno_sessionsruns_collection默认派生存储每次运行的集合名默认agno_runs若指定了 session_collection 则为session_collection_runsmemory_collection默认值存储记忆的集合名metrics_collection默认值存储指标数据的集合名eval_collection默认值存储评估运行记录的集合名knowledge_collection默认值存储知识文档的集合名traces_collection/spans_collection默认值存储可观测性追踪与跨度数据的集合名schedules_collection/schedule_runs_collection默认值存储调度任务及其运行记录的集合名learnings_collection默认值存储学习learnings数据的集合名id由连接串与库名派生数据库实例 ID从BaseDb基类libs/agno/agno/db/base.py可以看出会话集合的默认名是agno_sessions运行集合默认是agno_runs如果你自定义了session_collectionmy_sessions则运行集合会自动派生为my_sessions_runs。版本提示较新的 Agno 存储模型采用会话集合 独立运行集合每个 run 一个文档的双集合设计并通过get_latest_schema_version/upsert_schema_version与MigrationManager配合管理 Schema 版本迁移完成后旧版本遗留在会话文档中的runs字段可通过cleanup_legacy_runs_field()清理详见 libs/agno/agno/db/mongo/mongo.py。自动创建集合与索引MongoDb会在首次写入时自动创建所需集合sessions、runs、memories、metrics、evals、knowledge、schedules、schedule_runs 等并依据 libs/agno/agno/db/mongo/schemas.py 中定义的模式为每个集合建立索引例如sessions 集合session_id唯一索引user_id、session_type、agent_id、team_id、workflow_id、created_at、updated_at普通索引runs 集合run_id唯一索引session_id、run_type、agent_id、team_id、workflow_id、user_id、parent_run_id、status、created_at普通索引以及(session_id, run_index)复合索引memories 集合memory_id唯一索引等metrics 集合(user_id, date, aggregation_period)复合唯一索引用于按用户聚合指标。索引创建逻辑位于 libs/agno/agno/db/mongo/utils.py 的create_collection_indexes异步版本为create_collection_indexes_async每个索引独立尝试创建单个索引失败例如遗留集合中存在选项冲突不会阻断其余索引同时会尝试清理旧的date_1_aggregation_period_1指标索引。这套设计让历史数据迁移与唯一性约束都能在 MongoDB 层得到保证。实战一Agent 使用 MongoDB 存储完整示例见 cookbook/06_storage/mongo/mongodb_for_agent.py核心代码from agno.agent import Agent from agno.db.mongo import MongoDb from agno.tools.websearch import WebSearchTools db_url mongodb://mongoadmin:secretlocalhost:27017 db MongoDb(db_urldb_url) agent Agent( dbdb, tools[WebSearchTools()], add_history_to_contextTrue, ) if __name__ __main__: agent.print_response(How many people live in Canada?) agent.print_response(What is their national anthem called?)运行前安装依赖uv pip install openai pymongo要点解读dbdb把 MongoDB 实例挂到 Agent 上每次对话的运行记录run都会写入agno_runs集合会话元数据写入agno_sessions集合add_history_to_contextTrue表示把历史对话记录加入上下文这样第二个问题加拿大国歌叫什么能够结合第一个问题加拿大有多少人形成连贯对话——这正是持久化存储与记忆配合的典型场景tools[WebSearchTools()]让 Agent 具备联网检索能力工具调用记录同样会被持久化。运行示例python cookbook/06_storage/mongo/mongodb_for_agent.py运行记录如何被存储底层upsert_run见 libs/agno/agno/db/mongo/mongo.py以run_id为主键执行replace_one(..., upsertTrue)的单文档 upsert属于 O(1) 操作专为 HITL人工介入或后台模式下频繁更新运行状态而优化查询历史时get_session会通过聚合管道按session_id拉取该会话的运行记录并支持runs_limit参数把最近 N 条上下文相关运行的筛选下推到数据库端完成见_get_session_runs_docs。实战二Team 使用 MongoDB 存储完整示例见 cookbook/06_storage/mongo/mongodb_for_team.py。该示例构建了一个由HackerNews 研究员与网络搜索员组成的 Team把 MongoDB 作为 Team 级共享存储from typing import List from agno.agent import Agent from agno.db.mongo import MongoDb from agno.models.openai import OpenAIChat from agno.team import Team from agno.tools.hackernews import HackerNewsTools from agno.tools.websearch import WebSearchTools from pydantic import BaseModel db_url mongodb://mongoadmin:secretlocalhost:27017 db MongoDb(db_urldb_url) class Article(BaseModel): title: str summary: str reference_links: List[str] hn_researcher Agent( nameHackerNews Researcher, modelOpenAIChat(gpt-5.6-luna), roleGets top stories from hackernews., tools[HackerNewsTools()], ) web_searcher Agent( nameWeb Searcher, modelOpenAIChat(gpt-5.6-luna), roleSearches the web for information on a topic, tools[WebSearchTools()], add_datetime_to_contextTrue, ) hn_team Team( nameHackerNews Team, modelOpenAIChat(gpt-5.6-luna), members[hn_researcher, web_searcher], dbdb, instructions[ First, search hackernews for what the user is asking about., Then, ask the web searcher to search for each story to get more information., Finally, provide a thoughtful and engaging summary., ], output_schemaArticle, markdownTrue, show_members_responsesTrue, add_member_tools_to_contextFalse, ) if __name__ __main__: hn_team.print_response(Write an article about the top 2 stories on hackernews)运行前安装依赖uv pip install openai ddgs newspaper4k lxml_html_clean agno要点解读Team与Agent使用相同的MongoDb实例说明会话/运行存储对二者是统一的抽象底层通过session_typeAGENT/TEAM/WORKFLOW区分不同组件output_schemaArticle让 Team 输出结构化结果标题、摘要、参考链接markdownTrue输出 Markdown 格式show_members_responsesTrue会展示各成员 Agent 的响应add_member_tools_to_contextFalse避免把成员工具重复注入上下文Team 的完整执行成员分工、工具调用、最终汇总作为一个整体 run 写入 MongoDBget_sessions(component_id...)可以通过team_id精确过滤某个 Team 的会话历史参见 libs/agno/agno/db/mongo/mongo.py 中get_sessions的$or过滤逻辑。进阶异步版本 AsyncMongoDbAgent / Team / Workflow异步示例位于 cookbook/06_storage/mongo/async_mongo/涵盖 Agent、Team、Workflow 三种场景适用于 FastAPI 等异步服务或高并发调用场景。异步 Agentcookbook/06_storage/mongo/async_mongo/async_mongodb_for_agent.pyimport asyncio from agno.agent import Agent from agno.db.mongo import AsyncMongoDb from agno.tools.websearch import WebSearchTools db_url mongodb://mongoadmin:secretlocalhost:27017 db AsyncMongoDb(db_urldb_url) agent Agent( dbdb, tools[WebSearchTools()], add_history_to_contextTrue, ) if __name__ __main__: asyncio.run(agent.aprint_response(How many people live in Canada?)) asyncio.run(agent.aprint_response(What is their national anthem called?))注意运行方式改为aprint_response并用asyncio.run包装。异步 Teamcookbook/06_storage/mongo/async_mongo/async_mongodb_for_team.py 与同步 Team 示例结构一致仅将MongoDb换为AsyncMongoDb、print_response换为aprint_responseimport asyncio from typing import List from agno.agent import Agent from agno.db.mongo import AsyncMongoDb from agno.models.openai import OpenAIChat from agno.team import Team from agno.tools.hackernews import HackerNewsTools from agno.tools.websearch import WebSearchTools from pydantic import BaseModel db_url mongodb://mongoadmin:secretlocalhost:27017 db AsyncMongoDb(db_urldb_url) class Article(BaseModel): title: str summary: str reference_links: List[str] hn_researcher Agent( nameHackerNews Researcher, modelOpenAIChat(gpt-5.6-luna), roleGets top stories from hackernews., tools[HackerNewsTools()], ) web_searcher Agent( nameWeb Searcher, modelOpenAIChat(gpt-5.6-luna), roleSearches the web for information on a topic, tools[WebSearchTools()], add_datetime_to_contextTrue, ) hn_team Team( nameHackerNews Team, modelOpenAIChat(gpt-5.6-luna), members[hn_researcher, web_searcher], dbdb, instructions[ First, search hackernews for what the user is asking about., Then, ask the web searcher to search for each story to get more information., Finally, provide a thoughtful and engaging summary., ], output_schemaArticle, markdownTrue, show_members_responsesTrue, add_member_tools_to_contextFalse, ) if __name__ __main__: asyncio.run( hn_team.aprint_response( Write an article about the top 2 stories on hackernews ) )异步 Workflowcookbook/06_storage/mongo/async_mongo/async_mongodb_for_workflow.py 展示了 Workflow 级别的存储一个内容创作工作流先由研究团队Hackernews Agent Web Agent完成调研再由内容规划 Agent基于调研产出四周内容排期。工作流本身挂载dbdb从而持久化整条执行链路import asyncio from agno.agent import Agent from agno.db.mongo import AsyncMongoDb from agno.models.openai import OpenAIChat from agno.team import Team from agno.tools.hackernews import HackerNewsTools from agno.tools.websearch import WebSearchTools from agno.workflow.step import Step from agno.workflow.workflow import Workflow db_url mongodb://mongoadmin:secretlocalhost:27017 db AsyncMongoDb(db_urldb_url) hackernews_agent Agent( nameHackernews Agent, modelOpenAIChat(idgpt-5.6-luna), tools[HackerNewsTools()], roleExtract key insights and content from Hackernews posts, ) web_agent Agent( nameWeb Agent, modelOpenAIChat(idgpt-5.6-luna), tools[WebSearchTools()], roleSearch the web for the latest news and trends, ) content_planner Agent( nameContent Planner, modelOpenAIChat(idgpt-5.6-luna), instructions[ Plan a content schedule over 4 weeks for the provided topic and research content, Ensure that I have posts for 3 posts per week, ], ) research_team Team( nameResearch Team, members[hackernews_agent, web_agent], instructionsResearch tech topics from Hackernews and the web, ) research_step Step( nameResearch Step, teamresearch_team, ) content_planning_step Step( nameContent Planning Step, agentcontent_planner, ) content_creation_workflow Workflow( nameContent Creation Workflow, descriptionAutomated content creation from blog posts to social media, dbdb, steps[research_step, content_planning_step], ) if __name__ __main__: asyncio.run( content_creation_workflow.aprint_response( inputAI trends in 2024, markdownTrue, ) )运行方式pip install openai ddgs pymongo motor python cookbook/06_storage/mongo/async_mongo/async_mongodb_for_workflow.pyAsyncMongoDb 的驱动选择从 libs/agno/agno/db/mongo/async_mongo.py 的源码可以看到AsyncMongoDb同时兼容两类异步驱动PyMongo 异步接口pymongo4.9提供的AsyncMongoClient推荐Motormotor.motor_asyncio.AsyncIOMotorClient传统方案。两者均未安装时导入AsyncMongoDb会直接抛出ImportError提示安装其一。AsyncMongoDb内部通过_detect_client_type自动识别传入 client 的类型行为与同步MongoDb保持高度一致含索引创建、迁移、指标聚合等全套能力对应create_collection_indexes_async等异步工具函数。存储模型与测试佐证集合与字段概览综合 libs/agno/agno/db/mongo/schemas.py 的定义MongoDb 会按需维护以下集合均自动建索引集合核心唯一索引主要用途agno_sessionssession_id会话元数据组件类型、所属 agent/team/workflow、用户agno_runsrun_id复合(session_id, run_index)每次运行含工具调用、消息、状态agno_memoriesmemory_id记忆条目agno_metrics(user_id, date, aggregation_period)用户维度指标聚合agno_evalsrun_id评估运行记录agno_knowledgeid知识文档agno_traces/agno_spanstrace_id/span_id可观测性追踪与跨度agno_schedules/agno_schedule_runsid调度任务与运行记录agno_learningslearning_id学习数据测试用例参考仓库为 MongoDB 存储提供了完整的单元测试与集成测试可作为理解行为边界的参考libs/agno/tests/unit/db/test_mongo.py —— 同步MongoDb的学习数据读写、唯一键 upsert、_id剥离、错误透传等行为libs/agno/tests/unit/db/test_async_mongo.py —— 异步版本的对应行为libs/agno/tests/integration/db/async_mongo/test_db.py 与 libs/agno/tests/integration/db/async_mongo/conftest.py —— 需要真实 MongoDB 的集成测试libs/agno/tests/unit/db/test_mongo_scheduler.py —— 调度任务在 MongoDB 上的生命周期。一处值得注意的底层细节MongoDb在查询会话时会对历史运行做上下文过滤HISTORY_SKIP_STATUSES定义的终止状态 run 会被跳过成员子 run 也会被过滤并把最近 N 条的排序与截取下推到 MongoDB 聚合管道执行_get_session_runs_docs。这意味着即使某个会话积累了海量运行记录add_history_to_contextTrue的上下文构建也只读取与当前对话相关的最近若干条从而控制 Token 开销与查询延迟。小结与最佳实践通过本文你已经在 Agno 中完成 MongoDB 存储的完整接入安装pymongo异步场景再加motor、用 Docker 或 cookbook/scripts/run_mongodb.sh 启动本地实例、创建MongoDb(db_url...)并挂载到 Agent/Team/Workflow。实践建议显式指定db_name默认库名为agno多环境dev/prod或多租户部署建议显式指定避免数据混库生产环境使用独立账号连接串中的账号应遵循最小权限原则只授予所需数据库的读写权限善用add_history_to_contextTrue与持久化存储配合即可获得跨会话记忆能力同时依靠底层的上下文过滤机制控制输入长度异步场景优先用 PyMongo 异步接口AsyncMongoDb已同时兼容 PyMongo async 与 Motor新项目推荐pymongo4.9的AsyncMongoClient关注 Schema 迁移Agno 为 MongoDB 提供了基于版本戳的迁移机制versions集合记录各表 Schema 版本升级 Agno 版本后建议按官方迁移说明执行MigrationManager(db).up()确认无误后再清理遗留字段。更多相关示例可继续阅读 cookbook/06_storage/README.md 与 cookbook/06_storage/mongo/async_mongo/README.md。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考