🔧 模块2：Agent 工具调用能力实战

2.1 工具调用的核心逻辑

工具是 AI Agent 的"手"——让 Agent 不再只是"说说而已"，而是真正能做事情。核心逻辑：

工具调用 vs 普通函数调用

2.2 完整工具系统代码

创建 tool_agent.py — 一个拥有完整工具注册、发现、调用、错误恢复系统的 Agent。

#!/usr/bin/env python3
"""
ToolAgent — 完整工具调用系统实战
==================================
核心特性：
  ✅ 工具自动注册与发现
  ✅ Claude Code 内置工具风格
  ✅ 自定义工具开发全流程
  ✅ 工具调用失败自动恢复
  ✅ 参数验证与错误提示

运行：python3 tool_agent.py
"""

import json
import os
import sys
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Any, Callable
from functools import wraps

from dotenv import load_dotenv
from anthropic import Anthropic
import requests

load_dotenv()


# ================================================================
# 第1部分：工具注册系统
# ================================================================

class ToolRegistry:
    """
    工具注册中心 — 管理所有可用工具
    灵感来自 Claude Code 的内置工具系统
    """

    def __init__(self):
        self._tools: dict[str, dict] = {}

    def register(
        self,
        name: str,
        description: str,
        parameters: dict[str, str] = None,
        category: str = "general"
    ):
        """
        工具注册装饰器
        用法：
          @registry.register("get_weather", "获取天气", {"city": "城市名"})
          def get_weather(city: str) -> str: ...
        """
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)

            self._tools[name] = {
                "func": wrapper,
                "description": description,
                "parameters": parameters or {},
                "category": category,
                "registered_at": datetime.now().isoformat(),
            }
            return wrapper
        return decorator

    def get_tool(self, name: str) -> Callable | None:
        """获取工具函数"""
        tool = self._tools.get(name)
        return tool["func"] if tool else None

    def get_description(self) -> str:
        """生成所有工具的说明文档（给 Claude 看的）"""
        lines = ["## 可用工具列表\n"]
        for name, tool in self._tools.items():
            params_desc = ", ".join(
                [f"{k}: {v}" for k, v in tool["parameters"].items()]
            ) if tool["parameters"] else "无参数"
            lines.append(
                f"### {name}\n"
                f"- 功能: {tool['description']}\n"
                f"- 参数: {params_desc}\n"
                f"- 分类: {tool['category']}\n"
            )
        return "\n".join(lines)

    def list_tools(self) -> list[str]:
        """列出所有工具名"""
        return list(self._tools.keys())

    def get_tool_schema(self, name: str) -> dict | None:
        """获取工具的详细信息"""
        tool = self._tools.get(name)
        if not tool:
            return None
        return {
            "name": name,
            "description": tool["description"],
            "parameters": tool["parameters"],
            "category": tool["category"],
        }


# 创建全局注册中心
registry = ToolRegistry()


# ================================================================
# 第2部分：注册内置工具
# ================================================================

