在大多数生产场景中,您都需要编写自己的工具。这就是为什么 Agno 专注于提供最佳的工具使用体验。
规则很简单:
- 任何 Python 函数都可以被 Agent 用作工具。
- 使用
@tool
装饰器来修改该工具调用之前和之后发生的事情。
任何 Python 函数都可以用作工具
例如,以下是如何将 get_top_hackernews_stories
函数用作工具:
import json
import httpx
from agno.agent import Agent
def get_top_hackernews_stories(num_stories: int = 10) -> str:
"""
使用此函数获取 Hacker News 的热门故事。
Args:
num_stories (int): 要返回的故事数量。默认为 10。
Returns:
str: 热门故事的 JSON 字符串。
"""
# 获取热门故事 ID
response = httpx.get('https://hacker-news.firebaseio.com/v0/topstories.json')
story_ids = response.json()
# 获取故事详情
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)
stories.append(story)
return json.dumps(stories)
agent = Agent(tools=[get_top_hackernews_stories], show_tool_calls=True, markdown=True)
agent.print_response("总结 Hacker News 的前 5 个故事?", stream=True)
要修改工具的行为,请使用 @tool
装饰器。一些值得注意的特性:
requires_confirmation=True
: 在执行前需要用户确认。
requires_user_input=True
: 在执行前需要用户输入。使用 user_input_fields
指定需要用户输入的字段。
external_execution=True
: 工具将在 Agent 的控制之外执行。
show_result=True
: 在 Agent 的响应中显示工具调用的输出。没有此标志,工具调用的结果将被发送到模型进行进一步处理。
stop_after_tool_call=True
: 在工具调用后停止 Agent 运行。
tool_hooks
: 在此工具调用之前和之后运行自定义逻辑。
cache_results=True
: 缓存工具结果以避免重复相同的调用。使用 cache_dir
和 cache_ttl
配置缓存。
这是一个使用 @tool
装饰器上许多可能参数的示例。
import httpx
from agno.agent import Agent
from agno.tools import tool
from typing import Any, Callable, Dict
def logger_hook(function_name: str, function_call: Callable, arguments: Dict[str, Any]):
"""包装工具执行的 Hook 函数"""
print(f"即将调用 {function_name},参数为:{arguments}")
result = function_call(**arguments)
print(f"函数调用完成,结果为:{result}")
return result
@tool(
name="fetch_hackernews_stories", # 工具的自定义名称(否则使用函数名称)
description="获取 Hacker News 的热门故事", # 自定义描述(否则使用函数文档字符串)
show_result=True, # 函数调用后显示结果
stop_after_tool_call=True, # 工具调用后立即返回结果并停止 Agent
tool_hooks=[logger_hook], # 在执行前后运行的 Hook
requires_confirmation=True, # 在执行前需要用户确认
cache_results=True, # 启用结果缓存
cache_dir="/tmp/agno_cache", # 自定义缓存目录
cache_ttl=3600 # 缓存的 TTL(以秒为单位,1 小时)
)
def get_top_hackernews_stories(num_stories: int = 5) -> str:
"""
从 Hacker News 获取热门故事。
Args:
num_stories: 要获取的故事数量(默认:5)
Returns:
str: 以文本格式的热门故事
"""
# 获取热门故事 ID
response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
story_ids = response.json()
# 获取故事详情
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()
stories.append(f"{story.get('title')} - {story.get('url', '无 URL')}")
return "\n".join(stories)
agent = Agent(tools=[get_top_hackernews_stories])
agent.print_response("给我看看 Hacker News 的热门新闻")
参数 | 类型 | 描述 |
---|
name | str | 函数名称的替代名称 |
description | str | 函数描述的替代描述 |
show_result | bool | 如果为 True,则在函数调用后显示结果 |
stop_after_tool_call | bool | 如果为 True,Agent 将在函数调用后停止 |
tool_hooks | list[Callable] | 包装函数执行的 Hook 列表 |
pre_hook | Callable | 在函数执行前运行的 Hook |
post_hook | Callable | 在函数执行后运行的 Hook |
requires_confirmation | bool | 如果为 True,则在执行前需要用户确认 |
requires_user_input | bool | 如果为 True,则在执行前需要用户输入 |
user_input_fields | list[str] | 需要用户输入的字段列表 |
external_execution | bool | 如果为 True,工具将在 Agent 的控制之外执行 |
cache_results | bool | 如果为 True,则启用函数结果的缓存 |
cache_dir | str | 用于存储缓存文件的目录 |
cache_ttl | int | 缓存结果的生存时间(以秒为单位,默认为 3600) |
Responses are generated using AI and may contain mistakes.