Agno : コンセプト : エージェント – エージェントの実行

by

エージェントを実行してレスポンスを取得する方法を学習します。Agent.run() 関数はエージェントを実行してレスポンスを RunResponse オブジェクトまたは RunResponse オブジェクトのストリームとして生成します。

Agno : ユーザガイド : コンセプト : エージェント – エージェントの実行

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

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

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

 

 

Agno ユーザガイド : コンセプト : エージェント – エージェントの実行

エージェントを実行してレスポンスを取得する方法を学習します。

Agent.run() 関数はエージェントを実行してレスポンスを RunResponse オブジェクトまたは RunResponse オブジェクトのストリームとして生成します。

多くの例では agent.print_response() を使用しています、これはレスポンスをターミナルで出力するヘルパーユーティリティです。それは内部的には agent.run() を使用しています。

 

エージェントの実行

エージェントの実行方法は以下の通りです。レスポンス (応答) は response でキャプチャされます。

from typing import Iterator
from agno.agent import Agent, RunResponse
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))

# Run agent and return the response as a variable
response: RunResponse = agent.run("Tell me a 5 second short story about a robot")

# Print the response in markdown format
pprint_run_response(response, markdown=True)

 

RunResponse

Agent.run() 関数は、ストリーミングでないときには RunResponse オブジェクトを返します。それは以下の属性を持ちます :

Note : メトリクスの理解 – メトリクスが収集され使用される方法の詳細説明については、メトリクス・ドキュメント を参照してください。

See detailed documentation in the RunResponse documentation.

 

ストリーミング・レスポンス

ストリーミングを有効にするには、run() を呼び出す際に stream=True を設定します。これは単一のレスポンスではなく、RunResponseEvent オブジェクトのイテレータを返します。

Note : From agno version 1.6.0, the Agent.run() function returns an iterator of RunResponseEvent, not of RunResponse objects.

from typing import Iterator
from agno.agent import Agent, RunResponseEvent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))

# Run agent and return the response as a stream
response_stream: Iterator[RunResponseEvent] = agent.run(
    "Tell me a 5 second short story about a lion",
    stream=True
)

# Print the response stream in markdown format
pprint_run_response(response_stream, markdown=True)

 

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

より詳細なストリーミングについては、stream_intermediate_steps=True を設定することで中間ステップを有効にできます。これはエージェントの中間プロセスについてリアルタイムの更新情報を提供します。

# Stream with intermediate steps
response_stream: Iterator[RunResponseEvent] = agent.run(
    "Tell me a 5 second short story about a lion",
    stream=True,
    stream_intermediate_steps=True
)

 

イベント処理

response stream を反復処理することでイベントが到着した時点で処理できます :

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

for event in response_stream:
    if event.event == "RunResponseContent":
        print(f"Content: {event.content}")
    elif event.event == "ToolCallStarted":
        print(f"Tool call started: {event.tool}")
    elif event.event == "ReasoningStep":
        print(f"Reasoning step: {event.content}")
    ...

You can see this behavior in action in our Playground.

 

イベントの保存

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

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"), store_events=True)

response = agent.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)

RunStarted
RunCompleted

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

例えば :

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"), store_events=True, events_to_skip=[RunEvent.run_started.value])

 

以上