Agno でエージェントの実行のフローを制御する方法を学習します。これはまた “Human in the Loop” とも呼ばれます。Agno のユーザ制御フローは “Human in the Loop” パターンの実装を可能にします、そこではエージェントの実行中に人間の監視や入力が必要となります。

Agno : ユーザガイド : コンセプト : エージェント – ユーザ制御フロー (Human in the Loop)

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

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

User Guide : Concepts : Agent – User Control Flows

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

Agno ユーザガイド : コンセプト : エージェント – ユーザ制御フロー (Human in the Loop)

Agno でエージェントの実行のフローを制御する方法を学習します。これはまた “Human in the Loop” とも呼ばれます。

Agno のユーザ制御フローは “Human in the Loop” パターンの実装を可能にします、そこではエージェントの実行中に人間の監視や入力が必要となります。これは以下のために重要です :

機密性の高い操作の検証
実行前にツール呼び出しをレビュー
意思決定のためにユーザ入力を収集する
外部ツール実行の管理

ユーザ制御フローの種類

Agno は 4 つの主要なタイプのユーザ制御フローをサポートします :

ユーザ確認 (Confirmation) : ツール呼び出しの実行前に明示的なユーザ承認を必要とします
ユーザ入力 : 実行中にユーザから特定の情報を収集します
動的ユーザ入力 : エージェントにユーザ入力を必要に応じて収集させます
外部ツール実行 : エージェントの制御外でツールを実行します

エージェント実行の一時停止

ユーザ制御フローはエージェントの実行を中断 (interrupt) し、人間の監視を必要とします。それから continue_run メソッドを呼び出すことで実行が継続されます。

例えば :

agent.run("Perform sensitive operation")

if agent.is_paused:
    # The agent will pause while human input is provided
    # ... perform other tasks

    # The user can then continue the run
    response = agent.continue_run()
    # or response = await agent.acontinue_run()

continue_run メソッドは、一時停止時のエージェントの状態を使用して継続します。また、特定の実行の run_response、または run_id を continue_run メソッドに渡すこともできます。

ユーザ確認

ユーザ確認 (confirmation) により、実行を一時停止して、ツール呼び出しを進める前に明示的なユーザの承認を要求することを可能にします。これは以下のために有用です :

機密性の高い操作
データを変更する API 呼び出し
重要な結果を伴うアクション

次の例はユーザ確認を実装する方法を示しています。

from agno.tools import tool
from agno.agent import Agent
from agno.models.openai import OpenAIChat

@tool(requires_confirmation=True)
def sensitive_operation(data: str) -> str:
    """Perform a sensitive operation that requires confirmation."""
    # Implementation here
    return "Operation completed"

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

# Run the agent
agent.run("Perform sensitive operation")

# Handle confirmation
if agent.is_paused:
    for tool in agent.run_response.tools_requiring_confirmation:
        # Get user confirmation
        print(f"Tool {tool.tool_name}({tool.tool_args}) requires confirmation")
        confirmed = input(f"Confirm? (y/n): ").lower() == "y"
        tool.confirmed = confirmed

  # Continue execution
  response = agent.continue_run()

ツールキットのどのツールが確認を必要とするか指定することもできます。

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
from agno.tools.yfinance import YFinanceTools
from agno.utils import pprint

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

agent.run("What is the current stock price of Apple?")
if agent.is_paused:
    for tool in agent.run_response.tools_requiring_confirmation:
        print(f"Tool {tool.tool_name}({tool.tool_args}) requires confirmation")
        confirmed = input(f"Confirm? (y/n): ").lower() == "y"

        if message == "n":
            tool.confirmed = False
        else:
            # We update the tools in place
            tool.confirmed = True

    run_response = agent.continue_run()
    pprint.pprint_run_response(run_response)

ユーザ入力

ユーザ入力フローは、実行中にユーザからの特定の情報を収集することを可能にします。これは以下のために有用です :

必要なパラメータの収集
ユーザ設定の取得
不足している情報の収集

次の例では、ユーザからの send_email ツール用のすべての入力を必要とします。

from typing import List
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.function import UserInputField

