人机协作

此示例展示了如何在您的 Agent 工作流中实现人工验证。它演示了：

• 执行前验证
• 执行后复核
• 交互式反馈循环
• 质量控制检查点

示例场景：

• 内容审核
• 重要决策审批
• 输出质量验证
• 安全检查
• 专家评审流程

代码

human_in_the_loop.py

import json
from textwrap import dedent
from typing import Iterator

import httpx
from agno.agent import Agent
from agno.exceptions import StopAgentRun
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 pre_hook(fc: FunctionCall):
    # 从控制台获取实时显示实例
    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="由于未获得许可，正在停止执行。",
        )


@tool(pre_hook=pre_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()

    # 产生故事详情
    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)
        yield json.dumps(story)


# 使用技术达人的个性和清晰的指令初始化 Agent
agent = Agent(
    description="一个获取和总结 Hacker News 故事的技术新闻助手",
    instructions=dedent("""\
        你是一个热情的科技记者

        你的职责：
        - 以引人入胜且信息丰富的方式呈现 Hacker News 故事
        - 提供清晰的信息摘要

        风格指南：
        - 使用表情符号使你的回复更具吸引力
        - 保持你的摘要简洁但信息丰富
        - 以友好的科技主题结尾\
    """),
    tools=[get_top_hackernews_stories],
    show_tool_calls=True,
    markdown=True,
)

# 尝试示例问题：
# - "现在 HN 的头条新闻是什么？"
# - "展示 Hacker News 的最新故事"
# - "获取前 5 条故事（您可以尝试接受和拒绝确认）"
agent.print_response(
    "What are the top 2 hackernews stories?", stream=True, console=console
)

用法

1. 创建虚拟环境

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

   Mac

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

2. 安装库

   pip install openai agno
   

3. 运行 Agent

   python human_in_the_loop.py