第 8 章 · 自定义工具与 Toolkit 开发
本章目标：掌握三种自定义工具的写法——普通函数、
@tool装饰器、Toolkit 子类，学会用装饰器参数控制工具行为，并建立正确的工具设计粒度观。
8.1 普通函数：最轻量的自定义工具
上一章已经用过普通函数作工具，这里补全工程细节。函数工具支持同步与异步两种定义：
python
# custom_func.py —— 同步/异步函数皆可作工具
import asyncio
import os
import sqlite3
from agno.agent import Agent
from agno.models.openai import OpenAIChat
def get_user(user_id: int) -> dict:
"""Get a user's name and email by user id.
Args:
user_id: The numeric id of the user.
"""
# 真实项目里这里通常是查数据库或调内部 API
return {"id": user_id, "name": "高同学", "email": "gao@example.com"}
async def query_report(report_date: str) -> str:
"""Fetch the daily report content for a given date.
Args:
report_date: Date string in YYYY-MM-DD format.
"""
await asyncio.sleep(0.1) # 模拟 IO
return f"{report_date} 的日报：今日新增用户 1024 人。"
agent = Agent(
model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
),
tools=[get_user, query_report], # 同步异步可以混搭
)
agent.print_response("查一下用户 42 的邮箱，再给我 2026-01-10 的日报要点")返回值约定
工具可以返回字符串、数字、字典、列表等可序列化值。返回结构化数据（dict/list）通常比拼接大段文本更好——模型能更准确地提取所需字段。若需要随文本一起返回媒体文件，可使用 ToolResult。
8.2 @tool 装饰器：给函数加上行为控制
agno.tools.tool 装饰器（v2 新增）可以在不改变函数体的前提下为工具附加执行策略。以下参数均已在 agno 2.9 源码中核实：
| 参数 | 默认 | 说明 |
|---|---|---|
name / description | None | 覆盖模型可见的工具名/描述 |
show_result | False | 执行后直接把结果展示给用户 |
stop_after_tool_call | False | 执行完该工具后终止本次 run |
cache_results / cache_dir / cache_ttl | False | 结果缓存及目录、有效期（秒） |
requires_confirmation | False | 执行前需人工确认（第 16 章展开） |
pre_hook / post_hook / tool_hooks | None | 工具执行前后的钩子 |
python
# tool_decorator.py —— 用 @tool 控制工具行为
import os
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
@tool(cache_results=True, cache_ttl=300) # 相同入参 5 分钟内直接走缓存
def slow_exchange_rate(base: str, quote: str) -> str:
"""Get the current exchange rate from base currency to quote currency.
Args:
base: The base currency code, e.g. USD.
quote: The quote currency code, e.g. CNY.
"""
time.sleep(2) # 模拟昂贵的第三方 API 调用
return f"1 {base} = 7.23 {quote}"
@tool(stop_after_tool_call=True, show_result=True)
def submit_ticket(title: str) -> str:
"""Create a support ticket and finish the conversation.
Args:
title: Short summary of the issue.
"""
return f"工单 #{8848} 已创建：{title}"
agent = Agent(
model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
),
tools=[slow_exchange_rate, submit_ticket],
)
agent.print_response("现在美元兑人民币汇率是多少？")
agent.print_response("帮我提交一个工单：笔记本无法连接 WiFi")两个参数的语义值得咀嚼：
stop_after_tool_call：适合"终态动作"。提交工单后没必要让模型再总结一轮，直接结束 run，省一次模型调用；show_result：适合"结果即答案"的确定性输出（计算器、查询类），跳过模型的复述环节。
8.3 Toolkit 子类：批量注册一组相关工具
当一个业务域有多个操作且需要共享状态（如统一的客户端、鉴权配置），就该写 Toolkit 了。写法是继承 agno.tools.Toolkit 并实现 __init__：
python
# todo_toolkit.py —— 自定义待办事项 Toolkit
import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import Toolkit
class TodoTools(Toolkit):
def __init__(self, **kwargs):
super().__init__(name="todo_tools", **kwargs)
self._todos: list[str] = [] # 工具包内共享状态
self.register(self.add_todo)
self.register(self.list_todos)
self.register(self.remove_todo)
def add_todo(self, task: str) -> str:
"""Add a task to the todo list.
Args:
task: The task description to add.
"""
self._todos.append(task)
return f"已添加：{task}"
def list_todos(self) -> str:
"""List all tasks in the todo list."""
if not self._todos:
return "清单为空"
return "\n".join(f"{i + 1}. {t}" for i, t in enumerate(self._todos))
def remove_todo(self, index: int) -> str:
"""Remove a task by its 1-based index.
Args:
index: The 1-based position of the task to remove.
"""
removed = self._todos.pop(index - 1)
return f"已移除：{removed}"
agent = Agent(
model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
),
tools=[TodoTools()],
instructions="管理用户的待办清单，增删查后都要展示当前完整列表。",
)
agent.print_response("添加三件事：买牛奶、写周报、跑步半小时")
agent.print_response("把第 2 项删掉")Toolkit 构造时还能按名字批量声明控制策略（同样来自 2.9 源码中的 Toolkit.__init__ 参数）：stop_after_tool_call_tools=["submit_order"]、show_result_tools=[...]、requires_confirmation_tools=[...] 以及包级 cache_results=True。这与 8.2 的装饰器效果相同，只是作用在 Toolkit 维度。
8.4 工具内的异常处理
工具会访问外部世界，失败是常态。Agno 会捕获工具抛出的异常并把错误信息回传给模型，让模型决定重试还是换路；你也可以通过 tool_hooks 做统一兜底：
python
# safe_tools.py —— 异常兜底示例
import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool
def retry_on_error(func_name: str, func, *args, **kwargs):
"""post_hook 风格的兜底：记录并转换异常。"""
try:
return func(*args, **kwargs)
except Exception as e:
# 返回给模型的错误信息应当"可行动"：告诉它可以怎么办
return f"调用失败({type(e).__name__)})：{e}。请检查参数后重试一次。"
@tool(post_hook=retry_on_error)
def divide(a: float, b: float) -> float:
"""Divide a by b.
Args:
a: Numerator.
b: Denominator, must not be zero.
"""
return a / b # b=0 时 ZeroDivisionError 会被钩子接住
agent = Agent(
model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
),
tools=[divide],
)
agent.print_response("计算 10 除以 0 的结果")经验法则：错误信息写给"人"和写给"模型"要分开设计。给模型的错误要说清楚下一步能做什么（"请检查参数""该城市暂无数据，请换一个城市"），而不是堆一段堆栈。
8.5 工具设计的粒度原则
初学者最容易犯的两个极端：
- ❌ 粒度太细：
get_id、get_name、get_email三个工具，模型要多轮调用拼装； - ❌ 粒度太粗：一个
manage_company(query: str)万能接口，schema 含糊导致模型乱传参。
推荐的做法是面向任务建模：一个工具完成一件模型视角下完整的事，参数尽量少而明确。对比：
python
# granularity.py —— 面向任务 vs 面向数据库
def get_weather_bad(city: str, field: str) -> str:
"""Get one field of weather. BAD: 模型要先猜 field 取值."""
def get_weather_good(city: str) -> dict:
"""Get current weather summary (temperature, condition, humidity) for a city.
Args:
city: City name in Chinese or English.
"""
# 一次性返回结构化全量字段，模型自己挑需要的用
return {"city": city, "temperature": 26, "condition": "多云", "humidity": 62}此外还有两条生产守则：
- 幂等与副作用分离：只读工具随便调；有副作用的工具（下单、发邮件）务必配
requires_confirmation=True或限流； - 超时与重试在工具内部解决：不要指望模型自己处理网络抖动。
本章小结
- 三种自定义姿势：普通函数（最轻）、
@tool装饰器（附加执行策略）、Toolkit子类（共享状态 + 批量注册）； @tool核心参数：cache_results缓存、stop_after_tool_call终止 run、show_result直出结果、requires_confirmation人工确认；- Toolkit 可用
*_tools列表参数按名字批量声明策略； - 工具异常会被 Agno 捕获回传给模型，错误文案要写成"可行动"的指引；
- 工具按任务粒度设计：一次调用解决一个完整问题，返回结构化数据。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 某工具用于"提交订单"，希望调用完成后立即结束本次运行，应使用哪个参数？
2. 关于 Toolkit 子类中 self.register(self.add_todo) 的作用，正确的是？
3. 工具执行时抛出了异常，Agno 的默认行为是？
4. 下列哪种工具设计最符合"面向任务的粒度"原则？
🛠️ 动手实践
- 为 8.3 节的
TodoTools增加clear_finished(done_index: int)方法并注册，验证模型能用自然语言正确传入"第几项"。 - 给某个模拟耗时 3 秒的行情工具分别开启/关闭
cache_results，连续提问两次，用日志对比两次运行的响应差异。 - 设计一个"文件统计"工具集：
count_lines(path)与search_in_file(path, keyword)二选一方案 A（两个细粒度工具）与方案 B（一个analyze_file复合工具），各跑 5 组提问，比较平均调用轮次。
完成动手实践后，进入第 9 章：多模态 Agent。