先读,再运行:该改循环的哪一处?
先读
main.py,再运行它,报告文件内容和运行结果。
文件内容是 print(6 * 7)。这次既要取得源文本,也要取得退出码和输出。思想篇已经拆开了工具说明与本地实现。现在落到开发上:保留第一课负责消息与停止的 run_loop,把传入的执行函数换成通用分发器。
本课实现由本册独立编写,放在公开仓库的 examples/s02-tool-use/。新增职责集中在三个函数:
build_tools(workspace)返回同一份注册表,连接工具说明、参数形状与本地函数。execute_tool(call, tools)接收模型请求,校验并调用注册项,返回关联结果。read_file(workspace, path)处理文件路径、大小与 UTF-8 读取;已有程序执行继续复用第一课。
下面各函数的正文从公开源文件按符号提取。代码与图共用已有调试游标:点行号设断点,再点“调试”,用“下一步”或“继续”观察真实执行轨迹。浏览器中的文件和进程使用可展开查看的教学替身;下载仓库后的默认入口会真实读取文件并启动 Python,模型仍使用离线固定响应。
build_tools:只维护一份工具注册
先决定工具需要什么信息。本课只用必填字符串参数和空参数,因此每项包含 description、parameters、run 三项。parameters 保存字段名与 Python 类型,run 保存可调用对象。
workspace 在建立注册时由宿主传入,闭包把它绑定到实现。模型只能提供 path,不能用另一个参数替换工作目录。创建闭包时不会读取文件;执行发生在分发器真正调用 run 时。
真实执行轨迹回放;文件与进程使用公开教学替身,不访问实际文件、不启动进程。
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),
},
}点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
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))- read_file 注册用途、path 字符串参数和读取入口。
- run_main 注册用途、空参数和运行入口。
- 返回字典;尚未执行工具。模型说明从同一份注册生成。
在 "run": lambda path: ... 处停下,再走到函数末尾。返回注册表只完成了能力装配,既没有运行 read_file,也没有启动 main.py。run_main 对应的 _run_main 是一个薄适配:调用第一课的运行函数,将失败转换为异常供分发层统一封装。
要把注册表交给模型,还需要剥离 Python 函数对象。实际代码用 tool_definitions 生成 API 所需的说明:
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) 会让错误字段直接进入调用,错误也难以归到明确的契约检查。这里把检查放在实际实现之前。
真实执行轨迹回放;文件与进程使用公开教学替身,不访问实际文件、不启动进程。
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}点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
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"])- 建立 tool_result 并保留 call.id。
- 查名称,找不到就返回错误。
- 验证字段集合和字段类型;不匹配就拒绝调用。
- 调用注册项的 run,取得输出或异常。
- 返回 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 解码。这里选择拒绝大文件,避免模型把截断内容误当完整文件。
真实执行轨迹回放;文件与进程使用公开教学替身,不访问实际文件、不启动进程。
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")点击行号设置断点,再点“调试”;也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口,在单独文件名中执行;断点只回放上方源码。
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)- resolve 解析工作目录和传入路径。
- 检查目标仍在工作目录中;越界立即拒绝,尚未打开文件。
- 最多读取 4097 字节,超过 4096 字节就报错。
- 以 UTF-8 解码;返回文本或交回异常。
在 with target.open("rb") 处设断点:正常输入能到达这里。切到“越界路径”后,同一个断点不会命中;异常在目录检查处产生,教学入口捕获并显示。切到“文件过大”,会完成有限读取,再在解码前拒绝。
真实实践中,这些异常由 execute_tool 捕获,封装成带原 ID 的错误结果。不存在的文件和无效 UTF-8 同样返回错误。本课的路径检查适合可信教学目录;它不消除检查与打开之间的文件系统竞态,也不限制 run_main 中程序的本机权限。两项工具的边界需要分别评估。
三个函数怎样接到现有入口
装配发生在公开入口 user_code/practice.py。它先确定工作目录,再建立注册表;同一份表分别传给模型适配器和分发器。循环继续接收两个依赖:ask(messages) 和 execute(call)。
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+ 标准库即可:
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默认模型替身按固定次序提出读取、运行、回答。文件读取和子进程是真实的,因此可以验证分发与记录,不能据此宣称已验证真实模型的工具选择。预期日志中的关键变化是:
模型调用 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提供分发思路;官方工具调用文档说明协议往返;本篇代码的注册结构、文件上限和拒绝策略由本册自行设计。