
agno Team 中各成员如何共享同一份 session state【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno在 agno 里构建多成员 Team 时一个常见需求是第一轮对话里传入的用户信息姓名、年龄或任务数据购物清单在后续轮次中仍然可读并且 Team 的各个成员 Agent 都能拿到同一份数据而不是各自维护副本。agno 的做法是把数据放在 Team 的session_state上用db把状态挂到一个会话由session_id/user_id标识上再通过上下文注入让成员读取。下面的操作路径全部来自仓库内cookbook/03_teams/21_state/目录的示例模型均使用OpenAIResponses示例中为gpt-5.2与gpt-5-mini。准备条件cookbook/03_teams/21_state/的 README 给出的运行前提通过direnv allow加载环境变量例如OPENAI_API_KEY使用.venvs/demo/bin/python运行 cookbook 示例部分示例需要额外服务PostgreSQL、LanceDB 等本目录的 state 示例不需要。状态存储用两种 db 之一InMemoryDb()内存态只在同一进程内有效SqliteDb(db_filetmp/agents.db)写入tmp/agents.db文件可跨进程保留。主路径用 session_state db 让成员读取同一份状态共享状态的关键配置有三个对应 state_sharing.py 的前半部分在Team上设置db会话状态按session_id/user_id持久化第一次print_response传入session_state{...}并设置add_session_state_to_contextTrue状态会被注入上下文成员 Agent 的instructions用{user_name}、{age}占位符引用状态中的字段成员由此读到 Team 层的同一份状态。from agno.agent import Agent from agno.db.in_memory import InMemoryDb from agno.models.openai import OpenAIResponses from agno.team import Team, TeamMode user_advisor Agent( roleUser Advisor, descriptionYou answer questions related to the user., modelOpenAIResponses(idgpt-5.2), instructionsUsers name is {user_name} and age is {age}, ) state_team Team( dbInMemoryDb(), modelOpenAIResponses(idgpt-5.2), instructionsYou are a team that answers questions related to the user. Delegate to the member agent to address user requests or answer any questions about the user., members[user_advisor], modeTeamMode.route, ) if __name__ __main__: # 第一轮传入 session_state state_team.print_response( Write a short poem about my name and age, session_idsession_1, user_iduser_1, session_state{user_name: John, age: 30}, add_session_state_to_contextTrue, ) # 第二轮同一 session_id/user_id不再传 session_state state_team.print_response( How old am I?, session_idsession_1, user_iduser_1, add_session_state_to_contextTrue, )在仓库根目录下用.venvs/demo/bin/python cookbook/03_teams/21_state/state_sharing.py运行该文件原文件还包含第二个 Team见后文“共享成员交互”一节。判断依据第二轮调用没有再传session_state如果成员仍能从第一轮传入的{user_name: John, age: 30}回答 “How old am I?”说明状态已持久化并被成员读到。如何验证共享已生效两种验证方式都出自本目录示例看日志中的上下文。TEST_LOG.md 记录了同一模式change_state_on_run.py的调试日志下面是文档示例输出可以看到instructions里的占位符已被session_state的值替换DEBUG system DEBUG Users name is John and age is 30 DEBUG user DEBUG What is my name? DEBUG assistant DEBUG Your name is John.调用team.get_session_state()打印 Team 当前存储的状态。agentic_session_state.py 和 overwrite_stored_session_state.py 都用print(fSession state: {team.get_session_state()})作为运行后的检查手段。可选分支让 Team 和成员在运行中修改状态默认情况下session_state由调用方传入如果希望模型自己在对话过程中更新状态例如维护购物清单需要开启enable_agentic_stateTrue并在 Team 和成员上都配置db、add_session_state_to_contextTruefrom agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.team import Team db SqliteDb(db_filetmp/agents.db) shopping_agent Agent( nameShopping List Agent, roleManage the shopping list, modelOpenAIResponses(idgpt-5-mini), dbdb, add_session_state_to_contextTrue, enable_agentic_stateTrue, ) team Team( members[shopping_agent], session_state{shopping_list: []}, dbdb, add_session_state_to_contextTrue, enable_agentic_stateTrue, descriptionYou are a team that manages a shopping list and chores, show_members_responsesTrue, ) if __name__ __main__: team.print_response(Add milk, eggs, and bread to the shopping list) team.print_response(I picked up the eggs, now whats on my list?) print(fSession state: {team.get_session_state()})session_state{shopping_list: []}在 Team 上给出初始值show_members_responsesTrue会打印成员 Agent 的响应便于观察成员是否基于共享状态作答。验证方式是最后一行team.get_session_state()的输出TEST_LOG.md 中该示例记录为 PASS。同一目录的 nested_shared_state.py 展示了更复杂的层级顶层 Team 定义session_state{shopping_list: [], chores: []}子 Teamshopping_mgmt_team、meal_planning_team下的成员通过带RunContext参数的工具函数add_item、get_ingredients等读写run_context.session_state各层因此操作的是同一份状态。可选分支按用户隔离状态与覆盖已存状态change_state_on_run.py 展示同一 Team 上按会话隔离状态每次print_response传入不同的session_id/user_id与不同的session_state同一session_id内的后续调用不传状态也能沿用首轮值。适合多用户场景每个用户的状态互不干扰。overwrite_stored_session_state.py 展示另一种行为设置overwrite_db_session_stateTrue后运行时传入的session_state会替换db 中已存的整份状态而不是合并team Team( modelOpenAIResponses(idgpt-5.2), dbSqliteDb(db_filetmp/agents.db), members[], markdownTrue, session_state{}, add_session_state_to_contextTrue, overwrite_db_session_stateTrue, ) team.print_response( Can you tell me whats in your session_state?, session_state{shopping_list: [Potatoes]}, streamTrue, ) print(fStored session state: {team.get_session_state()}) team.print_response( Can you tell me what is in your session_state?, session_state{secret_number: 43}, streamTrue, ) print(fStored session state: {team.get_session_state()})两次运行各传入一份session_state第二次之后存储的状态变为{secret_number: 43}。该示例在 TEST_LOG 中记录为 PASS。可选分支共享成员之间的交互share_member_interactionsstate_sharing.py 原文件里的第二个 Team 演示的是另一条机制——不是共享数据而是把成员 Agent 之间的交互历史共享给其他成员interaction_team Team( modelOpenAIResponses(idgpt-5-mini), dbSqliteDb(db_filetmp/agents.db), members[web_research_agent, report_agent], share_member_interactionsTrue, instructions[ You are a team of agents that can research the web and write a report., First, research the web for information about the topic., Then, use your report agent to write a report from the web research., ], show_members_responsesTrue, ) interaction_team.print_response(How are LEDs made?)share_member_interactionsTrue让report_agent能看到web_research_agent的检索交互从而基于同一批结果写报告show_members_responsesTrue用于打印各成员的响应。它与session_state的区别在于前者共享“交互记录”后者共享“结构化数据”两者可独立或同时使用。限制与已知情况InMemoryDb只在本进程内有效示例重启后状态即丢失需要落盘时使用SqliteDb(db_filetmp/agents.db)它会创建/写入tmp/agents.db文件。TEST_LOG.mdvalidation run 2026-02-15中change_state_on_run.py、agentic_session_state.py、overwrite_stored_session_state.py记录为 PASS而state_sharing.py与nested_shared_state.py记录为 “Timeout after 30s”。这是校验脚本 30 秒时限下的结果日志未给出排障结论state_sharing.py的第二个 Team 依赖WebSearchTools的网络调用本地运行耗时较长属预期可按需只保留文件中不依赖网络搜索的前半部分。本目录的示例模型统一为gpt-5.2/gpt-5-miniOpenAIResponses换模型时保持示例中的参数名不变即可仓库未对这些示例给出其他模型的兼容说明。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考