人机协作

人机协作 (HITL) 功能允许你在执行工具调用之前或之后获取用户的输入。

下面的示例展示了如何在执行工具调用之前使用工具钩子获取用户确认。

示例：使用工具钩子实现人机协作

本示例演示了如何：

• 为工具添加用户确认钩子
• 在工具执行期间处理用户输入
• 根据用户的选择优雅地取消操作

hitl.py

"""🤝 人机协作：为工具调用添加用户确认

本示例展示了如何在 Agno 工具中实现人机协作功能。
它演示了如何：
- 为工具添加用户确认钩子
- 在工具执行期间处理用户输入
- 根据用户的选择优雅地取消操作

一些实际应用：
- 在执行前确认敏感操作
- 在生成 API 调用前进行审查
- 验证数据转换
- 在关键系统中批准自动化操作

运行 `pip install openai httpx rich agno` 来安装依赖项。
"""

import json
from typing import Any, Callable, Dict, Iterator

import httpx
from agno.agent import Agent
from agno.exceptions import StopAgentRun
from agno.models.openai import OpenAIChat
from agno.tools import FunctionCall, tool
from rich.console import Console
from rich.pretty import pprint
from rich.prompt import Prompt

# 这是控制台用于 print_response 方法的实例
# 我们可以使用它来停止和重新启动实时显示并请求用户确认
console = Console()


def confirmation_hook(
    function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
    # 从控制台获取实时显示实例
    live = console._live

    # 暂时停止实时显示，以便我们可以请求用户确认
    live.stop()  # type: ignore

    # 请求确认
    console.print(f"\n即将运行 [bold blue]{fc.function.name}[/]")
    message = (
        Prompt.ask("是否继续?", choices=["y", "n"], default="y")
        .strip()
        .lower()
    )

    # 重新启动实时显示
    live.start()  # type: ignore

    # 如果用户不想继续，则引发 StopExecution 异常
    if message != "y":
        raise StopAgentRun(
            "用户已取消工具调用",
            agent_message="由于未获得权限，将停止执行。",
        )
    
    # 调用函数
    result = function_call(**arguments)

    # 可选地转换结果

    return result


@tool(tool_hooks=[confirmation_hook])
def get_top_hackernews_stories(num_stories: int) -> Iterator[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()

    # 返回故事详情
    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],
    markdown=True,
)

agent.print_response(
    "获取 Hacker News 的前 2 条热门故事？", stream=True, console=console
)