# We still provide a docstring to the tool; This will be used to populate the `user_input_schema`
@tool(requires_user_input=True)
def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email to the user.

    Args:
        to (str): The address to send the email to.
        subject (str): The subject of the email.
        body (str): The body of the email.
    """
    # Implementation here
    return f"Email sent to {to} with subject {subject} and body {body}"

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

agent.run("Send an email please")
if agent.is_paused:
    for tool in agent.run_response.tools_requiring_user_input:
        input_schema: List[UserInputField] = tool.user_input_schema

        for field in input_schema:
            # Display field information to the user
            print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")

            # Get user input
            user_value = input(f"Please enter a value for {field.name}: ")

            # Update the field value
            field.value = user_value

    run_response = (
        agent.continue_run()
    )

RunResponse オブジェクトはツールのリストを持ち、requires_user_input の場合、入力を必要とするツールには user_input_schema の値が設定されます。これは UserInputField オブジェクトのリストです。

class UserInputField:
    name: str  # The name of the field
    field_type: Type  # The required type of the field
    description: Optional[str] = None  # The description of the field
    value: Optional[Any] = None  # The value of the field. Populated by the agent or the user.

どのフィールドがユーザにより入力される必要がある指定することもできます、フィールドの残りはエージェントが提供します。

# You can either specify the user_input_fields leave empty for all fields to be provided by the user
@tool(requires_user_input=True, user_input_fields=["to_address"])
def send_email(subject: str, body: str, to_address: str) -> str:
    """
    Send an email.

    Args:
        subject (str): The subject of the email.
        body (str): The body of the email.
        to_address (str): The address to send the email to.
    """
    return f"Sent email to {to_address} with subject {subject} and body {body}"

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

agent.run("Send an email with the subject 'Hello' and the body 'Hello, world!'")
if agent.is_paused:
    for tool in agent.run_response.tools_requiring_user_input:
        input_schema: List[UserInputField] = tool.user_input_schema

        for field in input_schema:
            # Display field information to the user
            print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")

            # Get user input (if the value is not set, it means the user needs to provide the value)
            if field.value is None:
                user_value = input(f"Please enter a value for {field.name}: ")
                field.value = user_value
            else:
                print(f"Value provided by the agent: {field.value}")

    run_response = (
        agent.continue_run()
    )

動的ユーザ入力

このパターンはエージェントに、ユーザ入力を必要とするときを示すツールを提供します。それは以下のために最適です :

ユーザがエージェントとやり取りする方法が不明な場合
ユーザと form のようなインタラクションを望む場合

次の例では、エージェントが必要なときにユーザのフィードバックを収集できるような専用ツールを使用します。

from typing import Any, Dict

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
from agno.tools.toolkit import Toolkit
from agno.tools.user_control_flow import UserControlFlowTools
from agno.utils import pprint

# Example toolkit for handling emails
class EmailTools(Toolkit):
    def __init__(self, *args, **kwargs):
        super().__init__(
            name="EmailTools", tools=[self.send_email, self.get_emails], *args, **kwargs
        )

    def send_email(self, subject: str, body: str, to_address: str) -> str:
        """Send an email to the given address with the given subject and body.

        Args:
            subject (str): The subject of the email.
            body (str): The body of the email.
            to_address (str): The address to send the email to.
        """
        return f"Sent email to {to_address} with subject {subject} and body {body}"

    def get_emails(self, date_from: str, date_to: str) -> str:
        """Get all emails between the given dates.

        Args:
            date_from (str): The start date.
            date_to (str): The end date.
        """
        return [
            {
                "subject": "Hello",
                "body": "Hello, world!",
                "to_address": "test@test.com",
                "date": date_from,
            },
            {
                "subject": "Random other email",
                "body": "This is a random other email",
                "to_address": "john@doe.com",
                "date": date_to,
            },
        ]


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[EmailTools(), UserControlFlowTools()],
    markdown=True,
    debug_mode=True,
)

run_response = agent.run("Send an email with the body 'How is it going in Tokyo?'")

# We use a while loop to continue the running until the agent is satisfied with the user input
while run_response.is_paused:
    for tool in run_response.tools_requiring_user_input:
        input_schema: List[UserInputField] = tool.user_input_schema

        for field in input_schema:
            # Display field information to the user
            print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")

            # Get user input (if the value is not set, it means the user needs to provide the value)
            if field.value is None:
                user_value = input(f"Please enter a value for {field.name}: ")
                field.value = user_value
            else:
                print(f"Value provided by the agent: {field.value}")

    run_response = agent.continue_run(run_response=run_response)

    # If the agent is not paused for input, we are done
    if not run_response.is_paused:
        pprint.pprint_run_response(run_response)
        break

外部ツール実行

外部ツール実行は、エージェントの制御の外でツールを実行することを可能にします。これは以下のために有用です :

外部サービス呼び出し
データベース操作

import subprocess

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
from agno.utils import pprint


# We have to create a tool with the correct name, arguments and docstring for the agent to know what to call.
@tool(external_execution=True)
def execute_shell_command(command: str) -> str:
    """Execute a shell command.

    Args:
        command (str): The shell command to execute

    Returns:
        str: The output of the shell command
    """
    return subprocess.check_output(command, shell=True).decode("utf-8")


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

run_response = agent.run("What files do I have in my current directory?")
if run_response.is_paused:
    for tool in run_response.tools_awaiting_external_execution:
        if tool.tool_name == execute_shell_command.name:
            print(f"Executing {tool.tool_name} with args {tool.tool_args} externally")

            # We execute the tool ourselves. You can execute any function or process here and use the tool_args as input.
            result = execute_shell_command.entrypoint(**tool.tool_args)
            # We have to set the result on the tool execution object so that the agent can continue
            tool.result = result

    run_response = agent.continue_run()
    pprint.pprint_run_response(run_response)

以上

月	火	水	木	金	土	日
	1	2	3	4	5	6
7	8	9	10	11	12	13
14	15	16	17	18	19	20
21	22	23	24	25	26	27
28	29	30	31