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_file="tmp/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_context=True,状态会被注入上下文; - 成员 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( role="User Advisor", description="You answer questions related to the user.", model=OpenAIResponses(id="gpt-5.2"), instructions="User's name is {user_name} and age is {age}", ) state_team = Team( db=InMemoryDb(), model=OpenAIResponses(id="gpt-5.2"), instructions="You 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], mode=TeamMode.route, ) if __name__ == "__main__": # 第一轮:传入 session_state state_team.print_response( "Write a short poem about my name and age", session_id="session_1", user_id="user_1", session_state={"user_name": "John", "age": 30}, add_session_state_to_context=True, ) # 第二轮:同一 session_id/user_id,不再传 session_state state_team.print_response( "How old am I?", session_id="session_1", user_id="user_1", add_session_state_to_context=True, )在仓库根目录下用.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(f"Session state: {team.get_session_state()}")作为运行后的检查手段。
可选分支:让 Team 和成员在运行中修改状态
默认情况下session_state由调用方传入;如果希望模型自己在对话过程中更新状态(例如维护购物清单),需要开启enable_agentic_state=True,并在 Team 和成员上都配置db、add_session_state_to_context=True:
from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.team import Team db = SqliteDb(db_file="tmp/agents.db") shopping_agent = Agent( name="Shopping List Agent", role="Manage the shopping list", model=OpenAIResponses(id="gpt-5-mini"), db=db, add_session_state_to_context=True, enable_agentic_state=True, ) team = Team( members=[shopping_agent], session_state={"shopping_list": []}, db=db, add_session_state_to_context=True, enable_agentic_state=True, description="You are a team that manages a shopping list and chores", show_members_responses=True, ) if __name__ == "__main__": team.print_response("Add milk, eggs, and bread to the shopping list") team.print_response("I picked up the eggs, now what's on my list?") print(f"Session state: {team.get_session_state()}")session_state={"shopping_list": []}在 Team 上给出初始值;show_members_responses=True会打印成员 Agent 的响应,便于观察成员是否基于共享状态作答。验证方式是最后一行team.get_session_state()的输出,TEST_LOG.md 中该示例记录为 PASS。
同一目录的 nested_shared_state.py 展示了更复杂的层级:顶层 Team 定义session_state={"shopping_list": [], "chores": []},子 Team(shopping_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_state=True后,运行时传入的session_state会替换db 中已存的整份状态,而不是合并:
team = Team( model=OpenAIResponses(id="gpt-5.2"), db=SqliteDb(db_file="tmp/agents.db"), members=[], markdown=True, session_state={}, add_session_state_to_context=True, overwrite_db_session_state=True, ) team.print_response( "Can you tell me what's in your session_state?", session_state={"shopping_list": ["Potatoes"]}, stream=True, ) print(f"Stored session state: {team.get_session_state()}") team.print_response( "Can you tell me what is in your session_state?", session_state={"secret_number": 43}, stream=True, ) print(f"Stored session state: {team.get_session_state()}")两次运行各传入一份session_state,第二次之后存储的状态变为{"secret_number": 43}。该示例在 TEST_LOG 中记录为 PASS。
可选分支:共享成员之间的交互(share_member_interactions)
state_sharing.py 原文件里的第二个 Team 演示的是另一条机制——不是共享数据,而是把成员 Agent 之间的交互历史共享给其他成员:
interaction_team = Team( model=OpenAIResponses(id="gpt-5-mini"), db=SqliteDb(db_file="tmp/agents.db"), members=[web_research_agent, report_agent], share_member_interactions=True, 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_responses=True, ) interaction_team.print_response("How are LEDs made?")share_member_interactions=True让report_agent能看到web_research_agent的检索交互,从而基于同一批结果写报告;show_members_responses=True用于打印各成员的响应。它与session_state的区别在于:前者共享“交互记录”,后者共享“结构化数据”,两者可独立或同时使用。
限制与已知情况
InMemoryDb只在本进程内有效,示例重启后状态即丢失;需要落盘时使用SqliteDb(db_file="tmp/agents.db"),它会创建/写入tmp/agents.db文件。- TEST_LOG.md(validation 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-mini(OpenAIResponses),换模型时保持示例中的参数名不变即可,仓库未对这些示例给出其他模型的兼容说明。
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考