用户控制流

了解如何在 Agno 中控制代理的执行流程。这也称为“人工干预”。

Agno 中的用户控制流使你能够实现“人工干预”模式，在代理执行期间需要人工监督和输入。这对于以下方面至关重要：

• 验证敏感操作
• 在执行前审查工具调用
• 收集用户输入以进行决策
• 管理外部工具执行

用户控制流类型

Agno 支持四种主要的用户控制流类型：

1. 用户確認: 在执行工具调用前需要用户明确批准
2. 用户输入: 在执行期间从用户那里收集特定信息
3. 动态用户输入: 让代理根据需要收集用户输入
4. 外部工具执行: 执行代理控制之外的工具

暂停代理执行

用户控制流会中断代理的执行并需要人工监督。然后可以通过调用 continue_run 方法来继续执行。

例如：

agent.run("Perform sensitive operation")

if agent.is_paused:
    # 当提供人工输入时，代理将暂停
    # ... 执行其他任务

    # 然后用户可以继续执行
    response = agent.continue_run()
    # 或者 response = await agent.acontinue_run()

continue_run 方法会根据暂停时的代理状态继续执行。你也可以将特定运行的 run_response 或 run_id 传递给 continue_run 方法。

用户确认

用户确认允许你在继续工具调用之前暂停执行并需要用户明确批准。这对于以下情况很有用：

• 敏感操作
• 修改数据的 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:
    """执行需要确认的敏感操作。"""
    # 在此处实现
    return "Operation completed"

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

# 运行代理
agent.run("Perform sensitive operation")

# 处理确认
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"
        tool.confirmed = confirmed

  # 继续执行
  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:
            # 我们会就地更新工具
            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

# 我们仍然为工具提供文档字符串；这将用于填充 `user_input_schema`
@tool(requires_user_input=True)
def send_email(to: str, subject: str, body: str) -> dict:
    """向用户发送电子邮件。

    Args:
        to (str): 要发送电子邮件的地址。
        subject (str): 电子邮件的主题。
        body (str): 电子邮件的正文。
    """
    # 在此处实现
    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:
            # 向用户显示字段信息
            print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")

            # 获取用户输入
            user_value = input(f"Please enter a value for {field.name}: ")

            # 更新字段值
            field.value = user_value

    run_response = (
        agent.continue_run()
    )

RunResponse 对象有一个工具列表，在 requires_user_input 的情况下，需要输入的工具会填充 user_input_schema。 这是一个 UserInputField 对象列表。

class UserInputField:
    name: str  # 字段的名称
    field_type: Type  # 字段的必需类型
    description: Optional[str] = None  # 字段的描述
    value: Optional[Any] = None  # 字段的值。由代理或用户填充。

你还可以指定哪些字段应由用户填写，而代理将提供其余字段。

# 你可以指定 `user_input_fields`，将其他字段留空，由用户提供所有字段
@tool(requires_user_input=True, user_input_fields=["to_address"])
def send_email(subject: str, body: str, to_address: str) -> str:
    """
    发送电子邮件。

    Args:
        subject (str): 电子邮件的主题。
        body (str): 电子邮件的正文。
        to_address (str): 要发送电子邮件的地址。
    """
    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:
            # 向用户显示字段信息
            print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")

            # 获取用户输入（如果值未设置，则表示用户需要提供该值）
            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()
    )

动态用户输入

此模式为代理提供了在需要用户输入时发出信号的工具。它非常适合：

• 不确定用户如何与代理交互的情况
• 当你希望与用户进行表单式交互时

在以下示例中，我们使用了一个专用工具来允许代理在需要时收集用户反馈。

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

# 处理电子邮件的示例工具包
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:
        """使用给定的主题和正文向给定的地址发送电子邮件。

        Args:
            subject (str): 电子邮件的主题。
            body (str): 电子邮件的正文。
            to_address (str): 要发送电子邮件的地址。
        """
        return f"Sent email to {to_address} with subject {subject} and body {body}"

    def get_emails(self, date_from: str, date_to: str) -> str:
        """获取给定日期范围内的所有电子邮件。

        Args:
            date_from (str): 开始日期。
            date_to (str): 结束日期。
        """
        return [
            {
                "subject": "Hello",
                "body": "Hello, world!",
                "to_address": "[e-mail]",
                "date": date_from,
            },
            {
                "subject": "Random other email",
                "body": "This is a random other email",
                "to_address": "[e-mail]",
                "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?'")

# 我们使用 while 循环继续运行，直到代理对用户输入感到满意为止
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:
            # 向用户显示字段信息
            print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")

            # 获取用户输入（如果值未设置，则表示用户需要提供该值）
            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 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


# 我们必须创建一个具有正确名称、参数和文档字符串的工具，以便代理知道要调用哪个工具。
@tool(external_execution=True)
def execute_shell_command(command: str) -> str:
    """执行 shell 命令。

    Args:
        command (str): 要执行的 shell 命令

    Returns:
        str: shell 命令的输出
    """
    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")

            # 我们自己执行工具。你可以执行任何函数或进程，并使用 tool_args 作为输入。
            result = execute_shell_command.entrypoint(**tool.tool_args)
            # 我们必须将结果设置在 tool execution 对象上，以便代理可以继续
            tool.result = result

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

最佳实践

1. 净化用户输入: 始终验证和净化用户输入，以防止安全漏洞。
2. 错误处理: 始终为用户输入和外部调用实现适当的错误处理。
3. 输入验证: 在处理之前验证用户输入。

开发者资源

• 查看更多 示例
• 查看 Cookbook