​
代码

cookbook/agent_concepts/user_control_flows/user_input_required_async.py
import asyncio
from typing import List

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

@tool(requires_user_input=True, user_input_fields=["to_address"])
async def send_email(subject: str, body: str, to_address: str) -> str:
    """
    发送电子邮件。

    参数:
        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-mini"),
    tools=[send_email],
    markdown=True,
)

asyncio.run(
    agent.arun("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"\n字段: {field.name}")
            print(f"描述: {field.description}")
            print(f"类型: {field.field_type}")

            # 获取用户输入
            if field.value is None:
                user_value = input(f"请输入 {field.name} 的值: ")
                # 更新字段值
                field.value = user_value
            else:
                print(f"值: {field.value}")


    run_response = asyncio.run(agent.acontinue_run())
    pprint.pprint_run_response(run_response)

​
用法

1

创建虚拟环境

打开 Terminal 并创建一个 python 虚拟环境。

python3 -m venv .venv
source .venv/bin/activate
2

设置您的 API 密钥

export OPENAI_API_KEY=xxx
3

安装库

pip install -U agno openai
4

运行示例

python cookbook/agent_concepts/user_control_flows/user_input_required_async.py

​
主要特点

  • 使用 agent.arun() 进行异步代理执行
  • 实现 agent.acontinue_run() 进行异步继续
  • 保持与同步版本相同的用户输入流程
  • 演示如何处理具有用户输入收集的异步执行

​
用途

  • 非阻塞用户输入收集
  • 需要异步执行的高性能应用程序
  • 具有类似表单交互的 Web 应用程序
  • 需要用户输入的长时间运行操作