Agno : コンセプト : チーム – チームの実行

by

チームを実行しレスポンスを取得する方法を学習します。Team.run() 関数はチームを実行し、レスポンスを TeamRunResponse オブジェクトか TeamRunResponseEvent オブジェクトのストリームのいずれかとして生成します。

Agno : ユーザガイド : コンセプト : チーム – チームの実行

作成 : クラスキャット・セールスインフォメーション
作成日時 : 07/26/2025
バージョン : Agno 1.7.5

* 本記事は docs.agno.com の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています :

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

Agno ユーザガイド : コンセプト : チーム – チームの実行

チームを実行しレスポンスを取得する方法を学習します。

Team.run() 関数はチームを実行し、レスポンスを TeamRunResponse オブジェクトか TeamRunResponseEvent オブジェクトのストリームのいずれかとして生成します。

多くの例では、ターミナルに応答をプリントするヘルパーユーティリティである team.print_response() を使用しています。これは内部的には team.run() を使用しています。

チームを実行する方法が以下です。レスポンスは response と response_stream 変数で取得されます。

from agno.team import Team
from agno.models.openai import OpenAIChat

agent_1 = Agent(name="News Agent", role="Get the latest news")
agent_2 = Agent(name="Weather Agent", role="Get the weather for the next 7 days")
team = Team(name="News and Weather Team", mode="coordinate", members=[agent_1, agent_2])

response = team.run("What is the weather in Tokyo?")

# Synchronous execution
result = team.run("What is the weather in Tokyo?")

# Asynchronous execution
result = await team.arun("What is the weather in Tokyo?")

# Streaming responses
for chunk in team.run("What is the weather in Tokyo?", stream=True):
    print(chunk.content, end="", flush=True)

# Asynchronous streaming
async for chunk in await team.arun("What is the weather in Tokyo?", stream=True):
    print(chunk.content, end="", flush=True)

※ 訳註 : 非同期関数を実行するには、以下のようにする必要があります :

import asyncio
from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.agent import Agent

agent_1 = Agent(name="News Agent", role="Get the latest news")
agent_2 = Agent(name="Weather Agent", role="Get the weather for the next 7 days")
team = Team(name="News and Weather Team", mode="coordinate", members=[agent_1, agent_2])

# Synchronous execution
result = team.run("What is the weather in Tokyo?")

# Asynchronous execution
async def async_run():
    result = await team.arun("What is the weather in Tokyo?")

# Streaming responses
for chunk in team.run("What is the weather in Tokyo?", stream=True):
    print(chunk.content, end="", flush=True)

# Asynchronous streaming
async def async_streaming():
    async for chunk in await team.arun("What is the weather in Tokyo?", stream=True):
        print(chunk.content, end="", flush=True)

# 非同期関数を実行するには
asyncio.run(async_run())
asyncio.run(async_streaming())

 

中間ステップのストリーミング

チームの実行を通して複数のイベントが発生しますが、これらのイベントはチームの透過性の向上のためにリアルタイムに提供します。

stream_intermediate_steps=True を設定することで、中間ステップのストリーミングを有効にできます。

# Stream with intermediate steps
response_stream = team.run(
    "What is the weather in Tokyo?",
    stream=True,
    stream_intermediate_steps=True
)

 

イベントの処理

response stream に対して反復処理することで、イベントを到着したときに処理できます :

response_stream = team.run("Your prompt", stream=True, stream_intermediate_steps=True)

for event in response_stream:
    if event.event == "TeamRunResponseContent":
        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}")
    ...

 

イベントの保存

実行中に発生したすべてのイベントを RunResponse オブジェクトに保存することができます。

from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

team = Team(model=OpenAIChat(id="gpt-4o-mini"), members=[], store_events=True)

response = team.run("Tell me a 5 second short story about a lion", stream=True, stream_intermediate_steps=True)
pprint_run_response(response)

for event in agent.run_response.events:
    print(event.event)

デフォルトでは TeamRunResponseContentEvent と RunResponseContentEvent イベントは保存されません。events_to_skip パラメータを設定して、スキップするイベントを変更できます。

team = Team(model=OpenAIChat(id="gpt-4o-mini"), members=[], store_events=True, events_to_skip=[TeamRunEvent.run_started.value])

 

構造化入力

Team.run() や Team.print_response() に message パラメータとして渡すことで、チームに構造化入力 (i.e a pydantic モデル) を提供できます。

from typing import List

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from pydantic import BaseModel, Field


class ResearchTopic(BaseModel):
    """Structured research topic with specific requirements"""

    topic: str
    focus_areas: List[str] = Field(description="Specific areas to focus on")
    target_audience: str = Field(description="Who this research is for")
    sources_required: int = Field(description="Number of sources needed", default=5)


# Define agents
hackernews_agent = Agent(
    name="Hackernews Agent",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from Hackernews posts",
)

team = Team(
    name="Hackernews Team",
    model=OpenAIChat(id="gpt-4o-mini"),
    members=[hackernews_agent],
    mode="collaborate",
)

team.print_response(
    message=ResearchTopic(
        topic="AI",
        focus_areas=["AI", "Machine Learning"],
        target_audience="Developers",
        sources_required=5,
    )
)

 

以上