比LangGraph快5000倍還不夠？Agno Teams展示多智能體協(xié)作的終極形態(tài)！ 原創(chuàng)
本文針對于Teams概念進行深入研究。
概述
一個 Team（團隊） 是由多個智能體（或其他子團隊）組成的集合，它們協(xié)作完成任務。
下面是一個簡單示例：
from agno.team import Team
from agno.agent import Agent
team = Team(members=[
Agent(name="智能體 1", role="你用英文回答問題"),
Agent(name="智能體 2", role="你用中文回答問題"),
Team(name="團隊 1", members=[Agent(name="智能體 3", role="你用法語回答問題")], role="你協(xié)調團隊成員用法語回答問題"),
])團隊的領導者會根據成員的角色與任務性質，將任務分配給相應的成員。
與智能體類似，團隊也支持以下特性：
- 模型（Model）：可設置用于團隊領導者（team leader）的模型，用來決定如何將任務分配給團隊成員。
- 指令（Instructions）：可以對團隊領導者下達指令，指導其如何解決問題。 團隊成員的名稱、描述和角色會自動提供給團隊領導者。
- 工具（Tools）：如果團隊領導者需要直接使用工具，可以為團隊添加工具。
- 推理（Reasoning）：允許團隊領導者在作出回應或分配任務前進行“思考”，并在收到成員結果后進行“分析”。
- 知識（Knowledge）：如果團隊需要檢索信息，可以為團隊添加知識庫。知識庫由團隊領導者訪問。
- 存儲（Storage）：團隊的會話歷史和狀態(tài)會保存在數據庫中，使團隊可以從上次中斷處繼續(xù)對話，支持多輪、長期的交互。
- 記憶（Memory）：賦予團隊記憶能力，讓其能夠存儲并回憶先前交互中的信息，從而學習用戶偏好并個性化響應。
構建團隊（Building Teams）
要構建一個高效的團隊，應從簡單開始 —— 只包含模型（model）、成員（members）和指令（instructions）。 當基本功能正常后，再根據需要逐步增加復雜性。
以下是一個最簡單的帶有專職智能體的團隊示例：
# 文件名：news_weather_team.py
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
# 創(chuàng)建專職智能體
news_agent = Agent(
id="news-agent",
name="新聞智能體",
role="獲取最新新聞并提供摘要",
tools=[DuckDuckGoTools()]
)
weather_agent = Agent(
id="weather-agent",
name="天氣智能體",
role="獲取天氣信息和預報",
tools=[DuckDuckGoTools()]
)
# 創(chuàng)建團隊
team = Team(
name="新聞與天氣團隊",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o"),
instructions="與團隊成員協(xié)作，為用戶提供全面的信息。根據用戶請求分配任務。"
)
team.print_response("東京的最新新聞和天氣怎么樣？", stream=True)提示（Tip）建議為每個團隊成員明確指定 ?
?id???、??name??? 和 ??role??? 字段，以便團隊領導者更好地識別成員。 其中，??id?? 用于在團隊內部以及領導者上下文中標識該成員。注意（Note）當團隊成員未指定模型時，會從其父團隊繼承模型。 如果成員顯式指定了模型，則保留自身模型。 在嵌套團隊中，成員從其直接父級團隊繼承模型。 若團隊未指定模型，則默認使用 OpenAI 的 ?
?gpt-4o??。該繼承規(guī)則適用于以下字段：?
?model???、??reasoning_model???、??parser_model???、??output_model??。
運行團隊（Run your Team）
運行團隊時，可以使用 ??Team.print_response()?? 方法在終端中打印響應：
team.print_response("東京的最新新聞和天氣怎么樣？")此方法僅適用于開發(fā)階段，不推薦在生產環(huán)境中使用。 在生產環(huán)境中，請使用 ??Team.run()??? 或異步版本 ??Team.arun()??。例如：
from typing import Iterator
from agno.team import Team
from agno.agent import Agent
from agno.run.team import TeamRunOutputEvent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response
news_agent = Agent(name="新聞智能體", role="獲取最新新聞")
weather_agent = Agent(name="天氣智能體", role="獲取未來7天的天氣")
team = Team(
name="新聞與天氣團隊",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o")
)
# 運行團隊并返回響應變量
response = team.run("東京的天氣怎么樣？")
# 打印響應內容
print(response.content)
################ 流式響應（STREAM RESPONSE） #################
stream: Iterator[TeamRunOutputEvent] = team.run("東京的天氣怎么樣？", stream=True)
for chunk in stream:
if chunk.event == "TeamRunContent":
print(chunk.content)
################ 流式響應 + 美化打印（STREAM AND PRETTY PRINT） #################
stream: Iterator[TeamRunOutputEvent] = team.run("東京的天氣怎么樣？", stream=True)
pprint_run_response(stream, markdown=True)修改終端顯示內容
使用 ??print_response??? 方法時，默認只打印團隊中涉及工具調用的部分（通常是任務分配信息）。 如果希望同時打印各個成員（智能體）的響應內容，可以設置參數 ??show_members_responses=True??：
team.print_response("東京的天氣怎么樣？", show_members_responses=True)運行團隊（Running Teams）
可以通過調用 ??Team.run()??? 或 ??Team.arun()?? 來運行團隊。其工作流程如下：
- 團隊領導者構建要發(fā)送給模型的上下文（包括系統(tǒng)消息、用戶消息、對話歷史、用戶記憶、會話狀態(tài)及其他相關輸入）。
- 團隊領導者將該上下文發(fā)送給模型。
- 模型處理輸入，并決定是使用?
?delegate_task_to_members?? 工具將任務委派給團隊成員、調用其他工具，還是直接生成響應。 - 如果發(fā)生了任務委派，團隊成員會執(zhí)行各自的任務，并將結果返回給團隊領導者。
- 團隊領導者處理更新后的上下文，并生成最終響應。
- 團隊將該最終響應返回給調用方。
基本執(zhí)行（Basic Execution）
??Team.run()??? 函數運行團隊并返回輸出結果 —— 可以是一個 ??TeamRunOutput??? 對象， 也可以在啟用 ??stream=True??? 時，返回一個由 ??TeamRunOutputEvent???（以及成員智能體的 ??RunOutputEvent??）組成的流。
示例如下：
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response
news_agent = Agent(
name="新聞智能體",
model=OpenAIChat(id="gpt-4o"),
role="獲取最新新聞",
tools=[DuckDuckGoTools()]
)
weather_agent = Agent(
name="天氣智能體",
model=OpenAIChat(id="gpt-4o"),
role="獲取未來7天的天氣",
tools=[DuckDuckGoTools()]
)
team = Team(
name="新聞與天氣團隊",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o")
)
# 運行團隊并返回響應
response = team.run(input="東京的天氣怎么樣？")
# 以 Markdown 格式打印響應
pprint_run_response(response, markdown=True)提示：你也可以使用 ?
?Team.arun()?? 異步運行團隊。 當團隊領導者在一次請求中將任務委派給多個成員時，成員會并發(fā)執(zhí)行任務。
運行輸出（Run Output）
當未啟用流式（stream）模式時，??Team.run()??? 函數會返回一個 ??TeamRunOutput?? 對象。 該對象的核心屬性包括：
- ?
?run_id??：本次運行的唯一 ID。 - ?
?team_id??：團隊 ID。 - ?
?team_name??：團隊名稱。 - ?
?session_id??：會話 ID。 - ?
?user_id??：用戶 ID。 - ?
?content??：最終響應內容。 - ?
?content_type??：內容類型（若為結構化輸出，則為對應 Pydantic 模型的類名）。 - ?
?reasoning_content??：推理內容。 - ?
?messages??：發(fā)送給模型的消息列表。 - ?
?metrics??：本次運行的指標。 - ?
?model??：本次運行所使用的模型。 - ?
?member_responses???：團隊成員的響應列表（若??store_member_responses=True?? 時可用）。
注意：未指定模型的團隊成員會繼承其父團隊的模型。 這適用于：?
?model???、??reasoning_model???、??parser_model???、??output_model??。
流式運行（Streaming）
設置 ??stream=True??? 可啟用流式模式。此時，??run()??? 將返回一個 ??TeamRunOutputEvent?? 對象的迭代器，而非單一響應。
from typing import Iterator
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
news_agent = Agent(name="新聞智能體", role="獲取最新新聞")
weather_agent = Agent(name="天氣智能體", role="獲取未來7天的天氣")
team = Team(
name="新聞與天氣團隊",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o")
)
# 以流式方式運行團隊
stream: Iterator[TeamRunOutputEvent] = team.run("東京的天氣怎么樣？", stream=True)
for chunk in stream:
if chunk.event == "TeamRunContent":
print(chunk.content)提示：當使用 ?
?arun()?? 異步運行團隊時，如果團隊領導者將任務分派給多個成員，這些成員會并發(fā)執(zhí)行。 這意味著事件會并行產生，事件順序不一定有序。
流式所有事件（Streaming All Events）
默認情況下，流式輸出僅包含 ??RunContent??? 類型事件。 若要流式傳輸團隊內部所有事件，可設置 ??stream_events=True??：
# 啟用全部事件流式輸出
response_stream = team.run(
"東京的天氣怎么樣？",
stream=True,
stream_events=True
)這將實時輸出團隊的內部進程，如工具調用（tool call）或推理步驟（reasoning）。
處理事件（Handling Events）
你可以通過迭代響應流，逐個處理到達的事件：
response_stream = team.run("你的提示詞", stream=True, stream_events=True)
for event in response_stream:
if event.event == "TeamRunContent":
print(f"內容: {event.content}")
elif event.event == "TeamToolCallStarted":
print(f"開始調用工具: {event.tool}")
elif event.event == "ToolCallStarted":
print(f"成員開始調用工具: {event.tool}")
elif event.event == "ToolCallCompleted":
print(f"成員完成調用工具: {event.tool}")
elif event.event == "TeamReasoningStep":
print(f"推理步驟: {event.content}")
...注意：團隊成員事件會在團隊執(zhí)行期間產生。 若不希望接收這些事件，可設置 ?
?stream_member_events=False??。
存儲事件（Storing Events）
你可以在 ??RunOutput?? 對象中保存運行期間產生的所有事件。
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response
team = Team(
name="故事團隊",
members=[],
model=OpenAIChat(id="gpt-4o"),
store_events=True
)
response = team.run("講一個5秒鐘的關于獅子的短故事", stream=True, stream_events=True)
pprint_run_response(response)
for event in response.events:
print(event.event)默認情況下，??TeamRunContentEvent??? 和 ??RunContentEvent??? 不會被存儲。 你可以通過設置 ??events_to_skip?? 參數修改跳過的事件類型。例如：
team = Team(
name="故事團隊",
members=[],
model=OpenAIChat(id="gpt-4o"),
store_events=True,
events_to_skip=["TeamRunStarted"]
)事件類型（Event Types）
以下是 ??Team.run()??? 與 ??Team.arun()?? 根據配置可能產生的事件類型：
核心事件（Core Events）
事件類型 | 描述 |
? | 表示運行開始 |
? | 包含模型響應的文本塊 |
? | 表示內容流式傳輸結束 |
? | 包含模型的中間響應（當啟用 ? |
? | 表示運行成功完成 |
? | 表示運行過程中發(fā)生錯誤 |
? | 表示運行被取消 |
工具事件（Tool Events）
事件類型 | 描述 |
? | 團隊工具調用開始 |
? | 團隊工具調用完成（包含結果） |
推理事件（Reasoning Events）
事件類型 | 描述 |
? | 推理開始 |
? | 單個推理步驟 |
? | 推理完成 |
記憶事件（Memory Events）
事件類型 | 描述 |
? | 團隊記憶更新開始 |
? | 團隊記憶更新完成 |
會話摘要事件（Session Summary Events）
事件類型 | 描述 |
? | 會話摘要生成開始 |
? | 會話摘要生成完成 |
前置鉤子事件（Pre-Hook Events）
事件類型 | 描述 |
? | 前置鉤子開始執(zhí)行 |
? | 前置鉤子執(zhí)行完成 |
后置鉤子事件（Post-Hook Events）
事件類型 | 描述 |
? | 后置鉤子開始執(zhí)行 |
? | 后置鉤子執(zhí)行完成 |
解析模型事件（Parser Model Events）
事件類型 | 描述 |
? | 解析模型響應開始 |
? | 解析模型響應完成 |
輸出模型事件（Output Model Events）
事件類型 | 描述 |
? | 輸出模型響應開始 |
? | 輸出模型響應完成 |
自定義事件（Custom Events）
如果你編寫了自定義工具（custom tools），你可以定義并發(fā)送自定義事件。 這些事件會與 Agno 內置事件一同被處理。
可以通過繼承內置的 ??CustomEvent?? 類來自定義事件類型，例如：
from dataclasses import dataclass
from agno.run.team import CustomEvent
@dataclass
class CustomerProfileEvent(CustomEvent):
"""客戶資料的自定義事件"""
customer_name: Optional[str] = None
customer_email: Optional[str] = None
customer_phone: Optional[str] = None然后可以在自定義工具中產出該事件：
from agno.tools import tool
@tool()
async def get_customer_profile():
"""示例工具，生成自定義事件"""
yield CustomerProfileEvent(
customer_name="John Doe",
customer_email="john.doe@example.com",
customer_phone="1234567890",
)指定運行用戶與會話（Specify Run User and Session）
你可以通過 ??user_id??? 和 ??session_id?? 參數指定運行所屬的用戶和會話：
team.run("生成我的月度報告", user_id="john@example.com", session_id="session_123")傳入圖片 / 音頻 / 視頻 / 文件（Passing Images / Audio / Video / Files）
你可以通過 ??images???、??audio???、??video??? 或 ??files?? 參數向團隊傳入多模態(tài)內容。例如：
team.run("根據這張圖片講一個5秒鐘的短故事", images=[Image(url="https://example.com/image.jpg")])取消運行（Cancelling a Run）
可以通過調用 ??Team.cancel_run()?? 方法取消正在執(zhí)行的運行。
團隊成員指派（Team Member Delegation）
基本概念：團隊執(zhí)行流程
一個 ??Team??? 在內部其實有一個“團隊領導者”（team leader）agent。 當你調用 ??team.run()??? 或 ??team.arun()?? 時，這個領導者會分析用戶輸入并決定：
- 是否需要將任務分配（delegate）給團隊成員；
- 分配給誰；
- 或者是否自己直接回答。
整個執(zhí)行流程如下：
- 團隊接收到用戶輸入；
- 團隊領導者（Team Leader）分析輸入并拆解為子任務；
- 領導者將子任務分配給合適的成員；
- 各成員執(zhí)行任務并返回結果；
- 領導者可能再次委派任務，或綜合所有成員的結果；
- 最終返回完整的響應給用戶。
內部實現上，這個“分配任務”的動作是通過一個特殊的工具：?
?delegate_task_to_members?? 工具來完成的。
當使用異步執(zhí)行（??arun()??）時，如果領導者同時把任務分配給多個成員，這些成員將并發(fā)執(zhí)行。
不同的團隊執(zhí)行策略
Agno 提供了幾種常見的執(zhí)行模式，用來控制任務分配與響應方式。
1. 成員直接響應（respond_directly=True）
默認情況下，成員的回答會被領導者處理、總結，然后返回統(tǒng)一結果。 但如果你希望成員的回答直接返回給用戶（領導者不加工），就設置：
Team(respond_directly=True)這種模式下，團隊更像一個 “路由器（Router）”，負責把問題轉發(fā)給最合適的 agent。
例子：多語言團隊
multi_language_team = Team(
name="Multi Language Team",
respond_directly=True,
members=[english_agent, japanese_agent],
instructions=[
"You are a language router that directs questions to the appropriate language agent.",
],
)當 ??respond_directly=True?? 時：
- 團隊領導者不會再整合結果；
- 多個成員的回答會直接拼接返回；
- 不可與?
?delegate_task_to_all_members?? 同時使用。
2. 直接把用戶輸入傳給成員（determine_input_for_members=False）
默認情況下，領導者會重新組織輸入，決定給成員什么任務。 但如果你希望成員直接收到用戶的原始輸入，可以設置：
Team(determine_input_for_members=False)這對于成員輸入結構化（如 Pydantic 模型）特別有用。
例子：研究團隊
team = Team(
name="Hackernews Research Team",
determine_input_for_members=False,
members=[hackernews_agent],
)場景：當團隊接收結構化任務（如 JSON 或表單輸入）時，希望成員直接處理原始數據。
3. 任務同時分配給所有成員（delegate_task_to_all_members=True）
默認團隊領導者會選擇一個成員處理任務。 若你希望所有成員并行執(zhí)行相同任務，可以啟用：
Team(delegate_task_to_all_members=True)這會讓所有成員同時接收相同的輸入，執(zhí)行各自的分析，然后團隊領導者再整合結果。 在異步模式（??arun()??）下，所有成員將并發(fā)運行。
例子：多來源信息研究團隊
agent_team = Team(
name="Discussion Team",
delegate_task_to_all_members=True,
members=[
reddit_researcher,
hackernews_researcher,
academic_paper_researcher,
twitter_researcher,
],
)這種模式常用于：
- 信息聚合（多渠道新聞/研究源）；
- 共創(chuàng)（多 agent 協(xié)作生成結果）；
- 決策（各成員提出意見再綜合）。
三種模式的區(qū)別總結
模式 | 關鍵參數 | 行為特點 | 常用場景 |
成員直接響應 | ? | 領導者不加工結果，直接返回成員回答 | 路由團隊（如多語言、專業(yè)分類） |
輸入直傳成員 | ? | 用戶輸入直接送成員 | 結構化輸入、科研/開發(fā)任務 |
同時分配任務 | ? | 所有成員同時執(zhí)行任務 | 協(xié)作、共識形成、聚合分析 |
調試團隊（Debugging Teams）
為什么要用 Debug 模式？
Agno 的團隊（Team）設計允許多個智能體（Agent）協(xié)作，因此在開發(fā)和測試階段：
- 你需要知道每一步是誰執(zhí)行的；
- 模型之間傳了哪些消息；
- 是否正確調用了工具；
- token 消耗、執(zhí)行時間等指標如何。
Debug 模式能讓你 可視化整個團隊的執(zhí)行過程，包括中間步驟與錯誤信息，非常適合在調試或優(yōu)化階段使用。
Debug 模式的三種啟用方式
你可以通過三種方式開啟 debug 模式：
方式 | 說明 | 作用范圍 |
? （在 Team 實例化時） | 在創(chuàng)建團隊時啟用 | 全局生效（包括團隊成員） |
? （在運行時） | 在調用 ? | 只對當前運行有效 |
環(huán)境變量 ? | 在系統(tǒng)層面開啟 | 所有 Team 默認進入調試模式 |
示例：啟用 Debug 模式
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
# 定義兩個簡單的 Agent
news_agent = Agent(name="News Agent", role="Get the latest news")
weather_agent = Agent(name="Weather Agent", role="Get the weather for the next 7 days")
# 創(chuàng)建一個團隊
team = Team(
name="News and Weather Team",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o"),
debug_mode=True, # 啟用調試模式
# debug_level=2, # 可選，提供更詳細日志
)
# 調試運行
team.print_response("What is the weather in Tokyo?")Debug 輸出內容包括：
- 模型接收到的 prompt 與生成的響應；
- 調用了哪些工具（例如搜索、數據庫、API）；
- 任務分配給了哪個成員；
- token 用量、執(zhí)行時長；
- 報錯棧與上下文。
如果你設置 ?
?debug_level=2??，日志會更加詳細，包括每個子任務的上下文與中間輸出。
Debug 日志等級說明
參數值 | 說明 |
? | 默認，顯示關鍵步驟（輸入、輸出、調用鏈） |
? | 顯示所有細節(jié)（包括模型消息、內部調用、token 消耗等） |
在開發(fā)復雜團隊時建議開啟 ??debug_level=2??，能幫助你：
- 定位消息在何處被修改；
- 追蹤工具調用失敗的原因；
- 優(yōu)化 prompt 和上下文結構。
交互式 CLI 模式（Interactive CLI）
Agno 還提供了一個命令行交互式界面，可以讓你像聊天一樣與團隊實時對話，非常適合測試多輪會話和團隊協(xié)調效果。
from agno.team import Team
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
news_agent = Agent(name="News Agent", role="Get the latest news")
weather_agent = Agent(name="Weather Agent", role="Get the weather for the next 7 days")
team = Team(
name="News and Weather Team",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o"),
db=SqliteDb(db_file="tmp/data.db"), # 保存歷史記錄
add_history_to_context=True, # 將歷史對話加到上下文中
num_history_runs=3, # 保存最近3次記錄
)
# 啟動命令行交互模式
team.cli_app(stream=True)運行后你會看到一個交互式命令行界面，比如：
> Hello team!
[News Agent]: Here's the latest news...
[Weather Agent]: The forecast for Tokyo is sunny...支持：
- 多輪對話；
- 流式輸出；
- 歷史追蹤；
- 調試中實時查看任務流轉。
Teams其他的用法跟單個Agent用法類似，詳情可參考官方文檔。
本文轉載自??AI 博物院?? 作者：longyunfeigu
