Agno : コンセプト : ツール – 非同期ツール

Agno エージェントは複数のツールを同時に実行して、モデルが行なう関数呼び出しを効率的に処理できるようにします。これは関数が時間がかかる演算を含む場合に特に価値があります。応答性を向上させ、全体の実行時間を短縮します。

Agno : ユーザガイド : コンセプト : ツール – 非同期ツール

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

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

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

 

クラスキャット 人工知能 研究開発支援サービス ⭐️ リニューアルしました 😉

クラスキャット は人工知能に関する各種サービスを提供しています。お気軽にご相談ください :

  • 人工知能導入個別相談会(無償)実施中! [詳細]

  • 人工知能研究開発支援 [詳細]
    1. 自社特有情報を含むチャットボット構築支援
    2. 画像認識 (医療系含む) / 画像生成

  • PoC(概念実証)を失敗させないための支援 [詳細]

お問合せ : 下記までお願いします。

  • クラスキャット セールス・インフォメーション
  • sales-info@classcat.com
  • ClassCatJP

 

 

Agno ユーザガイド : コンセプト : ツール – 非同期ツール

Agno エージェントは複数のツールを同時に実行して、モデルが行なう関数呼び出しを効率的に処理できるようにします。これは関数が時間がかかる演算を含む場合に特に価値があります。応答性を向上させ、全体の実行時間を短縮します。

以下に例を示します :

async_tools.py

import asyncio
import time

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.log import logger

async def atask1(delay: int):
    """Simulate a task that takes a random amount of time to complete
    Args:
        delay (int): The amount of time to delay the task
    """
    logger.info("Task 1 has started")
    for _ in range(delay):
        await asyncio.sleep(1)
        logger.info("Task 1 has slept for 1s")
    logger.info("Task 1 has completed")
    return f"Task 1 completed in {delay:.2f}s"


async def atask2(delay: int):
    """Simulate a task that takes a random amount of time to complete
    Args:
        delay (int): The amount of time to delay the task
    """
    logger.info("Task 2 has started")
    for _ in range(delay):
        await asyncio.sleep(1)
        logger.info("Task 2 has slept for 1s")
    logger.info("Task 2 has completed")
    return f"Task 2 completed in {delay:.2f}s"


async def atask3(delay: int):
    """Simulate a task that takes a random amount of time to complete
    Args:
        delay (int): The amount of time to delay the task
    """
    logger.info("Task 3 has started")
    for _ in range(delay):
        await asyncio.sleep(1)
        logger.info("Task 3 has slept for 1s")
    logger.info("Task 3 has completed")
    return f"Task 3 completed in {delay:.2f}s"


async_agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[atask2, atask1, atask3],
    show_tool_calls=True,
    markdown=True,
)

asyncio.run(
    async_agent.aprint_response("Please run all tasks with a delay of 3s", stream=True)
)

 
How to use :

  1. エージェントにツールのリストを供給します、可能なら最適なパフォーマンスのために非同期が望ましいです。ただし、同期関数も使用できます、それらは別々のスレッドで同時に実行されるからです。

  2. arun か aprint_response メソッドのいずれかを使用してエージェントを実行し、ツール呼び出しの同時実行を可能にします。

この例では、gpt-4o は atask1, atask2 と atask3 への 3 つのツール呼び出しを同時に行います。通常はこれらのツール呼び出しは逐次実行されますが、aprint_response 関数を使用して、それらは同時に実行され実行時間を向上させます。

 

以上