Skip to content

第 19 章 · Team 进阶:共享状态与成员通信

本章目标:掌握 Team 的四种协作模式切换、成员间共享 session state 的正确姿势,以及如何通过流式事件观察团队内部分工。

19.1 四种协作模式总览

上一章我们用 mode=TeamMode.coordinate 让 leader 分派任务。Agno 2.x 把协作风格显式化为四种模式(from agno.team.mode import TeamMode):

模式配置行为
Coordinate(默认)不设 mode 或设 coordinateLeader 选人、改写任务、汇总结果
Routemode=TeamMode.routeLeader 路由给一个成员,直接返回该成员的回复
Broadcastmode=TeamMode.broadcast同一任务发给所有成员
Tasksmode=TeamMode.tasksLeader 维护共享任务清单循环执行,直到完成

关于遗留参数

旧版本的 respond_directly=Truedelegate_to_all_members=True 仍然兼容(分别映射到 routebroadcast),但新代码一律使用 mode=——当两者同时出现时 mode 会覆盖遗留标志。

19.2 Route 模式:直达回复

客服分流、语言路由这类场景不需要 leader 再加工结果,用 route 模式延迟最低:

python
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 模式

广播适合"多视角并行评审",任务循环适合目标驱动的长流程:

python
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 与成员共用同一个数据库

python
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,运行时自动替换:

python
team = Team(
    members=[],
    session_state={"user_name": "小明"},
    instructions="用户的名字是 {user_name},请用名字称呼他。",
    markdown=True,
)

19.5 观察分工:流式事件与成员响应

调试团队的关键是看清"谁在干活"。三个手段配合使用:

python
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)   # 成员启动、工具调用、内容增量都有对应事件

配套的最佳实践还有三条:

  1. 给成员设置显式 id:委派选择和运行追踪都用 member id 定位,比靠 name 稳定;
  2. 模型继承:成员未设置 model 时自动继承 team 的模型——这让"成员用小模型、leader 用大模型"的分档只需一行省略;
  3. callable factoriesmembers= 可以传函数,按 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 + 流式事件 + 显式 member id 是观察团队分工的三板斧。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 在 Agno 2.x 中,想让「leader 只负责选人并把该成员的回复直接返回」,推荐写法是?

2. Route 模式下要把用户原始输入不做改写地传给成员,应该?

3. 让成员能读取并自主修改团队 session state 的最小配置是?

4. 关于 Tasks 模式,下列说法正确的是?

🛠️ 动手实践

  1. 用 route 模式实现一个"中文/英文/代码问题"三路路由器,分别对比 determine_input_for_members 开关前后的成员收到的输入差异。
  2. 给 19.4 的购物清单团队再加一个"烹饪建议成员",验证两个成员是否都能读到对方写入的清单条目。
  3. 打开 show_members_responses=True 并用 stream=True 跑一次 coordinate 模式,记录事件序列(leader 委派→成员→汇总),写成一段观察笔记。

下一章我们把"确定性"请回来:当流程不该由模型临场发挥时,Workflow 是更好的编排工具。