@registry.register(
    "get_time",
    "获取当前日期时间，支持指定时区",
    {"timezone": "时区，如 Asia/Shanghai（可选，默认本地时区）"},
    category="系统"
)
def get_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间"""
    now = datetime.now()
    weekdays = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
    wd = weekdays[now.weekday()]
    return f"🕐 {now.strftime('%Y年%m月%d日')} {wd} {now.strftime('%H:%M:%S')}"


@registry.register(
    "calculator",
    "安全地计算数学表达式，支持加减乘除、括号、百分比",
    {"expression": "数学表达式，如 '(100 + 50) * 0.8'"},
    category="系统"
)
def calculator(expression: str) -> str:
    """安全计算器"""
    allowed_chars = set("0123456789+-*/().% ^")
    clean = expression.strip().replace(" ", "")
    if not all(c in allowed_chars for c in clean):
        return "❌ 表达式包含不支持的字符"
    try:
        result = eval(clean, {"__builtins__": {}}, {})
        return f"🧮 {expression} = {result}"
    except Exception as e:
        return f"❌ 计算错误: {e}"


@registry.register(
    "read_file",
    "读取指定文件的内容，自动判断文本/二进制",
    {"filepath": "文件路径"},
    category="文件"
)
def read_file(filepath: str) -> str:
    """读取文件"""
    path = Path(filepath).expanduser()
    if not path.exists():
        return f"❌ 文件不存在: {filepath}"
    if path.stat().st_size > 1024 * 1024:  # 大于 1MB
        return f"❌ 文件过大 ({path.stat().st_size / 1024:.0f}KB)，拒绝读取"
    try:
        content = path.read_text("utf-8")
        if len(content) > 3000:
            content = content[:3000] + f"\n...(共 {len(content)} 字符，已截断)"
        return f"📄 {path.name}:\n{content}"
    except UnicodeDecodeError:
        return f"📄 {path.name}: [二进制文件，大小 {path.stat().st_size} 字节]"
    except Exception as e:
        return f"❌ 读取出错: {e}"


@registry.register(
    "write_file",
    "将内容写入文件（会覆盖已有文件）",
    {"filepath": "文件路径", "content": "要写入的内容"},
    category="文件"
)
def write_file(filepath: str, content: str = "") -> str:
    """写入文件"""
    path = Path(filepath).expanduser()
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")
        return f"✅ 已写入: {path.absolute()} ({len(content)} 字符)"
    except Exception as e:
        return f"❌ 写入失败: {e}"


@registry.register(
    "run_command",
    "执行一条安全的系统命令（白名单限制）",
    {"command": "要执行的命令"},
    category="系统"
)
def run_command(command: str) -> str:
    """安全执行命令（白名单限制）"""
    # 安全白名单：只允许这些命令
    allowed_prefixes = ["ls", "cat", "echo", "wc", "head", "tail", "date", "pwd", "whoami"]
    cmd_parts = command.strip().split()
    if not cmd_parts or cmd_parts[0] not in allowed_prefixes:
        return f"❌ 命令 '{cmd_parts[0] if cmd_parts else ''}' 不在白名单中。允许: {', '.join(allowed_prefixes)}"
    try:
        result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=10)
        output = result.stdout if result.returncode == 0 else result.stderr
        return f"💻 $ {command}\n{output[:1000]}"
    except subprocess.TimeoutExpired:
        return "❌ 命令执行超时 (10秒)"
    except Exception as e:
        return f"❌ 执行出错: {e}"


@registry.register(
    "web_search",
    "搜索网页内容（使用 DuckDuckGo，无需 API Key）",
    {"query": "搜索关键词"},
    category="网络"
)
def web_search(query: str) -> str:
    """网页搜索（需要 requests 库）"""
    try:
        # 注意：实际使用时可能需要更稳定的搜索API
        url = f"https://api.duckduckgo.com/?q={query}&format=json&no_html=1"
        resp = requests.get(url, timeout=10, headers={"User-Agent": "ToolAgent/1.0"})
        data = resp.json()
        abstract = data.get("AbstractText", "")
        if abstract:
            return f"🔍 搜索结果 ({query}):\n{abstract[:500]}"
        related = data.get("RelatedTopics", [])
        if related:
            items = [f"  • {r.get('Text', '')[:100]}" for r in related[:5]]
            return f"🔍 相关主题:\n" + "\n".join(items)
        return f"🔍 未找到与 '{query}' 直接相关的结果"
    except Exception as e:
        return f"❌ 搜索失败: {e}。提示：确保网络连接正常，或改用其他搜索API。"


@registry.register(
    "send_http_request",
    "发送 HTTP 请求到指定 URL",
    {"url": "目标URL", "method": "GET 或 POST（可选，默认GET）"},
    category="网络"
)
def send_http_request(url: str, method: str = "GET") -> str:
    """发送 HTTP 请求"""
    try:
        if method.upper() == "POST":
            resp = requests.post(url, timeout=10)
        else:
            resp = requests.get(url, timeout=10, headers={"User-Agent": "ToolAgent/1.0"})
        resp.raise_for_status()
        content_type = resp.headers.get("content-type", "")
        if "json" in content_type:
            body = json.dumps(resp.json(), ensure_ascii=False, indent=2)[:1000]
        else:
            body = resp.text[:1000]
        return f"🌐 {method.upper()} {url} → 状态: {resp.status_code}\n{body}"
    except requests.RequestException as e:
        return f"❌ HTTP 请求失败: {e}"


# ================================================================
# 第3部分：工具调用 Agent
# ================================================================

class ToolAgent:
    """具备完整工具调用系统的 Agent"""

    def __init__(self, name: str = "ToolAgent"):
        self.name = name
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"
        self.max_retries = 3  # 工具调用失败最大重试次数
        self.call_history: list[dict] = []  # 工具调用历史

    def _build_system_prompt(self) -> str:
        return f"""你是 {self.name}，一个具备强大工具调用能力的智能助手。

{registry.get_description()}

【工具调用规则 - 重要！】
当需要使用工具时，你必须严格按以下 JSON 格式回复（不要加任何其他文字）：
{{"tool": "工具名", "args": {{"参数名": "参数值", ...}}}}

如果不需要工具，直接回复即可。一次只能调用一个工具。

【错误恢复规则】
如果工具返回了错误（以 ❌ 开头），请分析错误原因并：
1. 如果是参数错误 → 修正参数后重试
2. 如果是工具不可用 → 尝试用其他工具替代
3. 如果无法解决 → 向用户说明情况

