ツール呼び出し後に異なる指示のセットでモデルを「リトライ」したりエージェントを停止する必要がある場合、以下の例外の一つを発生させることができます。

Agno : ユーザガイド : コンセプト : ツール – 例外, フック

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

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

• User Guide : Concepts : Tools – Exceptions
• User Guide : Concepts : Tools – Hooks

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

 

 

 

Agno ユーザガイド : コンセプト : ツール – 例外

ツール呼び出し後に異なる指示のセットでモデルを「リトライ」したりエージェントを停止する必要がある場合、以下の例外の一つを発生させることができます :

• RetryAgentRun : 異なる指示のセットでエージェント実行をリトライしたい場合にこの例外を使用します。

• StopAgentRun : エージェントの実行を停止したい場合、この例外を使用します。

• AgentRunException : ツール呼び出しをリトライするために使用できる、一般的な例外です。

この例は、RetryAgentRun 例外を使用して追加の指示とともにエージェントをリトライする方法を示します。

retry_in_tool_call.py

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


def add_item(agent: Agent, item: str) -> str:
    """Add an item to the shopping list."""
    agent.session_state["shopping_list"].append(item)
    len_shopping_list = len(agent.session_state["shopping_list"])
    if len_shopping_list < 3:
        raise RetryAgentRun(
            f"Shopping list is: {agent.session_state['shopping_list']}. Minimum 3 items in the shopping list. "
            + f"Add {3 - len_shopping_list} more items.",
        )

    logger.info(f"The shopping list is now: {agent.session_state.get('shopping_list')}")
    return f"The shopping list is now: {agent.session_state.get('shopping_list')}"


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # Initialize the session state with empty shopping list
    session_state={"shopping_list": []},
    tools=[add_item],
    markdown=True,
)
agent.print_response("Add milk", stream=True)
print(f"Final session state: {agent.session_state}")

Note : Make sure to set AGNO_DEBUG=True to see the debug logs.

 

Agno ユーザガイド : コンセプト : ツール - フック

ツールフックを使用してツールの動作を変更する方法を学習します。

 

ツールフック

ツールフックを使用して、ツールが呼び出される前後に検証、ログ記録やその他のロジックを実行することができます。

ツールフックは、関数名、関数呼び出しと引数を受け取る関数です。オプションで、Agent や Team オブジェクトにもアクセスできます。ツールフック内では、関数呼び出しを呼び出して、結果を返す必要があります。
​
Note : It is important to use exact parameter names when defining a tool hook. agent, team, function_name, function_call, and arguments are available parameters.

例えば :

def logger_hook(
    function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
    """Log the duration of the function call"""
    start_time = time.time()

    # Call the function
    result = function_call(**arguments)
    
    end_time = time.time()
    duration = end_time - start_time
    
    logger.info(f"Function {function_name} took {duration:.2f} seconds to execute")

    # Return the result
    return result

または

def confirmation_hook(
    function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
    """Confirm the function call"""
    if function_name != "get_top_hackernews_stories":
        raise ValueError("This tool is not allowed to be called")
    return function_call(**arguments)

ツールフック内で Agent や Team オブジェクトにアクセスすることもできます。

def grab_customer_profile_hook(
    agent: Agent, function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
    cust_id = arguments.get("customer")
    if cust_id not in agent.session_state["customer_profiles"]:
        raise ValueError(f"Customer profile for {cust_id} not found")
    customer_profile = agent.session_state["customer_profiles"][cust_id]

    # Replace the customer with the customer_profile for the function call
    arguments["customer"] = json.dumps(customer_profile)
    # Call the function with the updated arguments
    result = function_call(**arguments)

    return result

 

複数のツールフック

複数のツールフックを一度に割り当てることもできます。割り当てられた順に適用されます。

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[DuckDuckGoTools()],
    tool_hooks=[logger_hook, confirmation_hook],  # The logger_hook will run on the outer layer, and the confirmation_hook will run on the inner layer
)

特定のカスタムツールにツールフックを割り当てることもできます。

@tool(tool_hooks=[logger_hook, confirmation_hook])
def get_top_hackernews_stories(num_stories: int) -> Iterator[str]:
    """Fetch top stories from Hacker News.

    Args:
        num_stories (int): Number of stories to retrieve
    """
    # Fetch top story IDs
    response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
    story_ids = response.json()

    # Yield story details
    final_stories = []
    for story_id in story_ids[:num_stories]:
        story_response = httpx.get(
            f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
        )
        story = story_response.json()
        if "text" in story:
            story.pop("text", None)
        final_stories.append(story)

    return json.dumps(final_stories)

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[get_top_hackernews_stories],
)

 

Pre and Post フック

pre フックと post フックはツールが呼び出される前後の動作を変更することを可能にします。これはツールフックの代替です。

@tool デコレータ内で pre_hook を設定すると、ツール呼び出しの前に関数を実行します。

@tool デコレータ内で post_hook を設定すると、ツール呼び出しの後に関数を実行します。

以下は、Agent Context とともに pre_hook, post_hook を使用するデモ例です。

pre_and_post_hooks.py

import json
from typing import Iterator

import httpx
from agno.agent import Agent
from agno.tools import FunctionCall, tool


def pre_hook(fc: FunctionCall):
    print(f"Pre-hook: {fc.function.name}")
    print(f"Arguments: {fc.arguments}")
    print(f"Result: {fc.result}")


def post_hook(fc: FunctionCall):
    print(f"Post-hook: {fc.function.name}")
    print(f"Arguments: {fc.arguments}")
    print(f"Result: {fc.result}")


@tool(pre_hook=pre_hook, post_hook=post_hook)
def get_top_hackernews_stories(agent: Agent) -> Iterator[str]:
    num_stories = agent.context.get("num_stories", 5) if agent.context else 5

    # Fetch top story IDs
    response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
    story_ids = response.json()

    # Yield story details
    for story_id in story_ids[:num_stories]:
        story_response = httpx.get(
            f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
        )
        story = story_response.json()
        if "text" in story:
            story.pop("text", None)
        yield json.dumps(story)


agent = Agent(
    context={
        "num_stories": 2,
    },
    tools=[get_top_hackernews_stories],
    markdown=True,
    show_tool_calls=True,
)
agent.print_response("What are the top hackernews stories?", stream=True)

以上