Tool Use

Tool Use:工具定义、注册与分发

从这一章开始,我们会关注一些工程化实现,确保后续工具具有可扩展性。

s01 通过循环连接模型与 Bash 工具。s02 新增的唯一机制是 工具注册与分发: Agent Loop 把模型返回的工具名称和参数交给运行时的 execute() 方法, 由运行时查询注册表,调用对应的工具实现。

本章将通用工具机制、具体能力和应用组装分开,使工具的实现与启用清单可以独立管理。 Agent Loop 通过参数接收工具运行时、模型客户端和模型名称,负责更新消息历史 并返回最终文本。调用方负责创建依赖和管理客户端的生命周期。

这样,同一个循环可以服务于不同的工具集,也可以在测试中使用模拟客户端。 模型决定调用什么工具,程序执行并回填结果;工具请求和结果通过 OpenAI 消息中的 调用标识关联。

flowchart LR
  accTitle: s02 工具注册与分发流程
  accDescr: Agent Loop 把模型生成的工具名称和参数交给工具运行时,运行时查询注册表并执行对应工具,循环将结果追加到消息历史后再次调用模型。

  user["用户提问:messages"] --> llm["调用大模型"]
  llm --> decision{"循环检查 tool_calls"}
  decision -->|否| final(["返回结果"])
  decision -->|是| dispatcher

  subgraph s02["s02:注册与分发"]
    direction LR
    dispatcher["ToolRuntime.execute"]
    dispatcher --> bash_tool["bash"]
    dispatcher --> read_tool["read_file"]
    dispatcher --> write_tool["write_file"]
    dispatcher --> edit_tool["edit_file"]
    dispatcher --> glob_tool["glob"]
  end

  bash_tool --> tool_result["工具结果追加到 messages"]
  read_tool --> tool_result
  write_tool --> tool_result
  edit_tool --> tool_result
  glob_tool --> tool_result
  tool_result --> llm

  class decision attention
  class dispatcher,bash_tool,read_tool,write_tool,edit_tool,glob_tool,tool_result added

项目代码

agentic-s02-tool-use
"""s02: 多工具注册与分发。

相对 s01,Agent Loop 的形状不变;本章只把硬编码的 bash 调用替换为工具分发器,
并增加 read_file、write_file、edit_file 和 glob。

运行:
    在 agentic-s02-tool-use/.env 配置 MODEL_API_KEY、BASE_URL,可选 MODEL
    uv run agentic-s02-tool-use/main.py
"""

from __future__ import annotations

import asyncio
import os
from pathlib import Path

from dotenv import load_dotenv
from openai import AsyncOpenAI
from tools import TOOLS, WORKDIR, execute_tool

ENV_FILE = Path(__file__).resolve().parent / ".env"
load_dotenv(ENV_FILE)

client = AsyncOpenAI(api_key=os.getenv("MODEL_API_KEY"), base_url=os.getenv("BASE_URL"))
MODEL = os.getenv("MODEL", "doubao-seed-evolving")
SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Use tools to solve tasks. Act, don't explain."
)


# 循环调用工具,直到模型不再请求工具
async def agent_loop(messages: list) -> None:
    while True:
        response = await client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS, tool_choice="auto"
        )
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))

        tool_calls = message.tool_calls or []
        if not tool_calls:  # 模型没有调用工具,本轮结束
            return

        for call in tool_calls:
            print(f"\033[33m[tool] {call.function.name} {call.function.arguments}\033[0m")
            output = await execute_tool(
                call.function.name,
                call.function.arguments,
            )
            print(output[:200])
            messages.append(
                {"role": "tool", "tool_call_id": call.id, "content": output}
            )


# 入口
async def main() -> None:
    print("s02: Tool Use")
    print("输入问题回车发送,输入 q 退出。\n")

    messages = [{"role": "system", "content": SYSTEM}]
    while True:
        try:
            query = input("s02 >> ")
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        messages.append({"role": "user", "content": query})
        await agent_loop(messages)
        print(messages[-1].get("content") or "")
        print()


if __name__ == "__main__":
    asyncio.run(main())

目录结构

