有多种方法可以在团队成员之间共享状态。
共享团队状态
Team Session State 支持跨 Agent 团队进行复杂的状态管理,具备共享和私有状态的功能。
团队通常需要在共享信息(例如购物清单)上进行协调,同时维护各自的私有指标或配置。Agno 为此提供了一个优雅的三层状态系统。
Agno 的团队状态管理提供了三个不同的级别:
- Team’s team_session_state - 所有团队成员都可以访问的共享状态。
- Team’s session_state - 仅团队领导者可以访问的私有状态。
- Agent’s session_state - 每个团队成员的私有状态。
如何使用 Team Session State
您可以设置 Team
上的 team_session_state
参数来在团队成员之间共享状态。
此状态对所有团队成员都可用,并在他们之间同步。
例如:
team = Team(
members=[agent1, agent2, agent3],
team_session_state={"shopping_list": []},
)
成员可以在工具中使用 team_session_state
属性来访问共享状态。
例如:
def add_item(agent: Agent, item: str) -> str:
"""将商品添加到购物清单并返回确认信息。
Args:
item (str): 要添加到购物清单的商品。
"""
# 如果商品不在列表中,则添加它
if item.lower() not in [
i.lower() for i in agent.team_session_state["shopping_list"]
]:
agent.team_session_state["shopping_list"].append(item)
return f"已将 '{item}' 添加到购物清单"
else:
return f"'{item}' 已在购物清单中"
这是一个关于团队管理共享购物清单的简单示例:
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
# 定义可处理共享团队状态的工具
def add_item(agent: Agent, item: str) -> str:
"""将商品添加到购物清单。"""
if item.lower() not in [
i.lower() for i in agent.team_session_state["shopping_list"]
]:
agent.team_session_state["shopping_list"].append(item)
return f"已将 '{item}' 添加到购物清单"
else:
return f"'{item}' 已在购物清单中"
def remove_item(agent: Agent, item: str) -> str:
"""从购物清单中移除商品。"""
for i, list_item in enumerate(agent.team_session_state["shopping_list"]):
if list_item.lower() == item.lower():
agent.team_session_state["shopping_list"].pop(i)
return f"已将 '{list_item}' 从购物清单中移除"
return f"在购物清单中找不到 '{item}'"
# 创建一个管理购物清单的代理
shopping_agent = Agent(
name="Shopping List Agent",
role="管理购物清单",
model=OpenAIChat(id="gpt-4o-mini"),
tools=[add_item, remove_item],
)
# 定义团队级别的工具
def list_items(team: Team) -> str:
"""列出购物清单中的所有商品。"""
# 访问共享状态(非私有状态)
shopping_list = team.team_session_state["shopping_list"]
if not shopping_list:
return "购物清单为空。"
items_text = "\n".join([f"- {item}" for item in shopping_list])
return f"当前购物清单:\n{items_text}"
def add_chore(team: Team, chore: str) -> str:
"""将已完成的家务添加到团队的私有日志中。"""
# 访问团队的私有状态
if "chores" not in team.session_state:
team.session_state["chores"] = []
team.session_state["chores"].append(chore)
return f"已记录家务: {chore}"
# 创建一个同时具有共享和私有状态的团队
shopping_team = Team(
name="Shopping Team",
mode="coordinate",
model=OpenAIChat(id="gpt-4o-mini"),
members=[shopping_agent],
# 共享状态 - 所有成员都可以访问
team_session_state={"shopping_list": []},
# 团队私有状态 - 仅团队可访问
session_state={"chores": []},
tools=[list_items, add_chore],
instructions=[
"你管理一个购物清单。",
"将添加/移除请求转发给 Shopping List Agent。",
"使用 list_items 显示当前列表。",
"使用 add_chore 记录已完成的任务。",
],
show_tool_calls=True,
)
# 用法示例
shopping_team.print_response("添加牛奶、鸡蛋和面包", stream=True)
print(f"共享状态: {shopping_team.team_session_state}")
shopping_team.print_response("我的列表里有什么?", stream=True)
shopping_team.print_response("我拿到了鸡蛋", stream=True)
print(f"共享状态: {shopping_team.team_session_state}")
print(f"团队私有状态: {shopping_team.session_state}")
请注意,共享工具使用 agent.team_session_state
,这使得状态可以在整个团队中传播和持久化——即使是团队内的子团队。这确保了所有成员拥有一致的共享状态。
相比之下,团队特定的工具使用 team.session_state
,允许私有、团队特定的状态。例如,团队领导者的工具会使用 team.session_state
来维护自己的会话状态。
请参阅完整示例 此处。
代理上下文 (Agentic Context)
团队领导者维护一个共享的上下文,该上下文由代理(即团队领导者)更新,并在需要时发送给团队成员。
代理上下文对于代理之间的有效信息共享和协作至关重要,团队响应的质量取决于团队领导者如何有效地管理这个共享的代理上下文。
这可能需要为团队领导者使用更高质量的模型,以确保团队响应的质量。
团队成员的任务和响应会自动添加到团队上下文中,但代理上下文需要由开发者启用。
启用代理上下文
要启用团队领导者维护代理上下文,请设置 enable_agentic_context=True
。
这将允许团队领导者在运行期间维护和更新团队上下文。
team = Team(
members=[agent1, agent2, agent3],
enable_agentic_context=True, # 启用团队领导者维护代理上下文
)
团队成员交互
Agent Teams 可以共享成员之间的交互,从而使代理能够从彼此的输出中学习:
team = Team(
members=[agent1, agent2, agent3],
share_member_interactions=True, # 分享交互
)
Responses are generated using AI and may contain mistakes.