程云来 / 杭州
返回文章·Agent 与 Harness
·约 9 分钟

实践|Tool Use,把工具接进同一个循环

展开 build_tools、execute_tool、read_file 的完整实现,用断点驱动注册、参数拒绝、路径检查与结果关联;沿用第一课循环运行两种工具。


先读,再运行:该改循环的哪一处?

先读 main.py,再运行它,报告文件内容和运行结果。

文件内容是 print(6 * 7)。这次既要取得源文本,也要取得退出码和输出。思想篇已经拆开了工具说明与本地实现。现在落到开发上:保留第一课负责消息与停止的 run_loop,把传入的执行函数换成通用分发器。

本课实现由本册独立编写,放在公开仓库的 examples/s02-tool-use/。新增职责集中在三个函数:

  1. build_tools(workspace) 返回同一份注册表,连接工具说明、参数形状与本地函数。
  2. execute_tool(call, tools) 接收模型请求,校验并调用注册项,返回关联结果。
  3. read_file(workspace, path) 处理文件路径、大小与 UTF-8 读取;已有程序执行继续复用第一课。

下面各函数的正文从公开源文件按符号提取。代码与图共用已有调试游标:点行号设断点,再点“调试”,用“下一步”或“继续”观察真实执行轨迹。浏览器中的文件和进程使用可展开查看的教学替身;下载仓库后的默认入口会真实读取文件并启动 Python,模型仍使用离线固定响应。

build_tools:只维护一份工具注册

先决定工具需要什么信息。本课只用必填字符串参数和空参数,因此每项包含 description、parameters、run 三项。parameters 保存字段名与 Python 类型,run 保存可调用对象。

workspace 在建立注册时由宿主传入,闭包把它绑定到实现。模型只能提供 path,不能用另一个参数替换工作目录。创建闭包时不会读取文件;执行发生在分发器真正调用 run 时。

真实执行轨迹回放;文件与进程使用公开教学替身,不访问实际文件、不启动进程。

Pys02/core/runtime.py · build_tools
未运行
def build_tools(workspace):
    """每项注册同时持有说明、参数形状和实现,供定义生成与分发共用。"""
    return {
        "read_file": {
            "description": "读取工作目录内最多 4096 字节的 UTF-8 文件,不修改文件。",
            "parameters": {"path": str},
            "run": lambda path: read_file(workspace, path),
        },
        "run_main": {
            "description": "运行工作目录的 main.py,返回退出码和输出,不接受参数。",
            "parameters": {},
            "run": lambda: _run_main(workspace),
        },
    }
输出

点击行号设置断点,再点“调试”;也可以直接运行。

read_filerun_mainNAME → INPUT → RUNNAME / SCHEMAdescriptionFILE READERpath → text工具说明一份注册本地实现
代码只读;执行发生在浏览器中,不会发送到博客服务器。
教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。

import io
from pathlib import PurePosixPath
_files = {"/teaching/main.py": b"print(6 * 7)\n", "/teaching/large": b"x" * 4097}
_reads = []
_runs = []
class _Path:
    def __init__(self, path):
        self.path = PurePosixPath(path)
    def resolve(self):
        parts = []
        for part in self.path.parts:
            if part == "..":
                if parts: parts.pop()
            elif part not in ("/", "."): parts.append(part)
        return _Path("/" + "/".join(parts))
    def __truediv__(self, path):
        return _Path(self.path / path)
    def is_relative_to(self, root):
        return self.path.is_relative_to(root.path)
    def open(self, mode):
        assert mode == "rb"
        _reads.append(str(self.path))
        return io.BytesIO(_files[str(self.path)])
workspace = _Path("/teaching")
def _run_main(workspace):
    _runs.append("main.py")
    return "exit_code=0\n42"

def read_file(workspace, path):
    """仅允许工作目录内的短 UTF-8 文件;路径约束不是进程沙箱。"""
    root = workspace.resolve()
    target = (root / path).resolve()
    if not target.is_relative_to(root):
        raise ValueError("路径越出工作目录。")
    with target.open("rb") as stream:
        data = stream.read(4097)
    if len(data) > 4096:
        raise ValueError("文件超过 4096 字节,请缩小读取范围。")
    return data.decode("utf-8")