s02-tool-use/ ├── s02.md # 本章学习笔记 ├── README.md # 运行说明 ├── main.py # Agent Loop 和命令行入口 ├── bootstrap.py # 应用选择工具并组装运行时 ├── workspace.py # 章节工作目录与安全路径解析 ├── tooling/ │ ├── __init__.py # 通用机制包,不执行注册 │ ├── contract.py # ToolDefinition 和 ToolHandler 契约 │ ├── registry.py # 工具索引、查重和 OpenAI Schema 生成 │ └── runtime.py # 工具查找、参数解析与执行分发 ├── tools/ │ ├── __init__.py # 具体工具包,不自动启用工具 │ ├── shell.py # bash 命令工具 │ └── filesystem.py # 文件读取、写入、编辑和 glob 工具 ├── tests/ │ ├── test_tooling.py │ ├── test_tools.py │ ├── test_agent_loop.py │ └── test_web.py └── examples/ └── web/ ├── app.py # FastAPI Web 适配示例 └── agent-chat.html # 单文件聊天前端

工具系统按职责分为三部分:

部分回答的问题负责的内容
tooling/工具系统如何工作契约、注册表、统一执行入口
tools/某个工具如何实现具体能力的 Schema、字段校验和 handler
bootstrap.py这个应用启用哪些工具显式选择工具,创建注册表与运行时

注册与分发是所有工具共用的机制,工具的种类和启用范围则由应用决定。 将它们分开后,注册表可以接收任意符合契约的工具,无需依赖 Bash 或文件工具的 具体模块。应用组装入口同时引用通用机制和具体工具,负责把二者连接起来。

下面从包的导入开始,再看契约、注册和调用数据流。

init

tooling/ 和 tools/ 各有一个 __init__.py,都只保留包说明:

# tooling/__init__.py """通用工具机制;导入包不会加载具体工具或创建注册表。"""
# tools/__init__.py """具体工具实现;声明工具不等于注册,启用清单由 bootstrap.py 决定。"""

Python 导入子模块前会先执行包的 __init__.py。如果在这里创建注册表, 那么导入契约类型也会触发应用组装,连带加载具体工具。 这会让单独使用契约、选择不同工具集或隔离测试变得困难。

包初始化只保留说明,可以让导入与组装相互独立。调用方按需要导入契约或运行时:

from tooling.contract import ToolDefinition from tooling.runtime import ToolRuntime

这些导入不会加载 tools.shell、tools.filesystem,也不会创建注册表。 入口需要工具时,再显式调用 build_runtime()。

导入 bootstrap.py 会加载它引用的具体工具声明,但要调用 build_runtime() 才会创建注册表与运行时。这里区分的是“声明能力”和 “为应用启用能力”。

contract

tooling/contract.py 规定“一个工具到底长什么样”。

from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any ToolHandler = Callable[[dict[str, Any]], Awaitable[str]] @dataclass(frozen=True) class ToolDefinition: name: str description: str parameters: dict[str, Any] handler: ToolHandler

ToolHandler 有三个约束:

  1. 入参是 dict[str, Any]。模型返回的 JSON 字符串由 ToolRuntime.execute() 负责解析;字段类型由具体 handler 检查。
  2. 调用 handler 得到 Awaitable[str],等待后取得文本结果。本章工具使用 async def 实现。同步阻塞操作仍需显式使用 asyncio.to_thread(),仅声明为 async 不会自动消除阻塞。
  3. 返回值是字符串,可以直接作为 role="tool" 消息的内容。可恢复的输入或 I/O 错误返回 Error: ...;编程错误不应被无差别吞掉。

@dataclass(frozen=True) 禁止通过普通赋值重新绑定 name、handler 等字段。 它不是深度不可变:parameters 仍然是字典,内部内容可以被修改, 因此工具声明创建后应当按只读数据使用。

ToolRegistry.schemas() 还会复制 parameters,避免调用方修改生成的 API 请求数据时,连带改动原工具声明。

整个数据流是“同一对象,两种用途”:

