一句话总结:给 Agent 加工具,就是加一个函数 + 一条配置。Agent Loop 本身一行都不用改。
一句话总结:给 Agent 加工具,就是加一个函数 + 一条配置。Agent Loop 本身一行都不用改。
safe_path 防止 AI 访问工作目录之外的文件(路径安全)回忆第 1 课,我们的 Agent 只有一个 bash 工具。这节课我们要加三个新工具:
| 工具名 | 功能 | 为什么需要 |
|---|---|---|
bash |
执行 shell 命令 | 万能工具,啥都能干 |
read_file |
读取文件内容 | 比 cat 更安全,有行数限制 |
write_file |
写入文件 | 比 echo > 更安全,有路径检查 |
edit_file |
精确替换文件内容 | 比整个文件重写更精准 |
你可能会问:有了 bash,为什么还要 read/write/edit?
答案:专用工具比通用工具更安全、更可控。
bash 里 AI 可以执行任何命令,你不知道它会干什么read_file 只能读文件,而且有路径安全限制write_file 只能在工作目录下写文件edit_file 只替换指定的文本,不会意外覆盖整个文件第 1 课的结构: 第 2 课的结构:
+-- 初始化配置 +-- 初始化配置
+-- System Prompt +-- System Prompt
+-- TOOLS(1个工具) +-- safe_path(新增!路径安全函数)
+-- run_bash(1个执行函数) +-- run_bash(和之前一样)
| +-- run_read(新增!)
| +-- run_write(新增!)
| +-- run_edit(新增!)
| +-- TOOL_HANDLERS(新增!工具分发表)
| +-- TOOLS(4个工具)
+-- agent_loop(核心循环) +-- agent_loop(几乎没变!)
+-- main +-- main关键发现: agent_loop 函数几乎没变!唯一的变化是查表调用工具,而不是写死调用 run_bash。
用户输入
|
v
+--------------------+
| 发送消息给 AI |<-----------------------------+
+--------------------+ |
| |
v |
+--------------------+ |
| AI 返回响应 | |
+--------------------+ |
| |
v |
+--------------------+ 是 |
| stop_reason 是 |---------> 遍历 tool_use 块 |
| "tool_use" ? | | |
+--------------------+ v |
| +-------------------+ |
| 不是 | TOOL_HANDLERS | |
v | 查表找到对应函数 | |
+----------------+ | 执行工具 | |
| 输出最终回答 | +-------------------+ |
+----------------+ | |
v |
把所有工具结果 |
拼成 user 消息 ----------------+和第 1 课的流程图几乎一样!唯一区别是中间多了一步"查表"。
和第 1 课一样:
bashpip install anthropic python-dotenv
.env 文件:
envANTHROPIC_API_KEY=sk-ant-xxxxx MODEL_ID=claude-sonnet-4-20250514 # ANTHROPIC_BASE_URL=https://your-proxy.com # 可选
python#!/usr/bin/env python3 """s02_tool_use.py - Tools(工具使用) 在第 1 课的基础上,新增了文件操作工具和工具分发表。 Agent Loop 本身几乎没变——这正是好架构的体现。 """ import os # 用于读取环境变量 import subprocess # 用于执行 shell 命令 from pathlib import Path # 用于安全的路径操作(比 os.path 更现代) from anthropic import Anthropic # Anthropic 官方 SDK from dotenv import load_dotenv # 从 .env 文件加载环境变量 # ============================================================ # 第一步:初始化配置(和第 1 课一样) # ============================================================ load_dotenv(override=True) if os.getenv("ANTHROPIC_BASE_URL"): os.environ.pop("ANTHROPIC_AUTH_TOKEN", None) # WORKDIR 记录工作目录,后面所有文件操作都限制在这个目录下 WORKDIR = Path.cwd() client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL")) MODEL = os.environ["MODEL_ID"] # ============================================================ # 第二步:System Prompt(和第 1 课类似) # ============================================================ SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain." # ============================================================ # 第三步:路径安全函数(新增!重要!) # ============================================================ def safe_path(p: str) -> Path: """ 把用户/AI 给的路径转换为绝对路径,并检查是否在工作目录内。 为什么需要这个? 假设 AI 要读取 "../../etc/passwd",这会跑到工作目录外面去。 safe_path 会检测到这种情况并报错,防止 AI 越权访问文件。 参数: p: AI 给的文件路径(可能是相对路径) 返回: 安全的绝对路径(Path 对象) 异常: ValueError: 如果路径跑到了工作目录外面 """ # WORKDIR / p:把相对路径拼接到工作目录下 # .resolve():解析掉所有的 ".."、"." 和软链接,得到真正的绝对路径 path = (WORKDIR / p).resolve() # is_relative_to:检查解析后的路径是不是在 WORKDIR 下面 # 如果 AI 用 "../../" 试图逃逸出去,resolve() 之后就不是 WORKDIR 的子路径了 if not path.is_relative_to(WORKDIR): raise ValueError(f"Path escapes workspace: {p}") return path # ============================================================ # 第四步:工具执行函数们 # ============================================================ # ---------- 工具 1:bash(和第 1 课一样) ---------- def run_bash(command: str) -> str: """执行一条 shell 命令,返回输出结果。""" # 安全检查:拦截危险命令 dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"] if any(d in command for d in dangerous): return "Error: Dangerous command blocked" try: r = subprocess.run(command, shell=True, cwd=WORKDIR, capture_output=True, text=True, timeout=120) out = (r.stdout + r.stderr).strip() return out[:50000] if out else "(no output)" except subprocess.TimeoutExpired: return "Error: Timeout (120s)" # ---------- 工具 2:read_file(新增!) ---------- def run_read(path: str, limit: int = None) -> str: """ 读取文件内容。 参数: path: 文件路径(相对于工作目录) limit: 最多读取多少行(可选,不传就读全部) 返回: 文件内容字符串,最多 50000 字符 """ try: # safe_path 确保路径安全,read_text 读取文件全部内容 text = safe_path(path).read_text() lines = text.splitlines() # 如果设了行数限制,只返回前 limit 行,并提示还有多少行没显示 if limit and limit < len(lines): lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"] # 拼回字符串,限制最大长度为 50000 字符(防止超大文件撑爆 token) return "\n".join(lines)[:50000] except Exception as e: return f"Error: {e}" # ---------- 工具 3:write_file(新增!) ---------- def run_write(path: str, content: str) -> str: """ 写入内容到文件。如果文件不存在就创建,如果父目录不存在也会自动创建。 参数: path: 文件路径(相对于工作目录) content: 要写入的内容 返回: 成功消息或错误信息 """ try: fp = safe_path(path) # mkdir(parents=True, exist_ok=True):自动创建所有缺失的父目录 # 比如写 "a/b/c.txt",如果 a/ 和 a/b/ 都不存在,会自动建好 fp.parent.mkdir(parents=True, exist_ok=True) fp.write_text(content) return f"Wrote {len(content)} bytes to {path}" except Exception as e: return f"Error: {e}" # ---------- 工具 4:edit_file(新增!) ---------- def run_edit(path: str, old_text: str, new_text: str) -> str: """ 在文件中查找 old_text,替换为 new_text(只替换第一次出现的)。 这比 write_file 更精确:不需要重写整个文件,只改你想改的部分。 AI 会先用 read_file 看到文件内容,然后精确指定要改哪一段。 参数: path: 文件路径 old_text: 要被替换的原始文本(必须精确匹配) new_text: 替换成的新文本 返回: 成功消息或错误信息 """ try: fp = safe_path(path) content = fp.read_text() # 必须能在文件中找到 old_text,否则报错 if old_text not in content: return f"Error: Text not found in {path}" # replace(..., 1):只替换第一次出现的,避免意外改到其他地方 fp.write_text(content.replace(old_text, new_text, 1)) return f"Edited {path}" except Exception as e: return f"Error: {e}" # ============================================================ # 第五步:工具分发表(新增!这是本课的设计亮点) # ============================================================ # 这是一个字典,key 是工具名称,value 是对应的执行函数 # 用 lambda 把参数名做一下转换/转发 TOOL_HANDLERS = { "bash": lambda **kw: run_bash(kw["command"]), "read_file": lambda **kw: run_read(kw["path"], kw.get("limit")), "write_file": lambda **kw: run_write(kw["path"], kw["content"]), "edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]), } # 为什么用分发表而不是 if-elif? # 1. 加新工具只需要加一行,不需要改 agent_loop # 2. 代码更清晰,一眼看到有哪些工具 # 3. 未来可以动态加载工具(比如从配置文件读取) # ============================================================ # 第六步:工具定义列表(给 AI 看的"菜单") # ============================================================ TOOLS = [ # 工具 1:bash(和第 1 课一样) { "name": "bash", "description": "Run a shell command.", "input_schema": { "type": "object", "properties": { "command": {"type": "string"} # 要执行的命令 }, "required": ["command"], }, }, # 工具 2:read_file(新增!) { "name": "read_file", "description": "Read file contents.", "input_schema": { "type": "object", "properties": { "path": {"type": "string"}, # 文件路径 "limit": {"type": "integer"}, # 最多读多少行(可选) }, "required": ["path"], }, }, # 工具 3:write_file(新增!) { "name": "write_file", "description": "Write content to file.", "input_schema": { "type": "object", "properties": { "path": {"type": "string"}, # 文件路径 "content": {"type": "string"}, # 要写入的内容 }, "required": ["path", "content"], }, }, # 工具 4:edit_file(新增!) { "name": "edit_file", "description": "Replace exact text in file.", "input_schema": { "type": "object", "properties": { "path": {"type": "string"}, # 文件路径 "old_text": {"type": "string"}, # 要替换的原始文本 "new_text": {"type": "string"}, # 替换成的新文本 }, "required": ["path", "old_text", "new_text"], }, }, ] # ============================================================ # 第七步:Agent Loop(核心循环——几乎没变!) # ============================================================ def agent_loop(messages: list): """ Agent 的核心循环。 和第 1 课对比,唯一的变化是: - 不再写死调用 run_bash - 而是从 TOOL_HANDLERS 字典里查找对应的处理函数 这就是"加工具不用改循环"的秘密。 """ while True: # 1. 发给 AI,带上所有工具定义 response = client.messages.create( model=MODEL, system=SYSTEM, messages=messages, tools=TOOLS, max_tokens=8000, ) # 2. 记录 AI 的回复 messages.append({"role": "assistant", "content": response.content}) # 3. AI 说完了?退出 if response.stop_reason != "tool_use": return # 4. 处理工具调用 results = [] for block in response.content: if block.type == "tool_use": # ---- 这里是唯一的变化 ---- # 从分发表里查找对应的处理函数 handler = TOOL_HANDLERS.get(block.name) # 如果找到了就执行,否则返回 "Unknown tool" 错误 output = handler(**block.input) if handler else f"Unknown tool: {block.name}" # 打印工具名和输出 print(f"> {block.name}:") print(output[:200]) results.append({ "type": "tool_result", "tool_use_id": block.id, "content": output, }) # 5. 把工具结果送回给 AI messages.append({"role": "user", "content": results}) # ============================================================ # 第八步:主程序入口(和第 1 课完全一样) # ============================================================ if __name__ == "__main__": history = [] # 对话历史 while True: try: query = input("\033[36ms02 >> \033[0m") except (EOFError, KeyboardInterrupt): break if query.strip().lower() in ("q", "exit", ""): break history.append({"role": "user", "content": query}) agent_loop(history) # 打印 AI 的最终文本回答 response_content = history[-1]["content"] if isinstance(response_content, list): for block in response_content: if hasattr(block, "text"): print(block.text) print()
pythondef safe_path(p: str) -> Path: path = (WORKDIR / p).resolve() if not path.is_relative_to(WORKDIR): raise ValueError(f"Path escapes workspace: {p}") return path
这三行代码是整个文件操作的安全基石。我们来看它怎么防止路径逃逸:
text假设 WORKDIR = /home/user/project 输入: "src/main.py" 拼接: /home/user/project/src/main.py resolve: /home/user/project/src/main.py 检查: 是 WORKDIR 的子路径 -> 通过 输入: "../../etc/passwd" 拼接: /home/user/project/../../etc/passwd resolve: /etc/passwd 检查: 不是 WORKDIR 的子路径 -> 报错!
为什么要用 .resolve()? 因为不 resolve 的话,路径里的 .. 不会被真正解析,你根本看不出它会跑到哪里去。
read_file —— 读文件
pythondef run_read(path: str, limit: int = None) -> str:
亮点在于 limit 参数。AI 读大文件时,不需要一次读完几万行。它可以先读前 50 行看看结构,再决定要不要继续读。这省 token,也更高效。
write_file —— 写文件
pythonfp.parent.mkdir(parents=True, exist_ok=True)
这一行是贴心设计。AI 想写 src/utils/helper.py,如果 src/utils/ 目录不存在,不会报错,而是自动创建。
edit_file —— 编辑文件
pythoncontent.replace(old_text, new_text, 1)
注意最后的 1——只替换第一次出现的位置。这很重要!如果文件里有多处相同的文本,我们不想把所有的都改掉,只改 AI 指定的那一处。
pythonTOOL_HANDLERS = { "bash": lambda **kw: run_bash(kw["command"]), "read_file": lambda **kw: run_read(kw["path"], kw.get("limit")), "write_file": lambda **kw: run_write(kw["path"], kw["content"]), "edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]), }
这是一个"名称 -> 函数"的映射表。AI 说"我要用 read_file 工具",我们就查表找到 run_read 函数去执行。
为什么不用 if-elif?
python# 不好的写法(每加一个工具就要改这里): if block.name == "bash": output = run_bash(block.input["command"]) elif block.name == "read_file": output = run_read(block.input["path"]) elif block.name == "write_file": output = run_write(block.input["path"], block.input["content"]) # ... 越来越长 # 好的写法(加工具只要在字典里加一行): handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input)
分发表的好处:
第 1 课的核心行:
pythonoutput = run_bash(block.input["command"]) # 写死调用 bash
第 2 课的核心行:
pythonhandler = TOOL_HANDLERS.get(block.name) # 查表找到对应函数 output = handler(**block.input) # 通用调用
就改了这两行。整个循环的骨架(while True、发消息、判断 stop_reason、收集结果、追加历史)一点没变。
这就是好架构的价值:核心循环是稳定的,工具是可插拔的。
text$ python s02_tool_use.py s02 >> 创建一个 hello.py 文件,写一个计算斐波那契数列的函数,然后运行它 > write_file: Wrote 190 bytes to hello.py > read_file: def fibonacci(n): if n <= 1: return n a, b = 0, 1 for _ in range(2, n + 1): a, b = b, a + b return b if __name__ == "__main__": for i in range(10): print(f"fib({i}) = {fibonacci(i)}") > bash: fib(0) = 0 fib(1) = 1 fib(2) = 1 fib(3) = 2 fib(4) = 3 fib(5) = 5 fib(6) = 8 fib(7) = 13 fib(8) = 21 fib(9) = 34 已经创建了 hello.py,包含一个斐波那契数列函数,并成功运行。 前 10 个斐波那契数为:0, 1, 1, 2, 3, 5, 8, 13, 21, 34。 s02 >> 把 range(10) 改成 range(15) > edit_file: Edited hello.py > bash: fib(0) = 0 fib(1) = 1 ... fib(14) = 377 已经把范围从 10 改成了 15,现在会输出前 15 个斐波那契数。 s02 >>
注意看第二次对话:AI 没有重写整个文件,而是精确地把 range(10) 替换成了 range(15)。这就是 edit_file 工具的威力。
TOOL_HANDLERS 里注册一下,在 TOOLS 里描述一下,就完事了。Agent Loop 不用动。bash 有危险命令黑名单,文件操作有 safe_path 路径检查。每个工具自己负责自己的安全。read_file 只能读文件不能执行命令,edit_file 只能改指定文本不能删文件。权限越小,风险越低。现在我们的 Agent 会用工具了,但每次对话都是从零开始。真正的 AI Agent 需要更多能力:
敬请期待后续课程!