【重要提示】
- 从用户输入中提取工具参数，不要自己编造
- 工具名和参数名必须与列表中的完全一致
- 优先使用中文回复
"""

    def execute_tool(self, tool_name: str, args: dict) -> str:
        """执行工具，带重试和错误恢复"""
        tool_func = registry.get_tool(tool_name)
        if not tool_func:
            return f"❌ 工具 '{tool_name}' 不存在。可用工具: {registry.list_tools()}"

        try:
            # 调用工具函数，传入参数
            first_arg = list(args.values())[0] if args else ""
            result = tool_func(first_arg)

            # 记录调用历史
            self.call_history.append({
                "tool": tool_name,
                "args": args,
                "result": result[:200],
                "time": datetime.now().isoformat(),
                "success": not result.startswith("❌"),
            })

            return result
        except Exception as e:
            return f"❌ 工具执行异常: {e}"

    def run(self, user_input: str) -> str:
        """运行 Agent（带工具调用自动重试）"""
        print(f"\n{'─'*50}")
        print(f"👤 用户: {user_input}")

        system_prompt = self._build_system_prompt()
        messages = [{"role": "user", "content": user_input}]
        retry_count = 0

        while retry_count <= self.max_retries:
            try:
                response = self.client.messages.create(
                    model=self.model,
                    max_tokens=1024,
                    system=system_prompt,
                    messages=messages,
                    temperature=0.2,
                )
                reply = response.content[0].text.strip()
            except Exception as e:
                print(f"🤖 {self.name}: ❌ API调用失败: {e}")
                return f"❌ {e}"

            # 尝试解析工具调用
            tool_result = self._try_parse_tool_call(reply)
            if tool_result is None:
                # 不是工具调用，直接返回回复
                print(f"🤖 {self.name}: {reply}")
                return reply

            # 是工具调用
            tool_name, args = tool_result
            print(f"🔧 [{retry_count+1}/{self.max_retries+1}] 调用工具: {tool_name}({args})")
            result = self.execute_tool(tool_name, args)

            if result.startswith("❌") and retry_count < self.max_retries:
                # 工具失败，让 Agent 重试
                print(f"  ⚠️ 工具返回错误，让 Agent 调整策略...")
                messages = [
                    {"role": "user", "content": user_input},
                    {"role": "assistant", "content": reply},
                    {"role": "user", "content": f"工具返回了错误: {result}\n请分析错误并重新选择工具或修正参数。"}
                ]
                retry_count += 1
                continue

            print(f"🤖 {self.name}: {result}")
            return result

        return "❌ 多次重试后仍然失败，请检查工具配置或重试。"

    def _try_parse_tool_call(self, reply: str):
        """尝试解析工具调用JSON，返回 (tool_name, args) 或 None"""
        try:
            json_str = reply
            for marker in ["```json", "```"]:
                if marker in reply:
                    parts = reply.split(marker)
                    if len(parts) > 1:
                        json_str = parts[1].split("```")[0]
                    break
            parsed = json.loads(json_str.strip())
            if "tool" in parsed:
                return parsed["tool"], parsed.get("args", {})
        except (json.JSONDecodeError, KeyError, IndexError):
            pass
        return None

    def show_tool_stats(self):
        """显示工具使用统计"""
        if not self.call_history:
            print("📊 暂无工具调用记录")
            return
        print("\n📊 工具调用统计:")
        tool_counts = {}
        for call in self.call_history:
            name = call["tool"]
            tool_counts[name] = tool_counts.get(name, 0) + 1
        for name, count in tool_counts.items():
            success = sum(1 for c in self.call_history if c["tool"] == name and c["success"])
            print(f"  🔧 {name}: {count}次 (成功{success}次)")


# ================================================================
# 第4部分：交互入口
# ================================================================

def main():
    print("=" * 50)
    print("🔧 ToolAgent — 工具调用系统实战")
    print("=" * 50)
    print(f"已注册工具 ({len(registry.list_tools())}个):")
    for name in registry.list_tools():
        schema = registry.get_tool_schema(name)
        print(f"  • {name} — {schema['description']}")
    print("\n试试这些命令：")
    print("  • 计算 138 * 25 + 47")
    print("  • 现在几点了？")
    print("  • 读取 test.txt 的内容")
    print("  • 搜索 Python 教程")
    print("  • tools → 查看工具列表")
    print("  • stats → 查看工具使用统计")
    print("  • quit → 退出")
    print("=" * 50)

    if not os.getenv("ANTHROPIC_API_KEY"):
        print("❌ 请先配置 ANTHROPIC_API_KEY！")
        return

    agent = ToolAgent()

    # 创建测试文件
    Path("test.txt").write_text(
        "Hello from ToolAgent!\n这是测试文件，用于验证文件读取工具。\n",
        encoding="utf-8"
    )

    while True:
        try:
            user_input = input("\n👤 你: ").strip()
            if not user_input:
                continue
            if user_input.lower() in ("quit", "exit", "q"):
                break
            if user_input.lower() == "tools":
                for name in registry.list_tools():
                    s = registry.get_tool_schema(name)
                    params = ", ".join([f"{k}:{v}" for k, v in s["parameters"].items()])
                    print(f"  🔧 {name}({params}) — {s['description']}")
                continue
            if user_input.lower() == "stats":
                agent.show_tool_stats()
                continue

            agent.run(user_input)

        except KeyboardInterrupt:
            break

    agent.show_tool_stats()
    print(f"\n👋 {agent.name} 已退出。")


if __name__ == "__main__":
    main()

2.3 工具调用失败优化方案