flowchart LR
  accTitle: ToolDefinition 的两种用途
  accDescr: 注册表保存完整工具定义;名称、描述和参数用于生成发送给模型的 Schema,执行时从同一个工具定义取出本地 handler。

  definition["ToolDefinition"]
  definition --> public["name / description / parameters"]
  definition --> registry["ToolRegistry 保存完整定义"]
  public --> schema["OpenAI tool schema"]
  schema --> model["发送给模型"]
  registry --> handler["tool.handler"]
  handler --> execute["本地执行"]

  class schema,registry added
  class handler attention

模型只会收到 name、description 和 parameters。handler 不属于 API schema,只保留在本地注册表中。

第一步:注册

应用将选中的 ToolDefinition 传给 ToolRegistry 构造函数。 注册表按名称建立实例自己的索引,并拒绝同一实例内的重名。 具体清单由 bootstrap.py 决定。

第二步:生成公开 schema

ToolRegistry.schemas() 从这个索引中的定义提取公开字段, 包装成 OpenAI Chat Completions 的 function tool 格式。 ToolRuntime.schemas() 暴露这份清单,Agent Loop 将它传给模型。

第三步:分发执行

模型返回 tool_calls 后,Agent Loop 将工具名和原始参数交给 ToolRuntime.execute()。运行时从同一个注册表找到 ToolDefinition, 解析参数并执行 handler,然后 Agent Loop 把结果追加为 role="tool" 消息。

公开清单和执行清单共用同一个索引,避免维护两份列表时出现“模型看得到, 本地却没有对应实现”的问题。

registry

tooling/registry.py 负责注册表机制:

  1. 接收应用提供的工具定义,建立名称索引并检查重复。
  2. 根据名称查找 ToolDefinition。
  3. 生成 OpenAI API 需要的 function tool schema。

它不导入具体工具,也不解析模型参数或执行 handler:

