2026年6月6日 4 分钟阅读

如何给 AI Agent 添加工具调用能力:从零实现 7 个核心工具

tinyash 0 条评论

你构建了一个基础 AI Agent:它能接收用户输入,调用大模型,维持对话上下文,然后输出回答。但你会发现一个问题——它只能聊天,不能做事

要让 Agent 真正有用,必须给它一种在环境中执行操作的能力。这就是工具调用(Tool Calling)——让 LLM 不仅能”说”,还能”做”。本文将手把手教你实现一个完整的工具系统,涵盖 Agent 最需要的 7 个核心工具。

为什么要自己实现工具系统?

你可能在想:直接用 Claude Code 或 Hermes Agent 不就好了?没错,但理解工具系统的实现原理有更深远的意义:

  • 可定制性:你能精确控制每个工具的安全边界(比如限制 bash 命令的执行范围)
  • 教育价值:掌握工具调用的底层逻辑后,调试 Agent 行为会轻松得多
  • 渐进式构建:从最简单的工具开始,逐步增加复杂能力

工具调用是如何工作的?

现代 LLM(如 Claude 4、GPT-5、DeepSeek V4)已经原生支持工具调用——它们经过微调,能输出结构化的 JSON 工具请求,而不是依赖不可靠的文本解析。流程如下:

  1. 系统提示词中声明可用工具及其参数 Schema
  2. Agent 判断是否需要调用工具来完成任务
  3. LLM 输出 JSON 格式的工具调用请求(含工具名和参数)
  4. Agent Harness 解析请求并执行对应函数
  5. 函数返回值作为新消息追加到对话中
  6. LLM 基于返回结果继续推理或输出最终回答

实现 7 个核心工具

以下所有工具基于 Python 实现。先创建 tools.py 子模块:

1. run_bash — 执行 Shell 命令

这是最强大的工具——允许 Agent 在宿主机器上执行任意 bash 命令。强大意味着危险,所以在实现时要注意安全边界。

import subprocess

def run_bash(command: str, timeout: int = 30) -> str:
    """Run a bash command and return its output."""
    try:
        result = subprocess.run(
            command, shell=True, capture_output=True,
            text=True, timeout=timeout
        )
        output = result.stdout
        if result.stderr:
            output += f"\n[stderr]\n{result.stderr}"
        return output[:10000]  # 截断以防输出过大
    except subprocess.TimeoutExpired:
        return f"Error: Command timed out after {timeout}s"
    except Exception as e:
        return f"Error: {e}"

2. read_file — 读取文件内容

读取文件是 Agent 了解项目结构的基础能力。支持行号偏移和数量限制,避免读取超大文件淹没上下文。

def read_file(path: str, offset: int = 1, limit: int = 200) -> str:
    """Read lines from a file, with optional offset and limit."""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            lines = f.readlines()
        total = len(lines)
        start = max(0, offset - 1)
        end = min(start + limit, total)
        content = ''.join(lines[start:end])
        return f"File: {path} ({total} lines total)\n{content}"
    except FileNotFoundError:
        return f"Error: File not found: {path}"
    except Exception as e:
        return f"Error: {e}"

3. find_files — 按 Glob 模式查找文件

Agent 需要先”看到”你的项目结构,才能知道该读什么文件。

import glob as glob_module

def find_files(pattern: str, root: str = '.') -> str:
    """Find files matching a glob pattern inside a directory."""
    matches = glob_module.glob(f"{root}/{pattern}", recursive=True)
    if not matches:
        return f"No files matching '{pattern}' in {root}"
    result = '\n'.join(matches[:50])
    if len(matches) > 50:
        result += f"\n... and {len(matches) - 50} more"
    return result

4. search_content — 在文件中搜索文本

配合文件查找使用:先找到文件,再搜索关键内容。

import re

def search_content(pattern: str, file_glob: str = '*', root: str = '.') -> str:
    """Search file contents for a regex pattern."""
    import glob as g
    matches = []
    for path in g.glob(f"{root}/**/{file_glob}", recursive=True):
        try:
            with open(path, 'r', encoding='utf-8') as f:
                for i, line in enumerate(f, 1):
                    if re.search(pattern, line, re.IGNORECASE):
                        matches.append(f"{path}:{i}: {line.rstrip()[:200]}")
        except (UnicodeDecodeError, IsADirectoryError):
            continue
    if not matches:
        return f"No matches for '{pattern}'"
    return '\n'.join(matches[:30])

5. write_file — 写入文件

能让 Agent 创建和修改文件,是它真正”做事”的关键。

import os

