Agno チームの実行方法を学習します。Team.run() または Team.arun() を呼び出すことでチームを実行します。
Agno 2.x : Learn : チーム – チームの実行
作成 : クラスキャット・セールスインフォメーション
作成日時 : 11/05/2025
バージョン : Agno 2.2.6
* 本記事は docs.agno.com の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています。スニペットはできる限り日本語を使用しています :
* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。
◆ お問合せ : 下記までお願いします。
- クラスキャット セールス・インフォメーション
- sales-info@classcat.com
- ClassCatJP
Agno 2.x : Learn : チーム – チームの実行
Agno チームの実行方法を学習します。
Team.run() または Team.arun() を呼び出すことでチームを実行します。その動作の仕組みは以下のようなものです :
- チームリーダーはモデルに送信するコンテキストを構築します (システムメッセージ、ユーザメッセージ、チャット履歴、ユーザメモリ、セッション状態、そしてその他の関連入力)
- チームリーダーはこのコンテキストをモデルに送信します。
- モデルは入力を処理して delegate_task_to_members ツールを使用してチームメンバーに委任するか、他のツールを呼び出すか、あるいは直接応答するかを決定します。
- 委任が発生する場合、チームメンバーはそれらのタスクを実行して結果をチームリーダーに返します。
- チームリーダーは更新されたコンテキストを処理して最終的なレスポンスを提供します。
- チームリーダーは最終的なレスポンスを呼び出し元に返します。
基本的な実行
Team.run() 関数はチームを実行して出力を – TeamRunOutput オブジェクトとしてか、(stream=True の場合) TeamRunOutputEvent 及び RunOutputEvent (メンバーエージェント用) オブジェクトのストリームとして返します。例えば :
Python
from agno.team import Team
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response
from agno.tools.duckduckgo import DuckDuckGoTools
news_agent = Agent(
name="News Agent",
model=OpenAIChat(id="gpt-4o"),
role="最新ニュースを取得する。",
tools=[DuckDuckGoTools()]
)
weather_agent = Agent(
name="Weather Agent",
model=OpenAIChat(id="gpt-4o"),
role="今後7日間の天気予報を取得する。",
tools=[DuckDuckGoTools()]
)
team = Team(
name="News and Weather Team",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o")
)
# Run team and return the response as a variable
response = team.run(input="東京の天気はどうですか？")
# Print the response in markdown format
pprint_run_response(response, markdown=True)
出力例
╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ 東京の現在の天気に関する詳細な情報は、以下のリンクからご確認いただけます。 │ │ │ │ 1 AccuWeatherの天気予報: 現在の天気や空気の状態を含む詳細な情報を提供しています。 │ │ 2 timeanddate.comの14日間の天気予報: 最大14日間の予報を含めた詳細情報が確認できます。 │ │ 3 The Weather Channelの10日間の予測: 高温や低温、降水確率などを含む10日間の予報を確認できます。 │ │ │ │ これらのリンクをクリックして、現在の気温、天気の種類、降水確率をご確認いただけます。 │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
Run Output
Team.run() 関数は、ストリーミングでない場合には、TeamRunOutput を返します。コア属性の一部は以下のようなのです :
- run_id: 実行 id
- team_id: チーム id
- team_name: チーム名
- session_id: セッション id
- user_id: ユーザ id
- content: レスポンス・コンテンツ
- content_type: コンテンツの種類。構造化出力の場合、これは pydantic モデルのクラス名になります。
- messages: モデルに送信されたメッセージのリスト。
- metrics: 実行メトリクス。詳細は メトリクス 参照。
- model: 実行に使用されたモデル。
- member_responses: メンバーのレスポンスのリスト。Team で store_member_responses=True の場合にオプションで追加されます。
See detailed documentation in the TeamRunOutput documentation.
ストリーミング
ストリーミングを有効にするには、run() を呼び出す際に stream=True を設定します。これは、単一のレスポンスではなく、TeamRunOutputEvent のイテレータを返します。
from typing import Iterator
from agno.team import Team, TeamRunOutputEvent
from agno.agent import Agent
from agno.models.openai import OpenAIChat
news_agent = Agent(name="News Agent", role="最新ニュースを取得する")
weather_agent = Agent(name="Weather Agent", role="今後7日間の天気予報を取得する")
team = Team(
name="News and Weather Team",
members=[news_agent, weather_agent],
model=OpenAIChat(id="gpt-4o")
)
# Run team and return the response as a stream
stream: Iterator[TeamRunOutputEvent] = team.run("東京の天気はどうですか？", stream=True)
for chunk in stream:
if chunk.event == "TeamRunContent":
print(chunk.content)
出力例
INFO Agent 'News Agent' inheriting model from Team: gpt-4o INFO Agent 'Weather Agent' inheriting model from Team: gpt-4o 現在 、 リア ル タイ ム の 天 気 予 報 を 提供 する こと は でき ません が 、 東京 の 今 後 の 天 気 を 知 り たい 場合 は 、 以下 の 方法 を 試 す こと を お 勧 め します ： 1 . ** 天 気 予 報 ウェ ブ サイト ** ： - 日本 気 象 協 会 （ ten ki .jp ） - ウ ェ ザ ーニ ュ ース - Yahoo ! 天 気 2 . ** スマ ート フォン の 天 気 ア プリ ** ： - Apple の 天 気 ア プリ - Google 天 気 - その他 、お 使 い の 携 帯 電話 に プリ イン スト ール され て いる 天 気 ア プリ これ ら の リ ソ ース で 最新 の 天 気 情報 が 確認 できます 。
すべてのイベントのストリーミング
デフォルトでは、レスポンスをストリーミングするとき、RunContent イベントだけがストリーミングされます。
stream_events=True を設定することで、すべての実行イベントをストリーミングすることもできます。
これは、ツール呼び出しや推論のような、チームの内部プロセスについてリアルタイムの更新を提供できます :
# Stream all events
response_stream = team.run(
"What is the weather in Tokyo?",
stream=True,
stream_events=True
)
イベント処理
レスポンス・ストリームに対して反復処理することで、イベントを受信するたびにそれらを処理できます :
response_stream = team.run("Your prompt", stream=True, stream_events=True)
for event in response_stream:
if event.event == "TeamRunContent":
print(f"Content: {event.content}")
elif event.event == "TeamToolCallStarted":
print(f"Tool call started: {event.tool}")
elif event.event == "ToolCallStarted":
print(f"Member tool call started: {event.tool}")
elif event.event == "ToolCallCompleted":
print(f"Member tool call completed: {event.tool}")
elif event.event == "TeamReasoningStep":
print(f"Reasoning step: {event.content}")
...
イベント・タイプ
以下のイベントは、チームの設定に応じて、Team.run() と Team.arun() 関数により送信されます :
コア・イベント
- TeamRunStarted – 実行の開始を示します
- TeamRunContent – モデルの応答テキストを個々のチャンクとして含みます
- TeamRunContentCompleted – コンテンツ・ストリーミングの完了シグナル
- TeamRunIntermediateContent – モデルの中間的な応答テキストを個々のチャンクとして含みます。これは output_model が設定されている場合に使用されます。
- TeamRunCompleted – 実行の正常完了シグナル
- TeamRunError – 実行中にエラー発生
- TeamRunCancelled – 実行がキャンセルされたシグナル
実行ユーザとセッションの指定
user_id と session_id パラメータを渡すことで、チームを実行する際に使用するユーザとセッションを指定できます。これは現在の実行が正しいユーザとセッションに関連付けられることを保証します。例えば :
team.run("Get me my monthly report", user_id="john@example.com", session_id="session_123")
For more information see the Team Sessions documentation.
画像 / 音声 / 動画 / ファイルを渡す
images, audio, video, or files パラメータを渡すことで、画像、音声、動画やファイルをチームに渡すことができます。例えば :
team.run("Tell me a 5 second short story about this image", images=[Image(url="https://example.com/image.jpg")])
For more information see the Multimodal documentation.
実行のキャンセル
Team.cancel_run() メソッドの呼び出しで実行をキャンセルできます。
See more details in the Cancelling a Run documentation.
以上