​
代码

cookbook/agent_concepts/user_control_flows/confirmation_required_async.py
import asyncio
import json

import httpx
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
from agno.utils import pprint
from rich.console import Console
from rich.prompt import Prompt

console = Console()

@tool(requires_confirmation=True)
async def get_top_hackernews_stories(num_stories: int) -> str:
    """从 Hacker News 获取热门故事。

    Args:
        num_stories (int): 要检索的故事数量

    Returns:
        str: 包含故事详情的 JSON 字符串
    """
    # 获取热门故事 ID
    response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
    story_ids = response.json()

    # 产出故事详情
    all_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)
        all_stories.append(story)
    return json.dumps(all_stories)

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

run_response = asyncio.run(agent.arun("Fetch the top 2 hackernews stories"))
if run_response.is_paused:
    for tool in run_response.tools_requiring_confirmation:
        # 请求确认
        console.print(
            f"Tool name [bold blue]{tool.tool_name}({tool.tool_args})[/] requires confirmation."
        )
        message = (
            Prompt.ask("Do you want to continue?", choices=["y", "n"], default="y")
            .strip()
            .lower()
        )

        if message == "n":
            tool.confirmed = False
        else:
            tool.confirmed = True

run_response = asyncio.run(agent.acontinue_run(run_response=run_response))
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 httpx rich openai
4

运行示例

python cookbook/agent_concepts/user_control_flows/confirmation_required_async.py

​
主要特点

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

​
用例

  • 非阻塞用户确认流程
  • 需要异步执行的高性能应用程序
  • 具有用户交互的 Web 应用程序
  • 带有用户输入的长时间运行操作