def write_file(path: str, content: str) -> str:
    """Write content to a file, creating parent directories if needed."""
    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    with open(path, 'w', encoding='utf-8') as f:
        f.write(content)
    return f"Written {len(content)} bytes to {path}"

6. patch_file — 精确替换文件内容

当你只需要改一行代码而不是重写整个文件时,精确替换比完整写回更安全。

def patch_file(path: str, old_string: str, new_string: str) -> str:
    """Replace the first occurrence of old_string with new_string."""
    with open(path, 'r', encoding='utf-8') as f:
        content = f.read()
    if old_string not in content:
        return f"Error: old_string not found in {path}"
    content = content.replace(old_string, new_string, 1)
    with open(path, 'w', encoding='utf-8') as f:
        f.write(content)
    return f"Patched {path}: replaced '{old_string[:30]}...'"

7. fetch_webpage — 抓取网页内容

让 Agent 能联网获取信息。使用 BeautifulSoup 去除 HTML 标签,只保留纯文本。

import urllib.request
from bs4 import BeautifulSoup

def fetch_webpage(url: str) -> str:
    """Fetch a URL and return its plain-text content (up to 2 MB)."""
    if not url.startswith(('http://', 'https://')):
        return "Error: Only http/https URLs allowed"
    try:
        req = urllib.request.Request(url, headers={
            'User-Agent': 'Mozilla/5.0'
        })
        resp = urllib.request.urlopen(req, timeout=15)
        html = resp.read(2 * 1024 * 1024).decode('utf-8', errors='replace')
        soup = BeautifulSoup(html, 'html.parser')
        text = soup.get_text(separator='\n', strip=True)
        return text[:8000]
    except Exception as e:
        return f"Error fetching {url}: {e}"

定义工具 Schema 并集成到 Agent

每个工具都需要一个 Schema,让 LLM 知道它能做什么、需要什么参数:

TOOLS = [
    {
        "name": "run_bash",
        "description": "Run a bash command on the user's machine and return the output.",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"type": "string", "description": "Bash command to execute"}
            },
            "required": ["command"]
        }
    },
    {
        "name": "read_file",
        "description": "Read lines from a file with line numbers.",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path"},
                "offset": {"type": "integer", "description": "First line to read (1-indexed)"},
                "limit": {"type": "integer", "description": "Max lines to return"}
            },
            "required": ["path"]
        }
    },
    # ... 其他工具的 Schema 类似
]

TOOL_MAP = {
    "run_bash": run_bash,
    "read_file": read_file,
    "find_files": find_files,
    "search_content": search_content,
    "write_file": write_file,
    "patch_file": patch_file,
    "fetch_webpage": fetch_webpage,
}

然后在 Agent 主循环中处理工具调用:

def execute_tool(name: str, args: dict) -> str:
    """Execute a tool and return its result as a string."""
    if name not in TOOL_MAP:
        return f"Error: Unknown tool '{name}'"
    try:
        return TOOL_MAP[name](**args)
    except Exception as e:
        return f"Error executing {name}: {e}"

实战测试

有了这些工具后,你的 Agent 已经可以做很多事情了:

你:帮我抓取 ruxu.dev 的首页,列出所有文章标题
Agent:我来帮你。

Agent 调用了 fetch_webpage("https://www.ruxu.dev/articles")
→ 获取页面内容

Agent 调用了 write_file("/tmp/ruxu_articles.md", "...")
→ 将文章列表写入文件

Agent:已完成!我抓取了 ruxu.dev 首页,提取了文章列表
并保存到 /tmp/ruxu_articles.md 文件中。

从原型到生产级

这 7 个工具构成了一个可用但简陋的 Agent 工具系统。要实现生产级体验,还需要:

  1. 安全沙箱:用 Docker 或 gVisor 隔离 bash 和文件操作
  2. 并发控制:支持多工具并行执行,提升响应速度
  3. 速率限制:防止 Agent 在无限循环中消耗大量 Token
  4. 审计日志:记录每个工具调用并支持回滚操作
  5. 错误重试:对临时性失败(网络波动、文件锁定)自动重试

本文的完整代码可以在 ruxu.dev 找到。这是一个系列教程的第三篇,后续还会覆盖 MCP 协议集成、安全加固和规划能力等内容。

总结

工具调用是 AI Agent 从”对话玩具”进化为”生产力工具”的关键能力。通过实现这 7 个基础工具——bash 执行、文件读写、搜索、网页抓取——你的 Agent 已经具备了完成大多数开发任务的基础设施。核心原则是:工具定义边界,Scheme 驱动调用,循环完成闭环。理解这个模式后,你可以轻松扩展任意自定义工具,让 Agent 真正成为你的开发伙伴。

发表评论

你的邮箱地址不会被公开,带 * 的为必填项。