# 调用入口
outcome = build_tools(workspace)
print("已注册:", ", ".join(outcome))
  1. read_file 注册用途、path 字符串参数和读取入口。
  2. run_main 注册用途、空参数和运行入口。
  3. 返回字典;尚未执行工具。模型说明从同一份注册生成。

在 "run": lambda path: ... 处停下,再走到函数末尾。返回注册表只完成了能力装配,既没有运行 read_file,也没有启动 main.py。run_main 对应的 _run_main 是一个薄适配:调用第一课的运行函数,将失败转换为异常供分发层统一封装。

要把注册表交给模型,还需要剥离 Python 函数对象。实际代码用 tool_definitions 生成 API 所需的说明:

python
def tool_definitions(tools):
    """本课只支持必填字符串参数;这是小型契约,不是通用 JSON Schema 校验器。"""
    return [{
        "name": name, "description": spec["description"],
        "input_schema": {
            "type": "object",
            "properties": {key: {"type": "string"} for key in spec["parameters"]},
            "required": list(spec["parameters"]), "additionalProperties": False,
        },
    } for name, spec in tools.items()]

这里不是一套通用 Schema 框架:生成器只处理本课的“必填字符串参数”。新增可选值、数字或嵌套对象时,定义生成和本地校验需要一起扩展。当前把名称和参数从同一份注册提取,能避免在两张表里分别改名后遗漏一处。

execute_tool:先验证,再调用

循环将完整的 tool_use 交给 execute_tool。其中 id 已由循环检查;分发层先把它放进 tool_result.tool_use_id,然后查名称、取参数规则。成功、拒绝、执行失败都沿用这个关联 ID。

参数检查分两次:字段集合必须相等,随后每个值的类型必须相符。只做 handler(**input) 会让错误字段直接进入调用,错误也难以归到明确的契约检查。这里把检查放在实际实现之前。

真实执行轨迹回放;文件与进程使用公开教学替身,不访问实际文件、不启动进程。

Pys02/core/runtime.py · execute_tool
未运行
def execute_tool(call, tools):
    """循环已检查调用 ID;这里查名称、验参数、执行并封装关联结果。"""
    result = {"type": "tool_result", "tool_use_id": call["id"]}
    name, arguments = call.get("name"), call.get("input")
    if not isinstance(name, str) or name not in tools:
        return {**result, "content": "未知工具。", "is_error": True}
    spec = tools[name]
    parameters = spec["parameters"]
    if not isinstance(arguments, dict) or set(arguments) != set(parameters):
        return {**result, "content": "参数字段不匹配。", "is_error": True}
    if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
        return {**result, "content": "参数类型不匹配。", "is_error": True}
    try:
        output = spec["run"](**arguments)
        return {**result, "content": output, "is_error": False}
    except (ValueError, OSError) as error:
        return {**result, "content": str(error), "is_error": True}
输出

点击行号设置断点,再点“调试”;也可以直接运行。

TOOL_USEread_fileid: call_1read_filerun_mainNAME → INPUT → RUNFILE READERpath → textTOOL_RESULTtool_use_id请求分发与校验实现函数关联结果
代码只读;执行发生在浏览器中,不会发送到博客服务器。
教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。

import io
from pathlib import PurePosixPath
_files = {"/teaching/main.py": b"print(6 * 7)\n", "/teaching/large": b"x" * 4097}
_reads = []
_runs = []
class _Path:
    def __init__(self, path):
        self.path = PurePosixPath(path)
    def resolve(self):
        parts = []
        for part in self.path.parts:
            if part == "..":
                if parts: parts.pop()
            elif part not in ("/", "."): parts.append(part)
        return _Path("/" + "/".join(parts))
    def __truediv__(self, path):
        return _Path(self.path / path)
    def is_relative_to(self, root):
        return self.path.is_relative_to(root.path)
    def open(self, mode):
        assert mode == "rb"
        _reads.append(str(self.path))
        return io.BytesIO(_files[str(self.path)])
workspace = _Path("/teaching")
def _run_main(workspace):
    _runs.append("main.py")
    return "exit_code=0\n42"

def read_file(workspace, path):
    """仅允许工作目录内的短 UTF-8 文件;路径约束不是进程沙箱。"""
    root = workspace.resolve()
    target = (root / path).resolve()
    if not target.is_relative_to(root):
        raise ValueError("路径越出工作目录。")
    with target.open("rb") as stream:
        data = stream.read(4097)
    if len(data) > 4096:
        raise ValueError("文件超过 4096 字节,请缩小读取范围。")
    return data.decode("utf-8")