from collections.abc import Iterable from copy import deepcopy from openai.types.chat import ChatCompletionToolParam from .contract import ToolDefinition class ToolRegistry: def __init__(self, definitions: Iterable[ToolDefinition]) -> None: self._tools: dict[str, ToolDefinition] = {} for tool in definitions: if tool.name in self._tools: raise ValueError(f"Duplicate tool name: {tool.name}") self._tools[tool.name] = tool def get(self, name: str) -> ToolDefinition | None: return self._tools.get(name) def schemas(self) -> list[ChatCompletionToolParam]: return [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": deepcopy(tool.parameters), }, } for tool in self._tools.values() ]

每个 ToolRegistry 实例通过自己的 _tools 保存工具索引。 两个注册表可以选择不同的工具集,也可以为同一个名字绑定不同的工具定义。

这种隔离针对工具索引。构造函数保存传入的 ToolDefinition 引用, 不会为每个实例复制 handler 或创建独立文件系统。默认工具共享本章的 WORKDIR,不能把多个运行时理解为多个沙箱。

deepcopy() 用来隔离返回的嵌套 Schema 数据,成本是生成 Schema 时多一次复制。 本章注册五个工具,Schema 结构也很小,这个成本换来了更明确的数据边界。

bootstrap

bootstrap.py 是应用组装入口,负责选择应用启用的工具,并创建注册表与运行时:

from tooling.registry import ToolRegistry from tooling.runtime import ToolRuntime from tools.filesystem import EDIT_FILE, GLOB, READ_FILE, WRITE_FILE from tools.shell import BASH def build_runtime() -> ToolRuntime: registry = ToolRegistry([BASH, READ_FILE, WRITE_FILE, EDIT_FILE, GLOB]) return ToolRuntime(registry)

这里采用显式注册,不使用自动扫描。文件中存在一个工具,不代表它会自动暴露给 模型;只有传入注册表的工具才会被启用。

例如,只提供文件读取和查找能力,可以组装另一个运行时:

from tooling.registry import ToolRegistry from tooling.runtime import ToolRuntime from tools.filesystem import GLOB, READ_FILE read_only_runtime = ToolRuntime(ToolRegistry([READ_FILE, GLOB]))

向这个运行时请求 write_file,会得到 Error: Unknown tool: write_file。 虽然 Python 进程里已经加载了写文件函数,它并没有进入这个运行时的可调用清单。 这是能力选择,还不是 s03 那样按一次调用的参数判断权限。

依赖关系如下,箭头表示“导入或依赖”,不表示运行时执行顺序:

flowchart LR
  accTitle: 工具机制、实现与应用组装的依赖关系
  accDescr: bootstrap 同时依赖具体工具和通用机制;具体工具与通用机制依赖同一个工具契约,通用机制不导入具体工具。

  bootstrap["bootstrap.py"]
  bootstrap --> concrete["tools:具体能力"]
  bootstrap --> mechanism["tooling:注册与运行"]
  concrete --> contract["tooling.contract"]
  mechanism --> contract

  class bootstrap added

具体工具的 Schema 和 handler 放在所属的能力模块内。它们共同描述一项能力, 修改参数时可以同时核对声明与实现;是否启用则交给应用组装入口决定。

显式组装要求入口创建并传递运行时实例,代价是多一层组装函数和依赖参数。 收益是工具集可以独立选择,注册与执行机制也可以用测试工具单独验证。 扩展同类工具时,在具体能力模块中定义工具,再将它加入组装清单即可。

runtime

tooling/runtime.py 负责一次工具调用的执行过程。 它通过构造函数接收注册表,既不导入具体工具,也不自己决定启用清单:

import json from openai.types.chat import ChatCompletionToolParam from .registry import ToolRegistry class ToolRuntime: def __init__(self, registry: ToolRegistry) -> None: self._registry = registry def schemas(self) -> list[ChatCompletionToolParam]: return self._registry.schemas() async def execute(self, name: str, raw_arguments: str) -> str: tool = self._registry.get(name) if tool is None: return f"Error: Unknown tool: {name}" try: arguments = json.loads(raw_arguments) except json.JSONDecodeError as exc: return f"Error: Invalid arguments: {exc}" if not isinstance(arguments, dict): return "Error: arguments must be an object" return await tool.handler(arguments)

未知工具、非法 JSON 和非对象参数属于可恢复的模型输入错误,因此将错误文本返回 给模型,让模型根据结果调整下一步。

json.loads() 成功不代表参数可执行:[]、null、123 都是合法 JSON, 但都不是 handler 契约要求的对象。即使已经是对象,具体字段仍需继续校验。

运行时这里只捕获 JSON 解码错误。handler 中没有被具体工具处理的异常继续抛出, 例如测试中的 RuntimeError,不会被统一改写成“模型参数错误”。

一次完整调用的数据流如下:

flowchart LR
  accTitle: 工具调用的注册表分发流程
  accDescr: 工具运行时查询实例注册表并解析 JSON 参数;未知工具和参数错误返回文本,成功查找后调用 handler,由 Agent Loop 将结果关联原调用并交给模型。

  tool_request["tool_call: name + arguments"]
  tool_request --> lookup{"注册表中存在"}
  lookup -->|否| unknown["Error: Unknown tool"]
  lookup -->|是| parse{"JSON 是对象"}
  parse -->|否| invalid["返回对应参数错误"]
  parse -->|是| handler["handler 校验并执行"]
  handler --> output["stdout / stderr 或工具结果"]
  unknown --> result["循环追加 role=tool"]
  invalid --> result
  output --> result
  result --> model["模型继续推理"]

  class handler,output,result added
  class lookup,parse,unknown,invalid attention

Agent Loop

循环通过参数接收依赖,客户端与运行时由调用方创建。 这样 CLI 和 Web 可以共用循环,并分别管理连接的创建与关闭。 下面是 main.py 中的核心逻辑,省略终端日志:

from typing import cast from openai import AsyncOpenAI, omit from openai.types.chat import ( ChatCompletionAssistantMessageParam, ChatCompletionMessageParam, ) from tooling.runtime import ToolRuntime async def agent_loop( messages: list[ChatCompletionMessageParam], runtime: ToolRuntime, *, client: AsyncOpenAI, model: str, ) -> str: schemas = runtime.schemas() while True: response = await client.chat.completions.create( model=model, messages=messages, tools=schemas or omit, tool_choice="auto" if schemas else omit, ) message = response.choices[0].message messages.append( cast( ChatCompletionAssistantMessageParam, message.model_dump(exclude_none=True), ) ) tool_calls = message.tool_calls or [] if not tool_calls: return message.content or "" for call in tool_calls: if call.type != "function": raise ValueError(f"Unsupported tool call type: {call.type}") output = await runtime.execute( call.function.name, call.function.arguments, ) messages.append( {"role": "tool", "tool_call_id": call.id, "content": output} )

这里有几个需要分清的细节:

  1. messages 是调用方传入的列表,循环直接向它追加消息,返回值则是最终文本。 工具运行时不负责保存对话历史。
  2. schemas 来自传入的运行时,在进入 while 前生成一次。空注册表使用 SDK 的 omit 省略请求字段,不发送空的工具清单或 null。
  3. tool_choice="auto" 允许模型决定是否调用工具,不代表一定会调用。 本章只暴露 function tools,代码对其他调用类型直接报错。
  4. model_dump(exclude_none=True) 把 SDK 响应序列化为消息字典; cast() 只向类型检查器说明用途,不做运行时转换或参数校验。
  5. 同一个响应中的调用通过 for 与 await 逐个执行。 使用异步函数不意味着这些调用已经并行。
  6. 每个调用的结果都使用该次调用的 call.id 回填。未知工具或参数错误同样产生 tool 消息,不能因为执行失败就跳过配对。

上游使用 Anthropic 的 tool_use / tool_result 内容块。本地使用 OpenAI-compatible Chat Completions:调用位于 assistant 的 tool_calls, 结果是独立的 role="tool" 消息,参数 function.arguments 是 JSON 字符串。

以“读取 README 的第一行”为例,下面是一次示意消息轨迹,不是真实模型运行记录:

[ { "role": "user", "content": "读取 README.md 的第一行" }, { "role": "assistant", "tool_calls": [ { "id": "call_read_1", "type": "function", "function": { "name": "read_file", "arguments": "{\"path\":\"README.md\",\"limit\":1}" } } ] }, { "role": "tool", "tool_call_id": "call_read_1", "content": "# s02: Tool Use" }, { "role": "assistant", "content": "README 的第一行是:# s02: Tool Use" } ]

先记录 assistant 发出的调用,再记录对应工具结果,然后继续请求模型。 模型只生成调用请求,真正的文件读取发生在本地 handler 中。

workspace

workspace.py 位于章节根目录,统一维护工作目录和文件路径边界:

from pathlib import Path WORKDIR = Path(__file__).resolve().parent def safe_path(path: str) -> Path: resolved = (WORKDIR / path).resolve() if not resolved.is_relative_to(WORKDIR): raise ValueError(f"Path escapes workspace: {path}") return resolved

__file__ 指向 workspace.py,.resolve() 得到它的绝对路径, .parent 取得文件所在的章节根目录。 工作区由文件位置确定,所以从仓库根目录或章节目录启动,工具作用范围都不会变化。

safe_path() 检查的是解析后的路径归属,而不是文本中是否包含 ../。 工作目录内的绝对路径可以通过,解析到外部的相对路径和符号链接会被拒绝。

read_file、write_file 和 edit_file 在实际读写前调用它。 glob 使用单独的匹配结果过滤,下面会说明两者的差别。 这个路径检查不是操作系统沙箱,不约束 Bash,也不保证检查与实际访问之间 不会发生外部文件系统变化。

shell

shell.py 定义 Bash 工具的 schema 和 handler。subprocess.run() 是同步阻塞 调用,因此通过 asyncio.to_thread() 执行,避免阻塞 Agent 的事件循环:

async def handler(arguments: dict[str, Any]) -> str: command = arguments.get("command") if not isinstance(command, str): return "Error: command must be a string" return await asyncio.to_thread(_run, command)

Bash 的字符串黑名单只能拦截少量明显危险命令,不构成权限系统。正式权限治理属于 s03。

_run() 使用 shell=True 和本章 WORKDIR 执行命令,超时为 120 秒。 check=False 表示非零退出码不会自动抛出异常;返回结果是 stdout 与 stderr 拼接后的文本,没有单独返回退出码,也不保留两路输出的交错顺序。

返回文本最多保留 50,000 个字符,终端日志只显示其中前 200 个字符。 这是返回和显示限制,不是子进程输出的内存上限:capture_output=True 仍会先收集输出。

filesystem

filesystem.py 提供四个工具:

工具作用
read_file读取 UTF-8 文件,可限制返回行数
write_file写入文件,并按需创建父目录
edit_file精确替换第一次出现的文本
glob按 glob 模式查找文件,** 表示递归

每个工具都包含 handler 和对应的 ToolDefinition。handler 先检查参数,再通过 asyncio.to_thread() 执行同步文件操作。

以 read_file 为例,字段校验发生在 handler 中:

async def read_file(arguments: dict[str, Any]) -> str: path = arguments.get("path") limit = arguments.get("limit") if not isinstance(path, str) or not path: return "Error: path must be a non-empty string" if limit is not None and ( not isinstance(limit, int) or isinstance(limit, bool) or limit <= 0 ): return "Error: limit must be a positive integer" return await asyncio.to_thread(_read, path, limit)

Python 的 bool 是 int 的子类,所以只写 isinstance(limit, int) 会把 True 也当成整数接受。这里显式排除了布尔值。

同一模块再把这个函数绑定到工具声明上:

READ_FILE = ToolDefinition( name="read_file", description="Read UTF-8 file contents.", parameters={ "type": "object", "properties": { "path": {"type": "string"}, "limit": {"type": "integer", "minimum": 1}, }, "required": ["path"], "additionalProperties": False, }, handler=read_file, )

handler=read_file 保存的是函数引用,没有在声明时读取文件。 实际调用要等应用注册了 READ_FILE,并且运行时收到相应工具请求后才发生。

JSON Schema 不能替代本地校验。本章运行时没有完整的 JSON Schema 校验器: 它检查 JSON 是否为对象,handler 再检查自己使用的字段。 例如 Schema 声明了 additionalProperties: False,但 handler 不统一检查多余键, 额外字段可能被忽略。因此,Schema 的声明约束与本地字段校验的覆盖范围并不相同, 模型生成的参数仍需视为不可信输入。

几个实现边界也要区分:

  • read_file 先读完整文件再切分行数,limit 限制返回内容,不限制文件读取量。
  • write_file 会按需创建父目录,并覆盖目标文件;返回信息统计的是字符数。
  • edit_file 替换第一次匹配,不检查旧文本是否只出现一次。
  • glob 不调用 safe_path(),而是先匹配,再过滤解析后不属于 WORKDIR 的结果。 这保证了返回清单中的路径归属,但不等于禁止了工作区外的扫描。 最多展示 200 条匹配也是返回限制,不是扫描数量上限。

CLI 和 Web 的生命周期

CLI 在启动时调用 build_runtime(),再通过 async with create_client() 创建并管理模型客户端。多轮提问复用这个运行时、客户端和进程内的 messages; 退出时关闭客户端连接。

Web 在 FastAPI 的 lifespan 中创建自己的运行时和客户端,服务关闭时释放连接。 运行时可以在请求间复用,但每次请求都会创建新的 messages 列表: 浏览器发送完整的 user/assistant 历史,服务端在前面补充 system 消息。

Web 示例没有服务端 Session。一次请求内部产生的工具调用和工具结果会用于 该次 Agent Loop,但响应只返回最终 assistant 文本,浏览器不会保存完整工具轨迹。 因此下一次请求带回的是对话文本历史,而不是上一轮的全部内部消息。

main.py 在导入时通过自身位置加载同目录 .env,已有进程环境变量优先; 导入本身不创建模型客户端或工具运行时。.env 的加载路径和工具 WORKDIR 是两个用途不同的配置,它们在本章都落在章节根目录。

行为验证

本章的 19 个测试验证工具隔离、调用协议和资源生命周期等设计约束:

测试文件验证内容
test_tooling.py注册表隔离、重复名称、未启用工具、参数解析、Schema 复制、异常传播和导入边界
test_tools.py五个工具的默认暴露、文件读写编辑、glob、路径越界、符号链接和 Bash 工作目录
test_agent_loop.py注入自定义工具后的顺序执行、错误结果配对、最终文本返回和空注册表
test_web.py客户端历史传递、请求间历史隔离、运行时复用和客户端关闭

模型响应通过本地 HTTP 模拟器提供。这些测试验证程序的分发和协议处理, 不证明真实模型一定会选择正确工具,也不构成完整的安全验证。 运行与检查命令见工程中的 README.md。