跳到主要内容

第 4 章 · 团队(Team)与工作流(Workflow)

单个 agent 不够用时,Agno 给两种「把多个 agent 组合起来」的方式。它们的哲学截然不同:Team 让 LLM 自己决定派谁干(灵活、不确定);Workflow 让你用代码画死流程图(可控、确定)。这章讲清两者各适合什么。

4.1 两种组合的根本区别​

Team(团队)Workflow(工作流)
谁决定下一步领队模型自主决定派给哪个成员你的代码(步骤图)决定
确定性不确定(模型每次可能不同)确定(同输入同路径)
适合开放问题、需要动态分工固定多步流程(如「检索→起草→审校→发布」)
入口文件team/team.pyworkflow/workflow.py

4.2 Team:领队委派给成员​

Team(team/team.py:73)本质是一个「领队 agent」,它的成员(members)是别的 Agent 或 Team。领队不直接干活,而是通过一个内置工具把任务委派出去。

那个工具就是 delegate_task_to_member(team/_default_tools.py:595),由 _get_delegate_task_function 生成(team/_default_tools.py:441)。从领队模型的视角看,它就是个普通工具:

# 示意,非源码 —— 领队模型看到的委派工具长这样
def delegate_task_to_member(member_id: str, task: str):
    """把一个子任务交给指定成员去做,返回成员的结果。"""
    ...

于是「团队协作」复用了第 1/2 章的同一套机器:领队的 Model 循环里,模型「调用 delegate_task_to_member」→ 框架真的去跑那个成员 agent → 把成员的输出当工具结果喂回领队 → 领队继续。多成员并行时还有 delegate_task_to_members(复数,_default_tools.py:958)。

几个关键开关(team/team.py):

  • respond_directly(:108)—— 为 True 时成员的回答直接返回给用户,领队不再二次加工。
  • determine_input_for_members(:112,默认 True)—— 让领队为每个成员重新组织输入(而不是原样转发用户问题)。关掉则用固定 input(见 _get_delegate_task_function 的 input 参数,team/team.py:1508)。

委派流程图(领队的循环视角):

  用户问题 ──▶ 领队模型
                 │ 决定:调 delegate_task_to_member(member_id, task)
                 ▼
           跑成员 agent(成员自己也有完整 run 循环)
                 │ 成员输出
                 ▼
           作为「工具结果」喂回领队 ──▶ 领队继续
                 │ 可能再委派、或综合后给最终答案
                 ▼
              最终回答

任务管理工具: Team 还可挂一组任务管理工具(team/_task_tools.py:92 的 _get_task_management_tools),让领队 create_task / update_task_status / execute_task / execute_tasks_parallel(_task_tools.py:178、:403、:757)——这让团队能显式地拆任务、并行执行、汇总。

4.3 Workflow:确定性步骤图​

当流程是固定的(不想让模型每次自由发挥),用 Workflow(workflow/workflow.py)。它由一串步骤组成,每个步骤可以是一个 Agent、一个 Team,或一段函数。

步骤类型(workflow/ 各文件,workflow.py:156-161 的映射表):

步骤文件作用
Stepworkflow/step.py单步:跑一个 agent/team/函数
Stepsworkflow/steps.py顺序步骤序列
Loopworkflow/loop.py循环执行直到条件满足
Parallelworkflow/parallel.py并行多个步骤
Conditionworkflow/condition.py条件分支
Routerworkflow/router.py按规则路由到不同步骤

数据在步骤间靠 StepInput / StepOutput 传递(workflow/types.py,workflow/step.py:64)。一个 Step 的执行器签名是 [StepInput] -> StepOutput(及其 async/iterator 变体,workflow/step.py:89-96),所以前一步的输出自然成为后一步的输入。

概念演示:

# 示意,非源码 —— 一个「检索→起草→审校」的确定性工作流
workflow = Workflow(steps=[
    Step(name="research", agent=researcher),   # 第1步:研究员 agent
    Step(name="draft",    agent=writer),        # 第2步:写手 agent
    Condition(                                   # 第3步:质量不够就回炉
        evaluator=quality_check,
        steps=[Step(name="revise", agent=editor)],
    ),
])
workflow.run("写一篇关于 X 的文章")

Workflow 也支持暂停/审校: 步骤层有 StepRequirement / StepPauseResult 和 StepOutputReviewEvent(workflow/workflow.py:88、:128、:135),即工作流可以在某步停下让人审,与第 2 章的 HITL 一脉相承。

4.4 怎么选​

  • 任务边界清晰、步骤固定 → Workflow。可控、可复现、好调试。
  • 需要模型动态判断「该叫谁、拆成什么」 → Team。灵活,但每次路径可能不同。
  • 两者可嵌套:Workflow 的某个 Step 里可以放一个 Team;Team 的成员也能是另一个 Team。

4.5 巧妙之处​

  • 委派复用工具机制。 Team 没发明新协议——「派活」就是领队模型调一个名叫 delegate_task_to_member 的普通工具(_default_tools.py:595),整个第 1/2 章的循环原封不动复用。
  • 领队会「翻译」任务。 determine_input_for_members 默认让领队替成员重写输入(team/team.py:112),而不是把用户原话硬塞给成员——更贴合「领队理解全局、成员专注子任务」。
  • Workflow 步骤是统一的函数签名。 [StepInput] -> StepOutput(workflow/step.py:89)让 agent、team、纯函数能无缝串在一条流水线上。

4.6 代码地图​

主题文件符号
团队libs/agno/agno/team/team.pyTeam、respond_directly、determine_input_for_members
委派工具libs/agno/agno/team/_default_tools.py_get_delegate_task_function、delegate_task_to_member、delegate_task_to_members
任务管理libs/agno/agno/team/_task_tools.py_get_task_management_tools、execute_task、execute_tasks_parallel
工作流libs/agno/agno/workflow/workflow.pyWorkflow、WorkflowStep
步骤类型libs/agno/agno/workflow/Step、Steps、Loop、Parallel、Condition、Router
步骤数据libs/agno/agno/workflow/types.pyStepInput、StepOutput、StepRequirement