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

2026-09-27 artificial-intelligence / agentic 1.9K 字 8 分钟

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

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

s02 新增的唯一机制是 工具注册与分发：Agent Loop 不再硬编码调用 run_bash()，而是把模型返回的工具名称和参数交给统一入口 execute_tool()。

新增工具时只需要定义工具并显式注册，不需要修改 Agent Loop。

flowchart LR
  accTitle: s02 工具注册与分发流程
  accDescr: s01 的 Agent Loop 保持不变；模型请求工具时，s02 新增的分发层根据工具名找到对应实现，执行结果追加到消息历史后再次交给模型。

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

  subgraph s02["s02 新增：REGISTRY 分发"]
    direction TB
    dispatcher["execute_tool"]
    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 和命令行入口
├── tools/
│   ├── __init__.py           # 工具包公共接口
│   ├── contract.py           # 工具定义的数据结构与类型契约
│   ├── registry.py           # 工具注册、OpenAI Schema 生成与调用分发
│   ├── workspace.py          # 工作目录配置与安全路径解析
│   ├── shell.py              # bash 命令工具
│   └── filesystem.py         # 文件读取、写入、编辑和 glob 工具
├── tests/
│   ├── test_tools.py
│   └── test_web.py
└── examples/
    └── web/
        ├── app.py            # FastAPI Web 适配示例
        └── agent-chat.html   # 单文件聊天前端

下面逐一看 tools 下的内容。

init

__init__.py 负责把内部模块的三个对象暴露出去：

  • TOOLS：符合 OpenAI function calling 格式的工具 schema 列表，可以直接传给模型。
  • execute_tool：统一工具执行入口，根据模型返回的名称和参数找到工具并执行。
  • WORKDIR：工具使用的工作目录，固定为章节根目录 s02-tool-use/。
"""工具包的公共接口。"""

from .registry import TOOLS, execute_tool
from .workspace import WORKDIR

__all__ = ["TOOLS", "WORKDIR", "execute_tool"]

这样 Agent Loop 只依赖工具包的公共接口，不需要知道具体工具位于哪个模块。

contract

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 字符串由 registry.execute_tool() 负责解析。
  2. handler 是异步函数，返回 Awaitable[str]。同步阻塞操作仍需显式使用 asyncio.to_thread()，仅声明为 async 不会自动消除阻塞。
  3. 返回值是字符串，可以直接作为 role="tool" 消息的内容。可恢复的输入或 I/O 错误返回 Error: ...；编程错误不应被无差别吞掉。

@dataclass(frozen=True) 表示工具定义创建后不可修改，避免运行期间 schema 与 handler 的对应关系发生变化。

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

flowchart LR
  accTitle: ToolDefinition 的两种用途
  accDescr: 工具定义中的名称、描述和参数被转换为 OpenAI schema，handler 则进入本地注册表用于执行。

  definition["ToolDefinition"]
  definition --> public["name / description / parameters"]
  definition --> handler["handler"]
  public --> schema["OpenAI tool schema"]
  schema --> model["发送给模型"]
  handler --> registry["REGISTRY[name]"]
  registry --> execute["本地执行"]

  class schema,registry added
  class handler attention

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

第一步：注册

TOOL_DEFINITIONS = [BASH, *FILESYSTEM_TOOLS]
REGISTRY = _build_registry(TOOL_DEFINITIONS)

_build_registry() 按名称建立索引，并在启动时拒绝重名。

第二步：生成公开 schema

TOOLS: list[ChatCompletionToolParam] = [
    {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.parameters,
        },
    }
    for tool in TOOL_DEFINITIONS
]

这里只提取需要发送给模型的三个字段。TOOLS 最终由 main.py 传给模型。

第三步：分发执行

async def execute_tool(name: str, raw_arguments: str) -> str:
    tool = REGISTRY.get(name)
    ...
    arguments = json.loads(raw_arguments)
    ...
    return await tool.handler(arguments)

模型返回 tool_calls 后，Agent Loop 将工具名和原始参数交给 execute_tool()。分发器找到对应 ToolDefinition，解析参数并执行 handler， 然后 Agent Loop 把结果追加为 role="tool" 消息。

registry

registry.py 负责三件事：

  1. 收集当前启用的工具定义。
  2. 转换出 OpenAI API 需要的 function tool schema。
  3. 根据模型返回的工具名称查找并执行 handler。
from .filesystem import TOOLS as FILESYSTEM_TOOLS
from .shell import TOOL as BASH

TOOL_DEFINITIONS = [BASH, *FILESYSTEM_TOOLS]

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

注册表按名称建立索引：

def _build_registry(
    definitions: list[ToolDefinition],
) -> dict[str, ToolDefinition]:
    registry: dict[str, ToolDefinition] = {}
    for definition in definitions:
        if definition.name in registry:
            raise ValueError(f"Duplicate tool name: {definition.name}")
        registry[definition.name] = definition
    return registry

统一分发入口处理模型输入边界：

async def execute_tool(name: str, raw_arguments: str) -> str:
    tool = 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 和非对象参数属于可恢复的模型输入错误，因此将错误文本返回 给模型，让模型根据结果调整下一步。

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

flowchart LR
  accTitle: 工具调用的注册表分发流程
  accDescr: 模型返回工具名称和 JSON 参数，分发器查询注册表并解析参数，调用对应 handler 后把文本结果回传模型。

  tool_request["tool_call: name + arguments"]
  tool_request --> lookup{"REGISTRY 中存在"}
  lookup -->|否| unknown["Error: Unknown tool"]
  lookup -->|是| parse{"JSON 是对象"}
  parse -->|否| invalid["Error: Invalid arguments"]
  parse -->|是| handler["await tool.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

workspace

workspace.py 统一维护工作目录和文件路径边界：

WORKDIR = Path(__file__).resolve().parents[1]


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

WORKDIR 固定为章节根目录，所以从仓库根目录或章节目录启动，工具作用范围都不会 变化。safe_path() 在路径解析后检查归属，可以阻止 ../ 和符号链接造成的目录 逃逸。

这个限制只覆盖文件工具，不约束 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)

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

filesystem

filesystem.py 提供四个工具：

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

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

文件工具统一使用 safe_path()，但 JSON Schema 不能替代运行时校验。模型生成的 参数仍是不可信输入。