第 19 章 · Team 进阶:共享状态与成员通信
本章目标:掌握 Team 的四种协作模式切换、成员间共享 session state 的正确姿势,以及如何通过流式事件观察团队内部分工。
19.1 四种协作模式总览
上一章我们用 mode=TeamMode.coordinate 让 leader 分派任务。Agno 2.x 把协作风格显式化为四种模式(from agno.team.mode import TeamMode):
| 模式 | 配置 | 行为 |
|---|---|---|
| Coordinate(默认) | 不设 mode 或设 coordinate | Leader 选人、改写任务、汇总结果 |
| Route | mode=TeamMode.route | Leader 路由给一个成员,直接返回该成员的回复 |
| Broadcast | mode=TeamMode.broadcast | 同一任务发给所有成员 |
| Tasks | mode=TeamMode.tasks | Leader 维护共享任务清单循环执行,直到完成 |
关于遗留参数
旧版本的 respond_directly=True 和 delegate_to_all_members=True 仍然兼容(分别映射到 route 和 broadcast),但新代码一律使用 mode=——当两者同时出现时 mode 会覆盖遗留标志。
19.2 Route 模式:直达回复
客服分流、语言路由这类场景不需要 leader 再加工结果,用 route 模式延迟最低:
import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team, TeamMode
model = OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
)
billing_agent = Agent(name="Billing Agent", role="处理账单、退款类问题", model=model)
tech_agent = Agent(name="Tech Agent", role="处理技术故障类问题", model=model)
router_team = Team(
name="Customer Router",
model=model,
mode=TeamMode.route,
members=[billing_agent, tech_agent],
# 关键开关:用户原始输入原样传给成员,不再由 leader 改写
determine_input_for_members=False,
instructions=["把请求路由到最合适的成员;无法判断时自行回答。"],
)
router_team.print_response("我的订单被扣款两次了", stream=True)determine_input_for_members=False 是 route 模式的重要细节:不设置它时 leader 默认仍会为成员"润色"任务描述,这在需要原话透传的场景反而引入失真。
19.3 Broadcast 与 Tasks 模式
广播适合"多视角并行评审",任务循环适合目标驱动的长流程:
from agno.team import Team, TeamMode
# 广播:三个评审员同时收到同一篇稿子
review_team = Team(
name="Review Panel",
model=model,
mode=TeamMode.broadcast, # 同一任务发给每个成员
members=[
Agent(name="Legal Reviewer", role="从合规角度评审", model=model),
Agent(name="UX Reviewer", role="从用户体验角度评审", model=model),
Agent(name="SEO Reviewer", role="从搜索优化角度评审", model=model),
],
instructions="汇总所有成员意见,输出一份合并后的修改清单。",
)
# 任务循环:leader 拆解任务清单并逐项推进
ops_team = Team(
name="Ops Team",
model=model,
mode=TeamMode.tasks,
max_iterations=6, # 限制任务循环轮数上限,防止失控烧 token
members=[tech_agent, billing_agent],
)Tasks 模式务必设置 max_iterations:任务清单是模型自己生成和勾选的,没有上限的循环等于把预算交给模型。
19.4 成员间共享 Session State
让多个成员围绕同一份可变状态协作(比如共同维护一张购物清单),核心是三件事:team 上声明 session_state、打开 add_session_state_to_context、team 与成员共用同一个数据库:
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.team import Team
db = SqliteDb(db_file="tmp/team.db")
model = OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
)
shopping_agent = Agent(
id="shopping-agent",
name="Shopping List Agent",
role="管理购物清单",
model=model,
db=db, # 与 team 共用同一 db 才能共享会话
add_session_state_to_context=True, # 让成员能看到 session state
enable_agentic_state=True, # 允许成员在对话中自主修改 state
)
team = Team(
name="Household Team",
members=[shopping_agent],
session_state={"shopping_list": []}, # 初始共享状态
db=db,
add_session_state_to_context=True,
enable_agentic_state=True,
description="你们是一个管理购物清单和家务的团队",
show_members_responses=True,
)
team.print_response("把牛奶、鸡蛋和面包加进购物清单")
team.print_response("鸡蛋我已经买了,现在清单上还有什么?")
print(f"最终状态: {team.get_session_state()}")另一个轻量技巧是模板注入——state 变量可以直接写进 instructions,运行时自动替换:
team = Team(
members=[],
session_state={"user_name": "小明"},
instructions="用户的名字是 {user_name},请用名字称呼他。",
markdown=True,
)19.5 观察分工:流式事件与成员响应
调试团队的关键是看清"谁在干活"。三个手段配合使用:
team = Team(
name="Research Team",
model=model,
members=[news_agent, finance_agent],
markdown=True,
show_members_responses=True, # 最终回复中附带各成员的原始回复
)
# stream=True 时可以观察到 leader 委派 -> 成员执行 -> leader 汇总的全过程
for event in team.run("AI 公司最近的动态和 NVDA 股价表现如何?", stream=True):
print(event) # 成员启动、工具调用、内容增量都有对应事件配套的最佳实践还有三条:
- 给成员设置显式
id:委派选择和运行追踪都用 member id 定位,比靠 name 稳定; - 模型继承:成员未设置
model时自动继承 team 的模型——这让"成员用小模型、leader 用大模型"的分档只需一行省略; - callable factories:
members=可以传函数,按session_state在每次运行时动态决定阵容(记得配cache_callables=False)。
19.6 本章小结
- 新代码统一用
mode=TeamMode.coordinate/route/broadcast/tasks,遗留标志respond_directly已被覆盖机制取代; - Route 模式配
determine_input_for_members=False可把用户输入原样透传; - 共享状态的完整条件:team 与成员共用
db+ 双方add_session_state_to_context=True+ 需要自主修改时开enable_agentic_state=True; - instructions 支持
{var}模板注入 session state; show_members_responses+ 流式事件 + 显式 memberid是观察团队分工的三板斧。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 在 Agno 2.x 中,想让「leader 只负责选人并把该成员的回复直接返回」,推荐写法是?
2. Route 模式下要把用户原始输入不做改写地传给成员,应该?
3. 让成员能读取并自主修改团队 session state 的最小配置是?
4. 关于 Tasks 模式,下列说法正确的是?
🛠️ 动手实践
- 用 route 模式实现一个"中文/英文/代码问题"三路路由器,分别对比
determine_input_for_members开关前后的成员收到的输入差异。 - 给 19.4 的购物清单团队再加一个"烹饪建议成员",验证两个成员是否都能读到对方写入的清单条目。
- 打开
show_members_responses=True并用stream=True跑一次 coordinate 模式,记录事件序列(leader 委派→成员→汇总),写成一段观察笔记。
下一章我们把"确定性"请回来:当流程不该由模型临场发挥时,Workflow 是更好的编排工具。