4.1 工具即函数:schema 生成、注册表与分发
前三章里 minagent 已经能"想"(模型)也能"停"(停止条件),但它还只会说、不会做。工具(tool)就是 Agent 的手:模型从不直接执行任何东西,它只产出一个结构化的"调用意图",真正动手的是我们自己写的 Python 函数。
本节把 minagent 推进到
minagent/tools.py:定义Tool与ToolRegistry,让"一个 Python 函数 + 一段 JSON Schema"变成模型可调用的工具,并跑通第一次按名分发。
4.1.1 Tool:把函数包成模型看得懂的东西
模型不认识 Python 函数,也不认识类型注解。它只认一段结构化的描述:这个工具叫什么、干什么用、需要哪些参数、每个参数是什么类型。这段描述的事实标准是 JSON Schema。
所以 Tool 这个 dataclass 要做的事很朴素——把"函数 + 元信息"绑成一个对象,并提供一个 schema() 方法,把元信息翻译成供应商 API 要的形态:
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any, Callable
@dataclass
class ToolCall:
id: str
name: str
arguments: dict[str, Any]
@dataclass
class Tool:
name: str
description: str
parameters: dict[str, Any] # JSON Schema
func: Callable[..., Any]
def schema(self) -> dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"input_schema": self.parameters,
}
input_schema 这个名字不是随便起的。Anthropic 的 Messages API 里,工具定义的字段就叫 input_schema(OpenAI 那边叫 parameters),两者内容都是同一个 JSON Schema 对象。把供应商差异挡在 schema() 里,上层循环就永远只看到一种形状。
三个字段各司其职,缺一不可:
| 字段 | 给谁看 | 作用 |
|---|---|---|
name | 模型 | 调用的唯一标识,也是注册表的键 |
description | 模型 | 决定"什么时候该用它",是提示词的一部分 |
input_schema | 模型 | 决定"参数怎么填",模型据此生成 arguments |
name 和 input_schema 是给运行时用的,description 却纯粹是给模型看的自然语言——它写得好不好,直接决定模型选不选得对工具。
ToolRegistry 是工具的容器,接口只有三个动作:加进去、把全部 schema 吐给模型、按 ToolCall 分发执行。
class ToolRegistry:
def __init__(self) -> None:
self._tools: dict[str, Tool] = {}
def add(self, tool: Tool) -> None:
self._tools[tool.name] = tool
def schemas(self) -> list[dict[str, Any]]:
return [t.schema() for t in self._tools.values()]
def call(self, tc: ToolCall) -> str:
tool = self._tools.get(tc.name)
if tool is None:
return json.dumps({"error": f"unknown tool: {tc.name}"}, ensure_ascii=False)
try:
result = tool.func(**tc.arguments)
except Exception as e:
return json.dumps({"error": f"{type(e).__name__}: {e}"}, ensure_ascii=False)
if isinstance(result, str):
return result
return json.dumps(result, ensure_ascii=False)
注意 call() 的返回值永远是字符串,且永远不抛异常。这两个约定后面会反复用到:字符串是为了能直接塞进 tool 角色的消息;不抛异常是为了让"工具出错"变成模型可读的一条反馈,而不是把整个循环打断。
用 dict 按名索引而不是遍历列表,是因为分发是热路径:每轮循环可能调用多个工具,O(1) 查找比 O(n) 扫描划算,也让"重名覆盖"成为天然行为——后注册的同名工具会覆盖先前的,方便测试时用桩函数替换真实实现。错误统一用 {"error": ...} 这个键名,是为了让模型和上层代码都只需判断一个字段就能区分成功与失败。
把这两个小类放进 minagent/tools.py,我们的运行时就有了"手"。接下来要解决的是:模型填的参数可能根本不对,怎么办。
4.1.2 两种 schema 生成方式
现在的问题是:parameters 那段 JSON Schema 从哪来?有两条路。
| 做法 | 写法 | 优点 | 缺点 |
|---|---|---|---|
| 手写 schema | 直接敲 dict 字面量 | 完全可控,能写复杂约束 | 和函数签名容易脱节,改一处忘一处 |
| 从类型注解推导 | 读 __annotations__ / get_type_hints | 单一事实来源,改签名即改 schema | 极简版处理不了默认值、嵌套类型 |
推导需要一个"Python 类型 → JSON Schema 类型"的映射表,下表就是极简版支持的全部范围:
| Python 注解 | JSON Schema type | 说明 |
|---|---|---|
str | string | |
int | integer | |
float | number | |
bool | boolean | 注意 bool 是 int 的子类,映射表要先查 bool |
| 其它 | string | 兜底,避免直接崩溃 |
真实场景的类型远不止这四个:list[str]、dict[str, int]、Literal[...]、可选参数都要各自处理。手工维护这张表正是"自己造轮子"的成本所在。
手写版长这样,一眼能看懂,但一旦函数签名改了,这段 dict 不会自己跟着变:
def add(a: int, b: int) -> int:
"""两数相加。"""
return a + b
ADD_SCHEMA = {
"type": "object",
"properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
"required": ["a", "b"],
"additionalProperties": False,
}
推导版用 typing.get_type_hints 读函数签名。下面是教学用的极简版,只做 Python 内置类型到 JSON Schema 类型的一对一映射——真实工程里这活交给 pydantic,4.2 节就见:
from typing import get_type_hints
_PY_TO_JSON = {str: "string", int: "integer", float: "number", bool: "boolean"}
def schema_from_func(func: Callable[..., Any]) -> dict[str, Any]:
hints = get_type_hints(func)
props: dict[str, Any] = {}
required: list[str] = []
for name, hint in hints.items():
if name == "return":
continue
props[name] = {"type": _PY_TO_JSON.get(hint, "string")}
required.append(name)
return {
"type": "object",
"properties": props,
"required": required,
"additionalProperties": False,
}
把两个工具注册进去,真跑一遍看生成的 schema:
def search_notes(keyword: str, limit: int = 5) -> list[str]:
"""按关键词检索笔记。"""
return [f"note-{i}: {keyword}" for i in range(limit)]
reg = ToolRegistry()
reg.add(Tool(name="add", description="两数相加",
parameters=schema_from_func(add), func=add))
reg.add(Tool(name="search_notes", description="按关键词检索笔记",
parameters=schema_from_func(search_notes), func=search_notes))
print(json.dumps(reg.schemas(), ensure_ascii=False, indent=2))
输出(实测,Python 3.14.6):
[
{
"name": "add",
"description": "两数相加",
"input_schema": {
"type": "object",
"properties": { "a": { "type": "integer" }, "b": { "type": "integer" } },
"required": ["a", "b"],
"additionalProperties": false
}
},
{
"name": "search_notes",
"description": "按关键词检索笔记",
"input_schema": {
"type": "object",
"properties": {
"keyword": { "type": "string" },
"limit": { "type": "integer" }
},
"required": ["keyword", "limit"],
"additionalProperties": false
}
}
]
这里藏着一个极简版的缺陷:search_notes 的 limit 有默认值 5,但推导版把它也标成了 required。这正是"自己写推导器"会踩的坑——get_type_hints 只告诉你类型,不告诉你默认值,要正确处理得再读 inspect.signature(func).parameters[name].default。这也是为什么工业界直接上 pydantic:它把默认值、可选性、嵌套模型、约束校验一次性解决。
4.1.3 分发:未知工具返回 error,而不是抛异常
模型是概率性的,它完全可能"幻觉"出一个根本不存在的工具名。如果 call() 遇到未知名字直接 raise KeyError,整个 Agent 循环就崩了。正确做法是把它当成一次可恢复的错误,包成 JSON 回灌:
print(reg.call(ToolCall(id="c1", name="add", arguments={"a": 2, "b": 3})))
print(reg.call(ToolCall(id="c2", name="divide", arguments={"a": 1, "b": 0})))
实测输出:
5
{"error": "unknown tool: divide"}
add 返回的是 5——注意它是 json.dumps(5) 的结果,一个字符串 "5"。divide 不在注册表里,于是返回 {"error": "unknown tool: divide"}。模型下一轮会看到这条 tool 消息,自己决定是换一个工具、还是换个参数再来。
4.1.4 分发只做转发,不做语义
call() 的职责边界要划清楚:它只负责"按名找到函数、把参数展开、把结果序列化",不负责判断这个调用是否合理。参数类型对不对是 4.2 节的校验层的事,工具名对不对是模型自己的事。分发层越薄越好。
再跑一次带默认值的工具,确认返回值的形状:
print(reg.call(ToolCall(id="c3", name="search_notes", arguments={"keyword": "rag"})))
实测输出:
["note-0: rag", "note-1: rag", "note-2: rag", "note-3: rag", "note-4: rag"]
返回的是 json.dumps 后的字符串,不是 Python list——call() 的返回类型永远是 str,这一点在第 3 章的循环里就约定好了。模型拿到这段 JSON 字符串,能直接当文本读,也能自己解析。
真实模型下会怎样:假模型(FakeModel)里工具名和参数都是我们脚本写死的,不会出现"幻觉工具名"。真实 Claude 在拿到 schemas() 后,绝大多数时候会挑正确的工具,但确实会在工具描述写得含糊时调用错工具、或编出名字相近的工具。所以"未知工具返回 error 而非抛异常"在真实场景里不是防御性编程的洁癖,而是线上稳态的必要条件——一次坏调用只应消耗一轮,不应终止整个会话。另一个真实差异是:真实模型会同时发出多个 tool_call(4.3 节的主题),而我们的分发层一次只处理一个。
阅读导航:上一节:3.3 抽成 Agent 类:状态、步骤与回调 · 下一节:4.2 参数校验与执行隔离 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。