agno Team 中各成员如何共享同一份 session state?
2026/9/11 16:41:09 网站建设 项目流程

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.2gpt-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 的前半部分:

  1. Team上设置db,会话状态按session_id/user_id持久化;
  2. 第一次print_response传入session_state={...}并设置add_session_state_to_context=True,状态会被注入上下文;
  3. 成员 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?”,说明状态已持久化并被成员读到。

如何验证共享已生效

两种验证方式都出自本目录示例:

  1. 看日志中的上下文。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.
  1. 调用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 和成员上都配置dbadd_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_itemget_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=Truereport_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.pyagentic_session_state.pyoverwrite_stored_session_state.py记录为 PASS,而state_sharing.pynested_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),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询