def build_tools(workspace):
    """每项注册同时持有说明、参数形状和实现,供定义生成与分发共用。"""
    return {
        "read_file": {
            "description": "读取工作目录内最多 4096 字节的 UTF-8 文件,不修改文件。",
            "parameters": {"path": str},
            "run": lambda path: read_file(workspace, path),
        },
        "run_main": {
            "description": "运行工作目录的 main.py,返回退出码和输出,不接受参数。",
            "parameters": {},
            "run": lambda: _run_main(workspace),
        },
    }


# 调用入口
outcome = execute_tool({"id": "call_1", "name": 'read_file', "input": {'path': 'main.py'}}, build_tools(workspace))
print("call_1", "is_error=" + str(outcome["is_error"]))
print(outcome["content"])
  1. 建立 tool_result 并保留 call.id。
  2. 查名称,找不到就返回错误。
  3. 验证字段集合和字段类型;不匹配就拒绝调用。
  4. 调用注册项的 run,取得输出或异常。
  5. 返回 content、tool_use_id、is_error,由循环保存。

先选“读取文件”,在 output = spec["run"](**arguments) 停下。此时已经通过名称和参数检查,但实现尚未执行。下一步取得输出,随后才返回 tool_result;写入 messages 仍由外层循环负责。

再分别选择“未知工具”和“参数错误”。前者找不到注册项;后者给 path 传入数字。两种情况都直接返回 is_error=True,不会触达读文件或运行程序的函数。这个事实在公开测试中通过“处理函数未被调用”的断言核对,而不只看错误文字。

本课将预期的文件与运行错误转换为工具反馈。编程错误不在捕获范围内,例如错误注册导致的 TypeError 会直接暴露,便于开发者修复;不能把所有异常都写成模型可以自行修正的输入问题。

read_file:字符串是对的,路径还可能不对

path 是字符串,只能说明参数类型合法。../secret.txt 也是字符串,但已经越出指定工作目录。读取函数先解析工作目录,再解析目标路径,检查目标是否仍位于目录内,通过后才打开文件。

这个顺序也覆盖了已有符号链接指向目录外的情况。随后以二进制最多读取 4097 字节,用多出的一个字节判断是否超过 4096 字节上限,再按 UTF-8 解码。这里选择拒绝大文件,避免模型把截断内容误当完整文件。

真实执行轨迹回放;文件与进程使用公开教学替身,不访问实际文件、不启动进程。

Pys02/core/runtime.py · read_file
未运行
def read_file(workspace, path):
    """仅允许工作目录内的短 UTF-8 文件;路径约束不是进程沙箱。"""
    root = workspace.resolve()
    target = (root / path).resolve()
    if not target.is_relative_to(root):
        raise ValueError("路径越出工作目录。")
    with target.open("rb") as stream:
        data = stream.read(4097)
    if len(data) > 4096:
        raise ValueError("文件超过 4096 字节,请缩小读取范围。")
    return data.decode("utf-8")
输出

点击行号设置断点,再点“调试”;也可以直接运行。

FILE PATHmain.pyresolve firstWORKSPACE / UTF-8UTF-8 TEXTmax 4096 bytes文件路径工作目录读取与解码
代码只读;执行发生在浏览器中,不会发送到博客服务器。
教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。

import io
from pathlib import PurePosixPath
_files = {"/teaching/main.py": b"print(6 * 7)\n", "/teaching/large": b"x" * 4097}
_reads = []
_runs = []
class _Path:
    def __init__(self, path):
        self.path = PurePosixPath(path)
    def resolve(self):
        parts = []
        for part in self.path.parts:
            if part == "..":
                if parts: parts.pop()
            elif part not in ("/", "."): parts.append(part)
        return _Path("/" + "/".join(parts))
    def __truediv__(self, path):
        return _Path(self.path / path)
    def is_relative_to(self, root):
        return self.path.is_relative_to(root.path)
    def open(self, mode):
        assert mode == "rb"
        _reads.append(str(self.path))
        return io.BytesIO(_files[str(self.path)])
workspace = _Path("/teaching")
def _run_main(workspace):
    _runs.append("main.py")
    return "exit_code=0\n42"

