{ "cells": [ { "cell_type": "code", "execution_count": 1, "id": "b4617320", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "True" ] }, "execution_count": 1, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from textwrap import dedent\n", "\n", "from agno.agent import Agent\n", "from agno.models.openai import OpenAILike\n", "from agno.exceptions import StopAgentRun\n", "from agno.tools import tool,FunctionCall\n", "from agno.tools.function import UserInputField\n", "from agno.utils import pprint\n", "\n", "from dotenv import load_dotenv \n", "\n", "from typing import Iterator,List\n", "\n", "from rich.console import Console\n", "from rich.prompt import Prompt\n", "\n", "import httpx\n", "import os\n", "import json\n", "load_dotenv()" ] }, { "cell_type": "markdown", "id": "72dce64b", "metadata": {}, "source": [ "**User Control Flows(Human in loop)**" ] }, { "cell_type": "code", "execution_count": 2, "id": "cb6ea5c6", "metadata": {}, "outputs": [ { "data": { "application/vnd.jupyter.widget-view+json": { "model_id": "26ccd24ca6fa453490acd99692ca1ba6", "version_major": 2, "version_minor": 0 }, "text/plain": [ "Output()" ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/html": [ "
\n" ], "text/plain": [] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/html": [ "\n",
"About to run get_top_hackernews_stories\n",
"
\n"
],
"text/plain": [
"\n",
"About to run \u001b[1;34mget_top_hackernews_stories\u001b[0m\n"
]
},
"metadata": {},
"output_type": "display_data"
},
{
"data": {
"text/html": [
"Do you want to continue? [y/n] (y):\n" ], "text/plain": [ "Do you want to continue? \u001b[1;35m[y/n]\u001b[0m \u001b[1;36m(y)\u001b[0m: " ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/html": [ "\n" ], "text/plain": [] }, "metadata": {}, "output_type": "display_data" } ], "source": [ "# This is the console instance used by the print_response method\n", "# We can use this to stop and restart the live display and ask for user confirmation\n", "console = Console()\n", "\n", "\n", "def pre_hook(fc: FunctionCall):\n", " # Get the live display instance from the console\n", " live = console._live\n", "\n", " # Stop the live display temporarily so we can ask for user confirmation\n", " live.stop() # type: ignore\n", "\n", " # Ask for confirmation\n", " console.print(f\"\\nAbout to run [bold blue]{fc.function.name}[/]\")\n", " message = (\n", " Prompt.ask(\"Do you want to continue?\", choices=[\"y\", \"n\"], default=\"y\")\n", " .strip()\n", " .lower()\n", " )\n", "\n", " # Restart the live display\n", " live.start() # type: ignore\n", "\n", " # If the user does not want to continue, raise a StopExecution exception\n", " if message != \"y\":\n", " raise StopAgentRun(\n", " \"Tool call cancelled by user\", \n", " agent_message=\"Stopping execution as permission was not granted.\",\n", " )\n", "\n", "\n", "@tool(pre_hook=pre_hook)\n", "def get_top_hackernews_stories(num_stories: int) -> Iterator[str]:\n", " \"\"\"Fetch top stories from Hacker News after user confirmation.\n", "\n", " Args:\n", " num_stories (int): Number of stories to retrieve\n", "\n", " Returns:\n", " str: JSON string containing story details\n", " \"\"\"\n", " # Fetch top story IDs\n", " response = httpx.get(\"https://hacker-news.firebaseio.com/v0/topstories.json\")\n", " story_ids = response.json()\n", "\n", " # Yield story details\n", " for story_id in story_ids[:num_stories]:\n", " story_response = httpx.get(\n", " f\"https://hacker-news.firebaseio.com/v0/item/{story_id}.json\"\n", " )\n", " story = story_response.json()\n", " if \"text\" in story:\n", " story.pop(\"text\", None)\n", " yield json.dumps(story)\n", "\n", "\n", "# Initialize the agent with a tech-savvy personality and clear instructions\n", "agent = Agent(\n", " model=OpenAILike(id=\"qwen3-30b-a3b\", \n", " api_key=os.getenv(\"BAILIAN_API_KEY\"), \n", " base_url=os.getenv(\"BAILIAN_API_BASE_URL\"),\n", " request_params={\"extra_body\": {\"enable_thinking\": False}},),\n", " description=\"A Tech News Assistant that fetches and summarizes Hacker News stories\",\n", " instructions=dedent(\"\"\"\\\n", " You are an enthusiastic Tech Reporter\n", "\n", " Your responsibilities:\n", " - Present Hacker News stories in an engaging and informative way\n", " - Provide clear summaries of the information you gather\n", "\n", " Style guide:\n", " - Use emoji to make your responses more engaging\n", " - Keep your summaries concise but informative\n", " - End with a friendly tech-themed sign-off\\\n", " \"\"\"),\n", " tools=[get_top_hackernews_stories],\n", " show_tool_calls=True,\n", " markdown=True,\n", ")\n", "\n", "# Example questions to try:\n", "# - \"What are the top 3 HN stories right now?\"\n", "# - \"Show me the most recent story from Hacker News\"\n", "# - \"Get the top 5 stories (you can try accepting and declining the confirmation)\"\n", "agent.print_response(\n", " \"What are the top 2 hackernews stories?\", stream=True, console=console\n", ")" ] }, { "cell_type": "code", "execution_count": null, "id": "9485bd62", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Field: subject\n", "Description: The subject of the email.\n", "Type:
╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮\n", "│ The email with the body \"Hello, world!\" has been successfully sent. Let me know if there's anything else I can │\n", "│ assist you with! │\n", "╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯\n", "\n" ], "text/plain": [ "\u001b[34m╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮\u001b[0m\n", "\u001b[34m│\u001b[0m The email with the body \"Hello, world!\" has been successfully sent. Let me know if there's anything else I can \u001b[34m│\u001b[0m\n", "\u001b[34m│\u001b[0m assist you with! \u001b[34m│\u001b[0m\n", "\u001b[34m╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯\u001b[0m\n" ] }, "metadata": {}, "output_type": "display_data" } ], "source": [ "# You can either specify the user_input_fields leave empty for all fields to be provided by the user\n", "@tool(requires_user_input=True, user_input_fields=[\"subject\", \"to_address\"])\n", "def send_email(subject: str, body: str, to_address: str) -> str:\n", " \"\"\"\n", " Send an email.\n", "\n", " Args:\n", " subject (str): The subject of the email.\n", " body (str): The body of the email.\n", " to_address (str): The address to send the email to.\n", " \"\"\"\n", " return f\"Sent email to {to_address} with subject {subject} and body {body}\"\n", "\n", "agent = Agent(\n", " model=OpenAILike(id=\"qwen3-32b\", \n", " api_key=os.getenv(\"BAILIAN_API_KEY\"), \n", " base_url=os.getenv(\"BAILIAN_API_BASE_URL\"),\n", " request_params={\"extra_body\": {\"enable_thinking\": False}},),\n", " tools=[send_email],\n", " show_tool_calls=True,\n", " markdown=True,\n", ")\n", "\n", "agent.run(\"Send an email with the body 'Hello, world!'\")\n", "if agent.is_paused and agent.run_response:\n", " for tool in agent.run_response.tools_requiring_user_input:\n", " if tool.user_input_schema:\n", " input_schema: List[UserInputField] = tool.user_input_schema\n", "\n", " for field in input_schema:\n", " # Display field information to the user\n", " print(f\"\\nField: {field.name}\")\n", " print(f\"Description: {field.description}\")\n", " print(f\"Type: {field.field_type}\")\n", "\n", " # Get user input\n", " if field.value is None:\n", " user_value = input(f\"Please enter a value for {field.name}: \")\n", " # Update the field value\n", " field.value = user_value\n", " else:\n", " print(f\"Value: {field.value}\")\n", "\n", " run_response = agent.continue_run()\n", " pprint.pprint_run_response(run_response)" ] }, { "cell_type": "code", "execution_count": null, "id": "fd9fbd3d", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.13" } }, "nbformat": 4, "nbformat_minor": 5 }