def build_tools(workspace):
    """每项注册同时持有说明、参数形状和实现,供定义生成与分发共用。"""
    return {
        "read_file": {
            "description": "读取工作目录内最多 4096 字节的 UTF-8 文件,不修改文件。",
            "parameters": {"path": str},
            "run": lambda path: read_file(workspace, path),
        },
        "run_main": {
            "description": "运行工作目录的 main.py,返回退出码和输出,不接受参数。",
            "parameters": {},
            "run": lambda: _run_main(workspace),
        },
    }


# 调用入口
try:
    outcome = read_file(workspace, 'main.py')
except (ValueError, OSError) as error:
    outcome = str(error)
print(outcome)
  1. resolve 解析工作目录和传入路径。
  2. 检查目标仍在工作目录中;越界立即拒绝,尚未打开文件。
  3. 最多读取 4097 字节,超过 4096 字节就报错。
  4. 以 UTF-8 解码;返回文本或交回异常。

在 with target.open("rb") 处设断点:正常输入能到达这里。切到“越界路径”后,同一个断点不会命中;异常在目录检查处产生,教学入口捕获并显示。切到“文件过大”,会完成有限读取,再在解码前拒绝。

真实实践中,这些异常由 execute_tool 捕获,封装成带原 ID 的错误结果。不存在的文件和无效 UTF-8 同样返回错误。本课的路径检查适合可信教学目录;它不消除检查与打开之间的文件系统竞态,也不限制 run_main 中程序的本机权限。两项工具的边界需要分别评估。

三个函数怎样接到现有入口

装配发生在公开入口 user_code/practice.py。它先确定工作目录,再建立注册表;同一份表分别传给模型适配器和分发器。循环继续接收两个依赖:ask(messages) 和 execute(call)。

python
tools = build_tools(workspace)
ask = lambda messages: ask_claude(client, model, messages, tools)
execute = lambda call: execute_tool(call, tools)
outcome = run_loop(ask, execute, max_calls=4, question=QUESTION)

这一段省略 CLI 和凭据读取;完整入口在公开仓库。ask_claude 仍只做一次 API 请求,其中 tools=tool_definitions(tools) 提供说明。run_loop 复用第一课的同一个函数,本次只加了可选 question 参数,以便初始化新问题;原来的调用方式和默认行为继续可用。

默认运行不需要 SDK 或 API Key。Python 3.10+ 标准库即可:

bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s02-tool-use/user_code/practice.py
python3 examples/s02-tool-use/user_code/verify_practice.py

默认模型替身按固定次序提出读取、运行、回答。文件读取和子进程是真实的,因此可以验证分发与记录,不能据此宣称已验证真实模型的工具选择。预期日志中的关键变化是:

text
模型调用 1 → read_file → print(6 * 7)
模型调用 2 → run_main → exit_code=0 / 42
模型调用 3 → answered

两次结果分别带 call_1、call_2,最终有六条消息:用户问题、读取请求、读取结果、运行请求、运行结果、最终回答。一轮里出现多个请求时,沿用第一课按原顺序逐个执行,再统一保存结果;本课没有新增并发执行器。

停止条件仍归循环管理

工具返回错误不等于任务结束:循环会保存错误,再调用模型。能否修正参数,要看后续响应。为避免反复失败,本课仍保留四次模型调用上限。

可以把入口改为 --max-calls 1。此时只完成读取并保存结果,状态为 call_limit,CLI 退出码为 1;没有得到运行结果,不能报告“已经运行”。收到可接受的最终文本才是 answered,而截断、协议矛盾或模型请求失败仍走第一课已有的停止分支。

如果要验证真实模型,先按公开 README 安装固定 SDK,在自己的终端配置凭据,再显式加 --live;本站浏览器不会发出模型请求。本章已验证离线选择、真实文件与进程、拒绝路径和 SDK 模拟 HTTP;尚未进行在线模型验证。

开篇任务现在能通过同一个循环取得两种证据。再加一个工具时,先定义它接受什么、会做什么、怎样失败,再把它接入注册表;循环中的记录关联与停止控制不应跟着复制一份。

资料核对于 2026-10-16:课程 Tool Use提供分发思路;官方工具调用文档说明协议往返;本篇代码的注册结构、文件上限和拒绝策略由本册自行设计。